Skip to content

API reference

The reference is generated from relayr's docstrings and type annotations. It documents each package's public API: the names in its __all__.

relayr.events

The bridged event types, and the projections that choose what each envelope becomes.

artifactr's events reach reflexr as types in the artifactr namespace, such as artifactr:message_posted, under ArtifactrEvent. Give BRIDGED to reflexr's Workspaces(events=...) with the application's own types, and register projections for the application's app_events, artifact kinds and deferred tools:

projections = Projections(app_events={"deploy": deployed}, artifacts={"timeline": timeline})

BRIDGED module-attribute 

BRIDGED: tuple[type[ArtifactrEvent], ...] = (
    MessagePosted,
    ArtifactCreated,
    ArtifactChanged,
    ArtifactArchived,
    ProposalCreated,
    ProposalResolved,
    TurnEnded,
    FeedbackGiven,
    DeferredAnswered,
)

Every bridged type: give them to reflexr's Workspaces(events=...), with the application's own, so the follower may publish them.

Projection 

Projection = Callable[[Envelope], Event | None]

Maps an artifactr envelope to the reflexr event it becomes, or to None to bridge nothing.

An exception, a pydantic ValidationError included, dead-letters the envelope.

Projections 

Projections(
    *,
    app_events: Mapping[str, Projection] | None = None,
    artifacts: Mapping[str, Projection] | None = None,
    deferred: Mapping[str, Projection] | None = None,
)

What the follower publishes for each envelope: the default set, and the application's.

Parameters:

Name Type Description Default
app_events Mapping[str, Projection] | None

A projection per app_event name. An app_event with none stays in artifactr.

 None
artifacts Mapping[str, Projection] | None

A projection per artifact kind, for its artifact_created, artifact_changed and artifact_archived, in place of the default set's events. Proposals on the kind are bridged as the default set's.

 None
deferred Mapping[str, Projection] | None

A projection per deferred tool's name, for the deferred_answered of a request to that tool, in place of the default set's artifactr:deferred_answered.

 None

project 

project(envelope: Envelope) -> Event | None

Return the event an envelope becomes, or None if it stays in artifactr.

Raises:

Type Description
Exception

Whatever a projection raises, a pydantic ValidationError included.

ArtifactArchived pydantic-model

Bases: ArtifactrEvent

An artifact was archived.

Fields:

proposal_id pydantic-field 

proposal_id: ProposalId | None = None

The accepted proposal that archived it, if one did.

ArtifactChanged pydantic-model

Bases: ArtifactrEvent

An artifact was changed by a patch.

Fields:

proposal_id pydantic-field 

proposal_id: ProposalId | None = None

The accepted proposal that changed it, if one did.

ArtifactCreated pydantic-model

Bases: ArtifactrEvent

An artifact was created at version 1.

Fields:

proposal_id pydantic-field 

proposal_id: ProposalId | None = None

The accepted proposal that created it, if one did.

ArtifactrEvent pydantic-model

Bases: Event

The base of the bridged types: what every one says about where it came from.

Config:

  • frozen: True
  • extra: forbid

Fields:

author pydantic-field 

author: Actor

Who acted in artifactr: the envelope's actor there.

thread_id pydantic-field 

thread_id: ThreadId | None = None

The thread it happened in, if any: a chain that begins with it carries the thread.

turn_id pydantic-field 

turn_id: RunId | None = None

The agent turn that did it, if one did: artifactr's run id.

DeferredAnswered pydantic-model

Bases: ArtifactrEvent

A paused turn's question was answered, or its tool call approved or declined.

Its author is who answered, and its turn_id the turn that asked. A chat message that answers is bridged as artifactr:message_posted too.

Fields:

tool_name pydantic-field 

tool_name: str | None = None

The tool whose call was answered, such as ask_user; None only for answers artifactr recorded before it named the tool.

approved pydantic-field 

approved: bool | None = None

Whether an approval was given; None for a question's answer.

FeedbackGiven pydantic-model

Bases: ArtifactrEvent

A person or an evaluator gave feedback in artifactr. Its author is who gave it.

Fields:

MessagePosted pydantic-model

Bases: ArtifactrEvent

A message, or a notice, was posted in a thread.

Fields:

ProposalCreated pydantic-model

Bases: ArtifactrEvent

A change was proposed.

Fields:

proposed_by pydantic-field 

proposed_by: Actor

Who proposed it: its author, named as on artifactr:proposal_resolved.

ProposalResolved pydantic-model

Bases: ArtifactrEvent

A proposal was accepted or rejected. Its author is who resolved it.

Fields:

changes pydantic-field 

changes: Patch | None = None

The changes made to the proposal as it was accepted, if any.

version pydantic-field 

version: int | None = None

The artifact's version the acceptance made.

TurnEnded pydantic-model

Bases: ArtifactrEvent

An agent turn ended: artifactr's run_ended.

Fields:

relayr.ledger

The ledger: relayr's own storage, as a port with an in-memory adapter.

A follower keeps its workspace's cursor and lease here, and the envelopes it could not bridge; the outbound adapter keeps the run behind each of its writes, so the follower continues its chain; and the tracker connector claims each key it writes by before it writes, and its followers keep what they read of each bound project. The SQL adapter is relayr.sql.

Clock module-attribute 

Clock = Callable[[], datetime]

Returns the current time, which leases expire by; injectable so tests control it.

CauseKind 

CauseKind = Literal[
    "proposal", "artifact", "edit", "message", "turn"
]

What a cause is recorded by: a proposal's, a created artifact's or a message's id; an edited artifact's id and the version the edit makes, as <id>@<version>; or the id of an agent turn a message started.

TrackedKind 

TrackedKind = Literal['item', 'comment']

What a tracker key names: an item a run files, or a comment it posts.

Cause pydantic-model

Bases: BaseModel

The reflexr run behind an artifactr write: what continues its causal chain.

Config:

  • frozen: True

Fields:

correlation_id pydantic-field 

correlation_id: str

The run's chain: the id of its first event.

depth pydantic-field 

depth: int

The depth of the run's own events: one deeper than what made its rule fire.

DeadLetter pydantic-model

Bases: BaseModel

An artifactr envelope the follower could not bridge, and why.

Config:

  • frozen: True

Fields:

seq pydantic-field 

seq: int

The envelope's position in the artifactr workspace's log.

event_type pydantic-field 

event_type: str

The artifactr event's type, such as message_posted.

error pydantic-field 

error: str

What went wrong: the exception's type and message.

Ledger

Bases: Protocol

Where followers keep their cursors, leases and dead letters, and runs their tracker keys.

acquire_lease async 

acquire_lease(
    workspace: WorkspaceKey, holder: str, ttl: timedelta
) -> bool

Take or renew the workspace's lease for ttl. Return False if another holder has it.

release_lease async 

release_lease(workspace: WorkspaceKey, holder: str) -> None

Release the workspace's lease, if holder has it.

cursor async 

cursor(workspace: WorkspaceKey) -> int

Return the seq of the last envelope the workspace's follower is done with, or 0.

save_cursor async 

save_cursor(workspace: WorkspaceKey, seq: int) -> None

Save that the follower is done with seq. A cursor only moves forward.

dead_letter async 

dead_letter(
    workspace: WorkspaceKey, letter: DeadLetter
) -> None

Record an envelope the follower could not bridge; once per seq, the first kept.

dead_letters async 

dead_letters(workspace: WorkspaceKey) -> list[DeadLetter]

Return the workspace's dead letters, in seq order.

record_cause async 

record_cause(
    workspace: WorkspaceKey,
    kind: CauseKind,
    key: str,
    cause: Cause,
) -> None

Record the run behind a write, by what the follower will find it by; the first kept.

cause async 

cause(
    workspace: WorkspaceKey, kind: CauseKind, key: str
) -> Cause | None

Return the run behind a write, or None if no run of a rule made it.

forget_causes async 

forget_causes(
    workspace: WorkspaceKey, *, older_than: timedelta
) -> None

Forget the causes recorded longer ago than older_than, by the ledger's clock.

followed async 

followed() -> list[WorkspaceKey]

Return every workspace a follower has held a lease on or saved a cursor for, in order.

These are the workspaces to follow again when the application starts.

claim_tracked async 

claim_tracked(
    workspace: WorkspaceKey,
    kind: TrackedKind,
    key: str,
    correlation_id: str,
    *,
    stale_after: timedelta,
) -> bool

Claim a key before writing by it: True if the claim is the caller's.

A new key is claimed with correlation_id. A key claimed longer ago than stale_after that has no record is claimed again, keeping its first correlation id. A key with a record, or claimed more recently, is not the caller's.

tracked async 

tracked(
    workspace: WorkspaceKey, kind: TrackedKind, key: str
) -> Tracked | None

Return a key's claim and record, or None if it was never claimed.

record_tracked async 

record_tracked(
    workspace: WorkspaceKey,
    kind: TrackedKind,
    key: str,
    ref: dict[str, JsonValue],
) -> dict[str, JsonValue] | None

Record what a claimed key made, the first kept, and return the record kept.

Returns None, recording nothing, if the key was never claimed.

every_tracked async 

every_tracked(
    workspace: WorkspaceKey,
) -> dict[TrackedKey, Tracked]

Return every key claimed in the workspace, and what each made, by kind, then key.

release_tracked async 

release_tracked(
    workspace: WorkspaceKey, kind: TrackedKind, key: str
) -> None

Release a key's claim that recorded nothing, after a write that certainly didn't land.

mark_tracked_deleted async 

mark_tracked_deleted(
    workspace: WorkspaceKey, key: str
) -> None

Mark an item's key deleted, so it files nothing again.

record_written async 

record_written(
    workspace: WorkspaceKey,
    key: str,
    fields: Mapping[str, JsonValue],
) -> None

Record the values relayr is about to set on an item's fields, by its key.

forget_written async 

forget_written(
    workspace: WorkspaceKey,
    key: str,
    fields: Collection[str],
) -> None

Forget the values recorded for an item's fields.

consume_written async 

consume_written(
    workspace: WorkspaceKey,
    key: str,
    values: Mapping[str, JsonValue],
) -> None

Forget the values recorded for an item's fields that still hold the ones given.

A field recorded since with another value keeps it.

record_tracker_write async 

record_tracker_write(
    workspace: WorkspaceKey,
    item_id: str,
    *,
    within: timedelta,
) -> int

Record a write relayr is about to make to an item, by the tracker's id for it.

Returns how many it recorded within within, by the ledger's clock, this one included.

tracker_writes async 

tracker_writes(
    workspace: WorkspaceKey, *, within: timedelta
) -> dict[str, int]

Return how many writes relayr recorded to each item within within, by its id.

An item with none within it is left out.

acquire_tracker_lease async 

acquire_tracker_lease(
    workspace: WorkspaceKey, holder: str, ttl: timedelta
) -> bool

Take or renew the workspace's tracker lease for ttl; False if another holder has it.

It is the tracker follower's, apart from the artifactr follower's lease.

release_tracker_lease async 

release_tracker_lease(
    workspace: WorkspaceKey, holder: str
) -> None

Release the workspace's tracker lease, if holder has it.

tracker_following async 

tracker_following(
    workspace: WorkspaceKey,
) -> TrackerFollowing | None

Return how far the workspace's tracker follower has read, or None if it never has.

save_tracker_following async 

save_tracker_following(
    workspace: WorkspaceKey, following: TrackerFollowing
) -> None

Save how far the workspace's tracker follower has read.

Within one project its cursor, its sweep and its last read only move forward; another project replaces them.

snapshots async 

snapshots(
    workspace: WorkspaceKey,
    item_ids: Collection[str] | None = None,
) -> dict[str, Snapshot]

Return the snapshots of the items given, or of every item, by their ids.

save_snapshots async 

save_snapshots(
    workspace: WorkspaceKey, snapshots: Collection[Snapshot]
) -> None

Save items' snapshots, in place of those before.

forget_snapshots async 

forget_snapshots(
    workspace: WorkspaceKey,
    item_ids: Collection[str] | None = None,
) -> None

Forget the items given, or every item: their snapshots and relayr's writes to them.

tracker_dead_letter async 

tracker_dead_letter(
    workspace: WorkspaceKey, letter: TrackerDeadLetter
) -> None

Record a change the follower could not publish; once per event id, the first kept.

tracker_dead_letters async 

tracker_dead_letters(
    workspace: WorkspaceKey,
) -> list[TrackerDeadLetter]

Return the workspace's tracker dead letters, by event id.

Snapshot pydantic-model

Bases: BaseModel

An item of a bound project as the tracker follower last read it.

Config:

  • frozen: True

Fields:

item_id pydantic-field 

item_id: str

The tracker's id for the item.

item pydantic-field 

item: dict[str, JsonValue]

The item, as the tracker port reads it.

comments_after pydantic-field 

comments_after: datetime | None = None

The comments created before this were read; None if none were.

Tracked pydantic-model

Bases: BaseModel

A key a run writes to its workspace's tracker by, and what the write made.

Config:

  • frozen: True

Fields:

correlation_id pydantic-field 

correlation_id: str

The chain of the run that claimed the key first.

ref pydantic-field 

ref: dict[str, JsonValue] | None = None

The item or comment the key made, as the tracker port names it; None while it is only claimed.

deleted pydantic-field 

deleted: bool = False

Whether the item was found deleted: its key files nothing again.

written pydantic-field 

written: dict[str, JsonValue] = {}

The value relayr last set on each of the item's fields, recorded before the write.

TrackedKey

Bases: NamedTuple

A tracker key, by what it names.

TrackerDeadLetter pydantic-model

Bases: BaseModel

A change in a tracker the follower could not publish, and why.

Config:

  • frozen: True

Fields:

event_id pydantic-field 

event_id: str

The id the change's event was derived to have.

event_type pydantic-field 

event_type: str

The event's type, such as tracker:item_changed.

item_id pydantic-field 

item_id: str

The tracker's id for the item that changed.

error pydantic-field 

error: str

What went wrong: the exception's type and message.

TrackerFollowing pydantic-model

Bases: BaseModel

How far a workspace's tracker follower has read its bound project.

Config:

  • frozen: True

Fields:

project pydantic-field 

project: str

The tracker project it reads: another one starts over.

cursor pydantic-field 

cursor: tuple[datetime, str] | None = None

The updated_at and id of the last item it read, or None if the project had none.

swept_at pydantic-field 

swept_at: datetime

When it last read the whole project, by the follower's clock.

read_at pydantic-field 

read_at: datetime | None = None

When it last read the project at all, by the follower's clock; None if no read said.

WorkspaceKey

Bases: NamedTuple

A workspace, by its tenant's id and its own: the same ids in artifactr and reflexr.

InMemoryLedger 

InMemoryLedger(*, clock: Clock = utc_now)

A ledger in this process's memory, lost when it ends.

Parameters:

Name Type Description Default
clock Clock

Returns the current time, which leases expire by. Defaults to the system clock.

 utc_now

acquire_lease async 

acquire_lease(
    workspace: WorkspaceKey, holder: str, ttl: timedelta
) -> bool

Take or renew the workspace's lease for ttl. Return False if another holder has it.

release_lease async 

release_lease(workspace: WorkspaceKey, holder: str) -> None

Release the workspace's lease, if holder has it.

cursor async 

cursor(workspace: WorkspaceKey) -> int

Return the seq of the last envelope the workspace's follower is done with, or 0.

save_cursor async 

save_cursor(workspace: WorkspaceKey, seq: int) -> None

Save that the follower is done with seq. A cursor only moves forward.

dead_letter async 

dead_letter(
    workspace: WorkspaceKey, letter: DeadLetter
) -> None

Record an envelope the follower could not bridge; once per seq, the first kept.

dead_letters async 

dead_letters(workspace: WorkspaceKey) -> list[DeadLetter]

Return the workspace's dead letters, in seq order.

record_cause async 

record_cause(
    workspace: WorkspaceKey,
    kind: CauseKind,
    key: str,
    cause: Cause,
) -> None

Record the run behind a write, by what the follower will find it by; the first kept.

cause async 

cause(
    workspace: WorkspaceKey, kind: CauseKind, key: str
) -> Cause | None

Return the run behind a write, or None if no run of a rule made it.

forget_causes async 

forget_causes(
    workspace: WorkspaceKey, *, older_than: timedelta
) -> None

Forget the causes recorded longer ago than older_than, by the ledger's clock.

followed async 

followed() -> list[WorkspaceKey]

Return every workspace a follower has held a lease on or saved a cursor for, in order.

claim_tracked async 

claim_tracked(
    workspace: WorkspaceKey,
    kind: TrackedKind,
    key: str,
    correlation_id: str,
    *,
    stale_after: timedelta,
) -> bool

Claim a key before writing by it: True if the claim is the caller's.

tracked async 

tracked(
    workspace: WorkspaceKey, kind: TrackedKind, key: str
) -> Tracked | None

Return a key's claim and record, or None if it was never claimed.

every_tracked async 

every_tracked(
    workspace: WorkspaceKey,
) -> dict[TrackedKey, Tracked]

Return every key claimed in the workspace, and what each made, by kind, then key.

record_tracked async 

record_tracked(
    workspace: WorkspaceKey,
    kind: TrackedKind,
    key: str,
    ref: dict[str, JsonValue],
) -> dict[str, JsonValue] | None

Record what a claimed key made, the first kept, and return the record kept.

release_tracked async 

release_tracked(
    workspace: WorkspaceKey, kind: TrackedKind, key: str
) -> None

Release a key's claim that recorded nothing, after a write that certainly didn't land.

mark_tracked_deleted async 

mark_tracked_deleted(
    workspace: WorkspaceKey, key: str
) -> None

Mark an item's key deleted, so it files nothing again.

record_written async 

record_written(
    workspace: WorkspaceKey,
    key: str,
    fields: Mapping[str, JsonValue],
) -> None

Record the values relayr is about to set on an item's fields, by its key.

forget_written async 

forget_written(
    workspace: WorkspaceKey,
    key: str,
    fields: Collection[str],
) -> None

Forget the values recorded for an item's fields.

consume_written async 

consume_written(
    workspace: WorkspaceKey,
    key: str,
    values: Mapping[str, JsonValue],
) -> None

Forget the values recorded for an item's fields that still hold the ones given.

record_tracker_write async 

record_tracker_write(
    workspace: WorkspaceKey,
    item_id: str,
    *,
    within: timedelta,
) -> int

Record a write relayr is about to make to an item; return how many are within.

tracker_writes async 

tracker_writes(
    workspace: WorkspaceKey, *, within: timedelta
) -> dict[str, int]

Return how many writes relayr recorded to each item within within, by its id.

acquire_tracker_lease async 

acquire_tracker_lease(
    workspace: WorkspaceKey, holder: str, ttl: timedelta
) -> bool

Take or renew the workspace's tracker lease for ttl; False if another holder has it.

release_tracker_lease async 

release_tracker_lease(
    workspace: WorkspaceKey, holder: str
) -> None

Release the workspace's tracker lease, if holder has it.

tracker_following async 

tracker_following(
    workspace: WorkspaceKey,
) -> TrackerFollowing | None

Return how far the workspace's tracker follower has read, or None if it never has.

save_tracker_following async 

save_tracker_following(
    workspace: WorkspaceKey, following: TrackerFollowing
) -> None

Save how far the workspace's tracker follower has read; forward only, per project.

snapshots async 

snapshots(
    workspace: WorkspaceKey,
    item_ids: Collection[str] | None = None,
) -> dict[str, Snapshot]

Return the snapshots of the items given, or of every item, by their ids.

save_snapshots async 

save_snapshots(
    workspace: WorkspaceKey, snapshots: Collection[Snapshot]
) -> None

Save items' snapshots, in place of those before.

forget_snapshots async 

forget_snapshots(
    workspace: WorkspaceKey,
    item_ids: Collection[str] | None = None,
) -> None

Forget the items given, or every item: their snapshots and relayr's writes to them.

tracker_dead_letter async 

tracker_dead_letter(
    workspace: WorkspaceKey, letter: TrackerDeadLetter
) -> None

Record a change the follower could not publish; once per event id, the first kept.

tracker_dead_letters async 

tracker_dead_letters(
    workspace: WorkspaceKey,
) -> list[TrackerDeadLetter]

Return the workspace's tracker dead letters, by event id.

utc_now 

utc_now() -> datetime

The system clock, in UTC.

relayr.sql

The ledger in SQL: PostgreSQL or SQLite, through SQLAlchemy 2's asyncio extension.

Install relayr-ai[postgres] (asyncpg) or relayr-ai[sqlite] (aiosqlite). Upgrade the database with migrate, then give the followers a SqlLedger:

engine = create_async_engine("postgresql+asyncpg://localhost/app")
await migrate(engine)
ledger = SqlLedger(engine)

With row-level security, migrate as the tables' owner, granting a role that owns nothing the tables, and give the ledger that role's engine, and the owner's for the one read across tenants:

owner = create_async_engine("postgresql+asyncpg://owner:secret@localhost/app")
await migrate(owner, application_role="app")
engine = create_async_engine("postgresql+asyncpg://app:secret@localhost/app")
ledger = SqlLedger(engine, row_level_security=True, across_tenants=owner)

For SQLite, create the engine with create_sqlite_engine. create_schema creates the tables without migrations, for tests and prototypes.

SqlLedger 

SqlLedger(
    engine: AsyncEngine,
    *,
    clock: Clock = utc_now,
    row_level_security: bool = False,
    across_tenants: AsyncEngine | None = None,
)

A ledger in a SQL database, through SQLAlchemy's asyncio extension.

Create the tables with migrate first. The same code runs on PostgreSQL (with its default READ COMMITTED isolation) and on SQLite, whose engines must come from create_sqlite_engine.

With row_level_security, each transaction names its tenant to PostgreSQL, whose policies, which the migrations create, then show it only that tenant's rows (relayr.sql.tenancy). The one read that spans tenants, followed, which starts the followers, goes to across_tenants instead: an engine whose role the policies don't hold, such as the tables' owner.

Parameters:

Name Type Description Default
engine AsyncEngine

The database to use. The ledger does not dispose of it.

 required
clock Clock

Returns the current time, which leases expire by. Defaults to the system clock.

 utc_now
row_level_security bool

Whether each transaction names its tenant to PostgreSQL.

 False
across_tenants AsyncEngine | None

The engine for the read across tenants; required with row_level_security, and engine without it.

 None

Raises:

Type Description
ValueError

If row_level_security is on without across_tenants.

acquire_lease async 

acquire_lease(
    workspace: WorkspaceKey, holder: str, ttl: timedelta
) -> bool

Take or renew the workspace's lease for ttl. Return False if another holder has it.

release_lease async 

release_lease(workspace: WorkspaceKey, holder: str) -> None

Release the workspace's lease, if holder has it.

cursor async 

cursor(workspace: WorkspaceKey) -> int

Return the seq of the last envelope the workspace's follower is done with, or 0.

save_cursor async 

save_cursor(workspace: WorkspaceKey, seq: int) -> None

Save that the follower is done with seq. A cursor only moves forward.

The workspace's row is locked while it is compared, so saves from several processes leave the furthest.

dead_letter async 

dead_letter(
    workspace: WorkspaceKey, letter: DeadLetter
) -> None

Record an envelope the follower could not bridge; once per seq, the first kept.

dead_letters async 

dead_letters(workspace: WorkspaceKey) -> list[DeadLetter]

Return the workspace's dead letters, in seq order.

record_cause async 

record_cause(
    workspace: WorkspaceKey,
    kind: CauseKind,
    key: str,
    cause: Cause,
) -> None

Record the run behind a write, by what the follower will find it by; the first kept.

cause async 

cause(
    workspace: WorkspaceKey, kind: CauseKind, key: str
) -> Cause | None

Return the run behind a write, or None if no run of a rule made it.

forget_causes async 

forget_causes(
    workspace: WorkspaceKey, *, older_than: timedelta
) -> None

Forget the causes recorded longer ago than older_than, by the ledger's clock.

followed async 

followed() -> list[WorkspaceKey]

Return every workspace a follower has held a lease on or saved a cursor for, in order.

claim_tracked async 

claim_tracked(
    workspace: WorkspaceKey,
    kind: TrackedKind,
    key: str,
    correlation_id: str,
    *,
    stale_after: timedelta,
) -> bool

Claim a key before writing by it: True if the claim is the caller's.

A new key is inserted, once; a stale claim is taken again by one update, which a concurrent claim can't also make.

tracked async 

tracked(
    workspace: WorkspaceKey, kind: TrackedKind, key: str
) -> Tracked | None

Return a key's claim and record, or None if it was never claimed.

every_tracked async 

every_tracked(
    workspace: WorkspaceKey,
) -> dict[TrackedKey, Tracked]

Return every key claimed in the workspace, and what each made, by kind, then key.

record_tracked async 

record_tracked(
    workspace: WorkspaceKey,
    kind: TrackedKind,
    key: str,
    ref: dict[str, JsonValue],
) -> dict[str, JsonValue] | None

Record what a claimed key made, the first kept, and return the record kept.

release_tracked async 

release_tracked(
    workspace: WorkspaceKey, kind: TrackedKind, key: str
) -> None

Release a key's claim that recorded nothing, after a write that certainly didn't land.

mark_tracked_deleted async 

mark_tracked_deleted(
    workspace: WorkspaceKey, key: str
) -> None

Mark an item's key deleted, so it files nothing again.

record_written async 

record_written(
    workspace: WorkspaceKey,
    key: str,
    fields: Mapping[str, JsonValue],
) -> None

Record the values relayr is about to set on an item's fields, by its key.

The item's row is locked while it changes, so concurrent writes keep each other's.

forget_written async 

forget_written(
    workspace: WorkspaceKey,
    key: str,
    fields: Collection[str],
) -> None

Forget the values recorded for an item's fields.

consume_written async 

consume_written(
    workspace: WorkspaceKey,
    key: str,
    values: Mapping[str, JsonValue],
) -> None

Forget the values recorded for an item's fields that still hold the ones given.

The item's row is locked while it changes, so a value recorded meanwhile is kept.

record_tracker_write async 

record_tracker_write(
    workspace: WorkspaceKey,
    item_id: str,
    *,
    within: timedelta,
) -> int

Record a write relayr is about to make to an item; return how many are within.

The item's row is locked while it changes, so concurrent writes are each counted.

tracker_writes async 

tracker_writes(
    workspace: WorkspaceKey, *, within: timedelta
) -> dict[str, int]

Return how many writes relayr recorded to each item within within, by its id.

acquire_tracker_lease async 

acquire_tracker_lease(
    workspace: WorkspaceKey, holder: str, ttl: timedelta
) -> bool

Take or renew the workspace's tracker lease for ttl; False if another holder has it.

release_tracker_lease async 

release_tracker_lease(
    workspace: WorkspaceKey, holder: str
) -> None

Release the workspace's tracker lease, if holder has it.

tracker_following async 

tracker_following(
    workspace: WorkspaceKey,
) -> TrackerFollowing | None

Return how far the workspace's tracker follower has read, or None if it never has.

save_tracker_following async 

save_tracker_following(
    workspace: WorkspaceKey, following: TrackerFollowing
) -> None

Save how far the workspace's tracker follower has read; forward only, per project.

The follower's row is locked while it is compared, so saves from several processes leave the furthest.

snapshots async 

snapshots(
    workspace: WorkspaceKey,
    item_ids: Collection[str] | None = None,
) -> dict[str, Snapshot]

Return the snapshots of the items given, or of every item, by their ids.

save_snapshots async 

save_snapshots(
    workspace: WorkspaceKey, snapshots: Collection[Snapshot]
) -> None

Save items' snapshots, in place of those before, in one transaction.

Each is inserted, or updated in place where its row is there, so relayr's writes to the item, which the row keeps too, are left as they are.

forget_snapshots async 

forget_snapshots(
    workspace: WorkspaceKey,
    item_ids: Collection[str] | None = None,
) -> None

Forget the items given, or every item: their snapshots and relayr's writes to them.

tracker_dead_letter async 

tracker_dead_letter(
    workspace: WorkspaceKey, letter: TrackerDeadLetter
) -> None

Record a change the follower could not publish; once per event id, the first kept.

tracker_dead_letters async 

tracker_dead_letters(
    workspace: WorkspaceKey,
) -> list[TrackerDeadLetter]

Return the workspace's tracker dead letters, by event id.

create_schema async 

create_schema(engine: AsyncEngine) -> None

Create relayr's tables straight from the models, for tests and prototypes.

It skips Alembic, so the database records no schema version and cannot be upgraded with migrate later. Use migrate for databases that last.

migrate async 

migrate(
    engine: AsyncEngine,
    *,
    application_role: str | None = None,
) -> None

Upgrade the database to relayr's latest schema, creating the tables if needed.

Run it when the application deploys or starts, as the tables' owner. It is safe to run on every start: a database that is already up to date is left alone.

Parameters:

Name Type Description Default
engine AsyncEngine

The database, as the tables' owner.

 required
application_role str | None

A role that owns nothing, which the application connects as with row-level security on: it is granted relayr's tables, on PostgreSQL. It must exist.

 None

create_sqlite_engine 

create_sqlite_engine(
    url: str | URL, **kwargs: Any
) -> AsyncEngine

Create an async SQLite engine whose transactions take turns.

Every transaction on the engine, reads included, holds the database's write lock, so transactions run one at a time, on the engine's one connection. A transaction waits for the connection for up to the engine's pool_timeout (30 seconds), and for another process's lock for up to the timeout connect argument of Python's sqlite3 (5 seconds unless connect_args says otherwise). Use a database file (or a shared-cache memory database).

Parameters:

Name Type Description Default
url str | URL

A SQLite URL for an async driver, such as sqlite+aiosqlite:///app.db.

 required
**kwargs Any

Passed to SQLAlchemy's create_async_engine. Give poolclass (with its own arguments) to pool connections differently.

 {}

Returns:

Type Description
AsyncEngine

The engine. Dispose of it when you are done.

relayr.sql.tenancy

Row-level security: each transaction tells PostgreSQL its tenant (lattice RFC-0001).

With the option on, SqlLedger opens its sessions as TenantSession, which sets SETTING to its tenant, for the transaction only, as the transaction begins. The policies the migrations create on PostgreSQL then show each transaction only its tenant's rows. A pooled connection never carries a tenant into its next transaction, and a session that names no tenant can't begin one.

lattice-core holds the tenant session, the policy and the grants, which every library's SQL storage shares (lattice ADR-0018); this module re-exports them.

CURRENT module-attribute 

CURRENT: Final = (
    f"current_setting('{TENANT_SETTING}', true)"
)

The transaction's tenant, in SQL: null, or empty, when it names none, which no row's is.

TENANT module-attribute 

TENANT: Final = 'tenant_id'

The key of a TenantSession's tenant in its info.

SETTING module-attribute 

SETTING: Final = 'lattice.tenant_id'

The PostgreSQL setting that holds a transaction's tenant.

TenantSession

Bases: Session

A session that tells PostgreSQL its tenant as each transaction begins.

It sets TENANT_SETTING to the tenant in its info. SQLite has no row-level security, so there it sets nothing, but a session must still name its tenant.

grant 

grant(
    connection: Connection,
    role: str,
    tables: Collection[str],
) -> None

Let a role use some tables as storage does, and their sequences, on PostgreSQL.

It grants the role USAGE on the current schema, reading and writing each table, and USAGE on the sequences the tables own. Granting again changes nothing.

tenant_policy 

tenant_policy(
    table: str, column: str = "tenant_id"
) -> tuple[str, ...]

The DDL that keeps a table's rows to their tenant, on PostgreSQL.

table and column go into the DDL as they are, unquoted, unlike grant's role and tables: pass only trusted identifiers, such as a model's own table name. The policy is enabled, not forced, so the tables' owner isn't held by it. A transaction sees, changes and adds only rows whose column is its tenant; one that names none, or an empty one, sees nothing.

relayr.inbound

The inbound adapter: a follower per workspace, bridging artifactr's events into reflexr.

followers = Followers(collaboration, automation, ledger, projections=projections)
await followers.resume()  # every workspace the ledger knows
...
await followers.aclose()

The followers start from the libraries' on_open hook, followers.watch, which the application composes with its other ones, such as the feedback mirrors'.

FOLLOWER module-attribute 

FOLLOWER: Final = artifactr.SystemActor(name='relayr')

Who a follower reads artifactr's log as.

SOURCE module-attribute 

SOURCE: Final = SourceActor(name='artifactr')

Who a follower publishes into reflexr as: the actor on every bridged envelope.

Followers 

Followers(
    artifactr: Workspaces,
    reflexr: Workspaces,
    ledger: Ledger,
    *,
    projections: Projections | None = None,
    holder: str | None = None,
    lease_ttl: timedelta = timedelta(seconds=30),
    save_every: int = 500,
    cause_ttl: timedelta = timedelta(days=7),
    tracer_provider: TracerProvider | None = None,
    meter_provider: MeterProvider | None = None,
)

Follows each workspace the application uses, from the first time it is opened.

Give watch to both libraries' Workspaces as their on_open, composed with any other, and call resume at startup to follow again every workspace the ledger knows.

Parameters:

Name Type Description Default
artifactr Workspaces

The workspaces whose logs are followed.

 required
reflexr Workspaces

The workspaces the events are published into. Their events must include BRIDGED, and the types the projections publish: a type they refuse stops the follower.

 required
ledger Ledger

Where each follower keeps its lease, cursor and dead letters.

 required
projections Projections | None

What each envelope becomes. Defaults to the default set alone.

 None
holder str | None

This process's name in leases. Defaults to a new random id; give each process its own.

 None
lease_ttl timedelta

How long a lease lasts unless renewed. A follower renews it every third of that, and a process that dies holding one lets another take it once it lapses.

 timedelta(seconds=30)
save_every int

How many envelopes that bridge nothing a follower reads before it saves its cursor anyway, so a restarted follower reads at most this many again.

 500
cause_ttl timedelta

How long the ledger keeps the run behind each of the outbound adapter's writes. A follower forgets older ones at most hourly while it holds its lease, once it has caught up with the log; a write bridged after its cause is forgotten starts a chain of its own.

 timedelta(days=7)
tracer_provider TracerProvider | None

Where the bridge's spans go. Defaults to the global provider.

 None
meter_provider MeterProvider | None

Where its counters go. Defaults to the global provider.

 None

watch async 

watch(tenant_id: str, workspace_id: str) -> None

Follow a workspace, unless it is followed already: the libraries' on_open hook.

It is called each time a workspace is opened, its follower's own openings included, so it does nothing for a workspace it follows already, nor once the followers are closed.

resume async 

resume() -> None

Follow every workspace the ledger knows: call it when the application starts.

aclose async 

aclose() -> None

Stop every follower, releasing its lease, and start no more.

relayr.outbound

The outbound adapter: the runs of rules act in artifactr, each as its rule's own participant.

A function action acts through Outbound.acting, an agent through ArtifactrCapability:

outbound = Outbound(collaboration, ledger, allowlists={"ops:timeline": Allowlist(direct=...)})


async def log_deploy(reaction: Reaction[Deps]) -> None:
    acting = reaction.deps.outbound.acting(reaction)
    await acting.edit(timeline_id, f"/entries/{reaction.run_id}", "deployed")
    await acting.notice("Added to the timeline.")

Writes are proposals and notices by default, in the threads of the events that made the rule fire; a rule's Allowlist grants more.

RULES module-attribute 

RULES: Final = 'reflexr:'

The prefix of a rule's actor's client_id in artifactr: reflexr:<rule>.

NotAllowed 

NotAllowed(message: str)

Bases: RunFailure

The rule's allowlist doesn't allow what its run tried. Permanent: a retry would try again.

Outbound 

Outbound(
    artifactr: Workspaces,
    ledger: Ledger,
    *,
    allowlists: Mapping[str, Allowlist] | None = None,
    runner: Runner[Any] | None = None,
)

Lets the runs of rules act in artifactr, each within its rule's allowlist.

Give it to the actions that act in artifactr, as their dependency or through ArtifactrCapability.

Parameters:

Name Type Description Default
artifactr Workspaces

The workspaces runs act in: the same tenant and id as the run's.

 required
ledger Ledger

Where the run behind each write is recorded, for the follower.

 required
allowlists Mapping[str, Allowlist] | None

Each rule's allowlist, by name, or by a whole namespace's, as "chat:*", for rules installed at runtime. A rule's own name comes first; a rule with neither gets the default.

 None
runner Runner[Any] | None

Starts the turns that rules allowed to send ask for. Without one, no rule may.

 None

Raises:

Type Description
ValueError

If a key of allowlists is neither a rule's name nor a whole namespace's, or names reflexr's own namespace.

acting 

acting(reaction: Reaction[Any]) -> RunActions

Return what an attempt of a run may do in artifactr; the same for the whole attempt.

Refused 

Refused(rejection: Rejection)

Bases: RunFailure

artifactr refused a write as invalid, or about something missing.

An agent's tools hand it back to the model, which can do otherwise; from a function action, it fails the attempt, which is retried. Its reason is artifactr's code: validation_failed, patch_failed or not_found.

RunActions 

RunActions(
    reaction: Reaction[Any],
    workspaces: Workspaces,
    ledger: Ledger,
    allowlist: Allowlist,
    runner: Runner[Any] | None,
)

What one attempt of a run does in artifactr. Get it with Outbound.acting.

A write without a thread goes to the chain's first thread: that of the first event that made the rule fire with one, or else the first the run created.

create async 

create(
    artifact: Artifact,
    *,
    thread_id: ThreadId | None = None,
    rationale: str | None = None,
) -> ArtifactId

Create an artifact, by proposal unless the allowlist lets the rule create its kind.

Returns:

Type Description
ArtifactId

The artifact's id, derived from the run, the kind and the data.

Raises:

Type Description
Refused

If artifactr finds the artifact invalid.

edit async 

edit(
    artifact_id: ArtifactId,
    path: str,
    value: JsonValue,
    *,
    summary: str | None = None,
    thread_id: ThreadId | None = None,
    rationale: str | None = None,
) -> None

Set the value at a JSON Pointer path in an artifact: a keyed edit.

It is a proposal unless the allowlist lets the rule edit the artifact's kind. It is skipped when the artifact already holds the value there, and re-read and made again, within the attempt, when the artifact changed meanwhile. A direct edit retried after someone else changed the value sets the run's value again: a direct change carries no id to tell an earlier attempt's from another's.

Raises:

Type Description
RunFailure

If the artifact kept changing, so the attempt is retried; or, for good, if it is archived.
Refused

If the artifact is missing, or the patch doesn't apply.

archive async 

archive(
    artifact_id: ArtifactId,
    *,
    thread_id: ThreadId | None = None,
    rationale: str | None = None,
) -> None

Archive an artifact, by proposal unless the allowlist lets the rule edit its kind.

It is skipped when the artifact is archived already, and re-read and made again, within the attempt, when the artifact changed meanwhile, as edit is.

Raises:

Type Description
RunFailure

If the artifact kept changing, so the attempt is retried.
Refused

If the artifact is missing.

notice async 

notice(
    content: str, *, thread_id: ThreadId | None = None
) -> MessageId

Post a notice: for people, starting and steering no turn.

Parameters:

Name Type Description Default
content str

What it says.

 required
thread_id ThreadId | None

Where; by default the chain's first thread.

 None

Returns:

Type Description
MessageId

The notice's message id, derived from the run, the thread and the text.

create_thread async 

create_thread(title: str = '') -> ThreadId

Create a thread, if the allowlist lets the rule, and act in it from then on.

Returns:

Type Description
ThreadId

The thread's id, derived from the run and the title.

send async 

send(
    content: str, *, thread_id: ThreadId | None = None
) -> MessageId

Send a message that starts the thread's next agent turn, if the allowlist lets it.

Returns:

Type Description
MessageId

The message's id, derived from the run, the thread and the text.

read async 

read(artifact_id: ArtifactId) -> Versioned[Artifact]

Return an artifact's current version, as the rule's actor reads it.

Raises:

Type Description
Refused

If there is no such artifact.

Allowlist pydantic-model

Bases: BaseModel

What a rule may do in artifactr, and in its workspace's tracker.

The default allows proposals and notices, in the threads of the events that made it fire, and nothing more.

Config:

  • frozen: True

Fields:

direct pydantic-field 

direct: frozenset[str]

The artifact kinds the rule creates and edits directly rather than by proposal. artifactr's write policy and the thread's mode still apply.

threads pydantic-field 

threads: Literal['chain', 'any'] = 'chain'

Where the rule may act: chain, the threads of the events that made it fire and those it creates, or any thread in the workspace.

create_threads pydantic-field 

create_threads: bool = False

Whether the rule may create threads.

turns pydantic-field 

turns: bool = False

Whether the rule may send messages that start an agent turn, rather than only notices.

trackers pydantic-field 

trackers: bool = False

Whether the rule may file, comment on and change items in its workspace's tracker (lattice RFC-0004).

ArtifactrCapability dataclass 

ArtifactrCapability(outbound: Outbound)

Bases: AbstractCapability[Reaction[Any]]

Give a rule's agent tools to read and write artifacts, post notices, and more if allowed.

Parameters:

Name Type Description Default
outbound Outbound

The adapter the tools act through.

 required

get_toolset 

get_toolset() -> FunctionToolset[Reaction[Any]]

Return the tools for acting in artifactr.

get_instructions 

get_instructions() -> str

Return the instructions: how the agent's writes reach people.

is_rule_actor 

is_rule_actor(actor: Actor) -> bool

Whether an artifactr actor is a rule's: the actor a rule's runs act as, for no person.

rule_actor 

rule_actor(rule: RuleName) -> ExternalAgentActor

Who a rule's runs act as in artifactr: an external agent of their own, for no person.

turn_cause async 

turn_cause(
    ledger: Ledger,
    workspace: WorkspaceKey,
    message_ids: Sequence[MessageId],
) -> Cause | None

Return the run behind a turn, or None if no run's message is among those it carries out.

Parameters:

Name Type Description Default
ledger Ledger

Where the run behind each message is recorded.

 required
workspace WorkspaceKey

The turn's workspace.

 required
message_ids Sequence[MessageId]

The messages the turn carries out, oldest first, as its run_started names them.

 required

relayr.trackers

The tracker connector (lattice RFC-0004): rules' runs file and change items in a task tracker.

It sits behind one port, Tracker, with adapters for Plane's Community edition in relayr.plane and Vikunja in relayr.vikunja, and one in memory, FakeTracker. A function action writes through Trackers.acting, an agent through TrackerCapability:

trackers = Trackers(
    ledger,
    connect=plane(http),
    integration=relayr.plane.INTEGRATION,
    credentials=tenant_credentials,
    bindings={WorkspaceKey("acme", "ops"): TrackerBinding(project=OPS, labels=("lattice",))},
    allowlists={"ops:dead-letters": Allowlist(trackers=True)},
)


async def file_dead_letter(reaction: Reaction[Deps]) -> None:
    event = reaction.events[0].event
    assert isinstance(event, RunDeadLettered)
    actions = reaction.deps.trackers.acting(reaction)
    item = await actions.ensure_item(
        ItemDraft(title=f"{event.rule} failed", priority="high"), key=f"dead-letter/{event.rule}"
    )
    if not isinstance(item, Deleted):
        await actions.comment(item, f"<p>Retried {event.attempts} times.</p>")

Every write is made once, by a key the ledger claims before anything is sent.

What people change in a bound project comes back as tracker: events (relayr.trackers.events), published by TrackerFollowers, one per bound workspace, which poll the project and diff what they find against the ledger; a webhook only makes one poll now (relayr.fastapi.tracker_router):

followers = TrackerFollowers(trackers, automation)
await followers.resume()  # every bound workspace; `followers.watch` is reflexr's on_open
app.include_router(tracker_router(followers))

CLAIM_TTL module-attribute 

CLAIM_TTL: Final = timedelta(seconds=30)

How long a key's claim holds without a record before another attempt may take it again: longer than WRITING, which bounds the writes made under it.

CREDENTIALS_TTL module-attribute 

CREDENTIALS_TTL: Final = timedelta(minutes=5)

How long a tenant's credentials are used before they are asked for again; a key the tracker refuses asks again at once.

LOOP_CAP module-attribute 

LOOP_CAP: Final = 20

How many writes relayr makes to one item within LOOP_WINDOW: the next fails its run for good, as tracker_loop. It backs the echo filter, for what the filter can't see.

LOOP_WINDOW module-attribute 

LOOP_WINDOW: Final = timedelta(hours=1)

What LOOP_CAP is per.

WRITING module-attribute 

WRITING: Final = timedelta(seconds=20)

How long a claim's writes may take, all told: well inside the claim, so a claim taken again never overlaps a write relayr is still making. A request the tracker finishes after relayr gave up on it can still land later.

LOOK_BACK module-attribute 

LOOK_BACK: Final = timedelta(minutes=2)

How far before its cursor a poll reads: a change can commit with an updated_at older than one already read.

Connect 

Connect = Callable[[TrackerCredentials, str], Tracker]

Makes a lattice workspace's adapter from its tenant's credentials and its source: plane(http), vikunja(http), or the fake's connect.

Credentials 

Credentials = Callable[[str], Awaitable[TrackerCredentials]]

Returns a tenant's tracker credentials, from wherever the deployment keeps them.

CommentKey 

CommentKey = str

What lattice posts a comment by, as an item's key.

ItemKey 

ItemKey = str

What lattice files an item by: Plane's external id, or Vikunja's marker.

Priority 

Priority = Literal[
    "none", "low", "medium", "high", "urgent"
]

How urgent an item is.

ProjectId 

ProjectId = str

A project in the tracker, by the tracker's own id.

State 

State = Literal[
    "backlog", "todo", "started", "done", "cancelled"
]

Where an item is. Plane's state groups map one to one, unstarted as todo; Vikunja's done flag reads as done or todo, and only done and cancelled set it.

Deleted pydantic-model

Bases: BaseModel

What ensure_item returns for a key whose item a person deleted: it files nothing.

Config:

  • frozen: True

Fields:

TrackerActions 

TrackerActions(
    reaction: Reaction[Any],
    trackers: Trackers,
    binding: TrackerBinding | None,
    allowlist: Allowlist,
    telemetry: Telemetry,
)

What one attempt of a run does in its workspace's tracker. Get it with Trackers.acting.

Every method needs the rule's allowlist to allow trackers, the workspace to be bound, and the integration to be enabled there; otherwise it is NotAllowed, before the ledger or the tracker is touched. Each fails the attempt, to be retried, when the tracker is unavailable or rate-limited; raises TrackerRefused when it refuses a write as invalid or about something missing; and fails the run for good when it refuses the tenant's key.

ensure_item async 

ensure_item(
    draft: ItemDraft, *, key: ItemKey | None = None
) -> ItemRef | Deleted

File an item, once: or return the one its key filed, unless a person deleted it.

The key is claimed in the ledger before the tracker is asked for it, then the tracker is asked for an item the key filed, and only then is one created. An item filed earlier is read first: a closed or archived one is returned, for the rule to reopen if it will; a deleted one is Deleted, and files nothing.

Parameters:

Name Type Description Default
draft ItemDraft

The item, with the binding's defaults for what it leaves unset.

 required
key ItemKey | None

What the item is filed by. By default it is derived from the run and the draft; a key of the rule's own, as dead-letter/<rule>, files later runs into the same item.

 None

Raises:

Type Description
RunFailure

If another attempt claimed the key within the last 30 s and hasn't recorded its item, so the attempt is retried. A write the tracker refused releases the key's claim, so the rule may file by it again at once.

comment async 

comment(
    item: ItemRef,
    body: str,
    *,
    key: CommentKey | None = None,
) -> CommentRef

Post a comment on an item, once.

Parameters:

Name Type Description Default
item ItemRef

The item.

 required
body str

What it says, as HTML.

 required
key CommentKey | None

What the comment is posted by; by default derived from the run, the item and the text.

 None

set_state async 

set_state(item: ItemRef, state: State) -> None

Move an item to a state, unless it is there: a keyed edit, as update makes.

update async 

update(item: ItemRef, fields: ItemFields) -> None

Set the fields given on an item lattice filed, unless each holds its value already.

The item is read, and only the fields that differ are sent. Their values are recorded on the item's key before they are, so what comes back from the tracker can be told from a person's change; a change the tracker refuses puts back what was recorded before it.

Raises:

Type Description
RunFailure

For good, if the item wasn't filed by lattice, which has no key for it.
TrackerRefused

If the item was deleted, or the tracker refuses the change.

read async 

read(item: ItemRef) -> Item | None

Return an item's snapshot, archived or not, or None if it was deleted.

filed async 

filed(key: ItemKey) -> ItemRef | None

Return the item a key filed, or None if it filed none, or a person deleted it.

link(item: ItemRef) -> str

Return the item's address in the tracker's web app.

TrackerBinding pydantic-model

Bases: BaseModel

The tracker project a workspace's runs file into, and whose changes its follower reads.

Config:

  • frozen: True

Fields:

state pydantic-field 

state: State | None = None

Where an item starts unless the rule says; by default the tracker's default state.

priority pydantic-field 

priority: Priority | None = None

An item's priority unless the rule says; by default none.

labels pydantic-field 

labels: tuple[str, ...] = ()

Labels every item filed has, beside the rule's.

since pydantic-field 

since: AwareDatetime | None = None

When its follower first reads the project, the items changed since are published as created; by default none is, and only what changes after is published.

TrackerRefused 

TrackerRefused(error: Refused | NotFound)

Bases: RunFailure

The tracker refused a write as invalid, or about something missing.

An agent's tools hand it back to the model, which can do otherwise; from a function action, it fails the attempt, which is retried. Its reason is tracker_refused or tracker_not_found.

Trackers 

Trackers(
    ledger: Ledger,
    *,
    connect: Connect,
    credentials: Credentials,
    bindings: Mapping[WorkspaceKey, TrackerBinding],
    allowlists: Mapping[str, Allowlist] | None = None,
    integration: str,
    integrations: Integrations = ALWAYS,
    budget: int = 40,
    clock: Clock = utc_now,
    meter_provider: MeterProvider | None = None,
)

Lets the runs of rules write to their workspace's tracker, within their allowlists.

Give it to the actions that file items, as their dependency or through TrackerCapability.

Parameters:

Name Type Description Default
ledger Ledger

Where keys are claimed before each write and what they made is recorded.

 required
connect Connect

Makes a workspace's adapter, as plane(http) or vikunja(http) does.

 required
credentials Credentials

Returns a tenant's credentials; asked again every five minutes, and after the tracker refuses a key.

 required
bindings Mapping[WorkspaceKey, TrackerBinding]

The tracker project each workspace files into. A workspace with none files nothing.

 required
allowlists Mapping[str, Allowlist] | None

Each rule's allowlist, as Outbound takes them. A rule writes to its tracker only if its allowlist's trackers is set.

 None
integration str

The tracker's key among the application's integrations: relayr.plane.INTEGRATION or relayr.vikunja.INTEGRATION.

 required
integrations Integrations

Whether the integration runs in a workspace, asked before anything is done there: lattice-core's port, such as grantr's Grants. By default it runs everywhere.

 ALWAYS
budget int

How many requests a minute each tenant's key may make from this process, all told: runs' writes spend it first, and the followers' polls take what is left. Plane allows 60 by default.

 40
clock Clock

Returns the current time, which credentials expire and the budget is measured by.

 utc_now
meter_provider MeterProvider | None

Where the count of writes refused by the cap on loops goes. Defaults to the global provider.

 None

Raises:

Type Description
ValueError

If a key of allowlists is neither a rule's name nor a whole namespace's.

ledger property 

ledger: Ledger

Where keys are claimed, and what the followers read is kept.

integration property 

integration: str

The tracker's key among the application's integrations, such as tracker:plane.

bindings property 

bindings: Mapping[WorkspaceKey, TrackerBinding]

The tracker project each bound workspace files into and reads.

budget property 

budget: Budget

The requests each tenant's key may make from this process a minute.

acting 

acting(reaction: Reaction[Any]) -> TrackerActions

Return what an attempt of a run may do in its tracker; the same for the attempt.

enabled async 

enabled(workspace: WorkspaceKey) -> bool

Whether the integration runs in a workspace: asked before anything is done there.

connection async 

connection(
    workspace: WorkspaceKey, *, fresh: bool = False
) -> Tracker

A workspace's adapter; with its tenant's credentials asked for again if fresh.

connected 

connected(workspace: WorkspaceKey) -> Tracker | None

A workspace's adapter if one was made, without asking for credentials.

Budget 

Budget(per_minute: int, clock: Clock)

Each tenant's requests in the last minute, in this process.

Parameters:

Name Type Description Default
per_minute int

How many requests a minute polls may bring a tenant's spending to.

 required
clock Clock

Returns the current time, which the minute is measured by.

 required

spend 

spend(tenant: str) -> None

Count a request a run makes: it is made whatever the budget has left.

take 

take(tenant: str) -> bool

Count a request a poll makes, if the budget has one left; return whether it had.

TrackerCapability dataclass 

TrackerCapability(trackers: Trackers)

Bases: AbstractCapability[Reaction[Any]]

Give a rule's agent tools to file, comment on, move and read items in its tracker.

Parameters:

Name Type Description Default
trackers Trackers

The connector the tools write through.

 required

get_toolset 

get_toolset() -> FunctionToolset[Reaction[Any]]

Return the tools for writing to the tracker.

get_instructions 

get_instructions() -> str

Return the instructions: how the agent names the items it files.

FakeConnection 

FakeConnection(
    store: _Store,
    credentials: TrackerCredentials,
    source: str,
)

A lattice workspace's adapter to a FakeTracker.

user instance-attribute 

user = f'relayr:{source}'

Who its writes are by.

find async 

find(project: ProjectId, key: ItemKey) -> ItemRef | None

Return the item lattice filed in a project by a key, or None if there is none.

create async 

create(
    project: ProjectId, key: ItemKey, draft: ItemDraft
) -> ItemRef

File an item by a key; or, if the tracker dedupes, return the one the key filed.

item async 

item(ref: ItemRef) -> Item | None

Return an item's snapshot, archived or not, or None if it was deleted.

update async 

update(ref: ItemRef, fields: ItemFields) -> None

Set the fields given on an item.

comment async 

comment(
    ref: ItemRef, key: CommentKey, body: str
) -> CommentRef

Post a comment by a key, or return the one the key already posted.

items async 

items(
    project: ProjectId, after: Cursor | None
) -> Page[Item]

Return a project's listed items changed after a cursor, oldest first.

comments async 

comments(
    ref: ItemRef, after: CommentRef | None
) -> list[Comment]

Return an item's comments after one, oldest first, or all of them.

nudge 

nudge(
    headers: Mapping[str, str], body: bytes
) -> Nudge | None

Verify a nudge's signature, and say which project to poll.

link(ref: ItemRef) -> str

Return the item's address.

holds 

holds(state: State) -> State

Every state, as Plane holds them.

FakeTracker 

FakeTracker(
    *,
    api_key: str = "key",
    dedupes: bool = False,
    clock: Clock = utc_now,
)

A tracker workspace in memory. Give connect to Trackers.

Parameters:

Name Type Description Default
api_key str

The key it accepts; a request with another is Unauthorized.

 'key'
dedupes bool

Whether a create by a key already used returns its item, as Plane's 409 does, rather than filing another, as Vikunja does.

 False
clock Clock

Returns the current time, which items are updated at.

 utc_now

created property 

created: int

How many items were filed, by lattice and by people.

connect 

connect(
    credentials: TrackerCredentials, source: str
) -> FakeConnection

Connect as a lattice workspace's adapter does, with the tenant's credentials.

rotate 

rotate(api_key: str) -> None

Accept another key, and refuse the one before.

fail 

fail(*errors: TrackerError | None) -> None

Fail the next requests, one error each; None lets one through.

lose_next_created 

lose_next_created() -> None

Make the next item lattice files, and lose the answer, as Unavailable.

add 

add(
    project: ProjectId,
    title: str,
    *,
    by: str | None = PERSON,
) -> ItemRef

File an item, as a person does, with no key.

edit 

edit(
    ref: ItemRef,
    fields: ItemFields,
    *,
    by: str | None = PERSON,
) -> None

Change an item, as a person does.

reply 

reply(
    ref: ItemRef, body: str, *, by: str | None = PERSON
) -> CommentRef

Comment on an item, as a person does.

archive 

archive(ref: ItemRef) -> None

Archive an item: it is read, but no longer listed.

delete 

delete(ref: ItemRef) -> None

Delete an item, as a person does: its key files again.

items_in 

items_in(project: ProjectId) -> list[Item]

Every item in a project, archived ones included, in the order they were filed.

TrackerFollowers 

TrackerFollowers(
    trackers: Trackers,
    reflexr: Workspaces,
    *,
    projections: Mapping[WorkspaceKey, TrackerProjection]
    | None = None,
    holder: str | None = None,
    lease_ttl: timedelta = timedelta(seconds=30),
    interval: timedelta = timedelta(seconds=30),
    sweep_every: timedelta = timedelta(days=1),
    clock: Clock = utc_now,
    tracer_provider: TracerProvider | None = None,
    meter_provider: MeterProvider | None = None,
)

Follows the tracker project each bound workspace reads, from when it is first opened.

Give watch to reflexr's Workspaces as its on_open, composed with any other, and call resume at startup, so every bound project is read whether or not its workspace is opened.

Parameters:

Name Type Description Default
trackers Trackers

The connector: its bindings, credentials, budget and integration switch.

 required
reflexr Workspaces

The workspaces the events are published into. Their events must include TRACKED and the types the projections return, and their publishers should reserve the namespace with PUBLISHERS.

 required
projections Mapping[WorkspaceKey, TrackerProjection] | None

What a bound workspace's events become, in place of the tracker: events themselves.

 None
holder str | None

This process's name in leases. Defaults to a new random id.

 None
lease_ttl timedelta

How long a lease lasts unless renewed; it is renewed every third of that.

 timedelta(seconds=30)
interval timedelta

How long a follower waits between polls, unless a webhook nudges it.

 timedelta(seconds=30)
sweep_every timedelta

How often a follower reads the whole project.

 timedelta(days=1)
clock Clock

Returns the current time, which sweeps are due by.

 utc_now
tracer_provider TracerProvider | None

Where the polls' spans go. Defaults to the global provider.

 None
meter_provider MeterProvider | None

Where the counters go. Defaults to the global provider.

 None

watch async 

watch(tenant_id: str, workspace_id: str) -> None

Follow a bound workspace's project, unless it is followed already: the on_open hook.

A workspace with no binding is left alone.

resume async 

resume() -> None

Follow every bound workspace's project: call it when the application starts.

poll async 

poll(tenant_id: str, workspace_id: str) -> bool

Read a bound workspace's project once, now, as its follower does each interval.

It takes the workspace's lease, and leaves it held until it lapses. Returns whether it read: not if the workspace is unbound, its integration is off, or another process holds its lease. A tracker's failure is logged, and read past.

nudge async 

nudge(
    tenant_id: str,
    tracker: str,
    headers: Mapping[str, str],
    body: bytes,
) -> bool

Verify a webhook, and wake the followers this process runs of the project it names.

The tenant's secret comes from the credentials its followers last asked for, never from the request: a tenant whose follower hasn't read yet, a switched-off workspace and a bad signature are refused alike, and counted together.

Returns whether the webhook was verified.

aclose async 

aclose() -> None

Stop every follower, releasing its lease, and start no more.

Comment pydantic-model

Bases: BaseModel

A comment on an item.

Config:

  • frozen: True

Fields:

body pydantic-field 

body: str

Its text as HTML, without Vikunja's marker.

author pydantic-field 

author: str | None

The tracker's id for whoever wrote it, if the tracker says.

key pydantic-field 

key: CommentKey | None = None

The key lattice posted it by, or None if lattice didn't.

CommentRef pydantic-model

Bases: BaseModel

A comment on an item.

Config:

  • frozen: True

Fields:

item pydantic-field 

item: str

The item's id.

id pydantic-field 

id: str

The tracker's id for the comment.

Cursor pydantic-model

Bases: BaseModel

Where a read of a project's items stopped: the last item's updated_at, then id.

Config:

  • frozen: True

Fields:

Item pydantic-model

Bases: BaseModel

A snapshot of an item.

Config:

  • frozen: True

Fields:

description pydantic-field 

description: str

Its body as HTML, without Vikunja's marker.

labels pydantic-field 

labels: tuple[str, ...]

Label names, sorted.

assignees pydantic-field 

assignees: tuple[str, ...]

The tracker's ids for the people it is assigned to, sorted.

updated_by pydantic-field 

updated_by: str | None = None

The tracker's id for whoever changed it last, if the tracker says.

ItemDraft pydantic-model

Bases: BaseModel

What a rule files: its title and body, and what it leaves unset takes the binding's.

Config:

  • frozen: True

Fields:

description pydantic-field 

description: str = ''

The item's body, as HTML, which both trackers keep.

state pydantic-field 

state: State | None = None

Where it starts; by default the binding's, or the tracker's default state.

priority pydantic-field 

priority: Priority | None = None

How urgent it is; by default the binding's, or none.

labels pydantic-field 

labels: tuple[str, ...] = ()

Label names, beside the binding's. Each is found or created by name.

ItemFields pydantic-model

Bases: BaseModel

The fields a change sets; those left None are left as they are.

Config:

  • frozen: True

Fields:

ItemRef pydantic-model

Bases: BaseModel

An item in a tracker project: enough to read it, change it and link to it.

Config:

  • frozen: True

Fields:

id pydantic-field 

id: str

The tracker's id for the item.

number pydantic-field 

number: int

The item's number in its project.

name pydantic-field 

name: str

What people call it, as the tracker shows it: OPS-12.

key pydantic-field 

key: ItemKey | None = None

The key lattice filed it by, or None if lattice didn't.

NotFound

Bases: TrackerError

What a request named isn't there.

Nudge pydantic-model

Bases: BaseModel

A verified webhook's news: a project changed, so poll it now.

Config:

  • frozen: True

Fields:

Page pydantic-model

Bases: BaseModel

What a read found after a cursor, oldest first, and the cursor after it.

Config:

  • frozen: True

Fields:

next pydantic-field 

next: Cursor | None

The cursor after the last item; None if there was none.

RateLimited 

RateLimited(retry_after: float | None)

Bases: TrackerError

The tracker's rate limit is reached; try again after retry_after seconds, if it says.

Refused 

Refused(reason: str)

Bases: TrackerError

The tracker refused a request as invalid, for reason.

Tracker

Bases: Protocol

One tenant's tracker workspace, as one of the tenant's lattice workspaces writes to it.

An adapter is made from the tenant's credentials and a source, lattice:<tenant>/<workspace>, which tells its keys from another lattice workspace's. Every method raises a TrackerError when the tracker refuses or fails a request.

find async 

find(project: ProjectId, key: ItemKey) -> ItemRef | None

Return the item lattice filed in a project by a key, or None if there is none.

create async 

create(
    project: ProjectId, key: ItemKey, draft: ItemDraft
) -> ItemRef

File an item by a key, with the draft's values; labels are found or created by name.

Plane returns an item its external ids already name, rather than making another.

item async 

item(ref: ItemRef) -> Item | None

Return an item's snapshot, or None if it was deleted; an archived item is returned.

update async 

update(ref: ItemRef, fields: ItemFields) -> None

Set the fields given on an item.

comment async 

comment(
    ref: ItemRef, key: CommentKey, body: str
) -> CommentRef

Post a comment on an item by a key, or return the one the key already posted.

items async 

items(
    project: ProjectId, after: Cursor | None
) -> Page[Item]

Return a project's items changed after a cursor, by updated_at then id, oldest first.

It reads what the tracker lists, which on Plane leaves out archived, draft and triage items.

comments async 

comments(
    ref: ItemRef, after: CommentRef | None
) -> list[Comment]

Return an item's comments after one, oldest first, or all of them.

nudge 

nudge(
    headers: Mapping[str, str], body: bytes
) -> Nudge | None

Verify a webhook against the tenant's secret, and say which project to poll.

Returns None for anything it can't verify.

link(ref: ItemRef) -> str

Return the item's address in the tracker's web app.

holds 

holds(state: State) -> State

Return the state an item set to state reads back as.

Plane holds every state; Vikunja's done flag holds two of the five.

TrackerCredentials pydantic-model

Bases: BaseModel

How to reach one tenant's tracker workspace. Never logged, recorded or traced.

Config:

  • frozen: True

Fields:

base_url pydantic-field 

base_url: str

The tracker's URL, such as https://plane.example.com.

api_key pydantic-field 

api_key: str

The key of the tenant's own tracker user.

webhook_secret pydantic-field 

webhook_secret: str

What the tracker signs its webhooks with.

workspace pydantic-field 

workspace: str = ''

The tenant's workspace slug, on Plane.

TrackerError

Bases: Exception

A tracker refused or failed a request.

Unauthorized

Bases: TrackerError

The tracker refused the key.

Unavailable

Bases: TrackerError

The tracker didn't answer, or failed: a request may or may not have taken effect.

event_id 

event_id(*parts: str) -> str

The id of a tracker: event: trk_ and 16 hex digits of a digest of what it says.

relayr.trackers.events

The tracker: events: what people change in a bound tracker project, as reflexr sees it.

Each type is declared under TrackerEvent, whose event_namespace="tracker" makes every type tracker:<name>, as relayr's bridged types are artifactr:<name> (relayr ADR-0006). Every event carries the tracker it came from, so a rule is written once for Plane and Vikunja. Give TRACKED to reflexr's Workspaces(events=...), and PUBLISHERS to its publishers=, which reserves the namespace to the followers' source:

workspaces = Workspaces(
    storage,
    events=[*BRIDGED, *TRACKED, *app_events],
    publishers={"artifactr": [relayr.inbound.SOURCE], **PUBLISHERS},
)

The application must never resolve a client to source:tracker (reflexr ADR-0057).

SOURCE module-attribute 

SOURCE: Final = SourceActor(name='tracker')

Who the tracker followers publish into reflexr as: the actor on every tracker: envelope.

PUBLISHERS module-attribute 

PUBLISHERS: Final[Mapping[str, tuple[SourceActor, ...]]] = (
    MappingProxyType({"tracker": (SOURCE,)})
)

The tracker namespace reserved to the followers' source, for reflexr's publishers=.

TRACKED module-attribute 

TRACKED: tuple[type[TrackerEvent], ...] = (
    ItemCreated,
    ItemChanged,
    ItemCommented,
    ItemRemoved,
)

Every tracker: type: give them to reflexr's Workspaces(events=...), with the application's own, so the followers may publish them.

Field 

Field = Literal[
    "title",
    "description",
    "state",
    "priority",
    "labels",
    "assignees",
]

What a change to an item names.

TrackerProjection 

TrackerProjection = Callable[[TrackerEvent], Event | None]

Maps a tracker: event to the application's own event, or to None to publish nothing.

What it returns is published with the tracker: event's id; an exception dead-letters it.

TrackerEvent pydantic-model

Bases: Event

The base of the tracker: types: the item each is about, in the tracker it came from.

Fields:

tracker pydantic-field 

tracker: str

The tracker, by its integration's name after tracker:: plane or vikunja.

item_id pydantic-field 

item_id: str

The tracker's id for the item.

number pydantic-field 

number: int

The item's number in its project.

name pydantic-field 

name: str

What people call it, as the tracker shows it: OPS-12.

link: str

The item's address in the tracker's web app.

key pydantic-field 

key: ItemKey | None = None

The key lattice filed it by, if lattice did.

ItemCreated pydantic-model

Bases: TrackerEvent

An item appeared in the project, filed by someone other than relayr.

Fields:

actor pydantic-field 

actor: str | None = None

The tracker's id for whoever filed it, if the tracker says.

ItemChanged pydantic-model

Bases: TrackerEvent

A field of an item differs from when it was last read, by someone other than relayr.

Changes are read, not delivered: two between reads are one, and one undone between reads is none.

Fields:

actor pydantic-field 

actor: str | None = None

The tracker's id for whoever changed it, if the tracker says and relayr changed nothing on the item since it was last read.

ItemCommented pydantic-model

Bases: TrackerEvent

Someone other than relayr commented on an item.

Fields:

body pydantic-field 

body: str

The comment's text, as HTML.

author pydantic-field 

author: str | None = None

The tracker's id for whoever wrote it, if the tracker says.

ItemRemoved pydantic-model

Bases: TrackerEvent

An item left the project's list: deleted, or on Plane archived.

Fields:

deleted pydantic-field 

deleted: bool | None = None

Whether it was deleted, rather than archived, when that is known: for an item lattice filed, which its key finds archived.

relayr.fastapi

FastAPI adapters: the trackers' webhooks, and the tracker connector's reads for the portal.

The webhooks nudge their followers; the reads say what runs filed, and how a follower is doing.

Install relayr-ai[fastapi], and mount the routers beside the libraries':

app.include_router(tracker_router(followers))  # POST /relayr/v1/trackers/{tenant}/{tracker}
app.include_router(  # GET /relayr/v1/workspaces/{workspace_id}/tracker, .../tracker/items
    tracker_reads_router(trackers, automation, resolve_actor=resolve_reflexr_actor)
)

MAX_BODY module-attribute 

MAX_BODY: Final = 1024 * 1024

The largest webhook body read, in bytes: 1 MB.

Filed pydantic-model

Bases: BaseModel

What the runs of a workspace filed in its tracker.

Config:

  • frozen: True

Fields:

tracker pydantic-field 

tracker: str

The tracker, by its integration's name after tracker:: plane or vikunja.

items pydantic-field 

items: list[FiledItem]

The newest first, by their number.

FiledItem pydantic-model

Bases: BaseModel

An item a run filed, and the chains that wrote to it.

Config:

  • frozen: True

Fields:

key pydantic-field 

key: str

The key it was filed by.

id pydantic-field 

id: str

The tracker's id for the item.

name pydantic-field 

name: str

What people call it, as the tracker shows it: OPS-12.

link: str | None

Its address in the tracker's web app, or None if the workspace is bound to no tracker now.

deleted pydantic-field 

deleted: bool

Whether a person deleted it: its key files nothing again.

chains pydantic-field 

chains: list[str]

The causal chains whose runs filed it or commented on it, the one that filed it first.

Following pydantic-model

Bases: BaseModel

How the tracker connector is doing in a workspace.

Config:

  • frozen: True

Fields:

tracker pydantic-field 

tracker: str

The tracker, by its integration's name after tracker:: plane or vikunja.

integration pydantic-field 

integration: str

Its key among the application's integrations, which switches it per workspace.

binding pydantic-field 

binding: TrackerBinding | None

The project the workspace is bound to, with its defaults; None if it is bound to none.

reading pydantic-field 

reading: Reading | None

How far its follower has read; None if it never has.

filed pydantic-field 

filed: int

How many items its runs filed.

dead_letters pydantic-field 

dead_letters: list[TrackerDeadLetter]

The changes its follower could not publish, by event id.

writes pydantic-field 

writes: list[Writes]

The items relayr wrote to within the cap's window, the most written first.

loop_cap pydantic-field 

loop_cap: int

How many writes to one item the cap allows in its window.

loop_window_seconds pydantic-field 

loop_window_seconds: int

The cap's window, in seconds.

Reading pydantic-model

Bases: BaseModel

How far a workspace's tracker follower has read its project.

Config:

  • frozen: True

Fields:

cursor_at pydantic-field 

cursor_at: datetime | None

The tracker's updated_at of the last item it read; None if the project had none.

swept_at pydantic-field 

swept_at: datetime

When it last read the whole project.

read_at pydantic-field 

read_at: datetime | None

When it last read the project at all; None until it reads again after an upgrade.

Writes pydantic-model

Bases: BaseModel

relayr's recent writes to one item, which the cap on loops counts.

Config:

  • frozen: True

Fields:

name pydantic-field 

name: str | None

What people call it, if the follower has read it.

writes pydantic-field 

writes: int

How many writes relayr made to it within the cap's window.

tracker_reads_router 

tracker_reads_router(
    trackers: Trackers,
    workspaces: Workspaces,
    *,
    resolve_actor: ResolveActor,
) -> APIRouter

Return the router of the tracker connector's reads, for the portal.

Parameters:

Name Type Description Default
trackers Trackers

The connector whose ledger, bindings and credentials are read.

 required
workspaces Workspaces

reflexr's workspaces, whose authorize decides each read.

 required
resolve_actor ResolveActor

Authenticates each request, as reflexr's router takes it.

 required

tracker_router 

tracker_router(followers: TrackerFollowers) -> APIRouter

Return the router the trackers' webhooks are registered to, for a tenant and a tracker.

Parameters:

Name Type Description Default
followers TrackerFollowers

The followers a verified webhook wakes, whose credentials verify it.

 required

relayr.integrations

Whether an integration runs in a workspace: lattice-core's port, answered without a store.

An integration is named by a key, such as tracker:plane or tracker:vikunja. Before it acts in a workspace, it asks lattice-core's Integrations port; one that says no writes nothing, reads nothing and keeps nothing. grantr's Grants answers it from a store of switches per deployment, tenant and workspace (lattice ADR-0027). Without one, every integration runs: ALWAYS.

ALWAYS module-attribute 

ALWAYS: Final[Integrations] = _Always()

Every integration runs everywhere, a tracker's key lattice-core doesn't declare included: the answer when the application gives no store.

relayr.plane

The Plane adapter (lattice RFC-0004): the tracker port over Plane Community edition's API.

Install relayr-ai[plane], and connect each workspace through one shared client:

http = httpx.AsyncClient()
trackers = Trackers(
    ledger, connect=plane(http), integration=INTEGRATION, credentials=..., bindings=...
)

INTEGRATION module-attribute 

INTEGRATION: Final = 'tracker:plane'

Plane's key among the application's integrations (relayr.integrations).

PlaneTracker 

PlaneTracker(
    credentials: TrackerCredentials,
    source: str,
    *,
    client: AsyncClient,
)

One tenant's Plane workspace, as one lattice workspace writes to it.

Parameters:

Name Type Description Default
credentials TrackerCredentials

The tenant's: Plane's URL, an API key of the tenant's own Plane user, its webhooks' secret, and its workspace's slug.

 required
source str

The lattice workspace, lattice:<tenant>/<workspace>: Plane's external_source on everything it writes.

 required
client AsyncClient

The HTTP client, which the application shares and closes.

 required

find async 

find(project: ProjectId, key: ItemKey) -> ItemRef | None

Return the item Plane's external ids name, archived or not, or None.

create async 

create(
    project: ProjectId, key: ItemKey, draft: ItemDraft
) -> ItemRef

Create a work item with the key as its external id; a used key returns its item.

item async 

item(ref: ItemRef) -> Item | None

Return a work item's snapshot, or None if it was deleted.

An item lattice filed is read by its external ids, which find it archived too; any other by its id, which doesn't.

update async 

update(ref: ItemRef, fields: ItemFields) -> None

Patch the fields given; labels replace the item's.

comment async 

comment(
    ref: ItemRef, key: CommentKey, body: str
) -> CommentRef

Post a comment with the key as its external id; a used key returns its comment.

The item is read first: Plane deletes a work item softly, and posts a comment on a deleted one with a 201, unseen.

items async 

items(
    project: ProjectId, after: Cursor | None
) -> Page[Item]

Return the project's listed work items updated after a cursor, by updated_at then id.

Plane CE filters nothing by time, so every page of the list is read, 1000 at a time, in the order items were made, which a change during the read doesn't reshuffle.

comments async 

comments(
    ref: ItemRef, after: CommentRef | None
) -> list[Comment]

Return a work item's comments after one, oldest first, or all of them.

nudge 

nudge(
    headers: Mapping[str, str], body: bytes
) -> Nudge | None

Verify a webhook's signature over its bytes and its workspace, and return its project.

link(ref: ItemRef) -> str

Return the work item's page: <base>/<workspace>/browse/<PROJECT>-<number>/.

holds 

holds(state: State) -> State

Every state: each is a state group of Plane's.

plane 

plane(client: AsyncClient) -> Connect

Connect each lattice workspace to its tenant's Plane, through one shared client.

relayr.vikunja

The Vikunja adapter (lattice RFC-0004): the tracker port over Vikunja's REST API.

Install relayr-ai[vikunja], and connect each workspace through one shared client:

http = httpx.AsyncClient()
trackers = Trackers(
    ledger, connect=vikunja(http), integration=INTEGRATION, credentials=..., bindings=...
)

INTEGRATION module-attribute 

INTEGRATION: Final = 'tracker:vikunja'

Vikunja's key among the application's integrations (relayr.integrations).

VikunjaTracker 

VikunjaTracker(
    credentials: TrackerCredentials,
    source: str,
    *,
    client: AsyncClient,
)

One tenant's Vikunja, as one lattice workspace writes to it.

Parameters:

Name Type Description Default
credentials TrackerCredentials

The tenant's: Vikunja's URL, an API token of the tenant's own user with the scopes the tracker guide lists, and its webhooks' secret.

 required
source str

The lattice workspace, lattice:<tenant>/<workspace>, which its markers name.

 required
client AsyncClient

The HTTP client, which the application shares and closes.

 required

find async 

find(project: ProjectId, key: ItemKey) -> ItemRef | None

Return the item whose description ends with the key's marker, or None.

create async 

create(
    project: ProjectId, key: ItemKey, draft: ItemDraft
) -> ItemRef

Make a task with the key's marker at the end of its description, then its labels.

item async 

item(ref: ItemRef) -> Item | None

Return a task's snapshot, or None if it was deleted.

update async 

update(ref: ItemRef, fields: ItemFields) -> None

Send the task whole with the fields given set, then add and remove labels.

comment async 

comment(
    ref: ItemRef, key: CommentKey, body: str
) -> CommentRef

Post a comment ending with the key's marker, unless one on the task has it.

items async 

items(
    project: ProjectId, after: Cursor | None
) -> Page[Item]

Return the project's tasks updated after a cursor, by updated then id.

Vikunja keeps updated to the second, so the filter asks from the second before the cursor's, and what isn't after it is dropped here. Pages are read by id, which a change during the read doesn't reshuffle.

comments async 

comments(
    ref: ItemRef, after: CommentRef | None
) -> list[Comment]

Return a task's comments after one, oldest first, or all of them.

nudge 

nudge(
    headers: Mapping[str, str], body: bytes
) -> Nudge | None

Verify a webhook's signature over its bytes, and return its project.

link(ref: ItemRef) -> str

Return the task's page: <base>/tasks/<id>.

holds 

holds(state: State) -> State

done for done and cancelled, which set the task's flag; todo for the others.

vikunja 

vikunja(client: AsyncClient) -> Connect

Connect each lattice workspace to its tenant's Vikunja, through one shared client.

relayr.rules

Rules from chat: a rule drafted in a thread goes live in reflexr once a person accepts it.

The chat agent drafts a rule with RuleDrafting's draft_rule, which checks it, previews it and proposes it as a rule artifact. When a person who may install rules accepts it, relayr's relayr:rules-from-chat rule installs it in reflexr, through Installer; archiving the artifact archives the rule.

ALLOWLISTS module-attribute 

ALLOWLISTS: Final[Mapping[RuleName, Allowlist]] = {
    RULE_DRIFT.name: Allowlist(threads="any")
}

What relayr's rules from chat may do beyond proposals and notices in their chains' threads: relayr:rule-drift proposes in the thread its rule was drafted in, which its chain doesn't hold. Give them to the outbound adapter with the application's own.

RULES module-attribute 

RULES: Final = [RULES_FROM_CHAT, RULE_DRIFT]

relayr's rules from chat, for the application's reflexr workspaces.

MayChange 

MayChange = Callable[
    [str, str, UserActor, RuleChange], Awaitable[bool]
]

Whether a person may make a change to stored rules in a tenant's workspace: (tenant, workspace, person, change).

RuleChange 

RuleChange = Literal[
    "install_rule", "update_rule", "archive_rule"
]

A change to a stored rule, named as reflexr's action for it is, without its prefix.

RuleDrafting dataclass 

RuleDrafting(
    automation: Workspaces,
    events: Mapping[str, type[Event]],
    window: timedelta = timedelta(days=7),
)

Bases: AbstractCapability[Session[Any]]

Give the chat agent draft_rule, which checks a rule, previews it, and proposes it.

The preview reads reflexr's log as the person the turn acts for, so it counts only what they may read.

Parameters:

Name Type Description Default
automation Workspaces

reflexr's workspaces, whose log the preview replays, and whose stored rules' configuration a draft is checked against.

 required
events Mapping[str, type[Event]]

The event types those workspaces accept, by name.

 required
window timedelta

How far back the preview looks.

 timedelta(days=7)

Raises:

Type Description
ValueError

If automation has no stored rules.

get_toolset 

get_toolset() -> FunctionToolset[Session[Any]]

Return draft_rule.

get_instructions 

get_instructions() -> str

Return when to draft a rule.

RuleDraft pydantic-model

Bases: Artifact

A reflexr rule, drafted in a thread and reviewed as a proposal before it goes live.

The type checks the rule's shape alone: artifactr validates an artifact with no access to the application's event types or its stored rules' configuration. check_draft checks the rest before the draft is proposed, and reflexr checks it all again at install.

Config:

  • extra: forbid
  • validate_assignment: True

Fields:

Installer 

Installer(
    outbound: Outbound,
    artifactr: Workspaces,
    *,
    may_change: MayChange,
)

The action of relayr's rule from chat. Register actions with the reactor.

Parameters:

Name Type Description Default
outbound Outbound

The adapter the runs post their notices through.

 required
artifactr Workspaces

The workspaces the runs re-read.

 required
may_change MayChange

Whether the person who accepted a draft, or archived it, may make the change to its rule. Pass anyone for an application without access control.

 required

actions property 

actions: Mapping[str, Action[Any]]

The rules' actions, by the names RULES run.

act async 

act(reaction: Reaction[Any]) -> None

Install or update the rule an accepted proposal drafted, or archive an artifact's.

drift async 

drift(reaction: Reaction[Any]) -> None

Propose to its artifact a change made to a rule from chat in reflexr directly.

A change relayr made itself, from an accepted draft, is in step already. A rule no draft installed has no artifact to keep in step.

Preview dataclass 

Preview(
    firings: int,
    replayed: int,
    since: datetime,
    complete: bool,
    errors: int = 0,
)

How often a rule would have fired, had it watched the log since since.

replayed instance-attribute 

replayed: int

How many envelopes it replayed.

complete instance-attribute 

complete: bool

Whether it replayed the whole window, rather than only its newest envelopes.

errors class-attribute instance-attribute 

errors: int = 0

How many envelopes the rule's evaluation recorded as errors, as reflexr would.

check_draft 

check_draft(
    rule: Rule,
    *,
    events: Mapping[str, type[Event]],
    stored_rules: StoredRules,
) -> list[str]

Return every problem with a draft as a stored rule in an application, or none.

It runs reflexr's two checks of a stored rule: what the rule refers to exists among events and the actions stored_rules allows, and the rule keeps within reflexr's limits on stored rules.

Parameters:

Name Type Description Default
rule Rule

The draft.

 required
events Mapping[str, type[Event]]

The event types the application's reflexr workspaces accept, by name.

 required
stored_rules StoredRules

The application's configuration of stored rules.

 required

anyone async 

anyone(
    _tenant_id: str,
    _workspace_id: str,
    _person: UserActor,
    _change: RuleChange,
) -> bool

Let any person make any change: the choice for an application without access control.

relayr.evals

Evaluating the bridge's rules: proposal outcomes as feedback, and measures from the logs.

Outcomes is the action of relayr:proposal-outcome, which gives each run whose proposal people resolved a ProposalOutcome. proposal_rates counts them by rule, rule_conversion follows rules from chat from draft to survival, and history gives an artifact's versions to evalr's measures, such as measure_rewrites.

EVALUATOR module-attribute 

EVALUATOR: Final = reflexr.core.EvaluatorActor(
    name="relayr:proposal-outcome", version="1"
)

Who gives the outcomes: relayr's evaluator of the log.

RULES module-attribute 

RULES: Final = [PROPOSAL_OUTCOME]

relayr's evaluation rules, for the application's reflexr workspaces.

ProposalRates dataclass 

ProposalRates(
    accepted: int = 0, edited: int = 0, rejected: int = 0
)

How people resolved one rule's proposals.

accepted class-attribute instance-attribute 

accepted: int = 0

Accepted as proposed.

edited class-attribute instance-attribute 

edited: int = 0

Accepted with changes of their own.

acceptance property 

acceptance: float

The share accepted, edited or not; 0 with none resolved.

edit_rate property 

edit_rate: float

The share of those accepted that people edited; 0 with none accepted.

RuleConversion dataclass 

RuleConversion(
    proposed: int,
    accepted: int,
    installed: int,
    eligible: int,
    survived: int,
)

How many rules drafted in chat were proposed, accepted, put live, and kept live.

installed instance-attribute 

installed: int

Rules relayr put live from accepted drafts.

eligible instance-attribute 

eligible: int

Of those, the ones first put live at least survival ago.

survived instance-attribute 

survived: int

Of those, the ones never archived since: in reflexr, from their artifact, or by anyone.

survival property 

survival: float

The share of eligible rules that survived; 0 with none eligible.

Outcomes 

Outcomes(artifactr: Workspaces, ledger: Ledger)

The action of relayr:proposal-outcome. Register actions with the reactor.

Parameters:

Name Type Description Default
artifactr Workspaces

The workspaces the runs re-read.

 required
ledger Ledger

The ledger relayr's adapters share, which holds the run behind each proposal.

 required

actions property 

actions: Mapping[str, Action[Any]]

The rule's action, by the name PROPOSAL_OUTCOME runs.

score async 

score(reaction: Reaction[Any]) -> None

Give the run behind a resolved proposal its outcome, once.

It re-reads the proposal in artifactr rather than trust the bridged event, and gives nothing for a proposal no rule's run made, nor one already scored. The run must be the proposer's own, so a proposal whose id someone else chose scores no rule's run.

ProposalOutcome pydantic-model

Bases: Feedback

How a person resolved a rule's proposal.

edited means they accepted it with changes of their own, so the artifact differs from what the proposal alone would have made.

Config:

  • frozen: True
  • extra: forbid

Fields:

proposal_id pydantic-field 

proposal_id: str

The proposal resolved: a run may make several, and each is scored once.

proposal_rates async 

proposal_rates(
    workspace: Workspace,
    *,
    since: datetime | None = None,
    until: datetime | None = None,
) -> dict[RuleName, ProposalRates]

The proposal outcomes relayr's evaluator gave, counted by the rule whose run proposed.

It reads the workspace's feedback facts and its runs, each once: reflexr reads facts by type alone. A measure runs offline, as an experiment or a report, so it reads the whole log rather than a bounded window of it.

Parameters:

Name Type Description Default
workspace Workspace

The workspace in reflexr.

 required
since datetime | None

Count only outcomes given at or after this time.

 None
until datetime | None

Count only outcomes given before this time.

 None

rule_conversion async 

rule_conversion(
    source: Workspace,
    workspace: Workspace,
    *,
    now: datetime,
    survival: timedelta = timedelta(days=30),
) -> RuleConversion

Count rules from chat through their life, from artifactr's proposals and reflexr's facts.

A rule survives while no reflexr:rule_archived follows its first install, whoever archived it: a person, through its artifact, or an operator in reflexr directly. Only what was logged before now counts, and the whole log is read, as for every measure.

Parameters:

Name Type Description Default
source Workspace

The workspace in artifactr, where rules are drafted.

 required
workspace Workspace

The same workspace in reflexr, where they run.

 required
now datetime

When to measure survival up to.

 required
survival timedelta

How long a rule must have been live to count.

 timedelta(days=30)

history async 

history(
    source: Workspace, artifact_id: ArtifactId
) -> History

Return an artifact's versions, for evalr's measure_rewrites and the like.

A version is its author's: an accepted proposal's version is its proposer's, not whoever accepted it, as artifactr records it. A person, or a client acting for one, is person; the system is system; an agent, a rule or a client acting for no one is agent. Each version's text is its data as canonical JSON, and its time is when artifactr logged it. Archiving changes no text, so an archive is no version here. It reads the workspace's whole log once, as a measure does, to find when each version was logged.

Raises:

Type Description
NotFound

If there is no such artifact.

relayr.telemetry

relayr's telemetry: the bridge's spans, its counters, and where its traces belong.

The follower publishes each bridged event in a span of its own, relayr.bridge <type>, linked to the span that committed the artifactr envelope, so a run in reflexr links back, through it, to the turn or request behind its event (ADR-0004: traces linked, not continued). It counts the followers that stop and the envelopes it dead-letters; and, for the tracker connector, the changes its followers dead-letter or find unannounced, the webhooks it refuses, and the writes its cap on loops refuses.

ENVELOPE_SEQ module-attribute 

ENVELOPE_SEQ: Final = 'relayr.envelope.seq'

The envelope's position in the artifactr workspace's log.

ERROR_TYPE module-attribute 

ERROR_TYPE: Final = 'error.type'

What stopped a follower: the type of its exception, in the semantic conventions' name.

EVENT_TYPE module-attribute 

EVENT_TYPE: Final = 'relayr.event.type'

The artifactr event's type, such as message_posted.

METRICS module-attribute 

METRICS: Mapping[str, Metric] = MappingProxyType(
    {
        metric.name: metric
        for metric in (
            FOLLOWERS_STOPPED,
            DEAD_LETTERS,
            TRACKER_DEAD_LETTERS,
            UNNUDGED,
            NUDGES_REFUSED,
            TRACKER_LOOPS,
        )
    }
)

Every metric relayr records, by name.

SCOPE module-attribute 

SCOPE: Final = 'relayr'

The instrumentation scope of relayr's own spans and metrics.

SCOPED module-attribute 

SCOPED: Final = frozenset({TENANT_ID, WORKSPACE_ID})

The tenancy attributes, which a deployment may drop.

Metric dataclass 

Metric(
    name: str,
    unit: str,
    description: str,
    attributes: frozenset[str],
)

One counter: what it counts, and the attributes it may carry.

Telemetry 

Telemetry(
    *,
    tracer_provider: TracerProvider | None = None,
    meter_provider: MeterProvider | None = None,
)

relayr's tracer, and a counter for every metric in the registry.

Parameters:

Name Type Description Default
tracer_provider TracerProvider | None

Where spans go. Defaults to the global provider.

 None
meter_provider MeterProvider | None

Where metrics go. Defaults to the global provider.

 None

count 

count(
    metric: Metric,
    attributes: Mapping[str, AttributeValue],
    amount: int = 1,
) -> None

Add to a counter, one by default, with only the attributes it may carry.

linked 

linked(traceparent: str | None) -> list[Link]

A link to the span a W3C traceparent names, if it names a valid one.

workspace_attributes 

workspace_attributes(
    tenant_id: str, workspace_id: str
) -> dict[str, AttributeValue]

The attributes that attribute work to a workspace.

relayr.otel

relayr's contribution to an application's telemetry, for the libraries' configure_telemetry.

Install relayr-ai[otel], and pass it beside reflexr's:

configure_telemetry(reflexr.otel.telemetry(), relayr.otel.telemetry(), service_name="app")

Its span filter keeps relayr's bridge spans in Langfuse's traces, and its views apply the libraries' cardinality policy to relayr's counters.

MetricsDetail module-attribute 

MetricsDetail = Literal['workspace', 'tenant', 'none']

How much tenancy detail the libraries' metrics keep: part of the contributions' contract.

Contribution dataclass 

Contribution(
    name: str,
    metric_views: Callable[[MetricsDetail], Sequence[View]],
    should_export_span: Callable[[ReadableSpan], bool],
    instrument: frozenset[Instrumented],
)

A library's contribution to an application's telemetry.

Parameters:

Name Type Description Default
name str

The library's name, which is its instrumentation scope.

 required
metric_views Callable[[MetricsDetail], Sequence[View]]

Returns the views that keep only the attributes the library's metrics may carry at a level of detail.

 required
should_export_span Callable[[ReadableSpan], bool]

Whether a span belongs in the library's traces.

 required
instrument frozenset[Instrumented]

The instrumentations the library advises.

 required

metric_views 

metric_views(
    detail: MetricsDetail = "workspace",
) -> list[View]

Return views that keep only the attributes relayr's counters may carry at detail.

With "tenant" the workspace is dropped, and with "none" both tenant and workspace are.

telemetry 

telemetry() -> Contribution

Return relayr's contribution, for either library's configure_telemetry.

relayr.langfuse

Langfuse across the bridge: the tags and user that join a causal chain and a thread.

Install relayr-ai[langfuse], and give the contexts to the libraries in place of their own:

reactor = Reactor(automation, actions=actions, run_context=bridged_run)
runner = Runner(agent, app=app, turn_context=bridged_turns(ledger))

bridged_run async 

bridged_run(
    reaction: Reaction[Any],
) -> AsyncGenerator[None]

Propagate a run's Langfuse attributes, with its threads and its person; a RunContext.

Give it to reflexr's Reactor(run_context=...) in place of langfuse_run.

bridged_turns 

bridged_turns(ledger: Ledger) -> TurnContext

Return a TurnContext that propagates a turn's attributes, with its rule and chain.

Give it to artifactr's Runner(turn_context=...) in place of langfuse_turn.

Parameters:

Name Type Description Default
ledger Ledger

The ledger relayr's adapters share, which holds the run behind each message.

 required

relayr.ui

The forms extra: ui:form_submitted and ui:app_submitted from artifactr's forms (ADR-0038).

What a person sends reaches reflexr's rules as one event, whichever way it was asked.

The forms extra, for an application with artifactr's forms and dashboards (artifactr.ui, artifactr-ai[forms]). relayr's follower bridges artifactr's log into reflexr; give reflexr UI_EVENTS and UI_PUBLISHERS beside the artifactr reservation, and the follower UI_PROJECTIONS::

automation = reflexr.Workspaces(
    storage,
    events=[*BRIDGED, *UI_EVENTS, *app_events],
    publishers={"artifactr": [SOURCE], **UI_PUBLISHERS},
)
followers = Followers(
    collaboration, automation, ledger, projections=Projections(**UI_PROJECTIONS)
)

An answer to ask_form and a submission to a form both become ui:form_submitted, so a rule is written once for both ways of asking; a submission to an app becomes ui:app_submitted. A submission's artifact_changed and artifact_archived become nothing. ui: is reserved to the follower's publisher, so no client publishes a submission itself (ADR-0038).

UI_EVENTS module-attribute 

UI_EVENTS: Final[tuple[type[UiEvent], ...]] = (
    FormSubmitted,
    AppSubmitted,
)

The forms' event types, for reflexr's Workspaces(events=...).

UI_PUBLISHERS module-attribute 

UI_PUBLISHERS: Final[
    Mapping[str, tuple[SourceActor, ...]]
] = MappingProxyType(
    {"ui": (SourceActor(name="artifactr"),)}
)

The ui namespace, reserved to relayr's follower, for reflexr's publishers=.

UI_PROJECTIONS module-attribute 

UI_PROJECTIONS: Final = UiProjections(
    artifacts=MappingProxyType(
        {Submission.kind: _submitted}
    ),
    deferred=MappingProxyType({ASK_FORM: _answered}),
)

The forms' projections, by the argument of relayr's Projections each is for: Projections(**UI_PROJECTIONS), or each merged with the application's own of that name.

UiEvent pydantic-model

Bases: Event

The base of the forms' events: who sent what, and where.

Fields:

author pydantic-field 

author: Actor

Who sent it in artifactr: the envelope's actor there.

thread_id pydantic-field 

thread_id: ThreadId | None = None

The thread it was sent in, if any.

values pydantic-field 

values: dict[str, JsonValue]

The values sent: to a form, checked against its schema before they were recorded.

FormSubmitted pydantic-model

Bases: UiEvent

A form was sent: a submission to a form artifact, or the answer to ask_form.

A submission carries form_id and version; an answer, the turn_id that asked and its tool_call_id.

Fields:

Validators:

  • _one_way

AppSubmitted pydantic-model

Bases: UiEvent

A submission to an app, against the version of it the person saw.

Fields:

UiProjections

Bases: TypedDict

The projections relayr's Projections takes, by its argument's name.