Skip to content

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/context.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def __init__(self, resources: RuntimeResources | None = None):
    self._artifacts: dict[str, Artifact[Any]] = {}
    self._store = ArtifactStore()
    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]] = {}
    # Per storage key: (version, compaction baseline) already persisted.
    # A Context can be saved under several keys (a session + named
    # branches), so this is keyed rather than a single global marker.
    self._persisted: dict[str, tuple[int, int]] = {}
    #: Per-step input cursors (watermark scheduling): step key →
    #: {artifact type: (latest id, max version)}. In-memory for now;
    #: persistence/replay of cursors is a later core milestone.
    self._step_cursors: dict[Any, dict[type, dict[str, int]]] = {}
    # Compatibility view for the single-session case (last marked key).
    self._persisted_version = -1
    self._persisted_compacted_at = 0

request property 

request: Mapping[str, Any]

The active turn's request mapping (empty when none was set).

Read-only view of reactifact.request.current_request() — the same data a produce sees as call.request. Exposed on Context so hooks that receive only a context (e.g. ChatAssistant(create_message=…)) can read per-turn metadata (category, tenant, user) without a hand-rolled ContextVar of their own.

persisted_version property 

persisted_version: int

Highest version written by the last mark_persisted call (-1 when this Context has never been persisted).

persisted_compacted_at property 

persisted_compacted_at: int

Compaction baseline written by the last mark_persisted call.

version property 

version: int

Current Context version (number of applied commits).

head_id property 

head_id: str | None

Id of the last commit (HEAD).

compacted_at property 

compacted_at: int

Highest version collapsed into the baseline (0 = never compacted).

persisted_state 

persisted_state(key: str) -> tuple[int, int]

(version, compacted_at) already written for key.

(-1, 0) when this Context has never been persisted under that key.

Source code in reactifact/context/context.py 
136
137
138
139
140
141
def persisted_state(self, key: str) -> tuple[int, int]:
    """`(version, compacted_at)` already written for `key`.

    `(-1, 0)` when this Context has never been persisted under that key.
    """
    return self._persisted.get(key, (-1, 0))

mark_persisted 

mark_persisted(
    version: int | None = None, *, key: str = ""
) -> None

Records that the Context has been persisted up to version (head when omitted) for key, including its current compaction baseline.

Source code in reactifact/context/context.py 
154
155
156
157
158
159
160
def mark_persisted(self, version: int | None = None, *, key: str = "") -> None:
    """Records that the Context has been persisted up to `version` (head
    when omitted) for `key`, including its current compaction baseline."""
    persisted = self.version if version is None else version
    self._persisted[key] = (persisted, self.compacted_at)
    self._persisted_version = persisted
    self._persisted_compacted_at = self.compacted_at

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/context.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
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/context.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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._store.create(data, artifact.id, created_at=artifact.created_at)
    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/context.py 
209
210
211
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/context.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
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)
    self._store.update(artifact_id, new_data)
    updated = self._store.current_artifact(artifact_id)
    assert updated is not None
    self._artifacts[artifact_id] = updated
    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 dependents_of(self, 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 updated

delete 

delete(artifact_id: str) -> bool

Deletes the artifact and generates ARTIFACT_DELETED.

Source code in reactifact/context/context.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
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._store.delete(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/context.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
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/context.py 
297
298
299
300
301
302
303
304
305
306
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/context.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
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(
    source_id: str, relation: str, target_id: str
) -> Relation

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

Source code in reactifact/context/context.py 
338
339
340
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(
    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/context.py 
342
343
344
345
346
347
348
349
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/context.py 
351
352
353
354
355
356
357
358
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/context.py 
360
361
362
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/context.py 
364
365
366
367
368
369
370
371
372
373
374
375
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/context.py 
377
378
379
380
381
382
383
384
385
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
    ]

drain_events 

drain_events() -> list[Event]

Drains and clears the event queue.

Source code in reactifact/context/context.py 
441
442
443
444
445
def drain_events(self) -> list[Event]:
    """Drains and clears the event queue."""
    events = self._events
    self._events = []
    return events

pending_events 

pending_events() -> list[Event]

A snapshot of the queued events, without consuming them.

Non-destructive counterpart of drain_events() (which stays for callers that explicitly want to flush the queue — tests, seeding). The runtime reads a batch with this, schedules the generation, and only calls consume_events() once the generation's patches are committed: a generation that raises leaves its triggers queued instead of silently dropping the consumers they would have woken, so a retry re-runs the work rather than settling at zero runs on an empty queue.

Source code in reactifact/context/context.py 
447
448
449
450
451
452
453
454
455
456
457
458
459
def pending_events(self) -> list[Event]:
    """A snapshot of the queued events, without consuming them.

    Non-destructive counterpart of `drain_events()` (which stays for
    callers that explicitly want to flush the queue — tests, seeding).
    The runtime reads a batch with this, schedules the generation, and
    only calls `consume_events()` once the generation's patches are
    committed: a generation that raises leaves its triggers queued
    instead of silently dropping the consumers they would have woken, so
    a retry re-runs the work rather than settling at zero runs on an
    empty queue.
    """
    return list(self._events)

consume_events 

consume_events(events: list[Event]) -> None

Removes a batch previously read with pending_events().

Matching is by identity, not by count: events appended while the generation ran — its commits' effects, staged for the next generation — are preserved, and it stays correct even if a produce drained or reordered the queue directly during the generation.

Source code in reactifact/context/context.py 
461
462
463
464
465
466
467
468
469
470
471
472
def consume_events(self, events: list[Event]) -> None:
    """Removes a batch previously read with `pending_events()`.

    Matching is by identity, not by count: events appended while the
    generation ran — its commits' effects, staged for the *next*
    generation — are preserved, and it stays correct even if a produce
    drained or reordered the queue directly during the generation.
    """
    if not events:
        return
    consumed = {id(event) for event in events}
    self._events = [event for event in self._events if id(event) not in consumed]

clone 

clone() -> Context

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

Source code in reactifact/context/context.py 
474
475
476
477
478
def clone(self) -> Context:
    """Deep copy of this context's live state. See `reactifact.branching`."""
    from ..branching.ops 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/context.py 
480
481
482
483
484
def merge_from(self, other: Context) -> None:
    """Two-way merge, no conflict detection. See `reactifact.branching`."""
    from ..branching.ops 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/context.py 
486
487
488
489
490
491
492
493
494
495
496
497
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.ops 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/context.py 
499
500
501
502
503
504
505
506
507
508
509
510
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.ops 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/context.py 
512
513
514
515
516
517
518
519
520
521
522
523
524
525
def log_commit(self, commit: Commit) -> None:
    """Applies the commit to the repository: fills in parent/version, moves head."""
    self._log.append(commit)
    self._store.bind_commit(commit)
    for write in commit.writes:
        if write.op_type == "delete":
            continue
        refreshed = self._store.current_artifact(write.artifact_id)
        if refreshed is not None:
            self._artifacts[write.artifact_id] = refreshed
    # 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)

apply_commit 

apply_commit(
    commit: Commit, *, reindex: bool = True
) -> None

Applies a commit's operations to the live state and logs it.

This is the load path for the append-only session format (snapshot + commit log): it replays operations without emitting events (a restored session must not re-trigger its own agents). reindex=False skips the final full re-index — use it when applying several commits and call rebuild_indices() once at the end.

Source code in reactifact/context/context.py 
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
def apply_commit(self, commit: Commit, *, reindex: bool = True) -> None:
    """Applies a commit's operations to the live state and logs it.

    This is the load path for the append-only session format (snapshot +
    commit log): it replays operations *without* emitting events (a
    restored session must not re-trigger its own agents). `reindex=False`
    skips the final full re-index — use it when applying several commits
    and call `rebuild_indices()` once at the end.
    """
    for op in commit.operations:
        if isinstance(op, Create):
            artifact_id = op.artifact_id or op.id
            if artifact_id is None:
                raise ValueError(
                    "cannot apply a Create without artifact_id/id to a Context"
                )
            existing = self._artifacts.get(artifact_id)
            if existing is not None:
                self._by_type.get(type(existing.data), set()).discard(artifact_id)
                self._store.update(
                    artifact_id,
                    op.data,
                    commit_id=commit.id,
                    created_at=commit.timestamp,
                )
                self._by_type.setdefault(type(op.data), set()).add(artifact_id)
            else:
                self._store.create(
                    op.data,
                    artifact_id,
                    commit_id=commit.id,
                    created_at=commit.timestamp,
                )
                self._by_type.setdefault(type(op.data), set()).add(artifact_id)
            applied = self._store.current_artifact(artifact_id)
            assert applied is not None
            self._artifacts[artifact_id] = applied
        elif isinstance(op, Update):
            artifact = self._artifacts.get(op.artifact_id)
            if artifact is None:
                continue
            old_type = type(artifact.data)
            self._store.update(
                op.artifact_id,
                op.new_data,
                commit_id=commit.id,
                created_at=commit.timestamp,
            )
            applied = self._store.current_artifact(op.artifact_id)
            assert applied is not None
            self._artifacts[op.artifact_id] = applied
            new_type = type(op.new_data)
            if new_type is not old_type:
                self._by_type.get(old_type, set()).discard(op.artifact_id)
                self._by_type.setdefault(new_type, set()).add(op.artifact_id)
        elif isinstance(op, Delete):
            artifact = self._artifacts.pop(op.artifact_id, None)
            if artifact is not None:
                self._store.delete(op.artifact_id)
                self._by_type.get(type(artifact.data), set()).discard(
                    op.artifact_id
                )
        elif isinstance(op, Link):
            self.link(op.artifact_id, op.relation, op.target_id)
        elif isinstance(op, Unlink):
            self.unlink(op.artifact_id, op.relation, op.target_id)
        else:
            raise ValueError(f"Unknown operation: {op}")
    self.log_commit(commit)
    if reindex:
        self.rebuild_indices()

rebuild_indices 

rebuild_indices() -> None

Rebuilds the type/staleness indices from the live state.

Used after a bulk load (append-only replay) that bypasses the incremental create/update/delete paths.

Source code in reactifact/context/context.py 
599
600
601
602
603
604
605
606
def rebuild_indices(self) -> None:
    """Rebuilds the type/staleness indices from the live state.

    Used after a bulk load (append-only replay) that bypasses the
    incremental `create`/`update`/`delete` paths.
    """
    recompute_stale(self)
    reindex_by_type(self)

replace_events 

replace_events(events: Sequence[Event]) -> None

Replaces the pending-event queue (used by the session load path).

Source code in reactifact/context/context.py 
608
609
610
def replace_events(self, events: Sequence[Event]) -> None:
    """Replaces the pending-event queue (used by the session load path)."""
    self._events = list(events)

commit_graph 

commit_graph() -> dict[str, list[str]]

The commit DAG as {commit id: [parent ids]}.

Source code in reactifact/context/context.py 
615
616
617
def commit_graph(self) -> dict[str, list[str]]:
    """The commit DAG as `{commit id: [parent ids]}`."""
    return self._log.graph()

history 

history() -> list[Commit]

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

Source code in reactifact/context/context.py 
629
630
631
def history(self) -> list[Commit]:
    """History: an ordered chain of commits from the oldest to head."""
    return self._log.history()

compact 

compact(
    *, keep_commits: int, keep_versions: int | None = None
) -> int

Bounds memory: squash old commit history into a baseline snapshot.

Keeps the last keep_commits commits replayable and collapses everything older into a baseline; the absolute version/head_id and therefore context_hash are unchanged (this is the history of how state got here, not the state). With keep_versions, each artifact's retained version history is also trimmed to its last keep_versions (0 drops it) — Artifact.version itself does not change.

Irreversible, like squashing git history: checkout/diff below the resulting baseline raise, and provenance for artifacts last written before it is no longer available. Call it between turns. Returns the number of commits collapsed.

Source code in reactifact/context/context.py 
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
def compact(self, *, keep_commits: int, keep_versions: int | None = None) -> int:
    """Bounds memory: squash old commit history into a baseline snapshot.

    Keeps the last `keep_commits` commits replayable and collapses
    everything older into a baseline; the absolute `version`/`head_id` and
    therefore `context_hash` are unchanged (this is the history of *how*
    state got here, not the state). With `keep_versions`, each artifact's
    retained version history is also trimmed to its last `keep_versions`
    (0 drops it) — `Artifact.version` itself does not change.

    Irreversible, like squashing git history: `checkout`/`diff` below the
    resulting baseline raise, and provenance for artifacts last written
    before it is no longer available. Call it between turns. Returns the
    number of commits collapsed.
    """
    dropped = self._log.compact(keep_commits)
    if keep_versions is not None:
        for artifact_id in self._artifacts:
            self._store.trim_history(artifact_id, keep_versions)
    if dropped:
        recompute_stale(self)
    return dropped

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/context.py 
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
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 version_a < self._log.baseline_version:
        raise ValueError(
            f"cannot diff from version {version_a}: history below the "
            f"compaction baseline ({self._log.baseline_version}) was discarded"
        )
    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/context.py 
692
693
694
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()}

artifact_version 

artifact_version(
    artifact_id: str, version: int
) -> ArtifactVersion | None

One immutable artifact revision from the version store.

version is the artifact's absolute version counter (0 = create). Compaction is authoritative: trimmed versions return None. The current version resolves the producing commit lazily when the record was written before the commit existed (the runtime path).

Source code in reactifact/context/context.py 
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
def artifact_version(
    self, artifact_id: str, version: int
) -> ArtifactVersion | None:
    """One immutable artifact revision from the version store.

    `version` is the artifact's absolute version counter (0 = create).
    Compaction is authoritative: trimmed versions return `None`. The
    current version resolves the producing commit lazily when the record
    was written before the commit existed (the runtime path).
    """
    artifact = self._artifacts.get(artifact_id)
    record = self._store.version(artifact_id, version)
    if record is None:
        return None
    if (
        record.commit_id is None
        and artifact is not None
        and version == artifact.version
    ):
        producing = self._log.producing_commit(artifact_id)
        if producing is not None:
            return ArtifactVersion(
                artifact_id=record.artifact_id,
                version=record.version,
                data_type=record.data_type,
                data=record.data,
                content_hash=record.content_hash,
                commit_id=producing.id,
                created_at=producing.timestamp,
            )
    return record

version_history 

version_history(artifact_id: str) -> list[ArtifactVersion]

All readable revisions of an artifact, oldest to newest.

Source code in reactifact/context/context.py 
765
766
767
768
769
770
771
772
773
774
775
def version_history(self, artifact_id: str) -> list[ArtifactVersion]:
    """All readable revisions of an artifact, oldest to newest."""
    stored = self._store.history(artifact_id)
    if not stored:
        return []
    versions: list[ArtifactVersion] = []
    for version in range(stored[0].version, stored[-1].version + 1):
        item = self.artifact_version(artifact_id, version)
        if item is not None:
            versions.append(item)
    return versions

find_versions 

find_versions(
    content_hash: str, *, data_type: str | None = None
) -> list[ArtifactVersion]

Every stored revision with this payload hash (content addressing).

Source code in reactifact/context/context.py 
777
778
779
780
781
def find_versions(
    self, content_hash: str, *, data_type: str | None = None
) -> list[ArtifactVersion]:
    """Every stored revision with this payload hash (content addressing)."""
    return self._store.find_versions(content_hash, data_type)

at 

at(version: int) -> Context

An immutable snapshot view of this Context at a commit version.

self is not mutated (unlike checkout); the returned Context is a detached copy with no pending events and no step cursors. Useful for reading a historical state without rewinding the live run.

Source code in reactifact/context/context.py 
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
def at(self, version: int) -> Context:
    """An immutable snapshot view of this Context at a commit version.

    `self` is not mutated (unlike `checkout`); the returned Context is a
    detached copy with no pending events and no step cursors. Useful for
    reading a historical state without rewinding the live run.
    """
    if not (0 <= version <= self.version):
        raise ValueError(
            f"Invalid snapshot version: {version} (head={self.version})"
        )
    if version < self.compacted_at:
        raise ValueError(
            f"cannot snapshot version {version}: history below the "
            f"compaction baseline ({self.compacted_at}) was discarded"
        )
    snapshot = self.clone()
    snapshot.checkout(version)
    snapshot.replace_events([])
    snapshot._step_cursors = {}
    return snapshot

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/context.py 
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
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})"
        )
    if version < self._log.baseline_version:
        raise ValueError(
            f"cannot checkout version {version}: history below the "
            f"compaction baseline ({self._log.baseline_version}) was discarded"
        )
    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)
    replayed = self._replay_store(version)
    replayed.extend(self._store, ids=set(self._artifacts) - touched)
    self._store = replayed
    self._artifacts = {}
    for artifact_id in replayed._current:
        current = replayed.current_artifact(artifact_id)
        if current is not None:
            self._artifacts[artifact_id] = current

    # 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)
    recompute_stale(self)
    reindex_by_type(self)

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/context.py 
879
880
881
882
883
884
885
886
887
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, cast(dict[str, Any], 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/context.py 
889
890
891
892
893
@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 dataclass 

Artifact(
    data: TData,
    id: str | None = None,
    created_by_commit: str | None = None,
    *,
    version: int = 0,
    content_hash: str | None = None,
    created_at: datetime | None = None,
    updated_at: datetime | None = None,
    store: ArtifactStore | None = None,
)

Bases: Generic[TData]

An immutable snapshot/handle for one artifact version.

data is a Pydantic payload; _store is an internal reference used by the deprecated history/at/diff bridges. Mutation goes through Context.

Source code in reactifact/artifacts.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def __init__(
    self,
    data: TData,
    id: str | None = None,
    created_by_commit: str | None = None,
    *,
    version: int = 0,
    content_hash: str | None = None,
    created_at: datetime | None = None,
    updated_at: datetime | None = None,
    store: ArtifactStore | None = None,
) -> None:
    now = datetime.now(UTC)
    object.__setattr__(self, "id", id or str(uuid.uuid4()))
    object.__setattr__(self, "data", data)
    object.__setattr__(self, "data_type", type_id_of(type(data)))
    object.__setattr__(self, "version", version)
    object.__setattr__(
        self, "content_hash", content_hash or artifact_content_hash(data)
    )
    object.__setattr__(self, "created_at", created_at or now)
    object.__setattr__(self, "updated_at", updated_at or now)
    object.__setattr__(self, "created_by_commit", created_by_commit)
    object.__setattr__(self, "_store", store)

history property 

history: list[TData]

Previous versions (deprecated bridge; use Context.version_history).

get_all_versions 

get_all_versions() -> list[TData]

All retained versions, including the current one (deprecated bridge).

Source code in reactifact/artifacts.py 
133
134
135
def get_all_versions(self) -> list[TData]:
    """All retained versions, including the current one (deprecated bridge)."""
    return self.history + [self.data]

at 

at(version: int) -> TData

One retained version (deprecated bridge; use Context.artifact_version).

Source code in reactifact/artifacts.py 
137
138
139
140
141
142
143
144
145
146
147
148
def at(self, version: int) -> TData:
    """One retained version (deprecated bridge; use `Context.artifact_version`)."""
    if version == self.version:
        return self.data
    if self._store is not None:
        record = self._store.version(self.id, version)
        if record is not None:
            return cast(TData, record.data)
    raise ValueError(
        f"version {version} is not available for artifact {self.id!r} "
        f"(current={self.version})"
    )

diff 

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

Diff two retained versions by list index (deprecated bridge).

Source code in reactifact/artifacts.py 
150
151
152
153
154
155
156
157
158
159
def diff(self, old_version: int, new_version: int) -> dict[str, Any]:
    """Diff two retained versions by list index (deprecated bridge)."""
    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}"
        )
    return compute_dict_diff(
        versions[old_version].model_dump(), versions[new_version].model_dump()
    )

to_dict 

to_dict() -> dict[str, Any]

Legacy v1 serialization; Context.to_dict still uses it.

Source code in reactifact/artifacts.py 
161
162
163
164
165
166
167
168
169
170
171
172
def to_dict(self) -> dict[str, Any]:
    """Legacy v1 serialization; `Context.to_dict` still uses it."""
    return {
        "id": self.id,
        "data_type": self.data_type,
        "data": self.data.model_dump(mode="json"),
        "version": self.version,
        "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,
    }

Effects

Effects 

Effects(context: Context)

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

Source code in reactifact/effects.py 
84
85
86
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 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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 = 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 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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 = 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 = 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 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
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 = 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 
162
163
164
165
166
167
168
169
170
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 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
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 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
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,
    )

act 

act(
    kind: str,
    *,
    key: str,
    payload: dict[str, Any] | None = None,
    id: str | None = None,
) -> Handle | None

Records an outbound side effect intent — a PendingAction (outbox).

Returns None if an action with the same id already exists (idempotent re-run, §42). The produce must not perform the I/O itself: the runtime dispatches pending actions via Runtime(dispatcher=...) after the intent is committed, so replay/retry/merge cannot re-send it.

key is the stable idempotency key (derived from the action's content and context, e.g. f"notify:{order_id}:{channel}"); the artifact id defaults to f"action:{key}", and key travels in the payload to the external system.

Source code in reactifact/effects.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def act(
    self,
    kind: str,
    *,
    key: str,
    payload: dict[str, Any] | None = None,
    id: str | None = None,
) -> Handle | None:
    """Records an outbound side effect *intent* — a `PendingAction` (outbox).

    Returns `None` if an action with the same id already exists (idempotent
    re-run, §42). The produce must **not** perform the I/O itself: the
    runtime dispatches pending actions via `Runtime(dispatcher=...)` after
    the intent is committed, so replay/retry/merge cannot re-send it.

    `key` is the stable idempotency key (derived from the action's content
    and context, e.g. `f"notify:{order_id}:{channel}"`); the artifact id
    defaults to `f"action:{key}"`, and `key` travels in the payload to the
    external system.
    """
    from .interrupt import PendingAction

    data = PendingAction(
        kind=kind, idempotency_key=key, payload=dict(payload or {})
    )
    return self.create_once(data, id=id or f"action:{key}")

resume 

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

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

Source code in reactifact/effects.py 
247
248
249
250
251
252
253
254
255
256
257
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 
275
276
277
278
279
280
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 
47
48
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 
60
61
62
63
64
65
66
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 
88
89
90
91
92
93
94
95
96
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,
    *,
    steps: Sequence[Step[Any]] | None = None,
    wakes: bool
    | Callable[[Event, Context | None], bool]
    | 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 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def __init__(
    self,
    name: str | None = None,
    triggers: list[Trigger] | None = None,
    priority: int | None = None,
    *,
    steps: Sequence[Step[Any]] | None = None,
    wakes: bool | Callable[[Event, Context | None], bool] | 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)
    )

    existing_steps = self.__dict__.get("steps")
    existing_produces = self.__dict__.get("produces")
    resolved: Sequence[Step[Any]] | None
    if steps is not None:
        resolved = steps
    elif existing_steps is not None:
        resolved = existing_steps
    elif existing_produces is not None:
        resolved = existing_produces
    else:
        resolved = getattr(self.__class__, "steps", None)
        if resolved is None:
            resolved = getattr(self.__class__, "produces", None)
    self.steps = list(resolved) if resolved is not None else None
    # Deprecated instance-level alias: keep it in sync with `steps`.
    self.produces = self.steps

    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()
    elif self.steps:
        self.triggers = self._generate_triggers_from_steps()
    else:
        self.triggers = []

    self.wakes = (
        wakes if wakes is not None else getattr(self.__class__, "wakes", True)
    )

    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 
183
184
185
186
187
188
189
190
191
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. When the agent has no explicit consumes, this is the union of its steps' own consumes.

Source code in reactifact/agents.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
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.
    When the agent has no explicit `consumes`, this is the union of its
    steps' own `consumes`.
    """
    if self.consumes is None and self.steps:
        inputs: list[Artifact[Any]] = []
        seen: set[str] = set()
        for step in self.steps:
            for artifact in step.collect_inputs(context):
                if artifact.id in seen:
                    continue
                seen.add(artifact.id)
                inputs.append(artifact)
        return inputs
    return self._collect_inputs(context)

run async 

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

Default: runs self.steps via execute() (the effects-first path — write a Step subclass or @step 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 
231
232
233
234
235
236
237
238
239
240
241
242
243
async def run(self, event: Event, context: Context) -> Patch | None:
    """Default: runs `self.steps` via `execute()` (the effects-first
    path — write a `Step` subclass or `@step` 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): class steps write effects.*, function steps 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 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
async def execute(self, context: Context, event: Event | None = None) -> None:
    """Runs the agent's produces (usually on an event).

    Effects-first (§24): class steps write `effects.*`, function steps `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.steps:
        return None
    inputs = self._collect_inputs(context)
    slot = current_effects()
    log = _produce_runs.get()
    for p in self.steps:
        runs, trigger = self._resolve_produce_call(p, event, context)
        if not runs:
            continue
        step_inputs = p.collect_inputs(context) if p.consumes else inputs
        scope_value = _scope_value(p.scope, trigger)
        cursor_artifacts = (
            [*step_inputs, trigger] if trigger is not None else step_inputs
        )
        input_cursor = self._input_cursor(p, cursor_artifacts, scope_value)
        cursor_key = (self.name, p.name, scope_value)
        trigger_cursor = (
            {type(trigger.data): {trigger.id: trigger.version}}
            if trigger is not None
            else {}
        )
        if (
            trigger_cursor
            and context._step_cursors.get(cursor_key) == trigger_cursor
        ):
            continue
        before = len(slot.operations) if slot is not None else 0
        started_at = datetime.now(UTC)
        started = time.monotonic()
        call = StepCall(
            context=context,
            inputs=step_inputs,
            event=event,
            trigger=trigger,
            scope=scope_value,
            cursor={t: dict(v) for t, v in input_cursor.items()},
        )
        produce_token = set_current_produce(p.name)
        context_token = set_step_context(context)
        call_token = set_step_call(call)
        try:
            result = await invoke_step(p, call)
            p._apply_result(result, call.effects)
            if trigger_cursor:
                context._step_cursors[cursor_key] = trigger_cursor
        finally:
            reset_step_call(call_token)
            reset_step_context(context_token)
            reset_current_produce(produce_token)
            # Record the produce even when it raises — a failing produce is
            # exactly what a trace should show (the span is then the error
            # span, with this produce's partial operations/latency).
            if log is not None:
                operations = (
                    len(slot.operations) if slot is not None else 0
                ) - before
                log.append(
                    (
                        p.name,
                        operations,
                        (time.monotonic() - started) * 1000,
                        started_at,
                    )
                )
    return None

Produce

Produce module-attribute 

Produce = Step

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,
    include_subclasses: 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).

include_subclasses (default False): wake on artifacts whose type is a subclass of artifact_type, not only the exact type — the declarative form of a marker base / protocol base (Consume(ChatMessage, include_subclasses=True) fires for UserQuery and Answer). collect() already reads by issubclass (Context.list_artifacts), so the flag only affects triggering. Opt-in so an agent subscribed to both a base and a concrete type never wakes twice for the same event.

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 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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,
    include_subclasses: 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
    self.include_subclasses = (
        include_subclasses
        if include_subclasses is not None
        else self.__class__.include_subclasses
    )
    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 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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,
            include_subclasses=self.include_subclasses,
        )
        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 
120
121
122
123
124
125
126
127
128
129
130
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 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@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_base classmethod 

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

Creates a Consume that also wakes on subclasses of artifact_type.

Sugar for Consume(artifact_type, include_subclasses=True) — e.g. a marker base ChatMessage with concrete UserQuery/Answer artifacts.

Source code in reactifact/consume.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@classmethod
def by_base(
    cls,
    artifact_type: ArtifactType,
    event_types: Sequence[EventType] = (
        EventType.ARTIFACT_CREATED,
        EventType.ARTIFACT_UPDATED,
    ),
    *,
    wakes: bool = True,
    debounce: bool = False,
) -> Consume:
    """Creates a Consume that also wakes on subclasses of `artifact_type`.

    Sugar for `Consume(artifact_type, include_subclasses=True)` — e.g. a
    marker base `ChatMessage` with concrete `UserQuery`/`Answer` artifacts.
    """
    return cls(
        artifact_type,
        event_types=event_types,
        wakes=wakes,
        debounce=debounce,
        include_subclasses=True,
    )

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 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
@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",
    dispatcher: Dispatcher | None = None,
    on_dispatch_error: DispatchErrorHandler | None = None,
)
Source code in reactifact/runtime/runtime.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
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",
    dispatcher: Dispatcher | None = None,
    on_dispatch_error: DispatchErrorHandler | None = None,
):
    self.context = context
    self.agents = agents or []
    self.max_concurrency = max_concurrency
    self.session = session
    self.budget = budget
    self.scheduler = scheduler
    # Outbox (§42): when set, the runtime performs every committed-but-
    # undispatched `PendingAction` after the generation commits, once per
    # stable id, and records the outcome as state. Replay never runs the
    # runtime, so it never re-dispatches; retries/merges reuse a stable id
    # and produce no new intent. None (default) = no outbox.
    self.dispatcher = dispatcher
    #: Called when the dispatcher raises (the action is marked `failed` and
    #: the exception still propagates — fail-loud, §69). A place to alert or
    #: re-arm; retry policy/backoff is the application's.
    self.on_dispatch_error = on_dispatch_error
    self._outbox = Outbox(context, dispatcher, on_dispatch_error)
    self._patches = PatchApplicator(context)
    # "per_commit" (default): persist at every generation boundary, after
    # that generation's trigger batch is consumed — so the session survives
    # a crash between generations and the saved queue and artifacts commit
    # as one consistent snapshot (including settled generations, which
    # commit nothing but still consume a batch; see `_arun_once_impl`).
    # "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 generation (a
    # multi-stage pipeline easily produces several generations 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._turn_budget = TurnBudget()
    self._turn_started = False
    self._no_runs_warned = False
    # Not reentrant: `arun`/`arun_once`/`astream` all mutate shared,
    # instance-level turn state (`_runs_used`, `outcome`, `_deadline`,
    # and the per-turn copy of `context.resources`). 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
    # Per-turn LLM usage counters and the wrapper installed to fill them
    # (see `_begin_turn`/`_end_turn_resources`). Only set when the active
    # budget tracks tokens/cost.
    self._tracker: BudgetTracker | None = None
    self._original_llm: Any | None = None
    #: The app's `RuntimeResources` before `_begin_turn` installs a
    #: per-turn shallow copy (budget/deadline/tracker/LLM wrappers live on
    #: that copy, so a `RuntimeResources` shared across sessions/turns is
    #: never mutated by concurrent turns). Restored on turn end.
    self._original_resources: RuntimeResources | None = None
    #: Reset token for the turn's log correlation context (see `_begin_turn`).
    self._log_token: Any | None = None
    #: Generation counter within the current turn (log correlation).
    self._generation = 0
    #: This turn's id (logs/registry); set in `_begin_turn`.
    self._run_id = ""
    #: The task running the current turn (for `ashutdown`).
    self._turn_task: asyncio.Task[Any] | None = None
    #: Set by `request_stop()`; checked between generations (§59/§69).
    self._stop_requested = asyncio.Event()

in_flight property 

in_flight: bool

Whether this Runtime is currently running a turn.

request_stop 

request_stop() -> None

Ask the run to stop at the next generation boundary (§59).

The in-flight generation finishes (its commit lands); no new generation starts, and the run ends with RunOutcome.STOPPED. For a bounded wait plus a forced cancel, use ashutdown().

Source code in reactifact/runtime/runtime.py 
220
221
222
223
224
225
226
227
def request_stop(self) -> None:
    """Ask the run to stop at the next generation boundary (§59).

    The in-flight generation finishes (its commit lands); no new generation
    starts, and the run ends with `RunOutcome.STOPPED`. For a bounded wait
    plus a forced cancel, use `ashutdown()`.
    """
    self._stop_requested.set()

ashutdown async 

ashutdown(*, timeout: float | None = None) -> None

Graceful shutdown: stop at a boundary, then wait for the turn to end.

Calls request_stop() first, then awaits the in-flight turn (if any). With timeout, force-cancels the turn after that many seconds. Does not close shared RuntimeResources — close those yourself (a ResourceScope, or await resources.aclose() in your own lifespan).

Source code in reactifact/runtime/runtime.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
async def ashutdown(self, *, timeout: float | None = None) -> None:
    """Graceful shutdown: stop at a boundary, then wait for the turn to end.

    Calls `request_stop()` first, then awaits the in-flight turn (if any).
    With `timeout`, force-cancels the turn after that many seconds. Does
    **not** close shared `RuntimeResources` — close those yourself (a
    `ResourceScope`, or `await resources.aclose()` in your own lifespan).
    """
    self.request_stop()
    task = self._turn_task
    if task is None or task is asyncio.current_task() or task.done():
        return
    if timeout is None:
        with contextlib.suppress(Exception):
            await asyncio.shield(task)
        return
    try:
        await asyncio.wait_for(asyncio.shield(task), timeout)
    except TimeoutError:
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task

flush_pending_actions async 

flush_pending_actions() -> int

Dispatches committed-but-undispatched actions without a generation.

The explicit outbox flush after context.merge() (merge is a pure state operation and never dispatches). Returns the number attempted.

Source code in reactifact/runtime/runtime.py 
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
async def flush_pending_actions(self) -> int:
    """Dispatches committed-but-undispatched actions without a generation.

    The explicit outbox flush after `context.merge()` (merge is a pure
    state operation and never dispatches). Returns the number attempted.
    """
    if self.dispatcher is None:
        return 0
    self._enter_turn()
    try:
        pending = self._outbox.pending_count()
        if pending:
            await self._outbox.drain_actions()
            if self.session is not None:
                await self.session.save()
        return pending
    finally:
        self._exit_turn()

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/runtime.py 
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
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())
            try:
                finished, _ = await asyncio.wait(
                    {get_event, wait_done}, return_when=asyncio.FIRST_COMPLETED
                )
                if get_event in finished:
                    yield get_event.result()
            finally:
                # Always cancel *and await* the losing future. A task that is
                # cancelled but never awaited stays pending until the loop
                # closes, producing "Task was destroyed but it is pending!"
                # (`Queue.get`/`Event.wait`) on every early generator close
                # (e.g. a dropped SSE client).
                for fut in (get_event, wait_done):
                    if not fut.done():
                        fut.cancel()
                await asyncio.gather(get_event, wait_done, return_exceptions=True)
        # 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()
        # Settle the runner too, so an early generator close (client
        # disconnect) does not leave the run task pending at loop shutdown.
        # Bounded: an agent that swallows `CancelledError` must never hang
        # the caller's disconnect/shutdown path.
        with contextlib.suppress(Exception):
            await asyncio.wait_for(
                asyncio.gather(task, return_exceptions=True),
                timeout=_CANCEL_GRACE_SECONDS,
            )
        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,
    trace_truncate: int
    | None
    | _UnsetTraceTruncate = _UNSET_TRACE_TRUNCATE,
    record_tool_args: bool = False,
    id_factory: IdFactory | None = None,
    pricer: Pricer | None = None,
    metrics: MetricsSink | None = None,
    principal: Principal | None = None,
    authorizer: Authorizer | None = None,
    guardrails: GuardrailPolicy | None = None,
    quota: QuotaTracker | None = None,
    **additional: Any,
)
Source code in reactifact/resources.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
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,
    trace_truncate: int | None | _UnsetTraceTruncate = _UNSET_TRACE_TRUNCATE,
    record_tool_args: bool = False,
    id_factory: IdFactory | None = None,
    pricer: Pricer | None = None,
    metrics: MetricsSink | None = None,
    principal: Principal | None = None,
    authorizer: Authorizer | None = None,
    guardrails: GuardrailPolicy | None = None,
    quota: QuotaTracker | None = None,
    **additional: Any,
):
    self.llm = llm
    self.embedder = embedder
    self.sources = sources or {}
    # A domain metrics sink (`reactifact.metrics.Metrics` satisfies the
    # protocol). Defaults to a no-op, so a produce can always call
    # `call.context.resources.metrics.increment("my_domain_total", ...)`.
    self.metrics: MetricsSink = metrics if metrics is not None else _NullMetrics()
    # Who this run acts for, and the policy that gates what it may do
    # (`reactifact.authz`). Both `None` by default: with no authorizer every
    # action is allowed, so authz is opt-in and non-breaking. Enforced by
    # the runtime on agent runs and artifact writes, by the tool loop on
    # tool execution, and available to produces as
    # `resources.require_authorized(action, resource)`.
    self.principal = principal
    self.authorizer = authorizer
    # Artifact-boundary guardrails (`reactifact.guardrails`), applied to
    # every agent-produced Create/Update before it is committed. `None`
    # disables them.
    self.guardrails = guardrails
    # Cross-turn, per-principal usage limits (`reactifact.quota`). `None`
    # disables quota enforcement.
    self.quota = quota
    # Applies `Budget.max_cost`: `(model, prompt_tokens, completion_tokens)
    # -> cost`. `None` (default) makes a cost limit inert — the framework
    # ships no built-in prices on purpose (see `reactifact.pricing`).
    self.pricer = pricer
    # 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
    # Opt-in: record tool-call arguments on `ToolRun.args` (scrubbed by
    # `redactor` when set). Off by default because args may carry secrets.
    self.record_tool_args = record_tool_args
    # How many characters of an artifact's JSON (and an LLM response) are
    # kept in a trace. `None` disables truncation. Deprecated: configure
    # `Tracer(limits=TraceLimits(max_data_chars=...))` instead; this stays
    # as a fallback for older code.
    if isinstance(trace_truncate, _UnsetTraceTruncate):
        self.trace_truncate: int | None = 1500
    else:
        warnings.warn(
            "RuntimeResources(trace_truncate=...) is deprecated; use "
            "Tracer(limits=TraceLimits(max_data_chars=...)) instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        self.trace_truncate = trace_truncate
    # 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 on a shallow copy of this instance (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. `budget_tracker` is the per-turn
    # token/cost counters (set only when the budget tracks tokens/cost),
    # read the same way. The copy keeps a resources instance shared across
    # sessions/turns from being mutated by concurrent turns.
    self.budget: Budget | None = None
    self.budget_deadline: float | None = None
    self.budget_tracker: BudgetTracker | None = None

registered property 

registered: frozenset[Any]

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

authorize 

authorize(action: str, resource: str) -> bool

Whether the current principal may action on resource.

True when no authorizer is configured (authz is opt-in).

Source code in reactifact/resources.py 
198
199
200
201
202
203
204
205
def authorize(self, action: str, resource: str) -> bool:
    """Whether the current principal may `action` on `resource`.

    `True` when no authorizer is configured (authz is opt-in).
    """
    if self.authorizer is None:
        return True
    return self.authorizer.authorize(self.principal, action, resource)

require_authorized 

require_authorized(action: str, resource: str) -> None

authorize or raise AuthorizationError — the produce-side gate.

Source code in reactifact/resources.py 
207
208
209
210
def require_authorized(self, action: str, resource: str) -> None:
    """`authorize` or raise `AuthorizationError` — the produce-side gate."""
    if not self.authorize(action, resource):
        raise AuthorizationError(action, resource, self.principal)

quota_key 

quota_key() -> str

The key a QuotaTracker counts against: the principal, else the run.

Source code in reactifact/resources.py 
212
213
214
215
216
def quota_key(self) -> str:
    """The key a `QuotaTracker` counts against: the principal, else the run."""
    if self.principal is not None:
        return self.principal.id
    return "default"

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 
220
221
222
223
224
225
226
227
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 
235
236
237
238
239
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 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
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 
256
257
258
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 
265
266
267
268
269
270
271
272
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 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
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 
293
294
295
296
297
298
@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 
40
41
42
43
44
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 
46
47
48
49
50
51
52
53
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 
55
56
57
58
59
60
61
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 
63
64
65
66
@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 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/base.py 
267
268
269
270
@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/tools/use/blocking.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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/tools/use/hitl.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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

tools.native

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/tools/native.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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/tools/native.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
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/tools/native.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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/tools/agent.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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: Step[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 Step(PendingQuestion)/Step(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(), Step(PendingQuestion)]
Source code in reactifact/verify.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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 
68
69
70
71
72
73
74
75
76
77
78
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,
    max_observation_chars: int | None = None,
    redact_tool_outputs: bool = False,
    retry_tool_errors: bool = False,
    tool_retry_attempts: int = 2,
    tool_retry_delay: float = 0.5,
    parse_calls: Callable[[LLMResponse], list[ToolCall]]
    | None = None,
    fallback: Callable[
        [Context, list[dict[str, Any]], str],
        Awaitable[str | None],
    ]
    | None = None,
    on_event: OnToolLoopEvent | 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.
  • The turn's Budget (context.resources.budget) is honoured between rounds, like ToolUse's own loop: max_tool_calls caps executed calls, max_seconds (via budget_deadline) and token/cost limits stop new rounds. The exhausted calls are reported back as tool errors, then the loop falls through to the forced answer.
  • max_observation_chars truncates each tool result before it goes back to the model (the returned observations keep the full text).
  • redact_tool_outputs=True applies context.resources.redactor to tool outputs/errors before they reach the model.
  • retry_tool_errors=True retries a tool call when it returns a ToolError(retryable=True), up to tool_retry_attempts total attempts with exponential tool_retry_delay backoff. Retries count as tool executions for Budget.max_tool_calls.
  • parse_calls overrides how a response's tool calls are extracted (e.g. a textual DSML/XML codec); fallback(ctx, messages, text) runs when no calls were parsed and the text looks like tool-call markup, and its string (when not None) becomes the final answer.
  • on_event(name, payload) and Context.announce(kind="agent") report progress for streaming: round_start, tool_call, tool_result, final.
  • 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 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
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,
    max_observation_chars: int | None = None,
    redact_tool_outputs: bool = False,
    retry_tool_errors: bool = False,
    tool_retry_attempts: int = 2,
    tool_retry_delay: float = 0.5,
    parse_calls: Callable[[LLMResponse], list[ToolCall]] | None = None,
    fallback: Callable[[Context, list[dict[str, Any]], str], Awaitable[str | None]]
    | None = None,
    on_event: OnToolLoopEvent | 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.
    - The turn's `Budget` (`context.resources.budget`) is honoured between
      rounds, like `ToolUse`'s own loop: `max_tool_calls` caps executed calls,
      `max_seconds` (via `budget_deadline`) and token/cost limits stop new
      rounds. The exhausted calls are reported back as tool errors, then the
      loop falls through to the forced answer.
    - `max_observation_chars` truncates each tool result before it goes back
      to the model (the returned `observations` keep the full text).
    - `redact_tool_outputs=True` applies `context.resources.redactor` to tool
      outputs/errors before they reach the model.
    - `retry_tool_errors=True` retries a tool call when it returns a
      `ToolError(retryable=True)`, up to `tool_retry_attempts` total attempts
      with exponential `tool_retry_delay` backoff. Retries count as tool
      executions for `Budget.max_tool_calls`.
    - `parse_calls` overrides how a response's tool calls are extracted (e.g.
      a textual DSML/XML codec); `fallback(ctx, messages, text)` runs when no
      calls were parsed and the text looks like tool-call markup, and its
      string (when not `None`) becomes the final answer.
    - `on_event(name, payload)` and `Context.announce(kind="agent")` report
      progress for streaming: `round_start`, `tool_call`, `tool_result`,
      `final`.
    - 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}
    parse = parse_calls or parse_tool_calls
    redactor = context.resources.redactor if redact_tool_outputs else None
    if tool_retry_attempts < 1:
        raise ValueError("tool_retry_attempts must be >= 1")
    retry_attempts = tool_retry_attempts if retry_tool_errors else 1
    retry_delay = tool_retry_delay if retry_tool_errors else 0.0
    required = {mandatory} if isinstance(mandatory, str) else set(mandatory or ())
    budget = context.resources.budget
    max_tool_calls = budget.max_tool_calls if budget is not None else None
    deadline = context.resources.budget_deadline
    tracker = context.resources.budget_tracker
    messages: list[dict[str, Any]] = [
        {"role": "system", "content": system},
        {"role": "user", "content": user},
    ]
    observations: list[ToolObservation] = []
    called: set[str] = set()
    rounds = 0
    executed = 0

    def budget_spent() -> bool:
        return (deadline is not None and time.monotonic() >= deadline) or (
            tracker is not None
            and budget is not None
            and tracker.exhausted(budget) is not None
        )

    async def emit(name: str, payload: dict[str, Any]) -> None:
        message = str(payload.get("message", name))
        data = {key: value for key, value in payload.items() if key != "message"}
        context.announce(message, kind="agent", **data)
        if on_event is not None:
            result = on_event(name, payload)
            if inspect.isawaitable(result):
                await result

    def sanitize(observation: ToolObservation) -> ToolObservation:
        if redactor is None:
            return observation
        return ToolObservation(
            call=observation.call,
            output=redactor.redact(observation.output),
            error=redactor.redact(observation.error),
            error_info=observation.error_info,
            attempts=observation.attempts,
        )

    stop_reason: ToolLoopStopReason = "completed"
    for _ in range(max(0, max_rounds)):
        if budget_spent():
            stop_reason = "budget"
            break  # time/token/cost budget spent — fall through to the answer
        await emit("round_start", {"round": rounds + 1})
        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,
                stop_reason="provider_error",
                error="provider call failed or no provider is configured",
            )
        rounds += 1
        calls = parse(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
            if fallback is not None and looks_like_tool_markup(response.text):
                fallback_text = await fallback(context, messages, response.text)
                if fallback_text is not None:
                    await emit("final", {"text": fallback_text, "fallback": True})
                    return ToolLoopResult(
                        fallback_text,
                        messages,
                        observations,
                        rounds,
                        stop_reason="fallback",
                    )
            await emit("final", {"text": response.text})
            return ToolLoopResult(response.text, messages, observations, rounds)

        messages.append(
            {
                "role": "assistant",
                "content": response.text or "",
                "tool_calls": [_as_message(c) for c in calls],
            }
        )
        to_run = calls
        skipped: list[ToolCall] = []
        if max_tool_calls is not None:
            remaining = max(0, max_tool_calls - executed)
            to_run, skipped = calls[:remaining], calls[remaining:]
        for call in to_run:
            await emit(
                "tool_call",
                {
                    "message": f"Calling tool '{call.name}'…",
                    "tool": call.name,
                    "args": call.args,
                },
            )
        batch = (
            await _execute(
                tool_map,
                to_run,
                parallel=parallel,
                retry_attempts=retry_attempts,
                retry_delay=retry_delay,
            )
            if to_run
            else []
        )
        batch.extend(
            ToolObservation(
                call=call,
                error="tool-call budget exhausted; answer from the data already gathered",
            )
            for call in skipped
        )
        batch = [sanitize(observation) for observation in batch]
        executed += sum(observation.attempts for observation in batch[: len(to_run)])
        observations.extend(batch)
        called.update(c.name for c in to_run)
        for observation in batch:
            content = _content(observation)
            await emit(
                "tool_result",
                {
                    "message": (
                        f"Tool '{observation.name}' failed"
                        if observation.error
                        else f"Tool '{observation.name}' returned {len(content)} chars"
                    ),
                    "tool": observation.name,
                    "error": bool(observation.error),
                    "chars": len(content),
                },
            )
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": observation.call.id,
                    "content": _truncate(content, max_observation_chars),
                }
            )
        if skipped:
            stop_reason = "budget"
            break  # tool-call budget spent — force the answer now

    # out of rounds (or budget) — force a plain answer without tools
    if stop_reason == "completed":
        stop_reason = "max_rounds"
    response = await native_complete(
        context, messages=messages, temperature=temperature, max_tokens=max_tokens
    )
    text = response.text if response is not None else ""
    if response is None and stop_reason == "max_rounds":
        stop_reason = "provider_error"
    await emit("final", {"text": text})
    return ToolLoopResult(
        text,
        messages,
        observations,
        rounds,
        stop_reason=stop_reason,
        error=(
            "provider call failed or no provider is configured"
            if response is None
            else None
        ),
    )

Budgets

Budget

Budget

Bases: BaseModel

Resource limit for a single run (turn), applied by the runtime.

tracks_llm_usage property 

tracks_llm_usage: bool

Whether this budget needs an LLM-usage tracker (tokens/cost limits).

BudgetTracker

BudgetTracker 

BudgetTracker()

Per-turn LLM usage counters, checked against a Budget.

The runtime creates one per turn when the budget has token/cost limits and records every LLM response's usage into it — independent of tracing, which is optional (RecordingLLM is only installed when a tracer is set).

Source code in reactifact/budget.py 
91
92
93
94
def __init__(self) -> None:
    self.prompt_tokens = 0
    self.completion_tokens = 0
    self.cost = 0.0

exhausted 

exhausted(budget: Budget | None) -> RunOutcome | None

The budget-exceeded outcome, or None while still within budget.

Source code in reactifact/budget.py 
107
108
109
110
111
112
113
114
115
def exhausted(self, budget: Budget | None) -> RunOutcome | None:
    """The budget-exceeded outcome, or `None` while still within budget."""
    if budget is None:
        return None
    if budget.max_tokens is not None and self.total_tokens >= budget.max_tokens:
        return RunOutcome.BUDGET_TOKENS_EXCEEDED
    if budget.max_cost is not None and self.cost >= budget.max_cost:
        return RunOutcome.BUDGET_COST_EXCEEDED
    return None

Schema types

register

register 

register(
    model: type[BaseModel],
    *,
    type_id: str | None = None,
    aliases: Iterable[str] = (),
    migrate: Migrate | None = None,
) -> type[BaseModel]

Registers model's stable type_id (default: its TYPE_ID classvar, or the qualified name), aliases (previously-persisted ids — e.g. a model's old qualified name after a move), and an optional migrate hook.

Returns model, so it works as a decorator. Re-registering the same model replaces its spec; two different models claiming one id is an error.

Source code in reactifact/types.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def register(
    model: type[BaseModel],
    *,
    type_id: str | None = None,
    aliases: Iterable[str] = (),
    migrate: Migrate | None = None,
) -> type[BaseModel]:
    """Registers `model`'s stable `type_id` (default: its `TYPE_ID` classvar, or
    the qualified name), `aliases` (previously-persisted ids — e.g. a model's old
    qualified name after a move), and an optional `migrate` hook.

    Returns `model`, so it works as a decorator. Re-registering the same model
    replaces its spec; two *different* models claiming one id is an error.
    """
    resolved_id = type_id or getattr(model, "TYPE_ID", None) or _qualname(model)
    spec = TypeSpec(
        model=model,
        type_id=str(resolved_id),
        aliases=tuple(aliases),
        migrate=migrate,
    )
    _unregister(model)
    for key in (spec.type_id, *spec.aliases):
        existing = _BY_ID.get(key)
        if existing is not None and existing.model is not model:
            raise ValueError(
                f"type id {key!r} is already registered to {existing.model!r}; "
                f"cannot also map it to {model!r}"
            )
    _SPECS[model] = spec
    _BY_ID[spec.type_id] = spec
    for alias in spec.aliases:
        _BY_ID[alias] = spec
    return model

resolve

resolve 

resolve(type_id: str) -> type[BaseModel]

The class for a persisted type_id, or raise TypeNotFoundError.

Registry first, then a registered alias, then importing the id as a qualified name (so pre-registry payloads keep loading).

Source code in reactifact/types.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def resolve(type_id: str) -> type[BaseModel]:
    """The class for a persisted `type_id`, or raise `TypeNotFoundError`.

    Registry first, then a registered alias, then importing the id as a
    qualified name (so pre-registry payloads keep loading).
    """
    spec = _BY_ID.get(type_id)
    if spec is not None:
        return spec.model
    module_name, sep, class_name = type_id.rpartition(".")
    if sep:
        try:
            candidate = getattr(importlib.import_module(module_name), class_name)
        except (ImportError, AttributeError, ValueError):
            candidate = None
        if isinstance(candidate, type) and issubclass(candidate, BaseModel):
            return candidate
    raise TypeNotFoundError(
        f"unknown artifact type {type_id!r}: no registered type id/alias resolves "
        f"it, and it is not importable as a qualified name. Register it with "
        f"reactifact.types.register(Model, type_id=..., aliases=[{type_id!r}]) to "
        f"keep old sessions/recordings loadable."
    )

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 
 99
100
101
102
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 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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}"
            )

assert_golden_file

assert_golden_file 

assert_golden_file(
    context: Context,
    path: str | Path,
    *,
    trace: RunTrace | None = None,
    update: bool | None = None,
) -> GoldenRun

Path-aware snapshot assertion: writes the snapshot when it is missing (or update/$REACTIFACT_GOLDEN_UPDATE is set), otherwise compares it.

The first run seeds the file, so a golden test is a one-liner and later drift fails. Returns the GoldenRun now on disk.

Source code in reactifact/testing/golden.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def assert_golden_file(
    context: Context,
    path: str | Path,
    *,
    trace: RunTrace | None = None,
    update: bool | None = None,
) -> GoldenRun:
    """Path-aware snapshot assertion: writes the snapshot when it is missing
    (or `update`/`$REACTIFACT_GOLDEN_UPDATE` is set), otherwise compares it.

    The first run seeds the file, so a golden test is a one-liner and later
    drift fails. Returns the `GoldenRun` now on disk.
    """
    target = Path(path)
    if _update_requested(update) or not target.exists():
        golden = capture(context, trace=trace)
        golden.save(target)
        return golden
    golden = GoldenRun.load(target)
    assert_golden(context, golden, trace=trace)
    return golden

Testing harness

ScenarioLab

ScenarioLab 

ScenarioLab(
    agents: list[Agent],
    *,
    resources: RuntimeResources
    | Callable[[], RuntimeResources]
    | None = None,
    budget: Budget | None = None,
    max_concurrency: int | None = None,
    isolate_errors: bool = True,
    mode: Mode = "live",
    recording_path: str | Path = "scenario_calls.jsonl",
)

Runs agents against seeded artifacts, once per run() call.

mode controls the LLM behind resources.llm (see reactifact.testing.record): - "live" (default): the real provider, unchanged. - "record": wraps it in ReplayLLM(mode="record"), appending every call to recording_path. - "replay": wraps it in ReplayLLM(mode="replay") — no live network calls; a call that diverges from the recording raises ReplayMiss rather than being answered with a guess (§59).

isolate_errors defaults to True here (unlike Runtime's own default) so a scenario can assert on a crashed agent via result.errors instead of the whole run() raising; pass False to let an unexpected agent exception fail the test the ordinary way.

Source code in reactifact/testing/lab.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
def __init__(
    self,
    agents: list[Agent],
    *,
    resources: RuntimeResources | Callable[[], RuntimeResources] | None = None,
    budget: Budget | None = None,
    max_concurrency: int | None = None,
    isolate_errors: bool = True,
    mode: Mode = "live",
    recording_path: str | Path = "scenario_calls.jsonl",
) -> None:
    self._agents = agents
    self._resources = resources
    self._budget = budget
    self._max_concurrency = max_concurrency
    self._isolate_errors = isolate_errors
    self._mode: Mode = mode
    self._recording_path = Path(recording_path)
    self._faults: list[ToolFault] = []
    self._resource_faults: list[ResourceFault] = []

fail 

fail(
    tool_name: str,
    error: BaseException | Callable[[], BaseException],
    *,
    times: int | None = None,
    when: Callable[[dict[str, Any]], bool] | None = None,
    delay: float = 0.0,
) -> None

Queues a fault for the next run()/.turn().

tool_name raises error (or the result of calling it, if callable — useful for a fresh exception instance per call) instead of executing. times=None (default) faults every call; times=N faults the first N calls, then delegates to the real tool; when(args) narrows it to matching calls; delay seconds are slept before raising (a slow-then-fail tool). Queued faults are one-shot — consumed by the next run()/.turn() only.

Source code in reactifact/testing/lab.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
def fail(
    self,
    tool_name: str,
    error: BaseException | Callable[[], BaseException],
    *,
    times: int | None = None,
    when: Callable[[dict[str, Any]], bool] | None = None,
    delay: float = 0.0,
) -> None:
    """Queues a fault for the next `run()`/`.turn()`.

    `tool_name` raises `error` (or the result of calling it, if callable —
    useful for a fresh exception instance per call) instead of executing.
    `times=None` (default) faults every call; `times=N` faults the first
    `N` calls, then delegates to the real tool; `when(args)` narrows it to
    matching calls; `delay` seconds are slept before raising (a
    slow-then-fail tool). Queued faults are one-shot — consumed by the next
    `run()`/`.turn()` only.
    """
    self._faults.append(
        ToolFault(tool_name, error, times=times, when=when, delay=delay)
    )

stub_tool 

stub_tool(
    tool_name: str,
    *,
    text: str = "",
    error: str | None = None,
    times: int | None = None,
    when: Callable[[dict[str, Any]], bool] | None = None,
    delay: float = 0.0,
) -> None

Queues a canned-response stub for tool_name — the non-failing counterpart of fail(): the tool returns ToolOutput(text=..., error=...) instead of executing. Same times/when/delay semantics.

Source code in reactifact/testing/lab.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
def stub_tool(
    self,
    tool_name: str,
    *,
    text: str = "",
    error: str | None = None,
    times: int | None = None,
    when: Callable[[dict[str, Any]], bool] | None = None,
    delay: float = 0.0,
) -> None:
    """Queues a canned-response stub for `tool_name` — the non-failing
    counterpart of `fail()`: the tool returns `ToolOutput(text=..., error=...)`
    instead of executing. Same `times`/`when`/`delay` semantics.
    """
    self._faults.append(
        ToolFault(
            tool_name,
            times=times,
            when=when,
            delay=delay,
            output=ToolOutput(text=text, error=error or ""),
        )
    )

fail_resource 

fail_resource(
    resource: str | type[Any] | ResourceKey[Any],
    error: BaseException | Callable[[], BaseException],
    *,
    method: str | None = None,
    times: int | None = None,
    when: Callable[[tuple[Any, ...], dict[str, Any]], bool]
    | None = None,
    delay: float = 0.0,
) -> None

Queues a fault for a resource — the general-purpose analog of fail() for anything that isn't a tool.

resource addresses it either by string name — "llm", "embedder", a source id (resources.sources[id]), or a name set via resources.set(name, ...) — or by a typed key: a class or a ResourceKey registered with resources.register(...).

Wraps the resource in a duck-typed proxy for the next run()/ .turn(): method=None (default) fails every callable on it; naming one method (e.g. "embed", "search") faults only that method. times=None faults every call; times=N faults the first N, then delegates; when(args, kwargs) narrows it to matching calls; delay seconds are slept first. Raises ScenarioError at run time if resource doesn't match any resource, or matches one that's None (nothing configured to fail).

Source code in reactifact/testing/lab.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
def fail_resource(
    self,
    resource: str | type[Any] | ResourceKey[Any],
    error: BaseException | Callable[[], BaseException],
    *,
    method: str | None = None,
    times: int | None = None,
    when: Callable[[tuple[Any, ...], dict[str, Any]], bool] | None = None,
    delay: float = 0.0,
) -> None:
    """Queues a fault for a resource — the general-purpose analog of
    `fail()` for anything that isn't a tool.

    `resource` addresses it either by **string name** — `"llm"`,
    `"embedder"`, a source id (`resources.sources[id]`), or a name set via
    `resources.set(name, ...)` — or by a **typed key**: a class or a
    `ResourceKey` registered with `resources.register(...)`.

    Wraps the resource in a duck-typed proxy for the next `run()`/
    `.turn()`: `method=None` (default) fails every callable on it;
    naming one method (e.g. `"embed"`, `"search"`) faults only that
    method. `times=None` faults every call; `times=N` faults the first
    `N`, then delegates; `when(args, kwargs)` narrows it to matching calls;
    `delay` seconds are slept first. Raises `ScenarioError` at run time if
    `resource` doesn't match any resource, or matches one that's `None`
    (nothing configured to fail).
    """
    self._resource_faults.append(
        ResourceFault(
            resource, error, method=method, times=times, when=when, delay=delay
        )
    )

stub_resource 

stub_resource(
    resource: str | type[Any] | ResourceKey[Any],
    *,
    returns: Any = UNSET,
    side_effect: list[Any]
    | Callable[..., Any]
    | None = None,
    method: str | None = None,
    times: int | None = None,
    when: Callable[[tuple[Any, ...], dict[str, Any]], bool]
    | None = None,
    delay: float = 0.0,
) -> None

Queues a stub for a resource — the non-failing counterpart of fail_resource() for the same addressing (string name or typed key).

returns is returned on every stubbed call; or side_effect walks a sequence/list (unittest.mock semantics: an item that is an Exception raises, anything else is returned) or a callable called as side_effect(*args, **kwargs). times/when/delay/method behave as on fail_resource.

Source code in reactifact/testing/lab.py 
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
def stub_resource(
    self,
    resource: str | type[Any] | ResourceKey[Any],
    *,
    returns: Any = UNSET,
    side_effect: list[Any] | Callable[..., Any] | None = None,
    method: str | None = None,
    times: int | None = None,
    when: Callable[[tuple[Any, ...], dict[str, Any]], bool] | None = None,
    delay: float = 0.0,
) -> None:
    """Queues a stub for a resource — the non-failing counterpart of
    `fail_resource()` for the same addressing (string name or typed key).

    `returns` is returned on every stubbed call; or `side_effect` walks a
    sequence/list (`unittest.mock` semantics: an item that is an
    `Exception` raises, anything else is returned) or a callable called as
    `side_effect(*args, **kwargs)`. `times`/`when`/`delay`/`method` behave
    as on `fail_resource`.
    """
    self._resource_faults.append(
        ResourceFault(
            resource,
            method=method,
            times=times,
            returns=returns,
            side_effect=side_effect,
            when=when,
            delay=delay,
        )
    )

run async 

run(
    *seed: Any, max_iterations: int = 100
) -> ScenarioResult

Seeds seed artifacts into a fresh Context, runs the agents to completion (or budget/iteration exhaustion), and returns the result.

Source code in reactifact/testing/lab.py 
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
async def run(self, *seed: Any, max_iterations: int = 100) -> ScenarioResult:
    """Seeds `seed` artifacts into a fresh `Context`, runs the agents to
    completion (or budget/iteration exhaustion), and returns the result.
    """
    context = Context(resources=self._build_resources())
    for data in seed:
        context.create(data)
    return await _execute_turn(
        agents=self._agents,
        faults=self._take_faults(),
        resource_faults=self._take_resource_faults(),
        budget=self._budget,
        max_concurrency=self._max_concurrency,
        isolate_errors=self._isolate_errors,
        context=context,
        max_iterations=max_iterations,
    )

run_sync 

run_sync(
    *seed: Any, max_iterations: int = 100
) -> ScenarioResult

Synchronous run() — mirrors Runtime.run()/run_once() for plain (non-async) pytest tests. Uses asyncio.run, so it can't be called from inside an already-running event loop.

Source code in reactifact/testing/lab.py 
528
529
530
531
532
def run_sync(self, *seed: Any, max_iterations: int = 100) -> ScenarioResult:
    """Synchronous `run()` — mirrors `Runtime.run()`/`run_once()` for plain
    (non-async) pytest tests. Uses `asyncio.run`, so it can't be called from
    inside an already-running event loop."""
    return asyncio.run(self.run(*seed, max_iterations=max_iterations))

scenario 

scenario() -> Scenario

Starts a multi-turn scenario: one Context reused across .turn() calls, for flows that need more than one round of user input to reach the state under test (e.g. a design pick, then a plan, then an approval — each its own turn on the same project). The record/replay LLM wrapper is built once here, so its call index persists correctly across turns.

Source code in reactifact/testing/lab.py 
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
def scenario(self) -> Scenario:
    """Starts a multi-turn scenario: one `Context` reused across `.turn()`
    calls, for flows that need more than one round of user input to reach
    the state under test (e.g. a design pick, then a plan, then an
    approval — each its own turn on the same project). The record/replay
    LLM wrapper is built once here, so its call index persists correctly
    across turns.
    """
    context = Context(resources=self._build_resources())
    return Scenario(
        agents=self._agents,
        budget=self._budget,
        max_concurrency=self._max_concurrency,
        isolate_errors=self._isolate_errors,
        take_faults=self._take_faults,
        take_resource_faults=self._take_resource_faults,
        context=context,
    )

RelationAssertions

RelationAssertions 

RelationAssertions(context: Context)

Assertions over the artifact graph's relations (Context.relations()).

ArtifactAssertions checks an artifact's fields; this checks the edges: that provenance exists, that it doesn't, and how many. Any filter left None is a wildcard.

Source code in reactifact/testing/assertions.py 
195
196
def __init__(self, context: Context) -> None:
    self._context = context