Honcho CloudEvents Glossary

Reference for every CloudEvent emitted by Honcho and the on-disk shape of the Parquet archive that Xatu writes to S3.

Honcho instances emit telemetry as CloudEvents to a configurable endpoint (TELEMETRY.ENDPOINT). In the managed-SaaS deployment that endpoint is Xatu’s POST /v1/events. Xatu enriches each event with tenant_id (resolved by Fly app name → HonchoInstance.app_name), publishes to Redpanda, and the consumer archives batches to S3 as zstd-compressed Parquet partitioned by date.

Source of truth:

Event classes: repos/honcho/src/telemetry/events/
Emitter (envelope construction): repos/honcho/src/telemetry/emitter.py
Ingestion: repos/groudon/xatu/ingestion/main.py
Parquet writer: repos/groudon/xatu/consumer/parquet_writer.py

Related: [[cloudevents-enrichment|cloudevents-enrichment.md]] spec for the design history and rev notes.

CloudEvent envelope

All Honcho events ship with the standard CloudEvents v1.0 attributes:

Attribute	Value	Example
`specversion`	Always `"1.0"`	`"1.0"`
`id`	Deterministic SHA-256-derived ID (`evt_{base64url}`) — same `(event_type, timestamp, resource_id)` always produces the same ID, enabling idempotent retries	`evt_AbCd...`
`source`	`/honcho/{namespace}/{category}` where `namespace` = `TELEMETRY.NAMESPACE` (Fly app name in SaaS) and `category` is one of `representation`, `dream`, `dialectic`, `agent`, `deletion`, `reconciliation`, `llm`, `api`, `trace`	`/honcho/honcho-acme-prod/dialectic`
`type`	The event type string (see table below)	`dialectic.completed`
`time`	ISO-8601 UTC timestamp	`2026-05-14T03:21:09.482Z`
`datacontenttype`	`"application/json"`
`dataschema`	`https://honcho.dev/schemas/{type}/v{schema_version}`	`https://honcho.dev/schemas/dialectic.completed/v2`
`data`	The event payload (per-event fields below)

Xatu adds two enrichment fields on top of the envelope before publishing to Redpanda and writing to Parquet:

namespace — extracted from the source field (Fly app name)
tenant_id — looked up from namespace against the Groudon database

The Parquet schema (see Querying) stores data as a JSON string, not a typed struct, so payload fields are accessed via json_extract / -> in DuckDB/Trino.

The emitter also injects honcho_version into the event body (so events from different deploys don’t dedupe).

Volume & sampling

Each event declares a volume class. Ground-truth events (the aggregate envelopes like representation.completed, dialectic.completed, dream.run) always emit. High-volume per-call events (llm.call.completed, embedding.call.completed, agent.tool.call.completed) are subject to TELEMETRY.HIGH_VOLUME_SAMPLE_RATE, sampled deterministically on run_id so a whole agentic trace is either fully kept or fully dropped. At a sample rate < 1.0, aggregates still carry totals but per-call children are undercounted — rebuild cost from the aggregates, not the sampled children.

Trace payloads (full-fidelity stream)

A separate, opt-in stream carries replay-grade payloads — the exact context each model saw — content-addressed so payload size stays O(N). These trace-category events (llm.call.traced, embedding.call.traced, trace.content) go through a distinct emitter (emit_trace() / _trace_emitter) gated by TELEMETRY.TRACE_PAYLOADS_ENABLED, and are ground-truth (never sampled). llm.call.traced deliberately does not join to llm.call.completed — cost is computable from the trace alone, keeping the billing and audit streams decoupled.

Event catalog

`representation.completed`

Category: representation · Schema: v2 · Source file: src/telemetry/events/representation.py

Emitted when the deriver finishes processing a batch of messages and writes the resulting conclusions for a peer’s representation. This is the workhorse event in the memory-formation pipeline — one per processed batch per observed peer.

Field	Type	Description
`workspace_name`	str	Workspace
`session_name`	str	Session
`observed`	str	Peer being observed
`queue_items_processed`	int	Number of `QueueItem` rows dequeued
`earliest_message_id`	str	First message in the batch
`latest_message_id`	str	Last message in the batch
`message_count`	int	Messages processed
`explicit_conclusion_count`	int	Explicit conclusions extracted
`context_preparation_ms`	float	Time preparing context
`llm_call_ms`	float	Time in the LLM call
`total_duration_ms`	float	End-to-end duration
`input_tokens`	int	Queued-message tokens (the messages actually reasoned about); downstream metering key — do not rename
`total_input_tokens`	int	Total tokens sent to the LLM (queued + extra context + scaffold)
`output_tokens`	int	LLM output tokens
`queued_message_count`	int	Messages in the batch that were the actual queue items
`prompt_message_count`	int	Total messages in the prompt (queued + extra interleaving context)
`prompt_message_tokens`	int	Sum of `token_count` across all prompt messages
`extra_context_message_count`	int	`prompt_message_count - queued_message_count`
`extra_context_tokens`	int	`prompt_message_tokens - input_tokens` — token cost of extra context
`prompt_scaffold_tokens`	int	Estimated tokens for the system/scaffold portion of the prompt
`batch_max_tokens`	int	`DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS` at fetch time
`max_input_tokens`	int	`DERIVER.MAX_INPUT_TOKENS` at call time
`was_flush_enabled`	bool	`DERIVER.FLUSH_ENABLED` snapshot at batch time
`hit_batch_token_cap`	bool	True when the batcher clamped the batch to fit `batch_max_tokens`
`hit_input_token_cap`	bool	True when the LLM call truncated input to fit `max_input_tokens`
`observer_count`	int	Number of observers this representation was saved against

Resource ID: {workspace_name}:{session_name}:{latest_message_id}

`dream.run`

Category: dream · Schema: v2 · Source file: src/telemetry/events/dream.py

Top-level event for a single dream orchestration. Aggregates totals across all specialists. Correlate with dream.specialist and agent.iteration events via run_id.

Field	Type	Description
`run_id`	str	Nanoid for cross-event correlation
`workspace_name`	str	Workspace
`session_name`	str \| null	Session if scoped to one
`observer`	str	Observer peer
`observed`	str	Observed peer
`specialists_run`	list[str]	E.g. `["deduction", "induction"]`
`deduction_success`	bool
`induction_success`	bool
`surprisal_enabled`	bool	Whether surprisal sampling ran
`surprisal_conclusion_count`	int	High-surprisal conclusions found
`total_iterations`	int	LLM iterations across all specialists
`total_input_tokens`	int
`total_output_tokens`	int
`total_duration_ms`	float
`dream_type`	str \| null	DreamType slug (currently `omni`; future `deductive`/`inductive`)
`enabled_types_count`	int	`len(DREAM.ENABLED_TYPES)` at run start
`trigger_reason`	str \| null	What tripped the schedule: `document_threshold` / `manual` / `surprisal`
`delay_reason`	str \| null	What governed when it fired: `idle_timeout` / `immediate` / `min_hours_gate`
`documents_since_last_dream_at_schedule`	int \| null	Document count snapshot at schedule time
`document_threshold`	int \| null	`DREAM.DOCUMENT_THRESHOLD` snapshot at schedule time

Resource ID: {run_id}

`dream.specialist`

Category: dream · Schema: v2

One per specialist (deduction, induction) within a dream run. Shares run_id with the parent dream.run.

Field	Type	Description
`run_id`	str	Same as parent dream.run
`specialist_type`	str	`"deduction"` or `"induction"`
`workspace_name`	str
`observer`	str
`observed`	str
`iterations`	int	LLM iterations
`tool_calls_count`	int	Tool calls made
`input_tokens`	int
`output_tokens`	int
`duration_ms`	float
`success`	bool
`created_observation_count`	int	Actual observations created (from `ToolResult.metadata.created_count`, not tool-name counting)
`deleted_observation_count`	int	Actual observations deleted (from `ToolResult.metadata.deleted_count`)
`created_counts_by_level`	dict[str, int]	Created observations per level (explicit/deductive/inductive/contradiction); treat missing keys as 0
`deleted_counts_by_level`	dict[str, int]	Deleted observations per level
`peer_card_updated`	bool	True when ≥1 `update_peer_card` call succeeded
`search_tool_calls_count`	int	`search_memory` / `search_messages` / `search_messages_temporal` invocations
`error_class`	str \| null	Exception class when `success=False`; null on success

Resource ID: {run_id}:{specialist_type}

`dialectic.completed`

Category: dialectic · Schema: v2 · Source file: src/telemetry/events/dialectic.py

Emitted when a /peers/{peer_id}/chat query finishes. User-facing latency and cost surface — most billing analytics start here.

Field	Type	Description
`run_id`	str	Correlates with `agent.iteration` and `agent.tool.*`
`workspace_name`	str
`peer_name`	str	Peer being queried about
`session_name`	str \| null	If session-scoped
`reasoning_level`	str	`minimal` / `low` / `medium` / `high` / `max`
`total_iterations`	int	LLM iterations (defaults 1)
`prefetched_conclusion_count`	int	Conclusions prefetched before agent loop
`tool_calls_count`	int	Tool calls made
`total_duration_ms`	float
`input_tokens`	int
`output_tokens`	int
`cache_read_tokens`	int	Prompt-cache read tokens
`cache_creation_tokens`	int	Prompt-cache write tokens
`hit_input_token_cap`	bool	True when an iteration’s input exceeded `DIALECTIC.MAX_INPUT_TOKENS`

Resource ID: {run_id}

`agent.iteration`

Category: agent · Schema: v2 · Source file: src/telemetry/events/agent.py

One per LLM call inside an agentic loop (dream specialists and dialectic agents). Use for per-iteration cost analysis and tool-call attribution.

Field	Type	Description
`run_id`	str	Correlates with parent `dream.run` / `dream.specialist` / `dialectic.completed`
`parent_category`	str	`"dream"` or `"dialectic"`
`agent_type`	str	`"deduction"`, `"induction"`, or `"dialectic"`
`workspace_name`	str
`observer`	str \| null	Set for dream agents
`observed`	str \| null	Set for dream agents
`peer_name`	str \| null	Set for dialectic agent
`iteration`	int	1-indexed iteration number
`tool_calls`	list[str]	Tool names called this iteration (can be empty)
`input_tokens`	int
`output_tokens`	int
`cache_read_tokens`	int
`cache_creation_tokens`	int

Resource ID: {run_id}:{iteration}

`agent.tool.conclusions.created`

Category: agent · Schema: v2

Emitted when the create_conclusions agent tool executes. Includes a per-conclusion level breakdown (explicit vs. deductive).

Field	Type	Description
`run_id`	str
`iteration`	int
`parent_category`	str
`agent_type`	str
`workspace_name`	str
`observer`	str
`observed`	str
`conclusion_count`	int
`levels`	list[str]	E.g. `["explicit", "deductive", "deductive"]`

Resource ID: {run_id}:{iteration}:conclusions_created

`agent.tool.conclusions.deleted`

Category: agent · Schema: v3

Emitted when the delete_conclusions agent tool executes — typically during dream consolidation.

Field	Type	Description
`run_id`	str
`iteration`	int
`parent_category`	str	Typically `"dream"`
`agent_type`	str	Typically `"deduction"`
`workspace_name`	str
`observer`	str
`observed`	str
`conclusion_count`	int
`levels`	list[str]

Resource ID: {run_id}:{iteration}:conclusions_deleted

`agent.tool.peer_card.updated`

Category: agent · Schema: v2

Emitted when the update_peer_card agent tool executes during dream consolidation.

Field	Type	Description
`run_id`	str
`iteration`	int
`parent_category`	str	Typically `"dream"`
`agent_type`	str	`"deduction"` or `"induction"`
`workspace_name`	str
`observer`	str
`observed`	str
`facts_count`	int	Number of facts in the updated peer card

Resource ID: {run_id}:{iteration}:peer_card_updated

`agent.tool.summary.created`

Category: agent · Schema: v3

Emitted when a session summary is created. run_id / iteration are null when summarization runs outside an agentic loop.

Field	Type	Description
`run_id`	str \| null	Null when not in an agentic run
`iteration`	int \| null	Null when not in an agentic run
`parent_category`	str
`agent_type`	str
`workspace_name`	str
`session_name`	str
`message_id`	str	Message the summary covers up to
`message_count`	int	Messages summarized
`message_seq_in_session`	int	Sequence number of the base message
`summary_type`	str	`"short"` or `"long"`
`input_tokens`	int	Provider-side input tokens for the summary LLM call
`output_tokens`	int	Summary token count
`previous_summary_tokens`	int	Tokens of the previous summary fed back as context (0 if first)
`message_tokens`	int	Sum of `Message.token_count` across summarized messages
`prompt_scaffold_tokens`	int	Estimated tokens for the static scaffold portion of the prompt

Resource ID: {message_id}:{summary_type}:summary_created

`agent.tool.call.completed`

Category: agent · Schema: v1 · Volume: high (sampled) · Source file: src/telemetry/events/agent.py

Lightweight per-call record for every tool the Dialectic/Dreamer agents invoke — including read-only tools (search_memory, get_recent_history, …) that have no dedicated event. One per tool call.

Field	Type	Description
`run_id`	str	Run id for correlation
`iteration`	int	Iteration number (1-indexed)
`tool_call_seq`	int	0-indexed position in the iteration’s tool batch
`provider_tool_call_id`	str \| null	Provider-supplied tool call id (e.g. Anthropic `toolu_*`)
`parent_category`	str	`"dream"` or `"dialectic"`
`agent_type`	str	`"deduction"`, `"induction"`, or `"dialectic"`
`workspace_name`	str
`tool_name`	str	Tool as invoked
`duration_ms`	float	Handler wall-clock duration
`is_error`	bool	True if the handler raised
`result_chars`	int	Length of the result string returned to the LLM
`result_chars_before_truncation`	int \| null	Original size when truncated; null otherwise
`result_tokens_estimate`	int	tiktoken-based size proxy (estimate)
`was_truncated`	bool	True when the result was clamped to a size budget
`query_tokens`	int \| null	Search tools only — tiktoken estimate of the query
`top_k`	int \| null	Search tools only
`results_count`	int \| null	Search tools only — results returned
`used_embedding`	bool \| null	Search tools only — true on vector lookup vs. metadata filter
`embedding_query_count`	int	Embedding API calls the handler made

Resource ID: {run_id}:{iteration}:{tool_call_seq}

`deletion.completed`

Category: deletion · Schema: v1 · Source file: src/telemetry/events/deletion.py

Emitted when an async deletion task completes (workspace, session, or conclusions). For workspace/session deletions, cascade counts are populated.

Field	Type	Description
`workspace_name`	str
`deletion_type`	str	`"workspace"`, `"session"`, or `"conclusions"`
`resource_id`	str	ID of the deleted resource
`success`	bool
`peers_deleted`	int	Workspace deletions only
`sessions_deleted`	int	Workspace deletions only
`messages_deleted`	int	Workspace or session deletions
`conclusions_deleted`	int	Workspace or session deletions
`error_message`	str \| null	Populated on failure

Resource ID: {workspace_name}:{deletion_type}:{resource_id}

`reconciliation.sync_vectors.completed`

Category: reconciliation · Schema: v1 · Source file: src/telemetry/events/reconciliation.py

Periodic global maintenance — syncs documents and message embeddings to external vector stores. No workspace context (operates across all workspaces in the instance).

Field	Type	Description
`documents_synced`	int
`documents_failed`	int
`documents_cleaned`	int	Soft-deleted docs cleaned during sync
`message_embeddings_synced`	int
`message_embeddings_failed`	int
`total_duration_ms`	float

Resource ID: sync_vectors (fixed)

`reconciliation.cleanup_stale_items.completed`

Category: reconciliation · Schema: v1

Periodic global cleanup of soft-deleted documents and expired queue items. No workspace context.

Field	Type	Description
`documents_cleaned`	int
`queue_items_cleaned`	int
`total_duration_ms`	float

Resource ID: cleanup_stale_items (fixed)

`message.created`

Category: api · Schema: v1 · Source file: src/telemetry/events/api.py

Canonical event for counting created messages (including messages created from file uploads). One per create batch.

Field	Type	Description
`workspace_name`	str
`session_name`	str
`message_count`	int	Messages created
`total_tokens`	int	Total tokens across the batch
`source`	str	`"api"` or `"file_upload"`
`last_message_id`	str	public_id of the trailing message (stable batch key)

Resource ID: {workspace_name}:{session_name}:{source}:{last_message_id}

`file.uploaded`

Category: api · Schema: v1 · Source file: src/telemetry/events/api.py

Emitted when an uploaded file is converted into messages — captures file-side metadata. Use message.created for message counts (avoids double-counting).

Field	Type	Description
`workspace_name`	str
`session_name`	str
`peer_name`	str	Peer that uploaded the file
`file_id`	str	Generated file identifier
`filename`	str \| null
`content_type`	str \| null
`file_size_bytes`	int \| null
`message_count`	int	Messages created from the file
`total_tokens`	int	Tokens across those messages

Resource ID: {workspace_name}:{session_name}:{file_id}

`context.retrieved`

Category: api · Schema: v1 · Source file: src/telemetry/events/api.py

Emitted when context is retrieved for a session or peer (the get_context endpoints). Captures what was requested and what was returned.

Field	Type	Description
`workspace_name`	str
`context_scope`	str	`"session"` or `"peer"`
`session_name`	str \| null	Session-scoped only
`peer_name`	str \| null	Observer peer
`target_name`	str \| null	Observed peer
`tokens_requested`	int \| null	Caller’s `tokens` param (null = endpoint default)
`message_count`	int	Messages returned
`has_summary`	bool
`has_representation`	bool
`has_peer_card`	bool
`search_query_provided`	bool
`search_top_k`	int \| null
`search_max_distance`	float \| null
`include_most_frequent`	bool \| null
`max_conclusions`	int \| null
`include_summary`	bool \| null
`limit_to_session`	bool	Representation retrieval was session-limited
`peer_perspective_provided`	bool	`peer_perspective` supplied for session context
`total_duration_ms`	float

Resource ID: session → {workspace}:session:{session}:{peer}:{target} · peer → {workspace}:peer:{peer}:{target}

`llm.call.completed`

Category: llm · Schema: v1 · Volume: high (sampled) · Source file: src/telemetry/events/llm.py

One per provider hit (success, error, or cancellation), with full cost-attribution context. Streaming calls report token counts as 0 placeholders — use the aggregate envelopes for streamed-call accuracy.

Field	Type	Description
`workspace_name`	str \| null	Null for system calls without workspace context
`call_purpose`	str \| null	Closed enum: `deriver.representation`, `dialectic.answer`, `dream.deduction`, `dream.induction`, `summary.short`, `summary.long`
`parent_category`	str \| null	`representation` \| `dialectic` \| `dream` \| `summary`
`transport`	str	SDK transport: `anthropic` \| `openai` \| `gemini`
`provider_label`	str \| null	Best-effort vendor for relay setups (e.g. OpenRouter); null when unclear
`model`	str	Model id as sent to the provider
`effective_max_output_tokens`	int	`max_tokens` used
`provider_input_tokens`	int	0 on stream placeholder
`provider_output_tokens`	int	0 on stream placeholder
`cache_read_tokens`	int	Tokens read from prompt cache
`cache_creation_tokens`	int	Tokens written to prompt cache
`finish_reason`	str \| null
`outcome`	str	`success` \| `error` \| `cancelled` (exclude `cancelled` from error alerts)
`is_final_attempt`	bool	True on the last allowed attempt; pair with `outcome='error'` for exhausted
`error_class`	str \| null	Exception class on error/cancellation
`attempt`	int	1-indexed attempt number
`retry_attempts`	int	Total attempts allowed
`was_fallback`	bool	True when this attempt used the fallback model
`duration_ms`	float	Provider call wall-clock
`has_tools`	bool
`tool_call_count`	int	Tool calls the model requested
`was_stream`	bool	True for the streaming path (token counts are placeholders)
`run_id`	str \| null	Agent run id; null for non-agent calls
`iteration`	int \| null	Iteration within an agentic loop

Resource ID: {run_id|none}:{iteration|0}:{attempt}:{transport}:{model}

`embedding.call.completed`

Category: llm · Schema: v1 · Volume: high (sampled) · Source file: src/telemetry/events/llm.py

One per embedding-provider call (real per-token spend). Search tools, observation creation, embedding sync, and deriver/summarizer paths all hit this.

Field	Type	Description
`workspace_name`	str \| null
`call_purpose`	str \| null	Closed enum: `search_memory`, `search_messages`, `create_observations`, `vector_sync`, `summary`, …
`parent_category`	str \| null	e.g. `dialectic`, `representation`
`provider`	str	`openai` \| `gemini`
`model`	str
`input_count`	int	Texts embedded in this call (batch size)
`input_tokens_estimate`	int	tiktoken-based proxy (estimate only)
`duration_ms`	float
`outcome`	str	`success` \| `error` \| `cancelled`
`is_final_attempt`	bool
`error_class`	str \| null
`run_id`	str \| null	Agent run id; null for sync/CRUD paths

Resource ID: {run_id|none}:{call_purpose|unknown}:{provider}:{model}:{input_count}

`llm.call.traced`

Category: trace · Schema: v1 · Volume: ground-truth (never sampled) · Source file: src/telemetry/events/trace.py

Opt-in replay-grade record, one per LLM call (gated by TELEMETRY.TRACE_PAYLOADS_ENABLED). Content fields are references (content hashes into the trace.content store), never inline bytes. Deliberately not joined to llm.call.completed — cost is computable from the trace alone.

Field	Type	Description
`trace_id`	str \| null	Span-tree correlation
`span_id`	str \| null
`parent_span_id`	str \| null
`iteration`	int \| null
`step_seq`	int	Default 0
`attempt`	int	1-indexed attempt
`was_fallback`	bool
`parent_event_id`	str \| null
`call_purpose`	str \| null
`parent_category`	str \| null
`session_id`	str \| null	Used for grouping traces
`transport`	str	`anthropic` / `openai` / `gemini`
`provider_label`	str \| null
`model`	str
`input_message_refs`	list[str]	Content hashes for the context window
`system_prompt_ref`	str \| null	Content hash
`tool_schema_refs`	list[str]	Content hashes
`tool_choice`	any
`output_content_ref`	str \| null	Content hash of replay-grade output
`output_tool_calls`	list[dict]
`output_thinking_ref`	str \| null	Content hash
`output_signatures`	list[str]
`raw_response_ref`	str \| null	Reserved
`finish_reason`	str \| null
`provider_input_tokens`	int	Self-contained accounting copy
`provider_output_tokens`	int
`cache_read_tokens`	int
`cache_creation_tokens`	int
`was_truncated`	bool

Resource ID: {span_id}:{iteration}:{attempt}:{step_seq}

`embedding.call.traced`

Category: trace · Schema: v1 · Volume: ground-truth (never sampled) · Source file: src/telemetry/events/trace.py

Trace-stream record, one per embedding-provider call (gated by TELEMETRY.TRACE_PAYLOADS_ENABLED).

Field	Type	Description
`trace_id`	str \| null	Span-tree correlation
`span_id`	str \| null
`parent_span_id`	str \| null
`iteration`	int \| null
`step_seq`	int	Default 0
`attempt`	int	1-indexed attempt
`session_id`	str \| null
`call_purpose`	str \| null
`parent_category`	str \| null
`provider`	str
`model`	str
`provider_input_tokens`	int	v1: tiktoken estimate (no authoritative count plumbed)
`provider_output_tokens`	int	Always 0 (embeddings produce none)
`input_count`	int	Texts embedded in this call
`was_truncated`	bool

Resource ID: {span_id}:embedding:{call_purpose}:{input_count}

`trace.content`

Category: trace · Schema: v1 · Volume: ground-truth (never sampled) · Source file: src/telemetry/events/trace.py

One unique message in the content store, emitted once per run and referenced by content_hash. The CloudEvent id is content-addressed with no timestamp (content_{digest}), so re-sends of identical content dedupe at the transport layer.

Field	Type	Description
`content_hash`	str	Covers full message identity `{role, content, tool_call_id}`
`role`	str
`content`	any \| null	Message text, normalized across providers
`tool_call_id`	str \| null
`tool_calls`	list[dict]	Unified `{id, name, input}` shape (provider-agnostic)
`honcho_authored`	bool	Tags Honcho-authored content (system prompts, scaffold) for access control

Resource ID: {content_hash}

Quick reference

Event type	Category	Schema	Resource ID pattern
`representation.completed`	representation	v2	`{workspace}:{session}:{latest_message_id}`
`dream.run`	dream	v2	`{run_id}`
`dream.specialist`	dream	v2	`{run_id}:{specialist_type}`
`dialectic.completed`	dialectic	v2	`{run_id}`
`agent.iteration`	agent	v2	`{run_id}:{iteration}`
`agent.tool.conclusions.created`	agent	v2	`{run_id}:{iteration}:conclusions_created`
`agent.tool.conclusions.deleted`	agent	v3	`{run_id}:{iteration}:conclusions_deleted`
`agent.tool.peer_card.updated`	agent	v2	`{run_id}:{iteration}:peer_card_updated`
`agent.tool.summary.created`	agent	v3	`{message_id}:{summary_type}:summary_created`
`agent.tool.call.completed`	agent	v1	`{run_id}:{iteration}:{tool_call_seq}`
`deletion.completed`	deletion	v1	`{workspace}:{deletion_type}:{resource_id}`
`reconciliation.sync_vectors.completed`	reconciliation	v1	`sync_vectors`
`reconciliation.cleanup_stale_items.completed`	reconciliation	v1	`cleanup_stale_items`
`message.created`	api	v1	`{workspace}:{session}:{source}:{last_message_id}`
`file.uploaded`	api	v1	`{workspace}:{session}:{file_id}`
`context.retrieved`	api	v1	`{workspace}:{scope}:…:{peer}:{target}`
`llm.call.completed`	llm	v1	`{run_id}:{iteration}:{attempt}:{transport}:{model}`
`embedding.call.completed`	llm	v1	`{run_id}:{call_purpose}:{provider}:{model}:{input_count}`
`llm.call.traced`	trace	v1	`{span_id}:{iteration}:{attempt}:{step_seq}`
`embedding.call.traced`	trace	v1	`{span_id}:embedding:{call_purpose}:{input_count}`
`trace.content`	trace	v1	`{content_hash}`

Querying the Parquet archive

Storage layout

Xatu’s consumer writes batches to S3 with the following key layout (see xatu/consumer/parquet_writer.py):

s3://{XATU_S3_BUCKET}/year={YYYY}/month={MM}/day={DD}/{content_hash}.parquet

content_hash is a 12-char SHA-256 prefix over all event IDs in the batch — same batch always overwrites the same key (idempotent within a batch; cross-batch dupes are still possible and should be deduplicated in queries via DISTINCT id).
Compression: zstd. Statistics are written.

Parquet schema

All columns are strings — data is a serialized JSON string, not a typed struct.

Column	Type	Notes
`specversion`	string	Always `"1.0"`
`id`	string	Deterministic; use `DISTINCT id` to dedupe
`source`	string	`/honcho/{namespace}/{category}`
`type`	string	Event type — filter on this
`time`	string	ISO-8601 UTC; cast with `CAST(time AS TIMESTAMP)`
`datacontenttype`	string
`dataschema`	string	`https://honcho.dev/schemas/{type}/v{n}` — parse for schema version
`data`	string	JSON payload; use `json_extract` / `data->>`
`namespace`	string	Fly app name (== `HonchoInstance.app_name`)
`tenant_id`	string	Groudon tenant — primary partition key for billing queries

Querying with DuckDB

Best for local ad-hoc analysis from a laptop or notebook. Install DuckDB (brew install duckdb), configure S3 credentials, and read the Hive-partitioned Parquet directly.

Setup

INSTALL httpfs; LOAD httpfs;
SET s3_region='us-east-1';
SET s3_access_key_id='...';
SET s3_secret_access_key='...';
 
-- Define a reusable view across all partitions.
CREATE OR REPLACE VIEW events AS
SELECT *
FROM read_parquet(
    's3://xatu-events/year=*/month=*/day=*/*.parquet',
    hive_partitioning = true
);

With hive_partitioning=true, the year, month, day partition keys become first-class columns — push them down in WHERE clauses to prune scans:

WHERE year = '2026' AND month IN ('04', '05')

Recipes

1. Distinct event count for a tenant in May 2026

SELECT COUNT(DISTINCT id) AS events
FROM events
WHERE year = '2026' AND month = '05'
  AND tenant_id = 'tnt_abc123';

2. Token totals by tenant, last 7 days

SELECT
    tenant_id,
    SUM(CAST(json_extract_string(data, '$.input_tokens')  AS BIGINT)) AS input_tokens,
    SUM(CAST(json_extract_string(data, '$.output_tokens') AS BIGINT)) AS output_tokens
FROM events
WHERE year = '2026' AND month = '05'
  AND type IN ('representation.completed', 'dialectic.completed', 'dream.specialist')
  AND CAST(time AS TIMESTAMP) >= now() - INTERVAL 7 DAY
GROUP BY tenant_id
ORDER BY input_tokens DESC;

3. Dialectic latency percentiles by reasoning level

SELECT
    json_extract_string(data, '$.reasoning_level') AS reasoning_level,
    COUNT(*) AS calls,
    QUANTILE_CONT(CAST(json_extract_string(data, '$.total_duration_ms') AS DOUBLE), 0.50) AS p50_ms,
    QUANTILE_CONT(CAST(json_extract_string(data, '$.total_duration_ms') AS DOUBLE), 0.95) AS p95_ms,
    QUANTILE_CONT(CAST(json_extract_string(data, '$.total_duration_ms') AS DOUBLE), 0.99) AS p99_ms
FROM events
WHERE type = 'dialectic.completed'
  AND year = '2026' AND month = '05'
GROUP BY reasoning_level
ORDER BY p95_ms DESC;

4. Dream success rate per tenant

SELECT
    tenant_id,
    COUNT(*) AS runs,
    SUM(CASE WHEN json_extract_string(data, '$.deduction_success') = 'true' THEN 1 ELSE 0 END) * 1.0 / COUNT(*) AS deduction_rate,
    SUM(CASE WHEN json_extract_string(data, '$.induction_success') = 'true' THEN 1 ELSE 0 END) * 1.0 / COUNT(*) AS induction_rate
FROM events
WHERE type = 'dream.run'
  AND year = '2026' AND month = '05'
GROUP BY tenant_id
ORDER BY runs DESC;

5. Cross-event correlation by run_id — full dialectic breakdown

WITH dialectic AS (
    SELECT
        json_extract_string(data, '$.run_id')         AS run_id,
        tenant_id,
        json_extract_string(data, '$.peer_name')      AS peer_name,
        CAST(json_extract_string(data, '$.total_duration_ms') AS DOUBLE) AS total_ms
    FROM events
    WHERE type = 'dialectic.completed' AND year = '2026' AND month = '05'
),
iterations AS (
    SELECT
        json_extract_string(data, '$.run_id')        AS run_id,
        COUNT(*)                                     AS iters,
        SUM(CAST(json_extract_string(data, '$.input_tokens')  AS BIGINT)) AS in_tokens,
        SUM(CAST(json_extract_string(data, '$.output_tokens') AS BIGINT)) AS out_tokens
    FROM events
    WHERE type = 'agent.iteration' AND year = '2026' AND month = '05'
      AND json_extract_string(data, '$.parent_category') = 'dialectic'
    GROUP BY 1
)
SELECT d.*, i.iters, i.in_tokens, i.out_tokens
FROM dialectic d
LEFT JOIN iterations i USING (run_id)
ORDER BY d.total_ms DESC
LIMIT 50;

6. Deletion audit — which workspaces have been removed and what cascaded

SELECT
    CAST(time AS TIMESTAMP)                                            AS at,
    tenant_id,
    json_extract_string(data, '$.workspace_name')                      AS workspace,
    json_extract_string(data, '$.deletion_type')                       AS deletion_type,
    json_extract_string(data, '$.success')                             AS success,
    CAST(json_extract_string(data, '$.peers_deleted')       AS BIGINT) AS peers,
    CAST(json_extract_string(data, '$.sessions_deleted')    AS BIGINT) AS sessions,
    CAST(json_extract_string(data, '$.messages_deleted')    AS BIGINT) AS messages,
    CAST(json_extract_string(data, '$.conclusions_deleted') AS BIGINT) AS conclusions,
    json_extract_string(data, '$.error_message')                       AS error
FROM events
WHERE type = 'deletion.completed' AND year = '2026'
ORDER BY at DESC;

7. Reconciliation health — failed vector syncs over time

SELECT
    DATE_TRUNC('day', CAST(time AS TIMESTAMP)) AS day,
    namespace,
    SUM(CAST(json_extract_string(data, '$.documents_failed')           AS BIGINT)) AS doc_failures,
    SUM(CAST(json_extract_string(data, '$.message_embeddings_failed')  AS BIGINT)) AS msg_failures
FROM events
WHERE type = 'reconciliation.sync_vectors.completed'
  AND year = '2026'
GROUP BY 1, 2
HAVING doc_failures + msg_failures > 0
ORDER BY day DESC, doc_failures + msg_failures DESC;

DuckDB tips

Always filter by partition keys (year, month, day) first — they prune entire S3 prefixes from the scan.
Always DISTINCT id when computing counts — cross-batch retries can produce duplicates.
Cast JSON-extracted numerics — json_extract_string returns text; wrap with CAST(... AS BIGINT) / CAST(... AS DOUBLE).
Booleans round-trip as the strings "true" / "false" because data is a JSON-serialized string column.

Querying with ClickHouse

Best when you want the archive available to dashboards alongside the rest of the analytics stack. Two ergonomic options:

s3() table function for ad-hoc reads (no schema migration required).
S3 engine table for repeatable dashboard queries, optionally fronted by a MergeTree materialization for low-latency joins.

ClickHouse’s JSON functions are typed at the call site (JSONExtractString, JSONExtractInt, JSONExtractFloat, JSONExtractBool) — no CAST needed after extraction.

Setup

Ad-hoc reads via s3() — requires use_hive_partitioning=1 (ClickHouse 24.x+) so year/month/day become first-class columns:

SET use_hive_partitioning = 1;
 
SELECT count()
FROM s3(
    'https://xatu-events.s3.us-east-1.amazonaws.com/year=*/month=*/day=*/*.parquet',
    '<aws_access_key_id>',
    '<aws_secret_access_key>',
    'Parquet'
)
WHERE year = '2026' AND month = '05';

Reusable external table:

CREATE TABLE events_s3
(
    specversion      String,
    id               String,
    source           String,
    type             String,
    time             String,
    datacontenttype  String,
    dataschema       Nullable(String),
    data             String,
    namespace        String,
    tenant_id        String,
    year             String,
    month            String,
    day              String
)
ENGINE = S3(
    'https://xatu-events.s3.us-east-1.amazonaws.com/year=*/month=*/day=*/*.parquet',
    '<aws_access_key_id>',
    '<aws_secret_access_key>',
    'Parquet'
)
SETTINGS use_hive_partitioning = 1;
 
CREATE VIEW events AS SELECT * FROM events_s3;

Optional MergeTree materialization for fast repeat queries (project the JSON fields you actually use as typed columns):

CREATE TABLE events_mt
(
    event_time     DateTime64(3) DEFAULT parseDateTime64BestEffort(time, 3),
    id             String,
    type           LowCardinality(String),
    tenant_id      LowCardinality(String),
    namespace      LowCardinality(String),
    source         String,
    dataschema     Nullable(String),
    data           String,
    -- projected for common filters:
    workspace_name LowCardinality(String) MATERIALIZED JSONExtractString(data, 'workspace_name'),
    run_id         String                  MATERIALIZED JSONExtractString(data, 'run_id'),
    input_tokens   UInt64                  MATERIALIZED JSONExtractUInt(data, 'input_tokens'),
    output_tokens  UInt64                  MATERIALIZED JSONExtractUInt(data, 'output_tokens')
)
ENGINE = ReplacingMergeTree(event_time)
PARTITION BY toYYYYMM(event_time)
ORDER BY (tenant_id, type, event_time, id);
 
-- Backfill or incremental load
INSERT INTO events_mt (event_time, id, type, tenant_id, namespace, source, dataschema, data)
SELECT parseDateTime64BestEffort(time, 3), id, type, tenant_id, namespace, source, dataschema, data
FROM events_s3
WHERE year = '2026' AND month = '05';

ReplacingMergeTree(event_time) keyed on (tenant_id, type, event_time, id) collapses cross-batch duplicates on merge. Use FINAL or SELECT DISTINCT id for exact counts before merges complete.

Recipes

These assume events is the view defined above. If you materialized to events_mt, drop the JSONExtract* calls in favor of the typed projected columns.

1. Distinct event count for a tenant in May 2026

SELECT uniqExact(id) AS events
FROM events
WHERE year = '2026' AND month = '05'
  AND tenant_id = 'tnt_abc123';

2. Token totals by tenant, last 7 days

SELECT
    tenant_id,
    sum(JSONExtractUInt(data, 'input_tokens'))  AS input_tokens,
    sum(JSONExtractUInt(data, 'output_tokens')) AS output_tokens
FROM events
WHERE year = '2026' AND month = '05'
  AND type IN ('representation.completed', 'dialectic.completed', 'dream.specialist')
  AND parseDateTime64BestEffort(time, 3) >= now() - INTERVAL 7 DAY
GROUP BY tenant_id
ORDER BY input_tokens DESC;

3. Dialectic latency percentiles by reasoning level

SELECT
    JSONExtractString(data, 'reasoning_level') AS reasoning_level,
    count() AS calls,
    quantile(0.50)(JSONExtractFloat(data, 'total_duration_ms')) AS p50_ms,
    quantile(0.95)(JSONExtractFloat(data, 'total_duration_ms')) AS p95_ms,
    quantile(0.99)(JSONExtractFloat(data, 'total_duration_ms')) AS p99_ms
FROM events
WHERE type = 'dialectic.completed'
  AND year = '2026' AND month = '05'
GROUP BY reasoning_level
ORDER BY p95_ms DESC;

4. Dream success rate per tenant

SELECT
    tenant_id,
    count() AS runs,
    avg(JSONExtractBool(data, 'deduction_success')) AS deduction_rate,
    avg(JSONExtractBool(data, 'induction_success')) AS induction_rate
FROM events
WHERE type = 'dream.run'
  AND year = '2026' AND month = '05'
GROUP BY tenant_id
ORDER BY runs DESC;

JSONExtractBool returns 0/1, so avg(...) is the success rate directly.

5. Cross-event correlation by run_id — full dialectic breakdown

WITH
    dialectic AS (
        SELECT
            JSONExtractString(data, 'run_id')              AS run_id,
            tenant_id,
            JSONExtractString(data, 'peer_name')           AS peer_name,
            JSONExtractFloat(data, 'total_duration_ms')    AS total_ms
        FROM events
        WHERE type = 'dialectic.completed' AND year = '2026' AND month = '05'
    ),
    iterations AS (
        SELECT
            JSONExtractString(data, 'run_id')          AS run_id,
            count()                                    AS iters,
            sum(JSONExtractUInt(data, 'input_tokens'))  AS in_tokens,
            sum(JSONExtractUInt(data, 'output_tokens')) AS out_tokens
        FROM events
        WHERE type = 'agent.iteration' AND year = '2026' AND month = '05'
          AND JSONExtractString(data, 'parent_category') = 'dialectic'
        GROUP BY run_id
    )
SELECT d.*, i.iters, i.in_tokens, i.out_tokens
FROM dialectic AS d
LEFT JOIN iterations AS i USING (run_id)
ORDER BY total_ms DESC
LIMIT 50;

6. Deletion audit — which workspaces have been removed and what cascaded

SELECT
    parseDateTime64BestEffort(time, 3)                     AS at,
    tenant_id,
    JSONExtractString(data, 'workspace_name')              AS workspace,
    JSONExtractString(data, 'deletion_type')               AS deletion_type,
    JSONExtractBool(data, 'success')                       AS success,
    JSONExtractUInt(data, 'peers_deleted')                 AS peers,
    JSONExtractUInt(data, 'sessions_deleted')              AS sessions,
    JSONExtractUInt(data, 'messages_deleted')              AS messages,
    JSONExtractUInt(data, 'conclusions_deleted')           AS conclusions,
    JSONExtractString(data, 'error_message')               AS error
FROM events
WHERE type = 'deletion.completed' AND year = '2026'
ORDER BY at DESC;

7. Reconciliation health — failed vector syncs over time

SELECT
    toStartOfDay(parseDateTime64BestEffort(time, 3)) AS day,
    namespace,
    sum(JSONExtractUInt(data, 'documents_failed'))          AS doc_failures,
    sum(JSONExtractUInt(data, 'message_embeddings_failed')) AS msg_failures
FROM events
WHERE type = 'reconciliation.sync_vectors.completed'
  AND year = '2026'
GROUP BY day, namespace
HAVING doc_failures + msg_failures > 0
ORDER BY day DESC, doc_failures + msg_failures DESC;

ClickHouse tips

use_hive_partitioning = 1 is required to get year/month/day as columns. Without it, partition values are buried in the file path and you’d be glob-only.
uniqExact(id) is the safe count when reading directly from s3() (cross-batch dedupe). count() is fine on events_mt after OPTIMIZE TABLE ... FINAL or with SELECT count() FROM events_mt FINAL.
JSONExtract* is typed at call site — no follow-up CAST. Use JSONExtractInt / JSONExtractUInt / JSONExtractFloat / JSONExtractBool / JSONExtractString.
JSONExtractBool returns 0/1, so avg(...) directly gives the rate.
For dashboards, project hot JSON fields as MATERIALIZED columns on events_mt. The MergeTree ordering key (tenant_id, type, event_time, id) makes per-tenant filters effectively a primary-key scan.
Timestamp parsing: parseDateTime64BestEffort(time, 3) for millisecond precision; persist as DateTime64(3) if materializing.

General tips (both engines)

Always filter by year/month/day first — these are partition keys, not data columns; filtering on them prunes whole S3 prefixes.
Dedupe by id when counting. Same-batch retries hit the same Parquet key (idempotent), but cross-batch retries can produce duplicate rows across files.
Booleans live as the JSON strings "true" / "false" inside data. DuckDB needs string comparison; ClickHouse JSONExtractBool handles it natively.
Schema versions are encoded in the dataschema column (https://honcho.dev/schemas/{type}/v{n}). Filter on it when a field’s semantics changed across versions.
Trino / Athena equivalents: json_extract_scalar(data, '$.foo') mirrors DuckDB’s json_extract_string; external tables stand in for the events_s3 pattern.
Local dev: Xatu’s xatu/scripts/load_test.py seeds synthetic events — useful for iterating on query shapes without hitting prod.

Plastic Labs Internal Wiki

Explorer

Recently Updated

Company

Operating Model — Decisions Log

Operating Model

Security & Compliance

ideal-reasoned-representations

Honcho CloudEvents Glossary

CloudEvent envelope

Volume & sampling

Trace payloads (full-fidelity stream)

Event catalog

representation.completed

dream.run

dream.specialist

dialectic.completed

agent.iteration

agent.tool.conclusions.created

agent.tool.conclusions.deleted

agent.tool.peer_card.updated

agent.tool.summary.created

agent.tool.call.completed

deletion.completed

reconciliation.sync_vectors.completed

reconciliation.cleanup_stale_items.completed

message.created

file.uploaded

context.retrieved

llm.call.completed

embedding.call.completed

llm.call.traced

embedding.call.traced

trace.content

Quick reference

Querying the Parquet archive

Storage layout

Parquet schema

Querying with DuckDB

Setup

Recipes

DuckDB tips

Querying with ClickHouse

Setup

Recipes

ClickHouse tips

General tips (both engines)

Graph View

Table of Contents

Backlinks

`representation.completed`

`dream.run`

`dream.specialist`

`dialectic.completed`

`agent.iteration`

`agent.tool.conclusions.created`

`agent.tool.conclusions.deleted`

`agent.tool.peer_card.updated`

`agent.tool.summary.created`

`agent.tool.call.completed`

`deletion.completed`

`reconciliation.sync_vectors.completed`

`reconciliation.cleanup_stale_items.completed`

`message.created`

`file.uploaded`

`context.retrieved`

`llm.call.completed`

`embedding.call.completed`

`llm.call.traced`

`embedding.call.traced`

`trace.content`