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

Read-only PG query functions backing StorageReader: list_tasks, read_run_report, read_inbox, query_events, replay (cursor-paginated), query_lessons, and other lookups over the event-store tables.

#trace.src.orxtra.trace._reader

#trace.src.orxtra.trace._reader

#_inbox_record_to_dict

python
def _inbox_record_to_dict(record: asyncpg.Record) -> dict[str, Any]

Coerce an inbox_items row for InboxItem validation.

options and tags are jsonb columns. A pool without a jsonb codec (the plain asyncpg.create_pool used in production and tests) returns them as JSON text, so parse defensively -- matching read_run_config.

#list_tasks

python
async def list_tasks(pool: asyncpg.Pool, run_id: UUID) -> list[TaskSummary]

#read_task_attempt

python
async def read_task_attempt(pool: asyncpg.Pool, task_id: UUID, attempt: int) -> TaskAttempt | None

#read_latest_attempt

python
async def read_latest_attempt(pool: asyncpg.Pool, task_id: UUID) -> TaskAttempt | None

#list_iterations

python
async def list_iterations(pool: asyncpg.Pool, task_id: UUID) -> list[IterationResult]

#read_transcript

python
async def read_transcript(pool: asyncpg.Pool, session_id: UUID) -> list[dict[str, Any]]

#search_transcript

python
async def search_transcript(pool: asyncpg.Pool, session_id: UUID, query: str) -> list[dict[str, Any]]

#read_run_report

python
async def read_run_report(pool: asyncpg.Pool, run_id: UUID) -> RunReport | None

#list_runs

python
async def list_runs(pool: asyncpg.Pool) -> list[RunSummary]

#read_inbox

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

#read_notepad

python
async def read_notepad(pool: asyncpg.Pool, run_id: UUID) -> list[NotepadEntry]

#read_active_constraints

python
async def read_active_constraints(pool: asyncpg.Pool, run_id: UUID) -> list[dict[str, Any]]

Read active constraints for a run.

#read_task_attempts

python
async def read_task_attempts(pool: asyncpg.Pool, task_id: UUID) -> list[TaskAttempt]

Read all attempts for a task, ordered by attempt number.

#query_events

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

Query events for a run with optional filters.

#read_inbox_item

python
async def read_inbox_item(pool: asyncpg.Pool, item_id: UUID) -> InboxItem | None

Read a single inbox item by ID.

#read_run_config

python
async def read_run_config(pool: asyncpg.Pool, run_id: UUID) -> dict[str, Any] | None

Read the config snapshot for a run.

#read_session_token_counts

python
async def read_session_token_counts(pool: asyncpg.Pool, session_id: UUID) -> list[dict[str, Any]]

Read token counts from transcripts for a session.

#read_session_turn_count

python
async def read_session_turn_count(pool: asyncpg.Pool, session_id: UUID) -> int

Read the number of transcript turns for a session.

#query_relevant_lessons

python
async def query_relevant_lessons(pool: asyncpg.Pool, tags: list[str]) -> list[dict[str, Any]]

Query lessons matching any of the given relevance tags.

#read_decisions

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

Read decisions for a run, newest first.

#read_constraints

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

Read constraints for a run, optionally filtered to active only.

#read_assumptions

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

Read assumptions for a run, optionally filtered by status.

#query_lessons

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

Query lessons with optional filters.

#read_workflow_status

python
async def read_workflow_status(pool: asyncpg.Pool, workflow_id: UUID) -> dict[str, Any] | None

Read overseer workflow status for a workflow.

#read_event

python
async def read_event(pool: asyncpg.Pool, event_id: UUID) -> dict[str, Any] | None

Read a single event by its ID.

Returns the event as a dict, or None if not found.

#replay

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

Replay events with optional filters and cursor-based pagination.

Uses since_id (UUIDv7, time-ordered) as cursor for monotonic pagination.

Search