diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index ac5de9d4..904eb9a2 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -71,9 +71,8 @@ jobs:
 
       # gpui_linux (wayland + x11), font-kit's fontconfig, reqwest's default
       # native-tls, and rodio's ALSA backend (`preview::playback`) are the only
-      # things that need system libraries.
-      # mold replaces the single-threaded default linker; macOS already ships a parallel one and
-      # Windows-MSVC cannot select rust-lld on stable.
+      # things that need system libraries. mold replaces the single-threaded
+      # default linker; macOS already ships a parallel one.
       - name: Install Linux system libraries
         if: runner.os == 'Linux'
         run: |
@@ -91,6 +90,14 @@ jobs:
       - name: Install cargo-nextest
         uses: taiki-e/install-action@nextest
 
+      # rust-lld ships with Rust and accepts the MSVC linker interface, avoiding the slower link.exe
+      # without another installation step. This runs before rust-cache so its key includes the flag.
+      - name: Configure Windows linker
+        if: runner.os == 'Windows'
+        shell: pwsh
+        run: |
+          'RUSTFLAGS=-C linker=rust-lld' >> $env:GITHUB_ENV
+
       - name: Cache cargo build
         uses: Swatinem/rust-cache@v2
 
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 578b86e0..15dbc1fc 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -146,6 +146,10 @@ jobs:
   build-windows:
     needs: version
     runs-on: windows-latest
+    # rust-lld ships with Rust and accepts the MSVC linker interface, avoiding the slower link.exe
+    # without another installation step.
+    env:
+      RUSTFLAGS: -C linker=rust-lld
     steps:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@stable
diff --git a/AGENTS.md b/AGENTS.md
index 6cff7f7e..c1b90a36 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -1,152 +1,31 @@
 # Agent instructions
 
-For any agent runtime, not just one vendor. Two unrelated jobs live here — pick the one you are
-actually doing.
+- **Changing qrate code?** Read [CLAUDE.md](CLAUDE.md).
+- **Reviewing the project open in qrate?** Use bridge protocol 2 below. The saved `.qrate` file can omit unsaved on-screen edits.
 
-- **Changing qrate's code?** Read [CLAUDE.md](CLAUDE.md). It is the contributor guide: crate map,
-  build and test commands, code style, git workflow. It is not Claude-specific despite the name,
-  and nothing in it is repeated below.
-- **Reading the metadata in a running qrate?** That is the rest of this file.
+## Live qrate bridge
 
-## Reading a running qrate
+qrate publishes `agent-bridge.json` in the platform application-data directory. It contains `url`, a per-launch `token`, and `bridge_protocol: 2`. If it is missing, qrate is not running or Settings ▸ Agent is off. Reject any other protocol version.
 
-qrate can hand you the metadata it currently holds in memory, including cell edits the archivist
-has not saved yet. This is the only route to that state — the `.qrate` file on disk is the last
-save, not what is on screen.
+POST one JSON request to `url` with `Authorization: Bearer <token>`, `Content-Type: application/json`, and `X-Agent: codex`. The agent name is a visible label, not authority. Re-read the endpoint after `forbidden` because the token changes each launch.
 
-### The bridge cannot change a cell, and that is the point
+The bridge never changes a cell. `stage_findings` only replaces that agent's draft findings in the Problems panel and optionally offers whole-cell replacements in the Fixes menu.
 
-Nothing in the protocol writes data. You cannot fix a row — you can only stage what you would
-change and let the archivist decide, one click at a time, in their own app. Never tell the user you
-have corrected something.
+qrate's bundled Pi runs as a contained child process. Stop, Restart, panel teardown, and app exit terminate its process tree; do not rely on an older Pi process surviving a restart.
 
-The archive is somebody's real collection. When you report a problem, quote the row index and the
-column name so they can find it, and say what the data says rather than what you assume it means —
-a date that looks wrong may be exactly what the source object is stamped with.
+### Protocol 2 methods
 
-### Connecting
+- `overview`: compact project, column, selection, diagnostic-count, and revision information.
+- `query`: bounded live rows or diagnostics. Sources are `all_rows`, `selected_rows`, `rows`, `search`, and filtered `diagnostics`. Operations include `select`, `where`, `distinct`, `group_by`, `order_by`, `limit`, and revision-bound `cursor`.
+- `program_save`: validate and activate a confined Luau function without running it.
+- `program_run`: run the saved function once against an immutable snapshot at an exact revision.
+- `thumbnails`: return at most four qrate-generated 512-pixel PNG derivatives by source row/page.
+- `stage_findings`: publish a complete advisory draft batch.
 
-qrate listens by default, but the user can switch it off under Settings ▸ Agent. If the endpoint
-file is missing, qrate is either not running or the bridge is off — say which you think it is and
-ask the user to check, rather than guessing at data.
+Queries default to 20 and allow at most 50 records. Collection results send one `fields` array and positional `items`, plus `returned`, `remaining`, `truncated`, and `next_cursor`. Ask for the smallest useful field set and follow a cursor only when more evidence could change the answer.
 
-```bash
-# Windows
-ENDPOINT="$LOCALAPPDATA/qrate/agent-bridge.json"
-# macOS: ~/Library/Application Support/qrate/agent-bridge.json
-# Linux: ~/.local/share/qrate/agent-bridge.json
+Programs have no network, filesystem, process, clock, randomness, plugin storage, UI, raw qrate objects, or staging access. Their only linked-file methods are qrate-resolved bounded UTF-8 reads and PDF text search. Original paths and bytes are never returned.
 
-URL=$(sed -n 's/.*"url":"\([^"]*\)".*/\1/p' "$ENDPOINT")
-TOKEN=$(sed -n 's/.*"token":"\([^"]*\)".*/\1/p' "$ENDPOINT")
+Every staged finding needs the exact response `revision`, source `row`, `column`, severity, one-sentence message, and exact current whole-cell `expected`; `replacement` is optional and must be the whole proposed value. A stale cell is rejected. Never report that staging corrected data.
 
-AGENT=your-runtime-name   # see below — substitute yours, do not send this literally
-
-ask() {
-  curl -s -X POST "$URL" -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
-    -H "X-Agent: $AGENT" -d "$1"
-}
-```
-
-Send `X-Agent` on every call. It is what the archivist sees in qrate's Agent panel beside each of
-your calls, and without it you appear as `unnamed agent`.
-
-Name the runtime, not the task: `claude-code`, `pi`, `codex`. Add a suffix only to tell concurrent
-sessions of the same runtime apart — `claude-code/review-2` — because the panel groups by this
-string, so two sessions sharing a name read as one agent reconnecting. If your runtime came with a
-qrate skill or plugin, that is where its name is written down; this file is vendor-neutral and
-cannot name it for you.
-
-It is a label, not a credential: qrate cannot check it, so never treat it as authorisation for
-anything, and never send a name that is not yours.
-
-`sed` rather than a JSON parser on purpose — the endpoint file is two flat fields, and a machine
-running qrate is not guaranteed to have `python` or `jq`.
-
-The endpoint also carries `"bridge_protocol": 1`. Callers should reject a different protocol
-version rather than guessing at request semantics. The token is regenerated every launch. Re-read
-the file if a call starts returning `forbidden`.
-
-If your runtime cannot run a shell, any HTTP client will do: POST the request JSON to `url` with the
-`Authorization: Bearer <token>` and `X-Agent` headers. The bridge ignores the method and path, so it
-is only HTTP to the extent that an ordinary client can talk to it.
-
-### The requests
-
-One JSON object per call: `{"method": ..., "params": ...}`. `params` is omitted for the methods
-that take none.
-
-| Call | Returns |
-|---|---|
-| `{"method":"project_summary"}` | project name, row and column counts, whether a files folder is set |
-| `{"method":"columns"}` | every column with its configured data type and notes |
-| `{"method":"rows","params":{"rows":[0,1,2]}}` | those rows as column/value pairs — max 100 per call |
-| `{"method":"search_rows","params":{"query":"1974","limit":20}}` | rows whose column name or value contains the query, case-insensitive |
-| `{"method":"diagnostics"}` | what qrate's own validators already flagged, with severity and source |
-| `{"method":"selected_rows"}` | source-row indices the archivist has selected right now |
-| `{"method":"stage_findings","params":{…}}` | publishes your findings as drafts — see below |
-
-Every response carries a `revision`. If it changes between calls, the archivist edited something
-mid-review — re-read anything you are about to quote.
-
-Row indices are **source** rows and ignore any active filter, so an index from `selected_rows` or
-`diagnostics` can be passed straight to `rows`.
-
-### Handing findings back
-
-`stage_findings` is how a review reaches the archivist. Each finding lands in qrate's Problems panel
-beside its own validators' output, and a finding that carries a `replacement` also puts that value
-in the cell's right-click **Fixes** menu — which the archivist has to open and click. Staging writes
-nothing.
-
-```json
-{"method":"stage_findings","params":{
-  "revision": 12,
-  "findings": [{
-    "row": 41,
-    "column": "Date",
-    "severity": "warning",
-    "message": "the title says 1962 but this column reads 1926 — likely transposed digits",
-    "expected": "1926",
-    "replacement": "1962"
-  }]
-}}
-```
-
-- `revision` is the one from the reads this batch is based on. Echo it; do not invent it.
-- `expected` is the cell text you judged, copied exactly from what `rows` gave you. qrate drops any
-  finding whose cell no longer says that, and withholds the fix later if the cell changes before the
-  archivist opens the menu. This is the guard that stops you proposing an edit to text nobody read —
-  getting `expected` wrong silently loses the finding.
-- `replacement` is the cell's **whole** new text, not a fragment. Omit it for an observation with no
-  fix to offer; most findings should omit it.
-- `severity`: `error`, `warning`, or `note`. Use `note` freely and `error` only for something
-  demonstrably wrong.
-
-The reply is `{"accepted": n, "stale": [indices]}`, where `stale` indexes into the batch you sent.
-Re-read those rows and re-stage them if they still matter.
-
-Each batch **replaces** the last one you staged. Send everything you stand by in one call, and send
-an empty `findings` list to retract. Max 200 findings per batch, 512 characters per message —
-the panel shows one line per finding, so write one sentence.
-
-### Working order
-
-1. `project_summary` and `columns` first — the column names and notes tell you what the collection
-   is and what each field is supposed to hold. Review nothing before reading them.
-2. `diagnostics` next. qrate already found the malformed dates and misspellings; repeating them adds
-   nothing. Your value is what a validator cannot see — a title that contradicts its own date field,
-   a creator spelled two ways across rows, a description that belongs to a different object.
-3. Then read rows. `selected_rows` when the ask is about "these", `search_rows` when it is about a
-   term, explicit indices when the diagnostics pointed at them.
-4. Report by row index and column name. Separate "this is wrong" from "this looks inconsistent".
-5. `stage_findings` last, once, with everything you stand by — so the archivist sees the review in
-   the app rather than only in your transcript. Tell them it is staged and still theirs to accept.
-
-Errors come back as `{"error": ...}`: `project_unavailable` means no project is open,
-`table_unavailable` means no table is loaded yet, and the rest are bad requests — check the row
-count and query limits above before retrying.
-
-The contract's types are `crates/ai/src/agent.rs`, the transport is `crates/app/src/agent_bridge.rs`,
-the live-table adapter is `crates/table/src/agent.rs`, and the panel that shows the archivist what
-you did is `crates/workspace/src/panels/agent.rs`. Change any of them and this file is part of the
-change.
+The contract is `crates/ai/src/agent.rs`, transport is `crates/app/src/agent_bridge.rs`, live adapter is `crates/table/src/agent.rs`, private runner is `crates/plugin-host/src/agent_program.rs`, and the audit panel is `crates/workspace/src/panels/agent.rs`. Keep this file synchronized when those surfaces change.
diff --git a/Cargo.lock b/Cargo.lock
index c389d1a1..89b8db20 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -121,12 +121,13 @@ dependencies = [
 
 [[package]]
 name = "agent-runtime"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "alacritty_terminal",
  "gpui",
  "log",
  "settings",
+ "windows 0.62.2",
 ]
 
 [[package]]
@@ -153,7 +154,7 @@ dependencies = [
 
 [[package]]
 name = "ai"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -252,7 +253,7 @@ checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
 
 [[package]]
 name = "app"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "agent-runtime",
  "ai",
@@ -1018,7 +1019,7 @@ dependencies = [
 
 [[package]]
 name = "checks"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "diagnostics",
  "edtf",
@@ -1602,7 +1603,7 @@ checksum = "0c87e182de0887fd5361989c677c4e8f5000cd9491d6d563161a8f3a5519fc7f"
 
 [[package]]
 name = "data-exchange"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "base64",
  "calamine",
@@ -1712,7 +1713,7 @@ dependencies = [
 
 [[package]]
 name = "diagnostics"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "gpui",
  "gpui-component",
@@ -5493,7 +5494,7 @@ checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6"
 
 [[package]]
 name = "plugin-api"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "gpui",
  "serde_json",
@@ -5502,13 +5503,14 @@ dependencies = [
 
 [[package]]
 name = "plugin-host"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "diagnostics",
  "gpui",
  "log",
  "mlua",
  "plugin-api",
+ "preview",
  "reqwest",
  "serde_json",
  "settings",
@@ -5655,7 +5657,7 @@ dependencies = [
 
 [[package]]
 name = "preview"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "anyhow",
  "block2 0.6.2",
@@ -5739,7 +5741,7 @@ dependencies = [
 
 [[package]]
 name = "project-wizard"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "anyhow",
  "csv",
@@ -6934,7 +6936,7 @@ dependencies = [
 
 [[package]]
 name = "settings"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "anyhow",
  "dirs",
@@ -7149,7 +7151,7 @@ dependencies = [
 
 [[package]]
 name = "spellcheck"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "diagnostics",
  "gpui",
@@ -7686,18 +7688,22 @@ dependencies = [
 
 [[package]]
 name = "table"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "ai",
+ "base64",
  "diagnostics",
  "gpui",
  "gpui-component",
  "log",
  "plugin-api",
+ "plugin-host",
+ "preview",
  "regex",
  "serde",
  "serde_json",
  "settings",
+ "sha2",
 ]
 
 [[package]]
@@ -9085,7 +9091,7 @@ checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
 
 [[package]]
 name = "window-wrapper"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "gpui",
  "gpui-component",
@@ -9819,7 +9825,7 @@ dependencies = [
 
 [[package]]
 name = "workspace"
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 dependencies = [
  "agent-runtime",
  "diagnostics",
diff --git a/Cargo.toml b/Cargo.toml
index 31e42662..f553329c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -20,7 +20,7 @@ members = [
 default-members = ["crates/app"]
 
 [workspace.package]
-version = "0.3.0-alpha.2"
+version = "0.4.0-alpha.1"
 edition = "2024"
 
 # Dev runtime was crawling on the 2k-row table because unoptimized gpui/text-shaping
diff --git a/crates/agent-runtime/Cargo.toml b/crates/agent-runtime/Cargo.toml
index 27c2377a..0f7be232 100644
--- a/crates/agent-runtime/Cargo.toml
+++ b/crates/agent-runtime/Cargo.toml
@@ -9,3 +9,5 @@ gpui.workspace = true
 log.workspace = true
 settings = { path = "../settings" }
 
+[target."cfg(windows)".dependencies]
+windows = { version = "0.62.2", features = ["Win32_Foundation", "Win32_Security", "Win32_System_JobObjects", "Win32_System_Threading"] }
diff --git a/crates/agent-runtime/src/lib.rs b/crates/agent-runtime/src/lib.rs
index 633dfdb7..83d774e0 100644
--- a/crates/agent-runtime/src/lib.rs
+++ b/crates/agent-runtime/src/lib.rs
@@ -4,4 +4,4 @@ mod runtime;
 mod terminal;
 
 pub use runtime::{AgentRuntime, init};
-pub use terminal::{AgentTerminal, TerminalBackground, TerminalPalette, TerminalScreen};
+pub use terminal::{AgentTerminal, TerminalPalette, TerminalScreen};
diff --git a/crates/agent-runtime/src/runtime.rs b/crates/agent-runtime/src/runtime.rs
index 45d4d225..fe8c6bd5 100644
--- a/crates/agent-runtime/src/runtime.rs
+++ b/crates/agent-runtime/src/runtime.rs
@@ -25,6 +25,7 @@ pub fn init(cx: &mut App) {
                 "embedded Pi {PI_VERSION} ready at {}",
                 runtime.program.display()
             );
+            crate::terminal::warm_global_credential(runtime.program.clone());
             cx.set_global(runtime);
         }
         Err(err) => log::warn!("embedded Pi is unavailable: {err}"),
diff --git a/crates/agent-runtime/src/terminal.rs b/crates/agent-runtime/src/terminal.rs
index e482b8a3..cc1d225d 100644
--- a/crates/agent-runtime/src/terminal.rs
+++ b/crates/agent-runtime/src/terminal.rs
@@ -2,11 +2,24 @@ use std::borrow::Cow;
 use std::collections::{HashMap, HashSet};
 use std::ops::Range;
 use std::process::Command;
-use std::sync::{Arc, mpsc};
+use std::sync::{Arc, OnceLock, mpsc};
 
 #[cfg(windows)]
 use std::os::windows::process::CommandExt as _;
 
+#[cfg(windows)]
+use windows::Win32::Foundation::{CloseHandle, HANDLE};
+#[cfg(windows)]
+use windows::Win32::System::JobObjects::{
+    AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
+    JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JobObjectExtendedLimitInformation,
+    SetInformationJobObject,
+};
+#[cfg(windows)]
+use windows::Win32::System::Threading::TerminateProcess;
+#[cfg(windows)]
+use windows::core::PCWSTR;
+
 use alacritty_terminal::event::{Event, EventListener, Notify as _, OnResize as _, WindowSize};
 use alacritty_terminal::event_loop::{EventLoop, Msg, Notifier};
 use alacritty_terminal::grid::{Dimensions, Scroll};
@@ -44,16 +57,21 @@ pub struct TerminalPalette {
 pub struct TerminalScreen {
     pub text: String,
     pub highlights: Vec<(Range<usize>, HighlightStyle)>,
-    pub backgrounds: Vec<TerminalBackground>,
     pub auth_urls: Vec<String>,
 }
 
-#[derive(Clone, Copy)]
-pub struct TerminalBackground {
-    pub line: usize,
-    pub start_column: usize,
-    pub end_column: usize,
-    pub color: Hsla,
+static GLOBAL_CREDENTIAL: OnceLock<Option<String>> = OnceLock::new();
+
+/// Resolve the fallback credential once, on a background thread.
+///
+/// Each probe spawns Pi, which costs a second or more cold; doing that inside `start` froze the
+/// window on every session start. Called from `init`, so it is settled long before a panel opens.
+// ponytail: a start that beats the probe loses the fallback for that run and says so in the log.
+// Make it a real async handoff if that ever bites; blocking the UI to close the race is worse.
+pub fn warm_global_credential(program: std::path::PathBuf) {
+    std::thread::spawn(move || {
+        let _ = GLOBAL_CREDENTIAL.set(global_openrouter_credential(&program));
+    });
 }
 
 /// Ask the user's ordinary Pi installation to resolve its OpenRouter credential. The credential
@@ -153,6 +171,7 @@ fn terminal_style(
     let foreground = if inverse { cell.bg } else { cell.fg };
     let mut style = HighlightStyle {
         color: Some(resolve_color(foreground, palette, dynamic)),
+        background_color: terminal_background(cell, palette, dynamic),
         ..Default::default()
     };
     if cell.flags.contains(Flags::BOLD) {
@@ -306,14 +325,97 @@ struct Session {
     notifier: Notifier,
     events: mpsc::Receiver<Event>,
     running: bool,
+    #[cfg(windows)]
+    process: WindowsProcess,
 }
 
 impl Drop for Session {
     fn drop(&mut self) {
+        #[cfg(windows)]
+        if self.running {
+            self.process.terminate();
+        }
         let _ = self.notifier.0.send(Msg::Shutdown);
     }
 }
 
+#[cfg(windows)]
+struct WindowsProcess {
+    process: HANDLE,
+    job: Option<HANDLE>,
+    pid: Option<u32>,
+}
+
+#[cfg(windows)]
+impl WindowsProcess {
+    fn attach(pty: &tty::Pty) -> Self {
+        let watcher = pty.child_watcher();
+        let process = HANDLE(watcher.raw_handle());
+        let pid = watcher.pid().map(std::num::NonZeroU32::get);
+        Self::attach_handle(process, pid)
+    }
+
+    fn attach_handle(process: HANDLE, pid: Option<u32>) -> Self {
+        let job = Self::job(process).inspect_err(|err| {
+            log::warn!(
+                "could not contain embedded Pi process {} in a Windows job object: {err}; stop will terminate only Pi itself",
+                pid.map_or_else(|| "unknown".to_owned(), |pid| pid.to_string())
+            );
+        }).ok();
+        Self { process, job, pid }
+    }
+
+    fn job(process: HANDLE) -> windows::core::Result<HANDLE> {
+        unsafe {
+            let job = CreateJobObjectW(None, PCWSTR::null())?;
+            let mut limits = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default();
+            limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
+            if let Err(err) = SetInformationJobObject(
+                job,
+                JobObjectExtendedLimitInformation,
+                std::ptr::from_ref(&limits).cast(),
+                std::mem::size_of_val(&limits) as u32,
+            )
+            .and_then(|()| AssignProcessToJobObject(job, process))
+            {
+                let _ = CloseHandle(job);
+                return Err(err);
+            }
+            Ok(job)
+        }
+    }
+
+    fn terminate(&mut self) {
+        let result = unsafe {
+            match self.job.take() {
+                Some(job) => CloseHandle(job),
+                None => TerminateProcess(self.process, 1),
+            }
+        };
+        match result {
+            Ok(()) => log::info!(
+                "terminated embedded Pi process {}",
+                self.pid
+                    .map_or_else(|| "unknown".to_owned(), |pid| pid.to_string())
+            ),
+            Err(err) => log::warn!(
+                "could not terminate embedded Pi process {}: {err}",
+                self.pid
+                    .map_or_else(|| "unknown".to_owned(), |pid| pid.to_string())
+            ),
+        }
+    }
+}
+
+#[cfg(windows)]
+impl Drop for WindowsProcess {
+    fn drop(&mut self) {
+        if let Some(job) = self.job.take() {
+            let _ = unsafe { CloseHandle(job) };
+        }
+    }
+}
+
 pub struct AgentTerminal {
     session: Option<Session>,
     status: String,
@@ -387,7 +489,9 @@ impl AgentTerminal {
             "--provider".to_owned(),
             "openrouter".to_owned(),
             "--model".to_owned(),
-            "openrouter/free".to_owned(),
+            // Pi removes the pinned provider prefix before resolving the model. Doubling it makes
+            // the remaining ID exactly `openrouter/free` instead of the fuzzy pattern `free`.
+            "openrouter/openrouter/free".to_owned(),
             "--no-extensions".to_owned(),
             "--extension".to_owned(),
             runtime.extension.to_string_lossy().into_owned(),
@@ -404,15 +508,21 @@ impl AgentTerminal {
         }
 
         let mut env: HashMap<String, String> = std::env::vars().collect();
-        if let Some(credential) = global_openrouter_credential(&runtime.program) {
-            // A qrate-local auth.json still takes priority inside Pi. This is only a fallback to the
-            // user's global Pi login, and avoids duplicating a refresh token on disk.
-            env.insert("OPENROUTER_API_KEY".to_owned(), credential);
-            log::info!("embedded Pi can use the global Pi OpenRouter login as an auth fallback");
-        } else {
-            log::info!(
+        match GLOBAL_CREDENTIAL.get() {
+            Some(Some(credential)) => {
+                // A qrate-local auth.json still takes priority inside Pi. This is only a fallback
+                // to the user's global Pi login, and avoids duplicating a refresh token on disk.
+                env.insert("OPENROUTER_API_KEY".to_owned(), credential.clone());
+                log::info!(
+                    "embedded Pi can use the global Pi OpenRouter login as an auth fallback"
+                );
+            }
+            Some(None) => log::info!(
                 "no global Pi OpenRouter login is available; embedded Pi will use its isolated login or OpenRouter's anonymous free router"
-            );
+            ),
+            None => log::warn!(
+                "the global Pi login was still being resolved when this session started; embedded Pi will use its isolated login for this run"
+            ),
         }
         // Keep Pi's provider login under its isolated profile, but preserve the ordinary process
         // environment for TLS, proxies, locale, and platform support.
@@ -465,6 +575,8 @@ impl AgentTerminal {
                 return;
             }
         };
+        #[cfg(windows)]
+        let process = WindowsProcess::attach(&pty);
         let event_loop = match EventLoop::new(terminal.clone(), listener, pty, true, false) {
             Ok(event_loop) => event_loop,
             Err(err) => {
@@ -480,10 +592,14 @@ impl AgentTerminal {
             notifier,
             events,
             running: true,
+            #[cfg(windows)]
+            process,
         });
         log::info!("embedded Pi process started");
+        // A resumed session says nothing: Pi's own banner is already on screen, and the status line
+        // is worth the row it costs only when something needs saying.
         self.status = if continue_previous {
-            "Resuming this project's Pi session…".to_owned()
+            String::new()
         } else {
             "Started a new Pi session.".to_owned()
         };
@@ -512,6 +628,20 @@ impl AgentTerminal {
         changed
     }
 
+    /// Clear the conversation and start over.
+    ///
+    /// Pi's own `/new` does this in place. Respawning the process costs a cold start and drops the
+    /// scrollback, so only fall back to that when there is no running session to ask.
+    pub fn new_session(&mut self, cx: &App) {
+        match &self.session {
+            Some(session) if session.running => {
+                session.notifier.notify(Cow::Borrowed(b"/new\r".as_slice()));
+                self.status = "Started a new Pi session.".to_owned();
+            }
+            _ => self.start(false, cx),
+        }
+    }
+
     pub fn stop(&mut self) {
         if self.is_running() {
             log::info!("stopping embedded Pi");
@@ -573,10 +703,13 @@ impl AgentTerminal {
         };
         let cols = cols.max(2);
         let lines = lines.max(2);
-        session
-            .terminal
-            .lock()
-            .resize(SizedTerminal { lines, cols });
+        {
+            let mut terminal = session.terminal.lock();
+            if terminal.columns() == cols && terminal.screen_lines() == lines {
+                return;
+            }
+            terminal.resize(SizedTerminal { lines, cols });
+        }
         if session.running {
             session.notifier.on_resize(WindowSize {
                 num_lines: lines.min(u16::MAX as usize) as u16,
@@ -623,7 +756,6 @@ impl AgentTerminal {
             return TerminalScreen {
                 text: String::new(),
                 highlights: Vec::new(),
-                backgrounds: Vec::new(),
                 auth_urls: Vec::new(),
             };
         };
@@ -640,7 +772,6 @@ impl AgentTerminal {
             .map(|_| Vec::with_capacity(COLS))
             .collect();
         let mut auth_urls = Vec::new();
-        let mut backgrounds: Vec<TerminalBackground> = Vec::new();
         let mut seen_urls = HashSet::new();
         let top = -(content.display_offset as i32);
         for indexed in content.display_iter {
@@ -649,24 +780,6 @@ impl AgentTerminal {
                 .ok()
                 .and_then(|line| lines.get_mut(line))
             {
-                let line = usize::try_from(line).expect("visible terminal line is non-negative");
-                let column = indexed.point.column.0;
-                if let Some(color) = terminal_background(indexed.cell, palette, content.colors) {
-                    if let Some(previous) = backgrounds.last_mut()
-                        && previous.line == line
-                        && previous.end_column == column
-                        && previous.color == color
-                    {
-                        previous.end_column += 1;
-                    } else {
-                        backgrounds.push(TerminalBackground {
-                            line,
-                            start_column: column,
-                            end_column: column + 1,
-                            color,
-                        });
-                    }
-                }
                 if let Some(hyperlink) = indexed.cell.hyperlink() {
                     let uri = hyperlink.uri();
                     if is_auth_url(uri) && seen_urls.insert(uri.to_owned()) {
@@ -738,7 +851,6 @@ impl AgentTerminal {
         TerminalScreen {
             text,
             highlights,
-            backgrounds,
             auth_urls,
         }
     }
@@ -746,11 +858,24 @@ impl AgentTerminal {
 
 #[cfg(test)]
 mod tests {
+    #[cfg(windows)]
+    use std::os::windows::io::AsRawHandle as _;
+    #[cfg(windows)]
+    use std::os::windows::process::CommandExt as _;
+    #[cfg(windows)]
+    use std::process::Command;
+    #[cfg(windows)]
+    use std::time::{Duration, Instant};
+
     use alacritty_terminal::term::cell::{Cell, Flags};
     use alacritty_terminal::term::color::Colors;
     use alacritty_terminal::vte::ansi::{Color, Rgb};
     use gpui::hsla;
+    #[cfg(windows)]
+    use windows::Win32::Foundation::HANDLE;
 
+    #[cfg(windows)]
+    use super::{CREATE_NO_WINDOW, WindowsProcess};
     use super::{TerminalPalette, is_auth_url, terminal_background, terminal_style};
 
     fn palette() -> TerminalPalette {
@@ -778,6 +903,33 @@ mod tests {
         assert!(!is_auth_url("javascript:alert(1)"));
     }
 
+    #[cfg(windows)]
+    #[test]
+    fn terminating_a_windows_process_guard_stops_its_child() {
+        let mut child = Command::new("cmd.exe")
+            .args(["/d", "/s", "/c", "ping -n 30 127.0.0.1 > nul"])
+            .creation_flags(CREATE_NO_WINDOW)
+            .spawn()
+            .expect("spawn a long-lived child");
+        let mut process =
+            WindowsProcess::attach_handle(HANDLE(child.as_raw_handle()), Some(child.id()));
+        assert!(
+            process.job.is_some(),
+            "test child was not assigned to a job"
+        );
+
+        process.terminate();
+        let deadline = Instant::now() + Duration::from_secs(3);
+        while Instant::now() < deadline {
+            if child.try_wait().expect("read child status").is_some() {
+                return;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        let _ = child.kill();
+        panic!("the guarded child survived termination");
+    }
+
     #[test]
     fn dim_text_and_explicit_backgrounds_survive_rendering() {
         let mut cell = Cell::default();
@@ -789,6 +941,11 @@ mod tests {
         assert_eq!(terminal_background(&cell, palette(), &colors), None);
 
         cell.bg = Color::Spec(Rgb { r: 1, g: 2, b: 3 });
+        // The background rides the same style run as the glyph, so gpui places the two together.
+        assert_eq!(
+            terminal_style(&cell, palette(), &colors).background_color,
+            terminal_background(&cell, palette(), &colors)
+        );
         assert!(terminal_background(&cell, palette(), &colors).is_some());
     }
 }
diff --git a/crates/ai/src/agent.rs b/crates/ai/src/agent.rs
index f3dba0e0..1e39b898 100644
--- a/crates/ai/src/agent.rs
+++ b/crates/ai/src/agent.rs
@@ -1,77 +1,151 @@
-//! The narrow contract between a running qrate app and an external agent: reads of live project
-//! state, and one call that hands findings back as drafts without changing a cell.
+//! Protocol v2 between a running qrate app and an external review agent.
 //!
-//! This module deliberately knows nothing about MCP, GPUI, or a model provider. A transport and
-//! the app-side adapter can both depend on these serializable messages without the `ai` crate
-//! depending back on the table or UI.
+//! Reads are progressive: a cheap overview, a bounded declarative query, and optional scratch
+//! programs and thumbnails. The only state-changing request publishes draft findings; it never
+//! changes a table cell.
 
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 
-/// An agent-visible table revision. A later state-changing request must echo this value so qrate
-/// can reject a proposal made against stale cell contents.
 #[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
 #[serde(transparent)]
 pub struct TableRevision(pub u64);
 
-/// What an external agent may ask of the bridge. Every variant but [`Request::StageFindings`] is
-/// a read; that one publishes advisory diagnostics and still changes no cell.
-#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(tag = "method", content = "params", rename_all = "snake_case")]
 pub enum Request {
-    ProjectSummary,
-    Columns,
-    Rows {
-        rows: Vec<usize>,
+    Overview,
+    Query(Query),
+    ProgramSave {
+        source: String,
     },
-    SearchRows {
-        query: String,
-        limit: usize,
+    ProgramRun {
+        revision: TableRevision,
+        #[serde(default)]
+        args: Value,
+    },
+    Thumbnails {
+        items: Vec<ThumbnailRequest>,
     },
-    Diagnostics,
-    SelectedRows,
-    /// Hand back what a review found, as drafts. `revision` is the table the batch was judged
-    /// against, echoed from the read that produced it.
     StageFindings {
         revision: TableRevision,
         findings: Vec<Finding>,
     },
 }
 
-/// One draft finding: where it is, why, and — optionally — what the agent proposes the cell should
-/// say instead.
-///
-/// A proposal is never applied by staging it. `expected` is the cell text the agent judged, and a
-/// draft is only ever offered against a cell that still says exactly that, so an archivist who
-/// edited the cell in the meantime is not offered a correction to text nobody checked.
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+pub struct Query {
+    #[serde(default)]
+    pub source: QuerySource,
+    #[serde(default)]
+    pub select: Vec<String>,
+    #[serde(default, rename = "where")]
+    pub filters: Vec<QueryFilter>,
+    #[serde(default)]
+    pub group_by: Vec<String>,
+    /// Return each exact value of one field once. Mutually exclusive with `select` and `group_by`.
+    pub distinct: Option<String>,
+    pub order_by: Option<OrderBy>,
+    #[serde(default = "default_query_limit")]
+    pub limit: usize,
+    pub cursor: Option<String>,
+}
+
+impl Default for Query {
+    fn default() -> Self {
+        Self {
+            source: QuerySource::default(),
+            select: Vec::new(),
+            filters: Vec::new(),
+            group_by: Vec::new(),
+            distinct: None,
+            order_by: None,
+            limit: DEFAULT_QUERY_LIMIT,
+            cursor: None,
+        }
+    }
+}
+
+fn default_query_limit() -> usize {
+    DEFAULT_QUERY_LIMIT
+}
+
+#[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq, Serialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum QuerySource {
+    #[default]
+    AllRows,
+    SelectedRows,
+    Rows {
+        rows: Vec<usize>,
+    },
+    Search {
+        text: String,
+    },
+    Diagnostics {
+        #[serde(default)]
+        severities: Vec<Severity>,
+        #[serde(default)]
+        sources: Vec<String>,
+    },
+}
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+pub struct QueryFilter {
+    pub field: String,
+    pub op: FilterOp,
+    pub value: Option<String>,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FilterOp {
+    Equals,
+    NotEquals,
+    Contains,
+    StartsWith,
+    EndsWith,
+    IsBlank,
+    IsNotBlank,
+}
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+pub struct OrderBy {
+    pub field: String,
+    #[serde(default)]
+    pub descending: bool,
+}
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+pub struct ThumbnailRequest {
+    pub row: usize,
+    #[serde(default)]
+    pub page: usize,
+}
+
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
 pub struct Finding {
-    /// Source-row index, as handed out by [`Request::Rows`] — never a filtered-view index.
     pub row: usize,
     pub column: String,
     pub severity: Severity,
-    /// Why this is a finding, in the archivist's terms. This is what the Problems panel shows.
     pub message: String,
     pub expected: String,
-    /// The cell's whole new text, not a fragment. `None` is an observation with no fix to offer.
     pub replacement: Option<String>,
 }
 
 impl Request {
-    /// Reject oversized or ambiguous requests before the app reads live project data.
     pub fn validate(&self) -> Result<(), RequestError> {
         match self {
-            Self::Rows { rows } if rows.is_empty() => Err(RequestError::EmptyRows),
-            Self::Rows { rows } if rows.len() > MAX_ROWS_PER_REQUEST => {
-                Err(RequestError::TooManyRows)
-            }
-            Self::SearchRows { query, .. } if query.trim().is_empty() => {
-                Err(RequestError::EmptyQuery)
+            Self::Query(query) => query.validate(),
+            Self::ProgramSave { source } if source.trim().is_empty() => {
+                Err(RequestError::EmptyProgram)
             }
-            Self::SearchRows { query, .. } if query.len() > MAX_QUERY_LEN => {
-                Err(RequestError::QueryTooLong)
+            Self::ProgramSave { source } if source.len() > MAX_PROGRAM_BYTES => {
+                Err(RequestError::ProgramTooLarge)
             }
-            Self::SearchRows { limit, .. } if *limit == 0 || *limit > MAX_ROWS_PER_REQUEST => {
-                Err(RequestError::InvalidSearchLimit)
+            Self::Thumbnails { items } if items.is_empty() => Err(RequestError::EmptyThumbnails),
+            Self::Thumbnails { items } if items.len() > MAX_THUMBNAILS => {
+                Err(RequestError::TooManyThumbnails)
             }
             Self::StageFindings { findings, .. } if findings.len() > MAX_FINDINGS => {
                 Err(RequestError::TooManyFindings)
@@ -91,42 +165,71 @@ impl Request {
     }
 }
 
-pub const MAX_ROWS_PER_REQUEST: usize = 100;
-pub const MAX_QUERY_LEN: usize = 256;
+impl Query {
+    fn validate(&self) -> Result<(), RequestError> {
+        if self.limit == 0 || self.limit > MAX_QUERY_LIMIT {
+            return Err(RequestError::InvalidQueryLimit);
+        }
+        if self.select.len() > MAX_QUERY_FIELDS || self.group_by.len() > MAX_GROUP_FIELDS {
+            return Err(RequestError::TooManyQueryFields);
+        }
+        if self.distinct.is_some() && (!self.select.is_empty() || !self.group_by.is_empty()) {
+            return Err(RequestError::ConflictingQueryOperations);
+        }
+        if self.filters.len() > MAX_QUERY_FILTERS {
+            return Err(RequestError::TooManyQueryFilters);
+        }
+        if let QuerySource::Rows { rows } = &self.source
+            && (rows.is_empty() || rows.len() > MAX_EXPLICIT_ROWS)
+        {
+            return Err(RequestError::InvalidExplicitRows);
+        }
+        if let QuerySource::Search { text } = &self.source
+            && (text.trim().is_empty() || text.len() > MAX_QUERY_TEXT)
+        {
+            return Err(RequestError::InvalidSearchText);
+        }
+        Ok(())
+    }
+}
+
+pub const DEFAULT_QUERY_LIMIT: usize = 20;
+pub const MAX_QUERY_LIMIT: usize = 50;
+pub const MAX_QUERY_FIELDS: usize = 64;
+pub const MAX_GROUP_FIELDS: usize = 4;
+pub const MAX_QUERY_FILTERS: usize = 16;
+pub const MAX_EXPLICIT_ROWS: usize = 100;
+pub const MAX_QUERY_TEXT: usize = 256;
+pub const MAX_PROGRAM_BYTES: usize = 64 * 1024;
+pub const MAX_THUMBNAILS: usize = 4;
 pub const MAX_FINDINGS: usize = 200;
-/// A finding is a sentence for the Problems panel, not an essay — and the panel's row is one line.
 pub const MAX_MESSAGE_LEN: usize = 512;
+pub const MAX_RESULT_BYTES: usize = 16 * 1024;
 
-/// A successful read response. Every result carries the revision used to produce it.
-#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 pub struct Response {
     pub revision: TableRevision,
     #[serde(flatten)]
     pub result: ResultSet,
 }
 
-#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(tag = "result", rename_all = "snake_case")]
 pub enum ResultSet {
-    ProjectSummary(ProjectSummary),
-    Columns {
-        columns: Vec<Column>,
-    },
-    Rows {
-        rows: Vec<Row>,
-    },
-    Diagnostics {
-        diagnostics: Vec<Diagnostic>,
-    },
-    SelectedRows {
-        rows: Vec<usize>,
-    },
-    /// `stale` indexes into the submitted batch, so an agent learns which drafts to re-read rather
-    /// than only how many were dropped.
-    Staged {
-        accepted: usize,
-        stale: Vec<usize>,
-    },
+    Overview(Overview),
+    Query(QueryResult),
+    ProgramSaved { version: u64, hash: String },
+    ProgramRun(ProgramOutput),
+    Thumbnails { items: Vec<Thumbnail> },
+    Staged { accepted: usize, stale: Vec<usize> },
+}
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+pub struct Overview {
+    pub project: ProjectSummary,
+    pub columns: Vec<Column>,
+    pub selected_rows: usize,
+    pub diagnostics: DiagnosticSummary,
 }
 
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
@@ -144,26 +247,61 @@ pub struct Column {
     pub notes: String,
 }
 
-#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
-pub struct Row {
-    /// Source-row index, never a filtered-view index.
-    pub index: usize,
-    pub fields: Vec<Field>,
+#[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq, Serialize)]
+pub struct DiagnosticSummary {
+    pub errors: usize,
+    pub warnings: usize,
+    pub notes: usize,
+    pub sources: Vec<Count>,
 }
 
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
-pub struct Field {
-    pub column: String,
+pub struct Count {
     pub value: String,
+    pub count: usize,
+}
+
+/// A schema-once page. Every item uses the positions in `fields`.
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+pub struct QueryResult {
+    pub fields: Vec<String>,
+    pub items: Vec<Vec<Value>>,
+    pub returned: usize,
+    pub remaining: usize,
+    pub truncated: bool,
+    pub next_cursor: Option<String>,
+}
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+pub struct ProgramOutput {
+    pub version: u64,
+    pub hash: String,
+    pub value: Value,
+    pub logs: Vec<String>,
+    pub elapsed_ms: u64,
+    pub audit: ProgramAudit,
 }
 
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
-pub struct Diagnostic {
-    pub row: Option<usize>,
-    pub column: Option<String>,
-    pub severity: Severity,
-    pub source: String,
-    pub message: String,
+pub struct ProgramAudit {
+    pub run_id: String,
+    pub revision: TableRevision,
+    pub version: u64,
+    pub hash: String,
+    pub elapsed_ms: u64,
+    pub memory_limit_bytes: usize,
+    pub deadline_ms: u64,
+    pub output_limit_bytes: usize,
+    pub status: String,
+    pub output_bytes: usize,
+}
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+pub struct Thumbnail {
+    pub row: usize,
+    pub page: usize,
+    pub media_type: String,
+    pub data: String,
 }
 
 #[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
@@ -174,17 +312,30 @@ pub enum Severity {
     Note,
 }
 
-/// Errors that may be returned without reading or exposing project data.
-#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Serialize)]
-#[serde(rename_all = "snake_case")]
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[serde(tag = "code", content = "detail", rename_all = "snake_case")]
 pub enum RequestError {
     ProjectUnavailable,
     TableUnavailable,
-    EmptyRows,
-    TooManyRows,
-    EmptyQuery,
-    QueryTooLong,
-    InvalidSearchLimit,
+    InvalidQueryLimit,
+    TooManyQueryFields,
+    TooManyQueryFilters,
+    InvalidExplicitRows,
+    InvalidSearchText,
+    InvalidCursor,
+    StaleCursor,
+    UnknownField,
+    ConflictingQueryOperations,
+    ResultItemTooLarge,
+    EmptyProgram,
+    ProgramTooLarge,
+    ProgramUnavailable,
+    ProgramCompileFailed(String),
+    ProgramFailed(String),
+    ProgramOutputTooLarge(String),
+    EmptyThumbnails,
+    TooManyThumbnails,
+    ThumbnailUnavailable,
     TooManyFindings,
     EmptyFindingMessage,
     FindingMessageTooLong,
@@ -192,101 +343,74 @@ pub enum RequestError {
 
 #[cfg(test)]
 mod tests {
-    use super::{
-        Finding, MAX_FINDINGS, MAX_MESSAGE_LEN, MAX_QUERY_LEN, MAX_ROWS_PER_REQUEST, Request,
-        RequestError, ResultSet, Severity, TableRevision,
-    };
-
-    fn finding(message: &str) -> Finding {
-        Finding {
-            row: 0,
-            column: "Title".into(),
-            severity: Severity::Warning,
-            message: message.into(),
-            expected: "Harvest".into(),
-            replacement: Some("Harvest, 1962".into()),
-        }
-    }
+    use super::*;
 
-    fn staged(findings: Vec<Finding>) -> Result<(), RequestError> {
-        Request::StageFindings {
-            revision: TableRevision(1),
-            findings,
-        }
-        .validate()
-    }
-
-    /// A batch of drafts is bounded before any of it reaches the Problems panel: an agent must not
-    /// be able to bury the archivist's own findings under thousands of its own, or push an essay
-    /// into a one-line panel row.
     #[test]
-    fn staged_findings_are_bounded_and_must_say_something() {
-        assert_eq!(staged(vec![finding("date is a guess")]), Ok(()));
-        assert_eq!(staged(vec![]), Ok(()), "an empty batch retracts");
+    fn query_defaults_are_small_and_bounded() {
+        let request: Request =
+            serde_json::from_value(serde_json::json!({"method":"query","params":{}})).unwrap();
+        let Request::Query(query) = request else {
+            panic!()
+        };
+        assert_eq!(query.limit, DEFAULT_QUERY_LIMIT);
+        assert_eq!(query.source, QuerySource::AllRows);
+        assert_eq!(query.validate(), Ok(()));
         assert_eq!(
-            staged(vec![finding("x"); MAX_FINDINGS + 1]),
-            Err(RequestError::TooManyFindings)
-        );
-        assert_eq!(
-            staged(vec![finding("  ")]),
-            Err(RequestError::EmptyFindingMessage)
-        );
-        assert_eq!(
-            staged(vec![finding(&"x".repeat(MAX_MESSAGE_LEN + 1))]),
-            Err(RequestError::FindingMessageTooLong)
+            Query {
+                limit: MAX_QUERY_LIMIT + 1,
+                ..query
+            }
+            .validate(),
+            Err(RequestError::InvalidQueryLimit)
         );
     }
 
     #[test]
-    fn request_validation_bounds_agent_reads() {
+    fn dynamic_inputs_are_bounded() {
         assert_eq!(
-            Request::Rows { rows: vec![] }.validate(),
-            Err(RequestError::EmptyRows)
+            Request::Thumbnails { items: vec![] }.validate(),
+            Err(RequestError::EmptyThumbnails)
         );
         assert_eq!(
-            Request::Rows {
-                rows: vec![0; MAX_ROWS_PER_REQUEST + 1]
-            }
-            .validate(),
-            Err(RequestError::TooManyRows)
-        );
-        assert_eq!(
-            Request::SearchRows {
-                query: " ".into(),
-                limit: 1
-            }
-            .validate(),
-            Err(RequestError::EmptyQuery)
-        );
-        assert_eq!(
-            Request::SearchRows {
-                query: "x".repeat(MAX_QUERY_LEN + 1),
-                limit: 1
-            }
-            .validate(),
-            Err(RequestError::QueryTooLong)
+            Request::ProgramSave { source: " ".into() }.validate(),
+            Err(RequestError::EmptyProgram)
         );
         assert_eq!(
-            Request::SearchRows {
-                query: "title".into(),
-                limit: 0
+            Query {
+                distinct: Some("Title".into()),
+                select: vec!["Title".into()],
+                ..Query::default()
             }
             .validate(),
-            Err(RequestError::InvalidSearchLimit)
+            Err(RequestError::ConflictingQueryOperations)
         );
     }
 
     #[test]
-    fn protocol_tags_preserve_the_requested_operation() {
-        let request = serde_json::to_value(Request::SelectedRows).unwrap();
-        assert_eq!(request["method"], "selected_rows");
+    fn errors_have_a_stable_code_and_optional_bounded_detail() {
+        let value = serde_json::to_value(RequestError::ProgramCompileFailed(
+            "line 1: expected function".into(),
+        ))
+        .unwrap();
+        assert_eq!(value["code"], "program_compile_failed");
+        assert_eq!(value["detail"], "line 1: expected function");
+    }
 
-        let response = serde_json::to_value(super::Response {
+    #[test]
+    fn pages_are_schema_once() {
+        let value = serde_json::to_value(Response {
             revision: TableRevision(4),
-            result: ResultSet::SelectedRows { rows: vec![1, 3] },
+            result: ResultSet::Query(QueryResult {
+                fields: vec!["row".into(), "Title".into()],
+                items: vec![vec![Value::from(2), Value::from("Harvest")]],
+                returned: 1,
+                remaining: 0,
+                truncated: false,
+                next_cursor: None,
+            }),
         })
         .unwrap();
-        assert_eq!(response["revision"], 4);
-        assert_eq!(response["result"], "selected_rows");
+        assert_eq!(value["result"], "query");
+        assert_eq!(value["items"][0][1], "Harvest");
     }
 }
diff --git a/crates/app/src/about.rs b/crates/app/src/about.rs
index 49664247..8b349e10 100644
--- a/crates/app/src/about.rs
+++ b/crates/app/src/about.rs
@@ -86,9 +86,9 @@ impl Render for AboutWindow {
                         }
                         Check::Done(Some(UpdateStatus::Available {
                             version,
-                            download_url,
+                            download_page,
                         })) => {
-                            let url = download_url.clone();
+                            let url = download_page.clone();
                             h_flex()
                                 .gap_2()
                                 .items_center()
diff --git a/crates/app/src/agent_bridge.rs b/crates/app/src/agent_bridge.rs
index c0a1616e..0fd73fc1 100644
--- a/crates/app/src/agent_bridge.rs
+++ b/crates/app/src/agent_bridge.rs
@@ -109,7 +109,7 @@ pub fn init(cx: &mut App) {
             };
             loop {
                 match listener.accept() {
-                    Ok((stream, _)) => cx.update(|cx| serve(stream, token, &mut seen, cx)),
+                    Ok((stream, _)) => serve(stream, token, &mut seen, cx).await,
                     Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => break,
                     Err(err) => {
                         log::warn!("agent bridge dropped a connection: {err}");
@@ -168,7 +168,7 @@ fn start() -> Option<(TcpListener, String)> {
     })?;
     let port = listener.local_addr().map_or(0, |addr| addr.port());
     let endpoint = json!({
-        "bridge_protocol": 1,
+        "bridge_protocol": 2,
         "url": format!("http://127.0.0.1:{port}"),
         "token": token
     });
@@ -208,11 +208,11 @@ pub fn shutdown() {
     }
 }
 
-fn serve(
+async fn serve(
     mut stream: TcpStream,
     token: &str,
     seen: &mut HashMap<SharedString, Instant>,
-    cx: &mut App,
+    cx: &mut gpui::AsyncApp,
 ) {
     let framed = stream
         .set_read_timeout(Some(POLL))
@@ -243,10 +243,12 @@ fn serve(
         .as_ref()
         .is_ok_and(|request| request.credential.as_deref() == Some(token));
     if authenticated && !seen.contains_key(&agent) {
-        workspace::record_agent_call(
-            lifecycle(agent.clone(), "connected", "first call from this agent"),
-            cx,
-        );
+        cx.update(|cx| {
+            workspace::record_agent_call(
+                lifecycle(agent.clone(), "connected", "first call from this agent"),
+                cx,
+            )
+        });
     }
     if authenticated {
         seen.insert(agent.clone(), Instant::now());
@@ -277,7 +279,10 @@ fn serve(
             ),
             Ok(request) => {
                 (method, detail) = summarize(&request);
-                match table::respond_to_agent(request, cx) {
+                let response = cx
+                    .update(|cx| table::respond_to_agent_async(request, cx))
+                    .await;
+                match response {
                     Ok(response) => {
                         let outcome = describe(&response.result);
                         (
@@ -298,20 +303,22 @@ fn serve(
         },
     };
 
-    workspace::record_agent_call(
-        AgentCall {
-            agent,
-            label: method,
-            detail,
-            outcome,
-            entry: match refused {
-                true => AgentEntry::Refused,
-                false => AgentEntry::Answered,
+    cx.update(|cx| {
+        workspace::record_agent_call(
+            AgentCall {
+                agent,
+                label: method,
+                detail,
+                outcome,
+                entry: match refused {
+                    true => AgentEntry::Refused,
+                    false => AgentEntry::Answered,
+                },
+                took: started.elapsed(),
             },
-            took: started.elapsed(),
-        },
-        cx,
-    );
+            cx,
+        )
+    });
 
     let body = payload.to_string();
     let reply = format!(
@@ -334,12 +341,20 @@ fn summarize(request: &ai::agent::Request) -> (SharedString, SharedString) {
         .unwrap_or_else(|| "?".into());
 
     let detail = match request {
-        ai::agent::Request::Rows { rows } => format!("{} row(s)", rows.len()).into(),
-        ai::agent::Request::SearchRows { query, limit } => format!("“{query}”, max {limit}").into(),
+        ai::agent::Request::Query(query) => {
+            format!("{:?}, max {}", query.source, query.limit).into()
+        }
         ai::agent::Request::StageFindings { revision, findings } => {
             format!("{} finding(s) at revision {}", findings.len(), revision.0).into()
         }
-        _ => SharedString::default(),
+        ai::agent::Request::ProgramSave { source } => {
+            format!("{} source byte(s)", source.len()).into()
+        }
+        ai::agent::Request::ProgramRun { revision, .. } => {
+            format!("at revision {}", revision.0).into()
+        }
+        ai::agent::Request::Thumbnails { items } => format!("{} thumbnail(s)", items.len()).into(),
+        ai::agent::Request::Overview => SharedString::default(),
     };
     (method, detail)
 }
@@ -348,17 +363,23 @@ fn summarize(request: &ai::agent::Request) -> (SharedString, SharedString) {
 /// call happened; the archive's own data does not belong in a debugging list.
 fn describe(result: &ai::agent::ResultSet) -> SharedString {
     match result {
-        ai::agent::ResultSet::ProjectSummary(summary) => format!(
+        ai::agent::ResultSet::Overview(overview) => format!(
             "{} rows × {} columns",
-            summary.row_count, summary.column_count
+            overview.project.row_count, overview.project.column_count
         )
         .into(),
-        ai::agent::ResultSet::Columns { columns } => format!("{} columns", columns.len()).into(),
-        ai::agent::ResultSet::Rows { rows } => format!("{} rows", rows.len()).into(),
-        ai::agent::ResultSet::Diagnostics { diagnostics } => {
-            format!("{} diagnostics", diagnostics.len()).into()
+        ai::agent::ResultSet::Query(page) => {
+            format!("{} returned, {} remaining", page.returned, page.remaining).into()
+        }
+        ai::agent::ResultSet::ProgramSaved { version, hash } => {
+            format!("program v{version} {hash}").into()
+        }
+        ai::agent::ResultSet::ProgramRun(output) => {
+            format!("program v{} in {} ms", output.version, output.elapsed_ms).into()
+        }
+        ai::agent::ResultSet::Thumbnails { items } => {
+            format!("{} thumbnail(s)", items.len()).into()
         }
-        ai::agent::ResultSet::SelectedRows { rows } => format!("{} selected", rows.len()).into(),
         ai::agent::ResultSet::Staged { accepted, stale } => {
             format!("{accepted} staged, {} stale", stale.len()).into()
         }
@@ -369,7 +390,12 @@ fn describe(result: &ai::agent::ResultSet) -> SharedString {
 fn wire_name(error: &ai::agent::RequestError) -> SharedString {
     serde_json::to_value(error)
         .ok()
-        .and_then(|value| value.as_str().map(SharedString::from))
+        .and_then(|value| {
+            value
+                .get("code")
+                .and_then(|code| code.as_str())
+                .map(SharedString::from)
+        })
         .unwrap_or_else(|| "request_error".into())
 }
 
diff --git a/crates/app/src/app_settings/mod.rs b/crates/app/src/app_settings/mod.rs
index 993ed9cd..292b1a43 100644
--- a/crates/app/src/app_settings/mod.rs
+++ b/crates/app/src/app_settings/mod.rs
@@ -62,6 +62,29 @@ pub fn build_pages(cx: &App) -> Vec<SettingPage> {
                     }
                     .into_item(cx),
                 ),
+            )
+            .group(
+                SettingGroup::new()
+                    .title("Terminal")
+                    .item(
+                        Setting::Text {
+                            key: workspace::AGENT_FONT_KEY,
+                            label: "Font",
+                            description: "A monospace family installed on this machine, such as \
+                                          Cascadia Mono or Consolas. Leave empty to let qrate \
+                                          pick one. A proportional font will not line up.",
+                        }
+                        .into_item(cx),
+                    )
+                    .item(
+                        Setting::Dropdown {
+                            key: workspace::AGENT_FONT_SIZE_KEY,
+                            label: "Font size",
+                            description: "Larger type gives the agent fewer columns to draw in.",
+                            options: workspace::AGENT_FONT_SIZES,
+                        }
+                        .into_item(cx),
+                    ),
             ),
         SettingPage::new("Spelling").group(spelling_group(cx)),
         google_page(cx),
diff --git a/crates/app/src/logging.rs b/crates/app/src/logging.rs
index 1767dce7..ba28a01d 100644
--- a/crates/app/src/logging.rs
+++ b/crates/app/src/logging.rs
@@ -84,8 +84,9 @@ fn version() -> &'static str {
 /// Removes GPUI's unactionable focus-node notice and demotes window-teardown errors to debug.
 ///
 /// GPUI reports the missing focus node at info level every time focus reaches an internal element
-/// without an accessibility role. It is not actionable from a qrate log, so omit that exact notice
-/// while retaining other accessibility diagnostics.
+/// without an accessibility role, and narrates every accessibility tree update — one line per
+/// frame, which was four fifths of a session log. Neither is actionable from a qrate log, so omit
+/// those two notices while retaining other accessibility diagnostics.
 ///
 /// Closing a window always produces them: a detached per-window callback outlives the window,
 /// reads it, and `log_err`s the expected miss. Kept in the file for teardown debugging, but not at
@@ -93,11 +94,11 @@ fn version() -> &'static str {
 struct QuietGpuiNoise(Box<dyn Log>);
 
 fn is_accessibility_focus_noise(record: &Record) -> bool {
-    record.target() == "gpui::window::a11y"
-        && record
-            .args()
-            .to_string()
-            .starts_with("a11y: focused element")
+    if !record.target().starts_with("gpui::window") {
+        return false;
+    }
+    let message = record.args().to_string();
+    message.starts_with("a11y: focused element") || message.starts_with("Sending a11y tree update")
 }
 
 fn is_window_teardown(record: &Record) -> bool {
diff --git a/crates/app/src/main.rs b/crates/app/src/main.rs
index 7ed8d3fb..af342eeb 100644
--- a/crates/app/src/main.rs
+++ b/crates/app/src/main.rs
@@ -415,13 +415,11 @@ fn flush_all_state(cx: &mut gpui::App) {
 }
 
 fn main() {
+    // First, so failures in GPUI platform construction and startup still reach the log file.
+    logging::init();
     let app = gpui_platform::application().with_assets(assets::Assets);
 
     app.run(move |cx| {
-        // First, so a failure during the startup below — the case where no window ever appears to
-        // report it — still reaches the log file.
-        logging::init();
-
         gpui_component::init(cx);
 
         // Settings ------------------------------------
diff --git a/crates/app/src/status_items/mod.rs b/crates/app/src/status_items/mod.rs
index c01a6e8a..1dbb7b07 100644
--- a/crates/app/src/status_items/mod.rs
+++ b/crates/app/src/status_items/mod.rs
@@ -2,7 +2,6 @@ mod cell_location;
 pub mod markup;
 mod panel_buttons;
 mod plugin_bar;
-mod update_notice;
 
 use cell_location::CellLocation;
 use gpui::*;
@@ -10,7 +9,6 @@ use gpui_component::dock::DockArea;
 use panel_buttons::PanelButtons;
 use plugin_api::{Bar, BarContributions, Side};
 pub use plugin_bar::PluginBar;
-use update_notice::UpdateNotice;
 use window_wrapper::{BarRegistry, status_bar::StatusBarRegistry};
 use workspace::{BarSide, DockToggleButton, PANELS};
 
@@ -52,12 +50,6 @@ pub fn build_status_bar_registry(cx: &mut App, dock: WeakEntity<DockArea>) -> St
         !BarContributions::at(Bar::Status, Side::Right, cx).is_empty()
     });
 
-    // Dismissible "an update is available" text, shown only once `update_check::check` finds one.
-    let update_notice = cx.new(UpdateNotice::new);
-    registry
-        .items_mut()
-        .add_right_if(update_notice, UpdateNotice::occupied);
-
     // Text readout of the table's selected cell.
     let cell_location = cx.new(CellLocation::new);
     registry.items_mut().add_right(cell_location);
diff --git a/crates/app/src/title_items/mod.rs b/crates/app/src/title_items/mod.rs
index 144a207a..df49fb9f 100644
--- a/crates/app/src/title_items/mod.rs
+++ b/crates/app/src/title_items/mod.rs
@@ -1,3 +1,5 @@
+mod update_notice;
+
 use gpui::*;
 use gpui_component::{
     dock::{DockArea, DockPlacement},
@@ -7,6 +9,7 @@ use plugin_api::{Bar, Side};
 
 use crate::actions::{ToggleBottomDock, ToggleLeftDock, ToggleRightDock};
 use crate::status_items::PluginBar;
+use update_notice::UpdateNotice;
 use window_wrapper::{BarRegistry, title_bar::TitleBarRegistry};
 use workspace::DockToggleButton;
 
@@ -26,6 +29,13 @@ pub fn build_title_bar_registry(cx: &mut App, dock: WeakEntity<DockArea>) -> Tit
     let plugins = cx.new(|cx| PluginBar::new(Bar::Title, Side::Right, cx));
     registry.items_mut().add_right(plugins);
 
+    // Dismissible "an update is available" text, shown only once `update_check::check` finds one.
+    // Text before buttons, on the same reasoning as the plugin bar above.
+    let update_notice = cx.new(UpdateNotice::new);
+    registry
+        .items_mut()
+        .add_right_if(update_notice, UpdateNotice::occupied);
+
     // The label and action each button hovers with: the action is what makes the tooltip print
     // Ctrl or ⌘ to match whoever is reading it, rather than a string that is wrong on one platform.
     for (id, placement, label, action) in [
diff --git a/crates/app/src/status_items/update_notice.rs b/crates/app/src/title_items/update_notice.rs
similarity index 83%
rename from crates/app/src/status_items/update_notice.rs
rename to crates/app/src/title_items/update_notice.rs
index c95be05c..8b9c5b86 100644
--- a/crates/app/src/status_items/update_notice.rs
+++ b/crates/app/src/title_items/update_notice.rs
@@ -1,6 +1,6 @@
-//! Right-side status-bar widget: shown only while `update_check::AvailableUpdate` is set. Clicking
-//! the text opens the release's platform installer; the close button dismisses it for that
-//! version.
+//! Right-side title-bar widget: shown only while `update_check::AvailableUpdate` is set. Clicking
+//! the text opens the site's download page for the platform installer; the close button dismisses
+//! it for that version.
 
 use gpui::*;
 use gpui_component::button::{Button, ButtonVariants as _};
@@ -29,7 +29,7 @@ impl Render for UpdateNotice {
         let Some(update) = cx.try_global::<AvailableUpdate>().cloned() else {
             return div();
         };
-        let url = update.download_url.clone();
+        let url = update.download_page.clone();
         div().child(
             gpui_component::h_flex()
                 .id("update-notice")
diff --git a/crates/app/src/update_check.rs b/crates/app/src/update_check.rs
index 20c5d553..fbc8b91e 100644
--- a/crates/app/src/update_check.rs
+++ b/crates/app/src/update_check.rs
@@ -1,7 +1,7 @@
 //! One-shot GitHub release check at startup: is the latest tagged release newer than the version
 //! baked into this build? If so, the [`AvailableUpdate`] global carries the platform-matching
-//! download link for `status_items::UpdateNotice` to show. Never downloads or installs anything
-//! itself — `.github/workflows/release.yml` already builds the installer this links to.
+//! download link for `title_items::UpdateNotice` to show. Never downloads or installs anything
+//! itself — the link goes to the site's download page for the platform asset.
 
 use gpui::{App, AppContext as _, Global, Task};
 use serde::Deserialize;
@@ -9,6 +9,10 @@ use settings::AppSettings;
 
 const API_URL: &str = "https://api.github.com/repos/devnull03/qrate/releases/latest";
 
+/// Downloads go through the site rather than straight at the GitHub asset, so the reader lands on
+/// a page that can say what an unsigned build will do before their browser starts saving one.
+const DOWNLOAD_PAGE: &str = "https://qrate.dvnl.work/thanks";
+
 /// Matches the asset names `release.yml` packages under.
 #[cfg(target_os = "windows")]
 const ASSET_SUFFIX: &str = "-setup.exe";
@@ -29,13 +33,12 @@ struct Release {
 #[derive(Deserialize)]
 struct Asset {
     name: String,
-    browser_download_url: String,
 }
 
 #[derive(Clone)]
 pub struct AvailableUpdate {
     pub version: String,
-    pub download_url: String,
+    pub download_page: String,
 }
 
 impl Global for AvailableUpdate {}
@@ -59,7 +62,7 @@ pub enum UpdateStatus {
     UpToDate,
     Available {
         version: String,
-        download_url: String,
+        download_page: String,
     },
 }
 
@@ -79,7 +82,7 @@ pub fn check_now(cx: &App) -> Task<Option<UpdateStatus>> {
             .find(|a| a.name.ends_with(ASSET_SUFFIX))?;
         Some(UpdateStatus::Available {
             version: latest,
-            download_url: asset.browser_download_url.clone(),
+            download_page: format!("{DOWNLOAD_PAGE}?a={}", asset.name),
         })
     })
 }
@@ -92,7 +95,7 @@ pub fn check(cx: &mut App) {
     cx.spawn(async move |cx| {
         let Some(UpdateStatus::Available {
             version,
-            download_url,
+            download_page,
         }) = task.await
         else {
             return;
@@ -107,7 +110,7 @@ pub fn check(cx: &mut App) {
             }
             cx.set_global(AvailableUpdate {
                 version,
-                download_url,
+                download_page,
             });
         });
     })
diff --git a/crates/plugin-host/Cargo.toml b/crates/plugin-host/Cargo.toml
index 485fafdb..fb652f4c 100644
--- a/crates/plugin-host/Cargo.toml
+++ b/crates/plugin-host/Cargo.toml
@@ -10,6 +10,7 @@ settings = { path = "../settings" }
 gpui.workspace = true
 plugin-api = { path = "../plugin-api" }
 serde_json.workspace = true
+preview = { path = "../preview" }
 # `send` makes the VM `Send + Sync`, which is what lets a plugin run on the background executor.
 mlua = { version = "0.11", features = ["luau", "vendored", "serde", "send"] }
 # blocking, like data-exchange: the VM has no way to yield, so a host call has to return a value.
diff --git a/crates/plugin-host/src/agent_program.rs b/crates/plugin-host/src/agent_program.rs
new file mode 100644
index 00000000..bbdf8c6f
--- /dev/null
+++ b/crates/plugin-host/src/agent_program.rs
@@ -0,0 +1,359 @@
+//! One-shot, read-only Luau programs for the agent bridge.
+//!
+//! This deliberately does not load through the user plugin registry. Each call gets a fresh VM,
+//! an immutable snapshot userdata, no ambient I/O, and fixed resource/output budgets.
+
+use std::fs::{self, File};
+use std::io::Read as _;
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use gpui::SharedString;
+use mlua::{
+    Function, Lua, LuaOptions, LuaSerdeExt as _, StdLib, UserData, UserDataMethods, VmState,
+};
+use serde_json::Value as Json;
+
+pub const MEMORY_LIMIT: usize = 64 * 1024 * 1024;
+pub const DEADLINE: Duration = Duration::from_secs(2);
+pub const MAX_OUTPUT: usize = 16 * 1024;
+const MAX_LOGS: usize = 20;
+const MAX_LOG_BYTES: usize = 4 * 1024;
+const MAX_TEXT_READ: usize = 64 * 1024;
+const MAX_TEXT_FILE: u64 = 1024 * 1024;
+const MAX_ERROR: usize = 512;
+
+#[derive(Clone)]
+pub struct AgentDiagnostic {
+    pub row: Option<usize>,
+    pub column: Option<String>,
+    pub severity: String,
+    pub source: String,
+    pub message: String,
+}
+
+#[derive(Clone)]
+pub struct AgentSnapshot {
+    pub revision: u64,
+    pub columns: Vec<SharedString>,
+    pub rows: Vec<Vec<SharedString>>,
+    pub selected_rows: Vec<usize>,
+    pub diagnostics: Vec<AgentDiagnostic>,
+    /// Resolved by qrate. Scripts never receive these paths; host methods are the only access.
+    pub files: Vec<Option<PathBuf>>,
+}
+
+#[derive(Debug)]
+pub struct AgentProgramOutput {
+    pub value: Json,
+    pub logs: Vec<String>,
+    pub elapsed_ms: u64,
+    pub output_bytes: usize,
+}
+
+#[derive(Debug)]
+pub struct AgentProgramError {
+    pub kind: &'static str,
+    pub detail: String,
+    pub elapsed_ms: u64,
+}
+
+impl std::fmt::Display for AgentProgramError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.detail)
+    }
+}
+
+fn bounded_error(kind: &'static str, error: impl ToString, started: Instant) -> AgentProgramError {
+    let mut detail = error.to_string().replace('\\', "/");
+    if let Some(index) = detail.find("agent-scratch") {
+        detail = detail[index..].to_string();
+    }
+    detail.truncate(MAX_ERROR);
+    AgentProgramError {
+        kind,
+        detail,
+        elapsed_ms: started.elapsed().as_millis() as u64,
+    }
+}
+
+#[derive(Clone)]
+struct Snapshot(Arc<AgentSnapshot>);
+
+impl UserData for Snapshot {
+    fn add_methods<M: UserDataMethods<Self>>(methods: &mut M) {
+        methods.add_method("revision", |_, this, ()| Ok(this.0.revision));
+        methods.add_method("row_count", |_, this, ()| Ok(this.0.rows.len()));
+        methods.add_method("columns", |lua, this, ()| {
+            lua.create_sequence_from(this.0.columns.iter().map(|value| value.as_ref()))
+        });
+        methods.add_method("selected_rows", |lua, this, ()| {
+            lua.create_sequence_from(this.0.selected_rows.iter().copied())
+        });
+        methods.add_method("row", |lua, this, row: usize| {
+            let values = this
+                .0
+                .rows
+                .get(row)
+                .ok_or_else(|| mlua::Error::runtime("row out of range"))?;
+            let result = lua.create_table()?;
+            result.set("index", row)?;
+            for (column, value) in this.0.columns.iter().zip(values) {
+                result.set(column.as_str(), value.as_str())?;
+            }
+            Ok(result)
+        });
+        methods.add_method("cell", |_, this, (row, column): (usize, String)| {
+            let index = this
+                .0
+                .columns
+                .iter()
+                .position(|name| name == &column)
+                .ok_or_else(|| mlua::Error::runtime("unknown column"))?;
+            Ok(this
+                .0
+                .rows
+                .get(row)
+                .and_then(|values| values.get(index))
+                .map(ToString::to_string))
+        });
+        methods.add_method("diagnostics", |lua, this, ()| {
+            let result = lua.create_table()?;
+            for (index, diagnostic) in this.0.diagnostics.iter().enumerate() {
+                let item = lua.create_table()?;
+                item.set("row", diagnostic.row)?;
+                item.set("column", diagnostic.column.as_deref())?;
+                item.set("severity", diagnostic.severity.as_str())?;
+                item.set("source", diagnostic.source.as_str())?;
+                item.set("message", diagnostic.message.as_str())?;
+                result.set(index + 1, item)?;
+            }
+            Ok(result)
+        });
+        methods.add_method("file_info", |lua, this, row: usize| {
+            let path = this
+                .0
+                .files
+                .get(row)
+                .and_then(Option::as_ref)
+                .ok_or_else(|| mlua::Error::runtime("row has no linked file"))?;
+            let metadata = fs::metadata(path).map_err(mlua::Error::external)?;
+            let result = lua.create_table()?;
+            result.set(
+                "extension",
+                path.extension()
+                    .and_then(|value| value.to_str())
+                    .unwrap_or_default(),
+            )?;
+            result.set("bytes", metadata.len())?;
+            result.set("has_pdf_text", preview::has_text_layer(path))?;
+            Ok(result)
+        });
+        methods.add_method(
+            "read_text",
+            |_, this, (row, offset, limit): (usize, Option<usize>, Option<usize>)| {
+                let path = this
+                    .0
+                    .files
+                    .get(row)
+                    .and_then(Option::as_ref)
+                    .ok_or_else(|| mlua::Error::runtime("row has no linked file"))?;
+                if preview::has_text(path) {
+                    return Err(mlua::Error::runtime("use pdf_search for PDF text"));
+                }
+                let metadata = fs::metadata(path).map_err(mlua::Error::external)?;
+                if metadata.len() > MAX_TEXT_FILE {
+                    return Err(mlua::Error::runtime(
+                        "linked text file exceeds the 1 MiB policy limit",
+                    ));
+                }
+                let mut bytes = Vec::with_capacity(metadata.len() as usize);
+                File::open(path)
+                    .and_then(|file| file.take(MAX_TEXT_FILE + 1).read_to_end(&mut bytes))
+                    .map_err(mlua::Error::external)?;
+                let text = std::str::from_utf8(&bytes)
+                    .map_err(|_| mlua::Error::runtime("linked file is not UTF-8 text"))?;
+                let offset = offset.unwrap_or(0).min(text.len());
+                let mut start = offset;
+                while !text.is_char_boundary(start) {
+                    start += 1;
+                }
+                let requested = limit.unwrap_or(4096).min(MAX_TEXT_READ);
+                let mut end = (start + requested).min(text.len());
+                while !text.is_char_boundary(end) {
+                    end -= 1;
+                }
+                Ok(text[start..end].to_string())
+            },
+        );
+        methods.add_method(
+            "pdf_search",
+            |lua, this, (row, needle, limit): (usize, String, Option<usize>)| {
+                let path = this
+                    .0
+                    .files
+                    .get(row)
+                    .and_then(Option::as_ref)
+                    .ok_or_else(|| mlua::Error::runtime("row has no linked file"))?;
+                let result = lua.create_table()?;
+                for (index, found) in preview::search(path, &needle)
+                    .into_iter()
+                    .take(limit.unwrap_or(20).min(50))
+                    .enumerate()
+                {
+                    let item = lua.create_table()?;
+                    item.set("page", found.page)?;
+                    let mut line = found.line;
+                    line.truncate(512);
+                    item.set("text", line)?;
+                    result.set(index + 1, item)?;
+                }
+                Ok(result)
+            },
+        );
+    }
+}
+
+fn lua(started: Instant) -> Result<Lua, AgentProgramError> {
+    let deadline = Arc::new(Mutex::new(started + DEADLINE));
+    let lua = Lua::new_with(
+        StdLib::STRING | StdLib::TABLE | StdLib::MATH,
+        LuaOptions::default(),
+    )
+    .map_err(|error| bounded_error("compile_error", error, started))?;
+    lua.set_memory_limit(MEMORY_LIMIT)
+        .map_err(|error| bounded_error("compile_error", error, started))?;
+    lua.set_interrupt(move |_| {
+        if Instant::now() > *deadline.lock().unwrap() {
+            Err(mlua::Error::runtime("ran too long and was stopped"))
+        } else {
+            Ok(VmState::Continue)
+        }
+    });
+    let math: mlua::Table = lua
+        .globals()
+        .get("math")
+        .map_err(|error| bounded_error("compile_error", error, started))?;
+    math.set("random", mlua::Value::Nil)
+        .map_err(|error| bounded_error("compile_error", error, started))?;
+    math.set("randomseed", mlua::Value::Nil)
+        .map_err(|error| bounded_error("compile_error", error, started))?;
+    lua.sandbox(true)
+        .map_err(|error| bounded_error("compile_error", error, started))?;
+    Ok(lua)
+}
+
+pub fn validate_agent_program(source: &str) -> Result<(), AgentProgramError> {
+    let started = Instant::now();
+    let lua = lua(started)?;
+    let _: Function = lua
+        .load(source)
+        .set_name("agent-scratch")
+        .eval()
+        .map_err(|error| bounded_error("compile_error", error, started))?;
+    Ok(())
+}
+
+pub fn run_agent_program(
+    source: &str,
+    args: &Json,
+    snapshot: AgentSnapshot,
+) -> Result<AgentProgramOutput, AgentProgramError> {
+    let started = Instant::now();
+    let logs = Arc::new(Mutex::new(Vec::<String>::new()));
+    let lua = lua(started)?;
+    lua.globals()
+        .set(
+            "print",
+            lua.create_function({
+                let logs = logs.clone();
+                move |_, values: mlua::MultiValue| {
+                    let mut line = values
+                        .iter()
+                        .map(|value| format!("{value:?}"))
+                        .collect::<Vec<_>>()
+                        .join(" ");
+                    line.truncate(MAX_LOG_BYTES);
+                    let mut logs = logs.lock().unwrap();
+                    if logs.len() < MAX_LOGS {
+                        logs.push(line);
+                    }
+                    Ok(())
+                }
+            })
+            .map_err(|error| bounded_error("runtime_error", error, started))?,
+        )
+        .map_err(|error| bounded_error("runtime_error", error, started))?;
+
+    let function: Function = lua
+        .load(source)
+        .set_name("agent-scratch")
+        .eval()
+        .map_err(|error| bounded_error("runtime_error", error, started))?;
+    let argument = lua
+        .to_value(args)
+        .map_err(|error| bounded_error("runtime_error", error, started))?;
+    let value: mlua::Value = function
+        .call((Snapshot(Arc::new(snapshot)), argument))
+        .map_err(|error| bounded_error("runtime_error", error, started))?;
+    let value: Json = lua
+        .from_value(value)
+        .map_err(|error| bounded_error("runtime_error", error, started))?;
+    let output_bytes = serde_json::to_vec(&value).map_or(MAX_OUTPUT + 1, |bytes| bytes.len());
+    if output_bytes > MAX_OUTPUT {
+        return Err(bounded_error(
+            "output_too_large",
+            "program output exceeds 16 KiB; return a smaller projection",
+            started,
+        ));
+    }
+    Ok(AgentProgramOutput {
+        value,
+        logs: Arc::try_unwrap(logs).map_or_else(
+            |logs| logs.lock().unwrap().clone(),
+            |logs| logs.into_inner().unwrap(),
+        ),
+        elapsed_ms: started.elapsed().as_millis() as u64,
+        output_bytes,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn snapshot() -> AgentSnapshot {
+        AgentSnapshot {
+            revision: 7,
+            columns: vec!["Title".into()],
+            rows: vec![vec!["Harvest".into()], vec!["Wharf".into()]],
+            selected_rows: vec![1],
+            diagnostics: Vec::new(),
+            files: vec![None, None],
+        }
+    }
+
+    #[test]
+    fn program_reads_only_the_snapshot_api() {
+        let output = run_agent_program(
+            "return function(q, args) return { revision=q:revision(), title=q:cell(args.row, 'Title') } end",
+            &serde_json::json!({"row": 1}), snapshot(),
+        ).unwrap();
+        assert_eq!(
+            output.value,
+            serde_json::json!({"revision": 7, "title": "Wharf"})
+        );
+    }
+
+    #[test]
+    fn ambient_io_is_absent() {
+        let error = run_agent_program(
+            "return function() return io.open('x') end",
+            &Json::Null,
+            snapshot(),
+        )
+        .unwrap_err();
+        assert!(error.detail.contains("io"));
+    }
+}
diff --git a/crates/plugin-host/src/lib.rs b/crates/plugin-host/src/lib.rs
index 99cdb8e5..30eed454 100644
--- a/crates/plugin-host/src/lib.rs
+++ b/crates/plugin-host/src/lib.rs
@@ -10,8 +10,13 @@
 //! other `.lua` file beside its `init.lua`, and nothing else — the host reads the folder, so the
 //! sandbox never gains a way to name a file itself.
 
+mod agent_program;
 mod plugin;
 
+pub use agent_program::{
+    AgentDiagnostic, AgentProgramError, AgentProgramOutput, AgentSnapshot, DEADLINE, MAX_OUTPUT,
+    MEMORY_LIMIT, run_agent_program, validate_agent_program,
+};
 pub use plugin::{Env, LuaPlugin, PERMISSION_NET, Writes};
 // So the Settings window can render a plugin's knobs without depending on `plugin-api` directly.
 pub use plugin_api::{SettingKind, SettingScope, SettingSpec};
diff --git a/crates/preview/src/lib.rs b/crates/preview/src/lib.rs
index bd121915..bd31a22b 100644
--- a/crates/preview/src/lib.rs
+++ b/crates/preview/src/lib.rs
@@ -355,9 +355,31 @@ pub fn search(path: &Path, needle: &str) -> Vec<Match> {
     pdf::search(path, needle).unwrap_or_default()
 }
 
+/// The same cached 512px preview the UI uses, encoded as PNG for a bounded agent attachment.
+/// The caller already resolved `path` through qrate's linked-files index; this function never
+/// exposes the path or original bytes.
+pub fn thumbnail_png(path: &Path, page: usize) -> Option<Vec<u8>> {
+    let pixels = thumbnail_pixels(path, CARD, page)?;
+    let mut encoded = std::io::Cursor::new(Vec::new());
+    image::DynamicImage::ImageRgba8(pixels)
+        .write_to(&mut encoded, image::ImageFormat::Png)
+        .ok()?;
+    Some(encoded.into_inner())
+}
+
 /// Decode `path`, shrink it to fit `max_edge`, and hand back something gpui can draw. Runs on a
 /// background thread; `None` for anything that won't decode, which the caller turns into the icon.
 fn render(path: &Path, max_edge: u32, page: usize) -> Option<Arc<RenderImage>> {
+    let mut bgra = thumbnail_pixels(path, max_edge, page)?;
+    // `RenderImage` is documented as BGRA and gpui only swaps inside its own decode path, so an
+    // image built by hand has to arrive already swapped or every preview draws blue-for-red.
+    for px in bgra.pixels_mut() {
+        px.0.swap(0, 2);
+    }
+    Some(Arc::new(RenderImage::new([Frame::new(bgra)])))
+}
+
+fn thumbnail_pixels(path: &Path, max_edge: u32, page: usize) -> Option<image::RgbaImage> {
     // A full-size rendering is not cached: it would be a second copy of the original file on disk,
     // and it is wanted once, while someone is looking at it.
     let key = (max_edge != FULL)
@@ -374,13 +396,7 @@ fn render(path: &Path, max_edge: u32, page: usize) -> Option<Arc<RenderImage>> {
         }
     };
 
-    // `RenderImage` is documented as BGRA and gpui only swaps inside its own decode path, so an
-    // image built by hand has to arrive already swapped or every preview draws blue-for-red.
-    let mut bgra = scaled;
-    for px in bgra.pixels_mut() {
-        px.0.swap(0, 2);
-    }
-    Some(Arc::new(RenderImage::new([Frame::new(bgra)])))
+    Some(scaled)
 }
 
 /// The ladder: each tier is asked in turn and the first one that produces pixels wins.
diff --git a/crates/preview/src/native.rs b/crates/preview/src/native.rs
index 9eeb3307..fa9911a2 100644
--- a/crates/preview/src/native.rs
+++ b/crates/preview/src/native.rs
@@ -170,13 +170,13 @@ mod windows_shell {
             }
 
             // A DIB is BGRA; the rest of the pipeline works in RGBA and swaps once at the end.
-            for px in bytes.chunks_exact_mut(4) {
+            for px in bytes.as_chunks_mut::<4>().0 {
                 px.swap(0, 2);
             }
             // Some thumbnail providers return an opaque image with the alpha byte left at zero,
             // which would otherwise draw as nothing at all.
-            if bytes.chunks_exact(4).all(|px| px[3] == 0) {
-                for px in bytes.chunks_exact_mut(4) {
+            if bytes.as_chunks::<4>().0.iter().all(|px| px[3] == 0) {
+                for px in bytes.as_chunks_mut::<4>().0 {
                     px[3] = 255;
                 }
             }
diff --git a/crates/table/Cargo.toml b/crates/table/Cargo.toml
index e9f82c13..1e804fd7 100644
--- a/crates/table/Cargo.toml
+++ b/crates/table/Cargo.toml
@@ -14,6 +14,10 @@ ai = { path = "../ai" }
 diagnostics = { path = "../diagnostics" }
 plugin-api = { path = "../plugin-api" }
 settings = { path = "../settings" }
+plugin-host = { path = "../plugin-host" }
+preview = { path = "../preview" }
+base64 = "0.22"
+sha2 = "0.10"
 
 [dev-dependencies]
 gpui = { workspace = true, features = ["test-support"] }
diff --git a/crates/table/src/agent.rs b/crates/table/src/agent.rs
index 89c6c001..31892a8f 100644
--- a/crates/table/src/agent.rs
+++ b/crates/table/src/agent.rs
@@ -1,13 +1,18 @@
 //! Live-table adapter for the external-agent contract: the reads, and the one draft-only write.
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::atomic::{AtomicU64, Ordering};
 
 use ai::agent::{
-    Column, Diagnostic, Field, Finding, ProjectSummary, Request, RequestError, Response, ResultSet,
-    Row, Severity, TableRevision,
+    Column, Count, DiagnosticSummary, FilterOp, Finding, MAX_RESULT_BYTES, Overview, ProgramAudit,
+    ProgramOutput, ProjectSummary, Query, QueryResult, QuerySource, Request, RequestError,
+    Response, ResultSet, Severity, TableRevision,
 };
+use base64::Engine as _;
 use diagnostics::{DATASET_MAIN, FixProviders, Location, Source};
-use gpui::{App, Global, SharedString};
+use gpui::{App, Global, SharedString, Task};
+use sha2::{Digest as _, Sha256};
 
 use crate::{QrateTableDelegate, TableStateHandle};
 
@@ -15,6 +20,15 @@ use crate::{QrateTableDelegate, TableStateHandle};
 /// re-run replaces its own drafts wholesale instead of stacking a second copy beside them.
 pub const AGENT_SOURCE: &str = "agent";
 
+#[derive(Default)]
+struct AgentProgram {
+    source: String,
+    version: u64,
+    hash: String,
+}
+impl Global for AgentProgram {}
+static PROGRAM_RUNS: AtomicU64 = AtomicU64::new(1);
+
 /// Answer one validated agent request from qrate's live table state.
 ///
 /// This deliberately reads the delegate rather than the `.qrate` database, so responses include
@@ -24,8 +38,12 @@ pub const AGENT_SOURCE: &str = "agent";
 pub fn respond_to_agent(request: Request, cx: &mut App) -> Result<Response, RequestError> {
     request.validate()?;
 
-    if let Request::StageFindings { revision, findings } = request {
-        return stage_findings(revision, findings, cx);
+    match request {
+        Request::StageFindings { revision, findings } => {
+            return stage_findings(revision, findings, cx);
+        }
+        Request::ProgramSave { source } => return save_program(source, cx),
+        _ => {}
     }
 
     let project = cx
@@ -40,7 +58,381 @@ pub fn respond_to_agent(request: Request, cx: &mut App) -> Result<Response, Requ
     let revision = TableRevision(delegate.values_generation());
 
     let result = match request {
-        Request::ProjectSummary => ResultSet::ProjectSummary(ProjectSummary {
+        Request::Overview => ResultSet::Overview(overview(project, delegate, cx)),
+        Request::Query(query) => ResultSet::Query(run_query(query, revision, delegate, cx)?),
+        Request::ProgramRun {
+            revision: judged,
+            args,
+        } => {
+            if judged != revision {
+                return Err(RequestError::StaleCursor);
+            }
+            let program = cx
+                .try_global::<AgentProgram>()
+                .ok_or(RequestError::ProgramUnavailable)?;
+            let (version, hash, source) = (
+                program.version,
+                program.hash.clone(),
+                program.source.clone(),
+            );
+            let snapshot = program_snapshot(revision, delegate, cx);
+            let output =
+                plugin_host::run_agent_program(&source, &args, snapshot).map_err(|error| {
+                    log::warn!("agent scratch program {hash} failed: {error}");
+                    if error.kind == "output_too_large" {
+                        RequestError::ProgramOutputTooLarge(error.detail)
+                    } else {
+                        RequestError::ProgramFailed(error.detail)
+                    }
+                })?;
+            let run_id = format!(
+                "{}-{}-{}",
+                &hash[..12],
+                revision.0,
+                PROGRAM_RUNS.fetch_add(1, Ordering::Relaxed)
+            );
+            let audit = ProgramAudit {
+                run_id,
+                revision,
+                version,
+                hash: hash.clone(),
+                elapsed_ms: output.elapsed_ms,
+                memory_limit_bytes: plugin_host::MEMORY_LIMIT,
+                deadline_ms: plugin_host::DEADLINE.as_millis() as u64,
+                output_limit_bytes: plugin_host::MAX_OUTPUT,
+                status: "success".into(),
+                output_bytes: output.output_bytes,
+            };
+            log::info!(
+                "agent program run {} revision {} version {} sha256 {} status success elapsed_ms {} output_bytes {}",
+                audit.run_id,
+                revision.0,
+                version,
+                hash,
+                audit.elapsed_ms,
+                audit.output_bytes
+            );
+            ResultSet::ProgramRun(ProgramOutput {
+                version,
+                hash,
+                value: output.value,
+                logs: output.logs,
+                elapsed_ms: output.elapsed_ms,
+                audit,
+            })
+        }
+        Request::Thumbnails { items } => {
+            let mut thumbnails = Vec::with_capacity(items.len());
+            for item in items {
+                let path = delegate
+                    .row_image(item.row)
+                    .ok_or(RequestError::ThumbnailUnavailable)?;
+                let bytes = preview::thumbnail_png(path, item.page)
+                    .ok_or(RequestError::ThumbnailUnavailable)?;
+                thumbnails.push(ai::agent::Thumbnail {
+                    row: item.row,
+                    page: item.page,
+                    media_type: "image/png".into(),
+                    data: base64::engine::general_purpose::STANDARD.encode(bytes),
+                });
+            }
+            ResultSet::Thumbnails { items: thumbnails }
+        }
+        Request::ProgramSave { .. } | Request::StageFindings { .. } => unreachable!(),
+    };
+
+    Ok(Response { revision, result })
+}
+
+/// Capture live GPUI-owned state now, then do bounded analysis away from the UI thread.
+pub fn respond_to_agent_async(
+    request: Request,
+    cx: &mut App,
+) -> Task<Result<Response, RequestError>> {
+    if let Err(error) = request.validate() {
+        return Task::ready(Err(error));
+    }
+    match request {
+        Request::ProgramSave { source } => {
+            let Some(state) = cx
+                .try_global::<TableStateHandle>()
+                .and_then(|handle| handle.0.upgrade())
+            else {
+                return Task::ready(Err(RequestError::TableUnavailable));
+            };
+            let revision = TableRevision(state.read(cx).delegate().values_generation());
+            let version = cx
+                .try_global::<AgentProgram>()
+                .map_or(1, |program| program.version + 1);
+            let background = cx.background_executor().spawn(async move {
+                plugin_host::validate_agent_program(&source)
+                    .map_err(|error| RequestError::ProgramCompileFailed(error.detail))?;
+                let hash = format!("{:x}", Sha256::digest(source.as_bytes()));
+                Ok::<_, RequestError>((source, hash))
+            });
+            cx.spawn(async move |cx| {
+                let (source, hash) = background.await?;
+                cx.update(|cx| {
+                    cx.set_global(AgentProgram {
+                        source,
+                        version,
+                        hash: hash.clone(),
+                    })
+                });
+                Ok(Response {
+                    revision,
+                    result: ResultSet::ProgramSaved { version, hash },
+                })
+            })
+        }
+        Request::Query(query) => {
+            let Some(project) = cx.try_global::<settings::project::CurrentProject>() else {
+                return Task::ready(Err(RequestError::ProjectUnavailable));
+            };
+            let Some(state) = cx
+                .try_global::<TableStateHandle>()
+                .and_then(|handle| handle.0.upgrade())
+            else {
+                return Task::ready(Err(RequestError::TableUnavailable));
+            };
+            let state = state.read(cx);
+            let delegate = state.delegate();
+            let revision = TableRevision(delegate.values_generation());
+            let diagnostic_source = matches!(query.source, QuerySource::Diagnostics { .. });
+            let schema = if diagnostic_source {
+                ["row", "column", "severity", "source", "message"]
+                    .into_iter()
+                    .map(str::to_string)
+                    .collect()
+            } else {
+                std::iter::once("row".into())
+                    .chain(project.data.headers.iter().cloned())
+                    .collect()
+            };
+            let records = match &query.source {
+                QuerySource::Diagnostics {
+                    severities,
+                    sources,
+                } => diagnostic_records(cx, severities, sources),
+                source => row_records(delegate, source),
+            };
+            cx.background_executor().spawn(async move {
+                run_query_records(query, revision, records, schema, diagnostic_source).map(
+                    |result| Response {
+                        revision,
+                        result: ResultSet::Query(result),
+                    },
+                )
+            })
+        }
+        Request::ProgramRun {
+            revision: judged,
+            args,
+        } => {
+            let Some(state) = cx
+                .try_global::<TableStateHandle>()
+                .and_then(|handle| handle.0.upgrade())
+            else {
+                return Task::ready(Err(RequestError::TableUnavailable));
+            };
+            let state = state.read(cx);
+            let delegate = state.delegate();
+            let revision = TableRevision(delegate.values_generation());
+            if judged != revision {
+                return Task::ready(Err(RequestError::StaleCursor));
+            }
+            let Some(program) = cx.try_global::<AgentProgram>() else {
+                return Task::ready(Err(RequestError::ProgramUnavailable));
+            };
+            let (version, hash, source) = (
+                program.version,
+                program.hash.clone(),
+                program.source.clone(),
+            );
+            let snapshot = program_snapshot(revision, delegate, cx);
+            cx.background_executor().spawn(async move {
+                let run_id = format!(
+                    "{}-{}-{}",
+                    &hash[..12],
+                    revision.0,
+                    PROGRAM_RUNS.fetch_add(1, Ordering::Relaxed)
+                );
+                let output = match plugin_host::run_agent_program(&source, &args, snapshot) {
+                    Ok(output) => output,
+                    Err(error) => {
+                        log::warn!(
+                            "agent program run {} revision {} version {} sha256 {} status {} elapsed_ms {} output_bytes 0 memory_limit_bytes {} deadline_ms {} output_limit_bytes {}",
+                            run_id, revision.0, version, hash, error.kind, error.elapsed_ms,
+                            plugin_host::MEMORY_LIMIT, plugin_host::DEADLINE.as_millis(), plugin_host::MAX_OUTPUT
+                        );
+                        return Err(if error.kind == "output_too_large" {
+                            RequestError::ProgramOutputTooLarge(error.detail)
+                        } else {
+                            RequestError::ProgramFailed(error.detail)
+                        });
+                    }
+                };
+                let audit = ProgramAudit {
+                    run_id,
+                    revision,
+                    version,
+                    hash: hash.clone(),
+                    elapsed_ms: output.elapsed_ms,
+                    memory_limit_bytes: plugin_host::MEMORY_LIMIT,
+                    deadline_ms: plugin_host::DEADLINE.as_millis() as u64,
+                    output_limit_bytes: plugin_host::MAX_OUTPUT,
+                    status: "success".into(),
+                    output_bytes: output.output_bytes,
+                };
+                log::info!(
+                    "agent program run {} revision {} version {} sha256 {} status success elapsed_ms {} output_bytes {} memory_limit_bytes {} deadline_ms {} output_limit_bytes {}",
+                    audit.run_id, revision.0, version, hash, audit.elapsed_ms, audit.output_bytes,
+                    audit.memory_limit_bytes, audit.deadline_ms, audit.output_limit_bytes
+                );
+                Ok(Response {
+                    revision,
+                    result: ResultSet::ProgramRun(ProgramOutput {
+                        version,
+                        hash,
+                        value: output.value,
+                        logs: output.logs,
+                        elapsed_ms: output.elapsed_ms,
+                        audit,
+                    }),
+                })
+            })
+        }
+        Request::Thumbnails { items } => {
+            let Some(state) = cx
+                .try_global::<TableStateHandle>()
+                .and_then(|handle| handle.0.upgrade())
+            else {
+                return Task::ready(Err(RequestError::TableUnavailable));
+            };
+            let state = state.read(cx);
+            let delegate = state.delegate();
+            let revision = TableRevision(delegate.values_generation());
+            let paths: Result<Vec<_>, _> = items
+                .into_iter()
+                .map(|item| {
+                    delegate
+                        .row_image(item.row)
+                        .map(|path| (item, path.to_path_buf()))
+                        .ok_or(RequestError::ThumbnailUnavailable)
+                })
+                .collect();
+            let Ok(paths) = paths else {
+                return Task::ready(Err(RequestError::ThumbnailUnavailable));
+            };
+            cx.background_executor().spawn(async move {
+                let mut thumbnails = Vec::with_capacity(paths.len());
+                for (item, path) in paths {
+                    let bytes = preview::thumbnail_png(&path, item.page)
+                        .ok_or(RequestError::ThumbnailUnavailable)?;
+                    thumbnails.push(ai::agent::Thumbnail {
+                        row: item.row,
+                        page: item.page,
+                        media_type: "image/png".into(),
+                        data: base64::engine::general_purpose::STANDARD.encode(bytes),
+                    });
+                }
+                Ok(Response {
+                    revision,
+                    result: ResultSet::Thumbnails { items: thumbnails },
+                })
+            })
+        }
+        other => Task::ready(respond_to_agent(other, cx)),
+    }
+}
+
+fn program_snapshot(
+    revision: TableRevision,
+    delegate: &QrateTableDelegate,
+    cx: &App,
+) -> plugin_host::AgentSnapshot {
+    let columns: Vec<SharedString> = delegate
+        .row_fields(0)
+        .into_iter()
+        .map(|(column, _)| column)
+        .collect();
+    let rows = (0..delegate.row_count())
+        .map(|row| {
+            delegate
+                .row_fields(row)
+                .into_iter()
+                .map(|(_, value)| value)
+                .collect()
+        })
+        .collect();
+    let diagnostics = diagnostics::Diagnostics::all(cx)
+        .iter()
+        .map(|diagnostic| plugin_host::AgentDiagnostic {
+            row: diagnostic.location.row,
+            column: diagnostic.location.column.as_ref().map(ToString::to_string),
+            severity: format!("{:?}", diagnostic.severity).to_lowercase(),
+            source: diagnostic.source.label().to_string(),
+            message: diagnostic.message.to_string(),
+        })
+        .collect();
+    plugin_host::AgentSnapshot {
+        revision: revision.0,
+        columns,
+        rows,
+        selected_rows: delegate.selected_source_rows(),
+        diagnostics,
+        files: (0..delegate.row_count())
+            .map(|row| delegate.row_image(row).map(ToOwned::to_owned))
+            .collect(),
+    }
+}
+
+fn save_program(source: String, cx: &mut App) -> Result<Response, RequestError> {
+    let state = cx
+        .try_global::<TableStateHandle>()
+        .and_then(|handle| handle.0.upgrade())
+        .ok_or(RequestError::TableUnavailable)?;
+    let revision = TableRevision(state.read(cx).delegate().values_generation());
+    plugin_host::validate_agent_program(&source).map_err(|error| {
+        log::warn!("agent scratch program was not activated: {error}");
+        RequestError::ProgramCompileFailed(error.detail)
+    })?;
+    let hash = format!("{:x}", Sha256::digest(source.as_bytes()));
+    let version = cx.try_global::<AgentProgram>().map_or(1, |p| p.version + 1);
+    cx.set_global(AgentProgram {
+        source,
+        version,
+        hash: hash.clone(),
+    });
+    Ok(Response {
+        revision,
+        result: ResultSet::ProgramSaved { version, hash },
+    })
+}
+
+fn overview(
+    project: &settings::project::CurrentProject,
+    delegate: &QrateTableDelegate,
+    cx: &App,
+) -> Overview {
+    let mut summary = DiagnosticSummary::default();
+    let mut sources = BTreeMap::<String, usize>::new();
+    for diagnostic in diagnostics::Diagnostics::all(cx).iter() {
+        match diagnostic.severity {
+            diagnostics::Severity::Error => summary.errors += 1,
+            diagnostics::Severity::Warning => summary.warnings += 1,
+            diagnostics::Severity::Note => summary.notes += 1,
+        }
+        *sources
+            .entry(diagnostic.source.label().to_string())
+            .or_default() += 1;
+    }
+    summary.sources = sources
+        .into_iter()
+        .map(|(value, count)| Count { value, count })
+        .collect();
+    Overview {
+        project: ProjectSummary {
             name: project.display_name(),
             row_count: delegate.row_count(),
             column_count: delegate.column_count(),
@@ -49,72 +441,324 @@ pub fn respond_to_agent(request: Request, cx: &mut App) -> Result<Response, Requ
                 .values
                 .get(settings::project::FILES_FOLDER_KEY)
                 .is_some_and(|value| !value.text().trim().is_empty()),
-        }),
-        Request::Columns => ResultSet::Columns {
-            columns: project
-                .data
-                .headers
-                .iter()
-                .map(|name| {
-                    let configured = project
-                        .data
-                        .columns
-                        .iter()
-                        .find(|column| column.name == *name);
-                    Column {
-                        name: name.clone(),
-                        data_type: configured
-                            .map_or_else(String::new, |column| column.data_type.clone()),
-                        notes: configured.map_or_else(String::new, |column| column.notes.clone()),
-                    }
-                })
-                .collect(),
-        },
-        Request::Rows { rows } => ResultSet::Rows {
-            rows: rows
-                .into_iter()
-                .filter(|&row| row < delegate.row_count())
-                .map(|row| row_from(delegate, row))
-                .collect(),
         },
-        Request::SearchRows { query, limit } => {
-            let query = query.to_lowercase();
-            ResultSet::Rows {
-                rows: (0..delegate.row_count())
-                    .filter(|&row| {
-                        delegate.row_fields(row).iter().any(|(column, value)| {
-                            column.to_lowercase().contains(&query)
-                                || value.to_lowercase().contains(&query)
-                        })
-                    })
-                    .take(limit)
-                    .map(|row| row_from(delegate, row))
-                    .collect(),
+        columns: project
+            .data
+            .headers
+            .iter()
+            .map(|name| {
+                let configured = project
+                    .data
+                    .columns
+                    .iter()
+                    .find(|column| column.name == *name);
+                Column {
+                    name: name.clone(),
+                    data_type: configured
+                        .map_or_else(String::new, |column| column.data_type.clone()),
+                    notes: configured.map_or_else(String::new, |column| column.notes.clone()),
+                }
+            })
+            .collect(),
+        selected_rows: delegate.selected_source_rows().len(),
+        diagnostics: summary,
+    }
+}
+
+type Record = HashMap<String, serde_json::Value>;
+
+fn run_query(
+    query: Query,
+    revision: TableRevision,
+    delegate: &QrateTableDelegate,
+    cx: &App,
+) -> Result<QueryResult, RequestError> {
+    let diagnostic_source = matches!(query.source, QuerySource::Diagnostics { .. });
+    let source_schema: Vec<String> = if diagnostic_source {
+        ["row", "column", "severity", "source", "message"]
+            .into_iter()
+            .map(str::to_string)
+            .collect()
+    } else {
+        std::iter::once("row".into())
+            .chain(delegate.dataset_snapshot().0)
+            .collect()
+    };
+    let records = match &query.source {
+        QuerySource::Diagnostics {
+            severities,
+            sources,
+        } => diagnostic_records(cx, severities, sources),
+        source => row_records(delegate, source),
+    };
+    run_query_records(query, revision, records, source_schema, diagnostic_source)
+}
+
+fn run_query_records(
+    query: Query,
+    revision: TableRevision,
+    mut records: Vec<Record>,
+    source_schema: Vec<String>,
+    diagnostic_source: bool,
+) -> Result<QueryResult, RequestError> {
+    let fingerprint = query_fingerprint(&query);
+    let offset = cursor_offset(query.cursor.as_deref(), revision, fingerprint)?;
+    if let QuerySource::Search { text: needle } = &query.source {
+        let needle = needle.to_lowercase();
+        records.retain(|record| {
+            record.iter().any(|(field, value)| {
+                field.to_lowercase().contains(&needle)
+                    || text(Some(value)).to_lowercase().contains(&needle)
+            })
+        });
+    }
+    records.retain(|record| {
+        query
+            .filters
+            .iter()
+            .all(|filter| filter_matches(record, filter))
+    });
+
+    let referenced = query
+        .select
+        .iter()
+        .chain(query.group_by.iter())
+        .chain(query.distinct.iter())
+        .chain(query.filters.iter().map(|filter| &filter.field))
+        .chain(query.order_by.iter().map(|order| &order.field));
+    if referenced
+        .into_iter()
+        .any(|field| !source_schema.contains(field))
+    {
+        return Err(RequestError::UnknownField);
+    }
+
+    let mut fields = if let Some(distinct) = &query.distinct {
+        let mut values = records
+            .iter()
+            .map(|record| text(record.get(distinct)))
+            .collect::<Vec<_>>();
+        values.sort();
+        values.dedup();
+        records = values
+            .into_iter()
+            .map(|value| Record::from([(distinct.clone(), value.into())]))
+            .collect();
+        vec![distinct.clone()]
+    } else if query.group_by.is_empty() {
+        if query.select.is_empty() {
+            if diagnostic_source {
+                ["row", "column", "severity", "source", "message"]
+                    .into_iter()
+                    .map(str::to_string)
+                    .collect()
+            } else {
+                source_schema.clone()
             }
+        } else {
+            query.select.clone()
         }
-        Request::Diagnostics => ResultSet::Diagnostics {
-            diagnostics: diagnostics::Diagnostics::all(cx)
+    } else {
+        let mut fields = query.group_by.clone();
+        fields.push("count".into());
+        let mut grouped = BTreeMap::<Vec<String>, usize>::new();
+        for record in &records {
+            let key = query
+                .group_by
                 .iter()
-                .map(|diagnostic| Diagnostic {
-                    row: diagnostic.location.row,
-                    column: diagnostic.location.column.as_ref().map(ToString::to_string),
-                    severity: match diagnostic.severity {
-                        diagnostics::Severity::Error => Severity::Error,
-                        diagnostics::Severity::Warning => Severity::Warning,
-                        diagnostics::Severity::Note => Severity::Note,
-                    },
-                    source: diagnostic.source.label().to_string(),
-                    message: diagnostic.message.to_string(),
-                })
-                .collect(),
-        },
-        Request::SelectedRows => ResultSet::SelectedRows {
-            rows: delegate.selected_source_rows(),
-        },
-        Request::StageFindings { .. } => unreachable!("staging returned above"),
+                .map(|field| text(record.get(field)))
+                .collect();
+            *grouped.entry(key).or_default() += 1;
+        }
+        records = grouped
+            .into_iter()
+            .map(|(key, count)| {
+                let mut record = Record::new();
+                for (field, value) in query.group_by.iter().zip(key) {
+                    record.insert(field.clone(), value.into());
+                }
+                record.insert("count".into(), count.into());
+                record
+            })
+            .collect();
+        fields
     };
+    if fields.is_empty() {
+        fields.push("row".into());
+    }
+    if let Some(order) = &query.order_by {
+        if !fields.contains(&order.field)
+            && (query.distinct.is_some() || !query.group_by.is_empty())
+        {
+            return Err(RequestError::UnknownField);
+        }
+        records.sort_by_key(|record| text(record.get(&order.field)).to_lowercase());
+        if order.descending {
+            records.reverse();
+        }
+    }
 
-    Ok(Response { revision, result })
+    let total = records.len();
+    if offset > total {
+        return Err(RequestError::InvalidCursor);
+    }
+    let mut items = Vec::new();
+    // Reserve the schema and bounded pagination metadata, then serialize each candidate once.
+    let mut used = serde_json::to_vec(&fields).map_or(MAX_RESULT_BYTES, |value| value.len()) + 512;
+    let end = total.min(offset + query.limit);
+    for record in &records[offset..end] {
+        let item: Vec<_> = fields
+            .iter()
+            .map(|field| {
+                record
+                    .get(field)
+                    .cloned()
+                    .unwrap_or(serde_json::Value::Null)
+            })
+            .collect();
+        let prospective = serde_json::to_vec(&item).map_or(MAX_RESULT_BYTES + 1, |v| v.len());
+        if used + prospective > MAX_RESULT_BYTES && items.is_empty() {
+            return Err(RequestError::ResultItemTooLarge);
+        }
+        if used + prospective > MAX_RESULT_BYTES {
+            break;
+        }
+        used += prospective;
+        items.push(item);
+    }
+    let consumed = offset + items.len();
+    let remaining = total.saturating_sub(consumed);
+    let next_cursor =
+        (remaining > 0).then(|| format!("{}:{fingerprint:016x}:{consumed}", revision.0));
+    Ok(QueryResult {
+        fields,
+        returned: items.len(),
+        items,
+        remaining,
+        truncated: remaining > 0,
+        next_cursor,
+    })
+}
+
+fn row_records(delegate: &QrateTableDelegate, source: &QuerySource) -> Vec<Record> {
+    let rows: Vec<usize> = match source {
+        QuerySource::AllRows | QuerySource::Search { .. } => (0..delegate.row_count()).collect(),
+        QuerySource::SelectedRows => delegate.selected_source_rows(),
+        QuerySource::Rows { rows } => rows.clone(),
+        QuerySource::Diagnostics { .. } => unreachable!(),
+    };
+    rows.into_iter()
+        .filter(|row| *row < delegate.row_count())
+        .map(|row| {
+            let mut record = Record::from([("row".into(), row.into())]);
+            for (column, value) in delegate.row_fields(row) {
+                record.insert(column.to_string(), value.to_string().into());
+            }
+            record
+        })
+        .collect()
+}
+
+fn diagnostic_records(cx: &App, severities: &[Severity], sources: &[String]) -> Vec<Record> {
+    diagnostics::Diagnostics::all(cx)
+        .iter()
+        .filter_map(|diagnostic| {
+            let severity = match diagnostic.severity {
+                diagnostics::Severity::Error => Severity::Error,
+                diagnostics::Severity::Warning => Severity::Warning,
+                diagnostics::Severity::Note => Severity::Note,
+            };
+            let source = diagnostic.source.label().to_string();
+            if (!severities.is_empty() && !severities.contains(&severity))
+                || (!sources.is_empty() && !sources.contains(&source))
+            {
+                return None;
+            }
+            Some(Record::from([
+                (
+                    "row".into(),
+                    diagnostic
+                        .location
+                        .row
+                        .map_or(serde_json::Value::Null, Into::into),
+                ),
+                (
+                    "column".into(),
+                    diagnostic
+                        .location
+                        .column
+                        .as_ref()
+                        .map_or(serde_json::Value::Null, |v| v.to_string().into()),
+                ),
+                (
+                    "severity".into(),
+                    format!("{severity:?}").to_lowercase().into(),
+                ),
+                ("source".into(), source.into()),
+                ("message".into(), diagnostic.message.to_string().into()),
+            ]))
+        })
+        .collect()
+}
+
+fn filter_matches(record: &Record, filter: &ai::agent::QueryFilter) -> bool {
+    let actual = text(record.get(&filter.field));
+    let expected = filter.value.as_deref().unwrap_or_default();
+    match filter.op {
+        FilterOp::Equals => actual.eq_ignore_ascii_case(expected),
+        FilterOp::NotEquals => !actual.eq_ignore_ascii_case(expected),
+        FilterOp::Contains => actual.to_lowercase().contains(&expected.to_lowercase()),
+        FilterOp::StartsWith => actual.to_lowercase().starts_with(&expected.to_lowercase()),
+        FilterOp::EndsWith => actual.to_lowercase().ends_with(&expected.to_lowercase()),
+        FilterOp::IsBlank => actual.is_empty(),
+        FilterOp::IsNotBlank => !actual.is_empty(),
+    }
+}
+
+fn text(value: Option<&serde_json::Value>) -> String {
+    match value {
+        Some(serde_json::Value::String(value)) => value.clone(),
+        Some(value) => value.to_string(),
+        None => String::new(),
+    }
+}
+
+fn query_fingerprint(query: &Query) -> u64 {
+    let mut normalized = query.clone();
+    normalized.cursor = None;
+    let mut hasher = DefaultHasher::new();
+    serde_json::to_string(&normalized)
+        .unwrap_or_default()
+        .hash(&mut hasher);
+    hasher.finish()
+}
+
+fn cursor_offset(
+    cursor: Option<&str>,
+    revision: TableRevision,
+    fingerprint: u64,
+) -> Result<usize, RequestError> {
+    let Some(cursor) = cursor else { return Ok(0) };
+    let mut parts = cursor.split(':');
+    let cursor_revision = parts
+        .next()
+        .and_then(|v| v.parse::<u64>().ok())
+        .ok_or(RequestError::InvalidCursor)?;
+    let cursor_fingerprint = parts
+        .next()
+        .and_then(|v| u64::from_str_radix(v, 16).ok())
+        .ok_or(RequestError::InvalidCursor)?;
+    let offset = parts
+        .next()
+        .and_then(|v| v.parse::<usize>().ok())
+        .ok_or(RequestError::InvalidCursor)?;
+    if parts.next().is_some() || cursor_fingerprint != fingerprint {
+        return Err(RequestError::InvalidCursor);
+    }
+    if cursor_revision != revision.0 {
+        return Err(RequestError::StaleCursor);
+    }
+    Ok(offset)
 }
 
 /// One agent proposal, held until somebody opens the Fixes menu on its cell.
@@ -252,24 +896,13 @@ fn offer_staged(location: &Location, text: &str, cx: &App) -> Vec<diagnostics::F
         .unwrap_or_default()
 }
 
-fn row_from(delegate: &QrateTableDelegate, row: usize) -> Row {
-    Row {
-        index: row,
-        fields: delegate
-            .row_fields(row)
-            .into_iter()
-            .map(|(column, value)| Field {
-                column: column.to_string(),
-                value: value.to_string(),
-            })
-            .collect(),
-    }
-}
-
 #[cfg(test)]
 mod tests {
     // Never `use super::*` here — see the note on `note.rs`'s test module.
-    use ai::agent::{Finding, ProjectSummary, Request, RequestError, ResultSet, TableRevision};
+    use ai::agent::{
+        Finding, Overview, ProjectSummary, Query, QuerySource, Request, RequestError, ResultSet,
+        TableRevision,
+    };
     use diagnostics::{DATASET_MAIN, Location};
     use gpui::TestAppContext;
 
@@ -306,7 +939,7 @@ mod tests {
     fn no_open_project_does_not_leak_table_state(cx: &mut TestAppContext) {
         cx.update(|cx| {
             assert_eq!(
-                respond_to_agent(Request::ProjectSummary, cx),
+                respond_to_agent(Request::Overview, cx),
                 Err(RequestError::ProjectUnavailable)
             );
         });
@@ -320,29 +953,33 @@ mod tests {
 
         cx.update(|cx| {
             assert!(matches!(
-                respond_to_agent(Request::ProjectSummary, cx)
-                    .unwrap()
-                    .result,
-                ResultSet::ProjectSummary(ProjectSummary {
-                    row_count: 3,
-                    column_count: 2,
+                respond_to_agent(Request::Overview, cx).unwrap().result,
+                ResultSet::Overview(Overview {
+                    project: ProjectSummary {
+                        row_count: 3,
+                        column_count: 2,
+                        ..
+                    },
                     ..
                 })
             ));
 
-            let ResultSet::Columns { columns } =
-                respond_to_agent(Request::Columns, cx).unwrap().result
+            let ResultSet::Overview(Overview { columns, .. }) =
+                respond_to_agent(Request::Overview, cx).unwrap().result
             else {
                 panic!("columns answered with another result set");
             };
             let names: Vec<_> = columns.iter().map(|column| column.name.as_str()).collect();
             assert_eq!(names, ["Title", "Medium"]);
 
-            let ResultSet::Rows { rows } = respond_to_agent(
-                Request::SearchRows {
-                    query: "video".into(),
+            let ResultSet::Query(page) = respond_to_agent(
+                Request::Query(Query {
+                    source: QuerySource::Search {
+                        text: "video".into(),
+                    },
                     limit: 10,
-                },
+                    ..Query::default()
+                }),
                 cx,
             )
             .unwrap()
@@ -350,9 +987,35 @@ mod tests {
             else {
                 panic!("search answered with another result set");
             };
-            assert_eq!(rows.len(), 1, "case-insensitive search matched cell values");
-            assert_eq!(rows[0].index, 1);
-            assert_eq!(rows[0].fields[0].value, "Wharf");
+            assert_eq!(
+                page.items.len(),
+                1,
+                "case-insensitive search matched cell values"
+            );
+            assert_eq!(page.items[0][0], 1);
+            assert_eq!(page.items[0][1], "Wharf");
+
+            let ResultSet::Query(distinct) = respond_to_agent(
+                Request::Query(Query {
+                    distinct: Some("Medium".into()),
+                    limit: 20,
+                    ..Query::default()
+                }),
+                cx,
+            )
+            .unwrap()
+            .result
+            else {
+                panic!("distinct answered with another result set")
+            };
+            assert_eq!(distinct.fields, ["Medium"]);
+            assert_eq!(
+                distinct.items,
+                [
+                    vec![serde_json::Value::from("Film")],
+                    vec![serde_json::Value::from("Video")]
+                ]
+            );
         });
     }
 
@@ -380,25 +1043,29 @@ mod tests {
         cx.update(|cx| {
             assert!(
                 matches!(
-                    respond_to_agent(Request::ProjectSummary, cx)
-                        .unwrap()
-                        .result,
-                    ResultSet::ProjectSummary(ProjectSummary { row_count: 3, .. })
+                    respond_to_agent(Request::Overview, cx).unwrap().result,
+                    ResultSet::Overview(Overview {
+                        project: ProjectSummary { row_count: 3, .. },
+                        ..
+                    })
                 ),
                 "the filter hid a row from the count"
             );
-            assert_eq!(
-                respond_to_agent(Request::SelectedRows, cx).unwrap().result,
-                ResultSet::SelectedRows { rows: vec![2] }
-            );
-
-            let ResultSet::Rows { rows } = respond_to_agent(Request::Rows { rows: vec![2] }, cx)
-                .unwrap()
-                .result
+            let ResultSet::Query(page) = respond_to_agent(
+                Request::Query(Query {
+                    source: QuerySource::SelectedRows,
+                    limit: 20,
+                    ..Query::default()
+                }),
+                cx,
+            )
+            .unwrap()
+            .result
             else {
                 panic!("rows answered with another result set");
             };
-            assert_eq!(rows[0].fields[0].value, "Cannery");
+            assert_eq!(page.items[0][0], 2);
+            assert_eq!(page.items[0][1], "Cannery");
         });
     }
 
@@ -463,14 +1130,21 @@ mod tests {
                 "the fix is offered against the judged text"
             );
             assert_eq!(offered[0].replacement, "Harvest, 1962");
-            let ResultSet::Rows { rows } = respond_to_agent(Request::Rows { rows: vec![0] }, cx)
-                .unwrap()
-                .result
+            let ResultSet::Query(page) = respond_to_agent(
+                Request::Query(Query {
+                    source: QuerySource::Rows { rows: vec![0] },
+                    limit: 20,
+                    ..Query::default()
+                }),
+                cx,
+            )
+            .unwrap()
+            .result
             else {
                 panic!("rows answered with another result set");
             };
             assert_eq!(
-                rows[0].fields[0].value, "Harvest",
+                page.items[0][1], "Harvest",
                 "staging a replacement must not write the cell"
             );
             assert!(
diff --git a/crates/table/src/lib.rs b/crates/table/src/lib.rs
index a12adf5b..32962b6e 100644
--- a/crates/table/src/lib.rs
+++ b/crates/table/src/lib.rs
@@ -20,7 +20,7 @@ mod panel;
 pub mod photos;
 mod row_index;
 
-pub use agent::{AGENT_SOURCE, respond_to_agent};
+pub use agent::{AGENT_SOURCE, respond_to_agent, respond_to_agent_async};
 pub use delegate::{QrateTableDelegate, Selection, TableChanged};
 /// The grid's right-click menu, so a gallery card can raise the same one a row does rather than
 /// growing a second, quietly diverging copy.
diff --git a/crates/workspace/src/lib.rs b/crates/workspace/src/lib.rs
index 51569e4e..41368ca7 100644
--- a/crates/workspace/src/lib.rs
+++ b/crates/workspace/src/lib.rs
@@ -12,7 +12,10 @@ pub use viewer::{CloseViewerLayer, Scope as ViewerScope, VIEWER_CONTEXT, open_vi
 pub use dock_button::DockToggleButton;
 pub use panel_registry::{BarSide, PANELS, PanelMeta, PanelRegistry, bar_side};
 /// `record` is how the agent bridge in `app` files an entry for the Agent panel to show.
-pub use panels::{AgentCall, AgentEntry, DETAILS_META, record as record_agent_call};
+pub use panels::{
+    AGENT_FONT_KEY, AGENT_FONT_SIZE_KEY, AGENT_FONT_SIZES, AgentCall, AgentEntry, DETAILS_META,
+    record as record_agent_call,
+};
 pub use views::{ShowView, VIEW_MODE_KEY, VIEWS_CONTEXT, ViewMode, show_view};
 
 use std::sync::Arc;
diff --git a/crates/workspace/src/panels/agent.rs b/crates/workspace/src/panels/agent.rs
index 31fc5a02..7e19acb7 100644
--- a/crates/workspace/src/panels/agent.rs
+++ b/crates/workspace/src/panels/agent.rs
@@ -1,6 +1,5 @@
-use std::cell::Cell;
 use std::collections::{HashSet, VecDeque};
-use std::rc::Rc;
+use std::sync::OnceLock;
 use std::time::{Duration, Instant};
 
 use gpui::prelude::FluentBuilder as _;
@@ -10,7 +9,9 @@ use gpui_component::menu::{ContextMenuExt as _, PopupMenuItem};
 use gpui_component::scroll::ScrollableElement as _;
 use gpui_component::tab::{Tab, TabBar};
 use gpui_component::{
-    ActiveTheme as _, Disableable as _, IconName, Sizable as _, button::Button, h_flex, v_flex,
+    ActiveTheme as _, IconName, Sizable as _,
+    button::{Button, ButtonVariants as _},
+    h_flex, v_flex,
 };
 
 use crate::BottomDockCrop;
@@ -147,6 +148,100 @@ const LIVE_TICK: Duration = Duration::from_millis(33);
 /// a newly started session; the cost of *not* slowing down is paid from launch until quit.
 const IDLE_TICK: Duration = Duration::from_millis(250);
 
+/// The face Pi's terminal draws with. Empty means "whichever monospace font this machine has".
+pub const AGENT_FONT_KEY: &str = "agent_terminal_font";
+/// The point size of that face, as text, because that is what a dropdown stores.
+pub const AGENT_FONT_SIZE_KEY: &str = "agent_terminal_font_size";
+/// Sizes offered in Settings. The stored half is what [`terminal_type`] parses.
+pub const AGENT_FONT_SIZES: &[(&str, &str)] = &[
+    ("12", "12 px"),
+    ("13", "13 px"),
+    ("14", "14 px (default)"),
+    ("16", "16 px"),
+    ("18", "18 px"),
+    ("20", "20 px"),
+];
+const DEFAULT_FONT_SIZE: f32 = 14.;
+/// 14px type on a 16px line is what the panel shipped with, and a terminal wants the leading to
+/// track the type rather than stay put.
+const LINE_SPACING: f32 = 8. / 7.;
+
+/// The grid the PTY is told to use has to be the grid gpui will actually lay the glyphs out on, so
+/// `cell` is measured from the resolved font rather than assumed.
+fn terminal_dimensions(
+    width: Pixels,
+    height: Pixels,
+    crop: Pixels,
+    cell: Size<Pixels>,
+) -> (usize, usize) {
+    (
+        ((f32::from(width) - 16.) / f32::from(cell.width))
+            .floor()
+            .max(2.) as usize,
+        ((f32::from(height) - 8. - f32::from(crop)) / f32::from(cell.height))
+            .floor()
+            .max(2.) as usize,
+    )
+}
+
+/// The face and size the terminal draws with, after the archivist's preferences.
+///
+/// Everything that has to agree about the grid — the styled text, the cell measurement, the line
+/// height — reads this one function, so a setting cannot move the glyphs without moving the
+/// columns they were counted into.
+fn terminal_type(cx: &App) -> (SharedString, Pixels, Pixels) {
+    let configured = settings::effective_text(AGENT_FONT_KEY, cx);
+    let family = if configured.trim().is_empty() {
+        installed_monospace(cx)
+    } else {
+        configured
+    };
+    let size = settings::effective_text(AGENT_FONT_SIZE_KEY, cx)
+        .parse()
+        .unwrap_or(DEFAULT_FONT_SIZE);
+    (family, px(size), px((size * LINE_SPACING).round()))
+}
+
+/// `monospace` is a CSS generic, not a family any OS ships — asking for it silently lands on the
+/// proportional UI font, and then no two columns of the grid are the same width. Name a real face.
+fn installed_monospace(cx: &App) -> SharedString {
+    static FAMILY: OnceLock<String> = OnceLock::new();
+    FAMILY
+        .get_or_init(|| {
+            let available: HashSet<String> =
+                cx.text_system().all_font_names().into_iter().collect();
+            [
+                "Cascadia Mono",
+                "Consolas",
+                "SF Mono",
+                "Menlo",
+                "DejaVu Sans Mono",
+                "Liberation Mono",
+                "Noto Sans Mono",
+            ]
+            .into_iter()
+            .find(|family| available.contains(*family))
+            .unwrap_or_else(|| {
+                log::warn!(
+                    "no known monospace font is installed; Pi's terminal grid will be ragged"
+                );
+                "monospace"
+            })
+            .to_owned()
+        })
+        .clone()
+        .into()
+}
+
+/// The advance one terminal cell occupies in the font the terminal draws with.
+fn terminal_advance(window: &Window, family: SharedString, font_size: Pixels) -> Pixels {
+    let font_id = window.text_system().resolve_font(&font(family));
+    window
+        .text_system()
+        .advance(font_id, font_size, 'm')
+        .map_or(px(8.), |advance| advance.width)
+}
+
 /// File one thing that happened. Called by the transport, which is the only thing that sees them.
 pub fn record(call: AgentCall, cx: &mut App) {
     let history = cx.default_global::<AgentHistory>();
@@ -178,7 +273,7 @@ pub struct AgentPanel {
     view: View,
     scroll: ScrollHandle,
     terminal: agent_runtime::AgentTerminal,
-    terminal_size: Rc<Cell<(usize, usize)>>,
+    terminal_size: (usize, usize),
     opened_auth_urls: HashSet<String>,
     _terminal_task: Task<()>,
     /// Refreshes on any entry. One `observe_global` and no re-binding — the history is plain data
@@ -186,12 +281,14 @@ pub struct AgentPanel {
     _sub: Subscription,
     /// Re-renders when the bottom dock opens or closes, so the padding tracks it.
     _crop_sub: Subscription,
+    /// Re-renders when the terminal's font or size changes, in either scope.
+    _settings_sub: Subscription,
+    _project_sub: Subscription,
 }
 
 impl AgentPanel {
     pub fn new(_window: &mut Window, cx: &mut Context<Self>) -> Self {
         let scroll = ScrollHandle::new();
-        let terminal_size = Rc::new(Cell::new((100, 32)));
         let terminal_task = cx.spawn(async move |this, cx| {
             // The panel is built with the window, not when it is first opened, so this loop runs
             // for the app's whole life. At 30 Hz unconditionally it woke the main thread thirty
@@ -220,7 +317,7 @@ impl AgentPanel {
             view: View::Terminal,
             scroll: scroll.clone(),
             terminal: Default::default(),
-            terminal_size,
+            terminal_size: (100, 32),
             opened_auth_urls: HashSet::new(),
             _terminal_task: terminal_task,
             // ponytail: follows the tail unconditionally. Only worth remembering whether the
@@ -230,6 +327,13 @@ impl AgentPanel {
                 cx.notify();
             }),
             _crop_sub: cx.observe_global::<BottomDockCrop>(|_this: &mut Self, cx| cx.notify()),
+            // Changing the font resizes the grid, so the PTY has to be told before Pi redraws.
+            // Both stores, because the type can be set user-wide or per project.
+            _settings_sub: cx
+                .observe_global::<settings::AppSettings>(|_this: &mut Self, cx| cx.notify()),
+            _project_sub: cx.observe_global::<settings::project::CurrentProject>(
+                |_this: &mut Self, cx| cx.notify(),
+            ),
         }
     }
 }
@@ -262,28 +366,70 @@ impl Panel for AgentPanel {
         _w: &mut Window,
         cx: &mut Context<Self>,
     ) -> Option<impl IntoElement> {
+        let panel = cx.entity().downgrade();
         Some(
-            TabBar::new("agent-view")
-                .segmented()
-                .small()
-                .selected_index(
-                    View::ALL
-                        .iter()
-                        .position(|view| *view == self.view)
-                        .unwrap_or(0),
+            h_flex()
+                .gap_1()
+                .items_center()
+                .child(
+                    TabBar::new("agent-view")
+                        .segmented()
+                        .small()
+                        .selected_index(
+                            View::ALL
+                                .iter()
+                                .position(|view| *view == self.view)
+                                .unwrap_or(0),
+                        )
+                        .on_click(cx.listener(|this, ix: &usize, _w, cx| {
+                            this.view = View::ALL[*ix];
+                            if this.view == View::Terminal && !this.terminal.is_running() {
+                                this.opened_auth_urls.clear();
+                                this.terminal.start(true, cx);
+                            }
+                            cx.notify();
+                        }))
+                        .children(
+                            View::ALL
+                                .into_iter()
+                                .map(|view| Tab::new().icon(view.icon()).label(view.label())),
+                        ),
                 )
-                .on_click(cx.listener(|this, ix: &usize, _w, cx| {
-                    this.view = View::ALL[*ix];
-                    if this.view == View::Terminal && !this.terminal.is_running() {
-                        this.opened_auth_urls.clear();
-                        this.terminal.start(true, cx);
-                    }
-                    cx.notify();
-                }))
-                .children(
-                    View::ALL
-                        .into_iter()
-                        .map(|view| Tab::new().icon(view.icon()).label(view.label())),
+                // Starting a session is the common action, so it is the click; stopping and
+                // restarting an existing one are rare, so they are the right-click.
+                .child(
+                    Button::new("agent-new-session")
+                        .icon(IconName::Plus)
+                        .ghost()
+                        .xsmall()
+                        .tooltip("New Pi session")
+                        .on_click(cx.listener(|this, _, _, cx| {
+                            this.opened_auth_urls.clear();
+                            this.view = View::Terminal;
+                            this.terminal.new_session(cx);
+                            cx.notify();
+                        }))
+                        .context_menu(move |menu, _window, _cx| {
+                            let (stop, restart) = (panel.clone(), panel.clone());
+                            menu.item(PopupMenuItem::new("Stop").on_click(move |_, _, cx| {
+                                stop.update(cx, |this, cx| {
+                                    this.terminal.stop();
+                                    cx.notify();
+                                })
+                                .ok();
+                            }))
+                            .item(
+                                PopupMenuItem::new("Restart").on_click(move |_, _, cx| {
+                                    restart
+                                        .update(cx, |this, cx| {
+                                            this.opened_auth_urls.clear();
+                                            this.terminal.start(true, cx);
+                                            cx.notify();
+                                        })
+                                        .ok();
+                                }),
+                            )
+                        }),
                 ),
         )
     }
@@ -337,8 +483,9 @@ impl Render for AgentPanel {
             )
         };
         let crop = cx.try_global::<BottomDockCrop>().map_or(px(0.), |c| c.0);
-        let (terminal_cols, terminal_lines) = self.terminal_size.get();
+        let (terminal_cols, terminal_lines) = self.terminal_size;
         self.terminal.resize(terminal_cols, terminal_lines);
+        let agent_panel = cx.entity().downgrade();
         let terminal_screen = self.terminal.screen(agent_runtime::TerminalPalette {
             foreground,
             background,
@@ -374,9 +521,8 @@ impl Render for AgentPanel {
         } else {
             StyledText::new(terminal_screen.text).with_highlights(terminal_screen.highlights)
         };
-        let terminal_backgrounds = terminal_screen.backgrounds;
+        let (terminal_font, terminal_font_size, terminal_line_height) = terminal_type(cx);
         let terminal_status = self.terminal.status().to_owned();
-        let terminal_running = self.terminal.is_running();
 
         v_flex()
             .size_full()
@@ -396,54 +542,27 @@ impl Render for AgentPanel {
                     v_flex()
                         .size_full()
                         .min_h_0()
-                        .child(
-                            h_flex()
-                                .flex_shrink_0()
-                                .gap_1()
-                                .items_center()
-                                .px_2()
-                                .py_1()
-                                .border_b_1()
-                                .border_color(border)
-                                .child(
-                                    Button::new("agent-new-session")
-                                        .label("New")
-                                        .xsmall()
-                                        .on_click(cx.listener(|this, _, _, cx| {
-                                            this.opened_auth_urls.clear();
-                                            this.terminal.start(false, cx);
-                                        })),
-                                )
-                                .child(
-                                    Button::new("agent-stop")
-                                        .label("Stop")
-                                        .xsmall()
-                                        .disabled(!terminal_running)
-                                        .on_click(cx.listener(|this, _, _, cx| {
-                                            this.terminal.stop();
-                                            cx.notify();
-                                        })),
-                                )
-                                .child(
-                                    Button::new("agent-restart")
-                                        .label("Restart")
-                                        .xsmall()
-                                        .on_click(cx.listener(|this, _, _, cx| {
-                                            this.opened_auth_urls.clear();
-                                            this.terminal.start(true, cx);
-                                        })),
-                                )
-                                .child(
-                                    div()
-                                        .ml_1()
-                                        .min_w_0()
-                                        .text_xs()
-                                        .text_color(muted)
-                                        .overflow_hidden()
-                                        .text_ellipsis()
-                                        .child(terminal_status),
-                                ),
-                        )
+                        .when(!terminal_status.is_empty(), |terminal| {
+                            terminal.child(
+                                h_flex()
+                                    .flex_shrink_0()
+                                    .gap_1()
+                                    .items_center()
+                                    .px_2()
+                                    .py_1()
+                                    .border_b_1()
+                                    .border_color(border)
+                                    .child(
+                                        div()
+                                            .min_w_0()
+                                            .text_xs()
+                                            .text_color(muted)
+                                            .overflow_hidden()
+                                            .text_ellipsis()
+                                            .child(terminal_status),
+                                    ),
+                            )
+                        })
                         .child(
                             div()
                                 .relative()
@@ -454,9 +573,9 @@ impl Render for AgentPanel {
                                 .px_2()
                                 .pt_1()
                                 .pb(px(4.) + crop)
-                                .text_sm()
-                                .line_height(px(16.))
-                                .font_family("monospace")
+                                .text_size(terminal_font_size)
+                                .line_height(terminal_line_height)
+                                .font_family(terminal_font.clone())
                                 .on_scroll_wheel(cx.listener(
                                     |this, event: &ScrollWheelEvent, _window, cx| {
                                         let lines = match event.delta {
@@ -471,50 +590,33 @@ impl Render for AgentPanel {
                                     },
                                 ))
                                 .child({
-                                    let measured = self.terminal_size.clone();
                                     canvas(
                                         move |bounds, window, cx| {
-                                            let next = (
-                                                ((f32::from(bounds.size.width) - 16.) / 8.)
-                                                    .floor()
-                                                    .max(2.) as usize,
-                                                ((f32::from(bounds.size.height)
-                                                    - 8.
-                                                    - f32::from(crop))
-                                                    / 16.)
-                                                    .floor()
-                                                    .max(2.) as usize,
-                                            );
-                                            if measured.replace(next) != next {
-                                                window.defer(cx, |window, _| window.refresh());
-                                            }
-                                        },
-                                        move |bounds, _, window, _| {
-                                            for background in &terminal_backgrounds {
-                                                let bounds = Bounds {
-                                                    origin: point(
-                                                        bounds.origin.x
-                                                            + px(8.)
-                                                            + px(
-                                                                8. * background.start_column
-                                                                    as f32,
-                                                            ),
-                                                        bounds.origin.y
-                                                            + px(4.)
-                                                            + px(16. * background.line as f32),
+                                            let next = terminal_dimensions(
+                                                bounds.size.width,
+                                                bounds.size.height,
+                                                crop,
+                                                size(
+                                                    terminal_advance(
+                                                        window,
+                                                        terminal_font.clone(),
+                                                        terminal_font_size,
                                                     ),
-                                                    size: size(
-                                                        px(
-                                                            8. * (background.end_column
-                                                                - background.start_column)
-                                                                as f32,
-                                                        ),
-                                                        px(16.),
-                                                    ),
-                                                };
-                                                window.paint_quad(fill(bounds, background.color));
-                                            }
+                                                    terminal_line_height,
+                                                ),
+                                            );
+                                            let Some(panel) = agent_panel.upgrade() else {
+                                                return;
+                                            };
+                                            panel.update(cx, |panel, cx| {
+                                                if panel.terminal_size != next {
+                                                    panel.terminal_size = next;
+                                                    panel.terminal.resize(next.0, next.1);
+                                                    cx.notify();
+                                                }
+                                            });
                                         },
+                                        |_, _, _, _| {},
                                     )
                                     .absolute()
                                     .top_0()
@@ -655,10 +757,11 @@ mod tests {
     // the built-in one and expand into itself.
     use std::time::Duration;
 
-    use gpui::TestAppContext;
+    use gpui::{SharedString, TestAppContext, px, size};
 
     use crate::panels::agent::{
-        AgentCall, AgentHistory, AgentPanel, Entry, HISTORY_LIMIT, View, record,
+        AGENT_FONT_KEY, AGENT_FONT_SIZE_KEY, AgentCall, AgentHistory, AgentPanel, Entry,
+        HISTORY_LIMIT, View, record, terminal_dimensions, terminal_type,
     };
 
     fn call(label: &str, entry: Entry) -> AgentCall {
@@ -676,12 +779,36 @@ mod tests {
         }
     }
 
+    /// An unset size falls back to the shipped 14/16, and a chosen one carries its leading with it
+    /// — a font that grew without its line height would overlap rows.
+    #[gpui::test]
+    fn the_configured_font_size_takes_its_line_height_with_it(cx: &mut TestAppContext) {
+        cx.update(|cx| {
+            cx.set_global(settings::AppSettings::default());
+            let (_, size, line) = terminal_type(cx);
+            assert_eq!((size, line), (px(14.), px(16.)));
+
+            let mut app = settings::AppSettings::default();
+            app.values
+                .insert(AGENT_FONT_SIZE_KEY.into(), settings::Val::Text("20".into()));
+            app.values.insert(
+                AGENT_FONT_KEY.into(),
+                settings::Val::Text("Consolas".into()),
+            );
+            cx.set_global(app);
+            let (family, size, line) = terminal_type(cx);
+            assert_eq!(family, SharedString::from("Consolas"));
+            assert_eq!((size, line), (px(20.), px(23.)));
+        });
+    }
+
     /// The panel must render before anything has called, and after — the empty state is what an
     /// archivist sees on every launch that never starts the bridge. All three entry kinds render,
     /// because each takes its own colour branch.
     #[gpui::test]
     fn renders_empty_and_every_entry_kind(cx: &mut TestAppContext) {
         cx.update(gpui_component::init);
+        cx.update(|cx| cx.set_global(settings::AppSettings::default()));
         cx.add_window_view(AgentPanel::new);
 
         cx.update(|cx| {
@@ -697,6 +824,7 @@ mod tests {
     #[gpui::test]
     fn each_view_renders_and_the_terminal_is_the_default(cx: &mut TestAppContext) {
         cx.update(gpui_component::init);
+        cx.update(|cx| cx.set_global(settings::AppSettings::default()));
         cx.update(|cx| record(call("rows", Entry::Answered), cx));
 
         let (panel, cx) = cx.add_window_view(AgentPanel::new);
@@ -749,4 +877,23 @@ mod tests {
             );
         });
     }
+
+    #[test]
+    fn terminal_dimensions_follow_the_measured_panel() {
+        let cell = size(px(8.), px(16.));
+        assert_eq!(
+            terminal_dimensions(px(816.), px(520.), px(0.), cell),
+            (100, 32)
+        );
+        assert_eq!(
+            terminal_dimensions(px(416.), px(264.), px(0.), cell),
+            (50, 16)
+        );
+        assert_eq!(terminal_dimensions(px(8.), px(8.), px(30.), cell), (2, 2));
+        // A wider cell than the old hardcoded 8px must yield fewer columns, not the same count.
+        assert_eq!(
+            terminal_dimensions(px(816.), px(520.), px(0.), size(px(8.4), px(16.))),
+            (95, 32)
+        );
+    }
 }
diff --git a/crates/workspace/src/panels/mod.rs b/crates/workspace/src/panels/mod.rs
index 2814f328..0ecb5992 100644
--- a/crates/workspace/src/panels/mod.rs
+++ b/crates/workspace/src/panels/mod.rs
@@ -5,5 +5,8 @@
 pub mod agent;
 pub mod details;
 
-pub use agent::{AGENT_META, AgentCall, AgentPanel, Entry as AgentEntry, record};
+pub use agent::{
+    AGENT_FONT_KEY, AGENT_FONT_SIZE_KEY, AGENT_FONT_SIZES, AGENT_META, AgentCall, AgentPanel,
+    Entry as AgentEntry, record,
+};
 pub use details::{DETAILS_META, DetailsPanel};