Updated
On this page
Defines the StorageBackend protocol family: TaskStorage, EventStorage, RunStorage, RunControlStorage, OverseerStorage, InboxStorage, NotepadStorage, StorageReader, StorageLock, RecoveryOperations.
#trace.src.orxtra.trace._protocols
#trace.src.orxtra.trace._protocols
#TaskStorage
Task lifecycle operations.
#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#EventStorage
Event and transcript operations.
#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]#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#RunStorage
Run lifecycle operations.
#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#RunControlStorage
Run control subscription operations.
#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#OverseerStorage
Overseer state operations.
#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#InboxStorage
Inbox operations.
#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#NotepadStorage
Notepad operations.
#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#StorageReader
All read operations.
#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]]#StorageLock
Run locking operations.
#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#RecoveryOperations
Crash recovery operations.
#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#KnowledgeHashStorage
Persistence for knowledge file content hashes.
#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]#StorageBackend
Combined protocol for a complete storage backend.
A StorageBackend provides all storage operations: task lifecycle, events, runs, overseer state, inbox, notepad, reads, locks, and crash recovery. EventBus is separate because it has different lifecycle (long-lived connections for LISTEN/NOTIFY).