pond/
wire.rs

1use std::collections::BTreeMap;
2
3use chrono::{DateTime, Utc};
4use serde::{Deserialize, Serialize};
5use serde_json::{Map, Value};
6use uuid::Uuid;
7
8use crate::PROTOCOL_VERSION;
9use crate::adapter::Extracted;
10
11pub type ProviderOptions = BTreeMap<String, Value>;
12
13#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
14pub struct Session {
15    pub id: String,
16    #[serde(skip_serializing_if = "Option::is_none")]
17    pub parent_session_id: Option<String>,
18    /// spec.md#model-parent-pointer-coherence: when set, `parent_session_id`
19    /// MUST also be set. Spawn-only sources (claude-code subagents,
20    /// nanoclaw) leave this `None`; fork-with-cut-point sources
21    /// (pi-coding-agent) populate both pointers together.
22    #[serde(skip_serializing_if = "Option::is_none")]
23    pub parent_message_id: Option<String>,
24    pub source_agent: String,
25    pub created_at: DateTime<Utc>,
26    pub project: Extracted<String>,
27    #[serde(default)]
28    pub options: ProviderOptions,
29}
30
31#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
32#[serde(tag = "role", rename_all = "snake_case")]
33pub enum Message {
34    System {
35        id: String,
36        session_id: String,
37        timestamp: DateTime<Utc>,
38        /// `None` when the source row carried no content. The seal on
39        /// `Extracted<String>` means adapters CANNOT pass a synthesized
40        /// or sentinel string here - the value either flows from a
41        /// `Source` extraction or the field is `None`. Distinguishes
42        /// "source said content=''" (Some(extracted_empty)) from
43        /// "source had no content field" (None).
44        #[serde(default, skip_serializing_if = "Option::is_none")]
45        content: Option<Extracted<String>>,
46        #[serde(default)]
47        options: ProviderOptions,
48    },
49    User {
50        id: String,
51        session_id: String,
52        timestamp: DateTime<Utc>,
53        #[serde(default)]
54        options: ProviderOptions,
55    },
56    Assistant {
57        id: String,
58        session_id: String,
59        timestamp: DateTime<Utc>,
60        #[serde(default)]
61        options: ProviderOptions,
62    },
63    Tool {
64        id: String,
65        session_id: String,
66        timestamp: DateTime<Utc>,
67        #[serde(default)]
68        options: ProviderOptions,
69    },
70}
71
72impl Message {
73    pub fn id(&self) -> &str {
74        match self {
75            Self::System { id, .. }
76            | Self::User { id, .. }
77            | Self::Assistant { id, .. }
78            | Self::Tool { id, .. } => id,
79        }
80    }
81
82    pub fn session_id(&self) -> &str {
83        match self {
84            Self::System { session_id, .. }
85            | Self::User { session_id, .. }
86            | Self::Assistant { session_id, .. }
87            | Self::Tool { session_id, .. } => session_id,
88        }
89    }
90
91    pub fn role(&self) -> Role {
92        match self {
93            Self::System { .. } => Role::System,
94            Self::User { .. } => Role::User,
95            Self::Assistant { .. } => Role::Assistant,
96            Self::Tool { .. } => Role::Tool,
97        }
98    }
99
100    pub fn timestamp(&self) -> DateTime<Utc> {
101        match self {
102            Self::System { timestamp, .. }
103            | Self::User { timestamp, .. }
104            | Self::Assistant { timestamp, .. }
105            | Self::Tool { timestamp, .. } => *timestamp,
106        }
107    }
108
109    pub fn options(&self) -> &ProviderOptions {
110        match self {
111            Self::System { options, .. }
112            | Self::User { options, .. }
113            | Self::Assistant { options, .. }
114            | Self::Tool { options, .. } => options,
115        }
116    }
117
118    pub fn options_mut(&mut self) -> &mut ProviderOptions {
119        match self {
120            Self::System { options, .. }
121            | Self::User { options, .. }
122            | Self::Assistant { options, .. }
123            | Self::Tool { options, .. } => options,
124        }
125    }
126
127    pub fn system_content(&self) -> Option<&str> {
128        match self {
129            // Two layers of `as_deref`: the outer `Option<Extracted<String>>`
130            // becomes `Option<&Extracted<String>>`, then `Extracted: Deref`
131            // unwraps to `&str`.
132            Self::System { content, .. } => content.as_deref().map(|e| &**e),
133            Self::User { .. } | Self::Assistant { .. } | Self::Tool { .. } => None,
134        }
135    }
136}
137
138#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
139#[serde(rename_all = "snake_case")]
140pub enum Role {
141    System,
142    User,
143    Assistant,
144    Tool,
145}
146
147impl Role {
148    pub fn as_str(self) -> &'static str {
149        match self {
150            Self::System => "system",
151            Self::User => "user",
152            Self::Assistant => "assistant",
153            Self::Tool => "tool",
154        }
155    }
156}
157
158/// Whether a Part's content is conversation or harness-injected scaffolding
159/// (spec.md#model-part-provenance). No `Default` and no `#[serde(default)]` on the
160/// `Part.provenance` field below: constructing a Part without classifying it
161/// MUST be a compile error (spec.md#adapter-provenance-required).
162#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
163#[serde(rename_all = "snake_case")]
164pub enum Provenance {
165    Conversational,
166    Injected,
167}
168
169impl Provenance {
170    pub fn as_str(self) -> &'static str {
171        match self {
172            Self::Conversational => "conversational",
173            Self::Injected => "injected",
174        }
175    }
176}
177
178#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
179pub struct Part {
180    pub session_id: String,
181    pub id: String,
182    pub message_id: String,
183    pub ordinal: i32,
184    /// Conversation vs harness-injected (spec.md#model-part-provenance). Mandatory,
185    /// no serde default - search reads it to exclude injected scaffolding.
186    pub provenance: Provenance,
187    #[serde(default)]
188    pub options: ProviderOptions,
189    #[serde(flatten)]
190    pub kind: PartKind,
191}
192
193#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
194#[serde(tag = "type", rename_all = "snake_case")]
195pub enum PartKind {
196    Text {
197        /// `None` when the source row had no text field. The seal on
198        /// `Extracted<String>` means adapters CANNOT pass a synthesized
199        /// empty string or any other placeholder here - the value either
200        /// flows from a `Source` extraction or the field is `None`.
201        #[serde(default, skip_serializing_if = "Option::is_none")]
202        text: Option<Extracted<String>>,
203    },
204    Reasoning {
205        /// `None` when the source row had no reasoning text. Type-system
206        /// guard against `unwrap_or_default()`-style fallbacks: the
207        /// `Extracted<String>` seal forces the adapter to either get the
208        /// value from a `Source` or admit it is absent.
209        #[serde(default, skip_serializing_if = "Option::is_none")]
210        text: Option<Extracted<String>>,
211    },
212    File {
213        /// `None` when the source row carried no MIME hint. Sealed against
214        /// `unwrap_or("application/octet-stream")`-style fallbacks: an absent
215        /// type is faithfully absent, not a synthesized default
216        /// (spec.md#model-no-synthesis).
217        #[serde(default, skip_serializing_if = "Option::is_none")]
218        media_type: Option<String>,
219        #[serde(skip_serializing_if = "Option::is_none")]
220        file_name: Option<String>,
221        data: FileData,
222    },
223    ToolCall {
224        /// `None` when the source carried no call_id (rare; malformed).
225        /// Sealed via `Extracted<String>` - empty-string sentinels are
226        /// not constructable from adapter code.
227        #[serde(default, skip_serializing_if = "Option::is_none")]
228        call_id: Option<Extracted<String>>,
229        /// `None` when the source carried no tool name. claude-code
230        /// always carries it on `tool_use` rows; codex-cli sometimes
231        /// has placeholder shapes. The seal makes synthesized names
232        /// unconstructable from adapter code (spec.md#model-no-synthesis).
233        #[serde(default, skip_serializing_if = "Option::is_none")]
234        name: Option<Extracted<String>>,
235        params: Value,
236        provider_executed: bool,
237    },
238    ToolResult {
239        /// `None` when the source carried no `tool_use_id` link.
240        #[serde(default, skip_serializing_if = "Option::is_none")]
241        call_id: Option<Extracted<String>>,
242        /// `None` when the adapter could not resolve the tool name.
243        /// In claude-code, name lives only on the prior `tool_use` row;
244        /// the adapter resolves via a per-file `tool_use_id -> name`
245        /// map and surfaces a miss (e.g. compaction pruned the originating
246        /// call) as `None`, never as a fabricated string
247        /// (spec.md#model-no-synthesis).
248        #[serde(default, skip_serializing_if = "Option::is_none")]
249        name: Option<Extracted<String>>,
250        is_failure: bool,
251        result: Value,
252    },
253    ToolApprovalRequest {
254        approval_id: String,
255        tool_call_id: String,
256    },
257    ToolApprovalResponse {
258        approval_id: String,
259        approved: bool,
260        #[serde(skip_serializing_if = "Option::is_none")]
261        reason: Option<String>,
262    },
263}
264
265impl PartKind {
266    pub fn type_name(&self) -> &'static str {
267        match self {
268            Self::Text { .. } => "text",
269            Self::Reasoning { .. } => "reasoning",
270            Self::File { .. } => "file",
271            Self::ToolCall { .. } => "tool_call",
272            Self::ToolResult { .. } => "tool_result",
273            Self::ToolApprovalRequest { .. } => "tool_approval_request",
274            Self::ToolApprovalResponse { .. } => "tool_approval_response",
275        }
276    }
277}
278
279#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
280#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
281pub enum FileData {
282    String(String),
283    Bytes(Vec<u8>),
284    Url(String),
285}
286
287#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
288#[serde(rename_all = "snake_case")]
289pub enum ErrorCode {
290    ValidationFailed,
291    VersionUnsupported,
292    NotFound,
293    NamespaceUnknown,
294    StorageUnavailable,
295    Conflict,
296    Internal,
297}
298
299#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
300pub struct ErrorBody {
301    pub code: ErrorCode,
302    pub message: String,
303    #[serde(default)]
304    pub details: Value,
305}
306
307#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
308pub struct ErrorEnvelope {
309    pub error: ErrorBody,
310}
311
312#[expect(
313    clippy::large_enum_variant,
314    reason = "one per request and serialized immediately, never stored in bulk"
315)]
316#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
317#[serde(untagged)]
318pub enum GetEnvelope {
319    Success(GetResponse),
320    Error(ErrorEnvelope),
321}
322
323/// Whole-session read (spec.md#protocol). `id` names either kind: a session
324/// id reads that session; a message id resolves up to its parent session with
325/// the page anchored at that message
326/// (`GetResult::Session.resolved_from_message_id` records the resolution) -
327/// intent comes from the endpoint, so upcasting is always safe. The alias
328/// accepts this endpoint's own typed name; the other type's name is not an
329/// alias - cross-type forgiveness lives in the value, not the param name.
330#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
331pub struct GetSessionRequest {
332    pub protocol_version: u16,
333    #[serde(default)]
334    pub namespace: Option<String>,
335    #[serde(alias = "session_id")]
336    pub id: String,
337    /// Max messages per page.
338    #[serde(default = "default_get_limit")]
339    pub limit: usize,
340    /// Which end to read the first page from - `start` (oldest, default) or
341    /// `end` (most recent, e.g. post-compaction recovery). Pages stay
342    /// chronological. Ignored once an anchor below is set.
343    #[serde(default)]
344    pub from: SessionFrom,
345    /// Page forward - messages strictly after this id.
346    #[serde(default)]
347    pub after_message_id: Option<String>,
348    /// Page backward - messages strictly before this id.
349    #[serde(default)]
350    pub before_message_id: Option<String>,
351}
352
353/// Single-message read (spec.md#protocol): the target with its full part
354/// bodies plus conversational neighbors. `id` must be a message id; a session
355/// id cannot resolve to one message (which one?), so the handler rejects it
356/// with a hint naming the session read.
357#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
358pub struct GetMessageRequest {
359    pub protocol_version: u16,
360    #[serde(default)]
361    pub namespace: Option<String>,
362    #[serde(alias = "message_id")]
363    pub id: String,
364    /// Conversational sibling messages before the target (mirrors `grep -B`).
365    #[serde(default = "default_context")]
366    pub context_before: usize,
367    /// Conversational sibling messages after the target (mirrors `grep -A`).
368    #[serde(default = "default_context")]
369    pub context_after: usize,
370}
371
372/// Which end of a session `pond_get_session` reads its first page from
373/// (spec.md#protocol).
374#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
375#[serde(rename_all = "lowercase")]
376pub enum SessionFrom {
377    /// Oldest messages first (the session's start).
378    #[default]
379    Start,
380    /// Most recent messages (the session's tail), still chronological.
381    End,
382}
383
384/// The session header is always present; `result` carries the mode-specific
385/// payload, discriminated by a `scope` tag (spec.md#protocol). Flattened so a
386/// client reads `session` / `scope` / payload fields off one object - no
387/// `session.session` nesting.
388#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
389pub struct GetResponse {
390    pub session: GetSession,
391    #[serde(flatten)]
392    pub result: GetResult,
393}
394
395/// Trimmed session header (spec.md#protocol): adapter-redundant `options`,
396/// parent pointers (served by `restore_lineage`), and per-message session id
397/// dropped to keep get responses lean for agent context windows.
398#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
399pub struct GetSession {
400    pub id: String,
401    pub source_agent: String,
402    pub project: String,
403    pub created_at: DateTime<Utc>,
404}
405
406impl GetSession {
407    pub fn from_session(session: &Session) -> Self {
408        Self {
409            id: session.id.clone(),
410            source_agent: session.source_agent.clone(),
411            project: (*session.project).clone(),
412            created_at: session.created_at,
413        }
414    }
415}
416
417/// Per-message view in a get response (spec.md#protocol). Always
418/// conversational: `text`/`content` plus one-line part summaries. Full part
419/// bodies ride `GetResult::Message.target_parts`, reached by `message_id`
420/// scope - a session view never inlines them.
421#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
422pub struct MessageView {
423    pub id: String,
424    pub role: Role,
425    pub timestamp: DateTime<Utc>,
426    /// Conversational text (`search_text`); absent for carrier rows.
427    #[serde(default, skip_serializing_if = "Option::is_none")]
428    pub text: Option<String>,
429    /// System-message content string, when the source carried one.
430    #[serde(default, skip_serializing_if = "Option::is_none")]
431    pub content: Option<String>,
432    #[serde(default, skip_serializing_if = "Vec::is_empty")]
433    pub parts_summary: Vec<PartSummary>,
434}
435
436/// Compact per-part descriptor (spec.md#protocol): enough to tell what a
437/// message carries without paying for full content. `call_id` is populated
438/// for `tool_call` / `tool_result` only.
439#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
440pub struct PartSummary {
441    pub kind: String,
442    #[serde(default, skip_serializing_if = "Option::is_none")]
443    pub label: Option<String>,
444    #[serde(default, skip_serializing_if = "Option::is_none")]
445    pub call_id: Option<String>,
446}
447
448impl PartSummary {
449    /// Project a canonical [`PartKind`] into its compact response descriptor, or
450    /// `None` for a kind that does not earn a summary. Exhaustive on purpose - a
451    /// new `PartKind` variant must decide here. `call_id` is carried for
452    /// `tool_call` / `tool_result` only.
453    ///
454    /// `text` and `reasoning` return `None`: a text part's content already rides
455    /// the message's `text`/`content` (a summary would duplicate it), and
456    /// reasoning is deliberately not surfaced in the session/conversational view
457    /// (its full body is still rendered when a message is fetched by `message_id`
458    /// scope). The kinds that survive are exactly [`SUMMARY_PART_TYPES`].
459    pub fn for_kind(kind: &PartKind) -> Option<Self> {
460        let (label, call_id) = match kind {
461            PartKind::Text { .. } | PartKind::Reasoning { .. } => return None,
462            PartKind::File {
463                media_type,
464                file_name,
465                ..
466            } => (file_name.clone().or_else(|| media_type.clone()), None),
467            PartKind::ToolCall { name, call_id, .. } => {
468                (name.as_deref().cloned(), call_id.as_deref().cloned())
469            }
470            PartKind::ToolResult {
471                name,
472                call_id,
473                is_failure,
474                ..
475            } => {
476                let label = name.as_deref().map(|name| {
477                    if *is_failure {
478                        format!("{name} (failed)")
479                    } else {
480                        name.clone()
481                    }
482                });
483                (label, call_id.as_deref().cloned())
484            }
485            PartKind::ToolApprovalRequest { approval_id, .. } => (Some(approval_id.clone()), None),
486            PartKind::ToolApprovalResponse {
487                approval_id,
488                approved,
489                ..
490            } => {
491                let verb = if *approved { "approved" } else { "denied" };
492                (Some(format!("{approval_id} ({verb})")), None)
493            }
494        };
495        Some(Self {
496            kind: kind.type_name().to_owned(),
497            label,
498            call_id,
499        })
500    }
501}
502
503/// Canonical part `type` names that yield a [`PartSummary`] - every kind except
504/// `text` and `reasoning` (see [`PartSummary::for_kind`], the source of truth).
505/// The summary read paths filter the parts scan to these so a text/reasoning
506/// heavy session never loads parts that would summarize to nothing.
507pub const SUMMARY_PART_TYPES: &[&str] = &[
508    "file",
509    "tool_call",
510    "tool_result",
511    "tool_approval_request",
512    "tool_approval_response",
513];
514
515/// A `Part` as it rides a get response (spec.md#protocol): the canonical
516/// part minus `session_id` / `message_id`, which the enclosing session and
517/// message already identify. Built from a canonical [`Part`] in the handler.
518#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
519pub struct ResponsePart {
520    pub id: String,
521    pub ordinal: i32,
522    pub provenance: Provenance,
523    #[serde(default, skip_serializing_if = "ProviderOptions::is_empty")]
524    pub options: ProviderOptions,
525    #[serde(flatten)]
526    pub kind: PartKind,
527}
528
529impl ResponsePart {
530    pub fn from_part(part: Part) -> Self {
531        Self {
532            id: part.id,
533            ordinal: part.ordinal,
534            provenance: part.provenance,
535            options: part.options,
536            kind: part.kind,
537        }
538    }
539}
540
541/// Mode-specific get payload, tagged by `scope` and flattened into
542/// `GetResponse` alongside the shared session header.
543#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
544#[serde(tag = "scope", rename_all = "snake_case")]
545pub enum GetResult {
546    Session {
547        messages: Vec<MessageView>,
548        /// Conversational messages before the emitted page (the top marker's
549        /// `before_message_id` cursor exists when this is > 0).
550        before_remaining: usize,
551        /// Conversational messages after the emitted page (the bottom marker's
552        /// `after_message_id` cursor exists when this is > 0).
553        after_remaining: usize,
554        /// Set when the request's `session_id` was actually a message id that
555        /// the server resolved up to this session; the page is anchored at
556        /// that message.
557        #[serde(default, skip_serializing_if = "Option::is_none")]
558        resolved_from_message_id: Option<String>,
559    },
560    Message {
561        target: MessageView,
562        target_parts: Vec<ResponsePart>,
563        target_parts_remaining: usize,
564        /// `context_before` + `context_after` conversational messages around
565        /// the target (target excluded).
566        siblings: Vec<MessageView>,
567        /// Request echo so the rendered header can state the window size.
568        context_before: usize,
569        /// Request echo so the rendered header can state the window size.
570        context_after: usize,
571    },
572}
573
574#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
575#[serde(untagged)]
576pub enum SearchEnvelope {
577    Success(SearchResponse),
578    Error(ErrorEnvelope),
579}
580
581/// JSON shape is externally tagged: `{"contains": "pond"}` or
582/// `{"regex": "^/Users/.*"}`.
583#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
584#[serde(rename_all = "snake_case")]
585pub enum ProjectFilter {
586    Contains(String),
587    Regex(String),
588}
589
590#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
591pub struct SearchRequest {
592    pub protocol_version: u16,
593    #[serde(default)]
594    pub namespace: Option<String>,
595    pub query: String,
596    /// Retrieval arm (spec.md#search). `fts` (default) matches exact whole
597    /// words via BM25; `vector` matches on meaning and is available only when
598    /// the serving instance has `[embeddings].enabled = true` - otherwise it is
599    /// refused. The agent picks per query; there is no server-side fusion.
600    #[serde(default)]
601    pub mode: SearchModeWire,
602    /// Result ordering. `relevance` (default) ranks by match strength (vector:
603    /// cosine + a gentle recency tiebreaker; fts: BM25); `recency` ranks
604    /// strictly newest-first. A recency-sorted response is labeled so the
605    /// caller does not misread rank-1 as the best match.
606    #[serde(default)]
607    pub sort_by: SortBy,
608    #[serde(default)]
609    pub filters: SearchFilters,
610    #[serde(default = "default_limit")]
611    pub limit: usize,
612}
613
614/// Wire-level retrieval arm (spec.md#search). `fts` (default) matches exact
615/// whole words via BM25; `vector` matches on meaning and is available only when
616/// the serving instance has `[embeddings].enabled = true` - otherwise it is
617/// refused. The agent picks per query; there is no server-side fusion.
618#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
619#[serde(rename_all = "lowercase")]
620pub enum SearchModeWire {
621    #[default]
622    Fts,
623    Vector,
624}
625
626/// Result ordering for `pond_search` (spec.md#search).
627#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
628#[serde(rename_all = "lowercase")]
629pub enum SortBy {
630    /// Match strength: vector = cosine + recency tiebreaker, fts = BM25.
631    #[default]
632    Relevance,
633    /// Strictly newest-first; the response is labeled as recency-sorted.
634    Recency,
635}
636
637#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
638pub struct SearchFilters {
639    #[serde(default, skip_serializing_if = "Option::is_none")]
640    pub project: Option<ProjectFilter>,
641    #[serde(default, skip_serializing_if = "Option::is_none")]
642    pub session_id: Option<String>,
643    /// Filter to one source harness with exact-or-subpath semantics: the value
644    /// itself plus its `/`-subpaths (`openclaw` covers `openclaw/subagent`, not
645    /// `openclaw-x`). A source_agent filter also disables the default subagent
646    /// exclusion (spec.md#search) - the caller is scoping deliberately.
647    #[serde(default, skip_serializing_if = "Option::is_none")]
648    pub source_agent: Option<String>,
649    #[serde(default, skip_serializing_if = "Option::is_none")]
650    pub from_date: Option<String>,
651    #[serde(default, skip_serializing_if = "Option::is_none")]
652    pub to_date: Option<String>,
653}
654
655#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
656pub struct SearchResponse {
657    pub sessions: Vec<SearchSession>,
658    pub matched_total: usize,
659    /// How many messages with conversational text the caller's filters left
660    /// in scope - the universe the search actually ran over. The absence
661    /// signal: 0 means the filters excluded everything before retrieval, and
662    /// a small value warns that "no relevant hits" covers a thin slice.
663    #[serde(default)]
664    pub searchable_in_scope: usize,
665    pub has_more: bool,
666}
667
668#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
669pub struct SearchSession {
670    pub session_id: String,
671    pub project: String,
672    pub source_agent: String,
673    pub session_messages_count: usize,
674    pub matched_message_count: usize,
675    pub matches: Vec<SearchResult>,
676}
677
678#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
679pub struct SearchResult {
680    pub message_id: String,
681    pub role: Role,
682    pub timestamp: DateTime<Utc>,
683    pub text: String,
684    pub score: f64,
685    /// Populated only for user-role hits: distinguishes a plain-text prompt
686    /// from one carrying file attachments or multi-part scaffolding.
687    #[serde(default, skip_serializing_if = "Vec::is_empty")]
688    pub parts_summary: Vec<PartSummary>,
689}
690
691/// `POST /v1/x/sql` (spec.md#protocol): read-only SQL with JSON rows. The
692/// `/v1/x/` prefix puts it outside the additive wire contract - the request is
693/// versioned, but result columns follow the storage schema, which is not.
694#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
695pub struct SqlRequest {
696    pub protocol_version: u16,
697    #[serde(default)]
698    pub namespace: Option<String>,
699    /// One read-only SELECT/WITH (or EXPLAIN of one).
700    #[serde(alias = "sql")]
701    pub query: String,
702    /// Row cap on the response: default [`DEFAULT_INLINE_ROWS`], at most
703    /// [`MAX_INLINE_ROWS`].
704    ///
705    /// [`DEFAULT_INLINE_ROWS`]: crate::sql::DEFAULT_INLINE_ROWS
706    /// [`MAX_INLINE_ROWS`]: crate::sql::MAX_INLINE_ROWS
707    #[serde(default)]
708    pub limit: Option<usize>,
709    /// Execution timeout in seconds: default [`DEFAULT_QUERY_TIMEOUT_SECS`],
710    /// clamped to [`MAX_QUERY_TIMEOUT_SECS`].
711    ///
712    /// [`DEFAULT_QUERY_TIMEOUT_SECS`]: crate::sql::DEFAULT_QUERY_TIMEOUT_SECS
713    /// [`MAX_QUERY_TIMEOUT_SECS`]: crate::sql::MAX_QUERY_TIMEOUT_SECS
714    #[serde(default)]
715    pub timeout_seconds: Option<u64>,
716}
717
718/// One JSON object per row keyed by column name, NULL fields omitted, and
719/// timestamps as RFC3339 UTC with exactly six fractional digits (lossless
720/// keyset cursors). `columns` names every result column even when `rows` is
721/// empty. `row_count` counts the returned rows and `truncated` is true iff the
722/// row cap or the response byte budget cut the result - never whether the
723/// query's own `LIMIT` did; the budget always admits the first row.
724#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
725pub struct SqlResponse {
726    pub columns: Vec<String>,
727    pub rows: Vec<Value>,
728    pub row_count: usize,
729    pub truncated: bool,
730    pub elapsed_ms: u64,
731}
732
733#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
734#[serde(untagged)]
735pub enum SqlEnvelope {
736    Success(SqlResponse),
737    Error(ErrorEnvelope),
738}
739
740#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
741#[serde(untagged)]
742pub enum IngestEnvelope {
743    Success(IngestResponse),
744    Error(ErrorEnvelope),
745}
746
747#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
748pub struct IngestRequest {
749    pub protocol_version: u16,
750    #[serde(default)]
751    pub namespace: Option<String>,
752    pub events: Vec<crate::sessions::IngestEvent>,
753}
754
755/// `pond_ingest` response (spec.md#protocol). `accepted = inserted + matched`,
756/// `rejected = error`; both derived from `results`. Per-row `results[]` is
757/// the contract clients rely on to reconcile retries (the PK is echoed so
758/// the client can match outcomes back to its input even when `index` is not
759/// enough). Each result reports the input event's `index`, `kind`, `pk`,
760/// `status`, and an `error` body when `status = "error"`.
761#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
762pub struct IngestResponse {
763    pub accepted: usize,
764    pub rejected: usize,
765    pub results: Vec<IngestResult>,
766}
767
768/// One row of `pond_ingest` per-row output (spec.md#protocol).
769#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
770pub struct IngestResult {
771    /// Position in the request's `events` array (0-based).
772    pub index: usize,
773    /// `"session"` | `"message"` | `"part"`, matching `IngestEvent::kind`.
774    pub kind: String,
775    /// Echoed primary key: scalar for session, `[session_id, message_id]` for
776    /// message, `[session_id, message_id, part_id]` for part. Lets clients reconcile
777    /// against their own state on retry.
778    pub pk: Value,
779    pub status: IngestStatus,
780    /// Set only when `status = "error"`. Carries the same shape as the
781    /// envelope-level error body.
782    #[serde(default, skip_serializing_if = "Option::is_none")]
783    pub error: Option<ErrorBody>,
784}
785
786#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
787#[serde(rename_all = "snake_case")]
788pub enum IngestStatus {
789    /// New PK; `merge_insert` wrote a fresh row.
790    Inserted,
791    /// PK existed; `merge_insert` matched it (no-op per spec.md#adapter-integrity-additive-sync).
792    Matched,
793    /// Per-row failure: validation or storage error. See `error` field.
794    Error,
795}
796
797fn default_limit() -> usize {
798    10
799}
800
801pub fn new_request_id() -> String {
802    format!("req_{}", Uuid::now_v7())
803}
804
805pub const DEFAULT_NAMESPACE: &str = "local";
806
807pub fn default_namespace() -> String {
808    DEFAULT_NAMESPACE.to_owned()
809}
810
811fn default_get_limit() -> usize {
812    20
813}
814
815fn default_context() -> usize {
816    3
817}
818
819pub fn validate_protocol(version: u16) -> Result<(), ErrorEnvelope> {
820    if version == PROTOCOL_VERSION {
821        return Ok(());
822    }
823
824    Err(error(
825        ErrorCode::VersionUnsupported,
826        "unsupported protocol_version",
827        serde_json::json!({
828            "received": version,
829            "supported": [PROTOCOL_VERSION],
830        }),
831    ))
832}
833
834pub fn error(code: ErrorCode, message: impl Into<String>, details: Value) -> ErrorEnvelope {
835    ErrorEnvelope {
836        error: ErrorBody {
837            code,
838            message: message.into(),
839            details,
840        },
841    }
842}
843
844impl From<crate::Error> for ErrorEnvelope {
845    fn from(error_value: crate::Error) -> Self {
846        match error_value {
847            crate::Error::Validation {
848                message,
849                field,
850                value,
851                expected,
852            } => error(
853                ErrorCode::ValidationFailed,
854                message,
855                validation_details(field, value, expected),
856            ),
857            crate::Error::NotFound { message, kind, pk } => error(
858                ErrorCode::NotFound,
859                message,
860                serde_json::json!({ "kind": kind, "pk": pk }),
861            ),
862            crate::Error::NamespaceUnknown { namespace } => error(
863                ErrorCode::NamespaceUnknown,
864                "namespace unknown",
865                serde_json::json!({ "namespace": namespace }),
866            ),
867            crate::Error::Conflict { attempts } => error(
868                ErrorCode::Conflict,
869                "commit conflict after retries exhausted",
870                serde_json::json!({ "attempts": attempts }),
871            ),
872            crate::Error::Storage(error_value) => storage_error(error_value),
873            crate::Error::Internal(message) => {
874                error(ErrorCode::Internal, message, serde_json::json!({}))
875            }
876        }
877    }
878}
879
880fn validation_details(
881    field: Option<String>,
882    value: Option<Value>,
883    expected: Option<String>,
884) -> Value {
885    let mut details = Map::new();
886    if let Some(field) = field {
887        details.insert("field".to_owned(), Value::String(field));
888    }
889    if let Some(value) = value {
890        details.insert("value".to_owned(), value);
891    }
892    if let Some(expected) = expected {
893        details.insert("expected".to_owned(), Value::String(expected));
894    }
895    Value::Object(details)
896}
897
898pub fn storage_error(error_value: anyhow::Error) -> ErrorEnvelope {
899    // Lance's "fragment N referenced by an address-domain index result was not
900    // found" internal error means a compaction orphaned the timestamp
901    // zonemap's fragment references - any writer without the same-run
902    // self-heal does this, including this binary's own compaction in the
903    // window before its indices phase heals it. The store's data is intact
904    // and the index is recreatable; name the recovery instead of surfacing a
905    // bare storage failure.
906    let stale_address_index = error_value.chain().any(|cause| {
907        cause
908            .to_string()
909            .contains("referenced by an address-domain index result")
910    });
911    if stale_address_index {
912        return error(
913            ErrorCode::StorageUnavailable,
914            "date-filter index is stale: it references fragments that compaction rewrote; \
915             run `pond optimize --rebuild`, or the next `pond sync`/`pond optimize` from \
916             this pond version repairs it automatically",
917            serde_json::json!({ "underlying": format!("{error_value:#}") }),
918        );
919    }
920    error(
921        ErrorCode::StorageUnavailable,
922        "storage operation failed",
923        serde_json::json!({ "underlying": format!("{error_value:#}") }),
924    )
925}
926
927#[cfg(test)]
928mod tests {
929    #![expect(clippy::expect_used, reason = "tests fail by panicking")]
930
931    use super::*;
932    use serde_json::json;
933
934    #[test]
935    fn wire_envelope_carries_conflict_code_and_attempts_detail() {
936        let envelope: ErrorEnvelope = crate::Error::Conflict { attempts: 3 }.into();
937        assert_eq!(envelope.error.code, ErrorCode::Conflict);
938        assert_eq!(envelope.error.details, json!({ "attempts": 3 }));
939    }
940
941    #[test]
942    fn storage_error_names_the_stale_zonemap_recovery() {
943        let lance_internal = anyhow::anyhow!(
944            "Internal error: fragment 5225 referenced by an address-domain index result \
945             was not found in the dataset"
946        )
947        .context("scan failed");
948        let envelope = storage_error(lance_internal);
949        assert_eq!(envelope.error.code, ErrorCode::StorageUnavailable);
950        assert!(envelope.error.message.contains("pond optimize --rebuild"));
951        let underlying = envelope.error.details["underlying"]
952            .as_str()
953            .expect("underlying detail");
954        assert!(underlying.contains("fragment 5225"));
955
956        let generic = storage_error(anyhow::anyhow!("connection refused"));
957        assert_eq!(generic.error.message, "storage operation failed");
958    }
959}
pond/wire.rs

pond/
wire.rs
