Optimize project RPC paths and bound hot-path growth

OPTIMIZATION_TODOS.md

@@ -1,78 +1,104 @@
 # Optimization TODO list
 
-Full codebase review after the SQLite event-bus migration (`7915613`).
-Priorities: **P0** = correctness / unbounded growth; **P1** = hot-path performance;
-**P2** = medium impact; **P3** = polish / future.
+Full codebase review after project repository queries / RPC routing
+(`aaacba9`), plus unmerged hot-path fixes from earlier review rounds.
+Priorities: **P0** = correctness / unbounded growth; **P1** = hot-path
+performance; **P2** = medium impact; **P3** = polish / future.
 
 Stub crates (`retcon-worktrees`, `retcon-browser`, `retcon-checkpoints`,
 `retcon-filesystem`, `retcon-index`, `retcon-platform`, `retcon-permissions`,
-`retcon-diagnostics`, `retcon-updater`) and empty TypeScript packages have no
-meaningful optimization surface yet.
+`retcon-diagnostics`, `retcon-updater`) and empty TypeScript/Dart packages
+have no meaningful optimization surface yet.
 
 ---
 
 ## Done in this PR
 
-- [x] **P0** Cap browser-service console/network buffers with a ring buffer
-  (`apps/browser-service/src/browser.ts`)
-- [x] **P0** Remove timed-out / closed browser RPC entries from `pending`
-  (`crates/retcon-core/src/spikes/browser.rs`)
-- [x] **P0** Cap RPC frame size on read (`crates/retcon-core/src/frame.rs`,
-  `server.rs`, `spikes/browser.rs`)
-- [x] **P0** Periodic SQLite event retention (`crates/retcon-core/src/event.rs`)
-- [x] **P0** Evict finished jobs from memory + DB (`crates/retcon-core/src/jobs.rs`)
-- [x] **P0** Graceful browser-service shutdown before kill (`spikes/browser.rs`)
-- [x] **P0** Bound Git command output (`crates/retcon-git/src/lib.rs`)
-- [x] **P1** Don't hold event-bus mutex across SQLite insert (`event.rs` writer thread)
-- [x] **P1** Batch / coalesce terminal output events (`spikes/terminal.rs`)
-- [x] **P1** Move SQLite work off Tokio worker threads (`database.rs` `read_async`,
-  `artifacts.rs` async helpers)
-- [x] **P1** Add status indexes for recovery queries (migration `0003`)
-- [x] **P1** Enforce log pagination / tail in `browser.logs` (`browser.ts`)
-- [x] **P1** Default screenshot to viewport (`browser.ts`)
-- [x] **P1** Combine Git status into one process (`retcon-git/src/lib.rs`)
-- [x] **P2** Browser proc lock held across stdin writes — writer actor split (`browser.rs`)
-- [x] **P2** Clear stale `BrowserHandle.proc` + drain pending on reader exit (`browser.rs`)
-- [x] **P2** Artifact retention hash indexes (migration `0003`)
-- [x] **P2** Run artifact disk walks in `spawn_blocking` (`artifacts.rs`)
-- [x] **P2** Skip full rehash on duplicate artifact store (`artifacts.rs`)
-- [x] **P2** Binary-search event replay (`event.rs`)
-- [x] **P2** `Arc<EventEnvelope>` to cut clone cost on emit (`event.rs`)
-- [x] **P2** `prepare_cached` for hot statements (`event.rs`, `jobs.rs`)
-- [x] **P2** `PRAGMA quick_check` on normal open; full check for diagnostics (`database.rs`)
-- [x] **P2** `SQLITE_OPEN_NO_MUTEX` when Rust serializes the connection (`database.rs`)
-- [x] **P2** Tune `cache_size` / `mmap_size` / `wal_autocheckpoint` (`database.rs`)
-- [x] **P2** Debounce job persistence; avoid Debug-format enum names (`jobs.rs`)
-- [x] **P2** Don't hold terminal registry lock while killing (`spikes/terminal.rs`)
-- [x] **P2** Cache / async shell detection (`spikes/terminal.rs`)
-- [x] **P2** Cache agent/provider detection (`retcon-agents`)
-- [x] **P2** Await `child.wait()` instead of 250ms poll (`spikes/agent.rs`)
-- [x] **P2** Cache schema version after open (`state.rs`)
-- [x] **P2** Lazy-load event replay pages from DB at startup (`event.rs`)
-- [x] **P2** Pass browser temp profile dir via `launchPersistentContext` (`browser.ts`)
-- [x] **P2** Stdout backpressure on browser-service event emits (`main.ts`)
-- [x] **P2** Spawn `bun` directly (no `cmd /C` shell) on Windows (`spikes/browser.rs`)
-- [x] **P3** Deduplicate console/network stream vs buffer shipping (`browser.ts` — network
-  responses now land in the ring buffer)
-- [x] **P3** Method-aware concurrency for browser-service stdio RPC (`main.ts`)
-- [x] **P3** Cache log level at module load (`logging.ts`)
-- [x] **P3** Website asset generation: single Sharp pipeline + `Promise.all` for icons
-- [x] **P3** Website CLS: explicit image dimensions; defer Tally iframe (`index.astro`)
-- [x] **P3** CI-generate PWA icons referenced by `site.webmanifest` (`package.json` build)
+### From project RPC commit (`aaacba9`)
+
+- [x] **P0** Cap `project.clone` / project Git helpers via `retcon_git`
+  (no unbounded `.output()`) — `crates/retcon-core/src/projects.rs`
+- [x] **P1** Avoid full `git status --porcelain` in health; use
+  `rev-parse` + `diff --quiet` + capped `ls-files` —
+  `projects.rs`
+- [x] **P1** Run filesystem `analyze()` in `spawn_blocking` —
+  `projects.rs`
+- [x] **P1** Bound `project.list` (default 50 / max 200) + batch metadata
+  via `settings.list_by_key` (no N+1) — `projects.rs`, `repositories.rs`,
+  `projects_rpc.rs`
+- [x] **P1** Index active projects + settings-by-key (migration `0005`)
+- [x] **P3** Call `add_location` once on open (no duplicate upsert)
+
+### Carried from prior unmerged review work
+
+- [x] **P0** Cap RPC frame reads with chunked `fill_buf` (no unbounded
+  `read_until`) — `crates/retcon-core/src/frame.rs`
+- [x] **P0** Cap Git stderr; kill child on stdout/stderr overflow —
+  `crates/retcon-git/src/lib.rs`
+- [x] **P0** Bound event persist queue (`sync_channel` + `try_send`) —
+  `crates/retcon-core/src/event.rs`
+- [x] **P1** Drain completed RPC `JoinSet` tasks; cap concurrent
+  connections (64) — `crates/retcon-core/src/server.rs`
+- [x] **P1** Enforce outbound frame size (`MAX_FRAME_BYTES`) —
+  `server.rs`
 
 ---
 
-## Completed schema work
+## Still open
+
+### P1
+
+- [ ] **P1** Dedicated job-persistence writer actor (replace sync DB writes
+  on job update/finish) — `crates/retcon-core/src/jobs.rs`
+- [ ] **P1** Move event prune / lifecycle maintain / artifact cleanup off
+  async request & emit paths (`spawn_blocking` or writer-side prune) —
+  `event.rs`, `lifecycle.rs`
+- [ ] **P1** Browser reader cleanup generation ID (don't clear a newly
+  started process) — `crates/retcon-core/src/spikes/browser.rs`
+- [ ] **P1** Align Git 4MiB output cap with 1MiB protocol frame (page large
+  diffs as artifacts / lower per-method caps) — `retcon-git`, spikes/git
+- [ ] **P1** Run `project.clone` as a supervised background job with
+  progress events (RPC currently waits for the full clone)
+
+### P2
+
+- [ ] **P2** `jobs.list` SQL-backed ordered pagination (not in-memory
+  collect+sort) — `crates/retcon-core/src/jobs.rs`
+- [ ] **P2** Paginate `SessionRepository::list_for_project` —
+  `crates/retcon-storage/src/repositories.rs`
+- [ ] **P2** Artifact retention: stream/`EXISTS` instead of full hash set —
+  `crates/retcon-storage/src/artifacts.rs`
+- [ ] **P2** `CREATE INDEX … ON agent_events(created_at)` + age prune off
+  emit path — migrations + `event.rs`
+- [ ] **P2** Push `project.list` name/path filter into SQL (still filters
+  in Rust after the limited fetch)
+- [ ] **P2** Wrap `add_location` upsert + project `updated_at` in one
+  transaction — `repositories.rs`
+- [ ] **P2** Attach Playwright listeners to all pages (`context.on("page")`)
+  — `apps/browser-service/src/browser.ts`
+- [ ] **P2** Floating panel instance IDs + `ValueKey` for stateful bodies —
+  `apps/desktop/lib/src/workspace.dart`
+- [ ] **P2** Serialize/coalesce workspace layout persistence —
+  `apps/desktop/lib/src/workspace.dart`
+- [ ] **P2** Fixed-index ring buffer for browser logs (vs `splice`) —
+  `apps/browser-service/src/browser.ts`
+- [ ] **P2** BLOB-native fresh schema (skip TEXT→rebuild on new DBs)
+
+### P3
 
-- [x] **P3** UUID-as-BLOB schema migration (`retcon-storage` migration `0004`)
-  - Rebuilds v1-v3 databases transactionally, converts every UUID column to compact
-    16-byte BLOBs, recreates indexes, and updates active repository/core bindings.
+- [ ] **P3** Website asset generation mtime/hash short-circuit —
+  `apps/website/scripts/generate-brand-assets.mjs`
+- [ ] **P3** Guard redundant focus `setState` in design-system controls —
+  `packages/retcon-design-system/lib/src/components/controls.dart`
+- [ ] **P3** CI Playwright cache + path filters
 
 ---
 
-## Suggested order of attack (completed)
+## Suggested order of attack
 
-1. Frame-size caps + event retention + job eviction (bounds) ✅
-2. Event writer off async path + terminal coalesce (throughput) ✅
-3. Recovery indexes + Git/browser evidence polish (latency) ✅
-4. Storage pragma / statement-cache tuning (incremental) ✅
+1. Job writer actor + prune/maintain off async paths
+2. Browser process generation ID + Git/protocol size alignment
+3. `project.clone` as a supervised job
+4. Jobs/sessions pagination + event `created_at` index
+5. Desktop panel instance IDs + save coalescing
+6. Website/CI polish

crates/retcon-core/src/event.rs

@@ -4,7 +4,7 @@
 
 use std::collections::VecDeque;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::mpsc::{self, Sender};
+use std::sync::mpsc::{self, SyncSender, TrySendError};
 use std::sync::{Arc, Mutex};
 use std::thread::{self, JoinHandle};
 
@@ -20,6 +20,7 @@ use retcon_storage::Database;
 const DEFAULT_MAX_EVENTS: usize = 10_000;
 const DEFAULT_MAX_AGE_DAYS: i64 = 7;
 const PRUNE_EVERY_N_INSERTS: u64 = 100;
+const PERSIST_QUEUE_CAPACITY: usize = 2_048;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -77,7 +78,7 @@ struct EventBusInner {
     log: Mutex<EventLog>,
     sender: broadcast::Sender<Arc<EventEnvelope>>,
     database: Database,
-    persist_tx: Sender<PersistJob>,
+    persist_tx: SyncSender<PersistJob>,
     inserts_since_prune: AtomicU64,
     _writer: JoinHandle<()>,
 }
@@ -100,7 +101,7 @@ impl EventBus {
             )
         })?;
         let (sender, _) = broadcast::channel(1024);
-        let (persist_tx, persist_rx) = mpsc::channel();
+        let (persist_tx, persist_rx) = mpsc::sync_channel(PERSIST_QUEUE_CAPACITY);
         let writer_db = database.clone();
         let writer = thread::Builder::new()
             .name("event-writer".into())
@@ -173,21 +174,31 @@ impl EventBus {
         };
 
         if durable {
-            self.inner
-                .persist_tx
-                .send(PersistJob {
-                    event: Arc::clone(&event),
-                    encoded_payload: encoded,
-                })
-                .map_err(|error| internal_io(error.to_string()))?;
-            let count = self
-                .inner
-                .inserts_since_prune
-                .fetch_add(1, Ordering::Relaxed)
-                .saturating_add(1);
-            if count.is_multiple_of(PRUNE_EVERY_N_INSERTS) {
-                let _ = prune_database(&self.inner.database);
-                self.inner.inserts_since_prune.store(0, Ordering::Relaxed);
+            match self.inner.persist_tx.try_send(PersistJob {
+                event: Arc::clone(&event),
+                encoded_payload: encoded,
+            }) {
+                Ok(()) => {
+                    let count = self
+                        .inner
+                        .inserts_since_prune
+                        .fetch_add(1, Ordering::Relaxed)
+                        .saturating_add(1);
+                    if count.is_multiple_of(PRUNE_EVERY_N_INSERTS) {
+                        let _ = prune_database(&self.inner.database);
+                        self.inner.inserts_since_prune.store(0, Ordering::Relaxed);
+                    }
+                }
+                Err(TrySendError::Full(_)) => {
+                    tracing::warn!(
+                        sequence = event.sequence,
+                        kind = %event.kind,
+                        "event persist queue full; dropping durable write"
+                    );
+                }
+                Err(TrySendError::Disconnected(_)) => {
+                    return Err(internal_io("event writer disconnected".into()));
+                }
             }
         }
 

crates/retcon-core/src/frame.rs

@@ -7,6 +7,9 @@ use retcon_protocol::MAX_FRAME_BYTES;
 
 /// Read one newline-delimited frame, rejecting payloads larger than [`MAX_FRAME_BYTES`].
 ///
+/// Reads in bounded chunks via `fill_buf` so a missing newline cannot force unbounded
+/// allocation before the size check runs.
+///
 /// # Errors
 ///
 /// Returns [`CoreError`] when I/O fails or the frame exceeds the protocol limit.
@@ -15,30 +18,45 @@ pub async fn read_capped_line(
 ) -> Result<Option<String>, CoreError> {
     let mut buffer = Vec::new();
     loop {
-        let read = reader
-            .read_until(b'\n', &mut buffer)
+        let filled = reader
+            .fill_buf()
             .await
             .map_err(|error| CoreError::io("read RPC frame", error))?;
-        if read == 0 {
+        if filled.is_empty() {
             if buffer.is_empty() {
                 return Ok(None);
             }
             return trimmed_line(buffer);
         }
-        if buffer.len() > MAX_FRAME_BYTES + 1 {
-            return Err(CoreError::new(
-                ErrorCode::InvalidRequest,
-                ErrorSource::Rpc,
-                "Retcon received an oversized local request.",
-                format!("frame exceeds {MAX_FRAME_BYTES} bytes"),
-            ));
-        }
-        if buffer.ends_with(b"\n") {
+
+        if let Some(newline_at) = filled.iter().position(|&byte| byte == b'\n') {
+            let take = newline_at + 1;
+            if buffer.len().saturating_add(take) > MAX_FRAME_BYTES + 1 {
+                return oversized();
+            }
+            buffer.extend_from_slice(&filled[..take]);
+            reader.consume(take);
             return trimmed_line(buffer);
         }
+
+        let take = filled.len();
+        if buffer.len().saturating_add(take) > MAX_FRAME_BYTES + 1 {
+            return oversized();
+        }
+        buffer.extend_from_slice(filled);
+        reader.consume(take);
     }
 }
 
+fn oversized() -> Result<Option<String>, CoreError> {
+    Err(CoreError::new(
+        ErrorCode::InvalidRequest,
+        ErrorSource::Rpc,
+        "Retcon received an oversized local request.",
+        format!("frame exceeds {MAX_FRAME_BYTES} bytes"),
+    ))
+}
+
 fn trimmed_line(mut buffer: Vec<u8>) -> Result<Option<String>, CoreError> {
     if buffer.ends_with(b"\n") {
         buffer.pop();
@@ -58,3 +76,29 @@ fn trimmed_line(mut buffer: Vec<u8>) -> Result<Option<String>, CoreError> {
         )
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::BufReader;
+
+    #[tokio::test]
+    async fn rejects_oversize_frame_without_newline() {
+        let oversized = vec![b'a'; MAX_FRAME_BYTES + 2];
+        let mut reader = BufReader::new(oversized.as_slice());
+        let error = read_capped_line(&mut reader)
+            .await
+            .expect_err("must reject");
+        assert!(error.technical_message.contains("frame exceeds"));
+    }
+
+    #[tokio::test]
+    async fn reads_normal_frame() {
+        let mut reader = BufReader::new(b"{\"ok\":true}\n".as_slice());
+        let line = read_capped_line(&mut reader)
+            .await
+            .expect("ok")
+            .expect("frame");
+        assert_eq!(line, "{\"ok\":true}");
+    }
+}

crates/retcon-core/src/lifecycle.rs

@@ -273,7 +273,7 @@ mod tests {
             serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap();
         assert_eq!(health["result"]["status"], "healthy");
         assert_eq!(health["result"]["storage"]["status"], "healthy");
-        assert_eq!(health["result"]["storage"]["schema_version"], 4);
+        assert_eq!(health["result"]["storage"]["schema_version"], 5);
         assert_eq!(shutdown["result"]["accepted"], true);
 
         runtime.wait_for_shutdown_signal().await.unwrap();