Auto-generated reference

Every entry below is rendered straight from its docstring in the source — signature, type hints, and the actual rationale text, not a hand-maintained copy of it. If this page and api.md or a docstring ever disagree, the docstring is the truth; open an issue.

Grouped the same way as the roadmap's API-stability tiers: Core first, then "in the box". Recipes stay documented by example in recipes.md rather than duplicated here — a recipe's value is in how it's used, not its bare signature.

Core

Context

Context

Context(resources: RuntimeResources | None = None)

Central artifact store with an event queue.

Git-like model: every applied commit forms a new Context version (head). Commits chain through parent_id and carry a reads/writes trace — the actual agent linkage via consumes/produces.

Source code in reactifact/context.py

def __init__(self, resources: RuntimeResources | None = None):
    self._artifacts: dict[str, Artifact[Any]] = {}
    self._events: list[Event] = []
    self._log = CommitLog()
    self.resources = resources or RuntimeResources()
    self._hub = EventHub()
    self._relations = RelationGraph()
    self._base: Context | None = None
    self._fork_name: str = ""
    # Incrementally maintained (§ dependents index): artifact ids whose
    # producing commit read a source that has since moved to a newer
    # version. Kept up to date by `update()`/`log_commit()` so
    # `stale_artifacts()`/`has_stale()` never rescan the whole context.
    self._stale: set[str] = set()
    # Incrementally maintained: exact `type(data)` -> ids of that exact
    # type. Kept up to date by `create()`/`update()`/`delete()` so
    # `list_artifacts(T)` doesn't call `isinstance()` per artifact — it
    # unions the ids of every *distinct type ever created* that is an
    # `issubclass` of `T` (a small set — bounded by distinct types, not
    # artifact count) into an O(1) membership test, still walked in
    # insertion order to preserve tie-break order in callers' own sorts.
    # Any bulk rewrite that bypasses create/update/delete must call
    # `_reindex_by_type()` (mirrors `_recompute_stale()`, same reasoning).
    self._by_type: dict[type, set[str]] = {}

version property

version: int

Current Context version (number of applied commits).

head_id property

head_id: str | None

Id of the last commit (HEAD).

announce

announce(
    message: str, *, kind: str = "status", **data: Any
) -> None

Publishes a progress event to active streams (no-op without subscribers).

kind is a category for the application: "status" (domain agent statuses), "agent" (internal, from framework producers like ToolUse). The application itself decides which kinds to show the user.

Source code in reactifact/context.py

def announce(self, message: str, *, kind: str = "status", **data: Any) -> None:
    """Publishes a progress event to active streams (no-op without subscribers).

    `kind` is a category for the application: "status" (domain agent statuses),
    "agent" (internal, from framework producers like ToolUse).
    The application itself decides which kinds to show the user.
    """
    if self._hub.has_subscribers:
        self._hub.publish(
            ProgressEvent(
                kind=kind,
                message=message,
                data=data,
            )
        )

create

create(
    data: TData, id: str | None = None
) -> Artifact[TData]

Creates a new artifact and generates an ARTIFACT_CREATED event.

If a stable id is given and an artifact with it already exists, returns the existing one without creating a duplicate or an event (idempotency, §42).

Source code in reactifact/context.py

def create(self, data: TData, id: str | None = None) -> Artifact[TData]:
    """Creates a new artifact and generates an ARTIFACT_CREATED event.

    If a stable id is given and an artifact with it already exists, returns
    the existing one without creating a duplicate or an event (idempotency, §42).
    """
    if id is not None and id in self._artifacts:
        return self._artifacts[id]
    if id is None and self.resources.id_factory is not None:
        id = self.resources.id_factory(type(data).__name__)
    artifact = Artifact(data=data, id=id)
    self._artifacts[artifact.id] = artifact
    self._by_type.setdefault(type(data), set()).add(artifact.id)
    self._events.append(
        Event(
            type=EventType.ARTIFACT_CREATED,
            artifact_type=type(data),
            artifact_id=artifact.id,
        )
    )
    return artifact

get

get(artifact_id: str) -> Artifact[Any] | None

Returns the artifact by id or None.

Source code in reactifact/context.py

def get(self, artifact_id: str) -> Artifact[Any] | None:
    """Returns the artifact by id or None."""
    return self._artifacts.get(artifact_id)

update

update(
    artifact_id: str, new_data: TData
) -> Artifact[TData] | None

Updates artifact data, creates a new version, generates ARTIFACT_UPDATED.

If the data did not change, the version and the event are left untouched: no-op patches must not cascade into reactions (§41, §42).

Source code in reactifact/context.py

def update(self, artifact_id: str, new_data: TData) -> Artifact[TData] | None:
    """Updates artifact data, creates a new version, generates ARTIFACT_UPDATED.

    If the data did not change, the version and the event are left untouched:
    no-op patches must not cascade into reactions (§41, §42).
    """
    artifact = self._artifacts.get(artifact_id)
    if artifact is None:
        return None
    if artifact.data == new_data:
        return artifact
    old_type = type(artifact.data)
    artifact.update(new_data)
    new_type = type(new_data)
    if new_type is not old_type:
        self._by_type.get(old_type, set()).discard(artifact_id)
        self._by_type.setdefault(new_type, set()).add(artifact_id)
    self._events.append(
        Event(
            type=EventType.ARTIFACT_UPDATED,
            artifact_type=type(new_data),
            artifact_id=artifact.id,
        )
    )
    for dependent in self._dependents_of(artifact_id):
        self._events.append(
            Event(
                type=EventType.ARTIFACT_STALE,
                artifact_type=type(dependent.data),
                artifact_id=dependent.id,
            )
        )
        self._stale.add(dependent.id)
    return artifact

delete

delete(artifact_id: str) -> bool

Deletes the artifact and generates ARTIFACT_DELETED.

Source code in reactifact/context.py

def delete(self, artifact_id: str) -> bool:
    """Deletes the artifact and generates ARTIFACT_DELETED."""
    artifact = self._artifacts.pop(artifact_id, None)
    if artifact is None:
        return False
    self._stale.discard(artifact_id)
    self._by_type.get(type(artifact.data), set()).discard(artifact_id)
    self._events.append(
        Event(
            type=EventType.ARTIFACT_DELETED,
            artifact_type=type(artifact.data),
            artifact_id=artifact.id,
        )
    )
    return True

list_artifacts

list_artifacts(
    artifact_type: None = None,
) -> list[Artifact[Any]]

list_artifacts(
    artifact_type: type[TArtifact],
) -> list[Artifact[TArtifact]]

list_artifacts(
    artifact_type: type[TArtifact] | None = None,
) -> list[Artifact[Any]]

Returns a list of artifacts, optionally filtered by data type.

Preserves insertion order (matters: ties in a caller's own sort key, e.g. updated_at, break in creation order, same as before this method stopped isinstance-scanning every artifact).

Source code in reactifact/context.py

def list_artifacts(
    self, artifact_type: type[TArtifact] | None = None
) -> list[Artifact[Any]]:
    """Returns a list of artifacts, optionally filtered by data type.

    Preserves insertion order (matters: ties in a caller's own sort key,
    e.g. `updated_at`, break in creation order, same as before this
    method stopped `isinstance`-scanning every artifact).
    """
    if artifact_type is None:
        return list(self._artifacts.values())
    matching_ids: set[str] = set()
    for t, ids in self._by_type.items():
        if issubclass(t, artifact_type):
            matching_ids |= ids
    return [
        cast(Artifact[TArtifact], a)
        for aid, a in self._artifacts.items()
        if aid in matching_ids
    ]

latest

latest(
    artifact_type: type[TArtifact],
) -> Artifact[TArtifact] | None

The most recently created artifact of a type, or None.

Sugar over list_artifacts for "grab the latest answer/finding" — the common read after a run.

Source code in reactifact/context.py

def latest(self, artifact_type: type[TArtifact]) -> Artifact[TArtifact] | None:
    """The most recently created artifact of a type, or None.

    Sugar over `list_artifacts` for "grab the latest answer/finding" —
    the common read after a run.
    """
    artifacts = self.list_artifacts(artifact_type)
    if not artifacts:
        return None
    return max(artifacts, key=lambda a: a.created_at)

view

view(
    artifact_type: type[BaseModel]
    | tuple[type[BaseModel], ...]
    | None = None,
    *,
    condition: Callable[[Artifact[Any]], bool]
    | None = None,
    limit: int | None = None,
) -> View

Artifact projection for the agent (§27): by type/condition/limit.

The View does not copy state — it is references to artifacts plus serialization for the prompt. tokens_estimate lets the agent stay within the token budget (§58): build the view, check the estimate, reduce limit if needed.

Source code in reactifact/context.py

def view(
    self,
    artifact_type: type[BaseModel] | tuple[type[BaseModel], ...] | None = None,
    *,
    condition: Callable[[Artifact[Any]], bool] | None = None,
    limit: int | None = None,
) -> View:
    """Artifact projection for the agent (§27): by type/condition/limit.

    The View does not copy state — it is references to artifacts plus
    serialization for the prompt. `tokens_estimate` lets the agent stay within
    the token budget (§58): build the view, check the estimate, reduce `limit`
    if needed.
    """
    if artifact_type is None:
        artifacts = list(self._artifacts.values())
    else:
        artifacts = [
            a for a in self._artifacts.values() if isinstance(a.data, artifact_type)
        ]
    if condition is not None:
        artifacts = [a for a in artifacts if condition(a)]
    if limit is not None:
        artifacts = artifacts[:limit]
    return View(artifacts=artifacts)

link

link(
    source_id: str, relation: str, target_id: str
) -> Relation

Establishes a link source_id —relation→ target_id (idempotently, §42).

Source code in reactifact/context.py

def link(self, source_id: str, relation: str, target_id: str) -> Relation:
    """Establishes a link `source_id —relation→ target_id` (idempotently, §42)."""
    return self._relations.link(source_id, relation, target_id)

unlink

unlink(
    source_id: str,
    relation: str | None = None,
    target_id: str | None = None,
) -> int

Removes links; relation/target_id = None mean "any".

Source code in reactifact/context.py

def unlink(
    self,
    source_id: str,
    relation: str | None = None,
    target_id: str | None = None,
) -> int:
    """Removes links; `relation`/`target_id` = None mean "any"."""
    return self._relations.unlink(source_id, relation, target_id)

relations

relations(
    source_id: str | None = None,
    relation: str | None = None,
    target_id: str | None = None,
) -> list[Relation]

All links, optionally filtered by any edge component.

Source code in reactifact/context.py

def relations(
    self,
    source_id: str | None = None,
    relation: str | None = None,
    target_id: str | None = None,
) -> list[Relation]:
    """All links, optionally filtered by any edge component."""
    return self._relations.relations(source_id, relation, target_id)

incoming

incoming(
    target_id: str, relation: str | None = None
) -> list[Relation]

Links pointing at target_id (for provenance: who references what).

Source code in reactifact/context.py

def incoming(self, target_id: str, relation: str | None = None) -> list[Relation]:
    """Links pointing at `target_id` (for provenance: who references what)."""
    return self.relations(target_id=target_id, relation=relation)

related

related(
    source_id: str, relation: str | None = None
) -> list[Artifact[Any]]

Target artifacts of outgoing links (existing ones; "dangling" ones are skipped).

Source code in reactifact/context.py

def related(
    self, source_id: str, relation: str | None = None
) -> list[Artifact[Any]]:
    """Target artifacts of outgoing links (existing ones; "dangling" ones are skipped)."""
    targets: list[Artifact[Any]] = []
    seen: set[str] = set()
    for rel in self.relations(source_id=source_id, relation=relation):
        artifact = self._artifacts.get(rel.target_id)
        if artifact is not None and rel.target_id not in seen:
            targets.append(artifact)
            seen.add(rel.target_id)
    return targets

dangling_relations

dangling_relations() -> list[Relation]

Links with a non-existent source or target (§69): the state is visible, not hidden in a string.

Source code in reactifact/context.py

def dangling_relations(self) -> list[Relation]:
    """Links with a non-existent source or target (§69): the state is visible,
    not hidden in a string."""
    return [
        rel
        for rel in self._relations.values()
        if rel.source_id not in self._artifacts
        or rel.target_id not in self._artifacts
    ]

interrupt

interrupt(
    question: str,
    *,
    kind: str = "general",
    notes: dict[str, Any] | None = None,
) -> Artifact[PendingQuestion]

Poses a question to a human: creates a PendingQuestion in the context.

Source code in reactifact/context.py

def interrupt(
    self,
    question: str,
    *,
    kind: str = "general",
    notes: dict[str, Any] | None = None,
) -> Artifact[PendingQuestion]:
    """Poses a question to a human: creates a PendingQuestion in the context."""
    return self.create(
        PendingQuestion(question=question, kind=kind, notes=notes or {})
    )

pending_questions

pending_questions() -> list[Artifact[PendingQuestion]]

Unanswered questions awaiting the human.

Source code in reactifact/context.py

def pending_questions(self) -> list[Artifact[PendingQuestion]]:
    """Unanswered questions awaiting the human."""
    return [a for a in self.list_artifacts(PendingQuestion) if not a.data.answered]

resume

resume(
    question_id: str, answer: str
) -> Artifact[PendingQuestion] | None

A human's answer is a regular patch: marks the question as answered.

Generates ARTIFACT_UPDATED, which agents subscribed to PendingQuestion(answered=True) react to.

Source code in reactifact/context.py

def resume(self, question_id: str, answer: str) -> Artifact[PendingQuestion] | None:
    """A human's answer is a regular patch: marks the question as answered.

    Generates ARTIFACT_UPDATED, which agents subscribed to
    PendingQuestion(answered=True) react to.
    """
    artifact = self._artifacts.get(question_id)
    if artifact is None or not isinstance(artifact.data, PendingQuestion):
        return None
    updated = artifact.data.model_copy(
        update={
            "answered": True,
            "resolution": answer,
            "resolved_at": datetime.now(UTC),
        }
    )
    return self.update(question_id, updated)

drain_events

drain_events() -> list[Event]

Drains and clears the event queue.

Source code in reactifact/context.py

def drain_events(self) -> list[Event]:
    """Drains and clears the event queue."""
    events = self._events
    self._events = []
    return events

clone

clone() -> Context

Deep copy of this context's live state. See reactifact.branching.

Source code in reactifact/context.py

def clone(self) -> Context:
    """Deep copy of this context's live state. See `reactifact.branching`."""
    from .branching import clone_context

    return clone_context(self)

merge_from

merge_from(other: Context) -> None

Two-way merge, no conflict detection. See reactifact.branching.

Source code in reactifact/context.py

def merge_from(self, other: Context) -> None:
    """Two-way merge, no conflict detection. See `reactifact.branching`."""
    from .branching import merge_context_from

    merge_context_from(self, other)

branch

branch(*, name: str = '') -> Context

Forks an isolated copy for alternative state exploration (§39).

The fork records a snapshot of its base, so a later merge of two fork-mates can detect diverged artifacts three-way (§40). The branch shares resources with the parent but is otherwise fully independent: subsequent changes on either side do not affect the other. Algorithm lives in reactifact.branching.fork_context.

Source code in reactifact/context.py

def branch(self, *, name: str = "") -> Context:
    """Forks an isolated copy for alternative state exploration (§39).

    The fork records a snapshot of its base, so a later `merge` of two
    fork-mates can detect diverged artifacts three-way (§40). The branch
    shares `resources` with the parent but is otherwise fully independent:
    subsequent changes on either side do not affect the other. Algorithm
    lives in `reactifact.branching.fork_context`.
    """
    from .branching import fork_context

    return fork_context(self, name=name)

merge

merge(
    other: Context, *, message: str = "Merged branch"
) -> None

Merges other into self with explicit conflicts, atomically (§40).

Three-way merge against the shared fork base (the fork snapshot of other, or of self when other has none) — raises MergeConflict (reactifact.branching.MergeConflict, re-exported as reactifact.MergeConflict) rather than silently choosing a side. Algorithm lives in reactifact.branching.merge_contexts.

Source code in reactifact/context.py

def merge(self, other: Context, *, message: str = "Merged branch") -> None:
    """Merges `other` into `self` with explicit conflicts, atomically (§40).

    Three-way merge against the shared fork base (the fork snapshot of
    `other`, or of `self` when `other` has none) — raises `MergeConflict`
    (`reactifact.branching.MergeConflict`, re-exported as `reactifact.MergeConflict`)
    rather than silently choosing a side. Algorithm lives in
    `reactifact.branching.merge_contexts`.
    """
    from .branching import merge_contexts

    merge_contexts(self, other, message=message)

log_commit

log_commit(commit: Commit) -> None

Applies the commit to the repository: fills in parent/version, moves head.

Source code in reactifact/context.py

def log_commit(self, commit: Commit) -> None:
    """Applies the commit to the repository: fills in parent/version, moves head."""
    self._log.append(commit)
    # A commit that (re-)writes an artifact refreshes it against its
    # current reads — it can no longer be in the stale set.
    for write in commit.writes:
        self._stale.discard(write.artifact_id)

history

history() -> list[Commit]

History: an ordered chain of commits from the oldest to head.

Source code in reactifact/context.py

def history(self) -> list[Commit]:
    """History: an ordered chain of commits from the oldest to head."""
    return self._log.history()

diff

diff(version_a: int, version_b: int) -> dict[str, Any]

State delta between two Context versions.

Returns {"added": {id: data}, "removed": {id: data}, "changed": {id: {old, new}}}. The diff compares the versioned state (commits); artifacts created directly outside commits (the "working tree") do not participate.

Source code in reactifact/context.py

def diff(self, version_a: int, version_b: int) -> dict[str, Any]:
    """State delta between two Context versions.

    Returns {"added": {id: data}, "removed": {id: data}, "changed": {id: {old, new}}}.
    The diff compares the versioned state (commits); artifacts created directly
    outside commits (the "working tree") do not participate.
    """
    if not (0 <= version_a <= version_b <= self._log.version):
        raise ValueError(
            f"Invalid versions: {version_a}..{version_b} (head={self._log.version})"
        )
    snap_a = self._log.replay_state(version_a)
    snap_b = self._log.replay_state(version_b)
    result: dict[str, Any] = {"added": {}, "removed": {}, "changed": {}}
    for aid in snap_b.keys() - snap_a.keys():
        result["added"][aid] = snap_b[aid].model_dump()
    for aid in snap_a.keys() - snap_b.keys():
        result["removed"][aid] = snap_a[aid].model_dump()
    for aid in snap_a.keys() & snap_b.keys():
        if snap_a[aid] != snap_b[aid]:
            result["changed"][aid] = {
                "old": snap_a[aid].model_dump(),
                "new": snap_b[aid].model_dump(),
            }
    return result

snapshot

snapshot() -> dict[str, Any]

Consistent snapshot of the current artifact state (id → data).

Source code in reactifact/context.py

def snapshot(self) -> dict[str, Any]:
    """Consistent snapshot of the current artifact state (id → data)."""
    return {aid: art.data.model_dump() for aid, art in self._artifacts.items()}

stale_artifacts

stale_artifacts() -> list[Artifact[Any]]

Artifacts whose parents (reads in the producing commit) are now newer versions.

Dependencies are built from the actual reads recorded by the runtime via consumes — a link derived from execution, not an author-drawn graph. Backed by the incrementally-maintained _stale set (kept current by update()/log_commit()), not a rescan of every artifact.

Source code in reactifact/context.py

def stale_artifacts(self) -> list[Artifact[Any]]:
    """Artifacts whose parents (reads in the producing commit) are now newer versions.

    Dependencies are built from the actual reads recorded by the runtime via
    consumes — a link derived from execution, not an author-drawn graph.
    Backed by the incrementally-maintained `_stale` set (kept current by
    `update()`/`log_commit()`), not a rescan of every artifact.
    """
    return [self._artifacts[aid] for aid in self._stale if aid in self._artifacts]

checkout

checkout(version: int) -> None

Moves head back to a previous version (rollback along the commit chain).

Artifacts not part of the versioned history (created directly outside commits — the "working tree") are preserved.

Source code in reactifact/context.py

def checkout(self, version: int) -> None:
    """Moves head back to a previous version (rollback along the commit chain).

    Artifacts not part of the versioned history (created directly outside
    commits — the "working tree") are preserved.
    """
    if not (0 <= version <= self._log.version):
        raise ValueError(
            f"Invalid checkout version: {version} (head={self._log.version})"
        )
    touched: set[str] = set()
    for commit in self._log.commits_from(version):
        for op in commit.operations:
            if isinstance(op, (Create, Update, Delete)) and op.artifact_id:
                touched.add(op.artifact_id)
    rebuilt = self._rebuild_artifacts_from_commits(version)
    for aid, art in self._artifacts.items():
        if aid not in touched:
            rebuilt[aid] = art
    self._artifacts = rebuilt

    # Relations: those committed up to `version` plus the "working tree"
    # (created directly outside commits) are kept, as with artifacts. Links
    # introduced by commits in the [version:] range are rolled back.
    committed_now = self._log.replay_relations(self._log.version)
    working_tree_rels = {
        key: rel for key, rel in self._relations.items() if key not in committed_now
    }
    self._relations = RelationGraph.from_mapping(
        {**self._log.replay_relations(version), **working_tree_rels}
    )

    self._events = []
    self._log.truncate(version)
    self._recompute_stale()
    self._reindex_by_type()

to_kv async

to_kv(backend: KVBackend, key: str) -> None

Serializes and stores this context under key in a KV backend.

The one to_dict() round-trip shared by SessionStore/BranchStore (session_id / branch keys are just a naming convention over the same backend, §39) — call this instead of hand-rolling backend.set(key, context.to_dict()).

Source code in reactifact/context.py

async def to_kv(self, backend: KVBackend, key: str) -> None:
    """Serializes and stores this context under `key` in a KV backend.

    The one `to_dict()` round-trip shared by `SessionStore`/`BranchStore`
    (session_id / branch keys are just a naming convention over the same
    backend, §39) — call this instead of hand-rolling `backend.set(key,
    context.to_dict())`.
    """
    await backend.set(key, self.to_dict())

from_kv async classmethod

from_kv(backend: KVBackend, key: str) -> Context | None

Loads a context previously stored with to_kv, or None if absent.

Source code in reactifact/context.py

@classmethod
async def from_kv(cls, backend: KVBackend, key: str) -> Context | None:
    """Loads a context previously stored with `to_kv`, or None if absent."""
    data = await backend.get(key)
    return cls.from_dict(data) if data is not None else None

Artifact

Artifact

Artifact(
    data: TData,
    id: str | None = None,
    created_by_commit: str | None = None,
)

Bases: Generic[TData]

Wrapper around a Pydantic model with versioning.

Source code in reactifact/artifacts.py

def __init__(
    self,
    data: TData,
    id: str | None = None,
    created_by_commit: str | None = None,
):
    self.id = id or str(uuid.uuid4())
    self.data = data
    self.data_type = f"{type(data).__module__}.{type(data).__qualname__}"
    self.created_at = datetime.now(UTC)
    self.updated_at = self.created_at
    self.created_by_commit = created_by_commit
    self._history: list[
        TData
    ] = []  # previous data versions (excluding the current one)
    # Memoized to_dict(), keyed by version: re-derived on every save()
    # otherwise (session persistence saves after every commit, §reactifact.session)
    # even though most artifacts in a large context are unchanged since the
    # last save — see reactifact/session.py for the calling context.
    self._dict_cache: tuple[int, dict[str, Any]] | None = None

history property

history: list[TData]

Returns a copy of the list of previous versions (excluding the current one).

version property

version: int

Current version (0 – original, 1 – after the first update, etc.)

update

update(new_data: TData) -> None

Saves the current version to history and replaces the data.

Source code in reactifact/artifacts.py

def update(self, new_data: TData) -> None:
    """Saves the current version to history and replaces the data."""
    self._history.append(self.data)
    self.data = new_data
    self.updated_at = datetime.now(UTC)

get_all_versions

get_all_versions() -> list[TData]

Returns all data versions, including the current one, from oldest to newest.

Source code in reactifact/artifacts.py

def get_all_versions(self) -> list[TData]:
    """Returns all data versions, including the current one, from oldest to newest."""
    return self._history + [self.data]

diff

diff(old_version: int, new_version: int) -> dict[str, Any]

Returns a diff between two versions by their numbers (0 – the oldest).

Source code in reactifact/artifacts.py

def diff(self, old_version: int, new_version: int) -> dict[str, Any]:
    """Returns a diff between two versions by their numbers (0 – the oldest)."""
    versions = self.get_all_versions()
    if old_version < 0 or new_version >= len(versions) or old_version > new_version:
        raise ValueError(
            f"Invalid version indices: old={old_version}, new={new_version}"
        )
    old_data = versions[old_version].model_dump()
    new_data = versions[new_version].model_dump()
    return compute_dict_diff(old_data, new_data)

to_dict

to_dict() -> dict[str, Any]

Serializes the artifact, including its full version history.

Memoized per version (bumped by update()): an unchanged artifact returns the same dict instance on a later call instead of re-walking model_dump() over its entire history again — the history only grows monotonically, so a cache keyed by version can never go stale.

Source code in reactifact/artifacts.py

def to_dict(self) -> dict[str, Any]:
    """Serializes the artifact, including its full version history.

    Memoized per `version` (bumped by `update()`): an unchanged artifact
    returns the same dict instance on a later call instead of re-walking
    `model_dump()` over its entire history again — the history only grows
    monotonically, so a cache keyed by version can never go stale.
    """
    if self._dict_cache is not None and self._dict_cache[0] == self.version:
        return self._dict_cache[1]
    d = {
        "id": self.id,
        "data_type": self.data_type,
        "data": self.data.model_dump(mode="json"),
        "created_at": self.created_at.isoformat(),
        "updated_at": self.updated_at.isoformat(),
        "history": [v.model_dump(mode="json") for v in self._history],
        "created_by_commit": self.created_by_commit,
    }
    self._dict_cache = (self.version, d)
    return d

Effects

Effects

Effects(context: Context)

The current produce's effect set (creates/updates/links to commit once).

Source code in reactifact/effects.py

def __init__(self, context: Context):
    self._context = context
    self.operations: list[Operation] = []

create

create(data: Any, *, id: str | None = None) -> Handle

Plans a new artifact; returns a linkable/updatable handle (§38).

If id names an artifact that already exists when this effect is applied, the runtime treats it as a refresh (a new version of the same logical entity, §42/§43) rather than creating a duplicate — the same rule Runtime._apply_patch documents. Use upsert at the call site when that "may already exist" intent should be explicit instead of implicit in a plain create(..., id=...).

Source code in reactifact/effects.py

def create(self, data: Any, *, id: str | None = None) -> Handle:
    """Plans a new artifact; returns a linkable/updatable handle (§38).

    If `id` names an artifact that already exists when this effect is
    applied, the runtime treats it as a refresh (a new version of the
    same logical entity, §42/§43) rather than creating a duplicate — the
    same rule `Runtime._apply_patch` documents. Use `upsert` at the call
    site when that "may already exist" intent should be explicit instead
    of implicit in a plain `create(..., id=...)`.
    """
    if id is None:
        factory = self._context.resources.id_factory
        id = (
            factory(type(data).__name__)
            if factory is not None
            else _auto_id(type(data).__name__)
        )
    self.operations.append(Create(data, id=id))
    return Handle(self, id, data)

create_once

create_once(data: Any, *, id: str) -> Handle | None

Idempotent create (§42): None if id already exists in the context, otherwise the same as create(data, id=id).

Folds the "already done" guard every produce needs for a re-derived id (f"answer:{qid}") into the call itself:

handle = self.effects.create_once(Answer(...), id=f"answer:{qid}")
if handle is None:
    return None  # already answered — nothing to do

instead of a separate if context.get(f"answer:{qid}") is not None: return None above the call — one less place to get the id string wrong between the guard and the create.

Source code in reactifact/effects.py

def create_once(self, data: Any, *, id: str) -> Handle | None:
    """Idempotent create (§42): `None` if `id` already exists in the
    context, otherwise the same as `create(data, id=id)`.

    Folds the "already done" guard every produce needs for a re-derived
    id (`f"answer:{qid}"`) into the call itself:

        handle = self.effects.create_once(Answer(...), id=f"answer:{qid}")
        if handle is None:
            return None  # already answered — nothing to do

    instead of a separate `if context.get(f"answer:{qid}") is not None:
    return None` above the call — one less place to get the id string
    wrong between the guard and the create.
    """
    if self._context.get(id) is not None:
        return None
    return self.create(data, id=id)

create_once_from

create_once_from(
    source: Artifact[Any] | Handle | str,
    data: Any,
    *,
    prefix: str | None = None,
) -> Handle | None

Idempotent create whose id is derived from another artifact (§42).

Folds the single most common way a re-derivable id is built by hand — f"answer:{question.id}", f"review:{pr.id}" — into the call, so a produce doesn't have to invent the id string (nor get it subtly wrong between its guard and its create):

answer = self.effects.create_once_from(question, Answer(...))
if answer is None:
    return None  # already answered this question

prefix defaults to the created model's class name, lowercased (Answer → answer:{source_id}). source may be an Artifact, an effects Handle, or a plain id string. None back means an artifact with the derived id already exists — skip, exactly like create_once.

Source code in reactifact/effects.py

def create_once_from(
    self,
    source: Artifact[Any] | Handle | str,
    data: Any,
    *,
    prefix: str | None = None,
) -> Handle | None:
    """Idempotent create whose id is derived from another artifact (§42).

    Folds the single most common way a re-derivable id is built by hand —
    `f"answer:{question.id}"`, `f"review:{pr.id}"` — into the call, so a
    produce doesn't have to invent the id string (nor get it subtly wrong
    between its guard and its `create`):

        answer = self.effects.create_once_from(question, Answer(...))
        if answer is None:
            return None  # already answered this question

    `prefix` defaults to the created model's class name, lowercased
    (`Answer` → `answer:{source_id}`). `source` may be an `Artifact`, an
    effects `Handle`, or a plain id string. `None` back means an artifact
    with the derived id already exists — skip, exactly like `create_once`.
    """
    source_id = _id_of(source)
    artifact_id = f"{prefix or type(data).__name__.lower()}:{source_id}"
    return self.create_once(data, id=artifact_id)

upsert

upsert(data: Any, *, id: str) -> Handle

Explicit create-or-refresh: same effect as create(data, id=id).

Purely a call-site name: identical to create(..., id=...), but says at the call site that a refresh is an expected outcome, not a surprise — reach for it when the artifact may already exist (e.g. a re-derived id like f"answer:{qid}").

Source code in reactifact/effects.py

def upsert(self, data: Any, *, id: str) -> Handle:
    """Explicit create-or-refresh: same effect as `create(data, id=id)`.

    Purely a call-site name: identical to `create(..., id=...)`, but says
    at the call site that a refresh is an expected outcome, not a
    surprise — reach for it when the artifact may already exist (e.g. a
    re-derived id like `f"answer:{qid}"`).
    """
    return self.create(data, id=id)

update

update(artifact: Artifact[Any], **fields: Any) -> Effects

Bumps fields of an existing artifact (a new version).

Note the name means something different here than on Patch.update/ Context.update/Artifact.update (a full data replacement) — this is the one intentional exception, matching Patch.update_fields instead. Effects is the everyday authoring surface where "update some fields" is the common case (§18 above), so it gets the short name; the lower-level, less-used Patch/Context/Artifact surface keeps update for the operation it's actually named after.

Source code in reactifact/effects.py

def update(self, artifact: Artifact[Any], **fields: Any) -> Effects:
    """Bumps fields of an *existing* artifact (a new version).

    Note the name means something different here than on `Patch.update`/
    `Context.update`/`Artifact.update` (a full data replacement) — this is
    the one intentional exception, matching `Patch.update_fields`
    instead. `Effects` is the everyday authoring surface where "update
    some fields" is the common case (§18 above), so it gets the short
    name; the lower-level, less-used `Patch`/`Context`/`Artifact` surface
    keeps `update` for the operation it's actually named after.
    """
    new_data = artifact.data.model_copy(update=fields)
    self.operations.append(Update(artifact.id, new_data))
    return self

ask

ask(
    question: str,
    *,
    kind: str = "general",
    notes: dict[str, Any] | None = None,
    id: str | None = None,
) -> Handle

Poses a question to a human (HITL, §60): creates a PendingQuestion.

Returns a handle you can link later; the human answer is recorded via effects.resume(question_art, resolution) (§60). Pass id for a stable, re-derivable question (e.g. f"steer:{qid}:{round}") so a guard like if context.get(id) is not None: return None can stop the produce from asking again while the question is still unanswered.

Source code in reactifact/effects.py

def ask(
    self,
    question: str,
    *,
    kind: str = "general",
    notes: dict[str, Any] | None = None,
    id: str | None = None,
) -> Handle:
    """Poses a question to a human (HITL, §60): creates a `PendingQuestion`.

    Returns a handle you can link later; the human answer is recorded via
    `effects.resume(question_art, resolution)` (§60). Pass `id` for a
    stable, re-derivable question (e.g. `f"steer:{qid}:{round}"`) so a
    guard like `if context.get(id) is not None: return None` can stop the
    produce from asking again while the question is still unanswered.
    """
    from .interrupt import PendingQuestion

    return self.create(
        PendingQuestion(question=question, kind=kind, notes=dict(notes or {})),
        id=id,
    )

resume

resume(question: Any, resolution: str) -> Effects

Records the human answer on a PendingQuestion (HITL, §60).

Source code in reactifact/effects.py

def resume(self, question: Any, resolution: str) -> Effects:
    """Records the human answer on a `PendingQuestion` (HITL, §60)."""
    from datetime import UTC, datetime

    self.update(
        question,
        answered=True,
        resolution=resolution,
        resolved_at=datetime.now(UTC),
    )
    return self

to_patch

to_patch() -> Patch

Compiles the effects into a Patch (the runtime's transport).

Source code in reactifact/effects.py

def to_patch(self) -> Patch:
    """Compiles the effects into a `Patch` (the runtime's transport)."""
    patch = Patch()
    for op in self.operations:
        patch.add(op)
    return patch

Patch

Patch

Patch(operations: list[Operation] | None = None)

An ordered set of operations to apply to the Context (§12).

Source code in reactifact/patches.py

def __init__(self, operations: list[Operation] | None = None):
    self.operations: list[Operation] = operations if operations is not None else []

update_fields

update_fields(
    artifact: Artifact[Any], **fields: Any
) -> Patch

Update artifact fields without rebuilding the model.

Sugar over update: it does the model_copy(update=fields) itself and puts the full new model into Update (preserving commit replayability).

Source code in reactifact/patches.py

def update_fields(self, artifact: Artifact[Any], **fields: Any) -> Patch:
    """Update artifact fields without rebuilding the model.

    Sugar over `update`: it does the `model_copy(update=fields)` itself and puts
    the full new model into `Update` (preserving commit replayability).
    """
    return self.update(artifact.id, artifact.data.model_copy(update=fields))

merge

merge(*patches: Patch | None) -> Patch

Adds operations from patches to this patch; None are skipped.

Returns self (chaining, like add/create/update/delete).

Source code in reactifact/patches.py

def merge(self, *patches: Patch | None) -> Patch:
    """Adds operations from `patches` to this patch; None are skipped.

    Returns `self` (chaining, like `add`/`create`/`update`/`delete`).
    """
    for patch in patches:
        if patch is not None:
            self.operations.extend(patch.operations)
    return self

Agent

Agent

Agent(
    name: str | None = None,
    triggers: list[Trigger] | None = None,
    priority: int | None = None,
)

Bases: ABC

Base container: consumes some artifacts and produces others.

If run is not overridden, automatically collects inputs according to consumes and calls produce on every produce, merging the patches.

Source code in reactifact/agents.py

def __init__(
    self,
    name: str | None = None,
    triggers: list[Trigger] | None = None,
    priority: int | None = None,
):
    self.name = name or self.name or self.__class__.__name__

    self.priority = (
        priority if priority is not None else getattr(self.__class__, "priority", 0)
    )

    if triggers is not None:
        self.triggers = list(triggers)
    elif self.triggers:
        self.triggers = list(self.triggers)
    elif self.consumes is not None:
        self.triggers = self._generate_triggers_from_consumes()
    else:
        self.triggers = []

    self._validate_contracts()

matching_triggers

matching_triggers(
    event: Event, context: Context | None = None
) -> list[Trigger]

Every trigger that matches event — matches() is just bool(...) of this. Runtime._arun_once_impl uses the full list (not just the bool) to decide whether every matching trigger asks for debouncing (Trigger.debounce) before collapsing repeat events into one run.

Source code in reactifact/agents.py

def matching_triggers(
    self, event: Event, context: Context | None = None
) -> list[Trigger]:
    """Every trigger that matches `event` — `matches()` is just
    `bool(...)` of this. `Runtime._arun_once_impl` uses the full list
    (not just the bool) to decide whether *every* matching trigger asks
    for debouncing (`Trigger.debounce`) before collapsing repeat events
    into one run."""
    return [trigger for trigger in self.triggers if trigger.matches(event, context)]

collect_inputs

collect_inputs(context: Context) -> list[Artifact[Any]]

Public access to the consumed artifacts.

Used by the runtime to record the reads linkage (provenance) on run.

Source code in reactifact/agents.py

def collect_inputs(self, context: Context) -> list[Artifact[Any]]:
    """Public access to the consumed artifacts.

    Used by the runtime to record the reads linkage (provenance) on run.
    """
    return self._collect_inputs(context)

run async

run(event: Event, context: Context) -> Patch | None

Default: runs self.produces via execute() (the effects-first path — write a Produce subclass or @produce function instead of overriding this).

Overriding run() to return a Patch by hand is a low-level, internal escape hatch for cases effects genuinely can't express — not a third everyday produce style (see reactifact.produce's module docstring for the two you should reach for first). No example in this repo overrides it; only this repo's own tests do.

Source code in reactifact/agents.py

async def run(self, event: Event, context: Context) -> Patch | None:
    """Default: runs `self.produces` via `execute()` (the effects-first
    path — write a `Produce` subclass or `@produce` function instead of
    overriding this).

    Overriding `run()` to return a `Patch` by hand is a low-level,
    internal escape hatch for cases `effects` genuinely can't express —
    not a third everyday produce style (see `reactifact.produce`'s module
    docstring for the two you should reach for first). No example in
    this repo overrides it; only this repo's own tests do.
    """
    await self.execute(context, event)
    return None

execute async

execute(
    context: Context, event: Event | None = None
) -> None

Runs the agent's produces (usually on an event).

Effects-first (§24): produces write self.effects.*/call.effects.* and return None; the runtime compiles the effect slot into one atomic patch. This method only runs the produces — it does not build a patch. (run remains the agent-level escape hatch for custom Agent subclasses that assemble a change-set by hand; the runtime merges its result after the effects.)

Source code in reactifact/agents.py

async def execute(self, context: Context, event: Event | None = None) -> None:
    """Runs the agent's produces (usually on an event).

    Effects-first (§24): produces write `self.effects.*`/`call.effects.*`
    and return None; the *runtime* compiles the effect slot into one
    atomic patch. This method
    only *runs* the produces — it does not build a patch. (`run` remains the
    agent-level escape hatch for custom Agent subclasses that assemble a
    change-set by hand; the runtime merges its result after the effects.)
    """
    if not self.produces:
        return None
    inputs = self._collect_inputs(context)
    for p in self.produces:
        runs, trigger = self._resolve_produce_call(p, event, context)
        if not runs:
            continue
        call = ProduceCall(
            context=context, inputs=inputs, event=event, trigger=trigger
        )
        await p.produce(call)
    return None

Produce

Produce

Produce(
    artifact_type: ArtifactType | None = None,
    *,
    also_creates: Sequence[ArtifactType] | None = None,
    reacts_to: ArtifactType
    | Sequence[ArtifactType]
    | None = None,
)

Bases: Generic[TOut]

Describes the produced artifact type and how it is created.

artifact_type is auto-derived from the generic when a subclass is written as class X(Produce[Foo]) — write it explicitly only to override or when the class has no generic (e.g. programmatic Produce(Foo)).

A produce whose produce() body legitimately creates more than one artifact type declares the extra ones in also_creates — either as a class attribute (also_creates = (Bar, Baz)) or a constructor argument (Produce(Foo, also_creates=[Bar])). Runtime._validate_patch_types checks every Create op an agent's generation produces against the union of artifact_type/also_creates across that agent's produces list — without declaring Bar here, a self.effects.create(Bar(...)) inside artifact_type = Foo's own produce() raises "not declared in produces" at commit time, since nothing recorded that this produce is allowed to write it. Before also_creates existed, the only way to widen that set was an inert second Produce(Bar) placeholder added to the agent's produces list purely so its unused artifact_type got unioned in — correct, but nothing at the call site said why that placeholder was there; also_creates puts the declaration on the produce that actually does the writing.

The input-side mirror of that is reacts_to: an agent with several consumes and several produces runs every produce on every matching event by default (Agent.execute() has no idea which of an agent's several Consumes a given produce actually cares about) — every produce ends up guarding itself by hand, if call.event is None or not isinstance(call.trigger.data, TheOneTypeICareAbout): return None, at the top of its own body. Declaring reacts_to = (TheType,) (a tuple, same convention as also_creates — more than one entry for more than one type) moves that guard to where the intent already lives — the class declaration — and Agent.execute() simply skips calling produce() at all for an event none of reacts_to matches, the same way Trigger.artifact_type gates whether the agent wakes up at all. Two produces sharing one artifact_type (the same output) but different reacts_to (different triggers) is exactly why this can't just reuse artifact_type for both directions — see examples/product code with a FinalizeWithDocuments/DirectFinalize-shaped pair, both producing the same result type from two different upstream events. None (the default) means unrestricted — today's behavior, unchanged, so this is purely additive.

Once reacts_to narrows which event a produce runs for, resolving that event's own artifact is still work every such produce repeats — ProduceCall.trigger does it once, in Agent.execute(), for every produce that reads it. For a produce that also declares reacts_to, this is a real guarantee, not best-effort: Agent.execute() also skips calling it if context.get(event.artifact_id) no longer resolves for a CREATED/UPDATED/STALE event (the artifact was deleted by another agent earlier in the same generation — the same race Trigger.matches() documents) — so call.trigger is never None when such a produce actually runs, and the body needs no guard at all. This guarantee does not apply to a DELETED event on the produce's own reacts_to type: there, context.get(...) correctly returning None is the event, not a race, so call.trigger is None and the produce still runs — deletion-reacting code is expected to handle that itself (call.event still carries artifact_id/artifact_type there). Without reacts_to, call.trigger is still resolved on a best-effort basis whenever call.event is not None, but never a reason to skip the call — there is no per-type contract to enforce.

Source code in reactifact/produce.py

def __init__(
    self,
    artifact_type: ArtifactType | None = None,
    *,
    also_creates: Sequence[ArtifactType] | None = None,
    reacts_to: ArtifactType | Sequence[ArtifactType] | None = None,
):
    self.artifact_type = artifact_type or self.__class__.artifact_type
    if self.artifact_type is None:
        raise ValueError(
            "artifact_type must be provided either as class attribute or constructor argument"
        )
    self.also_creates = (
        tuple(also_creates)
        if also_creates is not None
        else self.__class__.also_creates
    )
    resolved_reacts_to = (
        reacts_to if reacts_to is not None else self.__class__.reacts_to
    )
    # Normalizes the constructor argument's wider "single type or
    # sequence" shape down to `reacts_to`'s own plain tuple-or-`None`
    # shape — a bare `Produce(Foo, reacts_to=Bar)` call site is
    # convenient; the class attribute (set by `self.__class__.reacts_to`
    # above when no constructor argument was given) is always already a
    # tuple, same convention as `also_creates`.
    if resolved_reacts_to is None:
        self.reacts_to = None
    elif isinstance(resolved_reacts_to, tuple):
        self.reacts_to = resolved_reacts_to
    else:
        self.reacts_to = (cast(ArtifactType, resolved_reacts_to),)

effects property

effects: Effects

The produce-scoped effect slot (authoring surface, §24).

Only meaningful inside produce(): the runtime pushes a fresh slot per execution. Returns an error outside a run. Equivalent to (and backed by the same contextvar as) call.effects on the ProduceCall the current produce() call received — kept here too so a class-style produce can write self.effects.create(...) without threading call through every helper method.

produce async

produce(call: ProduceCall) -> None

No-op by default (§24).

Subclass-style overrides write self.effects.* (or call.effects.*) and return None; a None return means "no work". A bare Produce(Model) (no override) is a valid, deliberate no-op — used e.g. to widen an agent's allowed Create types when the actual write happens via self.effects.ask(...) in another produce (see examples/supervisor).

Source code in reactifact/produce.py

async def produce(self, call: ProduceCall) -> None:
    """No-op by default (§24).

    Subclass-style overrides write `self.effects.*` (or `call.effects.*`)
    and return None; a `None` return means "no work". A bare
    `Produce(Model)` (no override) is a valid, deliberate no-op — used
    e.g. to widen an agent's allowed `Create` types when the actual
    write happens via `self.effects.ask(...)` in another produce (see
    `examples/supervisor`).
    """
    return None

Consume

Consume

Consume(
    artifact_type: ArtifactType | None = None,
    condition: Callable[[Artifact[Any]], bool]
    | None = None,
    event_types: Sequence[EventType] | None = None,
    *,
    wakes: bool | None = None,
    debounce: bool | None = None,
)

Describes the consumed artifact type, condition and triggering events.

All parameters can be set as class attributes (for inheritance) or passed to the constructor.

wakes (default True): whether this Consume also contributes to the agent's triggers (Agent.matches() — does the agent run at all). False means "read this as input, but don't wake up on it" — e.g. an agent that should run when a Question arrives but also wants the existing ConversationHistory as input, without re-running once per history artifact. For the declarative style (consumes/produces, the common case), this covers the one real reason to decouple "what wakes me" from "what I read" without reaching for Agent's separate triggers= override. triggers= itself still exists and is not deprecated — it's the only option for the imperative style (an Agent subclass that overrides run() directly and has no consumes at all to attach a condition to; see Agent.triggers's own docstring).

debounce (default False): when several events in the same generation would each independently wake this agent via this Consume (a fan-out step creating five Evidence artifacts in one commit, five separate ARTIFACT_CREATED events), collapse them into a single run instead of five — the fifth (last) event is what the runtime happens to pass as event, but a debounced produce should read inputs (collected fresh from Context regardless of which event triggered the run) rather than rely on event for "what changed", since that's exactly the information debouncing discards. See Runtime._arun_once_impl for where the collapsing actually happens — Consume/Trigger only carry the flag, they don't implement it, since collapsing needs to compare other events in the same drained batch, which is Runtime-level state neither of them has access to.

Source code in reactifact/consume.py

def __init__(
    self,
    artifact_type: ArtifactType | None = None,
    condition: Callable[[Artifact[Any]], bool] | None = None,
    event_types: Sequence[EventType] | None = None,
    *,
    wakes: bool | None = None,
    debounce: bool | None = None,
):
    self.artifact_type = artifact_type or self.__class__.artifact_type
    if self.artifact_type is None:
        raise ValueError(
            "artifact_type must be provided either as class attribute or constructor argument"
        )

    self.condition = (
        condition if condition is not None else self.__class__.condition
    )
    self.event_types = list(
        event_types if event_types is not None else self.__class__.event_types
    )
    self.wakes = wakes if wakes is not None else self.__class__.wakes
    self.debounce = debounce if debounce is not None else self.__class__.debounce
    if self.debounce and not self.wakes:
        raise ValueError(
            "Consume(debounce=True, wakes=False) is meaningless: debounce "
            "only collapses repeat wake-ups, and wakes=False means this "
            "Consume never wakes the agent in the first place"
        )

to_triggers

to_triggers() -> list[Trigger]

Converts into a list of triggers for automatic reaction.

Empty when wakes=False — this Consume still feeds Agent._collect_inputs(), it just never causes Agent.matches() to fire on its own.

Source code in reactifact/consume.py

def to_triggers(self) -> list[Trigger]:
    """Converts into a list of triggers for automatic reaction.

    Empty when `wakes=False` — this `Consume` still feeds
    `Agent._collect_inputs()`, it just never causes `Agent.matches()`
    to fire on its own.
    """
    if not self.wakes:
        return []
    return [
        Trigger(
            event_type, self.artifact_type, self.condition, debounce=self.debounce
        )
        for event_type in self.event_types
    ]

collect

collect(context: Context) -> list[Artifact[Any]]

The artifacts this Consume contributes to an agent's inputs.

Agent._collect_inputs() calls this once per entry in consumes — overridden by CorrelatedConsume to return a correlated group across several types instead of one type's own matching instances.

Source code in reactifact/consume.py

def collect(self, context: Context) -> list[Artifact[Any]]:
    """The artifacts this `Consume` contributes to an agent's inputs.

    `Agent._collect_inputs()` calls this once per entry in `consumes` —
    overridden by `CorrelatedConsume` to return a correlated group across
    several types instead of one type's own matching instances.
    """
    artifacts = context.list_artifacts(self.artifact_type)
    if self.condition:
        artifacts = [a for a in artifacts if self.condition(a)]
    return artifacts

by_status classmethod

by_status(
    artifact_type: ArtifactType,
    status: str,
    event_types: Sequence[EventType] = (
        EventType.ARTIFACT_CREATED,
        EventType.ARTIFACT_UPDATED,
    ),
    *,
    wakes: bool = True,
    debounce: bool = False,
) -> Consume

Creates a Consume with a condition on the equality of the status field.

Source code in reactifact/consume.py

@classmethod
def by_status(
    cls,
    artifact_type: ArtifactType,
    status: str,
    event_types: Sequence[EventType] = (
        EventType.ARTIFACT_CREATED,
        EventType.ARTIFACT_UPDATED,
    ),
    *,
    wakes: bool = True,
    debounce: bool = False,
) -> Consume:
    """Creates a Consume with a condition on the equality of the status field."""
    return cls(
        artifact_type,
        condition=lambda art: getattr(art.data, "status", None) == status,
        event_types=event_types,
        wakes=wakes,
        debounce=debounce,
    )

by_field classmethod

by_field(
    artifact_type: ArtifactType,
    field: str,
    value: Any,
    event_types: Sequence[EventType] = (
        EventType.ARTIFACT_CREATED,
        EventType.ARTIFACT_UPDATED,
    ),
    *,
    wakes: bool = True,
    debounce: bool = False,
) -> Consume

Creates a Consume with a condition on the equality of an arbitrary field.

Source code in reactifact/consume.py

@classmethod
def by_field(
    cls,
    artifact_type: ArtifactType,
    field: str,
    value: Any,
    event_types: Sequence[EventType] = (
        EventType.ARTIFACT_CREATED,
        EventType.ARTIFACT_UPDATED,
    ),
    *,
    wakes: bool = True,
    debounce: bool = False,
) -> Consume:
    """Creates a Consume with a condition on the equality of an arbitrary field."""
    return cls(
        artifact_type,
        condition=lambda art: getattr(art.data, field, None) == value,
        event_types=event_types,
        debounce=debounce,
        wakes=wakes,
    )

Runtime

Runtime

Runtime(
    context: Context,
    agents: list[Agent] | None = None,
    max_concurrency: int | None = None,
    session: Session | None = None,
    budget: Budget | None = None,
    tracer: Tracer | list[Tracer] | None = None,
    scheduler: Scheduler | None = None,
    isolate_errors: bool = False,
    on_agent_error: Callable[
        [Agent, Event, BaseException], None
    ]
    | None = None,
    session_save_policy: Literal[
        "per_commit", "per_turn"
    ] = "per_commit",
)

Source code in reactifact/runtime.py

def __init__(
    self,
    context: Context,
    agents: list[Agent] | None = None,
    max_concurrency: int | None = None,
    session: "Session | None" = None,
    budget: Budget | None = None,
    tracer: Tracer | list[Tracer] | None = None,
    scheduler: Scheduler | None = None,
    isolate_errors: bool = False,
    on_agent_error: Callable[[Agent, Event, BaseException], None] | None = None,
    session_save_policy: Literal["per_commit", "per_turn"] = "per_commit",
):
    self.context = context
    self.agents = agents or []
    self.max_concurrency = max_concurrency
    self.session = session
    self.budget = budget
    self.scheduler = scheduler
    # "per_commit" (default): git-like persist after every single commit
    # — the session survives a crash at the boundary of any agent
    # generation, not just a whole turn. "per_turn": save once, after
    # `arun()`/`astream()` (the whole run, every generation) fully
    # completes — trades that finer crash-resilience granularity for one
    # write per turn instead of one per commit (a multi-stage pipeline
    # easily produces 5-10 commits per turn, each a full Context
    # serialization through `session.save()`). Not read by `arun_once()`
    # on its own (a single generation has no well-defined "turn"
    # boundary) — only `arun()`/`astream()`'s own completion triggers the
    # deferred save.
    self.session_save_policy = session_save_policy
    # §69 "make illegal states visible" default: an agent's exception still
    # propagates out of arun()/astream() unless isolate_errors=True — opt in
    # to resilience explicitly rather than silently swallowing bugs.
    self.isolate_errors = isolate_errors
    self.on_agent_error = on_agent_error
    self.tracer: Tracer | CompositeTracer | None = (
        tracer
        if isinstance(tracer, Tracer) or tracer is None
        else CompositeTracer(tracer)
    )
    # Tracing is fully delegated to RunTracer (§54): span/trace building,
    # the RecordingLLM wrap, and the task→agent attribution it needs all
    # live there — Runtime just calls into it at a few points below.
    self._trace = RunTracer(context, self.tracer)
    self.outcome: RunOutcome = RunOutcome.COMPLETED
    self.last_stats: RunStats | None = None
    self._runs_used = 0
    self._deadline: float | None = None
    self._active_budget: Budget | None = None
    self._turn_started = False
    self._turn_started_at = 0.0
    self._no_runs_warned = False
    self._errors_used = 0
    # Not reentrant: `arun`/`arun_once`/`astream` all mutate shared,
    # instance-level turn state (`_runs_used`, `outcome`, `_deadline`,
    # and `context.resources.budget`/`budget_deadline` — a resource
    # *shared* by the Context). Two concurrent calls on the *same*
    # Runtime (e.g. `asyncio.gather(runtime.arun(), runtime.arun())`)
    # would race on that state — one call's budget/deadline silently
    # clobbers the other's mid-flight. Guarded in `_enter_turn`/
    # `_exit_turn`, checked only at the public entry points; `arun`'s own
    # internal loop calls `_arun_once_impl` directly (unguarded — it is
    # already inside the guarded region, not a second concurrent call).
    self._in_turn = False

astream async

astream(
    budget: Budget | None = None,
    max_iterations: int = 1000,
    *,
    request: Mapping[str, Any] | None = None,
) -> AsyncIterator[ProgressEvent]

Stream of a run: run_start → status (agent announces) → run_end.

Agents publish statuses via context.announce(...); the app re-renders them in the chat ("Thinking…", "Searching docs…", "Found N…"). At the end a run_end with a summary (outcome/runs/duration) is emitted.

Source code in reactifact/runtime.py

async def astream(
    self,
    budget: Budget | None = None,
    max_iterations: int = 1000,
    *,
    request: Mapping[str, Any] | None = None,
) -> AsyncIterator[ProgressEvent]:
    """Stream of a run: run_start → status (agent announces) → run_end.

    Agents publish statuses via `context.announce(...)`; the app
    re-renders them in the chat ("Thinking…", "Searching docs…", "Found N…").
    At the end a run_end with a summary (outcome/runs/duration) is emitted.
    """
    queue = self.context.subscribe()
    done = asyncio.Event()

    async def _runner() -> None:
        try:
            # request is installed on this task's context; the generation's
            # child tasks inherit it, so `call.request` works there too.
            await self.arun(
                max_iterations=max_iterations, budget=budget, request=request
            )
        finally:
            done.set()

    task = asyncio.create_task(_runner())
    try:
        yield ProgressEvent(kind="run_start", message="Processing started")
        while True:
            if done.is_set() and queue.empty():
                break
            get_event = asyncio.ensure_future(queue.get())
            wait_done = asyncio.ensure_future(done.wait())
            finished, _ = await asyncio.wait(
                {get_event, wait_done}, return_when=asyncio.FIRST_COMPLETED
            )
            if get_event in finished:
                yield get_event.result()
            else:
                get_event.cancel()
        # Re-raise any agent/runtime exception instead of silently dropping it:
        # an error inside a run must reach the caller, not hide in the task.
        await task
        stats = self.last_stats
        yield ProgressEvent(
            kind="run_end",
            message="Processing finished",
            data={
                "outcome": stats.outcome.value if stats is not None else None,
                "runs": stats.runs if stats is not None else 0,
                "duration": stats.duration if stats is not None else 0.0,
            },
        )
    finally:
        task.cancel()
        self.context.unsubscribe(queue)

RuntimeResources

RuntimeResources

RuntimeResources(
    llm: LLMProvider | None = None,
    embedder: EmbeddingProvider | None = None,
    sources: dict[str, Source] | None = None,
    context_builder: ContextBuilder | None = None,
    verification_threshold: float | None = None,
    redactor: Redactor | None = None,
    id_factory: IdFactory | None = None,
    **additional: Any,
)

Source code in reactifact/resources.py

def __init__(
    self,
    llm: LLMProvider | None = None,
    embedder: EmbeddingProvider | None = None,
    sources: dict[str, Source] | None = None,
    context_builder: ContextBuilder | None = None,
    verification_threshold: float | None = None,
    redactor: Redactor | None = None,
    id_factory: IdFactory | None = None,
    **additional: Any,
):
    self.llm = llm
    self.embedder = embedder
    self.sources = sources or {}
    # Injected id source for artifacts created without an explicit id
    # (`None` = the uuid default). A deterministic factory
    # (`reactifact.replay.counter_ids`) makes an unmodified app's
    # `context_hash` reproducible run to run — see `reactifact.replay.verify_run`.
    self.id_factory = id_factory
    # Applied to trace text only (artifact `data`, LLM messages/responses,
    # errors) before it reaches a sink — never to the live `Context` or a
    # persisted session. `None` (default) reproduces the pre-hook behavior.
    # See `reactifact.redaction`.
    self.redactor = redactor
    # Framework-wide pass/fail cutoff for `Verify` (verify.py): `None`
    # means "use Verify's own DEFAULT_THRESHOLD". A `Verify` instance's
    # own explicit `threshold=` still overrides this per agent.
    self.verification_threshold = verification_threshold
    # Runtime-level policy for what actually goes into an agent's inputs
    # (ranking/truncation) — `None` reproduces the old, unranked
    # behavior. See `context_builder.py`; applied inside
    # `Agent._collect_inputs` (`agents.py`), so both the runtime's
    # provenance (`Runtime._collect_reads`) and the agent's actual
    # produce inputs go through the same builder call and stay in sync.
    self.context_builder = context_builder
    # Typed resources: app collaborators (a knowledge store, decision
    # tools, a settings object) registered against their type or a
    # `ResourceKey`, so a produce reads `resources.require(Store)` and gets
    # `Store` — no `or None`, no duck-typing, no error surfacing three calls
    # later. `additional` stays as the string-keyed escape hatch.
    self._typed: dict[Any, Any] = {}
    self.additional = additional
    # Set by Runtime per turn (not a constructor param — the runtime, not
    # the caller, owns these): the active Budget and its wall-clock
    # deadline, read back by ToolUse's own inner loop (§ tool_use.py) to
    # enforce the tool-call/time budget between its own round-trips, not
    # just at the top-level Runtime._budget_exhausted check.
    self.budget: Budget | None = None
    self.budget_deadline: float | None = None

registered property

registered: frozenset[Any]

The keys of every registered typed resource (for diagnostics).

register

register(key: type[T] | ResourceKey[T], instance: T) -> T

Attaches instance under key (its type, or a ResourceKey).

Returns the instance, so store = resources.register(Store, Store(...)) reads as a one-liner. Re-registering a key overwrites it.

Source code in reactifact/resources.py

def register(self, key: type[T] | ResourceKey[T], instance: T) -> T:
    """Attaches `instance` under `key` (its type, or a `ResourceKey`).

    Returns the instance, so `store = resources.register(Store, Store(...))`
    reads as a one-liner. Re-registering a key overwrites it.
    """
    self._typed[key] = instance
    return instance

get

get(key: str) -> Any

get(key: type[T]) -> T | None

get(key: ResourceKey[T]) -> T | None

get(key: str | type[T] | ResourceKey[T]) -> Any

A string key reads additional; a type/ResourceKey reads typed.

Source code in reactifact/resources.py

def get(self, key: str | type[T] | ResourceKey[T]) -> Any:
    """A string key reads `additional`; a type/`ResourceKey` reads typed."""
    if isinstance(key, str):
        return self.additional.get(key)
    return self._typed.get(key)

require

require(key: type[T] | ResourceKey[T]) -> T

Like get, but a missing resource is a loud, early LookupError.

Reach for this inside a produce/agent for a resource the app must have configured — the failure names the missing type instead of being a None that blows up later.

Source code in reactifact/resources.py

def require(self, key: type[T] | ResourceKey[T]) -> T:
    """Like `get`, but a missing resource is a loud, early `LookupError`.

    Reach for this inside a produce/agent for a resource the app *must*
    have configured — the failure names the missing type instead of being a
    `None` that blows up later.
    """
    value = self._typed.get(key)
    if value is None:
        raise LookupError(
            f"resource {key!r} is not registered on RuntimeResources; "
            f"call resources.register({key!r}, ...) when building them"
        )
    return cast("T", value)

has

has(key: type[T] | ResourceKey[T]) -> bool

Whether a typed resource is registered (is_configured, explicitly).

Source code in reactifact/resources.py

def has(self, key: type[T] | ResourceKey[T]) -> bool:
    """Whether a typed resource is registered (`is_configured`, explicitly)."""
    return key in self._typed

typed_values

typed_values() -> list[Any]

Every registered typed resource value (read-only introspection).

registered gives the keys; this gives the values — e.g. so a generic tool scanner (reactifact.testing.fault) can find a list[Tool] registered with register(...), not just one stashed via set(...).

Source code in reactifact/resources.py

def typed_values(self) -> list[Any]:
    """Every registered typed resource value (read-only introspection).

    `registered` gives the keys; this gives the values — e.g. so a generic
    tool scanner (`reactifact.testing.fault`) can find a `list[Tool]`
    registered with `register(...)`, not just one stashed via `set(...)`.
    """
    return list(self._typed.values())

aclose async

aclose() -> None

Closes the llm/embedder clients if they support it.

Duck-typed: LLMProvider/EmbeddingProvider don't require aclose (a fake/no-op test double doesn't need one), so it's called only when present. Nothing in the runtime calls this automatically — resources are typically shared across many turns/runtimes, and closing them early would break whatever still holds a reference. Call it yourself once, at real shutdown: a FastAPI lifespan, or the end of a script. ChatAssistant is the one exception — see its docstring.

Source code in reactifact/resources.py

async def aclose(self) -> None:
    """Closes the llm/embedder clients if they support it.

    Duck-typed: `LLMProvider`/`EmbeddingProvider` don't require `aclose`
    (a fake/no-op test double doesn't need one), so it's called only when
    present. Nothing in the runtime calls this automatically — resources
    are typically shared across many turns/runtimes, and closing them
    early would break whatever still holds a reference. Call it yourself
    once, at real shutdown: a FastAPI `lifespan`, or the end of a script.
    `ChatAssistant` is the one exception — see its docstring.
    """
    for provider in (self.llm, self.embedder):
        aclose = getattr(provider, "aclose", None)
        if aclose is not None:
            await aclose()

scope classmethod

scope(
    factory: Callable[
        [], RuntimeResources | Awaitable[RuntimeResources]
    ],
) -> ResourceScope

async with RuntimeResources.scope(build) as resources: — see ResourceScope.

Source code in reactifact/resources.py

@classmethod
def scope(
    cls, factory: Callable[[], RuntimeResources | Awaitable[RuntimeResources]]
) -> ResourceScope:
    """`async with RuntimeResources.scope(build) as resources:` — see `ResourceScope`."""
    return ResourceScope(factory)

Source

Source

Source(source_id: str)

Bases: ABC

Source code in reactifact/sources.py

def __init__(self, source_id: str):
    self.source_id = source_id
    # Sorting hint for aggregating search: preferred sources
    # (e.g., vector RAG) are polled first, the rest fill in.
    self.preferred: bool = False

search

search(query: str, limit: int = 10) -> list[SourceRef]

Finds references to relevant content (by default — cannot do it).

The agent must not know the search mechanics: vector/keywords/SQL/CQL — that is the source's choice (§8). Sources without search simply return an empty list and stay available via resolve.

Source code in reactifact/sources.py

def search(self, query: str, limit: int = 10) -> list[SourceRef]:
    """Finds references to relevant content (by default — cannot do it).

    The agent must not know the search mechanics: vector/keywords/SQL/CQL
    — that is the source's choice (§8). Sources without search simply
    return an empty list and stay available via resolve.
    """
    return []

asearch async

asearch(query: str, limit: int = 10) -> list[SourceRef]

Asynchronous search (embeddings, API). Default — synchronous search.

Vector sources cannot compute the query embedding synchronously, so the aggregator (ScoutSources) uses asearch.

Source code in reactifact/sources.py

async def asearch(self, query: str, limit: int = 10) -> list[SourceRef]:
    """Asynchronous search (embeddings, API). Default — synchronous `search`.

    Vector sources cannot compute the query embedding synchronously,
    so the aggregator (`ScoutSources`) uses `asearch`.
    """
    return self.search(query, limit)

resolve abstractmethod async

resolve(ref: SourceRef) -> Any

Resolves a reference into materialized data (e.g., text, structure).

Source code in reactifact/sources.py

@abstractmethod
async def resolve(self, ref: SourceRef) -> Any:
    """Resolves a reference into materialized data (e.g., text, structure)."""
    ...

PendingQuestion (HITL)

PendingQuestion

Bases: BaseModel

Artifact awaiting a human response (HITL, constitution §60).

Created by an agent (or directly) to block a step until user input. A human answer is recorded via self.effects.resume(question, answer) (§60), after which agents subscribed to PendingQuestion(answered=True) continue.

In the box

Tool

Tool

Bases: ABC

Contract for an external operation: "Do this operation" (§46).

For an LLM agent (LLMAgent) a tool must provide a JSON schema of its arguments (schema), from which the model picks the operation and arguments. @tool builds it from the signature automatically.

execute abstractmethod async

execute(args: dict[str, Any]) -> ToolOutput

Execute the operation; failure is an exception or ToolOutput(error=...).

Source code in reactifact/tools.py

@abstractmethod
async def execute(self, args: dict[str, Any]) -> ToolOutput:
    """Execute the operation; failure is an exception or `ToolOutput(error=...)`."""
    ...

ToolUse

ToolUse

ToolUse(
    system: str,
    tools: Sequence[Tool] | dict[str, Tool],
    *,
    name: str = "llm",
    max_steps: int = 8,
    temperature: float | None = None,
    max_tokens: int | None = None,
    deferred_tool_groups: Sequence[DeferredToolGroup] = (),
)

Bases: _ToolLoopBase

Blocking loop "LLM decides → tool → … → answer" in a single produce.

Simple, no HITL: the LLM either calls a tool or answers. The logic lives here, not in the container agent. Destructive tools are not offered to the LLM.

Source code in reactifact/tool_use.py

def __init__(
    self,
    system: str,
    tools: Sequence[Tool] | dict[str, Tool],
    *,
    name: str = "llm",
    max_steps: int = 8,
    temperature: float | None = None,
    max_tokens: int | None = None,
    deferred_tool_groups: Sequence[DeferredToolGroup] = (),
):
    super().__init__(
        system, tools, name=name, temperature=temperature, max_tokens=max_tokens
    )
    self.max_steps = max_steps
    self.deferred_tool_groups = list(deferred_tool_groups)

ToolUseHITL

ToolUseHITL

ToolUseHITL(
    system: str,
    tools: Sequence[Tool] | dict[str, Tool],
    *,
    name: str = "llm",
    max_steps: int = 8,
    max_asks: int = 2,
    max_approvals: int = 3,
    resume_announce: Callable[[str], str] | None = None,
    temperature: float | None = None,
    max_tokens: int | None = None,
)

Bases: _ToolLoopBase

Reactive loop: step by step, can ask the human (HITL, §60).

The LLM may answer (answer), call a tool (tool_call, result goes into an Observation), or ask a clarifying question (ask → PendingQuestion). The human answer comes back into the loop as Observation(source="user").

Destructive tools are offered to the LLM (unlike ToolUse, which excludes them entirely) but never executed straight away: a tool_call targeting a destructive tool creates a PendingQuestion(kind="approve") instead, and the call only runs once a human resolves it with an affirmative answer (max_approvals bounds how many times one conversation can ask).

_history/the ask dedup check filter context.list_artifacts(Observation | PendingQuestion) by query_id in Python — O(count of that type in the whole context), not indexed by query_id. Bounded per conversation (max_steps, max_asks), so this is only a real cost if many long-running conversations share one Context — the standard one-Context-per-session pattern (SessionStore, every example in this repo) keeps each conversation's own artifact count small regardless of how many sessions exist. A real fix would need RelationGraph indexed by source_id (it currently isn't either) plus linking each Observation to its goal artifact instead of filtering by field — deliberately not done here; flag it if you're sharing one long-lived Context across many concurrent tool-use conversations.

Source code in reactifact/tool_use.py

def __init__(
    self,
    system: str,
    tools: Sequence[Tool] | dict[str, Tool],
    *,
    name: str = "llm",
    max_steps: int = 8,
    max_asks: int = 2,
    max_approvals: int = 3,
    resume_announce: Callable[[str], str] | None = None,
    temperature: float | None = None,
    max_tokens: int | None = None,
):
    super().__init__(
        system, tools, name=name, temperature=temperature, max_tokens=max_tokens
    )
    self.max_steps = max_steps
    self.max_asks = max_asks
    self.max_approvals = max_approvals
    # App callback: human answer → status message (kind="status").
    self.resume_announce = resume_announce

native_tool_use

tools_payload

tools_payload(
    tools: Sequence[Tool] | dict[str, Tool],
) -> list[dict[str, Any]]

Builds the OpenAI tools=[...] array from Tools — Tool.schema is already the right JSON-schema shape (tools.py), this just wraps it.

Source code in reactifact/native_tool_use.py

def tools_payload(tools: Sequence[Tool] | dict[str, Tool]) -> list[dict[str, Any]]:
    """Builds the OpenAI `tools=[...]` array from `Tool`s — `Tool.schema` is
    already the right JSON-schema shape (`tools.py`), this just wraps it."""
    values = tools.values() if isinstance(tools, dict) else tools
    return [
        {
            "type": "function",
            "function": {
                "name": t.name,
                "description": t.description,
                "parameters": t.schema,
            },
        }
        for t in values
    ]

native_complete async

native_complete(
    context: Context,
    *,
    messages: list[dict[str, Any]],
    tools: Sequence[Tool] | dict[str, Tool] = (),
    temperature: float | None = None,
    max_tokens: int | None = None,
) -> LLMResponse | None

One native tool-calling turn.

messages are raw OpenAI-format dicts (system/user/assistant-with- tool_calls/tool-with-tool_call_id) — not providers.Message, which has no field for either. The full, verbatim dicts go via LLMRequest.extra["messages"], which every OpenAICompatProvider-family provider applies after building its own messages list from LLMRequest.messages (providers/chat.py's _payload — request.extra is merged into the payload last, overriding anything built from typed fields). That's the documented escape hatch for provider-specific wire shapes, not a hack — see _payload's own comment on arbitrary extra fields.

LLMRequest.messages is also populated, with a best-effort typed rendering (role + content, tool_calls/tool_call_id dropped — Message has no field for either). OpenAICompatProvider ignores it in favor of the full-fidelity extra["messages"], but a LLMProvider that doesn't know this module's extra convention still sees a real conversation instead of an empty list.

Returns None if no provider is configured (§67 — the same honest no-provider fallback structured_llm uses) or the provider call itself raised (network/outage) — logged, not raised further. Bring your own history/retry/approval logic around this — see the module docstring for why that isn't this function's job.

Source code in reactifact/native_tool_use.py

async def native_complete(
    context: Context,
    *,
    messages: list[dict[str, Any]],
    tools: Sequence[Tool] | dict[str, Tool] = (),
    temperature: float | None = None,
    max_tokens: int | None = None,
) -> LLMResponse | None:
    """One native tool-calling turn.

    `messages` are raw OpenAI-format dicts (system/user/assistant-with-
    `tool_calls`/tool-with-`tool_call_id`) — not `providers.Message`, which
    has no field for either. The full, verbatim dicts go via
    `LLMRequest.extra["messages"]`, which every `OpenAICompatProvider`-family
    provider applies *after* building its own messages list from
    `LLMRequest.messages` (`providers/chat.py`'s `_payload` — `request.extra`
    is merged into the payload last, overriding anything built from typed
    fields). That's the documented escape hatch for provider-specific wire
    shapes, not a hack — see `_payload`'s own comment on arbitrary `extra`
    fields.

    `LLMRequest.messages` is *also* populated, with a best-effort typed
    rendering (role + content, `tool_calls`/`tool_call_id` dropped — `Message`
    has no field for either). `OpenAICompatProvider` ignores it in favor of
    the full-fidelity `extra["messages"]`, but a `LLMProvider` that doesn't
    know this module's `extra` convention still sees a real conversation
    instead of an empty list.

    Returns `None` if no provider is configured (§67 — the same honest
    no-provider fallback `structured_llm` uses) or the provider call itself
    raised (network/outage) — logged, not raised further. Bring your own
    history/retry/approval logic around this — see the module docstring
    for why that isn't this function's job.
    """
    llm = context.resources.llm
    if llm is None:
        return None
    extra: dict[str, Any] = {"messages": messages}
    payload_tools = tools_payload(tools)
    if payload_tools:
        extra["tools"] = payload_tools
        extra["tool_choice"] = "auto"
    typed_messages = [
        Message(
            role=cast(Role, m.get("role", "user")), content=str(m.get("content") or "")
        )
        for m in messages
    ]
    request = LLMRequest(
        messages=typed_messages,
        temperature=temperature,
        max_tokens=max_tokens,
        extra=extra,
    )
    try:
        return await llm.complete(request)
    except Exception as exc:
        logger.warning("native_tool_use: provider call failed: %r", exc)
        return None

parse_tool_calls

parse_tool_calls(response: LLMResponse) -> list[ToolCall]

Extracts message.tool_calls from the provider's raw response.

LLMResponse.raw is the full, unparsed JSON body every OpenAICompatProvider-family provider already returns (providers/chat.py) — this is deliberately the only place that reaches into it, so a provider shaping raw differently only needs to override this one function, not anything that calls it. Malformed arguments JSON parses to {} rather than raising (§67 — the model's mistake is not a crash), same tolerance structured_llm.parse_structured uses.

Source code in reactifact/native_tool_use.py

def parse_tool_calls(response: LLMResponse) -> list[ToolCall]:
    """Extracts `message.tool_calls` from the provider's raw response.

    `LLMResponse.raw` is the full, unparsed JSON body every
    `OpenAICompatProvider`-family provider already returns
    (`providers/chat.py`) — this is deliberately the only place that reaches
    into it, so a provider shaping `raw` differently only needs to override
    this one function, not anything that calls it. Malformed `arguments`
    JSON parses to `{}` rather than raising (§67 — the model's mistake is
    not a crash), same tolerance `structured_llm.parse_structured` uses.
    """
    if not isinstance(response.raw, dict):
        return []
    choices = response.raw.get("choices") or []
    if not choices:
        return []
    message = choices[0].get("message") or {}
    calls = message.get("tool_calls") or []
    parsed: list[ToolCall] = []
    for call in calls:
        function = call.get("function") or {}
        raw_args = function.get("arguments") or "{}"
        try:
            args = json.loads(raw_args)
        except ValueError:
            logger.debug(
                "native_tool_use: malformed tool_call arguments: %.160r", raw_args
            )
            args = {}
        parsed.append(
            ToolCall(id=call.get("id", ""), name=function.get("name", ""), args=args)
        )
    return parsed

AgentAsTool

AgentAsTool

AgentAsTool(
    *,
    name: str,
    description: str,
    agent_factory: Callable[[], Agent],
    resources: RuntimeResources,
    input_type: type[BaseModel] = SubTask,
    output_type: type[BaseModel] | None = None,
    max_runs: int = 20,
    destructive: bool = False,
)

Bases: Tool

Wraps agent_factory() as a callable tool (see module docstring).

Source code in reactifact/agent_tool.py

def __init__(
    self,
    *,
    name: str,
    description: str,
    agent_factory: Callable[[], Agent],
    resources: RuntimeResources,
    input_type: type[BaseModel] = SubTask,
    output_type: type[BaseModel] | None = None,
    max_runs: int = 20,
    destructive: bool = False,
):
    self.name = name
    self.description = description
    self.agent_factory = agent_factory
    self.resources = resources
    self.input_type = input_type
    self.output_type = output_type or ToolAnswer
    self.max_runs = max_runs
    self.destructive = destructive
    self.schema = {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "The task/question to delegate to the sub-agent.",
            }
        },
        "required": ["query"],
    }

ContextBuilder

ContextBuilder

Bases: ABC

Ranks/truncates the artifacts a single agent run would otherwise see.

Called once per agent run with every candidate matched by that agent's consumes (combined across all of them). Returning the list unchanged reproduces today's behavior.

TokenBudgetContextBuilder

TokenBudgetContextBuilder

TokenBudgetContextBuilder(
    *,
    max_tokens: int | None = None,
    token_counter: TokenCounter | None = None,
    rank_key: Callable[[Artifact[Any]], Any] | None = None,
    render: Callable[[Artifact[Any]], str] | None = None,
    exempt_types: tuple[type, ...] = (),
    min_keep: dict[type, int] | None = None,
)

Bases: ContextBuilder

Ranks candidates (newest-first by default) and keeps a prefix that fits max_tokens, estimated via token_counter.

At least one costed (non-exempt) artifact is always kept, even if it alone exceeds the budget — an agent silently getting zero real content is worse than one that gets an oversized item; max_tokens is a soft cap, not a hard clip.

exempt_types: artifact types that cost nothing and never trigger that clip — for a marker/trigger type an agent only consumes to wake up (little or no meaningful text), not to reason over. Without this, a freshly created marker can rank first (newest) and either eat the whole budget itself or — worse — silently consume the "at least one" slot, leaving zero real content if the budget is then too tight for the next (real) candidate. Exempt artifacts are still ranked and still kept (still wake the agent up); they just never count toward max_tokens or toward what counts as "at least one" content item.

min_keep: {type: count} — for a type with real content that must still survive the budget regardless of rank (unlike exempt_types, it does count toward max_tokens), guarantees its top-count-ranked instances a reserved slice of the budget, filled before the shared greedy pass runs over everything else. Without this, a low-volume type that happens to rank lower than a high-volume one sharing the same budget (e.g. the single triggering Question, if older than a pile of freshly-ranked Evidence) can be crowded out entirely — min_keep reserves its slot first instead of leaving it to rank order. The final list is still returned in overall rank order, not with reserved items forced to the front.

Source code in reactifact/context_builder.py

def __init__(
    self,
    *,
    max_tokens: int | None = None,
    token_counter: TokenCounter | None = None,
    rank_key: Callable[[Artifact[Any]], Any] | None = None,
    render: Callable[[Artifact[Any]], str] | None = None,
    exempt_types: tuple[type, ...] = (),
    min_keep: dict[type, int] | None = None,
):
    self.max_tokens = max_tokens
    self.token_counter = token_counter or HeuristicTokenCounter()
    self.rank_key = rank_key or (lambda a: a.updated_at)
    self.render = render or _default_render
    self.exempt_types = exempt_types
    self.min_keep = dict(min_keep or {})

Verify

Verify

Verify(
    *,
    metrics: Mapping[str, MetricFn] = core_metrics,
    threshold: float | None = None,
    required_metrics: tuple[str, ...] = (),
    on_fail: Literal["ask", "retry"] = "ask",
    answer_type: str = "Answer",
)

Bases: Produce[VerificationResult]

Scores a live Answer-shaped artifact with eval.py metrics.

threshold: explicit per-instance cutoff. None (default) resolves at produce-time from context.resources.verification_threshold, itself falling back to DEFAULT_THRESHOLD when that is also unset — so setting it once on RuntimeResources changes the bar for every Verify in the Runtime, while a specific agent can still opt out with its own value.

required_metrics: names that must score exactly 1.0 regardless of the weighted average — e.g. ("provenance_grounded",) so an ungrounded answer never passes just because other metrics compensate.

answer_type: the artifact class name (by name, no domain import — same convention as eval.py) this instance verifies; artifacts of any other type are ignored.

On top of VerificationResult, this creates a PendingQuestion (on_fail="ask") or a VerificationFailed (on_fail="retry") — the containing agent's produces must declare those too (a bare Produce(PendingQuestion)/Produce(VerificationFailed) widens the allowed Create types without adding real logic — same convention HITLLMAgent uses for Observation/PendingQuestion, llm_agent.py):

class Answerer(Agent):
    consumes = [Consume(Question)]
    produces = [BuildAnswer(), Verify(), Produce(PendingQuestion)]

Source code in reactifact/verify.py

def __init__(
    self,
    *,
    metrics: Mapping[str, MetricFn] = core_metrics,
    threshold: float | None = None,
    required_metrics: tuple[str, ...] = (),
    on_fail: Literal["ask", "retry"] = "ask",
    answer_type: str = "Answer",
):
    self.metrics = metrics
    self.threshold = threshold
    self.required_metrics = required_metrics
    self.on_fail = on_fail
    self.answer_type = answer_type
    super().__init__()

Audit & reproducibility

build_report

build_report

build_report(
    context: Context,
    answer: Artifact[Any] | str,
    *,
    relations: Sequence[str] | None = None,
    session_id: str = "",
) -> AuditReport

Walks the provenance chain behind answer into a verifiable report.

Breadth-first over context.relations(), following relations (all by default), so every artifact that contributed to the answer is included with its content hash and producing author. SourceRef locators in the chain become sources.

Source code in reactifact/audit.py

def build_report(
    context: Context,
    answer: Artifact[Any] | str,
    *,
    relations: Sequence[str] | None = None,
    session_id: str = "",
) -> AuditReport:
    """Walks the provenance chain behind `answer` into a verifiable report.

    Breadth-first over `context.relations()`, following `relations` (all by
    default), so every artifact that contributed to the answer is included with
    its content hash and producing author. `SourceRef` locators in the chain
    become `sources`.
    """
    followed = set(relations) if relations is not None else set(DEFAULT_RELATIONS)
    answer_artifact = _resolve_answer(context, answer)
    producers = _producers(context)

    entries: list[ProvenanceEntry] = []
    edges: list[tuple[str, str, str]] = []
    sources: list[str] = []
    visited: set[str] = set()
    queue = [answer_artifact]
    while queue:
        artifact = queue.pop(0)
        if artifact.id in visited:
            continue
        visited.add(artifact.id)
        entries.append(_entry(artifact, producers.get(artifact.id, "")))
        locator = getattr(artifact.data, "locator", None)
        if isinstance(locator, str) and locator and locator not in sources:
            sources.append(locator)
        for rel in context.relations(source_id=artifact.id):
            if rel.relation not in followed:
                continue
            edges.append((rel.source_id, rel.relation, rel.target_id))
            target = context.get(rel.target_id)
            if target is not None and target.id not in visited:
                queue.append(target)

    return AuditReport(
        session_id=session_id,
        context_version=context.version,
        context_sha256=context_hash(context),
        answer=entries[0],
        provenance=entries[1:],
        relations=edges,
        sources=sources,
    )

context_hash

context_hash

context_hash(context: Context) -> str

sha256 over the run's canonical state: version, artifacts, relations.

Includes each artifact's id, type, version and content hash, plus every relation edge — everything that makes two runs' states the same or different. Timestamps are deliberately excluded, so a re-run that reaches the same state hashes identically.

Source code in reactifact/audit.py

def context_hash(context: Context) -> str:
    """sha256 over the run's canonical state: version, artifacts, relations.

    Includes each artifact's id, type, version and content hash, plus every
    relation edge — everything that makes two runs' states the same or
    different. Timestamps are deliberately excluded, so a re-run that reaches
    the same state hashes identically.
    """
    artifacts = [
        {
            "id": artifact.id,
            "type": type(artifact.data).__name__,
            "version": artifact.version,
            "sha256": artifact_hash(artifact),
        }
        for artifact in sorted(context.list_artifacts(), key=lambda a: a.id)
    ]
    relations = sorted(
        (rel.source_id, rel.relation, rel.target_id) for rel in context.relations()
    )
    payload = {
        "version": context.version,
        "artifacts": artifacts,
        "relations": relations,
    }
    return hashlib.sha256(_canonical(payload).encode("utf-8")).hexdigest()

AuditReport

AuditReport

Bases: BaseModel

A reproducible record of an answer and the evidence behind it.

Redaction

RegexRedactor

RegexRedactor

RegexRedactor(
    patterns: Sequence[tuple[str, str]] | None = None,
    *,
    replacement: str = "[REDACTED:{name}]",
)

Pattern-based Redactor with conservative, ready-to-use defaults.

patterns is a sequence of (name, regex) pairs; None uses DEFAULT_PATTERNS. Everything that matches is replaced with replacement (default "[REDACTED:<name>]", so a reader can tell what was removed without seeing it).

Source code in reactifact/redaction.py

def __init__(
    self,
    patterns: Sequence[tuple[str, str]] | None = None,
    *,
    replacement: str = "[REDACTED:{name}]",
):
    self._patterns = [
        (name, re.compile(pattern))
        for name, pattern in (patterns or DEFAULT_PATTERNS)
    ]
    self._replacement = replacement

Tool loop

run_tool_loop

run_tool_loop async

run_tool_loop(
    context: Context,
    *,
    system: str,
    user: str,
    tools: Sequence[Tool] | dict[str, Tool],
    max_rounds: int = 6,
    parallel: bool = True,
    mandatory: str | Sequence[str] | None = None,
    temperature: float | None = None,
    max_tokens: int | None = None,
) -> ToolLoopResult

Runs native tool-calling for up to max_rounds, then forces an answer.

• A turn that returns no tool calls ends the loop with its text.
• Tool calls are executed (in parallel by default) and their results fed back as tool messages.
• mandatory (a tool name or names that must run): while none has been called and rounds remain, a reminder is injected so the model is nudged rather than the loop silently finishing without it.
• If max_rounds is reached without a final answer, one more call is made with no tools, to force a plain-text answer.
• No provider / a failed call yields text="" with whatever was observed — the same honest-failure contract the rest of the framework uses.

Source code in reactifact/recipes/tool_loop.py

async def run_tool_loop(
    context: Context,
    *,
    system: str,
    user: str,
    tools: Sequence[Tool] | dict[str, Tool],
    max_rounds: int = 6,
    parallel: bool = True,
    mandatory: str | Sequence[str] | None = None,
    temperature: float | None = None,
    max_tokens: int | None = None,
) -> ToolLoopResult:
    """Runs native tool-calling for up to `max_rounds`, then forces an answer.

    - A turn that returns no tool calls ends the loop with its text.
    - Tool calls are executed (in parallel by default) and their results fed
      back as `tool` messages.
    - `mandatory` (a tool name or names that *must* run): while none has been
      called and rounds remain, a reminder is injected so the model is nudged
      rather than the loop silently finishing without it.
    - If `max_rounds` is reached without a final answer, one more call is made
      with no tools, to force a plain-text answer.
    - No provider / a failed call yields `text=""` with whatever was observed
      — the same honest-failure contract the rest of the framework uses.
    """
    tool_map = dict(tools) if isinstance(tools, dict) else {t.name: t for t in tools}
    required = {mandatory} if isinstance(mandatory, str) else set(mandatory or ())
    messages: list[dict[str, Any]] = [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]
    observations: list[ToolObservation] = []
    called: set[str] = set()
    rounds = 0

    for _ in range(max(0, max_rounds)):
        response = await native_complete(
            context,
            messages=messages,
            tools=tool_map,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        if response is None:
            return ToolLoopResult("", messages, observations, rounds)
        rounds += 1
        calls = parse_tool_calls(response)
        if not calls:
            if required - called:
                # mandatory tool not used yet — nudge and keep going
                messages.append({"role": "assistant", "content": response.text})
                missing = ", ".join(sorted(required - called))
                messages.append(
                    {
                        "role": "user",
                        "content": f"You must call {missing} before answering.",
                    }
                )
                continue
            return ToolLoopResult(response.text, messages, observations, rounds)

        messages.append(
            {
                "role": "assistant",
                "content": response.text or "",
                "tool_calls": [_as_message(c) for c in calls],
            }
        )
        observations.extend(await _execute(tool_map, calls, parallel=parallel))
        called.update(c.name for c in calls)
        for observation in observations[-len(calls) :]:
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": observation.call.id,
                    "content": _content(observation),
                }
            )

    # out of rounds — force a plain answer without tools
    response = await native_complete(
        context, messages=messages, temperature=temperature, max_tokens=max_tokens
    )
    text = response.text if response is not None else ""
    return ToolLoopResult(text, messages, observations, rounds)

Golden runs

capture

capture

capture(
    context: Context, *, trace: RunTrace | None = None
) -> GoldenRun

Freezes context (and optionally a trace) into a GoldenRun.

Source code in reactifact/testing/golden.py

def capture(context: Context, *, trace: RunTrace | None = None) -> GoldenRun:
    """Freezes `context` (and optionally a trace) into a `GoldenRun`."""
    hashes = tuple(prompt_hashes(trace)) if trace is not None else ()
    return GoldenRun(context_sha256=context_hash(context), prompt_hashes=hashes)

assert_golden

assert_golden

assert_golden(
    context: Context,
    golden: GoldenRun,
    *,
    trace: RunTrace | None = None,
) -> None

Fails loudly when state (or, with trace, prompts) drifted.

Source code in reactifact/testing/golden.py

def assert_golden(
    context: Context,
    golden: GoldenRun,
    *,
    trace: RunTrace | None = None,
) -> None:
    """Fails loudly when state (or, with `trace`, prompts) drifted."""
    actual = context_hash(context)
    if actual != golden.context_sha256:
        raise AssertionFailure(
            f"context hash drifted: expected {golden.context_sha256}, got {actual}"
        )
    if golden.prompt_hashes:
        if trace is None:
            raise AssertionFailure(
                "golden has prompt hashes but no trace was passed to compare"
            )
        actual_hashes = tuple(prompt_hashes(trace))
        if actual_hashes != golden.prompt_hashes:
            raise AssertionFailure(
                "prompt hashes drifted: "
                f"expected {golden.prompt_hashes}, got {actual_hashes}"
            )