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

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.

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.

Tier

WriteOp

What the writer should do with a particular row.

Traits

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’s messages and yield every streaming-content row keyed by the chunk’s own id and agent_instance_hierarchy.

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?”).

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.

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 there’s no logged request for that hierarchy’s embedded response_id (no prior session).

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

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.

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.

write_agent_completion

write_function_execution

write_value

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

write_vector_completion

Type Aliases

RowsIter

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