okbot_core/
store.rs

1//! The conversation store port.
2
3use std::time::Duration;
4
5use jiff::Timestamp;
6use serde::{Deserialize, Serialize};
7
8use crate::agent::TokenUsage;
9use crate::caps::ToolPolicy;
10use crate::id::{ChannelId, ConversationKey, MessageId, RunId, UserId};
11use crate::peer::PeerTaskRecord;
12use crate::{BoxError, async_trait};
13
14/// A storage failure.
15#[derive(Debug, thiserror::Error)]
16#[non_exhaustive]
17pub enum StoreError {
18    /// The database was written by a newer okbot.
19    #[error("the database schema version {found} is newer than this okbot supports ({supported})")]
20    TooNew {
21        /// Version in the database.
22        found: u32,
23        /// Newest version this build knows.
24        supported: u32,
25    },
26    /// Any other failure.
27    #[error("storage error: {0}")]
28    Backend(#[source] BoxError),
29}
30
31/// Wrong pairing codes tolerated before pending codes are cancelled.
32pub const PAIRING_MAX_FAILURES: u32 = 5;
33
34/// Who wrote a stored message.
35#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
36#[serde(rename_all = "kebab-case")]
37pub enum MessageRole {
38    /// A chat user.
39    User,
40    /// The bot.
41    Bot,
42}
43
44/// A message to store.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct NewMessage {
47    /// Conversation.
48    pub conversation: ConversationKey,
49    /// Author role.
50    pub role: MessageRole,
51    /// Author id (the user, or the bot's own id when known).
52    pub author_id: Option<UserId>,
53    /// Author display name.
54    pub author_name: String,
55    /// Text (canonical Markdown for bot messages).
56    pub text: String,
57    /// Platform message id.
58    pub message_id: Option<MessageId>,
59    /// When it was sent.
60    pub at: Timestamp,
61}
62
63/// A stored message.
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub struct StoredMessage {
66    /// Author role.
67    pub role: MessageRole,
68    /// Author display name.
69    pub author_name: String,
70    /// Text.
71    pub text: String,
72    /// When it was sent.
73    pub at: Timestamp,
74}
75
76/// One agent run, for the audit table.
77#[derive(Debug, Clone, PartialEq, Eq)]
78pub struct AgentRunRecord {
79    /// Run id.
80    pub run_id: RunId,
81    /// Conversation.
82    pub conversation: ConversationKey,
83    /// Backend that produced the result (or the last one tried).
84    pub backend: String,
85    /// Model, when known.
86    pub model: Option<String>,
87    /// Tool policy of the run.
88    pub tools: ToolPolicy,
89    /// Token usage, when reported.
90    pub usage: Option<TokenUsage>,
91    /// Duration.
92    pub duration: Duration,
93    /// Error text, when the run failed.
94    pub error: Option<String>,
95    /// When the run started.
96    pub started_at: Timestamp,
97    /// Who the run was for ([`person_key`]), counted against `[limits]`; `None` for runs nobody
98    /// asked for in chat.
99    pub person: Option<String>,
100}
101
102/// The key a person's usage and audit entries are stored under: `person:<id>` for linked
103/// accounts (`[people]`, or the owner of a personal bot), else the account
104/// `platform:instance/user-id`.
105pub fn person_key(channel: &ChannelId, user: &UserId, person: Option<&str>) -> String {
106    match person {
107        Some(id) => format!("person:{id}"),
108        None => format!("{channel}/{user}"),
109    }
110}
111
112/// What one person used over a period.
113#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
114pub struct Usage {
115    /// Agent turns.
116    pub turns: u64,
117    /// Input plus output tokens, as reported.
118    pub tokens: u64,
119}
120
121/// What happened to a tool call (ADR 0011).
122#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
123#[non_exhaustive]
124pub enum ToolDecision {
125    /// Sent to the server.
126    Allowed,
127    /// Refused by the rules (role, room, tool lists).
128    Denied,
129    /// A person declined it (or did not answer) when asked to confirm.
130    Declined,
131}
132
133impl ToolDecision {
134    /// The name stored and shown.
135    pub fn as_str(self) -> &'static str {
136        match self {
137            Self::Allowed => "allowed",
138            Self::Denied => "denied",
139            Self::Declined => "declined",
140        }
141    }
142
143    /// Parses [`ToolDecision::as_str`].
144    pub fn parse(value: &str) -> Option<Self> {
145        match value {
146            "allowed" => Some(Self::Allowed),
147            "denied" => Some(Self::Denied),
148            "declined" => Some(Self::Declined),
149            _ => None,
150        }
151    }
152}
153
154/// One call of an `[[mcp]]` tool, by an agent or a chat command. Arguments and results are not
155/// kept: they may hold secrets.
156#[derive(Debug, Clone, PartialEq, Eq)]
157pub struct ToolCallRecord {
158    /// The agent run, when an agent made the call.
159    pub run_id: Option<RunId>,
160    /// The `[[command]]`, when a person ran one.
161    pub command: Option<String>,
162    /// Who it was for ([`person_key`]).
163    pub person: String,
164    /// Where.
165    pub conversation: ConversationKey,
166    /// The `[[mcp]]` server.
167    pub server: String,
168    /// The tool.
169    pub tool: String,
170    /// What happened.
171    pub decision: ToolDecision,
172    /// The transport error or the tool's own error flag, when the call failed.
173    pub error: Option<String>,
174    /// Time from the request to the answer (including a confirmation).
175    pub duration: Duration,
176    /// When the call was made.
177    pub at: Timestamp,
178}
179
180/// A role given to an account for a limited time (`okbot role grant`, `/grant`).
181#[derive(Debug, Clone, PartialEq, Eq)]
182pub struct RoleGrant {
183    /// A role from `[roles]`.
184    pub role: String,
185    /// The account, as `platform:instance/user-id`.
186    pub account: String,
187    /// Who granted it (an account, or `operator` for the terminal).
188    pub granted_by: String,
189    /// When it was granted.
190    pub granted_at: Timestamp,
191    /// When it ends.
192    pub expires_at: Timestamp,
193}
194
195/// Where one firing of a task is (design §5.6).
196#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
197pub enum FireState {
198    /// Claimed; being delivered or waiting for a retry.
199    Sending,
200    /// Delivered.
201    Sent,
202    /// Gave up after retries.
203    Failed,
204    /// Too late to deliver (the bot was down).
205    Missed,
206}
207
208impl FireState {
209    /// The name stored and shown.
210    pub fn as_str(self) -> &'static str {
211        match self {
212            Self::Sending => "sending",
213            Self::Sent => "sent",
214            Self::Failed => "failed",
215            Self::Missed => "missed",
216        }
217    }
218
219    /// Parses [`FireState::as_str`].
220    pub fn parse(value: &str) -> Option<Self> {
221        match value {
222            "sending" => Some(Self::Sending),
223            "sent" => Some(Self::Sent),
224            "failed" => Some(Self::Failed),
225            "missed" => Some(Self::Missed),
226            _ => None,
227        }
228    }
229
230    /// No more delivery attempts follow.
231    pub fn is_final(self) -> bool {
232        !matches!(self, Self::Sending)
233    }
234}
235
236/// One firing of a task: the task id and its planned time identify it.
237#[derive(Debug, Clone, PartialEq, Eq)]
238pub struct Fire {
239    /// Task id (`tasks/people/owner/t-7k2m9q`).
240    pub task_id: String,
241    /// The planned time.
242    pub fire_at: Timestamp,
243    /// Where it is.
244    pub state: FireState,
245    /// Delivery attempts so far.
246    pub attempts: u32,
247    /// Last change.
248    pub updated_at: Timestamp,
249}
250
251/// A sealed secret's identity, without the secret (for listings).
252#[derive(Debug, Clone, PartialEq, Eq)]
253pub struct SealedEntry {
254    /// What it is (`oauth-token`, `oauth-client`).
255    pub kind: String,
256    /// The `[[mcp]]` server.
257    pub server: String,
258    /// Whose it is: a [`person_key`], `shared`, or the issuer of a client registration.
259    pub subject: String,
260    /// Last change.
261    pub updated_at: Timestamp,
262}
263
264/// One run of an automation (ADR 0013).
265#[derive(Debug, Clone, PartialEq, Eq)]
266pub struct AutomationRun {
267    /// The `[[automation]]` name.
268    pub automation: String,
269    /// When it ran.
270    pub at: Timestamp,
271    /// New items acted on (0 when nothing was new).
272    pub items: u32,
273    /// What it did: `text`, `prompt`, `none` (nothing new) or `primed` (first poll).
274    pub action: String,
275    /// Why it failed, when it did.
276    pub error: Option<String>,
277}
278
279fn unsupported() -> StoreError {
280    StoreError::Backend("not supported by this store".into())
281}
282
283/// Stores conversations, deduplication marks and the audit trail (design §6.1).
284#[async_trait]
285pub trait ConversationStore: Send + Sync {
286    /// Records an inbound event id. Returns `false` when it was already seen within `ttl`.
287    async fn mark_seen(
288        &self,
289        channel: &ChannelId,
290        event_id: &str,
291        ttl: Duration,
292    ) -> Result<bool, StoreError>;
293
294    /// Appends a message to a conversation.
295    async fn append(&self, message: NewMessage) -> Result<(), StoreError>;
296
297    /// The newest `limit` messages after the last reset, oldest first.
298    async fn recent(
299        &self,
300        conversation: &ConversationKey,
301        limit: usize,
302    ) -> Result<Vec<StoredMessage>, StoreError>;
303
304    /// Starts a fresh history for the conversation (`/reset`). Old messages stay for audit
305    /// until retention removes them.
306    async fn reset(&self, conversation: &ConversationKey) -> Result<(), StoreError>;
307
308    /// The native agent session of a conversation: `(backend, session id)`.
309    async fn session(
310        &self,
311        conversation: &ConversationKey,
312    ) -> Result<Option<(String, String)>, StoreError>;
313
314    /// Remembers the native agent session of a conversation (cleared by [`ConversationStore::reset`]).
315    async fn set_session(
316        &self,
317        conversation: &ConversationKey,
318        backend: &str,
319        session: &str,
320    ) -> Result<(), StoreError>;
321
322    /// Records an agent run.
323    async fn record_run(&self, run: AgentRunRecord) -> Result<(), StoreError>;
324
325    /// Accounts of the owner of a personal bot (paired with [`ConversationStore::redeem_pairing_code`]).
326    async fn owners(&self) -> Result<Vec<(ChannelId, UserId)>, StoreError>;
327
328    /// Removes an owner account. Returns whether it was one.
329    async fn remove_owner(&self, channel: &ChannelId, user: &UserId) -> Result<bool, StoreError>;
330
331    /// Stores a one-time pairing code, valid for `ttl`.
332    async fn create_pairing_code(&self, code: &str, ttl: Duration) -> Result<(), StoreError>;
333
334    /// Consumes `code` and makes `(channel, user)` an owner account when the code is valid.
335    ///
336    /// A wrong code counts as a failed attempt against every pending code; after
337    /// [`PAIRING_MAX_FAILURES`] failures the pending codes are cancelled, so codes cannot be
338    /// guessed.
339    async fn redeem_pairing_code(
340        &self,
341        code: &str,
342        channel: &ChannelId,
343        user: &UserId,
344    ) -> Result<bool, StoreError>;
345
346    /// Claims a task firing: records it as [`FireState::Sending`] the first time and returns its
347    /// current record, so a firing that was already sent is never sent again.
348    async fn claim_fire(&self, task_id: &str, fire_at: Timestamp) -> Result<Fire, StoreError>;
349
350    /// Updates a claimed firing.
351    async fn set_fire(
352        &self,
353        task_id: &str,
354        fire_at: Timestamp,
355        state: FireState,
356        attempts: u32,
357    ) -> Result<(), StoreError>;
358
359    /// The newest `limit` firings, newest first (for `okbot doctor` and `okbot task`).
360    async fn recent_fires(&self, limit: usize) -> Result<Vec<Fire>, StoreError>;
361
362    /// Remembers the direct conversation of `user` on `channel` (reminders sent there).
363    async fn set_direct_chat(
364        &self,
365        channel: &ChannelId,
366        user: &UserId,
367        conversation: &ConversationKey,
368    ) -> Result<(), StoreError>;
369
370    /// The latest direct conversation of `user` on `channel`.
371    async fn direct_chat(
372        &self,
373        channel: &ChannelId,
374        user: &UserId,
375    ) -> Result<Option<ConversationKey>, StoreError>;
376
377    /// Turns and tokens of `person` ([`person_key`]) in runs started at or after `since`.
378    async fn usage_since(&self, person: &str, since: Timestamp) -> Result<Usage, StoreError>;
379
380    /// Records a tool call.
381    async fn record_tool_call(&self, call: ToolCallRecord) -> Result<(), StoreError>;
382
383    /// Tool calls made at or after `since`, newest first, at most `limit`.
384    async fn tool_calls(
385        &self,
386        since: Timestamp,
387        limit: usize,
388    ) -> Result<Vec<ToolCallRecord>, StoreError>;
389
390    /// Grants a role until `grant.expires_at`, replacing an earlier grant of the same role to the
391    /// same account.
392    async fn grant_role(&self, grant: RoleGrant) -> Result<(), StoreError>;
393
394    /// Ends a grant before it expires. Returns whether there was one.
395    async fn revoke_role(&self, role: &str, account: &str) -> Result<bool, StoreError>;
396
397    /// Grants that have not expired at `now`.
398    async fn role_grants(&self, now: Timestamp) -> Result<Vec<RoleGrant>, StoreError>;
399
400    /// Deletes messages, runs, tool calls, expired grants, dedupe marks, finished firings and
401    /// automation records older than `before`. Returns rows removed.
402    async fn purge_before(&self, before: Timestamp) -> Result<u64, StoreError>;
403
404    /// A sealed secret (ADR 0012): the runtime seals and opens it; the store only keeps bytes.
405    async fn sealed(
406        &self,
407        kind: &str,
408        server: &str,
409        subject: &str,
410    ) -> Result<Option<Vec<u8>>, StoreError> {
411        let _ = (kind, server, subject);
412        Err(unsupported())
413    }
414
415    /// Stores or replaces a sealed secret.
416    async fn set_sealed(
417        &self,
418        kind: &str,
419        server: &str,
420        subject: &str,
421        sealed: Vec<u8>,
422    ) -> Result<(), StoreError> {
423        let _ = (kind, server, subject, sealed);
424        Err(unsupported())
425    }
426
427    /// Deletes a sealed secret. Returns whether there was one.
428    async fn delete_sealed(
429        &self,
430        kind: &str,
431        server: &str,
432        subject: &str,
433    ) -> Result<bool, StoreError> {
434        let _ = (kind, server, subject);
435        Err(unsupported())
436    }
437
438    /// The sealed secrets of `kind`, without their values, by server and subject.
439    async fn sealed_entries(&self, kind: &str) -> Result<Vec<SealedEntry>, StoreError> {
440        let _ = kind;
441        Err(unsupported())
442    }
443
444    /// Records `items` as seen by `automation` and returns those that were not seen before, in
445    /// their order.
446    async fn claim_items(
447        &self,
448        automation: &str,
449        items: Vec<String>,
450    ) -> Result<Vec<String>, StoreError> {
451        let _ = (automation, items);
452        Err(unsupported())
453    }
454
455    /// Forgets that `automation` saw `items`, so they count as new again (items not acted on).
456    async fn release_items(&self, automation: &str, items: Vec<String>) -> Result<(), StoreError> {
457        let _ = (automation, items);
458        Err(unsupported())
459    }
460
461    /// Records a run of an automation.
462    async fn record_automation_run(&self, run: AutomationRun) -> Result<(), StoreError> {
463        let _ = run;
464        Err(unsupported())
465    }
466
467    /// Runs at or after `since`, of one automation or of all, newest first, at most `limit`.
468    async fn automation_runs(
469        &self,
470        automation: Option<&str>,
471        since: Timestamp,
472        limit: usize,
473    ) -> Result<Vec<AutomationRun>, StoreError> {
474        let _ = (automation, since, limit);
475        Err(unsupported())
476    }
477
478    /// Stores a peer task (ADR 0014), replacing the one with the same id and direction.
479    async fn put_peer_task(&self, task: PeerTaskRecord) -> Result<(), StoreError> {
480        let _ = task;
481        Err(unsupported())
482    }
483
484    /// A peer task by its id and direction.
485    async fn peer_task(
486        &self,
487        inbound: bool,
488        id: &str,
489    ) -> Result<Option<PeerTaskRecord>, StoreError> {
490        let _ = (inbound, id);
491        Err(unsupported())
492    }
493
494    /// The task a message created, if any (a retried request returns the same task).
495    async fn peer_task_by_message(
496        &self,
497        inbound: bool,
498        team: &str,
499        peer: &str,
500        message_id: &str,
501    ) -> Result<Option<PeerTaskRecord>, StoreError> {
502        let _ = (inbound, team, peer, message_id);
503        Err(unsupported())
504    }
505
506    /// Peer tasks, newest first, at most `limit`.
507    async fn peer_tasks(&self, limit: usize) -> Result<Vec<PeerTaskRecord>, StoreError> {
508        let _ = limit;
509        Err(unsupported())
510    }
511
512    /// Marks inbound tasks that were not finished as failed ("interrupted"), at start-up: work with
513    /// side effects is never re-run by itself. Returns how many.
514    async fn interrupt_peer_tasks(&self, now: Timestamp) -> Result<u64, StoreError> {
515        let _ = now;
516        Err(unsupported())
517    }
518}
okbot_core/store.rs

okbot_core/
store.rs
