From 11fb11d62bb0c22ec4a769964804940343cd39ac Mon Sep 17 00:00:00 2001 From: Owain Lewis Date: Thu, 30 Jul 2026 14:13:30 +0100 Subject: [PATCH] feat(storage): move runtime state to SQLite --- ARCHITECTURE.md | 58 ++- docs/channels/imessage.md | 11 +- docs/configuration.md | 17 +- docs/prd.md | 4 +- docs/security.md | 10 +- docs/services.md | 37 +- docs/telegram.md | 12 +- src/doctor.rs | 14 +- src/gateway/mod.rs | 34 +- src/gateway/tests.rs | 81 ++-- src/history.rs | 52 +- src/store.rs | 967 ++++++++++++++++++++++++++------------ 12 files changed, 917 insertions(+), 380 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 5889cb8..cee8639 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -49,11 +49,11 @@ that answer, and delivers it. │ │ worker queues │ │ │ └───────┬───────┘ │ │ ▼ │ -│ ┌──────────────┐ ┌───────────────┐ ┌───────────────────────────────┐ │ -│ │ state.json │ │ push.db │ │ audit.jsonl │ │ -│ │ cursors and │ │ conversations │ │ redacted operational events │ │ -│ │ sessions │ │ jobs/delivery │ │ │ │ -│ └──────────────┘ └───────────────┘ └───────────────────────────────┘ │ +│ ┌───────────────────────────────────────┐ ┌──────────────────────────┐ │ +│ │ push.db │ │ audit.jsonl │ │ +│ │ conversations, jobs, delivery, │ │ redacted operational │ │ +│ │ channel cursors, backend sessions │ │ events │ │ +│ └───────────────────────────────────────┘ └──────────────────────────┘ │ └────────────────────────────────────┬─────────────────────────────────────┘ │ subprocess ▼ @@ -174,14 +174,19 @@ Push uses separate stores because they have different update and query needs: | Store | Purpose | | --- | --- | -| `$PUSH_HOME/state.json` | small atomically replaced map of channel cursors and backend session IDs | -| `$PUSH_HOME/push.db` | transactional conversation, delivery, approval compatibility, and job-run history | +| `$PUSH_HOME/state.json` | retained legacy cursor and session source for one-time migration | +| `$PUSH_HOME/push.db` | transactional conversation, delivery, approval compatibility, job-run history, channel cursors, and channel-qualified backend sessions | | `$PUSH_HOME/state.json.slack-inbox.db` | Slack Socket Mode inbox committed before envelope acknowledgement | | `$PUSH_HOME/audit.jsonl` | append-only operational audit events | | `$PUSH_HOME/run/` | advisory job locks | | `$PUSH_HOME/cache/` | disposable agent handoff files | | assistant Git repository | user-owned identity, context, evals, jobs, and project resources | +`state_path` remains a compatibility input for one-time migration. Push never +uses the legacy JSON file as a live source of truth after its migration marker +commits. The Slack inbox remains separate because its rows are an +acknowledgement queue, not gateway cursor or session state. + Runtime databases, secrets, locks, sessions, and logs must stay outside the assistant repository. @@ -226,9 +231,11 @@ The gateway runs the required preflight checks before starting. ### Step 3: Build Shared State -`GatewayGroup::new` opens one shared `Store`, one shared `History`, one runner -per required backend, and one serialized audit-log lock. It then creates one -`Gateway` for each enabled channel. +`GatewayGroup::new` opens one shared transactional `Store` and one shared +`History` connection to `push.db`, one runner per required backend, and one +serialized audit-log lock. Opening the store also performs any one-time legacy +`state.json` migration before polling can begin. It then creates one `Gateway` +for each enabled channel. Each gateway owns its own: @@ -517,9 +524,16 @@ configured, unattended, and evaluator modes. ## 7. Durable Data Model -### `state.json` +### Runtime State in `push.db` -[`src/store.rs`](src/store.rs) owns a small JSON document: +[`src/store.rs`](src/store.rs) owns channel cursors and backend session +mappings in the canonical SQLite database. Cursor advancement uses a +monotonic upsert, so concurrent or repeated writes cannot move a channel +backwards. Session reads, backend changes, backend-owned ID updates, and +`/clear` rotation use transactions keyed by channel and thread. + +The first startup after upgrade imports the legacy document configured by +`state_path`: ```json { @@ -539,9 +553,18 @@ configured, unattended, and evaluator modes. } ``` -`last_row_id` and the field name `uuid` remain for backward compatibility. -Writes use a new owner-only temporary file followed by atomic rename. In-memory -state is rolled back if persistence fails. +`last_row_id` remains an iMessage fallback only when the JSON has no explicit +`cursors.imessage` value. Unqualified session keys are imported as iMessage +keys; an explicit channel-qualified key wins when both forms exist. Empty +session IDs are preserved so the next request starts fresh. + +The imported cursor/session rows and a source-path migration marker commit in +one immediate transaction. A crash or error before commit leaves no partial +import, and startup retries from the unchanged JSON file. After commit, later +starts ignore that source even though Push deliberately leaves it in place as +a private recovery copy. Restoring a pre-migration database, or removing the +database after preserving other needed data, allows the retained file to be +imported again. Corrupt JSON stops startup with recovery guidance. ### `push.db` @@ -556,6 +579,9 @@ timeout. | `approval_questions` | retained durable question and answer state | | `job_runs` | immutable job claims, bounded results, evaluation, scheduling, and delivery | | `gateway_control_actions` | idempotent control actions such as `/stop` targets | +| `channel_cursors` | monotonic per-channel polling checkpoints | +| `backend_sessions` | current backend session for each channel and thread | +| `legacy_state_migrations` | atomic record of each imported legacy JSON source | Important constraints: @@ -807,7 +833,7 @@ result before proactive delivery. | [`src/claude.rs`](src/claude.rs) | Claude Code CLI adapter | | [`src/codex.rs`](src/codex.rs) | Codex CLI adapter | | [`src/pi.rs`](src/pi.rs) | Pi CLI adapter | -| [`src/store.rs`](src/store.rs) | atomic cursor and session state | +| [`src/store.rs`](src/store.rs) | transactional SQLite cursor/session state and legacy JSON migration | | [`src/history.rs`](src/history.rs) | SQLite schema, conversation history, delivery, and migrations | | [`src/jobs.rs`](src/jobs.rs) | runbook validation, execution, evaluation, scheduler, and ledger | | [`src/audit.rs`](src/audit.rs) | redacted JSONL audit log | diff --git a/docs/channels/imessage.md b/docs/channels/imessage.md index a2ad829..b0bf894 100644 --- a/docs/channels/imessage.md +++ b/docs/channels/imessage.md @@ -75,12 +75,13 @@ See [configuration](../configuration.md#routing) for route precedence. On the first iMessage start, Push records the newest existing Messages row without running it. Send a new message after the gateway starts. Later starts -continue after the last completed row stored in `state.json`. +continue after the last completed row stored in `push.db`. -Push stores the last completed Messages row in `state.json` and accepted -conversation turns in `push.db`. It advances the cursor only after a row is -ignored or completed. An earlier in-flight row prevents later completed rows -from pushing the cursor past it. +Push stores both the last completed Messages row and accepted conversation +turns in `push.db`. It advances the cursor only after a row is ignored or +completed. An earlier in-flight row prevents later completed rows from pushing +the cursor past it. Existing `state.json` data is imported once on upgrade and +the original file remains as a recovery copy. If a generated outbound reply was stored before a crash, restart delivers the stored reply without generating a different second answer. A crash before the diff --git a/docs/configuration.md b/docs/configuration.md index d9d9fc9..d8207d4 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -256,8 +256,8 @@ requests. Review [permissions and security](security.md) before enabling jobs. | Setting | Default | Purpose | | --- | --- | --- | | `assistant_root` | required for new setups | Canonical root of the one assistant repository; `SOUL.md`, `context/`, `evals/`, and `jobs/` are derived | -| `state_path` | `$PUSH_HOME/state.json` | Channel cursors and backend session IDs | -| `database_path` | `$PUSH_HOME/push.db` | Canonical conversation, approval, and job history | +| `state_path` | `$PUSH_HOME/state.json` | Legacy JSON source retained for one-time cursor and session migration | +| `database_path` | `$PUSH_HOME/push.db` | Canonical history, jobs, delivery, channel cursors, and backend sessions | | `audit_log_path` | `$PUSH_HOME/audit.jsonl` | Structured local audit log | | `audit_log_content` | `false` | Include message and reply content in audit events | @@ -325,6 +325,13 @@ jobs path is exactly `/jobs`. For separate legacy paths, move Explicit `state_path`, `database_path`, `audit_log_path`, and `jobs_run_dir` settings remain supported for the rest of the 0.x release line. Push does not -move or rewrite data at those paths. A future removal would require an -announced major release and migration instructions. New installations should -omit them and let `PUSH_HOME` own the whole runtime layout. +move or rewrite data at the database, audit, or run-lock paths. On startup, +Push imports an existing JSON file at `state_path` into `database_path` once, +records the import in the same SQLite transaction, and leaves the JSON +untouched as a recovery copy. After that commit, Push reads and writes cursor +and backend-session state only in SQLite. Do not edit the retained JSON +expecting live state to change. + +A future removal of these settings would require an announced major release +and migration instructions. New installations should omit them and let +`PUSH_HOME` own the whole runtime layout. diff --git a/docs/prd.md b/docs/prd.md index 4f9996e..f503a57 100644 --- a/docs/prd.md +++ b/docs/prd.md @@ -161,10 +161,10 @@ user prompt. | `telegram.bot_token` | Telegram bot token stored in the private config. | | `telegram.allow_user_ids` | Allowed private-chat sender IDs. | | `telegram.allow_chat_ids` | Allowed private chat IDs. | -| `state_path` | JSON state path. | +| `state_path` | Legacy JSON cursor/session source retained for one-time migration compatibility. | | `audit_log_path` | Local JSONL audit log path. | | `audit_log_content` | Whether audit events include message and reply text. | -| `database_path` | Canonical SQLite history path; defaults to `$PUSH_HOME/push.db`. | +| `database_path` | Canonical SQLite history, job, delivery, cursor, and backend-session path; defaults to `$PUSH_HOME/push.db`. | ## Control Commands diff --git a/docs/security.md b/docs/security.md index 0b58c0c..d56b288 100644 --- a/docs/security.md +++ b/docs/security.md @@ -62,8 +62,8 @@ service environment using the narrowest policy that works. | --- | --- | | `$PUSH_HOME/config.toml` | allowlists, routes, paths, and possibly credentials | | `/` | Git-versioned identity, context, evals, jobs, and optional project skills | -| `$PUSH_HOME/state.json` | channel cursors and backend session IDs | -| `$PUSH_HOME/push.db` | conversation history, approvals, and job runs | +| `$PUSH_HOME/state.json` | retained legacy cursor and session recovery copy after migration | +| `$PUSH_HOME/push.db` | conversation history, approvals, jobs, delivery, channel cursors, and backend sessions | | `$PUSH_HOME/state.json.slack-inbox.db` | Slack envelopes accepted before gateway processing | | `$PUSH_HOME/audit.jsonl` | metadata, errors, handles, and optional content | | `$PUSH_HOME/run/` | local job lock files | @@ -110,6 +110,12 @@ error, and character counts. Message and reply content are omitted unless The redacted log is still sensitive because it can contain handles, thread IDs, file paths, and backend errors. Protect and rotate it like a service log. +The legacy state migration repairs the JSON file to owner-only permissions and +does not delete or rewrite it. Keep that recovery copy private until your +normal encrypted backup process has captured the migrated `push.db`. The +database is the live source of truth after migration, so backing up only +`state.json` is not sufficient. + ## Deployment checklist - allow only identities you control diff --git a/docs/services.md b/docs/services.md index 48ee7a5..43bc37e 100644 --- a/docs/services.md +++ b/docs/services.md @@ -22,6 +22,11 @@ Set one absolute `PUSH_HOME` in the service definition. It defaults to `~/.push` for interactive commands. The service user needs: - read and write access to `PUSH_HOME` +- access to the configured `config.toml` +- read access and owner control of an existing legacy `state_path` for migration +- write access to `audit_log_path` +- write access to `database_path` +- write access to `jobs_run_dir` - filesystem access to `assistant_root` as allowed by the selected agent - agent write access to `assistant_root/jobs/` when jobs should be created from chat - access to the selected `claude`, `codex`, or `pi` executable on `PATH` @@ -35,12 +40,14 @@ Set one absolute `PUSH_HOME` in the service definition. It defaults to `OPENAI_API_KEY` in the service environment, plus network access to `api.openai.com` -`$PUSH_HOME/state.json` stores independent cursors for each channel and backend -session IDs. `$PUSH_HOME/push.db` stores the canonical conversation and job -journal. The audit log, Slack recovery inbox, job locks, and cache are derived -from the same root. Chat agents run from `assistant_root`. Keep `PUSH_HOME` on -durable storage. Restarting the service resumes after the last completed row -and reuses existing backend sessions when the backend for that thread has not +`database_path` stores the canonical conversation journal, channel cursors, and +backend session mappings. `state_path` is only a legacy JSON migration source +and retained recovery copy; Push does not write live state to it. The audit +log, Slack recovery inbox, job locks, and cache remain separate paths derived +from `PUSH_HOME`, unless their documented compatibility settings override +them. Chat agents run from `assistant_root`. Keep these paths on durable +storage. Restarting the service resumes after the last completed row and +reuses existing backend sessions when the backend for that thread has not changed. Keep `assistant_root` in its own Git repository. Keep config secrets, state, @@ -209,6 +216,24 @@ pending result delivery; it does not catch up missed cron times or rerun interrupted agent execution. Use `push job runs` to distinguish execution state from delivery attempts. +## Backup and state migration recovery + +Stop the managed service before taking a filesystem copy of `push.db`, or use +SQLite's online backup tooling. The database contains conversation history, +job and delivery state, channel cursors, and backend session mappings. Back up +the audit log and assistant repository separately. Slack's durable inbox stays +in `.slack-inbox.db`; include it when preserving unprocessed Slack +events. + +After an upgrade, Push imports an existing configured `state_path` in one +transaction and leaves that JSON file unchanged. Keep it as a private recovery +copy until a verified database backup exists. If migration fails, fix or +restore the JSON and restart; Push will not poll while the import is +incomplete. If the post-migration database is lost, restore `push.db` from +backup. As a last resort, move the unusable database aside and restart with the +retained JSON to recover its older cursors and sessions, understanding that +conversation, job, and delivery records not present in JSON will be absent. + ## Agent-created jobs When asked, the agent writes jobs directly under `/jobs` and diff --git a/docs/telegram.md b/docs/telegram.md index 1e21e7e..0a962cb 100644 --- a/docs/telegram.md +++ b/docs/telegram.md @@ -120,16 +120,17 @@ state. ## Cursor and Restart Behavior -`state.json` stores independent `imessage` and `telegram` cursors. `push.db` -stores channel-qualified canonical conversations, so Telegram and iMessage -history cannot collide. On the first +`push.db` stores independent `imessage` and `telegram` cursors plus +channel-qualified canonical conversations and backend sessions, so Telegram +and iMessage state cannot collide. On the first Telegram start, Push asks Telegram for the newest pending update and records its id without running it. This explicit backlog skip prevents old bot messages from unexpectedly starting agent work. Later accepted and ignored updates advance only the Telegram cursor. Restarts continue from the next update id. As with iMessage, a crash after delivery but before cursor persistence can -repeat a reply. Keep `state.json` on durable storage. +repeat a reply. Keep `push.db` on durable storage. Existing `state.json` data +is imported once and retained unchanged as a recovery copy. ## Linux and Service Mode @@ -142,8 +143,7 @@ directly in a world-readable unit file. Protect these files as credentials or private assistant data: - the bot token and configuration -- `state.json` and the audit log -- `push.db` +- `push.db`, the retained legacy `state.json`, and the audit log - the private `assistant_root` repository, including identity, context, jobs, and optional project skills diff --git a/src/doctor.rs b/src/doctor.rs index c8589b2..6de49a1 100644 --- a/src/doctor.rs +++ b/src/doctor.rs @@ -5,7 +5,7 @@ use std::path::{Path, PathBuf}; use anyhow::{bail, Result}; -use crate::{channel::Channel, config, history, jobs}; +use crate::{channel::Channel, config, jobs}; use config::{SLACK_APP_TOKEN_ENV, SLACK_BOT_TOKEN_ENV, TELEGRAM_BOT_TOKEN_ENV}; /// Fails fast with actionable messages when the environment is not ready. @@ -256,16 +256,20 @@ fn check_writable_dir(name: &str, field: &str, dir: &Path, checks: &mut Vec) { - match history::History::open(&cfg.paths.database) { + match crate::store::Store::open(&cfg.paths) { Ok(_) => checks.push(Check::pass( "conversation database", - format!("{} is ready", cfg.paths.database.display()), + format!( + "{} is ready for history and runtime state", + cfg.paths.database.display() + ), )), Err(error) => checks.push(Check::fail( "conversation database", format!( - "cannot open {}: {error}. Choose a writable database_path and repair or remove an invalid database.", - cfg.paths.database.display() + "cannot prepare {} or migrate legacy state from {}: {error:#}. Repair the database or legacy JSON before starting Push.", + cfg.paths.database.display(), + cfg.paths.state.display() ), )), } diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 3f7a7f9..47fe528 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -116,7 +116,7 @@ struct WorkerState { impl GatewayGroup { pub fn new(cfg: Config) -> Result { let enabled = cfg.enabled_channel_kinds()?; - let store = Arc::new(Mutex::new(Store::open(&cfg.paths.state)?)); + let store = Arc::new(Mutex::new(Store::open(&cfg.paths)?)); let history = Arc::new(Mutex::new( History::open(&cfg.paths.database).with_context(|| { format!( @@ -333,7 +333,7 @@ where impl Gateway { #[cfg_attr(not(test), allow(dead_code))] pub fn new(cfg: Config) -> Result { - let store = Arc::new(Mutex::new(Store::open(&cfg.paths.state)?)); + let store = Arc::new(Mutex::new(Store::open(&cfg.paths)?)); let history = Arc::new(Mutex::new( History::open(&cfg.paths.database).with_context(|| { format!( @@ -493,7 +493,13 @@ impl Gateway { async fn skip_backlog(&self) -> Result<()> { let channel_id = self.channel.id(); - if self.store.lock().unwrap().has_cursor(channel_id) { + if self + .store + .lock() + .unwrap() + .has_cursor(channel_id) + .with_context(|| format!("check initial {channel_id} cursor"))? + { return Ok(()); } let max = self @@ -511,7 +517,16 @@ impl Gateway { } async fn tick(&mut self) { - let since = self.store.lock().unwrap().cursor(self.channel.id()); + let since = match self.store.lock().unwrap().cursor(self.channel.id()) { + Ok(cursor) => cursor, + Err(error) => { + error!( + "{} cursor read error; polling paused: {error:#}", + self.channel.id() + ); + return; + } + }; let msgs = match self.channel.poll(since).await { Ok(messages) => messages, Err(e) => { @@ -531,7 +546,16 @@ impl Gateway { async fn process_messages(&mut self, msgs: Vec) { self.recover_closed_workers(); persist_cursor(&self.store, &self.ack, self.channel.id()); - let since = self.store.lock().unwrap().cursor(self.channel.id()); + let since = match self.store.lock().unwrap().cursor(self.channel.id()) { + Ok(cursor) => cursor, + Err(error) => { + error!( + "{} cursor read error; message processing paused: {error:#}", + self.channel.id() + ); + return; + } + }; for m in &msgs { if m.row_id <= since { continue; diff --git a/src/gateway/tests.rs b/src/gateway/tests.rs index 92bc1e4..ef773f6 100644 --- a/src/gateway/tests.rs +++ b/src/gateway/tests.rs @@ -485,16 +485,22 @@ async fn cursor_save_failure_retries_without_rerunning_or_redelivering() { assert_eq!(calls.lock().unwrap().len(), 1); assert_eq!(gateway.ctx.sent_replies.lock().unwrap().len(), 1); - assert_eq!(gateway.store.lock().unwrap().cursor("imessage"), 0); + assert_eq!(gateway.store.lock().unwrap().cursor("imessage").unwrap(), 0); assert!(gateway.ack.lock().unwrap().completed.contains(&1)); gateway.tick_fake(vec![inbound]).await; assert_eq!(calls.lock().unwrap().len(), 1); assert_eq!(gateway.ctx.sent_replies.lock().unwrap().len(), 1); - assert_eq!(gateway.store.lock().unwrap().cursor("imessage"), 1); + assert_eq!(gateway.store.lock().unwrap().cursor("imessage").unwrap(), 1); assert!(gateway.ack.lock().unwrap().completed.is_empty()); - assert_eq!(Store::open(&state_path).unwrap().cursor("imessage"), 1); + assert_eq!( + Store::open_at(format!("{state_path}.db"), &state_path) + .unwrap() + .cursor("imessage") + .unwrap(), + 1 + ); let _ = std::fs::remove_file(&state_path); let _ = std::fs::remove_file(format!("{state_path}.db")); @@ -506,7 +512,9 @@ async fn cursor_save_failure_retries_without_rerunning_or_redelivering() { #[test] fn setup_failure_completion_unblocks_later_completed_rows() { let path = temp_state_path(); - let store = Arc::new(Mutex::new(Store::open(&path).unwrap())); + let store = Arc::new(Mutex::new( + Store::open_at(format!("{path}.db"), &path).unwrap(), + )); let ack = Arc::new(Mutex::new(AckState::default())); { let mut ack = ack.lock().unwrap(); @@ -526,12 +534,12 @@ fn setup_failure_completion_unblocks_later_completed_rows() { #[tokio::test] async fn session_lookup_failure_completes_in_flight_row() { - let blocker = temp_path("state-blocker"); - let state_path = blocker.join("state.json"); + let state_path = temp_state_path(); let store = Arc::new(Mutex::new( - Store::open(state_path.to_str().unwrap()).unwrap(), + Store::open_at(format!("{state_path}.db"), &state_path).unwrap(), )); - std::fs::write(&blocker, "not a directory").unwrap(); + store.lock().unwrap().fail_next_session_save_for_test(); + store.lock().unwrap().fail_next_cursor_save_for_test(); let ack = Arc::new(Mutex::new(AckState::default())); { let mut ack = ack.lock().unwrap(); @@ -554,13 +562,15 @@ async fn session_lookup_failure_completes_in_flight_row() { assert!(ack.in_flight.is_empty()); assert_eq!(ack.completed.iter().copied().collect::>(), [10, 11]); - let _ = std::fs::remove_file(blocker); + let _ = std::fs::remove_file(format!("{state_path}.db")); } #[tokio::test] async fn soul_read_failure_stops_backend_dispatch_and_completes_row() { let state_path = temp_state_path(); - let store = Arc::new(Mutex::new(Store::open(&state_path).unwrap())); + let store = Arc::new(Mutex::new( + Store::open_at(format!("{state_path}.db"), &state_path).unwrap(), + )); let sessions_dir = temp_path("soul-failure-sessions"); let assistant_dir = temp_path("soul-failure-assistant"); std::fs::create_dir_all(&assistant_dir).unwrap(); @@ -1181,7 +1191,7 @@ async fn canonical_history_failure_prevents_backend_dispatch_and_cursor_advance( assert!(gateway.handles.is_empty()); assert!(calls.lock().unwrap().is_empty()); assert!(gateway.ctx.sent_replies.lock().unwrap().is_empty()); - assert_eq!(gateway.store.lock().unwrap().cursor("imessage"), 0); + assert_eq!(gateway.store.lock().unwrap().cursor("imessage").unwrap(), 0); let _ = std::fs::remove_file(&state_path); let _ = std::fs::remove_file(format!("{state_path}.db")); @@ -1248,7 +1258,7 @@ async fn pending_outbound_is_delivered_after_restart_without_backend_rerun() { .status, DeliveryStatus::Delivered ); - assert_eq!(gateway.store.lock().unwrap().cursor("imessage"), 1); + assert_eq!(gateway.store.lock().unwrap().cursor("imessage").unwrap(), 1); let _ = std::fs::remove_file(&state_path); let _ = std::fs::remove_file(format!("{state_path}.db")); @@ -1262,9 +1272,7 @@ async fn session_state_save_failure_keeps_reply_for_restart_without_backend_reru let state_path = temp_state_path(); let sessions_dir = temp_path("session-save-failure-sessions"); let assistant_dir = temp_path("session-save-failure-assistant"); - let state_blocker = temp_path("session-save-failure-blocker"); std::fs::create_dir_all(&assistant_dir).unwrap(); - std::fs::write(&state_blocker, "not a directory").unwrap(); let first_calls = Arc::new(Mutex::new(Vec::new())); let mut gateway = Gateway::new(test_config( &state_path, @@ -1273,14 +1281,10 @@ async fn session_state_save_failure_keeps_reply_for_restart_without_backend_reru )) .unwrap(); let store = gateway.store.clone(); - let broken_state_path = state_blocker.join("state.json"); gateway.ctx.runners = Arc::new(fake_runners_with_hook( first_calls.clone(), Some(Arc::new(move || { - store - .lock() - .unwrap() - .set_path_for_test(broken_state_path.clone()); + store.lock().unwrap().fail_next_session_save_for_test(); })), )); gateway @@ -1339,12 +1343,14 @@ async fn session_state_save_failure_keeps_reply_for_restart_without_backend_reru "fake reply: hello\n\n-- sent by push".to_string() )] ); - assert_eq!(restarted.store.lock().unwrap().cursor("imessage"), 1); + assert_eq!( + restarted.store.lock().unwrap().cursor("imessage").unwrap(), + 1 + ); let _ = std::fs::remove_file(&state_path); let _ = std::fs::remove_file(format!("{state_path}.db")); let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); - let _ = std::fs::remove_file(state_blocker); let _ = std::fs::remove_dir_all(sessions_dir); let _ = std::fs::remove_dir_all(assistant_dir); } @@ -1373,7 +1379,7 @@ async fn exhausted_delivery_batch_retries_without_blocking_cursor() { assert_eq!(calls.lock().unwrap().len(), 1); assert_eq!(gateway.ctx.sent_replies.lock().unwrap().len(), 1); - assert_eq!(gateway.store.lock().unwrap().cursor("imessage"), 1); + assert_eq!(gateway.store.lock().unwrap().cursor("imessage").unwrap(), 1); let inbound_id = gateway .ctx .history @@ -1436,7 +1442,10 @@ async fn telegram_filters_before_agent_and_replies_to_originating_chat() { gateway.queues.clear(); gateway.drain_workers().await; - assert_eq!(gateway.store.lock().unwrap().cursor("telegram"), 12); + assert_eq!( + gateway.store.lock().unwrap().cursor("telegram").unwrap(), + 12 + ); assert_eq!(calls.lock().unwrap().len(), 1); assert_eq!( crate::prompt::current_message(&calls.lock().unwrap()[0].prompt).as_deref(), @@ -1539,8 +1548,8 @@ async fn enabled_channels_process_concurrently_with_isolated_state_and_origin_re tokio::join!(imessage[0].drain_workers(), telegram[0].drain_workers()); let store = imessage[0].store.lock().unwrap(); - assert_eq!(store.cursor("imessage"), 5); - assert_eq!(store.cursor("telegram"), 5); + assert_eq!(store.cursor("imessage").unwrap(), 5); + assert_eq!(store.cursor("telegram").unwrap(), 5); drop(store); assert_eq!( imessage[0].ctx.sent_replies.lock().unwrap().as_slice(), @@ -1561,9 +1570,19 @@ async fn enabled_channels_process_concurrently_with_isolated_state_and_origin_re .collect::>(); assert!(prompts.contains(&"from imessage".to_string())); assert!(prompts.contains(&"from telegram".to_string())); - let persisted = std::fs::read_to_string(&state_path).unwrap(); - assert!(persisted.contains("imessage:dm:+15551234567")); - assert!(persisted.contains("telegram:dm:7")); + let mut store = Store::open_at(format!("{state_path}.db"), &state_path).unwrap(); + assert_eq!( + store + .session_for("imessage:dm:+15551234567", "codex", "unused".to_string()) + .unwrap(), + ("fake-session".to_string(), false) + ); + assert_eq!( + store + .session_for("telegram:dm:7", "codex", "unused".to_string()) + .unwrap(), + ("fake-session".to_string(), false) + ); let _ = std::fs::remove_file(&state_path); let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); @@ -1604,7 +1623,7 @@ async fn telegram_voice_is_transcribed_and_gets_text_and_voice_replies() { gateway.ctx.sent_voice_replies.lock().unwrap().as_slice(), [("7".to_string(), vec![4, 5, 6])] ); - assert_eq!(gateway.store.lock().unwrap().cursor("telegram"), 1); + assert_eq!(gateway.store.lock().unwrap().cursor("telegram").unwrap(), 1); let history = gateway .ctx .history @@ -1652,7 +1671,7 @@ async fn telegram_voice_without_openai_key_falls_back_without_running_agent() { let reply = &replies[0].1; assert!(reply.contains("voice.openai_api_key")); assert!(reply.contains("OPENAI_API_KEY")); - assert_eq!(gateway.store.lock().unwrap().cursor("telegram"), 1); + assert_eq!(gateway.store.lock().unwrap().cursor("telegram").unwrap(), 1); let _ = std::fs::remove_file(&state_path); let _ = std::fs::remove_file(format!("{state_path}.db")); @@ -2138,7 +2157,7 @@ async fn failed_stop_history_write_retries_before_later_rows() { assert!(calls.lock().unwrap().is_empty()); assert!(gateway.queues.is_empty()); - assert_eq!(gateway.store.lock().unwrap().cursor("imessage"), 0); + assert_eq!(gateway.store.lock().unwrap().cursor("imessage").unwrap(), 0); gateway .ctx diff --git a/src/history.rs b/src/history.rs index dec8a8e..bae0342 100644 --- a/src/history.rs +++ b/src/history.rs @@ -12,7 +12,7 @@ use crate::approval::{parse_answer, AnswerOrigin, AnswerOutcome, NormalizedAnswe #[cfg(test)] use crate::approval::{DeliveryStatus as ApprovalDeliveryStatus, Question}; -const SCHEMA_VERSION: i64 = 10; +const SCHEMA_VERSION: i64 = 11; const RETIRED_JOB_APPROVAL_ERROR: &str = "job approval was removed; request direct job creation"; const MAX_HISTORY_READ_BYTES: usize = 8 * 1024; const READ_TRUNCATED: &str = "\n[truncated by push while reading history]"; @@ -825,6 +825,34 @@ fn migrate(conn: &Connection) -> Result<()> { )?; conn.execute_batch("PRAGMA user_version = 10;")?; } + if version <= 10 { + conn.execute_batch( + "CREATE TABLE channel_cursors ( + channel TEXT PRIMARY KEY, + cursor INTEGER NOT NULL, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + CHECK(length(trim(channel)) > 0) + ); + CREATE TABLE backend_sessions ( + channel TEXT NOT NULL, + thread_key TEXT NOT NULL, + backend TEXT NOT NULL, + session_id TEXT NOT NULL, + started INTEGER NOT NULL CHECK(started IN (0, 1)), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + PRIMARY KEY(channel, thread_key), + CHECK(length(trim(channel)) > 0), + CHECK(length(trim(thread_key)) > 0), + CHECK(length(trim(backend)) > 0) + ); + CREATE TABLE legacy_state_migrations ( + source_path TEXT PRIMARY KEY, + source_sha256 TEXT NOT NULL, + migrated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) + ); + PRAGMA user_version = 11;", + )?; + } conn.execute_batch("COMMIT;")?; Ok(()) } @@ -927,7 +955,11 @@ mod tests { let path = temp_path("message-delivery-v8-migration"); let history = History::open(path.to_str().unwrap()).unwrap(); history.execute_batch_for_test( - "ALTER TABLE messages DROP COLUMN delivery_chunk_index; PRAGMA user_version = 8;", + "ALTER TABLE messages DROP COLUMN delivery_chunk_index; + DROP TABLE legacy_state_migrations; + DROP TABLE backend_sessions; + DROP TABLE channel_cursors; + PRAGMA user_version = 8;", ); drop(history); @@ -964,7 +996,12 @@ mod tests { [&question.id], ) .unwrap(); - history.execute_batch_for_test("PRAGMA user_version = 9;"); + history.execute_batch_for_test( + "DROP TABLE legacy_state_migrations; + DROP TABLE backend_sessions; + DROP TABLE channel_cursors; + PRAGMA user_version = 9;", + ); drop(history); let mut reopened = History::open(path.to_str().unwrap()).unwrap(); @@ -1024,6 +1061,9 @@ mod tests { DROP TABLE job_runs; DROP TABLE approval_questions; ALTER TABLE messages DROP COLUMN delivery_chunk_index; + DROP TABLE legacy_state_migrations; + DROP TABLE backend_sessions; + DROP TABLE channel_cursors; PRAGMA user_version = 1;", ); drop(history); @@ -1053,6 +1093,9 @@ mod tests { DROP TABLE job_draft_proposals; DROP TABLE job_runs; ALTER TABLE messages DROP COLUMN delivery_chunk_index; + DROP TABLE legacy_state_migrations; + DROP TABLE backend_sessions; + DROP TABLE channel_cursors; PRAGMA user_version = 2;", ); drop(history); @@ -1112,6 +1155,9 @@ mod tests { ALTER TABLE job_runs DROP COLUMN evaluation_state; DROP TABLE gateway_control_actions; ALTER TABLE messages DROP COLUMN delivery_chunk_index; + DROP TABLE legacy_state_migrations; + DROP TABLE backend_sessions; + DROP TABLE channel_cursors; PRAGMA user_version = 6;", ); drop(history); diff --git a/src/store.rs b/src/store.rs index 40b66c9..dcab2a8 100644 --- a/src/store.rs +++ b/src/store.rs @@ -1,26 +1,33 @@ -//! Persisted gateway state: last processed message ROWID and the mapping from -//! conversation thread to the active agent backend session. +//! Transactional channel cursors and backend session mappings. +//! +//! New state lives in the canonical SQLite database. On startup, an existing +//! JSON state file is imported once in the same transaction as its migration +//! marker. The source file remains untouched as an operator recovery copy. use std::collections::HashMap; -use std::fs::OpenOptions; -use std::io::{ErrorKind, Write}; +use std::io::ErrorKind; use std::path::{Path, PathBuf}; +use std::time::Duration; -use anyhow::{anyhow, Context, Result}; -use serde::{Deserialize, Serialize}; +use anyhow::{bail, Context, Result}; +use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior}; +use serde::Deserialize; +use sha2::{Digest, Sha256}; +use crate::paths::PushPaths; use crate::util::non_empty_session_id; -#[derive(Serialize, Deserialize, Clone)] -pub struct SessionInfo { - pub uuid: String, + +#[derive(Deserialize, Clone)] +struct SessionInfo { + uuid: String, #[serde(default)] - pub started: bool, + started: bool, #[serde(default = "default_backend")] - pub backend: String, + backend: String, } -#[derive(Serialize, Deserialize, Default)] -struct State { +#[derive(Deserialize, Default)] +struct LegacyState { #[serde(default)] last_row_id: i64, #[serde(default)] @@ -29,85 +36,131 @@ struct State { sessions: HashMap, } -/// Store owns the persisted state and writes it atomically on every change. +/// Store owns transactional cursor and session state in `database_path`. pub struct Store { - path: PathBuf, - state: State, + database_path: PathBuf, + legacy_state_path: PathBuf, + conn: Connection, #[cfg(test)] cursor_save_failures_remaining: usize, + #[cfg(test)] + session_save_failures_remaining: usize, } impl Store { - pub fn open(path: impl AsRef) -> Result { - let p = path.as_ref().to_path_buf(); - let display = p.display(); - let state = match std::fs::read_to_string(&p) { - Ok(s) => { - crate::util::restrict_permissions(&p, false) - .with_context(|| format!("restrict state permissions {display}"))?; - serde_json::from_str(&s).with_context(|| format!("parse state {display}"))? - } - Err(e) if e.kind() == ErrorKind::NotFound => State::default(), - Err(e) => return Err(anyhow!("read state {display}: {e}")), - }; - Ok(Store { - path: p, - state, + pub fn open(paths: &PushPaths) -> Result { + Self::open_inner(&paths.database, &paths.state, false) + } + + #[cfg(test)] + pub(crate) fn open_at( + database_path: impl AsRef, + legacy_state_path: impl AsRef, + ) -> Result { + Self::open_inner(database_path.as_ref(), legacy_state_path.as_ref(), false) + } + + fn open_inner( + database_path: &Path, + legacy_state_path: &Path, + #[cfg_attr(not(test), allow(unused_variables))] fail_migration_before_commit: bool, + ) -> Result { + // History owns all forward schema migrations for the canonical database. + drop( + crate::history::History::open(database_path).with_context(|| { + format!( + "prepare canonical state database {}", + database_path.display() + ) + })?, + ); + + let database_path = database_path.to_path_buf(); + let conn = Connection::open(&database_path) + .with_context(|| format!("open state database {}", database_path.display()))?; + crate::util::restrict_permissions(&database_path, false).with_context(|| { + format!("restrict database permissions {}", database_path.display()) + })?; + conn.busy_timeout(Duration::from_secs(5)) + .context("configure state database busy timeout")?; + conn.execute_batch("PRAGMA foreign_keys = ON;") + .context("enable state database foreign keys")?; + + let mut store = Self { + database_path, + legacy_state_path: legacy_state_path.to_path_buf(), + conn, #[cfg(test)] cursor_save_failures_remaining: 0, - }) + #[cfg(test)] + session_save_failures_remaining: 0, + }; + store.migrate_legacy_state(fail_migration_before_commit)?; + Ok(store) } #[cfg(test)] pub fn last_row(&self) -> i64 { - self.cursor("imessage") + self.cursor("imessage").unwrap() } - pub fn has_cursor(&self, channel: &str) -> bool { - self.state.cursors.contains_key(channel) - || (channel == "imessage" && self.state.last_row_id != 0) + pub fn has_cursor(&self, channel: &str) -> Result { + self.conn + .query_row( + "SELECT 1 FROM channel_cursors WHERE channel = ?1", + [channel], + |_| Ok(()), + ) + .optional() + .with_context(|| { + format!( + "read {channel} cursor existence from {}", + self.database_path.display() + ) + }) + .map(|cursor| cursor.is_some()) } - pub fn cursor(&self, channel: &str) -> i64 { - self.state.cursors.get(channel).copied().unwrap_or_else(|| { - if channel == "imessage" { - self.state.last_row_id - } else { - 0 - } - }) + pub fn cursor(&self, channel: &str) -> Result { + self.conn + .query_row( + "SELECT cursor FROM channel_cursors WHERE channel = ?1", + [channel], + |row| row.get(0), + ) + .optional() + .with_context(|| { + format!( + "read {channel} cursor from {}", + self.database_path.display() + ) + }) + .map(|cursor| cursor.unwrap_or(0)) } pub fn set_cursor(&mut self, channel: &str, id: i64) -> Result<()> { - if self.has_cursor(channel) && id <= self.cursor(channel) { - return Ok(()); - } - let previous_cursor = self.state.cursors.insert(channel.to_string(), id); - let previous_last_row_id = self.state.last_row_id; - if channel == "imessage" { - self.state.last_row_id = id; - } + validate_channel(channel)?; #[cfg(test)] - let result = if self.cursor_save_failures_remaining > 0 { + if self.cursor_save_failures_remaining > 0 { self.cursor_save_failures_remaining -= 1; - Err(anyhow!("injected cursor save failure")) - } else { - self.save() - }; - #[cfg(not(test))] - let result = self.save(); - if let Err(error) = result { - match previous_cursor { - Some(previous) => { - self.state.cursors.insert(channel.to_string(), previous); - } - None => { - self.state.cursors.remove(channel); - } - } - self.state.last_row_id = previous_last_row_id; - return Err(error); + return Err(anyhow::anyhow!("injected cursor save failure")); } + self.conn + .execute( + "INSERT INTO channel_cursors (channel, cursor) + VALUES (?1, ?2) + ON CONFLICT(channel) DO UPDATE SET + cursor = excluded.cursor, + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE excluded.cursor > channel_cursors.cursor", + params![channel, id], + ) + .with_context(|| { + format!( + "advance {channel} cursor transactionally in {}", + self.database_path.display() + ) + })?; Ok(()) } @@ -119,121 +172,336 @@ impl Store { backend: &str, initial_id: String, ) -> Result<(String, bool)> { - self.migrate_legacy_imessage_session(thread)?; - if let Some(si) = self.state.sessions.get(thread) { - if si.backend == backend { - if si.uuid.trim().is_empty() { - self.state.sessions.insert( - thread.to_string(), - SessionInfo { - uuid: initial_id.clone(), - started: false, - backend: backend.to_string(), - }, - ); - self.save()?; - return Ok((initial_id, true)); - } - return Ok((si.uuid.clone(), !si.started)); + validate_backend(backend)?; + let (channel, thread_key) = split_thread(thread)?; + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .with_context(|| { + format!( + "begin session transaction in {}", + self.database_path.display() + ) + })?; + let existing = tx + .query_row( + "SELECT backend, session_id, started + FROM backend_sessions + WHERE channel = ?1 AND thread_key = ?2", + params![channel, thread_key], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, bool>(2)?, + )) + }, + ) + .optional() + .with_context(|| format!("read session for {thread:?}"))?; + + if let Some((stored_backend, session_id, started)) = existing { + if stored_backend == backend && !session_id.trim().is_empty() { + tx.commit() + .with_context(|| format!("commit session read for {thread:?}"))?; + return Ok((session_id, !started)); } } - self.state.sessions.insert( - thread.to_string(), - SessionInfo { - uuid: initial_id.clone(), - started: false, - backend: backend.to_string(), - }, - ); - self.save()?; + + #[cfg(test)] + if self.session_save_failures_remaining > 0 { + self.session_save_failures_remaining -= 1; + return Err(anyhow::anyhow!("injected session save failure")); + } + tx.execute( + "INSERT INTO backend_sessions ( + channel, thread_key, backend, session_id, started + ) VALUES (?1, ?2, ?3, ?4, 0) + ON CONFLICT(channel, thread_key) DO UPDATE SET + backend = excluded.backend, + session_id = excluded.session_id, + started = 0, + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + params![channel, thread_key, backend, initial_id], + ) + .with_context(|| format!("store fresh session for {thread:?}"))?; + tx.commit() + .with_context(|| format!("commit fresh session for {thread:?}"))?; Ok((initial_id, true)) } pub fn mark_started(&mut self, thread: &str, session_id: Option<&str>) -> Result<()> { - if let Some(si) = self.state.sessions.get_mut(thread) { - let session_id = session_id.and_then(non_empty_session_id); - if let Some(id) = session_id { - si.uuid = id.to_string(); - } - if !si.started { - if si.uuid.trim().is_empty() { - return Ok(()); - } - si.started = true; - return self.save(); - } - if session_id.is_some() { - return self.save(); - } + let (channel, thread_key) = split_thread(thread)?; + let session_id = session_id.and_then(non_empty_session_id); + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .with_context(|| { + format!( + "begin session activation transaction in {}", + self.database_path.display() + ) + })?; + let current = tx + .query_row( + "SELECT session_id, started + FROM backend_sessions + WHERE channel = ?1 AND thread_key = ?2", + params![channel, thread_key], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, bool>(1)?)), + ) + .optional() + .with_context(|| format!("read session for {thread:?}"))?; + let Some((current_id, started)) = current else { + tx.commit() + .with_context(|| format!("commit missing session read for {thread:?}"))?; + return Ok(()); + }; + let effective_id = session_id.unwrap_or(¤t_id); + let should_start = !started && !effective_id.trim().is_empty(); + if session_id.is_none() && !should_start { + tx.commit() + .with_context(|| format!("commit unchanged session for {thread:?}"))?; + return Ok(()); + } + + #[cfg(test)] + if self.session_save_failures_remaining > 0 { + self.session_save_failures_remaining -= 1; + return Err(anyhow::anyhow!("injected session save failure")); } + tx.execute( + "UPDATE backend_sessions + SET session_id = ?3, + started = CASE WHEN ?4 THEN 1 ELSE started END, + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE channel = ?1 AND thread_key = ?2", + params![channel, thread_key, effective_id, should_start], + ) + .with_context(|| format!("mark session started for {thread:?}"))?; + tx.commit() + .with_context(|| format!("commit session activation for {thread:?}"))?; Ok(()) } /// Assigns a fresh backend session to a thread (the `/clear` behavior). pub fn rotate(&mut self, thread: &str, backend: &str, initial_id: String) -> Result<()> { - self.state.sessions.insert( - thread.to_string(), - SessionInfo { - uuid: initial_id, - started: false, - backend: backend.to_string(), - }, - ); - self.save() + validate_backend(backend)?; + let (channel, thread_key) = split_thread(thread)?; + self.fail_session_write_for_test()?; + self.conn + .execute( + "INSERT INTO backend_sessions ( + channel, thread_key, backend, session_id, started + ) VALUES (?1, ?2, ?3, ?4, 0) + ON CONFLICT(channel, thread_key) DO UPDATE SET + backend = excluded.backend, + session_id = excluded.session_id, + started = 0, + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')", + params![channel, thread_key, backend, initial_id], + ) + .with_context(|| format!("rotate session for {thread:?}"))?; + Ok(()) } - fn migrate_legacy_imessage_session(&mut self, thread: &str) -> Result<()> { - let Some(legacy) = thread.strip_prefix("imessage:") else { + fn migrate_legacy_state(&mut self, _fail_before_commit: bool) -> Result<()> { + let path = self.legacy_state_path.clone(); + let source_path = path.to_string_lossy().to_string(); + if self + .conn + .query_row( + "SELECT 1 FROM legacy_state_migrations WHERE source_path = ?1", + [&source_path], + |_| Ok(()), + ) + .optional() + .with_context(|| format!("check legacy state migration for {}", path.display()))? + .is_some() + { return Ok(()); + } + + let data = match std::fs::read(&path) { + Ok(data) => data, + Err(error) if error.kind() == ErrorKind::NotFound => return Ok(()), + Err(error) => { + return Err(error).with_context(|| format!("read legacy state {}", path.display())); + } }; - if self.state.sessions.contains_key(thread) { + crate::util::restrict_permissions(&path, false) + .with_context(|| format!("restrict legacy state permissions {}", path.display()))?; + let legacy: LegacyState = serde_json::from_slice(&data).with_context(|| { + format!( + "parse legacy state {}. Restore valid JSON from backup, or move the file aside only if you intend to start without its cursors and sessions", + path.display() + ) + })?; + let source_sha256 = format!("{:x}", Sha256::digest(&data)); + + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .with_context(|| { + format!( + "begin legacy state migration into {}", + self.database_path.display() + ) + })?; + if tx + .query_row( + "SELECT 1 FROM legacy_state_migrations WHERE source_path = ?1", + [&source_path], + |_| Ok(()), + ) + .optional()? + .is_some() + { + tx.commit()?; return Ok(()); } - if let Some(session) = self.state.sessions.remove(legacy) { - self.state.sessions.insert(thread.to_string(), session); - self.save()?; + + for (channel, cursor) in &legacy.cursors { + insert_monotonic_cursor(&tx, channel, *cursor)?; + } + if !legacy.cursors.contains_key("imessage") && legacy.last_row_id != 0 { + insert_monotonic_cursor(&tx, "imessage", legacy.last_row_id)?; } - Ok(()) - } - fn save(&self) -> Result<()> { - if let Some(dir) = self.path.parent() { - std::fs::create_dir_all(dir).context("create state dir")?; + // Qualified entries win over legacy unqualified iMessage aliases, + // regardless of JSON map iteration order. + for (thread, session) in legacy + .sessions + .iter() + .filter(|(thread, _)| legacy_qualified_thread(thread).is_some()) + { + let (channel, thread_key) = legacy_qualified_thread(thread).expect("filtered"); + insert_migrated_session(&tx, channel, thread_key, session)?; } - let tmp = self - .path - .with_extension(format!("tmp-{}", uuid::Uuid::new_v4())); - let data = serde_json::to_string_pretty(&self.state)?; - let result = (|| -> Result<()> { - let mut options = OpenOptions::new(); - options.write(true).create_new(true); - #[cfg(unix)] - { - use std::os::unix::fs::OpenOptionsExt; - options.mode(0o600); - } - let mut file = options.open(&tmp).context("create temporary state")?; - crate::util::restrict_permissions(&tmp, false) - .context("restrict temporary state permissions")?; - file.write_all(data.as_bytes()).context("write state")?; - std::fs::rename(&tmp, &self.path).context("rename state")?; - Ok(()) - })(); - if result.is_err() { - let _ = std::fs::remove_file(&tmp); + for (thread, session) in legacy + .sessions + .iter() + .filter(|(thread, _)| legacy_qualified_thread(thread).is_none()) + { + insert_migrated_session(&tx, "imessage", thread, session)?; } - result - } - #[cfg(test)] - pub fn set_path_for_test(&mut self, path: PathBuf) { - self.path = path; + #[cfg(test)] + if _fail_before_commit { + bail!("injected migration interruption before commit"); + } + tx.execute( + "INSERT INTO legacy_state_migrations (source_path, source_sha256) + VALUES (?1, ?2)", + params![source_path, source_sha256], + ) + .with_context(|| format!("record migration of {}", path.display()))?; + tx.commit().with_context(|| { + format!( + "commit legacy state migration from {} into {}", + path.display(), + self.database_path.display() + ) + })?; + Ok(()) } #[cfg(test)] pub fn fail_next_cursor_save_for_test(&mut self) { self.cursor_save_failures_remaining += 1; } + + #[cfg(test)] + pub fn fail_next_session_save_for_test(&mut self) { + self.session_save_failures_remaining += 1; + } + + fn fail_session_write_for_test(&mut self) -> Result<()> { + #[cfg(test)] + if self.session_save_failures_remaining > 0 { + self.session_save_failures_remaining -= 1; + return Err(anyhow::anyhow!("injected session save failure")); + } + Ok(()) + } +} + +fn insert_monotonic_cursor( + tx: &rusqlite::Transaction<'_>, + channel: &str, + cursor: i64, +) -> Result<()> { + validate_channel(channel)?; + tx.execute( + "INSERT INTO channel_cursors (channel, cursor) + VALUES (?1, ?2) + ON CONFLICT(channel) DO UPDATE SET + cursor = excluded.cursor, + updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE excluded.cursor > channel_cursors.cursor", + params![channel, cursor], + )?; + Ok(()) +} + +fn insert_migrated_session( + tx: &rusqlite::Transaction<'_>, + channel: &str, + thread_key: &str, + session: &SessionInfo, +) -> Result<()> { + validate_channel(channel)?; + validate_thread_key(thread_key)?; + validate_backend(&session.backend)?; + tx.execute( + "INSERT INTO backend_sessions ( + channel, thread_key, backend, session_id, started + ) VALUES (?1, ?2, ?3, ?4, ?5) + ON CONFLICT(channel, thread_key) DO NOTHING", + params![ + channel, + thread_key, + session.backend, + session.uuid, + session.started + ], + )?; + Ok(()) +} + +fn split_thread(thread: &str) -> Result<(&str, &str)> { + let Some((channel, thread_key)) = thread.split_once(':') else { + bail!("backend session thread {thread:?} must be channel-qualified"); + }; + validate_channel(channel)?; + validate_thread_key(thread_key)?; + Ok((channel, thread_key)) +} + +fn legacy_qualified_thread(thread: &str) -> Option<(&str, &str)> { + let (channel, thread_key) = thread.split_once(':')?; + matches!(channel, "imessage" | "telegram" | "slack").then_some((channel, thread_key)) +} + +fn validate_channel(channel: &str) -> Result<()> { + if channel.trim().is_empty() { + bail!("channel cannot be empty"); + } + Ok(()) +} + +fn validate_thread_key(thread_key: &str) -> Result<()> { + if thread_key.trim().is_empty() { + bail!("backend session thread key cannot be empty"); + } + Ok(()) +} + +fn validate_backend(backend: &str) -> Result<()> { + if backend.trim().is_empty() { + bail!("backend cannot be empty"); + } + Ok(()) } fn default_backend() -> String { @@ -243,225 +511,336 @@ fn default_backend() -> String { #[cfg(test)] mod tests { use super::*; + use std::sync::{Arc, Barrier}; + use std::thread; use uuid::Uuid; - fn temp_state_path() -> String { - std::env::temp_dir() - .join(format!("push-store-test-{}.json", Uuid::new_v4())) - .to_string_lossy() - .to_string() + fn temp_paths() -> (String, String) { + let base = std::env::temp_dir().join(format!("push-store-test-{}", Uuid::new_v4())); + ( + base.with_extension("db").to_string_lossy().to_string(), + base.with_extension("json").to_string_lossy().to_string(), + ) + } + + fn open(database_path: &str, state_path: &str) -> Store { + Store::open_at(database_path, state_path).unwrap() + } + + fn cleanup(database_path: &str, state_path: &str) { + let _ = std::fs::remove_file(database_path); + let _ = std::fs::remove_file(format!("{database_path}-wal")); + let _ = std::fs::remove_file(format!("{database_path}-shm")); + let _ = std::fs::remove_file(state_path); } #[test] fn backend_change_starts_fresh_session() { - let path = temp_state_path(); - let mut store = Store::open(&path).unwrap(); + let (database_path, state_path) = temp_paths(); + let mut store = open(&database_path, &state_path); let first = store - .session_for("self:me", "claude", "claude-session".to_string()) + .session_for("imessage:self:me", "claude", "claude-session".to_string()) .unwrap(); assert_eq!(first, ("claude-session".to_string(), true)); - store.mark_started("self:me", None).unwrap(); + store.mark_started("imessage:self:me", None).unwrap(); let second = store - .session_for("self:me", "codex", String::new()) + .session_for("imessage:self:me", "codex", String::new()) .unwrap(); assert_eq!(second, (String::new(), true)); - - let _ = std::fs::remove_file(path); + cleanup(&database_path, &state_path); } #[test] fn mark_started_can_store_backend_owned_session_id() { - let path = temp_state_path(); - let mut store = Store::open(&path).unwrap(); - + let (database_path, state_path) = temp_paths(); + let mut store = open(&database_path, &state_path); store - .session_for("self:me", "codex", String::new()) + .session_for("telegram:dm:7", "codex", String::new()) .unwrap(); store - .mark_started("self:me", Some("codex-thread-id")) + .mark_started("telegram:dm:7", Some("codex-thread-id")) .unwrap(); let resumed = store - .session_for("self:me", "codex", String::new()) + .session_for("telegram:dm:7", "codex", String::new()) .unwrap(); assert_eq!(resumed, ("codex-thread-id".to_string(), false)); - - let _ = std::fs::remove_file(path); + cleanup(&database_path, &state_path); } #[test] - fn mark_started_ignores_empty_backend_session_id() { - let path = temp_state_path(); - let mut store = Store::open(&path).unwrap(); - - store - .session_for("self:me", "claude", "initial-session".to_string()) - .unwrap(); - store.mark_started("self:me", Some(" \t\n ")).unwrap(); + fn empty_session_ids_preserve_fresh_session_behavior() { + let (database_path, state_path) = temp_paths(); + std::fs::write( + &state_path, + r#"{"sessions":{"telegram:dm:7":{"uuid":"","started":true,"backend":"codex"}}}"#, + ) + .unwrap(); + let mut store = open(&database_path, &state_path); - let resumed = store - .session_for("self:me", "claude", "unused-session".to_string()) + let fresh = store + .session_for("telegram:dm:7", "codex", String::new()) .unwrap(); - assert_eq!(resumed, ("initial-session".to_string(), false)); - - let _ = std::fs::remove_file(path); + assert_eq!(fresh, (String::new(), true)); + store.mark_started("telegram:dm:7", Some(" \n ")).unwrap(); + assert_eq!( + store + .session_for("telegram:dm:7", "codex", String::new()) + .unwrap(), + (String::new(), true) + ); + cleanup(&database_path, &state_path); } #[test] - fn existing_empty_backend_session_id_starts_fresh_session() { - let path = temp_state_path(); + fn migrates_legacy_cursors_channels_backends_and_imessage_keys() { + let (database_path, state_path) = temp_paths(); std::fs::write( - &path, + &state_path, r#"{ + "last_row_id": 42, + "cursors": {"telegram": 99}, "sessions": { - "self:me": { - "uuid": "", - "started": true, - "backend": "claude" + "dm:+15551234567": { + "uuid": "legacy-imessage", "started": true, "backend": "claude" + }, + "telegram:dm:7": { + "uuid": "telegram-session", "started": true, "backend": "codex" + }, + "slack:dm:workspace:channel": { + "uuid": "slack-session", "started": false, "backend": "pi" } } }"#, ) .unwrap(); - let mut store = Store::open(&path).unwrap(); + let mut store = open(&database_path, &state_path); - let fresh = store - .session_for("self:me", "claude", "fresh-session".to_string()) - .unwrap(); - assert_eq!(fresh, ("fresh-session".to_string(), true)); - - let _ = std::fs::remove_file(path); + assert_eq!(store.cursor("imessage").unwrap(), 42); + assert_eq!(store.cursor("telegram").unwrap(), 99); + assert_eq!( + store + .session_for("imessage:dm:+15551234567", "claude", "unused".into()) + .unwrap(), + ("legacy-imessage".into(), false) + ); + assert_eq!( + store + .session_for("telegram:dm:7", "codex", "unused".into()) + .unwrap(), + ("telegram-session".into(), false) + ); + assert_eq!( + store + .session_for("slack:dm:workspace:channel", "pi", "unused".into()) + .unwrap(), + ("slack-session".into(), true) + ); + assert!(std::path::Path::new(&state_path).exists()); + cleanup(&database_path, &state_path); } #[test] - fn existing_empty_unstarted_backend_session_id_starts_fresh_session() { - let path = temp_state_path(); + fn explicit_imessage_cursor_wins_over_last_row_id() { + let (database_path, state_path) = temp_paths(); std::fs::write( - &path, - r#"{ - "sessions": { - "self:me": { - "uuid": "", - "started": false, - "backend": "claude" - } - } -}"#, + &state_path, + r#"{"last_row_id":42,"cursors":{"imessage":12}}"#, ) .unwrap(); - let mut store = Store::open(&path).unwrap(); - - let fresh = store - .session_for("self:me", "claude", "fresh-session".to_string()) - .unwrap(); - assert_eq!(fresh, ("fresh-session".to_string(), true)); - - let _ = std::fs::remove_file(path); + let store = open(&database_path, &state_path); + assert_eq!(store.cursor("imessage").unwrap(), 12); + cleanup(&database_path, &state_path); } #[test] - fn mark_started_does_not_activate_empty_backend_session_id() { - let path = temp_state_path(); - let mut store = Store::open(&path).unwrap(); + fn qualified_imessage_session_wins_over_legacy_alias() { + let (database_path, state_path) = temp_paths(); + std::fs::write( + &state_path, + r#"{"sessions":{ + "dm:+15551234567":{"uuid":"legacy","started":true,"backend":"claude"}, + "imessage:dm:+15551234567":{"uuid":"qualified","started":true,"backend":"codex"} +}}"#, + ) + .unwrap(); + let mut store = open(&database_path, &state_path); + assert_eq!( + store + .session_for("imessage:dm:+15551234567", "codex", "unused".into()) + .unwrap(), + ("qualified".into(), false) + ); + cleanup(&database_path, &state_path); + } + #[test] + fn migration_is_repeatable_without_regressing_new_state() { + let (database_path, state_path) = temp_paths(); + std::fs::write( + &state_path, + r#"{"cursors":{"telegram":10},"sessions":{ + "telegram:dm:7":{"uuid":"legacy","started":true,"backend":"codex"} +}}"#, + ) + .unwrap(); + let mut store = open(&database_path, &state_path); + store.set_cursor("telegram", 20).unwrap(); store - .session_for("self:me", "codex", String::new()) - .unwrap(); - store.mark_started("self:me", Some("")).unwrap(); - - let fresh = store - .session_for("self:me", "codex", String::new()) + .rotate("telegram:dm:7", "claude", "new".into()) .unwrap(); - assert_eq!(fresh, (String::new(), true)); + drop(store); - let _ = std::fs::remove_file(path); + let mut reopened = open(&database_path, &state_path); + assert_eq!(reopened.cursor("telegram").unwrap(), 20); + assert_eq!( + reopened + .session_for("telegram:dm:7", "claude", "unused".into()) + .unwrap(), + ("new".into(), true) + ); + cleanup(&database_path, &state_path); } #[test] - fn legacy_last_row_id_remains_the_imessage_cursor() { - let path = temp_state_path(); - std::fs::write(&path, r#"{"last_row_id":42,"sessions":{}}"#).unwrap(); - let store = Store::open(&path).unwrap(); - - assert!(store.has_cursor("imessage")); - assert_eq!(store.cursor("imessage"), 42); - assert_eq!(store.cursor("telegram"), 0); + fn interrupted_migration_rolls_back_and_retries_cleanly() { + let (database_path, state_path) = temp_paths(); + std::fs::write( + &state_path, + r#"{"cursors":{"telegram":10},"sessions":{ + "telegram:dm:7":{"uuid":"legacy","started":true,"backend":"codex"} +}}"#, + ) + .unwrap(); + let error = Store::open_inner(Path::new(&database_path), Path::new(&state_path), true) + .err() + .unwrap(); + assert!(format!("{error:#}").contains("injected migration interruption")); + assert!(std::path::Path::new(&state_path).exists()); - let _ = std::fs::remove_file(path); + let mut store = open(&database_path, &state_path); + assert_eq!(store.cursor("telegram").unwrap(), 10); + assert_eq!( + store + .session_for("telegram:dm:7", "codex", "unused".into()) + .unwrap(), + ("legacy".into(), false) + ); + cleanup(&database_path, &state_path); } #[test] - fn channel_cursors_persist_independently() { - let path = temp_state_path(); - let mut store = Store::open(&path).unwrap(); - - store.set_cursor("imessage", 12).unwrap(); - store.set_cursor("telegram", 99).unwrap(); - store.set_cursor("imessage", 13).unwrap(); - drop(store); - - let reopened = Store::open(&path).unwrap(); - assert_eq!(reopened.cursor("imessage"), 13); - assert_eq!(reopened.cursor("telegram"), 99); - assert!(std::fs::read_to_string(&path) - .unwrap() - .contains("\"last_row_id\": 13")); + fn corrupt_json_fails_with_recovery_guidance() { + let (database_path, state_path) = temp_paths(); + std::fs::write(&state_path, "{broken").unwrap(); + let error = Store::open_at(&database_path, &state_path).err().unwrap(); + let message = format!("{error:#}"); + assert!(message.contains("parse legacy state")); + assert!(message.contains("Restore valid JSON from backup")); + assert!(std::path::Path::new(&state_path).exists()); + cleanup(&database_path, &state_path); + } - let _ = std::fs::remove_file(path); + #[test] + fn database_open_failure_is_actionable_before_state_is_available() { + let (database_path, state_path) = temp_paths(); + let blocker = PathBuf::from(&database_path).with_extension("blocker"); + std::fs::write(&blocker, "not a directory").unwrap(); + let blocked_database = blocker.join("push.db"); + + let error = Store::open_at(&blocked_database, &state_path) + .err() + .unwrap(); + let message = format!("{error:#}"); + assert!(message.contains("prepare canonical state database")); + assert!( + message.contains("create database directory") + || message.contains("open conversation database") + ); + assert!(!std::path::Path::new(&state_path).exists()); + let _ = std::fs::remove_file(blocker); } #[test] - fn legacy_imessage_session_moves_to_channel_qualified_key() { - let path = temp_state_path(); - std::fs::write( - &path, - r#"{ - "sessions": { - "dm:+15551234567": { - "uuid": "existing-session", - "started": true, - "backend": "claude" + fn concurrent_cursor_writes_never_move_backwards() { + let (database_path, state_path) = temp_paths(); + drop(open(&database_path, &state_path)); + let barrier = Arc::new(Barrier::new(8)); + let mut handles = Vec::new(); + for cursor in [80, 10, 70, 20, 60, 30, 50, 40] { + let database_path = database_path.clone(); + let state_path = state_path.clone(); + let barrier = barrier.clone(); + handles.push(thread::spawn(move || { + let mut store = open(&database_path, &state_path); + barrier.wait(); + store.set_cursor("telegram", cursor).unwrap(); + })); + } + for handle in handles { + handle.join().unwrap(); + } + assert_eq!( + open(&database_path, &state_path) + .cursor("telegram") + .unwrap(), + 80 + ); + cleanup(&database_path, &state_path); } - } -}"#, - ) - .unwrap(); - let mut store = Store::open(&path).unwrap(); - let session = store - .session_for("imessage:dm:+15551234567", "claude", "unused".to_string()) - .unwrap(); + #[test] + fn clear_rotates_only_the_exact_channel_qualified_session() { + let (database_path, state_path) = temp_paths(); + let mut store = open(&database_path, &state_path); + for thread in ["imessage:dm:7", "telegram:dm:7"] { + store + .session_for(thread, "codex", format!("{thread}-old")) + .unwrap(); + store.mark_started(thread, None).unwrap(); + } - assert_eq!(session, ("existing-session".to_string(), false)); - let raw = std::fs::read_to_string(&path).unwrap(); - assert!(raw.contains("imessage:dm:+15551234567")); - assert!(!raw.contains("\"dm:+15551234567\"")); - let _ = std::fs::remove_file(path); + store + .rotate("telegram:dm:7", "codex", "telegram-new".into()) + .unwrap(); + assert_eq!( + store + .session_for("telegram:dm:7", "codex", "unused".into()) + .unwrap(), + ("telegram-new".into(), true) + ); + assert_eq!( + store + .session_for("imessage:dm:7", "codex", "unused".into()) + .unwrap(), + ("imessage:dm:7-old".into(), false) + ); + cleanup(&database_path, &state_path); } #[cfg(unix)] #[test] - fn state_file_is_private_and_existing_permissions_are_repaired() { + fn database_and_legacy_backup_are_private() { use std::os::unix::fs::PermissionsExt; - let path = temp_state_path(); - let mut store = Store::open(&path).unwrap(); - store.set_cursor("imessage", 1).unwrap(); + let (database_path, state_path) = temp_paths(); + std::fs::write(&state_path, "{}").unwrap(); + std::fs::set_permissions(&state_path, std::fs::Permissions::from_mode(0o666)).unwrap(); + drop(open(&database_path, &state_path)); assert_eq!( - std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, + std::fs::metadata(&database_path) + .unwrap() + .permissions() + .mode() + & 0o777, 0o600 ); - - std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o666)).unwrap(); - drop(store); - Store::open(&path).unwrap(); assert_eq!( - std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, + std::fs::metadata(&state_path).unwrap().permissions().mode() & 0o777, 0o600 ); - - let _ = std::fs::remove_file(path); + cleanup(&database_path, &state_path); } }