fix(plugins): don't block Find Usages on a failed plugin index batch (#2941 )

A plugin indexer batch that failed left its paths dirty forever, so every later Find Usages and rename failed with "Plugin indexes are still updating". Failed paths are now retried before the next usages or refactoring; usages go ahead with a "may be incomplete" note while an index is not up to date, and refactoring refuses with a message that says the index failed. Fixes #2913
fix(vcs): show changes in submodules whose git dir path exceeds MAX_PATH on Windows (#2902 )
2026-11-12 12:45:39 +02:00 · 2026-11-12 18:37:36 +08:00 · 2026-11-12 18:28:04 +08:00 · 2026-11-12 18:24:19 +08:00 · 2026-11-12 18:11:08 +08:00 · 2026-11-12 18:05:33 +08:00
@@ -0,0 +1,83 @@
+---
+name: eval
+description: Run the code-intelligence evals (`pnpm eval`), pick the worst failing scenario, diagnose it from the attached timeline, fix it, prove the fix with before / after numbers measured on this machine without regressing other scenarios, and open one PR. Use for "跑 eval", "优化 Java 代码智能", "eval 失败了", an issue labelled `eval`, or any change to Java navigation / usages / gutter markers that needs numbers.
+---
+
+# Eval loop: measure → fix → prove
+
+`pnpm eval` drives the production workbench in Chromium against a release `openidea-server`
+with real language servers (Java / jdtls for now) and judges every scenario on what the user
+sees: time to feedback, to the first result, to the final one, and the results themselves vs
+`evals/<language>/golden/`. One machine, before and after: numbers from different machines or
+from the nightly host don't judge a fix.
+
+## 0. Preconditions
+
+- `jdtls` on PATH and `JAVA_HOME` pointing at a JDK of the pinned `JAVA` version (projects
+  compile against it, and the JDK scenarios' results are in its sources; jdtls itself runs on
+  a newer installed JDK when it needs one). Pinned versions: `.github/tool-versions.env`. A
+  mismatch prints a warning and the report records both: say so in the PR when it applies.
+- First run builds `cargo build --release -p ide-host --bin openidea-server` and the perf
+  bundle; later iterations of a frontend-only change can use `--no-build` after a
+  `pnpm exec vite build --mode perf --outDir target/eval/dist`, a backend-only one after the
+  cargo build alone.
+- Run evals one at a time on this machine, with nothing heavy (builds, other evals) beside them.
+
+## 1. Baseline on the base commit
+
+```sh
+git switch --detach origin/main     # or the fix branch's merge base
+pnpm eval --save-baseline           # target/eval/baseline.json, this machine only
+```
+
+Read `target/eval/report.json`. Each verdict has `status`, `failures` (`kind`: `error`,
+`timeout`, `degraded`, `correctness`, `budget`, `golden`), `regressions`, `repro` (the command
+that reruns it) and, for failures, `diagnostics`: every command sent to the server in the
+slowest run's window (`cmd`, `start`, `ms`, `ok`, `bytes`, `items`), background tasks, server
+status changes and the server's log tail.
+
+## 2. Pick one problem
+
+Worst first: wrong or missing results and silent degradation, then timeouts, then the
+feedback budget (100 ms), then the slowest `completeMs`. One problem per PR. Say which
+scenario and metric, with its value vs budget.
+
+## 3. Diagnose from the timeline
+
+- Commands in the window ordered by `start`: what ran, what waited on what, what's serial
+  that could overlap, which request returned many `items`/`bytes`, which failed.
+- `feedbackMs` far above 16 ms with a fast `firstResultMs`: the UI shows nothing while it
+  waits (a hint or popup that only appears after a delay).
+- A task still running across the window: background work competing with the request.
+- For the frontend side rerun with `--trace` and open `target/eval/traces/<id>@<cond>.json`
+  in Chrome's Performance panel.
+- Name the cause at class level (as in `issue-fix`): it usually affects sibling scenarios.
+
+## 4. Fix, then rerun that scenario until it passes
+
+```sh
+pnpm eval --only <scenario> --condition <warm|cold>   # the verdict's `repro`
+```
+
+Correctness is enforced, so a faster answer that drops, truncates or serves stale results
+fails. Don't trade a budget miss for a degraded answer.
+
+## 5. Full suite, then PR
+
+```sh
+pnpm eval                           # vs the baseline from step 1: no new failures or regressions
+```
+
+PR per `issue-fix` §6, plus in **How it was verified** the before / after table from
+`target/eval/report.md` (both on this machine, commits named) and the tool versions line.
+
+## Guard rails
+
+- Goldens (`evals/*/golden/`) and budgets (`evals/budgets.json`) are human-owned. Don't edit
+  them, and don't run `--update-golden`, to make a scenario pass. When a golden looks wrong
+  (a tool bump, a deliberate behaviour change), stop and say so: a PR that touches them gets
+  `needs-decision`.
+- Don't weaken a scenario's checks or timeouts in `evals/<language>/scenarios.ts` to pass.
+  Adding a scenario (for a case the sweep or a user found) is fine; record its golden with
+  `--update-golden --only <id>` and ask a person to review the file.
+- A flaky timing (passes on rerun) is not a regression: rerun once before acting on one.
@@ -0,0 +1 @@
+../../.agents/skills/eval
@@ -0,0 +1,14 @@
+# Language server and tool versions for Language smoke (.github/workflows/language-smoke.yml)
+# and the code-intelligence evals (pnpm eval). Pinned so a release can't fail an unrelated
+# PR or move a golden result; bump them deliberately (and re-record goldens: pnpm eval --update-golden).
+VTSLS=0.3.0
+TYPESCRIPT=6.0.3
+BASEDPYRIGHT=1.40.2
+RUFF=0.17.3
+GOPLS=v0.24.0
+JDTLS=1.63.0-202610191322
+KOTLIN_LSP=263.7084.0
+GRADLE=8.14.3
+INTELEPHENSE=1.18.5
+# The JDK projects compile against (JAVA_HOME): library navigation opens its sources.
+JAVA=21
@@ -24,6 +24,7 @@ on:
      - "package.json"
      - "pnpm-lock.yaml"
      - ".github/workflows/language-smoke.yml"
+      - ".github/tool-versions.env"
  pull_request:
    branches: [main]
    types: [opened, synchronize, reopened]
@@ -38,16 +39,6 @@ concurrency:
 env:
  CARGO_TERM_COLOR: always
  CARGO_INCREMENTAL: 0
-  # Pinned so a server release can't fail an unrelated PR; bump them deliberately.
-  VTSLS: 0.3.0
-  TYPESCRIPT: 6.0.3
-  BASEDPYRIGHT: 1.40.2
-  RUFF: 0.17.3
-  GOPLS: v0.24.0
-  JDTLS: 1.63.0-202610191322
-  KOTLIN_LSP: 263.7084.0
-  GRADLE: 8.14.3
-  INTELEPHENSE: 1.18.5

 jobs:
  smoke:
@@ -61,6 +52,8 @@ jobs:
    timeout-minutes: 40
    steps:
      - uses: actions/checkout@v4
+      # Tool versions, shared with the evals (pnpm eval).
+      - run: grep -E '^[A-Z0-9_]+=' .github/tool-versions.env >> "$GITHUB_ENV"
      - uses: dtolnay/rust-toolchain@stable
        with: { components: "rust-analyzer, rustfmt" }
      # One build cache for every language; main saves it.
@@ -96,7 +89,7 @@ jobs:

      - if: matrix.language == 'java' || matrix.language == 'kotlin'
        uses: actions/setup-java@v4
-        with: { distribution: temurin, java-version: "21" }
+        with: { distribution: temurin, java-version: "${{ env.JAVA }}" }
      - if: matrix.language == 'java'
        run: |
          mkdir -p "$RUNNER_TEMP/jdtls"
@@ -40,6 +40,7 @@ CLAUDE.local.md
 .claude/skills/*
 !.claude/skills/verify
 !.claude/skills/issue-fix
+!.claude/skills/eval

 # Local reference screenshots
 pycharm-screenshots/
@@ -9,6 +9,7 @@
    "dist",
    "target",
    "src/bindings.ts",
+    "evals/*/golden",
    "src/editor/core/cm/angular/parser*.ts",
    "src/editor/core/cm/jsp/parser*.ts",
    "src/editor/core/cm/astro/parser*.ts",
@@ -24,8 +24,8 @@ Visual style: dark, low-contrast greys, one cool accent
  Plugin docs: `plugins/README.md`, `plugins/themes.md`.
 - `java/` — JDT LS extension bundles (`java-usages`, `kotlin-bridge`) and the patched
  Lombok agent, embedded into the app by `ide-sdk` / `ide-tools`.
- `e2e/` — Playwright UI tests (`pnpm e2e`). `fixtures/` — sample projects per language
-  for manual and automated testing.
+- `e2e/` — Playwright UI tests (`pnpm e2e`). `evals/` — code-intelligence evals (`pnpm eval`).
+  `fixtures/` — sample projects per language for manual and automated testing.
 - `docs/` — the Mintlify documentation site (`docs.json` navigation, `en/` and `zh/` pages);
  preview with `npx mint dev` in `docs/`.

@@ -59,9 +59,12 @@ Unclear product decisions → `needs-decision` and stop.

 ## Skills

+Do not use Impeccable or its related skills in this repository, including globally installed copies.
+
 Shared agent skills live in `.agents/skills/` (`.claude/skills/` links to them):
 `verify` drives the running app through Tauri MCP and captures screenshot evidence;
-`issue-fix` takes an issue from reproduction to one PR.
+`issue-fix` takes an issue from reproduction to one PR;
+`eval` takes a failing code-intelligence eval to a fix with before / after numbers.

 ## Commands

@@ -81,6 +84,11 @@ Shared agent skills live in `.agents/skills/` (`.claude/skills/` links to them):
  `pnpm gen:bindings`.
 - `pnpm perf` (UI interaction latency / frames / leaks), `pnpm perf:rust` (criterion
  benches vs budgets). Budgets: `perf/budgets.json`.
+- `pnpm eval [--only <id>] [--runs n] [--save-baseline]` — Java code intelligence (navigation,
+  usages, gutter markers) in the production workbench against a release `openidea-server` and
+  jdtls: time to feedback / first result / complete, results vs `evals/<language>/golden/`.
+  Report: `target/eval/report.json`. Goldens and `evals/budgets.json` are changed by people only.
+  Tool versions: `.github/tool-versions.env` (shared with Language smoke).
 - `pnpm release:prepare <version>` — the release commit's changes: app version everywhere,
  `Cargo.lock`, and the plugin API snapshot when `wit/` moved to a new version.

@@ -2637,7 +2637,6 @@ version = "1.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "337d46834ee672ab3e48caca2cb0c78cc174fb12b3a68d0d88f99a0519a5e36e"
 dependencies = [
- "generic-array 0.14.7",
 "rustversion",
 "typenum",
 ]
@@ -4145,7 +4144,7 @@ checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954"

 [[package]]
 name = "ide-agent"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "dunce",
 "encoding_rs",
@@ -4169,7 +4168,7 @@ dependencies = [

 [[package]]
 name = "ide-analysis"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ast-grep-core",
 "ide-core",
@@ -4183,7 +4182,7 @@ dependencies = [

 [[package]]
 name = "ide-android"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "serde",
@@ -4194,7 +4193,7 @@ dependencies = [

 [[package]]
 name = "ide-build"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "blake3",
 "chardetng",
@@ -4212,7 +4211,7 @@ dependencies = [

 [[package]]
 name = "ide-cli"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "dirs 6.0.0",
 "serde",
@@ -4226,7 +4225,7 @@ dependencies = [

 [[package]]
 name = "ide-clipboard"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "gtk",
 "objc2 0.6.5",
@@ -4236,7 +4235,7 @@ dependencies = [

 [[package]]
 name = "ide-core"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "dunce",
 "encoding_rs",
@@ -4253,7 +4252,7 @@ dependencies = [

 [[package]]
 name = "ide-coverage"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "ide-run",
@@ -4267,7 +4266,7 @@ dependencies = [

 [[package]]
 name = "ide-dap"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "futures",
 "ide-core",
@@ -4293,7 +4292,7 @@ dependencies = [

 [[package]]
 name = "ide-db"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "bb8",
 "blake3",
@@ -4303,6 +4302,7 @@ dependencies = [
 "ide-core",
 "ide-http",
 "ide-ssh",
+ "notify",
 "oracledb",
 "parking_lot",
 "reqwest",
@@ -4323,7 +4323,7 @@ dependencies = [

 [[package]]
 name = "ide-docker"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "serde",
@@ -4336,7 +4336,7 @@ dependencies = [

 [[package]]
 name = "ide-extension"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "base64 0.22.1",
 "blake3",
@@ -4357,7 +4357,7 @@ dependencies = [

 [[package]]
 name = "ide-fonts"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "objc2-core-foundation",
 "objc2-core-text",
@@ -4369,7 +4369,7 @@ dependencies = [

 [[package]]
 name = "ide-github"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "ide-vcs",
@@ -4382,7 +4382,7 @@ dependencies = [

 [[package]]
 name = "ide-heap"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "criterion",
 "ide-core",
@@ -4400,7 +4400,7 @@ dependencies = [

 [[package]]
 name = "ide-host"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "base64 0.22.1",
 "blake3",
@@ -4459,6 +4459,7 @@ dependencies = [
 "specta-typescript",
 "tempfile",
 "tokio",
+ "tokio-tungstenite",
 "toml 0.9.12+spec-1.1.0",
 "tracing",
 "tracing-subscriber",
@@ -4469,7 +4470,7 @@ dependencies = [

 [[package]]
 name = "ide-http"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "ide-vfs",
@@ -4488,7 +4489,7 @@ dependencies = [

 [[package]]
 name = "ide-index"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "blake3",
 "criterion",
@@ -4510,7 +4511,7 @@ dependencies = [

 [[package]]
 name = "ide-jdwp"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-jvm-inspect",
 "proptest",
@@ -4522,7 +4523,7 @@ dependencies = [

 [[package]]
 name = "ide-jsonrpc"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "encoding_rs",
 "futures",
@@ -4539,7 +4540,7 @@ dependencies = [

 [[package]]
 name = "ide-jvm-attach"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "libc",
 "tempfile",
@@ -4548,7 +4549,7 @@ dependencies = [

 [[package]]
 name = "ide-jvm-inspect"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "futures",
 "serde",
@@ -4559,7 +4560,7 @@ dependencies = [

 [[package]]
 name = "ide-language"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "blake3",
 "criterion",
@@ -4615,7 +4616,7 @@ dependencies = [

 [[package]]
 name = "ide-lsp"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "dunce",
 "encoding_rs",
@@ -4643,7 +4644,7 @@ dependencies = [

 [[package]]
 name = "ide-mcp"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "bytes",
 "dunce",
@@ -4662,7 +4663,7 @@ dependencies = [

 [[package]]
 name = "ide-notebook"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "futures",
 "ide-core",
@@ -4685,7 +4686,7 @@ dependencies = [

 [[package]]
 name = "ide-plugin"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "blake3",
 "dunce",
@@ -4708,7 +4709,7 @@ dependencies = [

 [[package]]
 name = "ide-profiler"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "ide-jvm-attach",
@@ -4722,7 +4723,7 @@ dependencies = [

 [[package]]
 name = "ide-remote"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "dunce",
 "globset",
@@ -4741,7 +4742,7 @@ dependencies = [

 [[package]]
 name = "ide-run"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "dunce",
 "encoding_rs",
@@ -4765,7 +4766,7 @@ dependencies = [

 [[package]]
 name = "ide-scaffold"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "ide-tools",
@@ -4781,7 +4782,7 @@ dependencies = [

 [[package]]
 name = "ide-sdk"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "dunce",
 "ide-core",
@@ -4802,7 +4803,7 @@ dependencies = [

 [[package]]
 name = "ide-search"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "criterion",
 "encoding_rs",
@@ -4823,7 +4824,7 @@ dependencies = [

 [[package]]
 name = "ide-spell"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "parking_lot",
@@ -4836,7 +4837,7 @@ dependencies = [

 [[package]]
 name = "ide-ssh"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "dirs 6.0.0",
 "flate2",
@@ -4861,7 +4862,7 @@ dependencies = [

 [[package]]
 name = "ide-syntax"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "ide-language",
@@ -4876,7 +4877,7 @@ dependencies = [

 [[package]]
 name = "ide-terminal"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "ide-core",
 "libc",
@@ -4889,7 +4890,7 @@ dependencies = [

 [[package]]
 name = "ide-text"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "criterion",
 "ide-core",
@@ -4902,7 +4903,7 @@ dependencies = [

 [[package]]
 name = "ide-tools"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "flate2",
 "hex",
@@ -4923,7 +4924,7 @@ dependencies = [

 [[package]]
 name = "ide-vcs"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "blake3",
 "dunce",
@@ -4944,7 +4945,7 @@ dependencies = [

 [[package]]
 name = "ide-vfs"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "chardetng",
 "dunce",
@@ -4968,7 +4969,7 @@ dependencies = [

 [[package]]
 name = "ide-web"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "dirs 6.0.0",
 "ide-core",
@@ -4984,7 +4985,7 @@ dependencies = [

 [[package]]
 name = "ide-workspace"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "blake3",
 "dunce",
@@ -6585,7 +6586,7 @@ dependencies = [

 [[package]]
 name = "openidea"
-version = "0.1.42"
+version = "0.1.43"
 dependencies = [
 "base64 0.22.1",
 "blake3",
@@ -7710,9 +7711,9 @@ dependencies = [

 [[package]]
 name = "russh"
-version = "0.64.1"
+version = "0.65.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba61e87b9ec9a39a59a6bbed4c0b8ff7fb07073405b24a767a1f2c8560ac8f58"
+checksum = "94be3d86c71b9ed220a675c405e35c7a02565a009bf8da2e9cc13a700320a5c5"
 dependencies = [
 "aes",
 "bitflags 2.13.2",
@@ -9867,7 +9868,10 @@ checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857"
 dependencies = [
 "futures-util",
 "log",
+ "rustls",
+ "rustls-pki-types",
 "tokio",
+ "tokio-rustls",
 "tungstenite",
 ]

@@ -10561,6 +10565,8 @@ dependencies = [
 "httparse",
 "log",
 "rand 0.9.5",
+ "rustls",
+ "rustls-pki-types",
 "sha1 0.10.7",
 "thiserror 2.0.21",
 "utf-8",
@@ -5,7 +5,7 @@ members = ["crates/*", "src-tauri"]
 exclude = ["sdk", "plugins"]

 [workspace.package]
-version = "0.1.42"
+version = "0.1.43"
 edition = "2024"
 rust-version = "1.96"
 license = "Apache-2.0"
@@ -68,6 +68,7 @@ parking_lot = "0.12"
 rayon = "1"
 dashmap = "6"
 futures = "0.3"
+tokio-tungstenite = { version = "0.28", default-features = false, features = ["connect", "__rustls-tls"] }

 # errors & diagnostics
 thiserror = "2"
@@ -1368,12 +1368,20 @@ impl AgentManager {
        let restamped = blocking(move || {
            let changes = at.changes(&before)?;
            let rev = at.rev()?;
+            let sources = match rev.as_deref() {
+                Some(rev) => worktree::rename_sources(at.cwd(), rev)?,
+                None => Default::default(),
+            };
            let mut restamped = Vec::new();
            for change in changes.iter().filter(|c| rel_paths.as_ref().is_none_or(|p| p.contains(&c.rel_path))) {
-                worktree::restore(at.cwd(), rev.as_deref(), change, &before, saved.as_deref())?;
-                // Back to the user's own edits: not a change of the task any more.
-                if before.contains_key(&change.rel_path) {
-                    restamped.push((change.rel_path.clone(), worktree::stamp(&change.path.0)));
+                // A rename goes back to its old path too; the new one is taken away first, so a
+                // case-only rename on a case-insensitive disk ends as the old file.
+                for change in std::iter::once(change).chain(sources.get(&change.rel_path)) {
+                    worktree::restore(at.cwd(), rev.as_deref(), change, &before, saved.as_deref())?;
+                    // Back to the user's own edits: not a change of the task any more.
+                    if before.contains_key(&change.rel_path) {
+                        restamped.push((change.rel_path.clone(), worktree::stamp(&change.path.0)));
+                    }
                }
            }
            Ok(restamped)
@@ -492,7 +492,7 @@ for line in sys.stdin:
        }
        let concurrent = async {
            if outcome == "session_report" {
-                tokio::time::timeout(Duration::from_secs(2), async {
+                tokio::time::timeout(Duration::from_secs(10), async {
                    while !root.join("initialized").exists() {
                        tokio::time::sleep(Duration::from_millis(10)).await;
                    }
@@ -564,7 +564,7 @@ for line in sys.stdin:
            };
            assert_eq!(subtypes, expected, "{outcome}");
            let pid = std::fs::read_to_string(root.join("pid")).unwrap();
-            tokio::time::timeout(Duration::from_secs(2), async {
+            tokio::time::timeout(Duration::from_secs(10), async {
                while std::process::Command::new("kill").args(["-0", pid.trim()]).output().unwrap().status.success() {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
@@ -3055,6 +3055,33 @@ async fn commits_made_during_a_local_task_are_kept_by_its_rollback() {
    }
 }

+#[tokio::test(flavor = "multi_thread")]
+async fn rolling_back_a_rename_brings_the_old_file_back() {
+    for eol in EOLS {
+        let repo = Checkout::new(eol);
+        repo.write("Old.java", "class Old {}\n");
+        repo.write("Case.java", "class Case {}\n");
+        repo.commit("init");
+        let (mgr, task) = repo.task().await;
+        repo.git(&["mv", "Old.java", "New.java"]);
+        repo.git(&["mv", "Case.java", "CASE.java"]);
+        // Renames list their new path only.
+        let listed = changed(&mgr, &task.id).await;
+        assert_eq!(listed, ["CASE.java", "New.java"], "{eol:?}");
+
+        mgr.rollback_changes(&task.id, Some(listed)).await.unwrap();
+        assert_eq!(repo.read("Old.java"), "class Old {}\n", "{eol:?}");
+        assert!(!repo.dir.join("New.java").exists(), "{eol:?}");
+        assert_eq!(repo.read("Case.java"), "class Case {}\n", "{eol:?}");
+        let names: Vec<String> = std::fs::read_dir(&repo.dir)
+            .unwrap()
+            .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
+            .filter(|n| n.ends_with(".java"))
+            .collect();
+        assert!(!names.contains(&"CASE.java".to_owned()), "{eol:?}: {names:?}");
+    }
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn a_local_task_in_a_repository_without_commits_lists_and_rolls_back_its_changes() {
    for eol in EOLS {
@@ -124,11 +124,41 @@ pub(crate) fn changes(cwd: &Path, base: &str) -> IdeResult<Vec<AgentChange>> {
    Ok(out)
 }

+/// Files renamed since `rev`, by their new repository-relative path: the old path, as a deletion
+/// to roll back along with the rename.
+pub(crate) fn rename_sources(cwd: &Path, rev: &str) -> IdeResult<BTreeMap<String, AgentChange>> {
+    let Some(repo) = discover(cwd)? else { return Ok(BTreeMap::new()) };
+    let root = repo.root().to_path_buf();
+    Ok(repo
+        .diff_with_worktree(rev)?
+        .into_iter()
+        .filter_map(|f| {
+            let old = f.old_path.filter(|old| *old != f.path)?;
+            let source = AgentChange {
+                path: IpcPath(root.join(&old)),
+                rel_path: old,
+                kind: ChangeKind::Deleted,
+                added: 0,
+                removed: 0,
+            };
+            Some((f.path, source))
+        })
+        .collect())
+}
+
 fn added(path: IpcPath, rel_path: String) -> AgentChange {
    let added = std::fs::read_to_string(&*path).map_or(0, |t| t.lines().count() as u32);
    AgentChange { path, rel_path, kind: ChangeKind::Added, added, removed: 0 }
 }

+/// Whether two paths name one file (a case-insensitive disk resolving both spellings).
+fn same_file(a: &Path, b: &Path) -> bool {
+    match (std::fs::canonicalize(a), std::fs::canonicalize(b)) {
+        (Ok(a), Ok(b)) => a == b,
+        _ => false,
+    }
+}
+
 fn crlf_to_lf(bytes: &[u8]) -> Vec<u8> {
    let mut out = Vec::with_capacity(bytes.len());
    for (i, &b) in bytes.iter().enumerate() {
@@ -245,34 +275,70 @@ pub(crate) fn accept(
    changes: &[AgentChange],
 ) -> IdeResult<crate::AgentAcceptOutcome> {
    let repo = repo_at(cwd)?;
-    let mut accepted = Vec::with_capacity(changes.len());
-    let mut skipped = Vec::new();
-    for change in changes {
-        let local = root.join(&change.rel_path);
-        let base_bytes = repo.file_bytes_at(&change.path.0, Some(base))?;
-        let current = std::fs::read(&change.path.0).ok();
+    // Change paths are repository-relative, and the task's project may be a folder inside it.
+    let local_root = discover(root)?.map_or_else(|| root.to_path_buf(), |r| r.root().to_path_buf());
+    let renamed_from: BTreeMap<String, String> =
+        repo.diff_with_worktree(base)?.into_iter().filter_map(|f| Some((f.path, f.old_path?))).collect();
+    // Brings one path of the checkout to the task's version, unless the checkout changed it itself.
+    let apply = |rel: &str| -> IdeResult<bool> {
+        let local = local_root.join(rel);
+        let in_task = repo.root().join(rel);
+        let base_bytes = repo.file_bytes_at(&in_task, Some(base))?;
+        let current = std::fs::read(&in_task).ok();
        let local_bytes = std::fs::read(&local).ok();
        if local_bytes == current {
-            accepted.push(change.rel_path.clone());
-        } else if local_bytes == base_bytes {
-            match current {
-                Some(bytes) => {
-                    if let Some(parent) = local.parent() {
-                        std::fs::create_dir_all(parent)?;
-                    }
-                    std::fs::write(&local, bytes)?;
+            return Ok(true);
+        }
+        // The checkout may hold the committed LF text with CRLF line ends (`core.autocrlf`).
+        let unchanged = match (&local_bytes, &base_bytes) {
+            (Some(now), Some(at_base)) => now == at_base || crlf_to_lf(now) == *at_base,
+            (now, at_base) => now == at_base,
+        };
+        if !unchanged {
+            return Ok(false);
+        }
+        match current {
+            Some(bytes) => {
+                if let Some(parent) = local.parent() {
+                    std::fs::create_dir_all(parent)?;
                }
-                None => match std::fs::remove_file(&local) {
-                    Err(e) if e.kind() != std::io::ErrorKind::NotFound => return Err(e.into()),
-                    _ => {}
-                },
+                std::fs::write(&local, bytes)?;
            }
-            accepted.push(change.rel_path.clone());
-        } else {
-            skipped.push(crate::AgentAcceptSkipped {
-                path: change.rel_path.clone(),
-                reason: "the checkout has changes of its own to it; accept it by hand".to_owned(),
-            });
+            None => match std::fs::remove_file(&local) {
+                Err(e) if e.kind() != std::io::ErrorKind::NotFound => return Err(e.into()),
+                _ => {}
+            },
+        }
+        Ok(true)
+    };
+    let mut accepted = Vec::with_capacity(changes.len());
+    let mut skipped = Vec::new();
+    let skip = |path: &str| crate::AgentAcceptSkipped {
+        path: path.to_owned(),
+        reason: "the checkout has changes of its own to it; accept it by hand".to_owned(),
+    };
+    for change in changes {
+        // A case-only rename on a case-insensitive disk: both names reach the same file, so
+        // rename it in place before bringing it to the task's version.
+        if let Some(old) = renamed_from.get(&change.rel_path)
+            && old != &change.rel_path
+            && old.to_lowercase() == change.rel_path.to_lowercase()
+            && local_root.join(old).exists()
+            && same_file(&local_root.join(old), &local_root.join(&change.rel_path))
+        {
+            std::fs::rename(local_root.join(old), local_root.join(&change.rel_path))?;
+        }
+        if !apply(&change.rel_path)? {
+            skipped.push(skip(&change.rel_path));
+            continue;
+        }
+        accepted.push(change.rel_path.clone());
+        // A rename also takes the file away from its old path (unless only its case changed).
+        if let Some(old) = renamed_from.get(&change.rel_path)
+            && old.to_lowercase() != change.rel_path.to_lowercase()
+            && !apply(old)?
+        {
+            skipped.push(skip(old));
        }
    }
    Ok(crate::AgentAcceptOutcome { accepted, skipped })
@@ -580,6 +646,57 @@ mod tests {
        without_untouched(changes(cwd, base).unwrap(), &BTreeMap::new())
    }

+    #[test]
+    fn accept_moves_a_renamed_file_and_maps_paths_from_a_project_folder() {
+        let (_tmp, repo, created) = accept_repo();
+        std::fs::create_dir(created.path.join("app")).unwrap();
+        git(&created.path, &["mv", "a.txt", "app/moved.txt"]);
+        std::fs::write(created.path.join("app/new.txt"), "n\n").unwrap();
+        git(&created.path, &["add", "."]);
+        git(&created.path, &["-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "-m", "move"]);
+
+        // The task's project is the repository's `app` folder.
+        let project = repo.join("app");
+        std::fs::create_dir(&project).unwrap();
+        let changes = listed(&created.path, &created.base_commit);
+        let outcome = accept(&project, &created.path, &created.base_commit, &changes).unwrap();
+        assert!(outcome.skipped.is_empty(), "{:?}", outcome.skipped);
+        assert_eq!(std::fs::read_to_string(repo.join("app/moved.txt")).unwrap(), "one\ntwo\n");
+        assert_eq!(std::fs::read_to_string(repo.join("app/new.txt")).unwrap(), "n\n");
+        assert!(!repo.join("a.txt").exists(), "the rename's old path is gone");
+        assert!(!repo.join("app/app").exists());
+    }
+
+    #[test]
+    fn accept_treats_a_crlf_checkout_of_the_base_as_unchanged() {
+        let (_tmp, repo, created) = accept_repo();
+        // The checkout holds the committed text with CRLF line ends, as core.autocrlf leaves it.
+        std::fs::write(repo.join("a.txt"), "one\r\ntwo\r\n").unwrap();
+        std::fs::write(created.path.join("a.txt"), "one\nTWO\n").unwrap();
+        git(&created.path, &["mv", "gone.txt", "kept.txt"]);
+        std::fs::write(repo.join("gone.txt"), "bye\r\n").unwrap();
+        let changes = listed(&created.path, &created.base_commit);
+        let outcome = accept(&repo, &created.path, &created.base_commit, &changes).unwrap();
+        assert!(outcome.skipped.is_empty(), "{:?}", outcome.skipped);
+        assert_eq!(std::fs::read_to_string(repo.join("a.txt")).unwrap(), "one\nTWO\n");
+        assert!(repo.join("kept.txt").exists() && !repo.join("gone.txt").exists());
+    }
+
+    #[test]
+    fn accept_applies_a_case_only_rename() {
+        let (_tmp, repo, created) = accept_repo();
+        git(&created.path, &["mv", "a.txt", "A.txt"]);
+        let changes = listed(&created.path, &created.base_commit);
+        let outcome = accept(&repo, &created.path, &created.base_commit, &changes).unwrap();
+        assert!(outcome.skipped.is_empty(), "{:?}", outcome.skipped);
+        let names: Vec<_> = std::fs::read_dir(&repo)
+            .unwrap()
+            .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
+            .filter(|n| n.eq_ignore_ascii_case("a.txt"))
+            .collect();
+        assert_eq!(names, ["A.txt"]);
+    }
+
    #[test]
    fn accept_copies_new_modified_and_deleted_files_into_the_checkout() {
        let (_tmp, repo, created) = accept_repo();
@@ -1203,6 +1203,8 @@ pub struct MavenSettings {
    pub threads: String,
    /// Active Maven profiles (`-P`), comma-separated; empty leaves profiles to the POM/settings.
    pub active_profiles: String,
+    /// Delegate IDE build/run actions to Maven: launches build with Maven (`javac`) instead of jdtls (ECJ).
+    pub delegate_build: bool,
 }

 impl MavenSettings {
@@ -1707,10 +1709,12 @@ mod maven_linkage_tests {
    fn maven_runner_jre_defaults_and_round_trips() {
        let old: MavenSettings = serde_json::from_str(r#"{"home":"/opt/maven"}"#).unwrap();
        assert!(old.runner_jre.is_empty());
+        assert!(!old.delegate_build);
        assert!(MavenSettings::default().runner_jre.is_empty());
-        let settings = MavenSettings { runner_jre: "/opt/jdk-21".into(), ..old };
+        let settings = MavenSettings { runner_jre: "/opt/jdk-21".into(), delegate_build: true, ..old };
        let encoded = serde_json::to_string(&settings).unwrap();
        assert!(encoded.contains(r#""runnerJre":"/opt/jdk-21""#));
+        assert!(encoded.contains(r#""delegateBuild":true"#));
        assert_eq!(serde_json::from_str::<MavenSettings>(&encoded).unwrap(), settings);
    }

@@ -105,9 +105,12 @@ fn builtin(adapter: &str) -> Option<DebuggerConfig> {
            "stdio",
            "Install netcoredbg (.NET debugger) in Settings › Languages & Frameworks › Language Servers",
        ),
-        "lldb-dap" => {
-            cfg("lldb-dap", &[], "stdio", "Install LLVM's lldb-dap (brew install llvm; on macOS `xcrun -f lldb-dap`)")
-        }
+        "lldb-dap" => cfg(
+            "lldb-dap",
+            &[],
+            "stdio",
+            "Install LLVM's lldb-dap (brew install llvm), or on macOS use the one found by xcrun -f lldb-dap",
+        ),
        // GDB's own DAP interpreter, stdio transport. The DAP support it needs shipped in GDB 14
        // (`prepare_gdb` refuses older ones with the required version in the message).
        "gdb" => cfg(
@@ -124,6 +127,12 @@ fn builtin(adapter: &str) -> Option<DebuggerConfig> {
            "tcp",
            "Install Microsoft java-debug (Settings › Languages & Frameworks › Language Servers)",
        ),
+        "bun-dap" => cfg(
+            "bun-dap",
+            &["${port}", "127.0.0.1"],
+            "tcp",
+            "To install bun-dap: git clone https://github.com/morf-bls/bun-dap, run bun install && bun run build, then place dist/bun-dap in your PATH",
+        ),
        _ => return None,
    })
 }
@@ -17,6 +17,7 @@ blake3.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 parking_lot.workspace = true
+notify.workspace = true
 futures.workspace = true
 sqlx.workspace = true
 tiberius.workspace = true
@@ -12,6 +12,8 @@ pub mod secrets;
 pub mod source;
 pub mod split;
 pub mod sql;
+mod watch;
+pub use watch::SourcesWatch;

 pub use conn::{Held, Pool, QueryResult, ResultColumn};
 pub use history::HistoryEntry;
@@ -235,6 +237,9 @@ pub struct DbManager {
    /// Manual-commit data sources, each with the connection of its open
    /// transaction (taken by the first statement after a commit / rollback).
    sessions: Mutex<HashMap<String, Session>>,
+    /// The data sources as last read from disk, for [`DbManager::reconcile_sources`]
+    /// (`None` until [`DbManager::watch_sources`]).
+    on_disk: Mutex<Option<Vec<DataSource>>>,
 }

 type Session = Arc<tokio::sync::Mutex<Option<Held>>>;
@@ -362,6 +367,7 @@ impl DbManager {
            tunnel_open: tokio::sync::Mutex::new(()),
            ssh,
            session_passwords: Mutex::default(),
+            on_disk: Mutex::default(),
            sessions: Mutex::default(),
        }
    }
@@ -474,9 +480,45 @@ impl DbManager {
                }
            }
        }
+        // Our own write is not reported as an edit on disk.
+        if let Some(known) = self.on_disk.lock().as_mut() {
+            *known = self.sources().unwrap_or(sources);
+        }
        Ok(())
    }

+    /// Watches the data source files (the project's and the global one) for edits made
+    /// outside the IDE; `on_change` runs on a background thread once they settle, and
+    /// should call [`DbManager::reconcile_sources`]. Stops when the result is dropped.
+    pub fn watch_sources(&self, on_change: Box<dyn Fn() + Send>) -> IdeResult<SourcesWatch> {
+        *self.on_disk.lock() = Some(self.sources().unwrap_or_default());
+        SourcesWatch::start(&self.root, self.global.as_ref().map(|(dir, _)| dir.as_path()), on_change)
+    }
+
+    /// Re-reads the data sources after their files changed on disk: connections of
+    /// removed or changed sources close, as on [`DbManager::save_sources`] (saved
+    /// passwords stay, the source may come back). Returns whether the list the tree
+    /// shows changed, or the files no longer parse (the tree reports the error).
+    pub async fn reconcile_sources(&self) -> bool {
+        let now = match self.sources() {
+            Ok(now) => now,
+            Err(e) => {
+                tracing::warn!("data sources: {}", e.message);
+                return true;
+            }
+        };
+        let before = self.on_disk.lock().replace(now.clone()).unwrap_or_default();
+        if before == now {
+            return false;
+        }
+        for o in &before {
+            if now.iter().find(|s| s.id == o.id) != Some(o) {
+                self.disconnect_source(&o.id).await;
+            }
+        }
+        true
+    }
+
    /// Remembers `password` for the session, and in the saved passwords when
    /// the data source saves passwords (`None` forgets it).
    pub async fn set_password(&self, id: &str, password: Option<String>, save: bool) -> IdeResult<()> {
@@ -8,7 +8,7 @@ use serde::{Deserialize, Serialize};
 use specta::Type;
 use std::path::Path;

-const FILE: &str = "data-sources.toml";
+pub(crate) const FILE: &str = "data-sources.toml";

 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Type)]
 #[serde(rename_all = "camelCase")]
@@ -0,0 +1,151 @@
+//! Data source files edited outside the Database tool window (by hand, by a coding
+//! agent, by `git checkout`) are picked up while the project is open, as IDEA does
+//! for `dataSources.xml`. The project file tree's watcher is no help here: it never
+//! reports a modification inside a gitignored `.openidea/`.
+
+use crate::source::FILE;
+use ide_core::{IdeError, IdeResult, project_dir};
+use notify::{RecommendedWatcher, RecursiveMode, Watcher};
+use parking_lot::Mutex;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Weak, mpsc};
+use std::time::Duration;
+
+/// How long a burst of writes (an editor's save, an agent's rewrite) settles before
+/// the files are read.
+const SETTLE: Duration = Duration::from_millis(200);
+
+/// Watches `<root>/.openidea/data-sources.toml` and `<global>/data-sources.toml`;
+/// stops when dropped.
+pub struct SourcesWatch {
+    _watcher: Arc<Mutex<RecommendedWatcher>>,
+}
+
+impl SourcesWatch {
+    /// Calls `on_change` once a burst of changes to either file settled, on a
+    /// background thread.
+    pub fn start(root: &Path, global: Option<&Path>, on_change: Box<dyn Fn() + Send>) -> IdeResult<Self> {
+        let config: PathBuf = root.join(project_dir::DIR);
+        let (tx, rx) = mpsc::channel::<()>();
+        let relevant = {
+            let config = config.clone();
+            move |p: &Path| p.file_name().is_some_and(|n| n == FILE) || p == config
+        };
+        let watcher = notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
+            if let Ok(e) = res
+                && e.paths.iter().any(|p| relevant(p))
+            {
+                let _ = tx.send(());
+            }
+        })
+        .map_err(|e| IdeError::internal(format!("data sources watcher: {e}")))?;
+        let watcher = Arc::new(Mutex::new(watcher));
+        {
+            let mut w = watcher.lock();
+            // The root (not recursively) sees `.openidea/` itself appear or go.
+            w.watch(root, RecursiveMode::NonRecursive)
+                .map_err(|e| IdeError::internal(format!("data sources watcher: {e}")))?;
+            if let Some(dir) = global.filter(|d| d.is_dir())
+                && let Err(e) = w.watch(dir, RecursiveMode::NonRecursive)
+            {
+                tracing::warn!("global data sources watcher: {e}");
+            }
+        }
+        let mut watching = watch_config(&watcher, &config, false);
+        let weak: Weak<Mutex<RecommendedWatcher>> = Arc::downgrade(&watcher);
+        std::thread::Builder::new().name("openidea-db-sources-watch".into()).spawn(move || {
+            while rx.recv().is_ok() {
+                // Wait until the writes stop.
+                loop {
+                    match rx.recv_timeout(SETTLE) {
+                        Ok(()) => {}
+                        Err(mpsc::RecvTimeoutError::Timeout) => break,
+                        Err(mpsc::RecvTimeoutError::Disconnected) => return,
+                    }
+                }
+                let Some(watcher) = weak.upgrade() else { return };
+                watching = watch_config(&watcher, &config, watching);
+                drop(watcher);
+                on_change();
+            }
+        })?;
+        Ok(Self { _watcher: watcher })
+    }
+}
+
+/// Watches `.openidea/` once it exists (again: a removed folder loses its watch);
+/// returns whether it is watched.
+fn watch_config(watcher: &Mutex<RecommendedWatcher>, config: &Path, watching: bool) -> bool {
+    if !config.is_dir() {
+        return false;
+    }
+    if watching {
+        return true;
+    }
+    match watcher.lock().watch(config, RecursiveMode::NonRecursive) {
+        Ok(()) => true,
+        Err(e) => {
+            tracing::warn!("data sources watcher {}: {e}", config.display());
+            false
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::time::Instant;
+
+    fn wait_for(count: &AtomicUsize, at_least: usize) -> bool {
+        let deadline = Instant::now() + Duration::from_secs(10);
+        while Instant::now() < deadline {
+            if count.load(Ordering::SeqCst) >= at_least {
+                return true;
+            }
+            std::thread::sleep(Duration::from_millis(20));
+        }
+        false
+    }
+
+    fn counting() -> (Arc<AtomicUsize>, Box<dyn Fn() + Send>) {
+        let count = Arc::new(AtomicUsize::new(0));
+        let c = count.clone();
+        (count, Box::new(move || _ = c.fetch_add(1, Ordering::SeqCst)))
+    }
+
+    #[test]
+    fn reports_project_and_global_files_and_a_config_folder_created_later() {
+        let tmp = tempfile::tempdir().unwrap();
+        let root = dunce::canonicalize(tmp.path()).unwrap();
+        let (project, global) = (root.join("p"), root.join("app"));
+        std::fs::create_dir_all(&project).unwrap();
+        std::fs::create_dir_all(&global).unwrap();
+        let (count, on_change) = counting();
+        let _watch = SourcesWatch::start(&project, Some(&global), on_change).unwrap();
+
+        // No `.openidea/` yet: it is created, then the file written into it.
+        std::fs::create_dir_all(project.join(".openidea")).unwrap();
+        assert!(wait_for(&count, 1));
+        let seen = count.load(Ordering::SeqCst);
+        std::fs::write(project.join(".openidea/data-sources.toml"), "").unwrap();
+        assert!(wait_for(&count, seen + 1), "project file");
+
+        let seen = count.load(Ordering::SeqCst);
+        std::fs::write(global.join("data-sources.toml"), "").unwrap();
+        assert!(wait_for(&count, seen + 1), "global file");
+    }
+
+    #[test]
+    fn ignores_other_files() {
+        let tmp = tempfile::tempdir().unwrap();
+        let root = dunce::canonicalize(tmp.path()).unwrap();
+        std::fs::create_dir_all(root.join(".openidea")).unwrap();
+        let (count, on_change) = counting();
+        let _watch = SourcesWatch::start(&root, None, on_change).unwrap();
+        std::fs::write(root.join(".openidea/settings.toml"), "").unwrap();
+        std::fs::write(root.join("main.py"), "").unwrap();
+        std::thread::sleep(SETTLE * 3);
+        assert_eq!(count.load(Ordering::SeqCst), 0);
+    }
+}
@@ -342,3 +342,31 @@ async fn applies_driver_properties_and_the_startup_script_on_connect() -> IdeRes
    assert_eq!(db.schemas("s1").await?, vec!["main"]);
    Ok(())
 }
+
+#[tokio::test]
+async fn data_sources_edited_on_disk_close_their_connections() -> IdeResult<()> {
+    let (tmp, db) = setup().await;
+    std::fs::File::create(tmp.path().join("other.db")).unwrap();
+    let (tx, rx) = std::sync::mpsc::channel();
+    let _watch = db.watch_sources(Box::new(move || _ = tx.send(())))?;
+    db.pool("s1").await?;
+    assert_eq!(db.connected(), vec!["s1"]);
+
+    // Our own save is no edit on disk.
+    let mut sources = db.sources()?;
+    sources[0].name = "renamed".into();
+    db.save_sources(sources.clone()).await?;
+    assert!(!db.reconcile_sources().await);
+
+    // Edited outside the IDE (by hand, by an agent): the watcher reports it, and the
+    // connection made with the old settings closes.
+    db.pool("s1").await?;
+    sources[0].file = Some("other.db".into());
+    ide_db::source::save(tmp.path(), &sources)?;
+    rx.recv_timeout(std::time::Duration::from_secs(10)).expect("the watcher reports the edit");
+    assert!(db.reconcile_sources().await);
+    assert!(db.connected().is_empty());
+    assert_eq!(db.sources()?[0].file.as_ref().map(|f| f.0.clone()), Some("other.db".into()));
+    assert!(!db.reconcile_sources().await);
+    Ok(())
+}
@@ -34,7 +34,7 @@ use std::path::{Component, Path, PathBuf};

 pub const MANIFEST_FILE: &str = "plugin.toml";
 /// The plugin API (engine) version this host implements.
-pub const ENGINE_VERSION: &str = "1.5.0";
+pub const ENGINE_VERSION: &str = "1.6.0";
 /// The 0.1 API, still accepted for contribution-only packages and 0.1 components.
 pub const LEGACY_API: &str = "0.1";
 /// Marketplace categories, in the order they are shown.
@@ -151,6 +151,8 @@ impl ActivationEvent {
 /// Workbench event kinds (`on-event:<kind>`, WIT `events.event-kind`).
 /// `scheduled` comes from `on-schedule` instead.
 pub const EVENT_KINDS: &[&str] = &[
+    "document-changed",
+    "websocket-ready",
    "run-exited",
    "terminal-exited",
    "content-tab-action",
@@ -366,6 +368,10 @@ impl Manifest {
        }
        let mut seen = BTreeSet::new();
        for event in &self.activation {
+            if matches!(event, ActivationEvent::Event(kind) if matches!(kind.as_str(), "document-changed" | "websocket-ready"))
+            {
+                self.needs_api(&format!("`{event}`"), 6)?;
+            }
            if matches!(event, ActivationEvent::Event(kind) if matches!(kind.as_str(), "file-closed" | "settings-changed" | "input-mode-changed" | "run-exited" | "terminal-exited" | "content-tab-action"))
            {
                self.needs_api(&format!("`{event}`"), 4)?;
@@ -582,6 +588,8 @@ mod tests {
            ActivationEvent::parse("on-event:files-changed"),
            Ok(ActivationEvent::Event("files-changed".into()))
        );
+        assert!(ActivationEvent::parse("on-event:document-changed").is_ok());
+        assert!(ActivationEvent::parse("on-event:websocket-ready").is_ok());
        let err = ActivationEvent::parse("on-event:keystroke").unwrap_err();
        assert!(err.contains("unknown event kind") && err.contains("action-performed"), "{err}");
        assert!(ActivationEvent::parse("on-event:scheduled").is_err(), "declared with on-schedule");
@@ -860,7 +868,7 @@ mod commands_api_tests {
 mod net_ask_api_tests {
    use super::*;
    #[test]
-    fn net_ask_requires_api_1_5() {
+    fn api_1_5_declarations_require_api_1_5() {
        for declaration in ["[capabilities]\nrequired = [\"net:ask\"]", "[capabilities]\noptional = [\"net:ask\"]"] {
            for engine in ["^1.4", "^1.5"] {
                let manifest = format!(
@@ -875,4 +883,23 @@ mod net_ask_api_tests {
            }
        }
    }
+
+    #[test]
+    fn realtime_events_require_api_1_6() {
+        for declaration in
+            ["activation = [\"on-event:document-changed\"]", "activation = [\"on-event:websocket-ready\"]"]
+        {
+            for engine in ["^1.5", "^1.6"] {
+                let manifest = format!(
+                    "id = 'test'\nname = 'Test'\nversion = '0.1.0'\nwasm = 'plugin.wasm'\nengines.openidea = '{engine}'\n{declaration}\n"
+                );
+                let parsed = Manifest::parse(&manifest);
+                if engine == "^1.6" {
+                    parsed.unwrap();
+                } else {
+                    assert!(parsed.unwrap_err().contains("new in plugin API 1.6"), "{engine}");
+                }
+            }
+        }
+    }
 }
@@ -14,6 +14,7 @@ encoding_rs.workspace = true
 ide-core.workspace = true
 ide-mcp.workspace = true
 futures.workspace = true
+tokio-tungstenite.workspace = true
 ide-text.workspace = true
 ide-vfs.workspace = true
 regex-syntax.workspace = true
@@ -176,14 +176,13 @@ fn apply_custom_migration_in(path: &Path, local: &Path) {

 /// Entries of a local-data tree that move to a newly chosen directory.
 /// Non-relocatable tool installs (virtual environments with absolute interpreter
-/// paths) stay behind and are reinstalled on demand, as in `migrate_entries`.
+/// paths) stay behind and are reinstalled on demand, as in `migrate_entries`;
+/// so do registry agents, whose saved commands are absolute paths into `agents/tools`.
 fn custom_entries(root: &Path) -> Vec<PathBuf> {
    let mut entries = local_entries(root);
-    for extra in ["logs", "agents/tools"] {
-        let entry = PathBuf::from(extra);
-        if !entries.contains(&entry) {
-            entries.push(entry);
-        }
+    let logs = PathBuf::from("logs");
+    if !entries.contains(&logs) {
+        entries.push(logs);
    }
    entries
 }
@@ -884,11 +883,13 @@ mod tests {
    }

    #[test]
-    fn custom_entries_cover_logs_and_agent_tools() {
+    fn custom_entries_cover_logs_and_keep_agent_installs() {
        let tmp = tempfile::tempdir().unwrap();
        let entries = custom_entries(tmp.path());
-        for expected in ["logs", "agents/tools", "index", "cache", "lsp", "agents/registry.json"] {
+        for expected in ["logs", "index", "cache", "lsp", "agents/registry.json"] {
            assert!(entries.contains(&PathBuf::from(expected)), "{entries:?}");
        }
+        // Registry agent files name their executables by absolute path.
+        assert!(!entries.iter().any(|e| e.starts_with("agents/tools")), "{entries:?}");
    }
 }
@@ -84,6 +84,14 @@ pub struct NotebookService {
    registry: Arc<Mutex<Registry>>,
    kernels: Arc<KernelManager>,
    sink: SharedSink,
+    observer: parking_lot::RwLock<Option<Arc<dyn NotebookObserver>>>,
+}
+
+/// Learns each open notebook's contents after every change of them, and when it closes (the
+/// language servers' copy of its code cells).
+pub trait NotebookObserver: Send + Sync {
+    fn changed(&self, path: &Path, snapshot: &NotebookSnapshot);
+    fn closed(&self, path: &Path);
 }

 impl NotebookService {
@@ -121,7 +129,23 @@ impl NotebookService {
                }
            }),
        );
-        Arc::new(Self { registry, kernels: Arc::new(kernels), sink })
+        Arc::new(Self { registry, kernels: Arc::new(kernels), sink, observer: Default::default() })
+    }
+
+    pub fn set_observer(&self, observer: Arc<dyn NotebookObserver>) {
+        *self.observer.write() = Some(observer);
+    }
+
+    /// Called with the document's state locked, so observers see its changes in order.
+    fn changed(&self, path: &Path, snapshot: &NotebookSnapshot) {
+        if let Some(observer) = self.observer.read().clone() {
+            observer.changed(path, snapshot);
+        }
+    }
+
+    /// The file of an open notebook.
+    pub fn path_of(&self, id: &str) -> IdeResult<PathBuf> {
+        Ok(self.resolve(id)?.path.clone())
    }

    pub fn open(&self, path: &Path) -> IdeResult<NotebookSnapshot> {
@@ -136,6 +160,7 @@ impl NotebookService {
        }
        let document = NotebookDocument::load(&path)?;
        let initial = document.snapshot();
+        self.changed(&path, &initial);
        let id = initial.id.clone();
        registry.paths.insert(path.clone(), id.clone());
        registry.documents.insert(
@@ -173,7 +198,9 @@ impl NotebookService {
        let document = self.resolve(id)?;
        let mut state = document.state.lock();
        mutable(&state)?;
-        state.document.edit(version, edit)
+        let snapshot = state.document.edit(version, edit)?;
+        self.changed(&document.path, &snapshot);
+        Ok(snapshot)
    }

    pub fn save(&self, id: &str) -> IdeResult<NotebookSnapshot> {
@@ -190,7 +217,9 @@ impl NotebookService {
        if state.active.is_some() {
            return Err(IdeError::unavailable("stop notebook execution before undo or redo"));
        }
-        if redo { state.document.redo() } else { state.document.undo() }
+        let snapshot = if redo { state.document.redo() } else { state.document.undo() }?;
+        self.changed(&document.path, &snapshot);
+        Ok(snapshot)
    }

    pub async fn interrupt(&self, id: &str) -> IdeResult<()> {
@@ -238,7 +267,10 @@ impl NotebookService {
                snapshot(&document)
            } else {
                self.kernels.shutdown_kernel(&document.path).await?;
-                document.state.lock().document.reload()
+                let mut state = document.state.lock();
+                let snapshot = state.document.reload()?;
+                self.changed(&document.path, &snapshot);
+                Ok(snapshot)
            }
        }
        .await;
@@ -284,6 +316,10 @@ impl NotebookService {
        let mut registry = self.registry.lock();
        registry.documents.remove(id);
        registry.paths.remove(&document.path);
+        drop(registry);
+        if let Some(observer) = self.observer.read().clone() {
+            observer.closed(&document.path);
+        }
        Ok(())
    }

@@ -572,6 +608,49 @@ mod tests {
        KernelOutput { kind: kind.into(), content, msg_id: "actual-parent-request".into() }
    }

+    #[tokio::test]
+    async fn the_observer_follows_every_change_of_an_open_notebook_and_its_close() {
+        #[derive(Default)]
+        struct Seen(Mutex<Vec<String>>);
+        impl NotebookObserver for Seen {
+            fn changed(&self, path: &Path, snapshot: &NotebookSnapshot) {
+                let sources: Vec<&str> = snapshot.cells.iter().map(|c| c.source.as_str()).collect();
+                let name = path.file_name().unwrap().to_string_lossy();
+                self.0.lock().push(format!("{name}: {}", sources.join(" | ")));
+            }
+            fn closed(&self, path: &Path) {
+                self.0.lock().push(format!("closed {}", path.file_name().unwrap().to_string_lossy()));
+            }
+        }
+        let (dir, service, initial) = fixture();
+        let seen = Arc::new(Seen::default());
+        service.set_observer(seen.clone());
+        service.close(&initial.id).await.unwrap();
+        let opened = service.open(&dir.path().join("session.ipynb")).unwrap();
+        assert_eq!(
+            service.path_of(&opened.id).unwrap(),
+            dunce::canonicalize(dir.path().join("session.ipynb")).unwrap()
+        );
+        let edit = NotebookEdit::SetSource { cell: "second".into(), source: "print(x + 1)".into() };
+        let edited = service.edit(&opened.id, opened.version, edit).unwrap();
+        service.undo(&opened.id, false).unwrap();
+        // A second view of the open notebook changes nothing.
+        service.open(&dir.path().join("session.ipynb")).unwrap();
+        assert!(edited.version > opened.version);
+        service.close(&opened.id).await.unwrap();
+        assert_eq!(
+            *seen.0.lock(),
+            [
+                "closed session.ipynb",
+                "session.ipynb: x = 1 | print(x) | notes",
+                "session.ipynb: x = 1 | print(x + 1) | notes",
+                "session.ipynb: x = 1 | print(x) | notes",
+                "closed session.ipynb",
+            ]
+        );
+        service.shutdown().await;
+    }
+
    #[tokio::test]
    async fn canonical_split_opens_share_incarnation_but_reopen_and_projects_do_not() {
        let (dir, service, initial) = fixture();
@@ -48,7 +48,7 @@ async fn resolve_launch(
    let build = policy != BuildPolicy::NoBuild;
    let full = policy == BuildPolicy::Rebuild;
    let build_kind = if full { compile::BuildKind::Rebuild } else { compile::BuildKind::Build };
-    let (model, external, delegated) = {
+    let (model, external, delegated, prefer_maven) = {
        let project = project.clone();
        let settings = settings.clone();
        let module = module.clone();
@@ -58,15 +58,17 @@ async fn resolve_launch(
            let non_java = build && compile::launch_has_non_java_sources(&project.root, &model, &module, test);
            let is_maven =
                model.maven.as_ref().is_some_and(|m| m.projects.iter().any(|p| project.root.join(&p.dir) == module));
+            let prefer_maven = settings.languages.java.maven.delegate_build;
            let delegated = build
                && delegates_to_external_build(
                    is_maven,
                    policy,
                    non_java,
                    cache::has_record(&project.root, &project.build_cache_dir(), &module),
+                    prefer_maven,
                );
            let external = non_java || delegated;
-            (model, external, delegated)
+            (model, external, delegated, prefer_maven)
        })
        .await
        .map_err(|e| IdeError::internal(e.to_string()))?
@@ -79,6 +81,7 @@ async fn resolve_launch(
    tokio::task::spawn_blocking(move || {
        let _external_build = external_build;
        let built = if delegated {
+            let policy = delegated_policy(policy, prefer_maven, info.error_count);
            let outputs = project.java_launch_snapshots.lock().build(
                &project.root,
                &model,
@@ -169,6 +172,17 @@ async fn resolve_launch(
    .map_err(|e| IdeError::internal(e.to_string()))?
 }

+/// With Settings › Maven › Delegate IDE build/run actions to Maven, javac's verdict replaces
+/// JDT's. Classes JDT wrote for sources with errors look up to date to Maven's incremental
+/// compiler, so a launch JDT reports errors for recompiles everything.
+fn delegated_policy(policy: BuildPolicy, prefer_maven: bool, jdt_errors: u32) -> BuildPolicy {
+    if prefer_maven && jdt_errors > 0 && matches!(policy, BuildPolicy::Automatic | BuildPolicy::Make) {
+        BuildPolicy::Rebuild
+    } else {
+        policy
+    }
+}
+
 fn apply_external_build_result(info: &mut JavaLaunchInfo, failure: Option<String>, clean: bool) {
    info.build = Some(if let Some(failure) = failure {
        info.hints.push(failure);
@@ -241,8 +255,15 @@ fn apply_library_classpath(
 /// Whether a launch should be built by the project's build system instead of jdtls.
 /// A Rebuild has to clean the output; jdtls' full build only recompiles sources, so
 /// Maven modules always delegate for it, even when every source is Java.
-fn delegates_to_external_build(is_maven: bool, policy: BuildPolicy, non_java: bool, has_record: bool) -> bool {
-    is_maven && (policy == BuildPolicy::Rebuild || non_java || has_record)
+/// `prefer_maven` is Settings › Maven › Delegate IDE build/run actions to Maven.
+fn delegates_to_external_build(
+    is_maven: bool,
+    policy: BuildPolicy,
+    non_java: bool,
+    has_record: bool,
+    prefer_maven: bool,
+) -> bool {
+    is_maven && (prefer_maven || policy == BuildPolicy::Rebuild || non_java || has_record)
 }

 fn same_entry(a: &str, b: &str, windows: bool) -> bool {
@@ -321,14 +342,29 @@ pub(crate) fn register(d: &mut Dispatcher) {
 mod tests {
    use super::*;

+    #[test]
+    fn delegated_launches_with_jdt_errors_recompile_with_maven() {
+        assert_eq!(delegated_policy(BuildPolicy::Make, true, 1), BuildPolicy::Rebuild);
+        assert_eq!(delegated_policy(BuildPolicy::Automatic, true, 2), BuildPolicy::Rebuild);
+        assert_eq!(delegated_policy(BuildPolicy::Make, true, 0), BuildPolicy::Make);
+        assert_eq!(delegated_policy(BuildPolicy::Make, false, 1), BuildPolicy::Make);
+        assert_eq!(
+            delegated_policy(BuildPolicy::MakeAndContinueOnErrors, true, 1),
+            BuildPolicy::MakeAndContinueOnErrors
+        );
+    }
+
    #[test]
    fn rebuild_delegates_maven_builds_to_the_build_system() {
-        assert!(delegates_to_external_build(true, BuildPolicy::Rebuild, false, false));
-        assert!(delegates_to_external_build(true, BuildPolicy::Automatic, true, false));
-        assert!(delegates_to_external_build(true, BuildPolicy::Make, false, true));
-        assert!(!delegates_to_external_build(true, BuildPolicy::Automatic, false, false));
-        assert!(!delegates_to_external_build(true, BuildPolicy::Make, false, false));
-        assert!(!delegates_to_external_build(false, BuildPolicy::Rebuild, false, false));
+        assert!(delegates_to_external_build(true, BuildPolicy::Rebuild, false, false, false));
+        assert!(delegates_to_external_build(true, BuildPolicy::Automatic, true, false, false));
+        assert!(delegates_to_external_build(true, BuildPolicy::Make, false, true, false));
+        assert!(!delegates_to_external_build(true, BuildPolicy::Automatic, false, false, false));
+        assert!(!delegates_to_external_build(true, BuildPolicy::Make, false, false, false));
+        assert!(!delegates_to_external_build(false, BuildPolicy::Rebuild, false, false, false));
+        assert!(delegates_to_external_build(true, BuildPolicy::Automatic, false, false, true));
+        assert!(delegates_to_external_build(true, BuildPolicy::Make, false, false, true));
+        assert!(!delegates_to_external_build(false, BuildPolicy::Automatic, false, false, true));
    }

    #[test]
@@ -314,10 +314,12 @@ pub async fn lsp_goto(host: &Host, window: &str, doc: DocId, pos: Pos, kind: Got
        && let Some(plugins) = crate::plugins::editor_plugin_host()
    {
        let project = host.project(window)?;
-        let (path, version) = project.documents.identity(doc)?;
-        if let Some(found) = plugins.navigation_at(&project.root, &path, version, pos, Some(window), |p| {
-            project.documents.by_path(p).and_then(|id| project.documents.version(id).ok())
-        }) {
+        // A notebook cell is a language-server document only: no plugin navigates in it.
+        if let Ok((path, version)) = project.documents.identity(doc)
+            && let Some(found) = plugins.navigation_at(&project.root, &path, version, pos, Some(window), |p| {
+                project.documents.by_path(p).and_then(|id| project.documents.version(id).ok())
+            })
+        {
            return Ok(found.targets);
        }
    }
@@ -19,6 +19,13 @@ pub fn get(host: &Host, window: &str, id: String) -> IdeResult<NotebookSnapshot>
    host.project(window)?.notebooks.get(&id)
 }

+/// The language-server documents of the notebook's code cells (empty when no server syncs notebooks).
+pub fn cell_docs(host: &Host, window: &str, id: String) -> IdeResult<Vec<ide_lsp::NotebookCellDoc>> {
+    let project = host.project(window)?;
+    let path = project.notebooks.path_of(&id)?;
+    Ok(project.lsp.notebook_cells(&path))
+}
+
 pub async fn edit(
    host: &Host,
    window: &str,
@@ -132,6 +139,7 @@ impl Drop for RequestGuard<'_> {
 pub(crate) fn register(d: &mut Dispatcher) {
    d.add("notebook_open", |c| async move { open(&c.host, &c.window, c.arg("path")?).await });
    d.add("notebook_get", |c| async move { get(&c.host, &c.window, c.arg("id")?) });
+    d.add("notebook_cell_docs", |c| async move { cell_docs(&c.host, &c.window, c.arg("id")?) });
    d.add("notebook_edit", |c| async move {
        edit(&c.host, &c.window, c.arg("id")?, c.arg("version")?, c.arg("edit")?).await
    });
@@ -111,8 +111,9 @@ pub trait PluginWorkspace: Send + Sync {
    /// Version and text of the open document at `path`.
    fn document(&self, path: &Path) -> Option<(u64, String)>;
    /// Applies `edits` to the open document at `path` if it is still at
-    /// `version`, as one change (one undo step in the editor).
-    fn apply_edits(&self, path: &Path, version: u64, edits: &[TextEdit]) -> Result<u64, HostError>;
+    /// `version`, as one change: one undo step in the editor, or none for a
+    /// `background` caller (a plugin event rather than a command).
+    fn apply_edits(&self, path: &Path, version: u64, edits: &[TextEdit], background: bool) -> Result<u64, HostError>;
    /// Code from the project may run (git hooks and config on fetch).
    fn trusted(&self, root: &Path) -> bool;
    /// Root of the project open in `window`.
@@ -180,7 +181,7 @@ impl SecretStore for FileSecretStore {
 /// Largest response body a plugin receives.
 const MAX_RESPONSE: usize = 8 << 20;
 /// Largest stdout / stderr kept from a process.
-const MAX_OUTPUT: usize = 4 << 20;
+const MAX_OUTPUT: usize = 16 << 20;
 /// Most processes one plugin keeps running at once.
 const MAX_PROCESSES: usize = 8;
 /// Most output one stream of a handle holds until the plugin reads it.
@@ -488,6 +489,7 @@ type SettingSnapshots = BTreeMap<(Option<String>, String), BTreeMap<String, Stri
 /// State shared by the host and every plugin's [`PluginApi`].
 pub(super) struct Shared {
    pub sink: Arc<dyn PluginEventSink>,
+    pub websockets: super::websocket::WebSocketHub,
    pub workspace: RwLock<Option<Arc<dyn PluginWorkspace>>>,
    /// `<plugins>/.data`: one directory per plugin id.
    pub data_dir: PathBuf,
@@ -565,7 +567,8 @@ impl Shared {
            ui,
            registry,
            locale: RwLock::new("en".into()),
-            sink,
+            sink: sink.clone(),
+            websockets: super::websocket::WebSocketHub::new(sink.clone()),
            workspace: RwLock::new(None),
            secrets: RwLock::new(Arc::new(FileSecretStore { dir: data_dir.clone() })),
            pins,
@@ -775,6 +778,7 @@ impl Shared {

        self.kill_processes(plugin);
        self.connections.clear(plugin);
+        self.websockets.disconnect_plugin(plugin);
        self.setting_snapshots.lock().retain(|(_, p), _| p != plugin);
        let channels: Vec<_> = {
            let mut output = self.output.lock();
@@ -2048,13 +2052,14 @@ impl HostApi for PluginApi {

    fn apply_edits(
        &self,
-        _client: Option<&str>,
+        client: Option<&str>,
        path: &Path,
        version: u64,
        edits: Vec<TextEdit>,
    ) -> Result<u64, HostError> {
        let workspace = self.shared.workspace().ok_or_else(|| HostError::NotFound("no project is open".into()))?;
-        workspace.apply_edits(path, version, &edits)
+        // Edits from a window's command are undoable; `editor.apply-edits-at` from a background event isn't.
+        workspace.apply_edits(path, version, &edits, client.is_none())
    }

    fn storage_get(&self, key: &str) -> Result<Option<String>, HostError> {
@@ -2275,6 +2280,49 @@ impl HostApi for PluginApi {
        })
    }

+    fn websocket_connect(
+        &self,
+        root: &Path,
+        request: ide_plugin::WebSocketRequest,
+        timeout: Duration,
+    ) -> Result<u64, HostError> {
+        let secrets = self.shared.secrets.read().clone();
+        let plugin = self.id().to_owned();
+        let headers = request
+            .headers
+            .into_iter()
+            .map(|(name, value)| {
+                let value = fill_secrets(&value, |name| secrets.get(&plugin, name))?;
+                Ok((name, value))
+            })
+            .collect::<Result<Vec<_>, HostError>>()?;
+        net_runtime()
+            .block_on(self.shared.websockets.connect(
+                &plugin,
+                &to_ipc(root),
+                &request.url,
+                &headers,
+                &request.subprotocol,
+                timeout,
+            ))
+            .map_err(|error| match error {
+                super::websocket::ConnectError::Invalid(message) => HostError::Invalid(message),
+                super::websocket::ConnectError::Failed(message) => HostError::Failed(message),
+            })
+    }
+
+    fn websocket_send(&self, connection: u64, bytes: Vec<u8>) -> Result<(), HostError> {
+        self.shared.websockets.send(self.id(), connection, bytes).map_err(HostError::Failed)
+    }
+
+    fn websocket_drain(&self, connection: u64, limit: usize) -> Result<ide_plugin::WebSocketBatch, HostError> {
+        self.shared.websockets.drain(self.id(), connection, limit).map_err(HostError::Invalid)
+    }
+
+    fn websocket_close(&self, connection: u64) {
+        let _ = self.shared.websockets.disconnect(self.id(), connection);
+    }
+
    /// The `net:ask` answer: a host the user allowed with "always" is
    /// remembered in the ledger (no dialog again, background included); a new
    /// host is asked about in `client`'s window when there is one. Dismissal
@@ -2443,9 +2491,14 @@ impl HostApi for PluginApi {
        let read = |pipe: Option<Box<dyn Read + Send>>| {
            std::thread::spawn(move || {
                let mut out = Vec::new();
+                let mut overflow = false;
                if let Some(p) = pipe {
                    let mut p = p;
-                    let _ = p.by_ref().take(MAX_OUTPUT as u64).read_to_end(&mut out);
+                    let _ = p.by_ref().take((MAX_OUTPUT + 1) as u64).read_to_end(&mut out);
+                    if out.len() > MAX_OUTPUT {
+                        overflow = true;
+                        out.truncate(MAX_OUTPUT);
+                    }
                    let _ = std::io::copy(&mut p, &mut std::io::sink());
                }
                let mut text = String::from_utf8_lossy(&out).into_owned();
@@ -2454,7 +2507,7 @@ impl HostApi for PluginApi {
                    end -= 1;
                }
                text.truncate(end);
-                text
+                (text, overflow)
            })
        };
        let stdout = read(child.stdout.take().map(|p| Box::new(p) as Box<dyn Read + Send>));
@@ -2474,11 +2527,18 @@ impl HostApi for PluginApi {
            }
            std::thread::sleep(Duration::from_millis(10));
        };
-        Ok(ProcessOutput {
-            status: status.code(),
-            stdout: stdout.join().unwrap_or_default(),
-            stderr: stderr.join().unwrap_or_default(),
-        })
+        let (stdout, stdout_overflow) = stdout.join().unwrap_or_default();
+        let (stderr, _) = stderr.join().unwrap_or_default();
+        if stdout_overflow {
+            tracing::warn!(
+                plugin = %self.id(),
+                program = %cmd.program,
+                limit_mib = MAX_OUTPUT >> 20,
+                "process stdout exceeded limit"
+            );
+            return Err(HostError::Failed(format!("{} stdout exceeded {} MiB limit", cmd.program, MAX_OUTPUT >> 20)));
+        }
+        Ok(ProcessOutput { status: status.code(), stdout, stderr })
    }

    fn subscribe(&self, kind: EventKind) {
@@ -3208,11 +3268,16 @@ impl PluginWorkspace for HostWorkspace {
        Some((u64::from(snap.version), snap.text))
    }

-    fn apply_edits(&self, path: &Path, version: u64, edits: &[TextEdit]) -> Result<u64, HostError> {
+    fn apply_edits(&self, path: &Path, version: u64, edits: &[TextEdit], background: bool) -> Result<u64, HostError> {
        let (project, doc) =
            self.open_document(path).ok_or_else(|| HostError::NotFound(format!("{} is not open", path.display())))?;
        let version = u32::try_from(version).map_err(|_| HostError::Stale)?;
-        match project.documents.apply_edits_at(doc, version, edits) {
+        let applied = if background {
+            project.documents.apply_background_edits_at(doc, version, edits)
+        } else {
+            project.documents.apply_edits_at(doc, version, edits)
+        };
+        match applied {
            Ok(Some(v)) => Ok(u64::from(v)),
            Ok(None) => Err(HostError::Stale),
            Err(e) if e.kind == ide_core::error::ErrorKind::InvalidInput => Err(HostError::Invalid(e.message)),
@@ -3542,7 +3607,7 @@ mod tests {
        fn document(&self, _: &Path) -> Option<(u64, String)> {
            None
        }
-        fn apply_edits(&self, _: &Path, _: u64, _: &[TextEdit]) -> Result<u64, HostError> {
+        fn apply_edits(&self, _: &Path, _: u64, _: &[TextEdit], _: bool) -> Result<u64, HostError> {
            Err(HostError::Stale)
        }
        fn trusted(&self, _: &Path) -> bool {
@@ -3606,7 +3671,7 @@ mod tests {
        fn document(&self, _: &Path) -> Option<(u64, String)> {
            None
        }
-        fn apply_edits(&self, _: &Path, _: u64, _: &[TextEdit]) -> Result<u64, HostError> {
+        fn apply_edits(&self, _: &Path, _: u64, _: &[TextEdit], _: bool) -> Result<u64, HostError> {
            Err(HostError::Stale)
        }
        fn trusted(&self, _: &Path) -> bool {
@@ -3807,10 +3872,11 @@ mod tests {
        let tmp = tempfile::tempdir().unwrap();
        let api = shell_api(tmp.path());
        let token = CancellationToken::new();
-        let cmd = shell("head -c 5242880 /dev/zero; head -c 5242880 /dev/zero >&2");
+        // stdout within the limit comes back whole; stderr beyond it is truncated (stdout beyond it fails).
+        let cmd = shell(&format!("head -c 5242880 /dev/zero; head -c {} /dev/zero >&2", MAX_OUTPUT + (1 << 20)));
        let out = api.run_process(cmd.clone(), None, None, &token, Duration::from_secs(5)).unwrap();
        assert_eq!(out.status, Some(0));
-        assert_eq!((out.stdout.len(), out.stderr.len()), (MAX_OUTPUT, MAX_OUTPUT));
+        assert_eq!((out.stdout.len(), out.stderr.len()), (5 << 20, MAX_OUTPUT));
        let id = api.start_process(cmd, None, None).unwrap();
        let out = api.wait_process(id, MAX_PENDING, Duration::from_secs(5), &token).unwrap();
        assert_eq!(out.status, Some(0));
@@ -4066,7 +4132,7 @@ mod workbench_tests {
        fn document(&self, _: &Path) -> Option<(u64, String)> {
            self.text.clone().map(|s| (1, s))
        }
-        fn apply_edits(&self, _: &Path, _: u64, _: &[TextEdit]) -> Result<u64, HostError> {
+        fn apply_edits(&self, _: &Path, _: u64, _: &[TextEdit], _: bool) -> Result<u64, HostError> {
            Err(HostError::Stale)
        }
        fn trusted(&self, _: &Path) -> bool {
@@ -262,7 +262,7 @@ mod tests {
        fn project_of(&self, _: &str) -> Option<PathBuf> {
            None
        }
-        fn apply_edits(&self, _: &Path, _: u64, _: &[ide_core::TextEdit]) -> Result<u64, HostError> {
+        fn apply_edits(&self, _: &Path, _: u64, _: &[ide_core::TextEdit], _: bool) -> Result<u64, HostError> {
            unreachable!()
        }
        fn publish_diagnostics(&self, _: &str, _: &Path, _: Vec<ide_core::Diagnostic>) -> Result<(), HostError> {
@@ -1,6 +1,7 @@
 //! Project-relative event filters, kept with runtime subscriptions across idle reclaim.
 use super::{LoadedPlugin, PluginHost};
 use globset::{GlobBuilder, GlobMatcher};
+use ide_extension::Capability;
 use ide_plugin::{EventKind, HostError, WorkbenchEvent};
 use std::{
    collections::{BTreeSet, HashMap},
@@ -21,9 +22,9 @@ pub(super) struct PathFilter {

 impl PathFilter {
    pub fn new(kind: &str, globs: Vec<String>) -> Result<Self, HostError> {
-        if !matches!(kind, "file-opened" | "file-saved" | "files-changed") {
+        if !matches!(kind, "file-opened" | "file-saved" | "files-changed" | "document-changed") {
            return Err(HostError::Invalid(
-                "subscribe-paths supports file-opened, file-saved and files-changed".into(),
+                "subscribe-paths supports file-opened, file-saved, files-changed and document-changed".into(),
            ));
        }
        if globs.len() > 64 {
@@ -95,13 +96,15 @@ impl PluginHost {
        events: &[WorkbenchEvent],
        subs: &Subscriptions,
    ) -> Vec<WorkbenchEvent> {
-        let Some(filters) = subs.paths.get(plugin.id()) else { return events.to_vec() };
+        let filters = subs.paths.get(plugin.id());
        let workspace = shared.workspace();
        events
            .iter()
            .filter_map(|event| {
                if let WorkbenchEvent::FilesChanged(batch) = event {
-                    let Some(filter) = filters.get(event.kind().name()) else { return Some(event.clone()) };
+                    let Some(filter) = filters.and_then(|filters| filters.get(event.kind().name())) else {
+                        return Some(event.clone());
+                    };
                    let root = workspace.as_ref()?.project_of(client?)?;
                    if plugin.dev_root.as_deref().is_some_and(|r| r != root) {
                        return None;
@@ -126,9 +129,9 @@ impl PluginHost {
                }
                let path = match event {
                    WorkbenchEvent::FileOpened(path) | WorkbenchEvent::FileSaved(path) => Path::new(path),
+                    WorkbenchEvent::DocumentChanged(change) => Path::new(&change.path),
                    _ => return Some(event.clone()),
                };
-                let Some(filter) = filters.get(event.kind().name()) else { return Some(event.clone()) };
                if !path.is_absolute() {
                    return None;
                }
@@ -147,7 +150,26 @@ impl PluginHost {
                if plugin.dev_root.as_deref().is_some_and(|r| r != root) {
                    return None;
                }
-                filter.matches(&root, path).then(|| event.clone())
+                if let WorkbenchEvent::DocumentChanged(change) = event {
+                    // Typed text: only for plugins that may read editors, and only for paths they named.
+                    let filter = filters.and_then(|filters| filters.get(event.kind().name()))?;
+                    let granted = &plugin.meta.lock().granted;
+                    if !granted.contains(&Capability::ProjectRead)
+                        || !granted.contains(&Capability::EditorRead)
+                        || !filter.matches(&root, path)
+                    {
+                        return None;
+                    }
+                    let real = ide_plugin::resolve_real(path);
+                    let relative = real.strip_prefix(ide_plugin::resolve_real(&root)).ok()?;
+                    let mut change = change.clone();
+                    change.path = relative.to_string_lossy().replace('\\', "/");
+                    return Some(WorkbenchEvent::DocumentChanged(change));
+                }
+                filters
+                    .and_then(|filters| filters.get(event.kind().name()))
+                    .is_none_or(|filter| filter.matches(&root, path))
+                    .then(|| event.clone())
            })
            .collect()
    }
@@ -167,6 +189,7 @@ mod tests {
        assert!(PathFilter::new("file-saved", vec!["./".repeat(513)]).is_err());
        assert!(PathFilter::new("file-saved", vec![]).is_ok());
        assert!(PathFilter::new("files-changed", vec![]).is_ok());
+        assert!(PathFilter::new("document-changed", vec!["src/**/*.rs".into()]).is_ok());
    }

    #[test]
@@ -507,7 +507,13 @@ mod tests {
        fn document(&self, _: &Path) -> Option<(u64, String)> {
            None
        }
-        fn apply_edits(&self, _: &Path, _: u64, _: &[ide_core::TextEdit]) -> Result<u64, ide_plugin::HostError> {
+        fn apply_edits(
+            &self,
+            _: &Path,
+            _: u64,
+            _: &[ide_core::TextEdit],
+            _: bool,
+        ) -> Result<u64, ide_plugin::HostError> {
            unreachable!()
        }
        fn trusted(&self, _: &Path) -> bool {
@@ -41,6 +41,7 @@ pub mod language_features;
 mod market;
 mod mcp;
 pub mod navigation;
+mod network_inbox;
 mod points;
 mod providers;
 pub mod remote;
@@ -54,6 +55,7 @@ mod types;
 pub mod ui_points;
 pub mod views;
 pub mod web;
+mod websocket;

 pub use api::{FileSecretStore, HostWorkspace, PluginWorkspace, SecretStore};
 pub(crate) use hover::merge as merge_hover;
@@ -1921,6 +1923,7 @@ impl PluginHost {
        for p in plugins {
            let wanted: Vec<WorkbenchEvent> = events
                .iter()
+                .filter(|event| !matches!(event, WorkbenchEvent::WebsocketReady(ready) if ready.plugin_id != p.id()))
                .filter(|e| Self::wants(&p, e.kind(), subs.kinds.get(p.id())))
                .filter(|e| {
                    !matches!(e, WorkbenchEvent::FilesChanged(_))
@@ -0,0 +1,203 @@
+//! Bounded inbox for the proposed background network API.
+//!
+//! Experimental transport-independent primitive. This module does not expose
+//! sockets to plugins. Overflow closes the inbox: silently dropping a document
+//! update would make a collaboration session appear healthy while diverging.
+use std::collections::VecDeque;
+use std::sync::Mutex;
+
+#[derive(Debug, PartialEq, Eq)]
+pub(super) enum Push {
+    /// The caller must schedule one plugin wake-up after releasing this lock.
+    Wake,
+    Queued,
+    /// This message was refused and the caller must schedule a wake-up.
+    OverflowWake,
+    /// This message was refused; a wake-up is already pending.
+    OverflowQueued,
+    Closed,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub(super) struct Batch {
+    pub messages: Vec<Vec<u8>>,
+    pub closed: bool,
+    pub more: bool,
+}
+
+#[derive(Default)]
+struct State {
+    messages: VecDeque<Box<[u8]>>,
+    bytes: usize,
+    closed: bool,
+    notified: bool,
+}
+
+pub(super) struct Inbox {
+    state: Mutex<State>,
+    max_messages: usize,
+    max_bytes: usize,
+    max_frame: usize,
+}
+
+impl Inbox {
+    pub fn new(max_messages: usize, max_bytes: usize, max_frame: usize) -> Self {
+        assert!(max_messages > 0 && max_bytes > 0 && max_frame > 0);
+        Self { state: Mutex::new(State::default()), max_messages, max_bytes, max_frame }
+    }
+
+    pub fn push(&self, message: Vec<u8>) -> Push {
+        let mut state = self.state.lock().unwrap();
+        if state.closed {
+            return Push::Closed;
+        }
+        let overflow = message.len() > self.max_frame
+            || state.messages.len() >= self.max_messages
+            || message.len() > self.max_bytes.saturating_sub(state.bytes);
+        if overflow {
+            state.closed = true;
+        } else {
+            state.bytes += message.len();
+            state.messages.push_back(message.into_boxed_slice());
+        }
+        let wake = !state.notified;
+        state.notified = true;
+        match (overflow, wake) {
+            (false, true) => Push::Wake,
+            (false, false) => Push::Queued,
+            (true, true) => Push::OverflowWake,
+            (true, false) => Push::OverflowQueued,
+        }
+    }
+
+    /// Returns whether the caller must schedule a wake-up. Idempotent.
+    pub fn close(&self) -> bool {
+        let mut state = self.state.lock().unwrap();
+        if state.closed {
+            return false;
+        }
+        state.closed = true;
+        let wake = !state.notified;
+        state.notified = true;
+        wake
+    }
+
+    /// A partial drain retains the notification; the consumer must continue
+    /// draining when `more` is true. An empty drain rearms the next wake-up.
+    pub fn drain(&self, limit: usize) -> Batch {
+        assert!(limit > 0);
+        let mut state = self.state.lock().unwrap();
+        let count = limit.min(state.messages.len());
+        let messages: Vec<Vec<u8>> = state.messages.drain(..count).map(|message| message.into_vec()).collect();
+        state.bytes -= messages.iter().map(Vec::len).sum::<usize>();
+        let more = !state.messages.is_empty();
+        state.notified = more;
+        Batch { messages, closed: state.closed, more }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn coalesces_and_rearms_wakeups() {
+        let inbox = Inbox::new(4, 16, 8);
+        assert_eq!(inbox.push(vec![1]), Push::Wake);
+        assert_eq!(inbox.push(vec![2]), Push::Queued);
+        assert!(inbox.drain(1).more);
+        assert_eq!(inbox.push(vec![3]), Push::Queued);
+        assert_eq!(inbox.drain(4).messages, vec![vec![2], vec![3]]);
+        assert_eq!(inbox.push(vec![4]), Push::Wake);
+    }
+
+    #[test]
+    fn overflow_preserves_accepted_messages_and_reports_close() {
+        for (messages, bytes, frame) in [(1, 8, 8), (4, 1, 8), (4, 8, 1)] {
+            let inbox = Inbox::new(messages, bytes, frame);
+            assert_eq!(inbox.push(vec![1]), Push::Wake);
+            assert_eq!(inbox.push(vec![2, 3]), Push::OverflowQueued);
+            let batch = inbox.drain(4);
+            assert_eq!(batch.messages, vec![vec![1]]);
+            assert!(batch.closed);
+            assert_eq!(inbox.push(vec![]), Push::Closed);
+        }
+    }
+
+    #[test]
+    fn accepted_frame_does_not_retain_excess_capacity() {
+        let inbox = Inbox::new(2, 8, 8);
+        let mut frame = Vec::with_capacity(1024 * 1024);
+        frame.push(42);
+        assert_eq!(inbox.push(frame), Push::Wake);
+        let state = inbox.state.lock().unwrap();
+        assert_eq!(state.bytes, 1);
+        assert_eq!(state.messages[0].len(), 1);
+        assert_eq!(std::mem::size_of_val(&*state.messages[0]), 1);
+    }
+
+    #[test]
+    fn close_during_partial_drain_remains_visible_without_duplicate_wakeup() {
+        let inbox = Inbox::new(3, 12, 4);
+        assert_eq!(inbox.push(vec![1]), Push::Wake);
+        assert_eq!(inbox.push(vec![2]), Push::Queued);
+        let first = inbox.drain(1);
+        assert_eq!(first.messages, vec![vec![1]]);
+        assert!(first.more);
+        assert!(!inbox.close());
+        let second = inbox.drain(1);
+        assert_eq!(second.messages, vec![vec![2]]);
+        assert!(second.closed);
+        assert!(!second.more);
+        assert_eq!(inbox.push(vec![3]), Push::Closed);
+    }
+
+    #[test]
+    fn draining_releases_byte_budget_for_later_messages() {
+        let inbox = Inbox::new(2, 4, 4);
+        assert_eq!(inbox.push(vec![1, 2, 3, 4]), Push::Wake);
+        assert_eq!(inbox.drain(1).messages, vec![vec![1, 2, 3, 4]]);
+        assert_eq!(inbox.push(vec![5, 6, 7, 8]), Push::Wake);
+        assert_eq!(inbox.drain(1).messages, vec![vec![5, 6, 7, 8]]);
+    }
+
+    #[test]
+    fn oversized_first_message_requests_wakeup_and_reports_refusal() {
+        let inbox = Inbox::new(2, 8, 1);
+        assert_eq!(inbox.push(vec![1, 2]), Push::OverflowWake);
+        let batch = inbox.drain(2);
+        assert!(batch.closed);
+        assert!(batch.messages.is_empty());
+    }
+
+    #[test]
+    fn close_wakes_once_even_without_messages() {
+        let inbox = Inbox::new(4, 16, 8);
+        assert!(inbox.close());
+        assert!(!inbox.close());
+        assert!(inbox.drain(4).closed);
+        assert_eq!(inbox.push(vec![1]), Push::Closed);
+    }
+
+    #[test]
+    fn concurrent_producers_remain_bounded_and_fifo_per_producer() {
+        let inbox = std::sync::Arc::new(Inbox::new(400, 800, 2));
+        std::thread::scope(|scope| {
+            for producer in 0..4u8 {
+                let inbox = inbox.clone();
+                scope.spawn(move || {
+                    for index in 0..100u8 {
+                        assert_ne!(inbox.push(vec![producer, index]), Push::Closed);
+                    }
+                });
+            }
+        });
+        let batch = inbox.drain(400);
+        assert_eq!(batch.messages.len(), 400);
+        assert!(!batch.closed);
+        for producer in 0..4u8 {
+            let sequence: Vec<_> = batch.messages.iter().filter(|m| m[0] == producer).map(|m| m[1]).collect();
+            assert_eq!(sequence, (0..100u8).collect::<Vec<_>>());
+        }
+    }
+}
@@ -155,6 +155,7 @@ pub fn forwarded(event: &PluginEvent) -> bool {
            | PluginEvent::TaskProgress { .. }
            | PluginEvent::TaskEnded { .. }
            | PluginEvent::SubscriptionsChanged
+            | PluginEvent::WebsocketReady { .. }
            | PluginEvent::ViewChanged { .. }
            | PluginEvent::ViewMessage { .. }
            | PluginEvent::OpenTerminal { .. }
@@ -275,6 +275,9 @@ pub enum PluginEvent {
    /// The kinds of workbench events plugins want changed: re-fetch
    /// `plugin_event_kinds` (the UI reports only those).
    SubscriptionsChanged,
+    /// A background WebSocket has messages ready for its owning plugin.
+    #[serde(rename_all = "camelCase")]
+    WebsocketReady { plugin_id: String, connection: u64 },
    #[serde(rename_all = "camelCase")]
    ThemeChanged { theme: Option<ActiveTheme> },
    /// A plugin needs capabilities the user has not granted yet (first use
@@ -0,0 +1,531 @@
+//! Host-owned background WebSocket connections for plugins.
+
+use super::{
+    PluginEvent, PluginEventSink,
+    network_inbox::{Inbox, Push},
+};
+use futures::{SinkExt, StreamExt};
+use ide_plugin::WebSocketBatch;
+use parking_lot::Mutex;
+use std::{
+    collections::HashMap,
+    sync::{
+        Arc, Weak,
+        atomic::{AtomicU64, Ordering},
+    },
+    time::Duration,
+};
+use tokio::sync::{Notify, watch};
+use tokio_tungstenite::{
+    Connector, connect_async_tls_with_config,
+    tungstenite::{
+        Message,
+        client::IntoClientRequest,
+        http::{HeaderName, HeaderValue},
+        protocol::WebSocketConfig,
+    },
+};
+
+const MAX_MESSAGES: usize = 256;
+const MAX_BYTES: usize = 8 << 20;
+const MAX_FRAME: usize = 1 << 20;
+const DRAIN_BATCH: usize = 32;
+const CLOSED_HANDLE_TTL: Duration = Duration::from_secs(60);
+const MAX_CONNECTIONS_PER_PLUGIN: usize = 8;
+
+pub type ConnectionId = u64;
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum ConnectError {
+    Invalid(String),
+    Failed(String),
+}
+
+struct Connection {
+    plugin: String,
+    outgoing: Arc<Inbox>,
+    incoming: Arc<Inbox>,
+    wake: Notify,
+    stop: watch::Sender<bool>,
+}
+
+struct Inner {
+    next: AtomicU64,
+    connections: Mutex<HashMap<ConnectionId, Arc<Connection>>>,
+    pending: Mutex<HashMap<String, usize>>,
+    sink: Option<Arc<dyn PluginEventSink>>,
+}
+
+#[derive(Clone)]
+pub struct WebSocketHub(Arc<Inner>);
+
+impl Default for WebSocketHub {
+    fn default() -> Self {
+        Self(Arc::new(Inner {
+            next: AtomicU64::new(1),
+            connections: Mutex::new(HashMap::new()),
+            pending: Mutex::new(HashMap::new()),
+            sink: None,
+        }))
+    }
+}
+
+impl WebSocketHub {
+    pub fn new(sink: Arc<dyn PluginEventSink>) -> Self {
+        Self(Arc::new(Inner {
+            next: AtomicU64::new(1),
+            connections: Mutex::new(HashMap::new()),
+            pending: Mutex::new(HashMap::new()),
+            sink: Some(sink),
+        }))
+    }
+
+    pub async fn connect(
+        &self,
+        plugin: &str,
+        _project: &str,
+        url: &str,
+        headers: &[(String, String)],
+        subprotocol: &str,
+        timeout: Duration,
+    ) -> Result<ConnectionId, ConnectError> {
+        validate_url(url).map_err(ConnectError::Invalid)?;
+        if subprotocol.len() > 256 {
+            return Err(ConnectError::Invalid("WebSocket subprotocol is too long".into()));
+        }
+        if timeout.is_zero() {
+            return Err(ConnectError::Failed("WebSocket connection timed out".into()));
+        }
+        self.reserve(plugin)?;
+        let result = async {
+            let mut request =
+                url.into_client_request().map_err(|_| ConnectError::Invalid("invalid WebSocket URL".into()))?;
+            for (name, value) in headers {
+                if protected_header(name) {
+                    return Err(ConnectError::Invalid(format!(
+                        "WebSocket header {name:?} is managed by the handshake"
+                    )));
+                }
+                let name: HeaderName =
+                    name.parse().map_err(|_| ConnectError::Invalid("invalid WebSocket header name".into()))?;
+                let value = HeaderValue::from_str(value)
+                    .map_err(|_| ConnectError::Invalid("invalid WebSocket header value".into()))?;
+                request.headers_mut().append(name, value);
+            }
+            if !subprotocol.is_empty() {
+                request.headers_mut().insert(
+                    "sec-websocket-protocol",
+                    subprotocol.parse().map_err(|_| ConnectError::Invalid("invalid WebSocket subprotocol".into()))?,
+                );
+            }
+            let config = WebSocketConfig::default()
+                .read_buffer_size(16 << 10)
+                .write_buffer_size(16 << 10)
+                .max_write_buffer_size(2 << 20)
+                .max_message_size(Some(MAX_FRAME))
+                .max_frame_size(Some(MAX_FRAME));
+            let tls = ide_http::client::tls_config()
+                .map(Arc::new)
+                .map(Connector::Rustls)
+                .map_err(|error| ConnectError::Failed(error.message))?;
+            let (socket, response) =
+                tokio::time::timeout(timeout, connect_async_tls_with_config(request, Some(config), false, Some(tls)))
+                    .await
+                    .map_err(|_| ConnectError::Failed("WebSocket connection timed out".into()))?
+                    .map_err(|_| ConnectError::Failed("WebSocket connection failed".into()))?;
+            if !subprotocol.is_empty()
+                && response.headers().get("sec-websocket-protocol").and_then(|value| value.to_str().ok())
+                    != Some(subprotocol)
+            {
+                return Err(ConnectError::Failed("WebSocket server did not accept the required subprotocol".into()));
+            }
+
+            let id = self.0.next.fetch_add(1, Ordering::Relaxed);
+            let (stop, stopped) = watch::channel(false);
+            let connection = Arc::new(Connection {
+                plugin: plugin.to_owned(),
+                outgoing: Arc::new(Inbox::new(MAX_MESSAGES, MAX_BYTES, MAX_FRAME)),
+                incoming: Arc::new(Inbox::new(MAX_MESSAGES, MAX_BYTES, MAX_FRAME)),
+                wake: Notify::new(),
+                stop,
+            });
+            self.0.connections.lock().insert(id, connection.clone());
+            tokio::spawn(run_socket(id, socket, connection, Arc::downgrade(&self.0), stopped));
+            Ok(id)
+        }
+        .await;
+        self.release(plugin);
+        result
+    }
+
+    pub fn send(&self, plugin: &str, connection: ConnectionId, bytes: Vec<u8>) -> Result<(), String> {
+        let connection = self.connection(plugin, connection)?;
+        match connection.outgoing.push(bytes) {
+            Push::Wake => {
+                connection.wake.notify_one();
+                Ok(())
+            }
+            Push::Queued => Ok(()),
+            Push::OverflowWake => {
+                connection.wake.notify_one();
+                Err("WebSocket send queue is full; message was rejected".into())
+            }
+            Push::OverflowQueued => Err("WebSocket send queue is full; message was rejected".into()),
+            Push::Closed => Err("WebSocket connection is closed; message was rejected".into()),
+        }
+    }
+
+    pub fn drain(&self, plugin: &str, connection: ConnectionId, limit: usize) -> Result<WebSocketBatch, String> {
+        if limit == 0 || limit > DRAIN_BATCH {
+            return Err("invalid drain limit".into());
+        }
+        let connection_state = self.connection(plugin, connection)?;
+        let batch = connection_state.incoming.drain(limit);
+        if batch.more {
+            self.emit_ready(connection, plugin);
+        }
+        if batch.closed && !batch.more {
+            self.0.connections.lock().remove(&connection);
+        }
+        Ok(WebSocketBatch { messages: batch.messages, closed: batch.closed })
+    }
+
+    pub fn disconnect(&self, plugin: &str, id: ConnectionId) -> Result<(), String> {
+        let connection = self
+            .0
+            .connections
+            .lock()
+            .get(&id)
+            .filter(|connection| connection.plugin == plugin)
+            .cloned()
+            .ok_or_else(|| "unknown WebSocket connection".to_owned())?;
+        connection.outgoing.close();
+        let _ = connection.stop.send(true);
+        connection.wake.notify_one();
+        self.0.connections.lock().remove(&id);
+        Ok(())
+    }
+
+    pub fn disconnect_plugin(&self, plugin: &str) {
+        let removed: Vec<_> = {
+            let mut connections = self.0.connections.lock();
+            let ids: Vec<_> =
+                connections.iter().filter(|(_, connection)| connection.plugin == plugin).map(|(id, _)| *id).collect();
+            ids.into_iter().filter_map(|id| connections.remove(&id)).collect()
+        };
+        for connection in removed {
+            connection.outgoing.close();
+            let _ = connection.stop.send(true);
+            connection.wake.notify_one();
+        }
+    }
+
+    fn connection(&self, plugin: &str, id: ConnectionId) -> Result<Arc<Connection>, String> {
+        self.0
+            .connections
+            .lock()
+            .get(&id)
+            .filter(|connection| connection.plugin == plugin)
+            .cloned()
+            .ok_or_else(|| "unknown WebSocket connection".into())
+    }
+
+    fn reserve(&self, plugin: &str) -> Result<(), ConnectError> {
+        let open = self.0.connections.lock().values().filter(|connection| connection.plugin == plugin).count();
+        let mut pending = self.0.pending.lock();
+        if open + pending.get(plugin).copied().unwrap_or_default() >= MAX_CONNECTIONS_PER_PLUGIN {
+            return Err(ConnectError::Invalid("at most 8 WebSocket connections per plugin".into()));
+        }
+        *pending.entry(plugin.to_owned()).or_default() += 1;
+        Ok(())
+    }
+
+    fn release(&self, plugin: &str) {
+        let mut pending = self.0.pending.lock();
+        if let Some(count) = pending.get_mut(plugin) {
+            *count -= 1;
+            if *count == 0 {
+                pending.remove(plugin);
+            }
+        }
+    }
+
+    fn emit_ready(&self, connection: ConnectionId, plugin: &str) {
+        if let Some(sink) = &self.0.sink {
+            sink.emit(None, PluginEvent::WebsocketReady { plugin_id: plugin.to_owned(), connection });
+        }
+    }
+}
+
+fn validate_url(url: &str) -> Result<(), String> {
+    let parsed = reqwest::Url::parse(url).map_err(|_| "invalid WebSocket URL")?;
+    // `host_str` keeps an IPv6 address in brackets.
+    let loopback = parsed.host_str().is_some_and(|host| {
+        let host = host.trim_start_matches('[').trim_end_matches(']');
+        host.eq_ignore_ascii_case("localhost") || host.parse::<std::net::IpAddr>().is_ok_and(|ip| ip.is_loopback())
+    });
+    if parsed.scheme() != "wss" && !(parsed.scheme() == "ws" && loopback) {
+        return Err("WebSocket requires WSS outside loopback".into());
+    }
+    if !parsed.username().is_empty() || parsed.password().is_some() || parsed.fragment().is_some() {
+        return Err("WebSocket URLs may not carry credentials or a fragment".into());
+    }
+    Ok(())
+}
+
+fn protected_header(name: &str) -> bool {
+    name.eq_ignore_ascii_case("host")
+        || name.eq_ignore_ascii_case("connection")
+        || name.eq_ignore_ascii_case("upgrade")
+        || name.get(.."sec-websocket-".len()).is_some_and(|prefix| prefix.eq_ignore_ascii_case("sec-websocket-"))
+}
+
+async fn run_socket<S>(
+    id: ConnectionId,
+    socket: tokio_tungstenite::WebSocketStream<S>,
+    connection: Arc<Connection>,
+    inner: Weak<Inner>,
+    mut stopped: watch::Receiver<bool>,
+) where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
+{
+    let (mut writer, mut reader) = socket.split();
+    loop {
+        tokio::select! {
+            _ = connection.wake.notified() => {
+                let batch = connection.outgoing.drain(DRAIN_BATCH);
+                let mut failed = false;
+                for bytes in batch.messages {
+                    if writer.send(Message::Binary(bytes.into())).await.is_err() {
+                        failed = true;
+                        break;
+                    }
+                }
+                if failed || batch.closed { break; }
+                if batch.more { connection.wake.notify_one(); }
+            }
+            changed = stopped.changed() => {
+                if changed.is_err() || *stopped.borrow() {
+                    let mut failed = false;
+                    loop {
+                        let batch = connection.outgoing.drain(DRAIN_BATCH);
+                        for bytes in batch.messages {
+                            if writer.send(Message::Binary(bytes.into())).await.is_err() {
+                                failed = true;
+                                break;
+                            }
+                        }
+                        if failed || !batch.more { break; }
+                    }
+                    break;
+                }
+            }
+            message = reader.next() => match message {
+                Some(Ok(Message::Binary(bytes))) => {
+                    let pushed = connection.incoming.push(bytes.to_vec());
+                    if matches!(pushed, Push::Wake | Push::OverflowWake)
+                        && let Some(inner) = inner.upgrade()
+                        && let Some(sink) = &inner.sink
+                    {
+                        sink.emit(None, PluginEvent::WebsocketReady { plugin_id: connection.plugin.clone(), connection: id });
+                    }
+                    if matches!(pushed, Push::OverflowWake | Push::OverflowQueued | Push::Closed) { break; }
+                }
+                Some(Ok(Message::Ping(bytes))) => if writer.send(Message::Pong(bytes)).await.is_err() { break; },
+                Some(Ok(Message::Close(_))) | Some(Err(_)) | None => break,
+                _ => {}
+            }
+        }
+    }
+    let _ = writer.close().await;
+    connection.outgoing.close();
+    let wake = connection.incoming.close();
+    if let Some(inner) = inner.upgrade() {
+        if wake && let Some(sink) = &inner.sink {
+            sink.emit(None, PluginEvent::WebsocketReady { plugin_id: connection.plugin.clone(), connection: id });
+        }
+        let retained = connection.clone();
+        tokio::spawn(async move {
+            tokio::time::sleep(CLOSED_HANDLE_TTL).await;
+            let mut connections = inner.connections.lock();
+            if connections.get(&id).is_some_and(|current| Arc::ptr_eq(current, &retained)) {
+                connections.remove(&id);
+            }
+        });
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::net::TcpListener;
+    use tokio_tungstenite::{
+        accept_hdr_async,
+        tungstenite::handshake::server::{ErrorResponse, Request, Response},
+    };
+
+    #[allow(clippy::result_large_err)] // Required by tungstenite's callback signature.
+    fn accept_test_protocol(request: &Request, mut response: Response) -> Result<Response, ErrorResponse> {
+        assert_eq!(request.headers()["authorization"], "Bearer test");
+        assert_eq!(request.headers()["sec-websocket-protocol"], "openidea.test.v1");
+        response.headers_mut().insert("sec-websocket-protocol", "openidea.test.v1".parse().unwrap());
+        Ok(response)
+    }
+
+    async fn server() -> (String, tokio::task::JoinHandle<Vec<Vec<u8>>>) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("ws://{}/socket", listener.local_addr().unwrap());
+        let task = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let socket = accept_hdr_async(stream, accept_test_protocol).await.unwrap();
+            let (_, mut reader) = socket.split();
+            let mut messages = Vec::new();
+            while let Ok(Some(message)) = tokio::time::timeout(Duration::from_secs(2), reader.next()).await {
+                match message.unwrap() {
+                    Message::Binary(bytes) => messages.push(bytes.to_vec()),
+                    Message::Close(_) => break,
+                    _ => {}
+                }
+            }
+            messages
+        });
+        (url, task)
+    }
+
+    #[tokio::test]
+    async fn outbound_message_reaches_the_websocket_server() {
+        let (url, server) = server().await;
+        let hub = WebSocketHub::default();
+        let connection = hub
+            .connect(
+                "plugin",
+                "/project",
+                &url,
+                &[("authorization".into(), "Bearer test".into())],
+                "openidea.test.v1",
+                Duration::from_secs(1),
+            )
+            .await
+            .unwrap();
+        hub.send("plugin", connection, b"message".to_vec()).unwrap();
+        hub.disconnect("plugin", connection).unwrap();
+        assert_eq!(server.await.unwrap(), [b"message"]);
+    }
+
+    #[tokio::test]
+    async fn inbound_messages_can_be_drained_and_disconnect_cleans_the_handle() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("ws://{}/socket", listener.local_addr().unwrap());
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut socket = accept_hdr_async(stream, accept_test_protocol).await.unwrap();
+            socket.send(Message::Binary(vec![1, 2, 3].into())).await.unwrap();
+            socket.close(None).await.unwrap();
+        });
+        let hub = WebSocketHub::default();
+        let connection = hub
+            .connect(
+                "plugin",
+                "/project",
+                &url,
+                &[("authorization".into(), "Bearer test".into())],
+                "openidea.test.v1",
+                Duration::from_secs(1),
+            )
+            .await
+            .unwrap();
+        server.await.unwrap();
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        assert_eq!(
+            hub.drain("plugin", connection, 1).unwrap(),
+            WebSocketBatch { messages: vec![vec![1, 2, 3]], closed: true }
+        );
+        assert!(hub.drain("plugin", connection, 1).is_err());
+    }
+
+    #[test]
+    fn send_rejects_full_queues_without_reporting_success() {
+        let hub = WebSocketHub::default();
+        let (stop, _) = watch::channel(false);
+        let connection = Arc::new(Connection {
+            plugin: "plugin".into(),
+            outgoing: Arc::new(Inbox::new(1, MAX_BYTES, MAX_FRAME)),
+            incoming: Arc::new(Inbox::new(1, MAX_BYTES, MAX_FRAME)),
+            wake: Notify::new(),
+            stop,
+        });
+        hub.0.connections.lock().insert(1, connection.clone());
+
+        assert!(hub.send("other-plugin", 1, b"cross-plugin".to_vec()).is_err());
+        assert!(hub.drain("other-plugin", 1, 1).is_err());
+        assert!(hub.send("plugin", 1, b"first".to_vec()).is_ok());
+        assert!(hub.send("plugin", 1, b"second".to_vec()).unwrap_err().contains("rejected"));
+        assert_eq!(connection.outgoing.drain(1).messages, [b"first"]);
+    }
+
+    #[test]
+    fn websocket_urls_require_tls_and_keep_credentials_out_of_the_url() {
+        assert!(validate_url("wss://relay.example/socket").is_ok());
+        assert!(validate_url("wss://relay.example/socket?channel=main&token=opaque").is_ok());
+        assert!(validate_url("ws://127.0.0.1:9000/socket").is_ok());
+        assert!(validate_url("ws://[::1]:9000/socket").is_ok());
+        assert!(validate_url("ws://localhost:9000/socket").is_ok());
+        assert!(validate_url("ws://relay.example/socket").unwrap_err().contains("WSS"));
+        assert!(validate_url("wss://[email protected]/socket").is_err());
+        assert!(validate_url("wss://relay.example/socket#fragment").is_err());
+        assert!(protected_header("Host"));
+        assert!(protected_header("Connection"));
+        assert!(protected_header("Upgrade"));
+        assert!(protected_header("Sec-WebSocket-Key"));
+        assert!(!protected_header("Cookie"));
+        assert!(!protected_header("X-API-Key"));
+    }
+
+    #[tokio::test]
+    #[allow(clippy::result_large_err)] // Required by tungstenite's callback signature.
+    async fn optional_subprotocol_and_custom_headers_reach_the_handshake() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("ws://{}/socket?channel=main", listener.local_addr().unwrap());
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut socket = accept_hdr_async(stream, |request: &Request, response: Response| {
+                assert_eq!(request.headers()["cookie"], "session=test");
+                assert_eq!(request.headers()["x-api-key"], "key");
+                assert!(request.headers().get("sec-websocket-protocol").is_none());
+                Ok(response)
+            })
+            .await
+            .unwrap();
+            socket.close(None).await.unwrap();
+        });
+        let hub = WebSocketHub::default();
+        let connection = hub
+            .connect(
+                "plugin",
+                "/project",
+                &url,
+                &[("cookie".into(), "session=test".into()), ("x-api-key".into(), "key".into())],
+                "",
+                Duration::from_secs(1),
+            )
+            .await
+            .unwrap();
+        hub.disconnect("plugin", connection).unwrap();
+        server.await.unwrap();
+    }
+
+    #[test]
+    fn connection_limit_is_enforced_before_connecting() {
+        let hub = WebSocketHub::default();
+        for _ in 0..MAX_CONNECTIONS_PER_PLUGIN {
+            hub.reserve("plugin").unwrap();
+        }
+        assert_eq!(
+            hub.reserve("plugin"),
+            Err(ConnectError::Invalid("at most 8 WebSocket connections per plugin".into()))
+        );
+        for _ in 0..MAX_CONNECTIONS_PER_PLUGIN {
+            hub.release("plugin");
+        }
+    }
+}
@@ -69,6 +69,8 @@ pub struct Project {
    /// watchers for (a virtualenv elsewhere, a path dependency).
    lsp_watchers: Mutex<Vec<(PathBuf, Watcher)>>,
    pub(crate) _scratch_watcher: Option<Watcher>,
+    /// Picks up data sources edited on disk outside the Database tool window.
+    _db_sources_watch: Option<ide_db::SourcesWatch>,
    /// A LightEdit window's loose files ([`Host::open_light`]): global settings only,
    /// nothing indexed, no language servers, repositories or project services.
    pub light: bool,
@@ -536,10 +538,12 @@ impl Host {
        tracing::info!("opened {} in window {window} in {} ms {laps:?}", root.display(), started.elapsed().as_millis());

        let (db, http_history, agents) = self.project_services(&root, &project_key, &settings);
+        let db_sources_watch = db_sources_watch(&db, &sink, &self.runtime);
        let notebooks = crate::notebook::NotebookService::new(
            self.local_data_dir.join("notebook-runtime").join(uuid::Uuid::new_v4().to_string()),
            sink.clone(),
        );
+        notebooks.set_observer(Arc::new(NotebookLspObserver(lsp.clone())));
        ide_agent::registry::maintain(crate::ops::agent::registry_paths(self));
        let plugin_diagnostics =
            crate::plugins::diagnostics::Diagnostics::new(sink.clone(), self.runtime.clone(), &documents, &root);
@@ -576,6 +580,7 @@ impl Host {
            content_watchers: Mutex::new(content_watchers),
            lsp_watchers: Mutex::new(Vec::new()),
            _scratch_watcher: scratch_watcher,
+            _db_sources_watch: db_sources_watch,
            files_events_active,
            light: false,
        });
@@ -738,6 +743,7 @@ impl Host {
            content_watchers: Mutex::new(Vec::new()),
            lsp_watchers: Mutex::new(Vec::new()),
            _scratch_watcher: None,
+            _db_sources_watch: None,
            files_events_active: Arc::new(AtomicBool::new(false)),
            light: true,
            root: dir,
@@ -790,6 +796,28 @@ fn documents(sink: &SharedSink, languages: &Arc<LanguageRegistry>, root: Option<
    )
 }

+/// Data sources edited on disk outside the Database tool window (by hand, by a coding
+/// agent): connections made with the old settings close and the tree re-reads them,
+/// without reopening the project.
+fn db_sources_watch(
+    db: &Arc<ide_db::DbManager>,
+    sink: &SharedSink,
+    runtime: &tokio::runtime::Handle,
+) -> Option<ide_db::SourcesWatch> {
+    let (weak, sink, runtime) = (Arc::downgrade(db), sink.clone(), runtime.clone());
+    db.watch_sources(Box::new(move || {
+        let Some(db) = weak.upgrade() else { return };
+        let sink = sink.clone();
+        runtime.spawn(async move {
+            if db.reconcile_sources().await {
+                sink.emit(IdeEvent::DbSourcesChanged);
+            }
+        });
+    }))
+    .inspect_err(|e| tracing::warn!("{}", e.message))
+    .ok()
+}
+
 fn light_watcher(
    dir: &Path,
    sink: &SharedSink,
@@ -1578,6 +1606,25 @@ impl DocumentObserver for LspObserver {
    }
 }

+/// Feeds the code cells of open notebooks to the Python language servers that sync notebooks.
+struct NotebookLspObserver(Arc<LspManager>);
+
+impl crate::notebook::NotebookObserver for NotebookLspObserver {
+    fn changed(&self, path: &Path, snapshot: &ide_notebook::NotebookSnapshot) {
+        let cells: Vec<ide_lsp::NotebookCell> = snapshot
+            .cells
+            .iter()
+            .filter(|c| c.cell_type == "code")
+            .map(|c| ide_lsp::NotebookCell { id: c.id.clone(), source: c.source.clone() })
+            .collect();
+        // Notebooks run Python kernels only.
+        self.0.sync_notebook(path, &LanguageId::new("python"), &cells);
+    }
+    fn closed(&self, path: &Path) {
+        self.0.close_notebook(path);
+    }
+}
+
 /// Feeds document edits into the per-project syntax tree cache.
 struct SyntaxObserver(Arc<SyntaxCache>);

@@ -39,7 +39,7 @@ impl PluginWorkspace for Workspace {
    fn document(&self, _: &Path) -> Option<(u64, String)> {
        None
    }
-    fn apply_edits(&self, _: &Path, _: u64, _: &[ide_core::TextEdit]) -> Result<u64, HostError> {
+    fn apply_edits(&self, _: &Path, _: u64, _: &[ide_core::TextEdit], _: bool) -> Result<u64, HostError> {
        Err(HostError::NotFound("not open".into()))
    }
    fn trusted(&self, _: &Path) -> bool {
@@ -154,9 +154,20 @@ impl PluginWorkspace for TestWorkspace {
        let snap = self.documents.snapshot(self.documents.by_path(path)?).ok()?;
        Some((u64::from(snap.version), snap.text))
    }
-    fn apply_edits(&self, path: &Path, version: u64, edits: &[ide_core::TextEdit]) -> Result<u64, HostError> {
+    fn apply_edits(
+        &self,
+        path: &Path,
+        version: u64,
+        edits: &[ide_core::TextEdit],
+        background: bool,
+    ) -> Result<u64, HostError> {
        let doc = self.documents.by_path(path).ok_or_else(|| HostError::NotFound("not open".into()))?;
-        match self.documents.apply_edits_at(doc, version as u32, edits) {
+        let applied = if background {
+            self.documents.apply_background_edits_at(doc, version as u32, edits)
+        } else {
+            self.documents.apply_edits_at(doc, version as u32, edits)
+        };
+        match applied {
            Ok(Some(v)) => Ok(u64::from(v)),
            Ok(None) => Err(HostError::Stale),
            Err(e) => Err(HostError::Failed(e.message)),
@@ -455,10 +466,10 @@ fn edits_are_refused_when_the_document_changed() {
        range: ide_core::Range::new(ide_core::Pos::new(0, 0), ide_core::Pos::new(0, 3)),
        new_text: "ONE".into(),
    };
-    assert_eq!(f.workspace.apply_edits(&editor.path, doc.0, std::slice::from_ref(&edit)), Err(HostError::Stale));
+    assert_eq!(f.workspace.apply_edits(&editor.path, doc.0, std::slice::from_ref(&edit), false), Err(HostError::Stale));
    assert_eq!(f.workspace.text("src/notes.txt"), "typed one two three\nfour five\n");
    let now = f.workspace.document(&editor.path).unwrap().0;
-    assert_eq!(f.workspace.apply_edits(&editor.path, now, &[edit]), Ok(now + 1));
+    assert_eq!(f.workspace.apply_edits(&editor.path, now, &[edit], false), Ok(now + 1));
    assert_eq!(f.workspace.text("src/notes.txt"), "ONEed one two three\nfour five\n");
    let _ = (
        std::any::type_name::<Runtime>(),
@@ -808,6 +819,28 @@ fn a_reporting_task_outlives_its_background_budget() {
    assert_eq!((info.status, info.stats.crashes, info.stats.errors), (PluginStatus::Active, 0, 0), "{info:?}");
 }

+#[test]
+fn background_tasks_scale_and_renew_cpu_fuel_budget() {
+    let f = fixture(Limits {
+        fuel: 3_500,
+        timeout: Duration::from_millis(500),
+        background: Duration::from_millis(5_000),
+        ..Limits::default()
+    });
+    f.install(&misbehave_pkg());
+    if f.host.set_capability("misbehave", "process:git", true).is_err() {
+        return;
+    }
+    // Warm up activation in its own call so `patient` starts with a fresh 3_500-fuel budget.
+    assert_eq!(f.run("misbehave", "echo", Some("warm")).unwrap().as_deref(), Some("warm"));
+
+    let out = f.run("misbehave", "patient", None).expect("background task succeeds").expect("output");
+    assert_eq!(out, "still here");
+
+    let err = f.run("misbehave", "spin", None).unwrap_err();
+    assert!(err.contains("exceeded its CPU budget (3500 fuel per call)"), "{err}");
+}
+
 #[test]
 fn private_directory_is_owned_persistent_and_removed_on_uninstall() {
    let f = fixture(Limits {
@@ -4807,8 +4840,14 @@ impl PluginWorkspace for EndpointWorkspace {
    fn project_of(&self, _: &str) -> Option<PathBuf> {
        self.roots.lock().first().cloned()
    }
-    fn apply_edits(&self, path: &Path, version: u64, edits: &[ide_core::TextEdit]) -> Result<u64, HostError> {
-        self.inner.apply_edits(path, version, edits)
+    fn apply_edits(
+        &self,
+        path: &Path,
+        version: u64,
+        edits: &[ide_core::TextEdit],
+        background: bool,
+    ) -> Result<u64, HostError> {
+        self.inner.apply_edits(path, version, edits, background)
    }
    fn publish_diagnostics(
        &self,
@@ -6648,6 +6687,73 @@ fn project_write_cannot_replace_the_git_file_of_a_linked_worktree() {
    }
 }

+#[cfg(unix)]
+#[test]
+fn process_timeout_returns_guest_error_without_disabling_plugin() {
+    use std::os::unix::fs::PermissionsExt;
+    let f = fixture(Limits {
+        timeout: Duration::from_millis(120),
+        max_failures: 3,
+        restart_backoff: Duration::ZERO,
+        ..Limits::default()
+    });
+    let package = f._tmp.path().join("api14-slow-proc");
+    copy_dir(&sample("api14"), &package);
+    let manifest = package.join("plugin.toml");
+    let text =
+        std::fs::read_to_string(&manifest).unwrap().replace("optional = [", "optional = [\"process:api14-tool\", ");
+    std::fs::write(manifest, text).unwrap();
+    f.install(&package);
+
+    let slow = f._tmp.path().join("slow-tool");
+    std::fs::write(&slow, "#!/bin/sh\nsleep 5\n").unwrap();
+    std::fs::set_permissions(&slow, std::fs::Permissions::from_mode(0o755)).unwrap();
+    let slow = slow.canonicalize().unwrap();
+    f.host.choose_program("api14", "process:api14-tool", Some(&f.project), Some(&slow)).unwrap();
+
+    // Even past `max_failures` (3), external subprocess timeouts unwind as guest errors
+    // instead of poisoning the Wasm instance and disabling the plugin for the session.
+    for _ in 0..4 {
+        let err = f.run("api14", "process-run", None).unwrap_err();
+        assert!(err.contains("api14-tool timed out"), "{err}");
+    }
+    let info = f.host.info_of("api14").unwrap();
+    assert_eq!(info.status, PluginStatus::Active);
+    assert_eq!(info.stats.crashes, 0);
+}
+
+#[cfg(unix)]
+#[test]
+fn process_stdout_accepts_large_payloads_and_reports_overflow_explicitly() {
+    use std::os::unix::fs::PermissionsExt;
+    let f = fixture(Limits { timeout: Duration::from_secs(10), ..Limits::default() });
+    let package = f._tmp.path().join("api14-large-proc");
+    copy_dir(&sample("api14"), &package);
+    let manifest = package.join("plugin.toml");
+    let text =
+        std::fs::read_to_string(&manifest).unwrap().replace("optional = [", "optional = [\"process:api14-tool\", ");
+    std::fs::write(manifest, text).unwrap();
+    f.install(&package);
+
+    let medium_bytes = 5 * 1024 * 1024;
+    let medium = f._tmp.path().join("medium-tool");
+    std::fs::write(&medium, "#!/bin/sh\ndd if=/dev/zero bs=1048576 count=5 2>/dev/null | tr '\\0' 'x'\n").unwrap();
+    std::fs::set_permissions(&medium, std::fs::Permissions::from_mode(0o755)).unwrap();
+    let medium = medium.canonicalize().unwrap();
+    f.host.choose_program("api14", "process:api14-tool", Some(&f.project), Some(&medium)).unwrap();
+    let out = f.run("api14", "process-run", None).unwrap().unwrap();
+    assert_eq!(out.len(), medium_bytes);
+
+    let huge = f._tmp.path().join("huge-tool");
+    std::fs::write(&huge, "#!/bin/sh\ndd if=/dev/zero bs=1048576 count=17 2>/dev/null | tr '\\0' 'x'\n").unwrap();
+    std::fs::set_permissions(&huge, std::fs::Permissions::from_mode(0o755)).unwrap();
+    let huge = huge.canonicalize().unwrap();
+    f.host.choose_program("api14", "process:api14-tool", Some(&f.project), Some(&huge)).unwrap();
+    let err = f.run("api14", "process-run", None).unwrap_err();
+    assert!(err.contains("stdout exceeded 16 MiB limit"), "{err}");
+    assert_eq!(f.host.info_of("api14").unwrap().stats.crashes, 0);
+}
+
 /// A link already in the project is written through, so it is judged by where it points.
 #[cfg(unix)]
 #[test]
@@ -29,17 +29,23 @@ pub fn build(proxy: &ProxySettings) -> IdeResult<reqwest::Client> {
    build_with_redirect(proxy, reqwest::redirect::Policy::default())
 }

-/// A client with an explicit redirect policy, for callers enforcing network grants.
-pub fn build_with_redirect(proxy: &ProxySettings, redirect: reqwest::redirect::Policy) -> IdeResult<reqwest::Client> {
+/// The backend's TLS configuration: ring with the operating system's trust store.
+pub fn tls_config() -> IdeResult<rustls::ClientConfig> {
    use rustls_platform_verifier::BuilderVerifierExt;
-    let fail = |e: String| IdeError::internal(format!("HTTP client: {e}"));
+    let fail = |e: String| IdeError::internal(format!("TLS client: {e}"));
    let provider = Arc::new(rustls::crypto::ring::default_provider());
-    let tls = rustls::ClientConfig::builder_with_provider(provider)
+    rustls::ClientConfig::builder_with_provider(provider)
        .with_safe_default_protocol_versions()
        .map_err(|e| fail(e.to_string()))?
        .with_platform_verifier()
-        .map_err(|e| fail(e.to_string()))?
-        .with_no_client_auth();
+        .map_err(|e| fail(e.to_string()))
+        .map(|builder| builder.with_no_client_auth())
+}
+
+/// A client with an explicit redirect policy, for callers enforcing network grants.
+pub fn build_with_redirect(proxy: &ProxySettings, redirect: reqwest::redirect::Policy) -> IdeResult<reqwest::Client> {
+    let fail = |e: String| IdeError::internal(format!("HTTP client: {e}"));
+    let tls = tls_config()?;
    // `localhost` is tried on IPv4 first: local servers (Ollama, dev servers)
    // usually listen there, and on Windows a refused `::1` costs ~2 s before
    // the system resolver's order falls back. Ports come from the URL.
@@ -5,10 +5,11 @@
 use std::path::{Path, PathBuf};

 /// The outermost ancestor of `file` containing a root marker, within
-/// `project_root` (any ancestor without one: loose files).
+/// `project_root` (any ancestor without one: loose files). A directory counts
+/// as its own ancestor: a module folder is served like the files in it.
 pub fn marked_root(project_root: Option<&Path>, file: &Path, markers: &[String]) -> Option<PathBuf> {
    let mut best = None;
-    for dir in file.ancestors().skip(1) {
+    for dir in file.ancestors().skip(usize::from(!file.is_dir())) {
        if project_root.is_some_and(|root| !dir.starts_with(root)) {
            break;
        }
@@ -112,6 +113,20 @@ mod tests {
        assert_eq!(marked_root(Some(root), &root.join("apps/api/main.ts"), &markers), None);
    }

+    #[test]
+    fn a_marked_directory_is_its_own_root() {
+        let tmp = tempfile::tempdir().unwrap();
+        let root = tmp.path();
+        let module = root.join("first");
+        std::fs::create_dir_all(module.join("src")).unwrap();
+        std::fs::write(module.join("pom.xml"), "").unwrap();
+        let markers = vec!["pom.xml".to_string()];
+        assert_eq!(marked_root(Some(root), &module.join("src/Main.java"), &markers), Some(module.clone()));
+        assert_eq!(marked_root(Some(root), &module, &markers), Some(module.clone()));
+        assert_eq!(marked_root(Some(root), &module.join("src"), &markers), Some(module));
+        assert_eq!(marked_root(Some(root), &root.join("second"), &markers), None);
+    }
+
    #[test]
    fn markers_above_the_project_root_only_count_without_one() {
        let tmp = tempfile::tempdir().unwrap();
@@ -163,7 +163,9 @@ pub fn client_capabilities() -> Value {
            "typeHierarchy": { "dynamicRegistration": false }
        },
        // rust-analyzer: report when the project has loaded (`quiescent`).
-        "experimental": { "serverStatusNotification": true }
+        "experimental": { "serverStatusNotification": true },
+        // Jupyter notebooks: code cells are synced with `notebookDocument/*` (see `notebook.rs`).
+        "notebookDocument": { "synchronization": { "dynamicRegistration": false, "executionSummarySupport": false } }
    })
 }

@@ -426,6 +428,34 @@ impl ServerCaps {
            .unwrap_or_default()
    }

+    /// Whether `notebookDocumentSync` takes Jupyter notebooks (`file:`) with cells of `language`:
+    /// a selector that names no notebook, or `jupyter-notebook` / `*`, and no cell languages or
+    /// `language` among them.
+    pub fn syncs_notebooks(&self, language: &str) -> bool {
+        let Some(selectors) = self.raw.pointer("/notebookDocumentSync/notebookSelector").and_then(Value::as_array)
+        else {
+            return false;
+        };
+        selectors.iter().any(|s| {
+            let notebook = match s.get("notebook") {
+                None | Some(Value::Null) => true,
+                Some(Value::String(kind)) => kind == "jupyter-notebook" || kind == "*",
+                Some(filter) => {
+                    filter
+                        .get("notebookType")
+                        .and_then(Value::as_str)
+                        .is_none_or(|t| t == "jupyter-notebook" || t == "*")
+                        && filter.get("scheme").and_then(Value::as_str).is_none_or(|s| s == "file")
+                }
+            };
+            let cells = match s.get("cells").and_then(Value::as_array) {
+                None => true,
+                Some(cells) => cells.iter().any(|c| c.get("language").and_then(Value::as_str) == Some(language)),
+            };
+            notebook && cells && (s.get("notebook").is_some() || s.get("cells").is_some())
+        })
+    }
+
    /// `workspace.fileOperations.<op>` registration options (`op`: `willRename`, `didRename`).
    pub fn file_operation(&self, op: &str) -> Option<&Value> {
        self.raw.pointer(&format!("/workspace/fileOperations/{op}")).filter(|v| v.is_object())
@@ -668,6 +698,26 @@ mod tests {
        assert_eq!(c.code_action_kinds(), Some(vec!["refactor.inline".into()]));
    }

+    #[test]
+    fn notebook_sync_selectors() {
+        let syncs = |selector: Value| {
+            ServerCaps::parse(json!({ "notebookDocumentSync": { "notebookSelector": [selector] } }))
+                .syncs_notebooks("python")
+        };
+        // basedpyright
+        assert!(syncs(json!({
+            "notebook": { "scheme": "file", "notebookType": "jupyter-notebook" }, "cells": [{ "language": "python" }]
+        })));
+        // Ruff: any notebook with Python cells.
+        assert!(syncs(json!({ "cells": [{ "language": "python" }] })));
+        assert!(syncs(json!({ "notebook": "*" })));
+        assert!(!syncs(json!({ "notebook": "interactive" })));
+        assert!(!syncs(json!({ "notebook": { "notebookType": "jupyter-notebook", "scheme": "untitled" } })));
+        assert!(!syncs(json!({ "cells": [{ "language": "r" }] })));
+        assert!(!syncs(json!({})));
+        assert!(!ServerCaps::parse(json!({ "textDocumentSync": 2 })).syncs_notebooks("python"));
+    }
+
    #[test]
    fn language_definitions_use_known_feature_names() {
        for lang in ide_language::LanguageRegistry::load(None).unwrap().all() {
@@ -966,10 +966,12 @@ pub fn inlay_hint(conv: &Conv, path: &Path, h: lsp::InlayHint) -> InlayHint {
    }
 }

-/// Decodes relative semantic token data into absolute UTF-16 quintuples.
-pub fn decode_semantic_tokens(data: &[u32], enc: Encoding, text: Option<&DocText>) -> Vec<u32> {
-    let mut out = Vec::with_capacity(data.len());
-    let (mut line, mut start) = (0u32, 0u32);
+/// Decodes relative semantic token data into absolute UTF-16 quintuples, sorted by position.
+/// Deltas are read as signed because some servers (JDT LS) emit tokens out of order; tokens that
+/// land before the document start are dropped and exact duplicates removed.
+pub fn decode_semantic_tokens(data: &[i64], enc: Encoding, text: Option<&DocText>) -> Vec<u32> {
+    let mut out: Vec<[u32; 5]> = Vec::with_capacity(data.len() / 5);
+    let (mut line, mut start) = (0i64, 0i64);
    for &[dl, ds, len, ty, mods] in data.as_chunks::<5>().0 {
        if dl == 0 {
            start = start.saturating_add(ds);
@@ -977,16 +979,21 @@ pub fn decode_semantic_tokens(data: &[u32], enc: Encoding, text: Option<&DocText
            line = line.saturating_add(dl);
            start = ds;
        }
+        let (Ok(line), Ok(start), Ok(len)) = (u32::try_from(line), u32::try_from(start), u32::try_from(len)) else {
+            continue;
+        };
        let (col, len16) = match (enc, text) {
            (Encoding::Utf16, _) | (_, None) => (start, len),
            (_, Some(t)) => {
-                let (a, b) = convert_offsets(t.line(line as usize), start as usize, (start + len) as usize, enc);
+                let (a, b) = convert_offsets(t.line(line as usize), start as usize, start as usize + len as usize, enc);
                (a, b - a)
            }
        };
-        out.extend_from_slice(&[line, col, len16, ty, mods]);
+        out.push([line, col, len16, ty as u32, mods as u32]);
    }
-    out
+    out.sort_by_key(|t| (t[0], t[1]));
+    out.dedup();
+    out.into_flattened()
 }

 pub fn code_lens(conv: &Conv, path: &Path, l: lsp::CodeLens) -> CodeLens {
@@ -1473,7 +1480,7 @@ mod tests {

    #[test]
    fn semantic_token_decoding() {
-        let data = [1, 2, 3, 0, 1, 0, 5, 2, 1, 0, 2, 0, 4, 2, 3];
+        let data: [i64; 15] = [1, 2, 3, 0, 1, 0, 5, 2, 1, 0, 2, 0, 4, 2, 3];
        let out = decode_semantic_tokens(&data, Encoding::Utf16, None);
        assert_eq!(out, vec![1, 2, 3, 0, 1, 1, 7, 2, 1, 0, 3, 0, 4, 2, 3]);
        // UTF-8 server: "😀ab" token "ab" at byte 4 len 2 → utf16 col 2 len 2.
@@ -1481,6 +1488,28 @@ mod tests {
        assert_eq!(decode_semantic_tokens(&[0, 4, 2, 0, 0], Encoding::Utf8, Some(&text)), vec![0, 2, 2, 0, 0]);
    }

+    #[test]
+    fn semantic_tokens_out_of_order() {
+        // JDT LS for `public class Repro<T extends Number> extends\n        ArrayList<T> {`: the bound's
+        // `extends` comes after `Number`, the superclass is emitted twice and the line goes backwards.
+        let data: [i64; 95] = [
+            0, 8, 4, 0, 0, 2, 7, 4, 0, 4096, 0, 5, 4, 0, 4096, 0, 5, 9, 1, 5184, 2, 0, 6, 11, 0, 0, 7, 5, 11, 0, 0, 6,
+            5, 1, 1104, 0, 6, 1, 6, 16, 0, 10, 6, 1, 65, 0, -8, 7, 11, 0, 1, 8, 9, 1, 1088, 0, 10, 1, 6, 2048, -1, 37,
+            7, 11, 0, 1, 8, 9, 1, 1088, 0, 10, 1, 6, 2048, 1, 4, 7, 11, 0, 0, 12, 5, 8, 144, 2, 9, 1, 7, 1040, 1, 8, 5,
+            8, 128,
+        ];
+        let out = decode_semantic_tokens(&data, Encoding::Utf16, None);
+        let positions: Vec<(u32, u32)> = out.chunks(5).map(|t| (t[0], t[1])).collect();
+        #[rustfmt::skip]
+        assert_eq!(positions, [
+            (0, 8), (2, 7), (2, 12), (2, 17),
+            (4, 0), (4, 7), (4, 13), (4, 19), (4, 21), (4, 29), (4, 37),
+            (5, 8), (5, 18), (6, 4), (6, 16), (8, 9), (9, 8),
+        ]);
+        // A token that would start before the document is dropped rather than wrapped.
+        assert_eq!(decode_semantic_tokens(&[0, -1, 2, 0, 0, 0, 5, 1, 0, 0], Encoding::Utf16, None), [0, 4, 1, 0, 0]);
+    }
+
    #[test]
    fn diagnostics_mapping() {
        let conv = Conv::utf16();
@@ -10,6 +10,7 @@ use ide_core::{IdeEvent, TaskId};
 use ide_jsonrpc::{Handler, codes};
 use lsp_types as lsp;
 use serde_json::{Value, json};
+use std::path::PathBuf;
 use std::sync::{Arc, Weak};
 use std::time::Duration;

@@ -229,7 +230,11 @@ impl ClientHandler {
                return;
            }
        };
-        let Some(path) = uri_to_path(params.uri.as_str()) else { return };
+        let Some(path) = uri_to_path(params.uri.as_str())
+            .or_else(|| crate::notebook::is_cell_uri(params.uri.as_str()).then(|| PathBuf::from(params.uri.as_str())))
+        else {
+            return;
+        };
        if self.server.name == "jdtls"
            && let Some(failure) = crate::jdtls::build_plan_failure(&path, &self.server.root, &params.diagnostics)
        {
@@ -25,6 +25,7 @@ mod kotlin_lsp;
 mod launch;
 mod library_source;
 mod loading;
+mod notebook;
 mod pull;
 mod scope;
 mod server;
@@ -42,6 +43,7 @@ pub use inspect::{FileInspection, InspectHost, InspectOptions};
 pub use jdtls::jvm_inspect::JdtJvmInspector;
 pub use kotlin_lsp::{KotlinDebugServer, KotlinDebugUnavailable};
 pub use launch::{LaunchError, Launcher, ProcessLauncher};
+pub use notebook::{NotebookCell, NotebookCellDoc, cell_uri};
 pub use scope::{CandidateFiles, CandidateFn, CandidateQuery, MAX_CANDIDATES};
 pub use uri::{is_virtual_uri, path_to_uri_string, uri_target, uri_to_path};

@@ -115,6 +117,8 @@ struct DocState {
    servers: Vec<Attached>,
    /// Open YAML document whose filename does not match a Compose glob and whose content is Compose.
    compose_detected: bool,
+    /// The notebook a code cell belongs to: synced through `notebookDocument/*`, not as a file.
+    notebook: Option<PathBuf>,
 }

 impl DocState {
@@ -229,6 +233,8 @@ pub struct LspManager {
    /// Contents fetched for virtual URIs.
    virtual_docs: Mutex<HashMap<String, VirtualDocument>>,
    injections: injections::State,
+    /// Open notebooks by path. Locked before `docs` when both are.
+    notebooks: Mutex<HashMap<PathBuf, notebook::NotebookState>>,
 }

 /// Fetched virtual documents kept before the cache is reset.
@@ -476,6 +482,7 @@ impl LspManager {
            virtual_origins: Mutex::new(HashMap::new()),
            virtual_docs: Mutex::new(HashMap::new()),
            injections: injections::State::default(),
+            notebooks: Mutex::new(HashMap::new()),
        })
    }

@@ -614,8 +621,13 @@ impl LspManager {
    /// enabled in settings.
    fn attach_fallbacks(&self) {
        let _gate = self.injections.gate.lock();
-        let open: Vec<(DocId, LanguageId, PathBuf)> =
-            self.docs.lock().iter().map(|(id, d)| (*id, d.language.clone(), d.path.clone())).collect();
+        let open: Vec<(DocId, LanguageId, PathBuf)> = self
+            .docs
+            .lock()
+            .iter()
+            .filter(|(_, d)| d.notebook.is_none())
+            .map(|(id, d)| (*id, d.language.clone(), d.path.clone()))
+            .collect();
        let wanted: Vec<(DocId, Vec<Arc<Server>>)> =
            open.into_iter().map(|(id, language, path)| (id, self.servers_for(&language, &path))).collect();
        let mut start = Vec::new();
@@ -648,6 +660,11 @@ impl LspManager {
        for (doc, server) in opened {
            self.schedule_pull(doc, Some(&server), Duration::ZERO);
        }
+        for server in self.attach_notebook_servers() {
+            if !start.iter().any(|s| s.id == server.id) {
+                start.push(server);
+            }
+        }
        for server in &start {
            self.ensure_started(server);
        }
@@ -756,7 +773,7 @@ impl LspManager {
            Err(LaunchError::Missing) => {
                let cfg = server.effective_config();
                let message =
-                    cfg.install_hint.clone().unwrap_or_else(|| format!("`{}` was not found on PATH", cfg.command));
+                    cfg.install_hint.clone().unwrap_or_else(|| format!("{} was not found on PATH", cfg.command));
                if server.fail_start(
                    generation,
                    Conn::Missing,
@@ -867,7 +884,7 @@ impl LspManager {
                false
            } else {
                let open_close = run.caps.read().open_close;
-                for (id, d) in docs.iter_mut() {
+                for (id, d) in docs.iter_mut().filter(|(_, d)| d.notebook.is_none()) {
                    let params = d.did_open_params();
                    for a in
                        d.servers.iter_mut().filter(|a| a.server.id == server.id && a.opened_gen != Some(generation))
@@ -889,6 +906,7 @@ impl LspManager {
        for doc in reopened {
            self.schedule_pull(doc, Some(&server), Duration::ZERO);
        }
+        self.reopen_notebooks(&server, &run);
        self.check_clang_tidy(&server).await;
        self.check_vue_navigation(&server).await;
        server.refresh_busy_status(&self.sink);
@@ -1083,6 +1101,7 @@ impl LspManager {
            text: Arc::new(DocText::new(text)),
            servers: servers.iter().map(|s| Attached { server: s.clone(), opened_gen: None }).collect(),
            compose_detected,
+            notebook: None,
        };
        let compose_added = {
            let mut docs = self.docs.lock();
@@ -1214,7 +1233,7 @@ impl LspManager {
            return;
        }
        let docs = self.docs.lock();
-        for d in docs.values() {
+        for d in docs.values().filter(|d| d.notebook.is_none()) {
            if !d.servers.iter().any(|a| a.server.id == server.id && a.opened_gen == Some(generation)) {
                continue;
            }
@@ -1288,13 +1307,23 @@ impl LspManager {
    fn snapshot(&self, doc: DocId) -> IdeResult<DocSnap> {
        let docs = self.docs.lock();
        let d = docs.get(&doc).ok_or_else(|| IdeError::not_found(format!("{doc} is not open")))?;
+        // A cell is asked only of the servers its notebook is open on (those that sync notebooks).
+        let servers = d
+            .servers
+            .iter()
+            .filter(|a| {
+                d.notebook.is_none()
+                    || a.opened_gen.is_some_and(|g| a.server.running().is_some_and(|run| run.generation == g))
+            })
+            .map(|a| a.server.clone())
+            .collect();
        Ok(DocSnap {
            version: d.version,
            path: d.path.clone(),
            uri: d.uri.clone(),
            language: d.language.clone(),
            text: d.text.clone(),
-            servers: d.servers.iter().map(|a| a.server.clone()).collect(),
+            servers,
        })
    }

@@ -2822,7 +2851,7 @@ impl LspManager {
        let params = json!({ "textDocument": snap.ident() });
        let r: Option<Value> = self.call(t, "textDocument/semanticTokens/full", params, None, DEFAULT_TIMEOUT).await?;
        let Some(data) = r.as_ref().and_then(|v| v.get("data")).and_then(Value::as_array) else { return Ok(None) };
-        let data: Vec<u32> = data.iter().map(|n| n.as_u64().unwrap_or(0) as u32).collect();
+        let data: Vec<i64> = data.iter().map(|n| n.as_i64().unwrap_or(0)).collect();
        let (enc, legend) = {
            let caps = t.caps();
            (caps.encoding, caps.semantic_legend.clone().unwrap_or_default())
@@ -3490,11 +3519,17 @@ impl LspManager {
            join_all(servers.iter().map(|s| self.stop_instance(s, true))).await;
            return;
        }
+        let existing: Vec<ide_core::ServerId> = self.servers.lock().iter().map(|s| s.id).collect();
        self.reconfigure_jdtls_workspace().await;
        self.stop_unlisted().await;
        let Some(me) = self.me.upgrade() else { return };
        let servers: Vec<Arc<Server>> = self.servers.lock().clone();
        for s in servers {
+            // Created in this pass (the jdtls of a new workspace layout): it starts with the
+            // current configuration, though it may not have recorded it yet.
+            if !existing.contains(&s.id) {
+                continue;
+            }
            // The settings hook may run tools (`npm root -g`), as when the server starts.
            let next = match tokio::task::spawn_blocking({
                let (me, s) = (me.clone(), s.clone());
@@ -0,0 +1,382 @@
+//! Jupyter notebooks on servers that sync them with LSP 3.17 `notebookDocument/*` (basedpyright,
+//! Ruff). Each code cell is a document of its own, `vscode-notebook-cell:<notebook path>#<cell id>`,
+//! kept in `docs` under a [`DocId`]: completion, hover and the other requests address a cell like
+//! any open file. Only opening, changing and closing go through the notebook methods; a change in
+//! the cells' order closes and reopens the notebook.
+
+use crate::server::{Running, Server};
+use crate::text::DocText;
+use crate::{Attached, DocState, LspManager, uri};
+use ide_core::{DocId, LanguageId, ServerId};
+use lsp_types as lsp;
+use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, utf8_percent_encode};
+use serde::Serialize;
+use serde_json::{Value, json};
+use specta::Type;
+use std::path::{Path, PathBuf};
+use std::str::FromStr;
+use std::sync::Arc;
+
+const CELL_SCHEME: &str = "vscode-notebook-cell";
+const NOTEBOOK_TYPE: &str = "jupyter-notebook";
+/// `NotebookCellKind.Code`.
+const CODE_CELL: u32 = 2;
+/// Escaped in a cell id: nbformat ids are `[a-zA-Z0-9-_]`, older notebooks may have others.
+const FRAGMENT: &AsciiSet = &NON_ALPHANUMERIC.remove(b'-').remove(b'_');
+
+/// A code cell as the notebook holds it.
+pub struct NotebookCell {
+    pub id: String,
+    pub source: String,
+}
+
+/// The document the language servers know a code cell by.
+#[derive(Clone, Debug, PartialEq, Serialize, Type)]
+#[serde(rename_all = "camelCase")]
+pub struct NotebookCellDoc {
+    pub cell: String,
+    pub doc: DocId,
+    /// Also the `path` of the cell's diagnostics and of locations in it.
+    pub uri: String,
+}
+
+pub(crate) struct NotebookState {
+    uri: lsp::Uri,
+    language: LanguageId,
+    version: i32,
+    /// Code cells in notebook order.
+    cells: Vec<(String, DocId)>,
+    servers: Vec<Arc<Server>>,
+    /// Servers it is open on, by instance generation.
+    opened: Vec<(ServerId, u64)>,
+}
+
+/// The URI of cell `cell` of the notebook at `path`: the notebook's `file:` URI with the cell
+/// scheme and the cell id as fragment (as VS Code names cells; basedpyright expects the scheme).
+pub fn cell_uri(path: &Path, cell: &str) -> String {
+    let file = uri::path_to_uri_string(path);
+    let rest = file.strip_prefix("file://").unwrap_or(&file);
+    // `file:///C:/a` → `…:/C:/a`; a share (`file://server/s/a`) keeps its authority.
+    let authority = if rest.starts_with('/') { "" } else { "//" };
+    format!("{CELL_SCHEME}:{authority}{rest}#{}", utf8_percent_encode(cell, FRAGMENT))
+}
+
+pub(crate) fn is_cell_uri(uri: &str) -> bool {
+    uri.strip_prefix(CELL_SCHEME).is_some_and(|rest| rest.starts_with(':'))
+}
+
+fn parse_uri(s: &str) -> Option<lsp::Uri> {
+    lsp::Uri::from_str(s).inspect_err(|e| tracing::debug!("not a valid URI {s}: {e}")).ok()
+}
+
+impl NotebookState {
+    fn opened_on(&self, server: ServerId, generation: u64) -> bool {
+        self.opened.contains(&(server, generation))
+    }
+
+    /// The servers it is open on whose instance is still the one it was opened on.
+    fn open_targets(&self) -> Vec<(Arc<Server>, Running)> {
+        self.servers
+            .iter()
+            .filter_map(|s| {
+                let run = s.running()?;
+                self.opened_on(s.id, run.generation).then(|| (s.clone(), run))
+            })
+            .collect()
+    }
+}
+
+impl LspManager {
+    /// Opens the notebook at `path` on its language's servers that sync notebooks, or brings
+    /// them up to date with `cells` (its code cells, in order).
+    pub fn sync_notebook(&self, path: &Path, language: &LanguageId, cells: &[NotebookCell]) {
+        let mut notebooks = self.notebooks.lock();
+        let mut changed = Vec::new();
+        let start = match notebooks.get_mut(path) {
+            None => {
+                let Some(notebook_uri) = parse_uri(&uri::path_to_uri_string(path)) else { return };
+                let servers = self.servers_for(language, path);
+                let mut state = NotebookState {
+                    uri: notebook_uri,
+                    language: language.clone(),
+                    version: 1,
+                    cells: Vec::new(),
+                    servers: servers.clone(),
+                    opened: Vec::new(),
+                };
+                self.set_cells(path, &mut state, cells);
+                self.open_notebook(&mut state, None);
+                notebooks.insert(path.to_owned(), state);
+                servers
+            }
+            Some(state) => {
+                let same_cells =
+                    state.cells.len() == cells.len() && state.cells.iter().zip(cells).all(|((id, _), c)| *id == c.id);
+                if same_cells {
+                    changed = self.change_cells(state, cells);
+                } else {
+                    self.close_notebook_on_servers(state);
+                    state.version += 1;
+                    self.set_cells(path, state, cells);
+                    self.open_notebook(state, None);
+                }
+                Vec::new()
+            }
+        };
+        drop(notebooks);
+        for server in &start {
+            self.ensure_started(server);
+        }
+        for doc in changed {
+            self.schedule_pull(doc, None, crate::pull::AFTER_EDIT);
+        }
+    }
+
+    /// Closes the notebook at `path` on its servers and forgets its cells.
+    pub fn close_notebook(&self, path: &Path) {
+        let Some(mut state) = self.notebooks.lock().remove(path) else { return };
+        self.close_notebook_on_servers(&mut state);
+        let removed: Vec<DocId> = state.cells.iter().map(|(_, doc)| *doc).collect();
+        self.remove_cells(&state.servers, &removed);
+    }
+
+    /// The documents of the notebook's code cells, in notebook order; empty when it isn't open.
+    pub fn notebook_cells(&self, path: &Path) -> Vec<NotebookCellDoc> {
+        let notebooks = self.notebooks.lock();
+        let Some(state) = notebooks.get(path) else { return Vec::new() };
+        let docs = self.docs.lock();
+        state
+            .cells
+            .iter()
+            .filter_map(|(cell, doc)| {
+                Some(NotebookCellDoc { cell: cell.clone(), doc: *doc, uri: docs.get(doc)?.uri.as_str().to_owned() })
+            })
+            .collect()
+    }
+
+    /// Opens the notebooks of `server`'s language on its instance that just became ready.
+    pub(crate) fn reopen_notebooks(&self, server: &Arc<Server>, run: &Running) {
+        let mut pull = Vec::new();
+        {
+            let mut notebooks = self.notebooks.lock();
+            for state in notebooks.values_mut() {
+                if state.servers.iter().any(|s| s.id == server.id) && !state.opened_on(server.id, run.generation) {
+                    self.open_notebook(state, Some(server));
+                    pull.extend(state.cells.iter().map(|(_, doc)| *doc));
+                }
+            }
+        }
+        for doc in pull {
+            self.schedule_pull(doc, Some(server), std::time::Duration::ZERO);
+        }
+    }
+
+    /// Adds servers of the notebooks' languages that were not wanted when they opened (a
+    /// fallback for a missing server, one enabled in settings) and returns them to be started.
+    pub(crate) fn attach_notebook_servers(&self) -> Vec<Arc<Server>> {
+        let mut start = Vec::new();
+        let mut notebooks = self.notebooks.lock();
+        for (path, state) in notebooks.iter_mut() {
+            let added: Vec<Arc<Server>> = self
+                .servers_for(&state.language, path)
+                .into_iter()
+                .filter(|s| !state.servers.iter().any(|known| known.id == s.id))
+                .collect();
+            if added.is_empty() {
+                continue;
+            }
+            {
+                let mut docs = self.docs.lock();
+                for (_, doc) in &state.cells {
+                    if let Some(d) = docs.get_mut(doc) {
+                        d.servers.extend(added.iter().map(|s| Attached { server: s.clone(), opened_gen: None }));
+                    }
+                }
+            }
+            state.servers.extend(added.iter().cloned());
+            for server in &added {
+                self.open_notebook(state, Some(server));
+            }
+            start.extend(added);
+        }
+        start
+    }
+
+    /// Replaces the cell documents with `cells`: kept cells keep their [`DocId`].
+    fn set_cells(&self, path: &Path, state: &mut NotebookState, cells: &[NotebookCell]) {
+        let mut docs = self.docs.lock();
+        let previous = std::mem::take(&mut state.cells);
+        for cell in cells {
+            let kept = previous.iter().find(|(id, _)| *id == cell.id).map(|(_, doc)| *doc);
+            if let Some(doc) = kept
+                && let Some(d) = docs.get_mut(&doc)
+            {
+                if d.text.as_str() != cell.source {
+                    Arc::make_mut(&mut d.text).set(&cell.source);
+                }
+                d.version += 1;
+                state.cells.push((cell.id.clone(), doc));
+                continue;
+            }
+            let uri_string = cell_uri(path, &cell.id);
+            let Some(cell_uri) = parse_uri(&uri_string) else { continue };
+            let doc = DocId::next();
+            let key = PathBuf::from(&uri_string);
+            docs.insert(
+                doc,
+                DocState {
+                    canonical: key.clone(),
+                    path: key,
+                    uri: cell_uri,
+                    language: state.language.clone(),
+                    version: 1,
+                    text: Arc::new(DocText::new(&cell.source)),
+                    servers: state.servers.iter().map(|s| Attached { server: s.clone(), opened_gen: None }).collect(),
+                    compose_detected: false,
+                    notebook: Some(path.to_owned()),
+                },
+            );
+            state.cells.push((cell.id.clone(), doc));
+        }
+        drop(docs);
+        let removed: Vec<DocId> = previous
+            .into_iter()
+            .filter(|(_, doc)| !state.cells.iter().any(|(_, d)| d == doc))
+            .map(|(_, d)| d)
+            .collect();
+        self.remove_cells(&state.servers, &removed);
+    }
+
+    /// Sends the cells whose text differs from `cells` as one change; returns their documents.
+    fn change_cells(&self, state: &mut NotebookState, cells: &[NotebookCell]) -> Vec<DocId> {
+        let mut docs = self.docs.lock();
+        let mut changed = Vec::new();
+        let mut content = Vec::new();
+        for ((_, doc), cell) in state.cells.iter().zip(cells) {
+            let Some(d) = docs.get_mut(doc) else { continue };
+            if d.text.as_str() == cell.source {
+                continue;
+            }
+            Arc::make_mut(&mut d.text).set(&cell.source);
+            d.version += 1;
+            content.push(json!({
+                "document": { "uri": d.uri, "version": d.version },
+                "changes": [{ "text": cell.source }],
+            }));
+            changed.push(*doc);
+        }
+        drop(docs);
+        if content.is_empty() {
+            return changed;
+        }
+        state.version += 1;
+        let params = json!({
+            "notebookDocument": { "uri": state.uri, "version": state.version },
+            "change": { "cells": { "textContent": content } },
+        });
+        for (server, run) in state.open_targets() {
+            if let Err(e) = run.peer.notify("notebookDocument/didChange", params.clone()) {
+                tracing::debug!("{}: notebookDocument/didChange: {e}", server.name);
+            }
+        }
+        changed
+    }
+
+    /// Opens the notebook on its running servers that sync notebooks (`only`: on that one).
+    fn open_notebook(&self, state: &mut NotebookState, only: Option<&Arc<Server>>) {
+        let mut docs = self.docs.lock();
+        let mut cells = Vec::new();
+        let mut texts = Vec::new();
+        for (_, doc) in &state.cells {
+            let Some(d) = docs.get(doc) else { continue };
+            cells.push(json!({ "kind": CODE_CELL, "document": d.uri }));
+            texts.push(json!({
+                "uri": d.uri,
+                "languageId": d.language.as_str(),
+                "version": d.version,
+                "text": d.text.as_str(),
+            }));
+        }
+        let params = json!({
+            "notebookDocument": { "uri": state.uri, "notebookType": NOTEBOOK_TYPE, "version": state.version, "cells": cells },
+            "cellTextDocuments": texts,
+        });
+        let servers: Vec<Arc<Server>> =
+            state.servers.iter().filter(|s| only.is_none_or(|o| o.id == s.id)).cloned().collect();
+        for server in servers {
+            let Some(run) = server.running() else { continue };
+            if state.opened_on(server.id, run.generation) || !run.caps.read().syncs_notebooks(state.language.as_str()) {
+                continue;
+            }
+            if let Err(e) = run.peer.notify("notebookDocument/didOpen", params.clone()) {
+                tracing::debug!("{}: notebookDocument/didOpen: {e}", server.name);
+                continue;
+            }
+            state.opened.retain(|(id, _)| *id != server.id);
+            state.opened.push((server.id, run.generation));
+            for (_, doc) in &state.cells {
+                if let Some(a) = docs.get_mut(doc).and_then(|d| d.servers.iter_mut().find(|a| a.server.id == server.id))
+                {
+                    a.opened_gen = Some(run.generation);
+                }
+            }
+        }
+    }
+
+    fn close_notebook_on_servers(&self, state: &mut NotebookState) {
+        let cells: Vec<Value> = {
+            let docs = self.docs.lock();
+            state.cells.iter().filter_map(|(_, doc)| docs.get(doc)).map(|d| json!({ "uri": d.uri })).collect()
+        };
+        let params = json!({ "notebookDocument": { "uri": state.uri }, "cellTextDocuments": cells });
+        for (server, run) in state.open_targets() {
+            if let Err(e) = run.peer.notify("notebookDocument/didClose", params.clone()) {
+                tracing::debug!("{}: notebookDocument/didClose: {e}", server.name);
+            }
+        }
+        state.opened.clear();
+        let mut docs = self.docs.lock();
+        for (_, doc) in &state.cells {
+            if let Some(d) = docs.get_mut(doc) {
+                d.servers.iter_mut().for_each(|a| a.opened_gen = None);
+            }
+        }
+    }
+
+    /// Forgets cell documents and clears what their servers reported about them: a server need
+    /// not clear the diagnostics of a cell that left the notebook.
+    fn remove_cells(&self, servers: &[Arc<Server>], removed: &[DocId]) {
+        let gone: Vec<DocState> = {
+            let mut docs = self.docs.lock();
+            removed.iter().filter_map(|doc| docs.remove(doc)).collect()
+        };
+        for d in gone {
+            for server in servers {
+                let run = server.running();
+                self.forget_pulled(server, &d.path, run.as_ref());
+                let enc = run.as_ref().map(|r| r.caps.read().encoding).unwrap_or_default();
+                let shown = server.set_diagnostics(d.path.clone(), Vec::new());
+                self.emit_diagnostics(server, d.path.clone(), shown, enc);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn cell_uris_name_the_notebook_file_and_the_cell() {
+        let root = if cfg!(windows) { PathBuf::from(r"C:\work\my nb") } else { PathBuf::from("/work/my nb") };
+        let expected = if cfg!(windows) {
+            "vscode-notebook-cell:/C:/work/my%20nb/a.ipynb#f3-x_1%2E"
+        } else {
+            "vscode-notebook-cell:/work/my%20nb/a.ipynb#f3-x_1%2E"
+        };
+        let uri = cell_uri(&root.join("a.ipynb"), "f3-x_1.");
+        assert_eq!(uri, expected);
+        assert!(is_cell_uri(&uri) && parse_uri(&uri).is_some());
+        assert!(!is_cell_uri("file:///a.ipynb") && !is_cell_uri("vscode-notebook-cellx:/a"));
+    }
+}
@@ -687,6 +687,8 @@ async fn java_classpath_waits_for_a_config_restart_that_is_reaping_startup() {
            assert!(result.unwrap_err().message.contains("restart configuration error"));
        } else {
            assert!(result.is_ok(), "a configuration restart is transient: {result:?}");
+            // The module folder resolves to the instance serving its files, not a second one.
+            assert_eq!(h.mgr.servers().len(), 1);
            assert_eq!(state.launches.load(Ordering::SeqCst), 2);
            assert_eq!(h.mgr.servers()[0].id != old, shared);
        }
@@ -1916,16 +1918,17 @@ async fn replacing_progress_keeps_cancellation_reports_and_indexing_on_the_new_t
    for _ in 0..2 {
        peer.notify("$/progress", json!({ "token": 7, "value": { "kind": "end" } })).unwrap();
    }
-    // A later task is a barrier for both end notifications.
+    h.wait_status("mock", ServerStatus::Ready).await;
+    let refs = tokio::time::timeout(Duration::from_secs(5), search).await.unwrap().unwrap().unwrap();
+    assert_eq!(refs.incomplete, None);
+    // A later task is a barrier for both end notifications. It starts only after the search
+    // finished: one that is still running would see the barrier as indexing.
    peer.notify("$/progress", json!({ "token": "barrier", "value": { "kind": "begin", "title": "Barrier" } })).unwrap();
    peer.notify("$/progress", json!({ "token": "barrier", "value": { "kind": "end" } })).unwrap();
    eventually(|| {
        h.events.lock().iter().any(|e| matches!(e, IdeEvent::Task { title, done: true, .. } if title == "Barrier"))
    })
    .await;
-    h.wait_status("mock", ServerStatus::Ready).await;
-    let refs = tokio::time::timeout(Duration::from_secs(5), search).await.unwrap().unwrap().unwrap();
-    assert_eq!(refs.incomplete, None);
    assert!(!ide_core::task::cancel(new));
    let events = h.events.lock();
    for id in [old, new] {
@@ -6781,3 +6784,142 @@ async fn gomod_files_are_served_by_gopls() {
    assert_eq!(items[0].message, "missing requirement");
    mgr.shutdown_all().await;
 }
+
+// ------------------------------------------------------------------ notebooks
+
+fn notebook_caps() -> Value {
+    let mut caps = incremental_caps();
+    caps["notebookDocumentSync"] = json!({
+        "notebookSelector": [{
+            "notebook": { "scheme": "file", "notebookType": "jupyter-notebook" },
+            "cells": [{ "language": "python" }]
+        }]
+    });
+    caps
+}
+
+fn cells(cells: &[(&str, &str)]) -> Vec<ide_lsp::NotebookCell> {
+    cells.iter().map(|(id, source)| ide_lsp::NotebookCell { id: (*id).into(), source: (*source).into() }).collect()
+}
+
+fn cell_texts(open: &Value) -> Vec<(String, String)> {
+    open["cellTextDocuments"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|c| (c["uri"].as_str().unwrap().to_owned(), c["text"].as_str().unwrap().to_owned()))
+        .collect()
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn notebook_cells_are_documents_of_the_servers_that_sync_notebooks() {
+    let h = harness_for("python", "py", &[("nb", notebook_caps()), ("plain", incremental_caps())], &[]);
+    let path = h.root.join("analysis.ipynb");
+    let python = LanguageId::new("python");
+    h.mgr.sync_notebook(&path, &python, &cells(&[("a", "import os\n"), ("b", "os.pa")]));
+    h.wait_status("nb", ServerStatus::Ready).await;
+    h.wait_status("plain", ServerStatus::Ready).await;
+    let (nb, plain) = (h.launcher.state("nb"), h.launcher.state("plain"));
+    eventually(|| !nb.notified("notebookDocument/didOpen").is_empty()).await;
+
+    let docs = h.mgr.notebook_cells(&path);
+    assert_eq!(docs.iter().map(|d| d.cell.as_str()).collect::<Vec<_>>(), ["a", "b"]);
+    assert_eq!(docs[0].uri, ide_lsp::cell_uri(&path, "a"));
+    let open = &nb.notified("notebookDocument/didOpen")[0];
+    assert_eq!(open["notebookDocument"]["uri"], uri_of(&path));
+    assert_eq!(open["notebookDocument"]["notebookType"], "jupyter-notebook");
+    assert_eq!(open["notebookDocument"]["cells"][1], json!({ "kind": 2, "document": docs[1].uri }));
+    assert_eq!(
+        cell_texts(open),
+        [(docs[0].uri.clone(), "import os\n".to_owned()), (docs[1].uri.clone(), "os.pa".to_owned())]
+    );
+    assert!(plain.notified("notebookDocument/didOpen").is_empty() && plain.notified("textDocument/didOpen").is_empty());
+
+    // Requests address a cell like a file, and go only to the server it is open on.
+    nb.respond("textDocument/completion", json!({ "isIncomplete": false, "items": [{ "label": "path" }] }));
+    plain.respond("textDocument/completion", json!({ "isIncomplete": false, "items": [{ "label": "wrong" }] }));
+    let list = h.mgr.completion(docs[1].doc, Pos::new(0, 5), None, Default::default()).await.unwrap();
+    assert_eq!(list.items.iter().map(|i| i.label.as_str()).collect::<Vec<_>>(), ["path"]);
+    assert_eq!(nb.requested("textDocument/completion")[0]["textDocument"]["uri"], docs[1].uri);
+    assert!(plain.requested("textDocument/completion").is_empty());
+
+    // Its diagnostics are reported under the cell's URI.
+    nb.peer()
+        .notify(
+            "textDocument/publishDiagnostics",
+            json!({ "uri": docs[1].uri, "diagnostics": [{
+                "range": {"start":{"line":0,"character":3},"end":{"line":0,"character":5}}, "message": "no attribute pa"
+            }] }),
+        )
+        .unwrap();
+    let cell_path = PathBuf::from(&docs[1].uri);
+    eventually(|| {
+        h.events
+            .lock()
+            .iter()
+            .any(|e| matches!(e, IdeEvent::Diagnostics { path, items, .. } if path.0 == cell_path && items.len() == 1))
+    })
+    .await;
+
+    h.mgr.close_notebook(&path);
+    eventually(|| !nb.notified("notebookDocument/didClose").is_empty()).await;
+    assert_eq!(nb.notified("notebookDocument/didClose")[0]["cellTextDocuments"].as_array().unwrap().len(), 2);
+    assert!(h.mgr.notebook_cells(&path).is_empty());
+    assert!(h.mgr.completion(docs[1].doc, Pos::new(0, 5), None, Default::default()).await.is_err());
+    let last = h
+        .events
+        .lock()
+        .iter()
+        .filter_map(|e| match e {
+            IdeEvent::Diagnostics { path, items, source } if path.0 == cell_path && source == "nb" => Some(items.len()),
+            _ => None,
+        })
+        .next_back();
+    assert_eq!(last, Some(0), "a closed notebook's diagnostics are cleared");
+    h.mgr.shutdown_all().await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn notebook_edits_change_cell_text_and_reorders_reopen_it() {
+    let h = harness_for("python", "py", &[("nb", notebook_caps())], &[]);
+    let path = h.root.join("n.ipynb");
+    let python = LanguageId::new("python");
+    h.mgr.sync_notebook(&path, &python, &cells(&[("a", "x = 1\n"), ("b", "print(x)\n")]));
+    h.wait_status("nb", ServerStatus::Ready).await;
+    let nb = h.launcher.state("nb");
+    eventually(|| nb.notified("notebookDocument/didOpen").len() == 1).await;
+    let before = h.mgr.notebook_cells(&path);
+
+    // Typing in one cell sends only that cell's text.
+    h.mgr.sync_notebook(&path, &python, &cells(&[("a", "x = 1\n"), ("b", "print(x + 1)\n")]));
+    h.mgr.sync_notebook(&path, &python, &cells(&[("a", "x = 1\n"), ("b", "print(x + 1)\n")]));
+    eventually(|| !nb.notified("notebookDocument/didChange").is_empty()).await;
+    let changes = nb.notified("notebookDocument/didChange");
+    assert_eq!(changes.len(), 1, "an unchanged notebook sends nothing");
+    let content = &changes[0]["change"]["cells"]["textContent"];
+    assert_eq!(content.as_array().unwrap().len(), 1);
+    assert_eq!(content[0]["document"]["uri"], before[1].uri);
+    assert_eq!(content[0]["changes"], json!([{ "text": "print(x + 1)\n" }]));
+    assert_eq!(changes[0]["notebookDocument"]["version"], 2);
+
+    // Inserting and moving cells reopens the notebook; kept cells keep their documents.
+    h.mgr.sync_notebook(&path, &python, &cells(&[("c", "import math\n"), ("b", "print(x + 1)\n"), ("a", "x = 1\n")]));
+    eventually(|| nb.notified("notebookDocument/didOpen").len() == 2).await;
+    assert_eq!(nb.notified("notebookDocument/didClose").len(), 1);
+    let after = h.mgr.notebook_cells(&path);
+    assert_eq!(after.iter().map(|d| d.cell.as_str()).collect::<Vec<_>>(), ["c", "b", "a"]);
+    assert_eq!((after[1].doc, after[2].doc), (before[1].doc, before[0].doc));
+    let reopened = &nb.notified("notebookDocument/didOpen")[1];
+    assert_eq!(
+        cell_texts(reopened).into_iter().map(|(_, t)| t).collect::<Vec<_>>(),
+        ["import math\n", "print(x + 1)\n", "x = 1\n"]
+    );
+
+    // A restarted server gets the notebook again.
+    let launches = nb.launches.load(Ordering::SeqCst);
+    let server = h.mgr.servers()[0].id;
+    h.mgr.restart(server).await.unwrap();
+    eventually(|| nb.launches.load(Ordering::SeqCst) > launches && nb.notified("notebookDocument/didOpen").len() == 3)
+        .await;
+    h.mgr.shutdown_all().await;
+}
@@ -920,3 +920,99 @@ async fn angular_language_server_end_to_end() {
    }
    tokio::time::timeout(Duration::from_secs(10), mgr.shutdown_all()).await.unwrap();
 }
+
+/// A notebook's code cells on basedpyright and Ruff (LSP 3.17 notebook sync): names flow from
+/// earlier cells, IPython magics parse, and locations and diagnostics name the cells.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore = "requires ruff and basedpyright on PATH"]
+async fn basedpyright_and_ruff_serve_notebook_cells() {
+    use ide_lsp::{NotebookCell, cell_uri};
+    let dir = tempfile::tempdir().unwrap();
+    let root = dir.path().canonicalize().unwrap();
+    let root = dunce::simplified(&root).to_path_buf();
+    std::fs::write(root.join("pyproject.toml"), "[tool.ruff.lint]\nselect = [\"F\"]\n").unwrap();
+    let path = root.join("analysis.ipynb");
+    std::fs::write(&path, "{}").unwrap();
+    let mut python = LanguageRegistry::load(None).unwrap().get(&LanguageId::new("python")).unwrap();
+    python.language_servers.retain(|server| matches!(server.name.as_str(), "basedpyright" | "ruff"));
+    let (mgr, events) = manager(&root, LanguageRegistry::from_configs(vec![python]));
+    let cell = |id: &str, source: &str| NotebookCell { id: id.into(), source: source.into() };
+    let first = "import os\nimport sys\n\ndef area(r: float) -> float:\n    return 3.14 * r * r\n";
+    let second = "%matplotlib inline\nsize: int = area(2)\nos.pa";
+    mgr.sync_notebook(&path, &LanguageId::new("python"), &[cell("first", first), cell("second", second)]);
+    for name in ["basedpyright", "ruff"] {
+        retry(name, Duration::from_secs(60), || {
+            let status = last_status(&events, name);
+            if let Some((ServerStatus::Crashed | ServerStatus::Missing | ServerStatus::Misconfigured, msg)) = &status {
+                panic!("{name} failed: {msg:?}");
+            }
+            async move { matches!(status, Some((ServerStatus::Ready | ServerStatus::Busy, _))).then_some(()) }
+        })
+        .await;
+    }
+    let docs = mgr.notebook_cells(&path);
+    let b = docs[1].doc;
+    let (a_uri, b_uri) = (cell_uri(&path, "first"), cell_uri(&path, "second"));
+    assert_eq!((docs[0].uri.as_str(), docs[1].uri.as_str()), (a_uri.as_str(), b_uri.as_str()));
+
+    // `os`, imported in the first cell, completes in the second.
+    let labels = retry("completion in a cell", Duration::from_secs(30), || {
+        let mgr = mgr.clone();
+        async move {
+            let list = mgr.completion(b, Pos::new(2, 5), Some(".".into()), Default::default()).await.ok()?;
+            let labels: Vec<String> = list.items.into_iter().map(|i| i.label).collect();
+            labels.iter().any(|l| l == "path").then_some(labels)
+        }
+    })
+    .await;
+    assert!(labels.iter().any(|l| l == "path"), "{labels:?}");
+
+    let hover = mgr.hover(b, Pos::new(1, 13), Default::default()).await.unwrap().expect("hover on `area`");
+    assert!(
+        hover.contents.value.contains("area") && hover.contents.value.contains("float"),
+        "{}",
+        hover.contents.value
+    );
+
+    let defs = mgr.goto(b, Pos::new(1, 13), GotoKind::Definition).await.unwrap();
+    assert_eq!(defs.len(), 1, "{defs:?}");
+    assert_eq!((defs[0].path.to_path_buf(), defs[0].range.start.line), (std::path::PathBuf::from(&a_uri), 3));
+
+    let diagnostics = |source: &'static str, uri: String| {
+        let events = events.clone();
+        move || {
+            let items = events.lock().iter().rev().find_map(|e| match e {
+                IdeEvent::Diagnostics { path, source: s, items } if s == source && path.0 == Path::new(&uri) => {
+                    Some(items.clone())
+                }
+                _ => None,
+            });
+            async move { items.filter(|items| !items.is_empty()) }
+        }
+    };
+    // `float` is not assignable to `int`; the magic on line 0 is no error.
+    let typed =
+        retry("basedpyright cell diagnostics", Duration::from_secs(30), diagnostics("basedpyright", b_uri)).await;
+    assert!(typed.iter().any(|d| d.range.start.line == 1), "{typed:?}");
+    assert!(typed.iter().all(|d| d.range.start.line != 0), "{typed:?}");
+    // F401: `sys` is unused.
+    let linted = retry("ruff cell diagnostics", Duration::from_secs(30), diagnostics("ruff", a_uri)).await;
+    assert!(linted.iter().any(|d| d.code.as_deref() == Some("F401") && d.range.start.line == 1), "{linted:?}");
+
+    // An edit reaches both servers: the error is fixed.
+    let fixed = "%matplotlib inline\nsize: float = area(2)\nos.path";
+    mgr.sync_notebook(&path, &LanguageId::new("python"), &[cell("first", first), cell("second", fixed)]);
+    retry("diagnostics after the fix", Duration::from_secs(30), || {
+        let uri = cell_uri(&path, "second");
+        let clear = events.lock().iter().rev().find_map(|e| match e {
+            IdeEvent::Diagnostics { path, source, items } if source == "basedpyright" && path.0 == Path::new(&uri) => {
+                Some(items.is_empty())
+            }
+            _ => None,
+        });
+        async move { (clear == Some(true)).then_some(()) }
+    })
+    .await;
+    mgr.close_notebook(&path);
+    tokio::time::timeout(Duration::from_secs(10), mgr.shutdown_all()).await.unwrap();
+}
@@ -55,7 +55,7 @@ use views_bindings::exports::openidea::plugin::view_provider as wit_view;
 use views_bindings::openidea::plugin::view_types as wit_vt;

 /// The API version of the world above, `major.minor` of [`host_api`].
-pub const API_VERSION: &str = "1.5";
+pub const API_VERSION: &str = "1.6";
 const EPOCH_TICK: Duration = Duration::from_millis(10);
 pub(crate) const MAX_TEXT: usize = 1 << 20;
 pub(crate) const MAX_TITLE: usize = 256;
@@ -432,6 +432,27 @@ pub trait HostApi: Send + Sync {
    fn publish_diagnostics(&self, path: &Path, items: Vec<ide_core::Diagnostic>) -> Result<(), HostError>;
    /// Performs `request` (its host is granted), filling in secret placeholders.
    fn fetch(&self, request: HttpRequest, cancel: &CancellationToken) -> Result<HttpResponse, HostError>;
+
+    fn websocket_connect(
+        &self,
+        root: &Path,
+        request: crate::WebSocketRequest,
+        timeout: Duration,
+    ) -> Result<u64, HostError> {
+        let _ = (root, request, timeout);
+        Err(HostError::Invalid("background WebSockets are unavailable on this host".into()))
+    }
+    fn websocket_send(&self, connection: u64, bytes: Vec<u8>) -> Result<(), HostError> {
+        let _ = (connection, bytes);
+        Err(HostError::Invalid("background WebSockets are unavailable on this host".into()))
+    }
+    fn websocket_drain(&self, connection: u64, limit: usize) -> Result<crate::WebSocketBatch, HostError> {
+        let _ = (connection, limit);
+        Err(HostError::Invalid("background WebSockets are unavailable on this host".into()))
+    }
+    fn websocket_close(&self, connection: u64) {
+        let _ = connection;
+    }
    /// The `net:ask` answer for a request to HTTPS host `host` of this plugin
    /// (`subject`): `Some(true)` allowed and remembered (a host the user
    /// already allowed connects without asking again, also in the background),
@@ -606,6 +627,10 @@ impl PluginEngine {
        let mut linker = Linker::new(&engine);
        wasmtime_wasi::p2::add_to_linker_sync(&mut linker).map_err(|e| e.to_string())?;
        bindings::Plugin::add_to_linker::<_, HasSelf<_>>(&mut linker, |s| s).map_err(|e| e.to_string())?;
+        // Cranelift caches fuel in a JIT local across `new_epoch` yields and only
+        // reloads `VMStoreContext::fuel_consumed` across host import calls, so
+        // `tasks` host calls refill `StoreContextMut::set_fuel` directly on return.
+        bind_task_fuel_refill(&mut linker).map_err(|e| e.to_string())?;

        let weak = engine.weak();
        std::thread::Builder::new()
@@ -799,7 +824,7 @@ impl CallCancel {
        self.token.lock().clone()
    }

-    fn is_set(&self) -> bool {
+    pub(crate) fn is_set(&self) -> bool {
        self.flag.load(Ordering::Acquire)
    }

@@ -835,6 +860,10 @@ pub struct StoreData {
    /// Tasks begun by the running call and not ended yet.
    pub(crate) open_tasks: Vec<u64>,
    pub(crate) last_report: Option<Instant>,
+    /// One-shot unwind window after a blocking host I/O call times out.
+    pub(crate) host_io_grace_used: bool,
+    /// Fuel budget active for the running call (scales when a background task begins).
+    pub(crate) active_fuel: u64,
 }

 impl WasiView for StoreData {
@@ -987,6 +1016,14 @@ const MAX_BATCH: usize = 1000;

 fn event_to_wit(e: WorkbenchEvent) -> wit_events::Event {
    match e {
+        WorkbenchEvent::DocumentChanged(value) => wit_events::Event::Other(wit_types::Extension {
+            kind: "document-changed".into(),
+            data: serde_json::to_string(&value).expect("document change serializes"),
+        }),
+        WorkbenchEvent::WebsocketReady(value) => wit_events::Event::Other(wit_types::Extension {
+            kind: "websocket-ready".into(),
+            data: serde_json::to_string(&value).expect("WebSocket ready event serializes"),
+        }),
        WorkbenchEvent::RunExited(value) => wit_events::Event::Other(wit_types::Extension {
            kind: "run-exited".into(),
            data: serde_json::to_string(&value).unwrap(),
@@ -1067,6 +1104,8 @@ impl Runtime {
            action_prompt_client: None,
            open_tasks: Vec::new(),
            last_report: None,
+            host_io_grace_used: false,
+            active_fuel: setup.limits.fuel,
        };
        let mut store = Store::new(&engine.engine, data);
        store.limiter(|d| &mut d.store_limits);
@@ -1083,7 +1122,7 @@ impl Runtime {
        });
        arm(&mut store, &setup.limits, setup.limits.timeout).map_err(|e| e.to_string())?;
        let api_hint = |e: wasmtime::Error| {
-            let msg = describe_trap(&e, &setup.limits);
+            let msg = describe_trap(&e, &setup.limits, setup.limits.fuel);
            if msg.contains("openidea:plugin/") { link_error(&msg) } else { msg }
        };
        let pre = engine.linker.instantiate_pre(component).map_err(api_hint)?;
@@ -1868,10 +1907,11 @@ impl Runtime {
        if !stderr.trim().is_empty() {
            tracing::info!(plugin = %data.plugin_id, "stderr: {}", stderr.trim_end());
        }
+        let active_fuel = data.active_fuel;
        match result {
            Ok(r) => Ok(r),
            Err(e) => {
-                let mut msg = describe_trap(&e, &self.limits);
+                let mut msg = describe_trap(&e, &self.limits, active_fuel);
                // Typically the panic message: "panicked at src/lib.rs:1:2:\nboom".
                let detail: Vec<&str> =
                    stderr.lines().filter(|l| !l.trim().is_empty() && !l.starts_with("note:")).collect();
@@ -1923,15 +1963,55 @@ fn arm(store: &mut Store<StoreData>, limits: &Limits, timeout: Duration) -> wasm
    data.deadline = now + timeout;
    data.timeout = timeout;
    data.last_report = None;
+    data.host_io_grace_used = false;
+    data.active_fuel = limits.fuel;
    data.cancel.reset();
    // The callback above runs every tick and decides.
    store.set_epoch_deadline(1);
    Ok(())
 }

-fn describe_trap(e: &wasmtime::Error, limits: &Limits) -> String {
+fn bind_task_fuel_refill(linker: &mut Linker<StoreData>) -> wasmtime::Result<()> {
+    use crate::wit_host::TaskRes;
+    use crate::wit_host::bindings::openidea::plugin::tasks::{Host, HostTask};
+    use wasmtime::StoreContextMut;
+    use wasmtime::component::Resource;
+
+    linker.allow_shadowing(true);
+    let tasks_iface = format!("openidea:plugin/tasks@{}", ide_extension::ENGINE_VERSION);
+    let mut inst = linker.instance(&tasks_iface)?;
+    inst.func_wrap("begin", |mut caller: StoreContextMut<'_, StoreData>, (title, cancellable): (String, bool)| {
+        let task = Host::begin(caller.data_mut(), title, cancellable)?;
+        let fuel = caller.data().active_fuel;
+        caller.set_fuel(fuel)?;
+        Ok((task,))
+    })?;
+    inst.func_wrap(
+        "[method]task.report",
+        |mut caller: StoreContextMut<'_, StoreData>,
+         (this, done, total, message): (Resource<TaskRes>, u32, Option<u32>, Option<String>)| {
+            HostTask::report(caller.data_mut(), this, done, total, message)?;
+            let fuel = caller.data().active_fuel;
+            caller.set_fuel(fuel)?;
+            Ok(())
+        },
+    )?;
+    inst.func_wrap(
+        "[method]task.is-cancelled",
+        |mut caller: StoreContextMut<'_, StoreData>, (this,): (Resource<TaskRes>,)| {
+            let cancelled = HostTask::is_cancelled(caller.data_mut(), this)?;
+            let fuel = caller.data().active_fuel;
+            caller.set_fuel(fuel)?;
+            Ok((cancelled,))
+        },
+    )?;
+    linker.allow_shadowing(false);
+    Ok(())
+}
+
+fn describe_trap(e: &wasmtime::Error, limits: &Limits, active_fuel: u64) -> String {
    match e.downcast_ref::<wasmtime::Trap>() {
-        Some(wasmtime::Trap::OutOfFuel) => format!("exceeded its CPU budget ({} fuel per call)", limits.fuel),
+        Some(wasmtime::Trap::OutOfFuel) => format!("exceeded its CPU budget ({active_fuel} fuel per call)"),
        Some(wasmtime::Trap::Interrupt) => CANCELLED.to_owned(),
        Some(wasmtime::Trap::UnreachableCodeReached) => "crashed (panic / unreachable)".to_owned(),
        Some(trap) => format!("trapped: {trap}"),
@@ -104,6 +104,8 @@ pub enum Prompt {
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, Type)]
 #[serde(rename_all = "kebab-case")]
 pub enum EventKind {
+    DocumentChanged,
+    WebsocketReady,
    FileClosed,
    SettingsChanged,
    InputModeChanged,
@@ -121,7 +123,9 @@ pub enum EventKind {
 }

 impl EventKind {
-    pub const ALL: [Self; 14] = [
+    pub const ALL: [Self; 16] = [
+        Self::DocumentChanged,
+        Self::WebsocketReady,
        Self::FileClosed,
        Self::SettingsChanged,
        Self::InputModeChanged,
@@ -140,6 +144,8 @@ impl EventKind {

    pub fn name(self) -> &'static str {
        match self {
+            Self::DocumentChanged => "document-changed",
+            Self::WebsocketReady => "websocket-ready",
            Self::FileClosed => "file-closed",
            Self::SettingsChanged => "settings-changed",
            Self::InputModeChanged => "input-mode-changed",
@@ -227,10 +233,53 @@ pub struct InputMode {
    pub sub: Option<String>,
 }

+/// One UTF-16 replacement in a locally edited editor document.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Type)]
+pub struct DocumentTextChange {
+    pub from: u32,
+    pub to: u32,
+    pub insert: String,
+}
+
+/// Local changes accepted by the editor's shared document model.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Type)]
+#[serde(rename_all = "camelCase")]
+pub struct DocumentChanged {
+    pub path: String,
+    pub base_version: u64,
+    pub version: u64,
+    pub changes: Vec<DocumentTextChange>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Type)]
+#[serde(rename_all = "camelCase")]
+pub struct WebsocketReady {
+    pub plugin_id: String,
+    pub connection: u64,
+}
+
+/// A host-owned background WebSocket request.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct WebSocketRequest {
+    pub url: String,
+    pub headers: Vec<(String, String)>,
+    pub subprotocol: String,
+}
+
+/// Frames received from a host-owned WebSocket and its transport state.
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, Type)]
+#[serde(rename_all = "camelCase")]
+pub struct WebSocketBatch {
+    pub messages: Vec<Vec<u8>>,
+    pub closed: bool,
+}
+
 /// A workbench event delivered to plugins.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Type)]
 #[serde(rename_all = "camelCase", tag = "kind", content = "value")]
 pub enum WorkbenchEvent {
+    DocumentChanged(DocumentChanged),
+    WebsocketReady(WebsocketReady),
    RunExited(crate::RunExited),
    TerminalExited(crate::TerminalExited),
    ContentTabAction(crate::ContentTabAction),
@@ -251,6 +300,8 @@ pub enum WorkbenchEvent {
 impl WorkbenchEvent {
    pub fn kind(&self) -> EventKind {
        match self {
+            Self::DocumentChanged(_) => EventKind::DocumentChanged,
+            Self::WebsocketReady(_) => EventKind::WebsocketReady,
            Self::RunExited(_) => EventKind::RunExited,
            Self::TerminalExited(_) => EventKind::TerminalExited,
            Self::ContentTabAction(_) => EventKind::ContentTabAction,
@@ -476,6 +527,18 @@ pub fn net_capability(url: &str) -> Result<ide_extension::Capability, String> {
    }
 }

+/// The grant a WebSocket URL requires: the same as its `https`/`http` counterpart.
+pub fn websocket_capability(url: &str) -> Result<ide_extension::Capability, String> {
+    let mut parsed = url::Url::parse(url).map_err(|e| format!("invalid URL: {e}"))?;
+    let scheme = match parsed.scheme() {
+        "wss" => "https",
+        "ws" => "http",
+        other => return Err(format!("unsupported WebSocket scheme {other:?}")),
+    };
+    parsed.set_scheme(scheme).map_err(|()| "invalid WebSocket URL".to_owned())?;
+    net_capability(parsed.as_str())
+}
+
 #[cfg(test)]
 mod tests {
    #[test]
@@ -510,6 +573,15 @@ mod tests {
            assert_eq!(net_host(url).unwrap(), host);
        }
        assert_ne!(net_host("https://аpi.example/").unwrap(), net_host("https://api.example/").unwrap());
+        assert_eq!(
+            websocket_capability("wss://Collab.example.com/socket?room=1").unwrap(),
+            ide_extension::Capability::Net("collab.example.com".into())
+        );
+        assert_eq!(
+            websocket_capability("ws://localhost:9000/").unwrap(),
+            ide_extension::Capability::Net("localhost".into())
+        );
+        assert!(websocket_capability("https://example.com/").is_err());
        assert!(net_host("file:///etc/passwd").is_err());
        assert!(net_host("example.com").is_err());
        assert!(net_host("https://*.example.com/").is_err());
@@ -26,6 +26,7 @@ pub(crate) mod bindings {
            "openidea:plugin/tasks.task": super::TaskRes,
            "openidea:plugin/process.child": super::ChildRes,
            "openidea:plugin/ipc.connection": super::ConnectionRes,
+            "openidea:plugin/net.websocket": super::WebSocketRes,
            "openidea:plugin/auth.loopback": super::LoopbackRes,
        },
    });
@@ -162,6 +163,19 @@ pub struct ChildRes {
    host: std::sync::Arc<dyn crate::HostApi>,
 }

+/// A project-bound background WebSocket owned by this plugin instance.
+/// Releasing the component resource closes the underlying transport.
+pub struct WebSocketRes {
+    id: u64,
+    host: std::sync::Arc<dyn crate::HostApi>,
+}
+
+impl Drop for WebSocketRes {
+    fn drop(&mut self) {
+        self.host.websocket_close(self.id);
+    }
+}
+
 impl Drop for ChildRes {
    fn drop(&mut self) {
        self.host.release_process(self.id);
@@ -183,6 +197,8 @@ pub struct TaskRes {
 const MAX_EDITS: usize = 10_000;
 const MAX_DIAGNOSTICS: usize = 5_000;
 const MAX_REQUEST_BODY: usize = 8 << 20;
+/// Upper bound for a WebSocket handshake; the call's remaining time can make it shorter.
+const WEBSOCKET_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
 const MAX_FIND: u32 = 10_000;
 const MAX_STATUS_TEXT: usize = 128;
 const MAX_PICK_ITEMS: usize = 5000;
@@ -822,6 +838,52 @@ impl workspace::HostProject for StoreData {
 }

 impl editor::Host for StoreData {
+    fn document_at(&mut self, project: Resource<ProjectRes>, path: String) -> Fallible<editor::Document> {
+        if let Err(error) = self.require(Capability::ProjectRead).and_then(|_| self.require(Capability::EditorRead)) {
+            return Ok(Err(error));
+        }
+        let root = self.table.get(&project)?.root.clone();
+        let path = match self.resolve_in_project(&root, &path) {
+            Ok(path) => path,
+            Err(error) => return Ok(Err(wit::Error::Invalid(error))),
+        };
+        Ok(match self.host.document(&path) {
+            Some((version, text)) => Ok(editor::Document { version, text }),
+            None => Err(wit::Error::NotFound(format!("{} is not open", path.display()))),
+        })
+    }
+
+    fn apply_edits_at(
+        &mut self,
+        project: Resource<ProjectRes>,
+        path: String,
+        version: u64,
+        edits: Vec<wit::TextEdit>,
+    ) -> Fallible<u64> {
+        if let Err(error) = self.require(Capability::ProjectRead).and_then(|_| self.require(Capability::EditorWrite)) {
+            return Ok(Err(error));
+        }
+        if edits.len() > MAX_EDITS {
+            return Ok(Err(wit::Error::Invalid(format!("more than {MAX_EDITS} edits"))));
+        }
+        if edits.iter().map(|edit| edit.text.len()).sum::<usize>() > crate::runtime::MAX_TEXT * 8 {
+            return Ok(Err(wit::Error::Invalid("edits too large".into())));
+        }
+        let root = self.table.get(&project)?.root.clone();
+        let path = match self.resolve_in_project(&root, &path) {
+            Ok(path) => path,
+            Err(error) => return Ok(Err(wit::Error::Invalid(error))),
+        };
+        let edits = edits
+            .into_iter()
+            .map(|edit| ide_core::TextEdit {
+                range: ide_core::Range::new(pos(edit.range.start), pos(edit.range.end)),
+                new_text: edit.text,
+            })
+            .collect();
+        Ok(self.host.apply_edits(None, &path, version, edits).map_err(to_wit))
+    }
+
    fn open_editors(&mut self) -> Fallible<Vec<editor::OpenEditor>> {
        if let Err(e) = self.require(Capability::EditorRead).and_then(|_| self.require_window()) {
            return Ok(Err(e));
@@ -857,6 +919,7 @@ impl editor::Host for StoreData {
 fn core_range(r: wit::Range) -> ide_core::Range {
    ide_core::Range::new(pos(r.start), pos(r.end))
 }
+
 impl StoreData {
    /// A `project` resource's path: inside the resource's root, or — for a
    /// project window group — inside any of its member roots (as the editor and
@@ -1196,6 +1259,11 @@ impl events::Host for StoreData {
        {
            return Ok(Err(to_wit(e)));
        }
+        if kind == EventKind::DocumentChanged {
+            return Ok(Err(wit::Error::Invalid(
+                "document-changed is delivered only for paths named with subscribe-paths".into(),
+            )));
+        }
        if let Some(cap) = kind.capability()
            && let Err(e) = self.require(cap)
        {
@@ -1209,6 +1277,14 @@ impl events::Host for StoreData {
        if let Err(e) = self.require(Capability::Events) {
            return Ok(Err(e));
        }
+        if kind == EventKind::DocumentChanged.name() {
+            if let Err(e) = self.host.require_api(&kind, 6) {
+                return Ok(Err(to_wit(e)));
+            }
+            if let Err(e) = self.require(Capability::ProjectRead).and_then(|_| self.require(Capability::EditorRead)) {
+                return Ok(Err(e));
+            }
+        }
        Ok(self.host.subscribe_paths(&kind, globs).map_err(to_wit))
    }

@@ -1267,6 +1343,11 @@ impl tasks::HostTask for StoreData {
    }
 }

+/// Headroom reserved before `StoreData::deadline` so a timed-out host subprocess
+/// or HTTP request can unwind as a recoverable `wit::Error` in guest code instead
+/// of immediately tripping the Wasm epoch trap on re-entry.
+const HOST_IO_UNWIND_GRACE: Duration = Duration::from_millis(250);
+
 impl StoreData {
    /// Lets the running call reach `from` plus the background budget (the
    /// longer of it and what was left), so a task that keeps saying it is alive
@@ -1277,6 +1358,8 @@ impl StoreData {
            self.deadline = extended;
            self.timeout = self.limits.background;
        }
+        let scale = (self.limits.background.as_millis() / self.limits.timeout.as_millis().max(1)).max(1) as u64;
+        self.active_fuel = self.limits.fuel.saturating_mul(scale);
    }

    /// What is left of the running call's wall-clock budget (host operations
@@ -1284,6 +1367,28 @@ impl StoreData {
    fn remaining(&self) -> Duration {
        self.deadline.saturating_duration_since(Instant::now())
    }
+
+    /// Budget handed to a blocking host I/O call (`process.run`, `net.fetch`),
+    /// leaving room before `self.deadline` when possible so a timed-out I/O call
+    /// returns `Err` to the guest instead of trapping the Wasm instance.
+    fn io_timeout(&self) -> Duration {
+        let rem = self.remaining();
+        if !self.host_io_grace_used && rem > HOST_IO_UNWIND_GRACE * 2 { rem - HOST_IO_UNWIND_GRACE } else { rem }
+    }
+
+    /// Grants at most one short unwind window per call when a blocking host I/O
+    /// call failed at or near `self.deadline` without caller cancellation. An
+    /// error well before the deadline leaves the window for a later call.
+    fn grant_io_error_grace(&mut self) {
+        if self.host_io_grace_used || self.cancel.is_set() {
+            return;
+        }
+        let min_deadline = Instant::now() + HOST_IO_UNWIND_GRACE;
+        if min_deadline > self.deadline {
+            self.deadline = min_deadline;
+            self.host_io_grace_used = true;
+        }
+    }
 }

 impl diagnostics::Host for StoreData {
@@ -1371,13 +1476,15 @@ impl net::Host for StoreData {
            url: req.url,
            headers: req.headers.into_iter().map(|h| (h.name, h.value)).collect(),
            body: req.body,
-            timeout: wanted.min(self.remaining()),
+            timeout: wanted.min(self.io_timeout()),
            grants,
        };
        let token = self.cancel.token();
-        Ok(self
-            .host
-            .fetch(request, &token)
+        let outcome = self.host.fetch(request, &token);
+        if outcome.is_err() {
+            self.grant_io_error_grace();
+        }
+        Ok(outcome
            .map(|r| net::Response {
                status: r.status,
                headers: r.headers.into_iter().map(|(name, value)| net::Header { name, value }).collect(),
@@ -1385,6 +1492,70 @@ impl net::Host for StoreData {
            })
            .map_err(to_wit))
    }
+
+    fn websocket_connect(
+        &mut self,
+        project: Resource<ProjectRes>,
+        req: net::WebsocketRequest,
+    ) -> Fallible<Resource<WebSocketRes>> {
+        let capability = match crate::types::websocket_capability(&req.url) {
+            Ok(capability) => capability,
+            Err(error) => return Ok(Err(wit::Error::Invalid(error))),
+        };
+        if let Err(error) = self.require(capability) {
+            return Ok(Err(error));
+        }
+        if req.headers.iter().any(|header| header.value.contains("{{secret:"))
+            && let Err(error) = self.require(Capability::Secrets)
+        {
+            return Ok(Err(error));
+        }
+        let request = crate::WebSocketRequest {
+            url: req.url,
+            headers: req.headers.into_iter().map(|header| (header.name, header.value)).collect(),
+            subprotocol: req.subprotocol,
+        };
+        let root = self.table.get(&project)?.root.clone();
+        let timeout = WEBSOCKET_HANDSHAKE_TIMEOUT.min(self.io_timeout());
+        match self.host.websocket_connect(&root, request, timeout) {
+            Ok(id) => self.push(WebSocketRes { id, host: self.host.clone() }).map(Ok),
+            Err(error) => {
+                self.grant_io_error_grace();
+                Ok(Err(to_wit(error)))
+            }
+        }
+    }
+}
+
+impl net::HostWebsocket for StoreData {
+    fn id(&mut self, this: Resource<WebSocketRes>) -> wasmtime::Result<u64> {
+        Ok(self.table.get(&this)?.id)
+    }
+
+    fn send(&mut self, this: Resource<WebSocketRes>, bytes: Vec<u8>) -> Fallible<()> {
+        let id = self.table.get(&this)?.id;
+        Ok(self.host.websocket_send(id, bytes).map_err(to_wit))
+    }
+
+    fn drain(&mut self, this: Resource<WebSocketRes>, limit: u32) -> Fallible<net::WebsocketBatch> {
+        let id = self.table.get(&this)?.id;
+        Ok(self
+            .host
+            .websocket_drain(id, limit as usize)
+            .map(|batch| net::WebsocketBatch { messages: batch.messages, closed: batch.closed })
+            .map_err(to_wit))
+    }
+
+    fn close(&mut self, this: Resource<WebSocketRes>) -> Fallible<()> {
+        let id = self.table.get(&this)?.id;
+        self.host.websocket_close(id);
+        Ok(Ok(()))
+    }
+
+    fn drop(&mut self, this: Resource<WebSocketRes>) -> wasmtime::Result<()> {
+        self.table.delete(this)?;
+        Ok(())
+    }
 }

 impl secrets::Host for StoreData {
@@ -1466,8 +1637,12 @@ impl process::Host for StoreData {
        };
        let cwd = self.cwd_of(cwd)?;
        let token = self.cancel.token();
-        let timeout = self.remaining();
-        Ok(self.host.run_process(cmd, cwd.as_deref(), stdin, &token, timeout).map(to_output).map_err(to_wit))
+        let timeout = self.io_timeout();
+        let outcome = self.host.run_process(cmd, cwd.as_deref(), stdin, &token, timeout);
+        if outcome.is_err() {
+            self.grant_io_error_grace();
+        }
+        Ok(outcome.map(to_output).map_err(to_wit))
    }

    fn run_command(
@@ -1482,8 +1657,12 @@ impl process::Host for StoreData {
        };
        let cwd = self.cwd_of(cwd)?;
        let token = self.cancel.token();
-        let timeout = self.remaining();
-        Ok(self.host.run_process(cmd, cwd.as_deref(), stdin, &token, timeout).map(to_output).map_err(to_wit))
+        let timeout = self.io_timeout();
+        let outcome = self.host.run_process(cmd, cwd.as_deref(), stdin, &token, timeout);
+        if outcome.is_err() {
+            self.grant_io_error_grace();
+        }
+        Ok(outcome.map(to_output).map_err(to_wit))
    }

    fn start(&mut self, cmd: process::Command, cwd: Option<Resource<ProjectRes>>) -> Fallible<Resource<ChildRes>> {
@@ -91,15 +91,15 @@ fn the_sdk_carries_the_hosts_wit() {

 #[test]
 fn the_version_check_runs_compatible_components_and_explains_the_rest() {
-    let host = v("1.5.0");
-    for ok in ["1.0.0", "1.3.0", "1.4.7", "1.5.0", "1.0.0-rc.1"] {
+    let host = v("1.6.0");
+    for ok in ["1.0.0", "1.3.0", "1.4.7", "1.5.0", "1.6.0", "1.0.0-rc.1"] {
        assert_eq!(check_api(&v(ok), &host), Ok(()), "{ok}");
    }
-    let newer = check_api(&v("1.6.0"), &host).unwrap_err();
-    assert!(newer.contains("1.6") && newer.contains("update OpenIDEA"), "{newer}");
+    let newer = check_api(&v("1.7.0"), &host).unwrap_err();
+    assert!(newer.contains("1.7") && newer.contains("update OpenIDEA"), "{newer}");
    assert!(check_api(&v("2.0.0"), &host).unwrap_err().contains("update OpenIDEA"));
    let older = check_api(&v("0.2.0"), &host).unwrap_err();
-    assert!(older.contains("rebuild it with the 1.5 SDK"), "{older}");
+    assert!(older.contains("rebuild it with the 1.6 SDK"), "{older}");
    // 0.x: every minor is its own API.
    assert_eq!(check_api(&v("0.2.0"), &v("0.2.5")), Ok(()));
    assert!(check_api(&v("0.1.0"), &v("0.2.0")).is_err());
@@ -1 +1 @@
-eb7d608bf1e7379620ef7d9ce5289dd9b927f137af2aafdd0861a84d60d677a2
+005d9da29eb7d713a97ff872051e408f653030434d837835843298553c9e12e0
@@ -1 +1 @@
-195cb4025508e7af6e6fb63d81660c4a977327fa3d101a2a77e5f448f43518f8
+6359ef4271f38374889412bd926d9aa735f59a8ce047ffcb06c47976444196cf
@@ -1 +1 @@
-49491ad2d57592cca2451b0abde5f88a3aedd3f5d36ceaf8259c3254f59c5023
+6f560f3d0d7737150d2dfa09c3b801bd3b4c3a52cef9893ed8fc045809fe33e1
@@ -1 +1 @@
-b24383f6e05ec531ef8db368b79af710433bc3782438f5a9e3578ebb730a1ce2
+4024af2d55d7fe4e8c456efd7e0c497ade4b2c5959e9108ff218f4a669e920f4
@@ -1 +1 @@
-ef01f199e668063b2c444e4ccf169d5b5407e567dbe8ff583cfdfc111b2438cf
+4ca977f53a18e9e8f869d34e027a15abece9e97b9331e23d4f71af721b104611
@@ -1 +1 @@
-a4dbe2a261e00809b64039ce9c89db681b186788db8a92b7456eca84b33c5945
+3d12a00f81e0db1b0902c2ddebfd6454a056f15b97add80b22a0333722568dc8
@@ -1 +1 @@
-3c9db6ffeaa9b782970c7a7a132878f862b0b5b18e9d11ad56649ee28317e3ca
+95d89e411216f50d0b34bee110c6bfc9b8fbcb2ffaae2e912c3609f7bf678b2c
@@ -1 +1 @@
-ac1dceb016b7963f44e876784ea62f776e90dd63fc6f660a5dad3dfbd7433b1c
+dd4b6c5153892a4f3facb3f23cb1d92da39a3977985b8abf4eaa9558decfa4a9
@@ -1 +1 @@
-11dc1a74c0650b493ed6ec88b5d15e08421ab08c5f2320099dabd5df1bdc5096
+325bc15cd28933d9687bc3c316008f921f48a4a69d7528c6b367fa0ff22489b8
@@ -1 +1 @@
-94c391142af832a67c694968d307217aabd7c139ae29e3299930b56ce644e45b
+abe706392e04ef5a14f9feab88e0c0d6a9e96bee7de771b61deeb65e7d9e0a7c
@@ -1 +1 @@
-e2c1893a160a0f27af7eb9e689d703577335cce32f746713c2435ae70fb9753d
+fb3677946021b30d226e2e3d422c352645c2065c93057c0a645246520860fc90
@@ -1 +1 @@
-83a8f59e0bde75c5c8fc725aa0fe34954279b31b7237acc868470ccf545ca229
+a01852ead9004ac1558028da19a1ae96ccd035775484089fd39f14263082daf1
@@ -1 +1 @@
-f9741cc8f7250a71311264613d636c5c844f1ae294362d4d331015ca73acbc57
+bf00050b7bb9911680a94fd36aaeb83fc96d94030e8ceef51d1723c3950253f9
@@ -1 +1 @@
-a6f65bb0d13527d726cef88fa50a5db19d48517b1ae516db36e9756dc87e75bc
+d00a07d9d51e7ca117cec59070d2812d4d1efb2730b60a6f171c8fd12903d055
@@ -1 +1 @@
-c9dbae18196b98eb4fad86fd5f36033c2b82a3087b4f51c4cba5a1362924d795
+0aa0ee56e8bacd4d218bafecdec7f4b6249381fad6e31f8b2ee89fd88b16d710
@@ -1 +1 @@
-1086cdc27def1958e8ec818e197af67edc2fd252b198e0ac358b149e0f5abfe8
+e06955818969808e6695fba9178727239e343bc4542efbf56262dd407175fd8a
@@ -1 +1 @@
-48fe38b954f91bf8adc01a4b7b41802040f663c16fde586bd5daa51be98ba51d
+7af67673471d49f97aa2cc131b5deddfc78ee7fc203901eba12f4970a5b13f75
@@ -1 +1 @@
-9ff2eaaa054ecdab56bf01220c8529da2b8d074b923cfb8b332d347767ba04f4
+e81d2454a61977ab69dc8b5c3c617627c559c4e340337a3d43309daeb9b1c12d
@@ -5,9 +5,9 @@ wit_bindgen::generate!({
    inline: "
        package fixture:misbehave;
        world misbehave {
-            include openidea:plugin/plugin-syntax@1.5.0;
-            export openidea:plugin/providers@1.5.0;
-            export openidea:plugin/file-actions@1.5.0;
+            include openidea:plugin/plugin-syntax@1.6.0;
+            export openidea:plugin/providers@1.6.0;
+            export openidea:plugin/file-actions@1.6.0;
        }
    ",
    path: "../../../../../wit",
@@ -1 +1 @@
-d72afe7db6283cdaecd7c673d1d26ff5af39f105733109ec73901ca37b00ea28
+123ae68df6baa3b5983349ba05fb441ec67fddcb32cab566b0d02e443bb4800e
@@ -1 +1 @@
-bccf0174932e661c40f05d779bf094e7aa4c91c359399688e68dd7a0a1d389b6
+7b26a4ec36b496f9a4f5e029ef557707961a281e31508863c989a5a5abd5e225
@@ -1 +1 @@
-77909d57469273d0abd56d382e408ebb9a6c5dd4463b0c2cbca8f27a19bf116c
+67f278ade12d11b1445ac3d448853ec46282f2f3e2b2299c6c8fe0dfb4fc86cd
@@ -1 +1 @@
-79beb3b8fdb9f412132dda4a76b88f55cfa859e3462c8cb020a0e431ad23fd72
+8cf9210844a1968f485e6e747f6dff4a1fdf0a628e2945f731d8289dd01f8385
@@ -1 +1 @@
-f2430f9afa0c8af35d96832a0cf7b1eade8e5977fb46eed93123b65f950f4f15
+263e6ab4b54593f09ae2cf71e9a191acb95ed744cc16bbaf9a9263e41614e99a
@@ -1 +1 @@
-65753d7c4938560bf2cbc2d1eb2d5588f455ae40f265369aa1126c4f1fd0e5ce
+682a712b6c75ad0c0da24476f87c77a76b248c201d4d31923a79fda053b76b90