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 system_content(&self) -> Option<&str> {
119        match self {
120            // Two layers of `as_deref`: the outer `Option<Extracted<String>>`
121            // becomes `Option<&Extracted<String>>`, then `Extracted: Deref`
122            // unwraps to `&str`.
123            Self::System { content, .. } => content.as_deref().map(|e| &**e),
124            Self::User { .. } | Self::Assistant { .. } | Self::Tool { .. } => None,
125        }
126    }
127}
128
129#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
130#[serde(rename_all = "snake_case")]
131pub enum Role {
132    System,
133    User,
134    Assistant,
135    Tool,
136}
137
138impl Role {
139    pub fn as_str(self) -> &'static str {
140        match self {
141            Self::System => "system",
142            Self::User => "user",
143            Self::Assistant => "assistant",
144            Self::Tool => "tool",
145        }
146    }
147}
148
149/// Whether a Part's content is conversation or harness-injected scaffolding
150/// (spec.md#model-part-provenance). No `Default` and no `#[serde(default)]` on the
151/// `Part.provenance` field below: constructing a Part without classifying it
152/// MUST be a compile error (spec.md#adapter-provenance-required).
153#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
154#[serde(rename_all = "snake_case")]
155pub enum Provenance {
156    Conversational,
157    Injected,
158}
159
160impl Provenance {
161    pub fn as_str(self) -> &'static str {
162        match self {
163            Self::Conversational => "conversational",
164            Self::Injected => "injected",
165        }
166    }
167}
168
169#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
170pub struct Part {
171    pub session_id: String,
172    pub id: String,
173    pub message_id: String,
174    pub ordinal: i32,
175    /// Conversation vs harness-injected (spec.md#model-part-provenance). Mandatory,
176    /// no serde default - search reads it to exclude injected scaffolding.
177    pub provenance: Provenance,
178    #[serde(default)]
179    pub options: ProviderOptions,
180    #[serde(flatten)]
181    pub kind: PartKind,
182}
183
184#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
185#[serde(tag = "type", rename_all = "snake_case")]
186pub enum PartKind {
187    Text {
188        /// `None` when the source row had no text field. The seal on
189        /// `Extracted<String>` means adapters CANNOT pass a synthesized
190        /// empty string or any other placeholder here - the value either
191        /// flows from a `Source` extraction or the field is `None`.
192        #[serde(default, skip_serializing_if = "Option::is_none")]
193        text: Option<Extracted<String>>,
194    },
195    Reasoning {
196        /// `None` when the source row had no reasoning text. Type-system
197        /// guard against `unwrap_or_default()`-style fallbacks: the
198        /// `Extracted<String>` seal forces the adapter to either get the
199        /// value from a `Source` or admit it is absent.
200        #[serde(default, skip_serializing_if = "Option::is_none")]
201        text: Option<Extracted<String>>,
202    },
203    File {
204        /// `None` when the source row carried no MIME hint. Sealed against
205        /// `unwrap_or("application/octet-stream")`-style fallbacks: an absent
206        /// type is faithfully absent, not a synthesized default
207        /// (spec.md#model-no-synthesis).
208        #[serde(default, skip_serializing_if = "Option::is_none")]
209        media_type: Option<String>,
210        #[serde(skip_serializing_if = "Option::is_none")]
211        file_name: Option<String>,
212        data: FileData,
213    },
214    ToolCall {
215        /// `None` when the source carried no call_id (rare; malformed).
216        /// Sealed via `Extracted<String>` - empty-string sentinels are
217        /// not constructable from adapter code.
218        #[serde(default, skip_serializing_if = "Option::is_none")]
219        call_id: Option<Extracted<String>>,
220        /// `None` when the source carried no tool name. claude-code
221        /// always carries it on `tool_use` rows; codex-cli sometimes
222        /// has placeholder shapes. The seal makes synthesized names
223        /// unconstructable from adapter code (spec.md#model-no-synthesis).
224        #[serde(default, skip_serializing_if = "Option::is_none")]
225        name: Option<Extracted<String>>,
226        params: Value,
227        provider_executed: bool,
228    },
229    ToolResult {
230        /// `None` when the source carried no `tool_use_id` link.
231        #[serde(default, skip_serializing_if = "Option::is_none")]
232        call_id: Option<Extracted<String>>,
233        /// `None` when the adapter could not resolve the tool name.
234        /// In claude-code, name lives only on the prior `tool_use` row;
235        /// the adapter resolves via a per-file `tool_use_id -> name`
236        /// map and surfaces a miss (e.g. compaction pruned the originating
237        /// call) as `None`, never as a fabricated string
238        /// (spec.md#model-no-synthesis).
239        #[serde(default, skip_serializing_if = "Option::is_none")]
240        name: Option<Extracted<String>>,
241        is_failure: bool,
242        result: Value,
243    },
244    ToolApprovalRequest {
245        approval_id: String,
246        tool_call_id: String,
247    },
248    ToolApprovalResponse {
249        approval_id: String,
250        approved: bool,
251        #[serde(skip_serializing_if = "Option::is_none")]
252        reason: Option<String>,
253    },
254}
255
256impl PartKind {
257    pub fn type_name(&self) -> &'static str {
258        match self {
259            Self::Text { .. } => "text",
260            Self::Reasoning { .. } => "reasoning",
261            Self::File { .. } => "file",
262            Self::ToolCall { .. } => "tool_call",
263            Self::ToolResult { .. } => "tool_result",
264            Self::ToolApprovalRequest { .. } => "tool_approval_request",
265            Self::ToolApprovalResponse { .. } => "tool_approval_response",
266        }
267    }
268}
269
270#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
271#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
272pub enum FileData {
273    String(String),
274    Bytes(Vec<u8>),
275    Url(String),
276}
277
278#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
279#[serde(rename_all = "snake_case")]
280pub enum ErrorCode {
281    ValidationFailed,
282    VersionUnsupported,
283    NotFound,
284    NamespaceUnknown,
285    StorageUnavailable,
286    Conflict,
287    Internal,
288}
289
290#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
291pub struct ErrorBody {
292    pub code: ErrorCode,
293    pub message: String,
294    #[serde(default)]
295    pub details: Value,
296}
297
298#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
299pub struct ErrorEnvelope {
300    pub error: ErrorBody,
301}
302
303// The success/error size gap is fine here: a `GetEnvelope` is one per-request
304// return value, serialized immediately - never stored in bulk where the gap
305// would waste memory.
306#[allow(clippy::large_enum_variant)]
307#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
308#[serde(untagged)]
309pub enum GetEnvelope {
310    Success(GetResponse),
311    Error(ErrorEnvelope),
312}
313
314#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
315pub struct GetRequest {
316    pub protocol_version: u16,
317    #[serde(default)]
318    pub namespace: Option<String>,
319    // Mutually exclusive scopes.
320    #[serde(default)]
321    pub session_id: Option<String>,
322    #[serde(default)]
323    pub message_id: Option<String>,
324    /// `message_id` mode only: symmetric window radius, mirroring `grep -C`.
325    /// Returns up to `2*context_depth` siblings around the target. Ignored in
326    /// session mode.
327    #[serde(default)]
328    pub context_depth: usize,
329    #[serde(default = "default_get_limit")]
330    pub limit: usize,
331    /// Ignored in `message_id` mode (that mode always returns the target with
332    /// its full parts, paginated).
333    #[serde(default)]
334    pub response_mode: ResponseMode,
335    /// Exclusive continuation anchor: last `message_id` in session mode, last
336    /// `part_id` in message mode. The append-only invariant means one id is
337    /// enough state - no opaque cursor.
338    #[serde(default)]
339    pub after_id: Option<String>,
340    /// Session mode only: which end to read `limit` messages from - `start`
341    /// (oldest, default) or `end` (most recent). Results stay chronological;
342    /// ignored in message mode.
343    #[serde(default)]
344    pub session_from: SessionFrom,
345}
346
347/// How much of each message `pond_get` materializes (spec.md#protocol).
348/// Ignored when `message_id` is set.
349#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
350#[serde(rename_all = "snake_case")]
351pub enum ResponseMode {
352    /// `search_text IS NOT NULL` conversational view + part summaries.
353    #[default]
354    Conversational,
355    /// All messages (including carriers) + part summaries.
356    Complete,
357    /// All messages + full parts inline (heaviest mode).
358    Verbatim,
359}
360
361/// Which end of a session `pond_get` reads its page from (spec.md#protocol).
362#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
363#[serde(rename_all = "lowercase")]
364pub enum SessionFrom {
365    /// Oldest messages first (the session's start).
366    #[default]
367    Start,
368    /// Most recent messages (the session's tail), still chronological.
369    End,
370}
371
372/// The session header is always present; `result` carries the mode-specific
373/// payload, discriminated by a `scope` tag (spec.md#protocol). Flattened so a
374/// client reads `session` / `scope` / payload fields off one object - no
375/// `session.session` nesting.
376#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
377pub struct GetResponse {
378    pub session: GetSession,
379    #[serde(flatten)]
380    pub result: GetResult,
381}
382
383/// Trimmed session header (spec.md#protocol): adapter-redundant `options`,
384/// parent pointers (served by `restore_lineage`), and per-message session id
385/// dropped to keep `pond_get` responses lean for agent context windows.
386#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
387pub struct GetSession {
388    pub id: String,
389    pub source_agent: String,
390    pub project: String,
391    pub created_at: DateTime<Utc>,
392}
393
394impl GetSession {
395    pub fn from_session(session: &Session) -> Self {
396        Self {
397            id: session.id.clone(),
398            source_agent: session.source_agent.clone(),
399            project: (*session.project).clone(),
400            created_at: session.created_at,
401        }
402    }
403}
404
405/// Per-message view in a `pond_get` response. `parts_summary` is always
406/// present (possibly empty); `parts` is populated only in Verbatim mode.
407#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
408pub struct MessageView {
409    pub id: String,
410    pub role: Role,
411    pub timestamp: DateTime<Utc>,
412    /// Conversational text (`search_text`); absent for carrier rows.
413    #[serde(default, skip_serializing_if = "Option::is_none")]
414    pub text: Option<String>,
415    /// System-message content string, when the source carried one.
416    #[serde(default, skip_serializing_if = "Option::is_none")]
417    pub content: Option<String>,
418    #[serde(default, skip_serializing_if = "Vec::is_empty")]
419    pub parts_summary: Vec<PartSummary>,
420    #[serde(default, skip_serializing_if = "Option::is_none")]
421    pub parts: Option<Vec<ResponsePart>>,
422}
423
424/// Compact per-part descriptor (spec.md#protocol): enough to tell what a
425/// message carries without paying for full content. `call_id` is populated
426/// for `tool_call` / `tool_result` only.
427#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
428pub struct PartSummary {
429    pub kind: String,
430    #[serde(default, skip_serializing_if = "Option::is_none")]
431    pub label: Option<String>,
432    #[serde(default, skip_serializing_if = "Option::is_none")]
433    pub call_id: Option<String>,
434}
435
436impl PartSummary {
437    /// Project a canonical [`PartKind`] into its compact response descriptor, or
438    /// `None` for a kind that does not earn a summary. Exhaustive on purpose - a
439    /// new `PartKind` variant must decide here. `call_id` is carried for
440    /// `tool_call` / `tool_result` only.
441    ///
442    /// `text` and `reasoning` return `None`: a text part's content already rides
443    /// the message's `text`/`content` (a summary would duplicate it), and
444    /// reasoning is deliberately not surfaced by default (its full body is still
445    /// reachable in verbatim mode). The kinds that survive are exactly
446    /// [`SUMMARY_PART_TYPES`].
447    pub fn for_kind(kind: &PartKind) -> Option<Self> {
448        let (label, call_id) = match kind {
449            PartKind::Text { .. } | PartKind::Reasoning { .. } => return None,
450            PartKind::File {
451                media_type,
452                file_name,
453                ..
454            } => (file_name.clone().or_else(|| media_type.clone()), None),
455            PartKind::ToolCall { name, call_id, .. } => {
456                (name.as_deref().cloned(), call_id.as_deref().cloned())
457            }
458            PartKind::ToolResult {
459                name,
460                call_id,
461                is_failure,
462                ..
463            } => {
464                let label = name.as_deref().map(|name| {
465                    if *is_failure {
466                        format!("{name} (failed)")
467                    } else {
468                        name.clone()
469                    }
470                });
471                (label, call_id.as_deref().cloned())
472            }
473            PartKind::ToolApprovalRequest { approval_id, .. } => (Some(approval_id.clone()), None),
474            PartKind::ToolApprovalResponse {
475                approval_id,
476                approved,
477                ..
478            } => {
479                let verb = if *approved { "approved" } else { "denied" };
480                (Some(format!("{approval_id} ({verb})")), None)
481            }
482        };
483        Some(Self {
484            kind: kind.type_name().to_owned(),
485            label,
486            call_id,
487        })
488    }
489}
490
491/// Canonical part `type` names that yield a [`PartSummary`] - every kind except
492/// `text` and `reasoning` (see [`PartSummary::for_kind`], the source of truth).
493/// The summary read paths filter the parts scan to these so a text/reasoning
494/// heavy session never loads parts that would summarize to nothing.
495pub const SUMMARY_PART_TYPES: &[&str] = &[
496    "file",
497    "tool_call",
498    "tool_result",
499    "tool_approval_request",
500    "tool_approval_response",
501];
502
503/// A `Part` as it rides a `pond_get` response (spec.md#protocol): the canonical
504/// part minus `session_id` / `message_id`, which the enclosing session and
505/// message already identify. Built from a canonical [`Part`] in the handler.
506#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
507pub struct ResponsePart {
508    pub id: String,
509    pub ordinal: i32,
510    pub provenance: Provenance,
511    #[serde(default, skip_serializing_if = "ProviderOptions::is_empty")]
512    pub options: ProviderOptions,
513    #[serde(flatten)]
514    pub kind: PartKind,
515}
516
517impl ResponsePart {
518    pub fn from_part(part: Part) -> Self {
519        Self {
520            id: part.id,
521            ordinal: part.ordinal,
522            provenance: part.provenance,
523            options: part.options,
524            kind: part.kind,
525        }
526    }
527}
528
529/// Mode-specific `pond_get` payload, tagged by `scope` and flattened into
530/// `GetResponse` alongside the shared session header.
531#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
532#[serde(tag = "scope", rename_all = "snake_case")]
533pub enum GetResult {
534    Session {
535        messages: Vec<MessageView>,
536        messages_remaining: usize,
537    },
538    Message {
539        target: MessageView,
540        target_parts: Vec<ResponsePart>,
541        target_parts_remaining: usize,
542        /// Up to `2*context_depth` messages around the target (target excluded).
543        siblings: Vec<MessageView>,
544    },
545}
546
547#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
548#[serde(untagged)]
549pub enum SearchEnvelope {
550    Success(SearchResponse),
551    Error(ErrorEnvelope),
552}
553
554/// JSON shape is externally tagged: `{"contains": "pond"}` or
555/// `{"regex": "^/Users/.*"}`.
556#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
557#[serde(rename_all = "snake_case")]
558pub enum ProjectFilter {
559    Contains(String),
560    Regex(String),
561}
562
563#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
564pub struct SearchRequest {
565    pub protocol_version: u16,
566    #[serde(default)]
567    pub namespace: Option<String>,
568    pub query: String,
569    // Server normally decides between hybrid and FTS-only from the embedder +
570    // embeddings-coverage state (spec.md#search); `mode_override` is the
571    // operator-tooling escape hatch consumed by the `scripts/search-benchmarks/`
572    // harness. Production callers (MCP, HTTP agents) should leave it `None`.
573    #[serde(default, skip_serializing_if = "Option::is_none")]
574    pub mode_override: Option<SearchModeWire>,
575    /// When set, retrieve messages similar to this stored message - pond uses
576    /// the message's stored `vector` directly as the query, runs vector-only
577    /// kNN, and ignores `query` and the FTS arm. The stored vector was
578    /// derived from `search_text` (`spec.md#session-embed-from-canonical`), so the
579    /// signal is already filtered of harness-injected parts. Filters and
580    /// `limit` still apply.
581    #[serde(default, skip_serializing_if = "Option::is_none")]
582    pub similar_to: Option<String>,
583    #[serde(default)]
584    pub filters: SearchFilters,
585    #[serde(default = "default_limit")]
586    pub limit: usize,
587    #[serde(default, skip_serializing_if = "Option::is_none")]
588    pub cursor: Option<String>,
589}
590
591/// Wire-level retrieval mode override (spec.md#search). Not normally set on
592/// the wire - the server decides hybrid vs FTS-only from embedding
593/// availability. The variant exists so operator tooling (`pond search --mode`,
594/// the embeddings-benchmark harness) can force one arm without an env-var
595/// backdoor.
596#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
597#[serde(rename_all = "lowercase")]
598pub enum SearchModeWire {
599    Fts,
600    Vector,
601    Hybrid,
602}
603
604#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
605pub struct SearchFilters {
606    #[serde(default, skip_serializing_if = "Option::is_none")]
607    pub project: Option<ProjectFilter>,
608    #[serde(default, skip_serializing_if = "Option::is_none")]
609    pub session_id: Option<String>,
610    #[serde(default, skip_serializing_if = "Option::is_none")]
611    pub source_agent: Option<String>,
612    #[serde(default, skip_serializing_if = "Option::is_none")]
613    pub from_date: Option<String>,
614    #[serde(default, skip_serializing_if = "Option::is_none")]
615    pub to_date: Option<String>,
616    #[serde(default, skip_serializing_if = "Option::is_none")]
617    pub role: Option<String>,
618    // Skip the default 0.0 so an unfiltered cursor/request stays compact.
619    #[serde(default, skip_serializing_if = "is_zero_f64")]
620    pub min_score: f64,
621    /// Include subagent sessions (`source_agent` like `claude-code/<name>`).
622    /// Default false: a search targets the human-facing main sessions.
623    #[serde(default, skip_serializing_if = "is_false")]
624    pub include_subagents: bool,
625}
626
627impl SearchFilters {
628    fn is_default(&self) -> bool {
629        *self == Self::default()
630    }
631}
632
633fn is_false(value: &bool) -> bool {
634    !*value
635}
636
637fn is_zero_f64(value: &f64) -> bool {
638    *value == 0.0
639}
640
641#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
642pub struct SearchResponse {
643    pub sessions: Vec<SearchSession>,
644    pub matched_total: usize,
645    pub has_more: bool,
646    #[serde(default, skip_serializing_if = "Option::is_none")]
647    pub next_cursor: Option<String>,
648}
649
650#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
651pub struct SearchSession {
652    pub session_id: String,
653    pub project: String,
654    pub source_agent: String,
655    pub session_messages_count: usize,
656    pub matched_message_count: usize,
657    pub matches: Vec<SearchResult>,
658}
659
660#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
661pub struct SearchResult {
662    pub message_id: String,
663    pub role: Role,
664    pub timestamp: DateTime<Utc>,
665    pub text: String,
666    pub score: f64,
667    /// Populated only for user-role hits: distinguishes a plain-text prompt
668    /// from one carrying file attachments or multi-part scaffolding.
669    #[serde(default, skip_serializing_if = "Vec::is_empty")]
670    pub parts_summary: Vec<PartSummary>,
671}
672
673#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
674pub struct SearchCursor {
675    pub query: String,
676    #[serde(default, skip_serializing_if = "Option::is_none")]
677    pub similar_to: Option<String>,
678    #[serde(default, skip_serializing_if = "SearchFilters::is_default")]
679    pub filters: SearchFilters,
680    pub offset: usize,
681}
682
683#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
684#[serde(untagged)]
685pub enum IngestEnvelope {
686    Success(IngestResponse),
687    Error(ErrorEnvelope),
688}
689
690#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
691pub struct IngestRequest {
692    pub protocol_version: u16,
693    #[serde(default)]
694    pub namespace: Option<String>,
695    pub events: Vec<crate::sessions::IngestEvent>,
696}
697
698/// `pond_ingest` response (spec.md#protocol). `accepted = inserted + matched`,
699/// `rejected = error`; both derived from `results`. Per-row `results[]` is
700/// the contract clients rely on to reconcile retries (the PK is echoed so
701/// the client can match outcomes back to its input even when `index` is not
702/// enough). Each result reports the input event's `index`, `kind`, `pk`,
703/// `status`, and an `error` body when `status = "error"`.
704#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
705pub struct IngestResponse {
706    pub accepted: usize,
707    pub rejected: usize,
708    pub results: Vec<IngestResult>,
709}
710
711/// One row of `pond_ingest` per-row output (spec.md#protocol).
712#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
713pub struct IngestResult {
714    /// Position in the request's `events` array (0-based).
715    pub index: usize,
716    /// `"session"` | `"message"` | `"part"`, matching `IngestEvent::kind`.
717    pub kind: String,
718    /// Echoed primary key: scalar for session, `[session_id, message_id]` for
719    /// message, `[session_id, message_id, part_id]` for part. Lets clients reconcile
720    /// against their own state on retry.
721    pub pk: Value,
722    pub status: IngestStatus,
723    /// Set only when `status = "error"`. Carries the same shape as the
724    /// envelope-level error body.
725    #[serde(default, skip_serializing_if = "Option::is_none")]
726    pub error: Option<ErrorBody>,
727}
728
729#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
730#[serde(rename_all = "snake_case")]
731pub enum IngestStatus {
732    /// New PK; `merge_insert` wrote a fresh row.
733    Inserted,
734    /// PK existed; `merge_insert` matched it (no-op per spec.md#adapter-integrity-additive-sync).
735    Matched,
736    /// Per-row failure: validation or storage error. See `error` field.
737    Error,
738}
739
740fn default_limit() -> usize {
741    10
742}
743
744pub fn new_request_id() -> String {
745    format!("req_{}", Uuid::now_v7())
746}
747
748pub const DEFAULT_NAMESPACE: &str = "local";
749
750pub fn default_namespace() -> String {
751    DEFAULT_NAMESPACE.to_owned()
752}
753
754fn default_get_limit() -> usize {
755    20
756}
757
758pub fn validate_protocol(version: u16) -> Result<(), ErrorEnvelope> {
759    if version == PROTOCOL_VERSION {
760        return Ok(());
761    }
762
763    Err(error(
764        ErrorCode::VersionUnsupported,
765        "unsupported protocol_version",
766        serde_json::json!({
767            "received": version,
768            "supported": [PROTOCOL_VERSION],
769        }),
770    ))
771}
772
773pub fn error(code: ErrorCode, message: impl Into<String>, details: Value) -> ErrorEnvelope {
774    ErrorEnvelope {
775        error: ErrorBody {
776            code,
777            message: message.into(),
778            details,
779        },
780    }
781}
782
783impl From<crate::Error> for ErrorEnvelope {
784    fn from(error_value: crate::Error) -> Self {
785        match error_value {
786            crate::Error::Validation {
787                message,
788                field,
789                value,
790                expected,
791            } => error(
792                ErrorCode::ValidationFailed,
793                message,
794                validation_details(field, value, expected),
795            ),
796            crate::Error::NotFound { message, kind, pk } => error(
797                ErrorCode::NotFound,
798                message,
799                serde_json::json!({ "kind": kind, "pk": pk }),
800            ),
801            crate::Error::NamespaceUnknown { namespace } => error(
802                ErrorCode::NamespaceUnknown,
803                "namespace unknown",
804                serde_json::json!({ "namespace": namespace }),
805            ),
806            crate::Error::Conflict { attempts } => error(
807                ErrorCode::Conflict,
808                "commit conflict after retries exhausted",
809                serde_json::json!({ "attempts": attempts }),
810            ),
811            crate::Error::Storage(error_value) => storage_error(error_value),
812            crate::Error::Internal(message) => {
813                error(ErrorCode::Internal, message, serde_json::json!({}))
814            }
815        }
816    }
817}
818
819fn validation_details(
820    field: Option<String>,
821    value: Option<Value>,
822    expected: Option<String>,
823) -> Value {
824    let mut details = Map::new();
825    if let Some(field) = field {
826        details.insert("field".to_owned(), Value::String(field));
827    }
828    if let Some(value) = value {
829        details.insert("value".to_owned(), value);
830    }
831    if let Some(expected) = expected {
832        details.insert("expected".to_owned(), Value::String(expected));
833    }
834    Value::Object(details)
835}
836
837pub fn storage_error(error_value: anyhow::Error) -> ErrorEnvelope {
838    error(
839        ErrorCode::StorageUnavailable,
840        "storage operation failed",
841        serde_json::json!({ "underlying": error_value.to_string() }),
842    )
843}
844
845#[cfg(test)]
846mod tests {
847    #![allow(clippy::expect_used, clippy::unwrap_used)]
848
849    use super::*;
850    use serde_json::json;
851
852    #[test]
853    fn wire_envelope_carries_conflict_code_and_attempts_detail() {
854        let envelope: ErrorEnvelope = crate::Error::Conflict { attempts: 3 }.into();
855        assert_eq!(envelope.error.code, ErrorCode::Conflict);
856        assert_eq!(envelope.error.details, json!({ "attempts": 3 }));
857    }
858}
pond/wire.rs

pond/
wire.rs
