orxtra v0.13.0 /trace.src.orxtra.trace._memory_backend
Updated
On this page

Dict-based InMemoryBackend implementing StorageBackend for tests, plus InMemoryEventBus. Deliberately enforces no foreign keys -- a documented parity boundary with PgBackend.

#trace.src.orxtra.trace._memory_backend

#trace.src.orxtra.trace._memory_backend

#InMemoryBackend

In-memory implementation of StorageBackend for tests and lightweight use.

Parity boundary with the PG backend: this backend enforces NO foreign keys. created_by (on runs) and principal_id (on events) are accepted and stored verbatim without validating that a matching principal row exists -- the PG backend rejects unknown principals via FK RESTRICT, this one does not. Tests that need FK behavior must use the PG backend.

#create_task

python
async def create_task(self, run_id: UUID, parent_task_id: UUID | None, name: str, task_type: str, config: dict[str, Any] | None=None) -> UUID

#transition_task

python
async def transition_task(self, task_id: UUID, new_status: str, reason: str | None=None, *, principal_id: UUID) -> None

#create_task_attempt

python
async def create_task_attempt(self, task_id: UUID, attempt: int) -> UUID

#complete_task_attempt

python
async def complete_task_attempt(self, attempt_id: UUID, agent_output: str, structured_output: dict[str, Any] | None, check_result: dict[str, Any] | None, check_verdict: str | None, session_id: UUID | None, input_tokens: int, output_tokens: int, reasoning_tokens: int, cache_read_tokens: int, cache_write_tokens: int, cost_usd: Decimal, duration_seconds: float) -> None

#fail_task_attempt

python
async def fail_task_attempt(self, attempt_id: UUID, error: str, session_id: UUID | None, input_tokens: int, output_tokens: int, reasoning_tokens: int, cache_read_tokens: int, cache_write_tokens: int, cost_usd: Decimal, duration_seconds: float) -> None

#create_iteration

python
async def create_iteration(self, task_id: UUID, index: int, item_value: object) -> UUID

#complete_iteration

python
async def complete_iteration(self, iteration_id: UUID, output: str | None, structured_output: dict[str, Any] | None, check_results: list[dict[str, Any]] | None) -> None

#fail_iteration

python
async def fail_iteration(self, iteration_id: UUID, error: str) -> None

#write_event

python
async def write_event(self, run_id: UUID | None, event_type: str, data: dict[str, Any], task_id: UUID | None=None, *, principal_id: UUID, idempotency_key: str | None=None) -> tuple[UUID, bool]

Insert an event. Returns (event_id, inserted).

principal_id is the acting principal; accepted unvalidated (no FK enforcement -- see the class docstring's parity boundary). Mirrors the PG backend's ON CONFLICT (idempotency_key) DO NOTHING semantics: when idempotency_key is provided and already seen, the event is not inserted a second time and (existing_event_id, False) is returned. With no key, every call inserts and returns (event_id, True).

#write_transcript_entry

python
async def write_transcript_entry(self, session_id: UUID, run_id: UUID, turn: int, role: str, content: str, tool_calls: dict[str, Any] | None=None, tokens: dict[str, Any] | None=None) -> None

#create_run

python
async def create_run(self, intent: str, config: dict[str, Any], autonomy_level: str, *, run_id: UUID, created_by: UUID) -> UUID

#transition_run

python
async def transition_run(self, run_id: UUID, new_status: str, reason: str | None=None, *, principal_id: UUID) -> None

#write_coherence_summary

python
async def write_coherence_summary(self, run_id: UUID, summary: str) -> None

#subscribe_run_control

python
async def subscribe_run_control(self, run_id: UUID, callback: Callable[[UUID, str], Awaitable[None]]) -> None

#unsubscribe_run_control

python
async def unsubscribe_run_control(self, run_id: UUID) -> None

#write_decision

python
async def write_decision(self, run_id: UUID, decision_type: str, choice: str, rationale: str | None=None) -> UUID

#write_constraint

python
async def write_constraint(self, run_id: UUID, text: str, tier: str, kind: str, args: dict[str, Any] | None=None) -> UUID

#write_assumption

python
async def write_assumption(self, run_id: UUID, text: str, scope: str, inbox_item_id: UUID | None=None) -> UUID

#write_lesson

python
async def write_lesson(self, run_id: UUID, text: str, relevance_tags: list[str], permanent: bool, source_files: list[str] | None=None) -> UUID

#update_workflow_status

python
async def update_workflow_status(self, workflow_id: UUID, current_step: str, health: str) -> None

#write_context_diff

python
async def write_context_diff(self, attempt_id: UUID, pre_refinement: str, refinement_diff: str) -> None

#create_inbox_item

python
async def create_inbox_item(self, run_id: UUID, decision_type: str, question: str, options: list[dict[str, Any]], assumed_option: str | None, work_proceeding: str | None, contradiction_impact: str | None, tags: list[str] | None=None, deadline: datetime | None=None, answer_event: str | None=None) -> UUID

#answer_inbox_item

python
async def answer_inbox_item(self, item_id: UUID, answer: str, *, resolved_by: UUID) -> None

#skip_inbox_item

python
async def skip_inbox_item(self, item_id: UUID, *, resolved_by: UUID) -> None

#reject_inbox_item

python
async def reject_inbox_item(self, item_id: UUID, reason: str, *, resolved_by: UUID) -> None

#expire_inbox_item

python
async def expire_inbox_item(self, item_id: UUID, *, resolved_by: UUID) -> None

#expire_due_inbox_items

python
async def expire_due_inbox_items(self, now: datetime) -> int

Expire all pending inbox items whose deadline has passed.

#write_notepad_entry

python
async def write_notepad_entry(self, run_id: UUID, task_name: str, agent_name: str, entry_type: str, text: str) -> None

#list_tasks

python
async def list_tasks(self, run_id: UUID) -> list[TaskSummary]

#read_task_attempt

python
async def read_task_attempt(self, task_id: UUID, attempt: int) -> TaskAttempt | None

#read_latest_attempt

python
async def read_latest_attempt(self, task_id: UUID) -> TaskAttempt | None

#list_iterations

python
async def list_iterations(self, task_id: UUID) -> list[IterationResult]

#read_transcript

python
async def read_transcript(self, session_id: UUID) -> list[dict[str, Any]]

#search_transcript

python
async def search_transcript(self, session_id: UUID, query: str) -> list[dict[str, Any]]

#read_run_report

python
async def read_run_report(self, run_id: UUID) -> RunReport | None

#list_runs

python
async def list_runs(self) -> list[RunSummary]

#read_inbox

python
async def read_inbox(self, run_id: UUID, status: str | None=None) -> list[InboxItem]

#read_notepad

python
async def read_notepad(self, run_id: UUID) -> list[NotepadEntry]

#read_active_constraints

python
async def read_active_constraints(self, run_id: UUID) -> list[dict[str, Any]]

#read_task_attempts

python
async def read_task_attempts(self, task_id: UUID) -> list[TaskAttempt]

#query_events

python
async def query_events(self, run_id: UUID, event_type: str | None=None, since: datetime | None=None, limit: int=100) -> list[dict[str, Any]]

#read_inbox_item

python
async def read_inbox_item(self, item_id: UUID) -> InboxItem | None

#read_run_config

python
async def read_run_config(self, run_id: UUID) -> dict[str, Any] | None

#read_session_token_counts

python
async def read_session_token_counts(self, session_id: UUID) -> list[dict[str, Any]]

#read_session_turn_count

python
async def read_session_turn_count(self, session_id: UUID) -> int

#query_relevant_lessons

python
async def query_relevant_lessons(self, tags: list[str]) -> list[dict[str, Any]]

#read_decisions

python
async def read_decisions(self, run_id: UUID, limit: int=10) -> list[dict[str, Any]]

#read_constraints

python
async def read_constraints(self, run_id: UUID, active_only: bool=True) -> list[dict[str, Any]]

#read_assumptions

python
async def read_assumptions(self, run_id: UUID, status: str | None=None) -> list[dict[str, Any]]

#query_lessons

python
async def query_lessons(self, run_id: UUID | None=None, tags: list[str] | None=None, permanent_only: bool=False) -> list[dict[str, Any]]

#read_workflow_status

python
async def read_workflow_status(self, workflow_id: UUID) -> dict[str, Any] | None

#replay

python
async def replay(self, *, event_types: list[str] | None=None, principal_id: UUID | None=None, since_id: UUID | None=None, limit: int=1000) -> list[dict[str, Any]]

#acquire_run_lock

python
async def acquire_run_lock(self, run_id: UUID) -> None

#release_run_lock

python
async def release_run_lock(self, run_id: UUID) -> None

#update_heartbeat

python
async def update_heartbeat(self, run_id: UUID) -> None

#is_lock_stale

python
async def is_lock_stale(self, run_id: UUID, threshold_seconds: float=300.0) -> bool

#reclaim_interrupted

python
async def reclaim_interrupted(self) -> int

#reevaluate_blocked

python
async def reevaluate_blocked(self) -> list[UUID]

#clean_orphaned

python
async def clean_orphaned(self) -> int

#write_knowledge_hash

python
async def write_knowledge_hash(self, run_id: UUID, path: str, file_hash: str) -> None

#read_knowledge_hashes

python
async def read_knowledge_hashes(self, run_id: UUID) -> dict[str, str]

#InMemoryEventBus

In-memory implementation of EventBus for tests and lightweight use.

#subscribe

python
async def subscribe(self, channel: str, callback: Callable[[str], Awaitable[None]]) -> None

#unsubscribe

python
async def unsubscribe(self, channel: str, callback: Callable[[str], Awaitable[None]]) -> None

Remove a specific callback from a channel by identity.

#publish

python
async def publish(self, channel: str, payload: str) -> None

#_now

python
def _now() -> datetime

Return current UTC datetime.

Search