Module logs 

Postgres-backed log writer.

The logs.* schema (defined in schema.sql) is a hybrid:

• Six tier blob tables (request + response × agent / vector / function) store the full chunk body as JSONB. Requests are written once on first chunk arrival; responses get UPDATEd per tick when their body changes.
• Fourteen streaming-content tables carry the per-message, per-part incrementally-updating content yielded by the SDK’s chunk-type log_rows() iterators.

Diff detection: every row written has a deterministic PK shape (response_id, index[, sub_index]). The writer holds a shadow map keyed identically and hashes each row’s body columns. Three verdicts come out: Insert (first sight), Update (changed body), Skip (unchanged). The SQL helpers in write dispatch flat INSERT or UPDATE per verdict — no ON CONFLICT ambiguity.

The writer is the sole caller and is single-instance per stream, so the shadow’s verdict is authoritative — concurrent races on the same row id can’t happen.

Structs

CompletionStatus

One nested agent completion’s liveness, borrowed from the folded cumulative chunk. finished = usage arrived (present only on a completion’s final chunk); errored = the completion carries its own in-band error (already persisted via WriterItem::Error). The mid-stream failure sweep logs the stream error for every completion that is neither.

ConversationTee

Handle held by the log writer. Cloneable — one spawn’s restart passes share one tee (one socket connection). Dropping every clone closes the channel; the RX task drains and exits.

FrameMapper

The writer’s row→event mapper. STATEFUL: head rows (the tool_response / request_message_tool / request_vector_choice metadata carriers) produce NO frame — they feed the maps below, and their payload rides every subsequent content event of their block (tool_call_id on the event’s boundary fields; the choice’s voting key per part). The writer emits each head strictly before its contents, so a lookup miss means a torn iterator — that content frame is skipped (reconnecting clients replay DB truth).

LogWriter

Background-task-fronted log writer.

SessionLookup

What lookup_session returns when a prior session exists for the queried agent_instance_hierarchy.

Shadow

Enums

MessageTable

The subset of RowTable that produces a objectiveai.messages event row when written. Maps 1:1 to the postgres objectiveai.message_table ENUM in schema.sql — same names, same order. The three response-blob tables are intentionally absent; they’re not events, just the latest snapshot. tool_response is also absent: its head row is written purely as the tool_call_id lookup for tool-response content rows (JOINed at read time) and emits no event of its own.

OwnedRowKey

Owned counterpart to RowKey. Stored in the shadow map. Built only on Insert.

RowBody

Owned body stored in the shadow map. Compared by PartialEq against an incoming RowValue via RowValue::body_eq.

RowKey

Borrowed key that identifies one row’s slot in postgres. Two RowKeys with equal variant + fields hash identically to an OwnedRowKey with the matching shape — that invariant lets the shadow’s hashbrown raw_entry_mut look up by borrowed key without converting to owned first.

RowTable

Every table in the logs.* schema, plus the synthetic MessageQueueContent variant for queue-consumption rows. The latter writes to objectiveai.messages with a "table" value chosen at write time via SQL CASE (one of message_queue_text, _image, _audio, _video, _file) — the Rust-side variant is kind-less because the dispatch happens in SQL.

RowValue

One streaming-content row to INSERT or UPDATE. Borrowed: every variant lifts string / media payloads from the owning chunk by reference. Every variant also carries the enclosing chunk’s agent_instance_hierarchy so the writer can populate objectiveai.messages.agent_instance_hierarchy and key the objectiveai.messages_queue downgrade against the right spawned agent.

TeeFrame

One JSONL line on conversation.sock. CLI-internal (the daemon is the only reader); the resolved wire type the daemon fans out is the SDK’s AgentInstanceEvent.

Tier

WriteOp

What the writer should do with a particular row.

WriterItem

One item from a top-level chunk walk: either a streaming-content RowValue or a per-AIH agent-completion token-usage snapshot. Usage rides the SAME single traversal as the content rows — the walk already descends into every nested agent completion, so the writer never re-walks the chunk tree for usage.

Traits

ChunkAgentRefs

Chunks that can surface (AIH, definition source) pairs for the objectiveai.agent_refs registry: every nested agent completion carrying agent_inline (each completion’s FIRST chunk). Per the registry’s rule, agent_remote present → the remote wins; otherwise the inline spec itself.

ChunkPush

CLI-side wrapper exposing the SDK’s intrinsic push(&mut self, other: &Self) method via a uniform trait. Each impl simply delegates to the chunk type’s inherent method — the SDK already guarantees push is a correct accumulator for the tier’s cumulative-state semantics.

WriterChunk

Functions

agent_completion_chunk_rows

Entry: walk an agent-completion chunk. Emits a WriterItem::Usage first when the chunk carries a non-None usage (its total_tokens snapshot), then every streaming-content row keyed by the chunk’s own id and agent_instance_hierarchy.

agent_completion_statuses

Status of the one completion an agent chunk IS.

any_pending_matching_kinds

Side-effect-free existence check used by agents logs read subscribe’s wait loop. Returns true iff objectiveai.messages_queue has at least one unread row past the watermark for any child of parent_agent_instance_hierarchy whose m."table" falls in kinds. When kinds is None or empty, the kind filter is dropped (existence check across all kinds — equivalent to “is there anything pending at all?”).

error_frame

Build the tee frame for one logged error — persist-before-return: the spawn path calls this AFTER insert_error committed the row.

function_execution_chunk_rows

Entry: walk every task in a function chunk and forward to the matching tier walker. Reasoning summary’s inner agent completion also flows through. Recursive: function tasks chain back into this function. Usage items interleave with rows via the leaf walker.

function_execution_statuses

Statuses of every nested agent completion in a function chunk — recursive (function tasks chain back in), reasoning summaries included. Mirrors function_execution_chunk_rows’s traversal.

get_agent_token_usage

Current most-recent agent-completion total_tokens snapshot for an AIH (objectiveai.agent_token_usage). None when no usage has been recorded for that AIH yet.

insert_error

Persist one error into an agent’s history. response_id is the response the failure belongs to when one existed, None for post-lock pre-stream failures. error is the CLI’s user-facing error value (Error::output_message()). Returns the objectiveai.errors.id — the row’s messages.row_index.

insert_plugin_message

INSERT one captured stderr line for a plugins run invocation.

insert_request_blob

INSERT the request blob. Called once per stream, on first chunk arrival. Request blobs don’t carry agent_instance_hierarchy — they’re shared across every agent that participates in the stream. The per-agent “the request was made for me” linkage lives in objectiveai.messages and is written separately by insert_request_messages_row the first time each agent appears in the chunk’s row iterator.

insert_request_messages_row

INSERT a objectiveai.messages row that registers this stream’s request blob in the agent’s history. Called once per (stream, agent) pair — the writer tracks which agents it has already seen and only emits this row the first time it encounters a new one in the row iterator. By postgres’s BIGSERIAL "index" assignment, this row is guaranteed to land earlier in the agent’s history than any subsequent streaming-content row that the same writer call sequences after it.

insert_response_blob

INSERT the response tier blob (first tick only). Response blobs don’t emit messages — they’re the latest snapshot, not events. Tier-symmetric: every tier’s response table now has the same (response_id, body, created_at, inserted_at) shape.

lookup_session

Resolve the session for agent_instance_hierarchy. Ok(None) means neither the agent_refs registry nor the legacy request-blob fallback holds a definition for that hierarchy (no prior session recorded).

read_all_for_hierarchy

Materialize every objectiveai.messages row for agent_instance_hierarchy (filtered by after_id / limit), coalesced into ResponseItem blocks.

read_by_id

Resolve one objectiveai.messages."index" to the matching SDK Response variant. Returns Ok(None) when no row exists at that index.

read_conversation_page

One page of an agent’s conversation as typed events, content inlined, in objectiveai.messages."index" order. Returns the events plus the after_id cursor for the next page (None when this page was the last). The caller (the daemon WS handler) loops pages, streaming each event as one frame — bounded memory for huge histories.

read_pending_for_parent

Materialize every unread objectiveai.messages row for the children spawned by parent_agent_instance_hierarchy (per objectiveai.messages_queue watermarks), coalesced into ResponseItem blocks. Bumps each affected child’s read_index to GREATEST(current, max_returned) atomically in the same SQL statement.

read_plugin_messages

Read captured stderr lines for one plugin coordinate, ascending by "index", skipping rows with "index" <= after_id and capping at limit (None = unlimited).

update_agent_token_usage

Overwrite an AIH’s most-recent agent-completion total_tokens (last-write-wins snapshot — not a running sum).

update_response_blob

UPDATE the response tier blob (subsequent ticks).

vector_completion_chunk_rows

Entry: walk every embedded per-agent completion in a vector chunk and forward to agent_completion_chunk_rows (usage included).

vector_completion_statuses

Statuses of every embedded per-agent completion in a vector chunk.

wait_for_logs_message_at

Block until the next objectiveai.messages INSERT whose agent_instance_hierarchy payload equals target_aih. Mismatching notifications are silently consumed.

wait_for_token_usage_change

Block until this AIH’s stored total_tokens differs from baseline, returning the new value.

write_agent_completion

write_function_execution

write_value

Dispatch SQL for value per op. Skip is a no-op.

write_vector_completion

Type Aliases

CompletionStatuses

Boxed iterator of CompletionStatuses.

RowsIter

Boxed-iterator alias used at the recursive boundaries (function execution → vector completion → agent completion). One Box per recursive descent, never per leaf row.

WriterItems

Boxed iterator of WriterItems — the return type of the three entry-point chunk walkers. Internal per-message helpers still yield bare RowValue via RowsIter; only the entry points interleave the usage item.