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

TraceWriter is the PG write path for every trace-owned table: run/task transitions, events, transcripts, inbox items, decisions/constraints/assumptions/lessons, iterations, and workflow status.

#trace.src.orxtra.trace._writer

#trace.src.orxtra.trace._writer

#TraceWriter

#subscribe_run_control

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

Register a callback for run control signals (pause/abort).

Handles the startup race: if the run is already paused or aborted when subscribing, the callback fires immediately.

#unsubscribe_run_control

python
async def unsubscribe_run_control(self, run_id: UUID) -> 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

Persist a run with a caller-supplied run_id and created_by.

The run_id is generated by the orchestrating layer before the run's principal is minted, so the row and its principal share the same id. created_by is the persisted principal of the actor that started the run.

#transition_run

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

#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

#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 that produced this event -- every event has an actor. When idempotency_key is provided the INSERT uses ON CONFLICT (idempotency_key) DO NOTHING so duplicate keys are silently deduplicated; inserted is False in that case and the existing row's id is fetched and returned (matching InMemoryBackend's (existing_id, False) contract).

#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

#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

#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.

Resolves the system principal from the principals table (same pattern as _recovery._resolve_system_principal_id) and attributes the bulk expiry to it. Returns the count of expired items.

#write_context_diff

python
async def write_context_diff(self, attempt_id: UUID, pre_refinement: str, refinement_diff: str) -> 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

#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

#update_workflow_status

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

#write_coherence_summary

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