fix(plugin-ai): validate context references before publishing turns

electron/plugins/ai/modelInvocationBroker.cjs

@@ -0,0 +1,263 @@
+"use strict";
+
+const { randomUUID } = require("node:crypto");
+const { PluginRpcError, RPC_ERRORS } = require("../rpcRouter.cjs");
+const { assertPluginJsonValue } = require("../jsonBoundary.cjs");
+const fail = (code, message) => new PluginRpcError(RPC_ERRORS[code], message);
+const LIMITS = Object.freeze({ chats: 32, perRuntime: 2, turnMs: 60_000, chatMs: 300_000,
+  inputBytes: 16_384, outputBytes: 32_768, outputTokens: 2_048, requestsPerMinute: 8,
+  reservedTokensPerMinute: 65_536, contexts: 4, chunkBytes: 4_096, events: 8_192, rateOwners: 128 });
+
+function boundedLimits(overrides) {
+  return Object.freeze(Object.fromEntries(Object.entries(LIMITS).map(([key, max]) => {
+    const value = overrides[key] ?? max;
+    if (!Number.isSafeInteger(value) || value < 1 || value > max) throw new TypeError(`Invalid ${key} limit`);
+    return [key, value];
+  })));
+}
+
+function abortable(promise, signal) {
+  signal.throwIfAborted();
+  return new Promise((resolve, reject) => {
+    const abort = () => reject(signal.reason);
+    signal.addEventListener("abort", abort, { once: true });
+    Promise.resolve(promise).then(resolve, reject).finally(() => signal.removeEventListener("abort", abort));
+  });
+}
+
+// In-memory, source-isolated handles. No AgentRuntime, tools, chat persistence,
+// provider configuration, URLs, credentials or arbitrary system prompts cross it.
+class ModelInvocationBroker {
+  constructor({ resolveDataOwner, selectModel, contexts, clock = Date.now, limits = {} }) {
+    if (typeof resolveDataOwner !== "function" || typeof selectModel !== "function") throw new TypeError("Host adapters required");
+    this.resolveDataOwner = resolveDataOwner;
+    this.selectModel = selectModel;
+    this.contexts = contexts;
+    this.clock = clock;
+    this.limits = boundedLimits(limits);
+    this.chats = new Map();
+    this.turns = new Map();
+    this.rates = new Map();
+    this.closed = false;
+    this.contextSubscription = contexts?.onDidInvalidate((handle) => {
+      for (const turn of [...this.turns.values()]) {
+        if (turn.references.some((reference) => reference.handle === handle)) {
+          this.#end(turn, fail("cancelled", "Selected context was revoked"));
+        }
+      }
+    });
+  }
+
+  async #binding(context) {
+    await context.assertActive();
+    const owner = await this.resolveDataOwner(context);
+    await context.assertActive();
+    if (!owner || (typeof owner !== "object" && typeof owner !== "string")) throw fail("permissionDenied", "Host data owner is unavailable");
+    return { owner, runtimeId: context.runtimeId, pluginVersion: context.pluginVersion };
+  }
+
+  async #ownedChat(chat, context) {
+    const binding = await this.#binding(context);
+    if (this.closed || this.chats.get(chat.handle) !== chat
+      || chat.owner !== binding.owner || chat.runtimeId !== binding.runtimeId || chat.pluginVersion !== binding.pluginVersion) {
+      throw fail("permissionDenied", "Plugin chat is unavailable");
+    }
+  }
+
+  async #check(chat, context) {
+    await this.#ownedChat(chat, context);
+    if (!await chat.model.isCurrent()) throw fail("permissionDenied", "Selected model changed");
+    const currentOwner = await this.resolveDataOwner(context);
+    await context.assertActive();
+    if (currentOwner !== chat.owner || this.closed || this.chats.get(chat.handle) !== chat || chat.expiresAt <= this.clock()) {
+      throw fail("permissionDenied", "Plugin chat is unavailable");
+    }
+  }
+
+  #prune() {
+    for (const chat of [...this.chats.values()]) if (chat.expiresAt <= this.clock()) this.#close(chat);
+    for (const [owner, rate] of this.rates) if (rate.startedAt + 60_000 <= this.clock()) this.rates.delete(owner);
+  }
+
+  async open(context) {
+    const binding = await this.#binding(context);
+    // Host chooses from configured models. Plugin parameters cannot override it.
+    const model = await this.selectModel(context);
+    await context.assertActive();
+    if (!model || typeof model.invoke !== "function" || typeof model.isCurrent !== "function"
+      || typeof model.label !== "string" || model.label.length > 160 || !await model.isCurrent()) {
+      throw fail("unavailable", "Choose a configured model in the host first");
+    }
+    await context.assertActive();
+    this.#prune();
+    if (this.closed || this.chats.size >= this.limits.chats
+      || [...this.chats.values()].filter((chat) => chat.runtimeId === context.runtimeId).length >= this.limits.perRuntime) {
+      throw fail("resourceExhausted", "Plugin chat limit reached");
+    }
+    const chat = { ...binding, handle: randomUUID(), model, expiresAt: this.clock() + this.limits.chatMs, turn: null };
+    this.chats.set(chat.handle, chat);
+    return { chatHandle: chat.handle, model: model.label, expiresAt: chat.expiresAt };
+  }
+
+  async start({ chatHandle, prompt, contexts = [] }, context) {
+    assertPluginJsonValue({ chatHandle, prompt, contexts }, { maxBytes: this.limits.inputBytes + 2048 });
+    if (typeof prompt !== "string" || !prompt.trim() || prompt.includes("\0") || !Array.isArray(contexts)
+      || contexts.length > this.limits.contexts || Buffer.byteLength(prompt) > this.limits.inputBytes) {
+      throw fail("invalidArgument", "Invalid model input");
+    }
+    const references = contexts.map((reference) => {
+      if (!reference || Array.isArray(reference) || Object.keys(reference).some((key) => !["handle", "version"].includes(key))
+        || typeof reference.handle !== "string" || !reference.handle || reference.handle.length > 128
+        || typeof reference.version !== "string" || !reference.version || reference.version.length > 128) {
+        throw fail("invalidArgument", "Invalid context reference");
+      }
+      return Object.freeze({ handle: reference.handle, version: reference.version });
+    });
+    const chat = this.chats.get(chatHandle);
+    if (!chat) throw fail("permissionDenied", "Plugin chat is unavailable");
+    await this.#check(chat, context);
+    if (chat.turn) throw fail("resourceExhausted", "A turn is already active");
+    const controller = new AbortController();
+    const turn = { handle: randomUUID(), chat, controller, reading: false, bytes: 0, sequence: 0, iterator: null, references: Object.freeze(references) };
+    chat.turn = turn;
+    this.turns.set(turn.handle, turn);
+    const abort = () => this.#end(turn, fail("cancelled", "Model turn cancelled"));
+    context.signal?.addEventListener("abort", abort, { once: true });
+    turn.cleanup = () => context.signal?.removeEventListener("abort", abort);
+    turn.timer = setTimeout(() => this.#end(turn, fail("deadlineExceeded", "Model turn timed out")), this.limits.turnMs);
+    try {
+      await this.#guard(turn, context);
+      const selected = [];
+      let bytes = Buffer.byteLength(prompt);
+      for (const reference of references) {
+        const result = await abortable(this.contexts.read(reference, { ...context, signal: controller.signal }), controller.signal);
+        bytes += Buffer.byteLength(result.text);
+        if (bytes > this.limits.inputBytes) throw fail("resourceExhausted", "Model input byte budget exceeded");
+        selected.push(result.text);
+      }
+      await this.#guard(turn, context);
+      this.#prune();
+      const rate = this.rates.get(chat.owner) ?? { startedAt: this.clock(), count: 0, tokens: 0 };
+      // Reserve conservatively using UTF-8 input bytes plus the output cap. Do
+      // not refund unknown provider usage after cancellation or disconnect.
+      const reserved = bytes + 1_024 + this.limits.outputTokens;
+      if ((!this.rates.has(chat.owner) && this.rates.size >= this.limits.rateOwners) || rate.count >= this.limits.requestsPerMinute || rate.tokens + reserved > this.limits.reservedTokensPerMinute) {
+        throw fail("resourceExhausted", "Model request/token budget reached");
+      }
+      rate.count++; rate.tokens += reserved; this.rates.set(chat.owner, rate);
+      await this.#guard(turn, context);
+      const iterator = await abortable(chat.model.invoke({ prompt, contexts: selected, signal: controller.signal,
+        maxOutputTokens: this.limits.outputTokens, maxChunkBytes: this.limits.chunkBytes }), controller.signal);
+      await this.#guard(turn, context);
+      if (!iterator || typeof iterator.next !== "function") throw fail("internal", "Model adapter failed");
+      turn.iterator = iterator;
+      return { turnHandle: turn.handle, maxOutputTokens: this.limits.outputTokens, reservedTokens: reserved };
+    } catch (error) {
+      this.#end(turn, error);
+      throw this.#safe(error);
+    }
+  }
+
+  async #guard(turn, context) {
+    turn.controller.signal.throwIfAborted();
+    await abortable(this.#check(turn.chat, context), turn.controller.signal);
+    for (const reference of turn.references) {
+      await abortable(this.contexts.assertCurrent(reference, { ...context, signal: turn.controller.signal }), turn.controller.signal);
+    }
+    turn.controller.signal.throwIfAborted();
+    if (this.turns.get(turn.handle) !== turn || turn.chat.turn !== turn) throw fail("cancelled", "Model turn ended");
+  }
+
+  #safe(error) {
+    // SDK/provider errors may contain headers, URLs or echoed request content.
+    return error instanceof PluginRpcError ? error : fail("unavailable", "Model invocation failed");
+  }
+
+  async read(turnHandle, context) {
+    const turn = this.turns.get(turnHandle);
+    if (!turn) throw fail("permissionDenied", "Model turn is unavailable");
+    // Validate owner before mutating/aborting another caller's turn.
+    await this.#ownedChat(turn.chat, context);
+    if (turn.reading || !turn.iterator) throw fail("resourceExhausted", "A model read is already pending");
+    turn.reading = true;
+    const abort = () => this.#end(turn, fail("cancelled", "Model read cancelled"));
+    context.signal?.addEventListener("abort", abort, { once: true });
+    try {
+      if (context.signal?.aborted) abort();
+      await this.#guard(turn, context);
+      if (turn.sequence >= this.limits.events) throw fail("resourceExhausted", "Model event budget exceeded");
+      const next = await abortable(turn.iterator.next(), turn.controller.signal);
+      await this.#guard(turn, context);
+      if (next.done) {
+        this.#end(turn);
+        return { kind: "end", sequence: turn.sequence++ };
+      }
+      const event = next.value;
+      if (event?.kind === "text" && typeof event.text === "string") {
+        const bytes = Buffer.byteLength(event.text);
+        if (bytes > this.limits.chunkBytes || turn.bytes + bytes > this.limits.outputBytes) {
+          throw fail("resourceExhausted", "Model output byte budget exceeded");
+        }
+        turn.bytes += bytes;
+        return { kind: "text", sequence: turn.sequence++, text: event.text };
+      }
+      if (event?.kind === "usage") {
+        const usage = {};
+        for (const key of ["inputTokens", "outputTokens", "totalTokens"]) {
+          if (event[key] !== undefined) {
+            if (!Number.isSafeInteger(event[key]) || event[key] < 0) throw fail("internal", "Invalid model usage");
+            usage[key] = event[key];
+          }
+        }
+        if ((usage.outputTokens ?? 0) > this.limits.outputTokens) throw fail("resourceExhausted", "Model output token budget exceeded");
+        return { kind: "usage", sequence: turn.sequence++, ...usage };
+      }
+      // A model's tool request is data, never host authorization. This phase
+      // exposes no execution path, and drops raw tool payloads/reasoning/errors.
+      throw fail("permissionDenied", "Model tools and unsupported events are disabled");
+    } catch (error) {
+      this.#end(turn, error);
+      throw this.#safe(error);
+    } finally { turn.reading = false; context.signal?.removeEventListener("abort", abort); }
+  }
+
+  async stop(turnHandle, context) {
+    const turn = this.turns.get(turnHandle);
+    if (!turn) return false;
+    await this.#ownedChat(turn.chat, context);
+    this.#end(turn, fail("cancelled", "Model turn cancelled"));
+    return true;
+  }
+
+  #end(turn, reason = fail("cancelled", "Model turn ended")) {
+    if (this.turns.get(turn.handle) !== turn) return;
+    this.turns.delete(turn.handle);
+    if (turn.chat.turn === turn) turn.chat.turn = null;
+    clearTimeout(turn.timer); turn.cleanup?.(); turn.controller.abort(reason);
+    // Never wait for a provider which ignores cancellation. Its late next()
+    // result is observed by abortable() and cannot re-enter the output path.
+    try { Promise.resolve(turn.iterator?.return?.()).catch(() => {}); } catch {}
+    turn.iterator = null;
+  }
+
+  #close(chat) {
+    if (chat.turn) this.#end(chat.turn);
+    this.chats.delete(chat.handle);
+  }
+  async close(chatHandle, context) {
+    const chat = this.chats.get(chatHandle);
+    if (!chat) return false;
+    await this.#ownedChat(chat, context); this.#close(chat); return true;
+  }
+  releaseRuntime(runtimeId) {
+    for (const chat of [...this.chats.values()]) if (chat.runtimeId === runtimeId) this.#close(chat);
+    this.contexts?.releaseRuntime(runtimeId);
+  }
+  releaseOwner(owner) {
+    for (const chat of [...this.chats.values()]) if (chat.owner === owner) this.#close(chat);
+    this.contexts?.releaseOwner(owner);
+  }
+  shutdown() { this.closed = true; for (const chat of [...this.chats.values()]) this.#close(chat); this.rates.clear(); this.contextSubscription?.dispose(); }
+}
+
+module.exports = { ModelInvocationBroker, MODEL_INVOCATION_LIMITS: LIMITS };

electron/plugins/ai/modelInvocationBroker.test.cjs

@@ -0,0 +1,211 @@
+"use strict";
+const assert = require("node:assert/strict");
+const test = require("node:test");
+const { ModelInvocationBroker } = require("./modelInvocationBroker.cjs");
+const { SelectedContextAuthority } = require("../context/selectedContextAuthority.cjs");
+const { RPC_ERRORS } = require("../rpcRouter.cjs");
+const deferred = () => { let resolve; const promise = new Promise((r) => { resolve = r; }); return { promise, resolve }; };
+const denied = (error) => error.code === RPC_ERRORS.permissionDenied;
+function fixture(options = {}) {
+  const owner = {}; let currentOwner = owner; let current = true; let active = true; let now = 0;
+  const request = new AbortController();
+  const context = { pluginId: "same-id", pluginVersion: "1", runtimeId: "r1", signal: request.signal,
+    assertActive: async () => { request.signal.throwIfAborted(); if (!active) throw Error("stopped"); } };
+  const calls = [];
+  const model = { label: "Host selected model", isCurrent: () => current,
+    invoke: (input) => { calls.push(input); return (async function* () { yield { kind: "text", text: "Summary" };
+      yield { kind: "usage", inputTokens: 7, outputTokens: 2, totalTokens: 9, secret: "DO_NOT_SERIALIZE" }; })(); }, ...options.model };
+  const resolveDataOwner = async () => currentOwner;
+  const contexts = new SelectedContextAuthority({ resolveDataOwner, clock: () => now });
+  const broker = new ModelInvocationBroker({ resolveDataOwner, selectModel: async () => model, contexts,
+    clock: () => now, limits: options.limits });
+  return { broker, context, contexts, calls, request, owner, model, setOwner: (value) => { currentOwner = value; },
+    setCurrent: (value) => { current = value; }, setActive: (value) => { active = value; }, setTime: (value) => { now = value; } };
+}
+
+test("selected synthetic Note traverses actual broker; text/usage are projected without configuration or secrets", async (t) => {
+  const f = fixture(); t.after(() => f.broker.shutdown());
+  const note = await f.contexts.share({ scope: "note", title: "Synthetic note", version: "v1",
+    readText: () => "Synthetic release: add retry button", isCurrent: () => true }, f.context);
+  assert.equal(f.calls.length, 0);
+  const chat = await f.broker.open(f.context);
+  const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Summarize", contexts: [{ handle: note.handle, version: note.version }] }, f.context);
+  assert.deepEqual(f.calls[0].contexts, ["Synthetic release: add retry button"]);
+  assert.equal(f.calls[0].tools, undefined);
+  const events = [await f.broker.read(turn.turnHandle, f.context), await f.broker.read(turn.turnHandle, f.context), await f.broker.read(turn.turnHandle, f.context)];
+  assert.equal(events[0].text, "Summary");
+  assert.deepEqual(events[1], { kind: "usage", sequence: 1, inputTokens: 7, outputTokens: 2, totalTokens: 9 });
+  assert.equal(events[2].kind, "end");
+  assert.doesNotMatch(JSON.stringify({ chat, turn, events, note }), /apiKey|config|DO_NOT_SERIALIZE/);
+});
+
+test("same pluginId with another source owner, runtime or version cannot use handles or stop victim turns", async (t) => {
+  const f = fixture(); t.after(() => f.broker.shutdown());
+  const chat = await f.broker.open(f.context);
+  const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context);
+  for (const changed of [{ runtimeId: "r2" }, { pluginVersion: "2" }]) {
+    await assert.rejects(f.broker.read(turn.turnHandle, { ...f.context, ...changed }), denied);
+    await assert.rejects(f.broker.stop(turn.turnHandle, { ...f.context, ...changed }), denied);
+  }
+  f.setOwner({});
+  await assert.rejects(f.broker.read(turn.turnHandle, f.context), denied);
+  await assert.rejects(f.broker.close(chat.chatHandle, f.context), denied);
+  f.setOwner(f.owner);
+  assert.equal((await f.broker.read(turn.turnHandle, f.context)).text, "Summary");
+});
+
+test("pending read cancels promptly and a provider's late result never escapes", async (t) => {
+  const pending = deferred(); let signal;
+  const f = fixture({ model: { invoke: (input) => { signal = input.signal; return { next: () => pending.promise }; } } });
+  t.after(() => f.broker.shutdown());
+  const chat = await f.broker.open(f.context);
+  const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context);
+  const reading = f.broker.read(turn.turnHandle, f.context);
+  await new Promise(setImmediate);
+  assert.equal(await f.broker.stop(turn.turnHandle, f.context), true);
+  await assert.rejects(reading, (e) => e.code === RPC_ERRORS.cancelled);
+  assert.equal(signal.aborted, true);
+  pending.resolve({ value: { kind: "text", text: "Late output" } });
+  await assert.rejects(f.broker.read(turn.turnHandle, f.context), denied);
+});
+
+test("runtime teardown, owner revoke and request abort discard pending results", async () => {
+  for (const action of ["runtime", "owner", "abort"]) {
+    const pending = deferred();
+    const f = fixture({ model: { invoke: () => ({ next: () => pending.promise }) } });
+    const chat = await f.broker.open(f.context);
+    const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context);
+    const reading = f.broker.read(turn.turnHandle, f.context);
+    await new Promise(setImmediate);
+    if (action === "runtime") f.broker.releaseRuntime("r1");
+    if (action === "owner") f.broker.releaseOwner(f.owner);
+    if (action === "abort") f.request.abort();
+    await assert.rejects(reading);
+    pending.resolve({ value: { kind: "text", text: "Late" } });
+    assert.equal(f.broker.turns.size, 0); f.broker.shutdown();
+  }
+});
+
+test("deadline also aborts providers that ignore AbortSignal", async () => {
+  const f = fixture({ model: { invoke: () => ({ next: () => new Promise(() => {}) }) }, limits: { turnMs: 20 } });
+  const chat = await f.broker.open(f.context);
+  const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context);
+  await assert.rejects(f.broker.read(turn.turnHandle, f.context), (e) => e.code === RPC_ERRORS.deadlineExceeded);
+  assert.equal(f.broker.turns.size, 0); f.broker.shutdown();
+});
+
+test("model switch and chat expiration reject results", async () => {
+  for (const action of ["model", "expiry"]) {
+    const pending = deferred(); const f = fixture({ model: { invoke: () => ({ next: () => pending.promise }) } });
+    const chat = await f.broker.open(f.context);
+    const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context);
+    const reading = f.broker.read(turn.turnHandle, f.context); await new Promise(setImmediate);
+    if (action === "model") f.setCurrent(false); else f.setTime(300_001);
+    pending.resolve({ value: { kind: "text", text: "Late" } });
+    await assert.rejects(reading, denied); f.broker.shutdown();
+  }
+});
+
+test("bounds cover request rate, output bytes/tokens, concurrent read and raw error serialization", async () => {
+  const f = fixture({ limits: { requestsPerMinute: 1 } });
+  const chat = await f.broker.open(f.context);
+  const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context);
+  await f.broker.stop(turn.turnHandle, f.context);
+  await assert.rejects(f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context), (e) => e.code === RPC_ERRORS.resourceExhausted);
+  f.broker.releaseRuntime("r1");
+  const again = await f.broker.open(f.context);
+  await assert.rejects(f.broker.start({ chatHandle: again.chatHandle, prompt: "Hi" }, f.context), (e) => e.code === RPC_ERRORS.resourceExhausted);
+  f.broker.shutdown();
+  for (const event of [{ kind: "text", text: "X".repeat(4097) }, { kind: "usage", outputTokens: 2049 }, { kind: "tool-call", command: "rm -rf /" }]) {
+    const instance = fixture({ model: { invoke: () => (async function* () { yield event; })() } });
+    const selected = await instance.broker.open(instance.context);
+    const started = await instance.broker.start({ chatHandle: selected.chatHandle, prompt: "Hi" }, instance.context);
+    await assert.rejects(instance.broker.read(started.turnHandle, instance.context));
+    assert.equal(instance.broker.turns.size, 0); instance.broker.shutdown();
+  }
+  const broken = fixture({ model: { invoke: () => { throw Error("api-key secret request body"); } } });
+  const selected = await broken.broker.open(broken.context);
+  await assert.rejects(broken.broker.start({ chatHandle: selected.chatHandle, prompt: "Hi" }, broken.context), (e) => !/secret|api-key/.test(e.message));
+  broken.broker.shutdown();
+});
+
+test("revoking selected context aborts an in-flight turn; changing its version suppresses late output", async () => {
+  for (const action of ["revoke", "version"]) {
+    let current = true; const pending = deferred();
+    const f = fixture({ model: { invoke: () => ({ next: () => pending.promise }) } });
+    const note = await f.contexts.share({ scope: "note", title: "Synthetic", version: "1",
+      readText: () => "Synthetic", isCurrent: () => current }, f.context);
+    const chat = await f.broker.open(f.context);
+    const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi", contexts: [{ handle: note.handle, version: note.version }] }, f.context);
+    const reading = f.broker.read(turn.turnHandle, f.context); await new Promise(setImmediate);
+    if (action === "revoke") f.contexts.release(note.handle); else current = false;
+    pending.resolve({ value: { kind: "text", text: "Late" } });
+    await assert.rejects(reading); f.broker.shutdown();
+  }
+});
+
+test("input, aggregate output, reserved token and concurrent read bounds", async () => {
+  const cases = [
+    { limits: { inputBytes: 2 }, prompt: "too big", start: true },
+    { limits: { reservedTokensPerMinute: 1 }, prompt: "Hi", start: true },
+    { limits: { outputBytes: 2 }, prompt: "Hi", start: false },
+  ];
+  for (const scenario of cases) {
+    const f = fixture({ limits: scenario.limits }); const chat = await f.broker.open(f.context);
+    const start = () => f.broker.start({ chatHandle: chat.chatHandle, prompt: scenario.prompt }, f.context);
+    if (scenario.start) await assert.rejects(start());
+    else { const turn = await start(); await assert.rejects(f.broker.read(turn.turnHandle, f.context)); }
+    assert.equal(f.broker.turns.size, 0); f.broker.shutdown();
+  }
+  const pending = deferred(); const f = fixture({ model: { invoke: () => ({ next: () => pending.promise }) } });
+  const chat = await f.broker.open(f.context); const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context);
+  const reading = f.broker.read(turn.turnHandle, f.context); await new Promise(setImmediate);
+  await assert.rejects(f.broker.read(turn.turnHandle, f.context), (e) => e.code === RPC_ERRORS.resourceExhausted);
+  pending.resolve({ value: { kind: "text", text: "OK" } }); assert.equal((await reading).text, "OK"); f.broker.shutdown();
+});
+
+test("the owner can stop or close after a model switch; read request cancellation aborts the provider", async () => {
+  const f = fixture(); const chat = await f.broker.open(f.context);
+  const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context);
+  f.setCurrent(false); assert.equal(await f.broker.stop(turn.turnHandle, f.context), true);
+  assert.equal(await f.broker.close(chat.chatHandle, f.context), true); f.broker.shutdown();
+  const pending = deferred(); const instance = fixture({ model: { invoke: () => ({ next: () => pending.promise }) } });
+  const selected = await instance.broker.open(instance.context);
+  const started = await instance.broker.start({ chatHandle: selected.chatHandle, prompt: "Hi" }, instance.context);
+  const controller = new AbortController();
+  const reading = instance.broker.read(started.turnHandle, { ...instance.context, signal: controller.signal });
+  await new Promise(setImmediate); controller.abort(); await assert.rejects(reading);
+  assert.equal(instance.broker.turns.size, 0); pending.resolve({ done: true }); instance.broker.shutdown();
+});
+
+test("event and handle caps prevent cheap empty streams and unbounded allocation", async () => {
+  const f = fixture({ limits: { perRuntime: 1, events: 1 }, model: { invoke: () => (async function* () {
+    yield { kind: "text", text: "" }; yield { kind: "text", text: "" };
+  })() } });
+  const chat = await f.broker.open(f.context); await assert.rejects(f.broker.open(f.context));
+  const turn = await f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi" }, f.context);
+  assert.equal((await f.broker.read(turn.turnHandle, f.context)).text, "");
+  await assert.rejects(f.broker.read(turn.turnHandle, f.context), (e) => e.code === RPC_ERRORS.resourceExhausted); f.broker.shutdown();
+});
+
+test("malformed context cannot publish a turn or interrupt two-handle runtime revocation", async (t) => {
+  const entered = deferred(); const resume = deferred(); let checks = 0;
+  const f = fixture({ model: { isCurrent: async () => { if (++checks === 3) { entered.resolve(); await resume.promise; } return true; } } });
+  t.after(() => { resume.resolve(); f.broker.shutdown(); });
+  const selection = { scope: "note", title: "Synthetic", version: "1", readText: () => "Synthetic", isCurrent: () => true };
+  await f.contexts.share(selection, f.context); await f.contexts.share(selection, f.context);
+  const chat = await f.broker.open(f.context);
+  const start = f.broker.start({ chatHandle: chat.chatHandle, prompt: "Hi", contexts: [null] }, f.context);
+  const outcome = start.then(() => "accepted", (error) => error);
+  // On the old implementation, reach the reported post-publication model guard.
+  // On the fixed implementation, validation rejects before that await exists.
+  await Promise.race([entered.promise, outcome]);
+  assert.doesNotThrow(() => f.contexts.releaseRuntime("r1"));
+  assert.equal(f.contexts.handles.size, 0);
+  resume.resolve();
+  const error = await outcome;
+  assert.equal(error.code, RPC_ERRORS.invalidArgument);
+  assert.equal(f.broker.turns.size, 0);
+  assert.equal(f.calls.length, 0);
+  f.broker.shutdown();
+});

electron/plugins/context/selectedContextAuthority.cjs

@@ -0,0 +1,165 @@
+"use strict";
+
+const { randomUUID } = require("node:crypto");
+const { PluginRpcError, RPC_ERRORS } = require("../rpcRouter.cjs");
+const { assertPluginJsonValue } = require("../jsonBoundary.cjs");
+const fail = (code, message) => new PluginRpcError(RPC_ERRORS[code], message);
+const SCOPES = new Set(["workspace", "tab", "session", "host", "note"]);
+const LIMITS = Object.freeze({ handles: 128, perRuntime: 8, contentBytes: 16_384, ttlMs: 300_000, readMs: 5_000 });
+
+// Internal host authority. Only trusted selection/review UI may call share().
+// A menu id, plugin supplied entity id or global inventory is never accepted.
+class SelectedContextAuthority {
+  constructor({ resolveDataOwner, clock = Date.now, limits = {} }) {
+    if (typeof resolveDataOwner !== "function") throw new TypeError("Host data-owner resolver required");
+    this.resolveDataOwner = resolveDataOwner;
+    this.clock = clock;
+    this.limits = Object.freeze(Object.fromEntries(Object.entries(LIMITS).map(([key, max]) => {
+      const value = limits[key] ?? max;
+      if (!Number.isSafeInteger(value) || value < 1 || value > max) throw new TypeError(`Invalid ${key} limit`);
+      return [key, value];
+    })));
+    this.handles = new Map();
+    this.closed = false;
+    this.listeners = new Set();
+  }
+
+  async binding(context) {
+    context.signal?.throwIfAborted();
+    await context.assertActive();
+    const owner = await this.resolveDataOwner(context);
+    await context.assertActive();
+    if (!owner || (typeof owner !== "object" && typeof owner !== "string")) {
+      throw fail("permissionDenied", "Host data owner is unavailable");
+    }
+    context.signal?.throwIfAborted();
+    return { owner, runtimeId: context.runtimeId, pluginVersion: context.pluginVersion };
+  }
+
+  async #check(entry, context) {
+    const binding = await this.binding(context);
+    if (this.closed || this.handles.get(entry.handle) !== entry || entry.expiresAt <= this.clock()
+      || binding.owner !== entry.owner || binding.runtimeId !== entry.runtimeId
+      || binding.pluginVersion !== entry.pluginVersion) throw fail("permissionDenied", "Selected context is unavailable");
+    context.signal?.throwIfAborted();
+    // The trusted reader checks the current window/resource generation and owner.
+    // Recheck our map after awaiting it: close/revoke/version changes can race.
+    const current = await entry.isCurrent();
+    const currentOwner = await this.resolveDataOwner(context);
+    await context.assertActive();
+    context.signal?.throwIfAborted();
+    if (!current || currentOwner !== entry.owner || this.handles.get(entry.handle) !== entry || this.closed || entry.expiresAt <= this.clock()) {
+      throw fail("permissionDenied", "Selected context changed");
+    }
+  }
+
+  #prune() {
+    for (const [handle, entry] of this.handles) if (entry.expiresAt <= this.clock()) this.release(handle);
+  }
+
+  async share(selection, context) {
+    if (this.closed) throw fail("unavailable", "Selected context is closed");
+    assertPluginJsonValue({ scope: selection.scope, title: selection.title, version: selection.version }, { maxBytes: 1024 });
+    if (!SCOPES.has(selection.scope) || typeof selection.title !== "string" || selection.title.length > 160
+      || typeof selection.version !== "string" || !selection.version || selection.version.length > 128
+      || typeof selection.readText !== "function" || typeof selection.isCurrent !== "function") {
+      throw fail("invalidArgument", "Invalid selected context");
+    }
+    // This API accepts a curated text reader, never a Host/Vault DTO. Passwords,
+    // private keys and arbitrary entity fields have no place in the descriptor.
+    if (Object.keys(selection).some((key) => !["scope", "title", "version", "readText", "isCurrent"].includes(key))) {
+      throw fail("invalidArgument", "Selected context contains unsupported fields");
+    }
+    const binding = await this.binding(context);
+    if (!await selection.isCurrent()) throw fail("permissionDenied", "Selection is no longer current");
+    await context.assertActive();
+    this.#prune();
+    if (this.closed || this.handles.size >= this.limits.handles
+      || [...this.handles.values()].filter((entry) => entry.runtimeId === context.runtimeId).length >= this.limits.perRuntime) {
+      throw fail("resourceExhausted", "Selected context handle limit reached");
+    }
+    const entry = { ...binding, ...selection, handle: randomUUID(), expiresAt: this.clock() + this.limits.ttlMs };
+    this.handles.set(entry.handle, entry);
+    return this.#summary(entry);
+  }
+
+  #summary(entry) {
+    return { handle: entry.handle, scope: entry.scope, title: entry.title, version: entry.version, expiresAt: entry.expiresAt };
+  }
+
+  async summaries(context) {
+    const binding = await this.binding(context);
+    this.#prune();
+    const summaries = [];
+    for (const entry of [...this.handles.values()]) {
+      if (entry.owner !== binding.owner || entry.runtimeId !== binding.runtimeId || entry.pluginVersion !== binding.pluginVersion) continue;
+      try { await this.#check(entry, context); summaries.push(this.#summary(entry)); }
+      catch (error) { if (context.signal?.aborted) throw error; this.release(entry.handle); }
+    }
+    await context.assertActive();
+    return summaries;
+  }
+
+  async #read({ handle, version }, context) {
+    const entry = this.handles.get(handle);
+    if (!entry || entry.version !== version) throw fail("permissionDenied", "Selected context is unavailable");
+    await this.#check(entry, context);
+    let text;
+    try { text = await entry.readText({ signal: context.signal, maxBytes: this.limits.contentBytes }); }
+    catch { throw fail("unavailable", "Selected context reader failed"); }
+    await this.#check(entry, context);
+    if (typeof text !== "string" || text.includes("\0") || Buffer.byteLength(text) > this.limits.contentBytes) {
+      throw fail("resourceExhausted", "Selected context exceeds its text limit");
+    }
+    // Do not disclose private key material even if a curated text reader errs.
+    if (/-----BEGIN (?:[A-Z0-9]+ )*PRIVATE KEY-----/.test(text)) {
+      throw fail("permissionDenied", "Private key material cannot be shared");
+    }
+    return { ...this.#summary(entry), text };
+  }
+
+  async read(reference, context) {
+    const controller = new AbortController();
+    const timeout = setTimeout(() => controller.abort(fail("deadlineExceeded", "Selected context read timed out")), this.limits.readMs);
+    const signal = context.signal ? AbortSignal.any([context.signal, controller.signal]) : controller.signal;
+    let abort;
+    try {
+      signal.throwIfAborted();
+      return await new Promise((resolve, reject) => {
+        abort = () => reject(signal.reason);
+        signal.addEventListener("abort", abort, { once: true });
+        this.#read(reference, { ...context, signal }).then(resolve, reject);
+      });
+    } finally {
+      clearTimeout(timeout);
+      if (abort) signal.removeEventListener("abort", abort);
+    }
+  }
+
+  async assertCurrent({ handle, version }, context) {
+    const entry = this.handles.get(handle);
+    if (!entry || entry.version !== version) throw fail("permissionDenied", "Selected context is unavailable");
+    await this.#check(entry, context);
+  }
+  onDidInvalidate(listener) {
+    this.listeners.add(listener);
+    return { dispose: () => this.listeners.delete(listener) };
+  }
+  release(handle) {
+    if (!this.handles.delete(handle)) return;
+    for (const listener of [...this.listeners]) {
+      // Revocation must finish even when an observer fails. Never log selected
+      // text or let an observer exception retain another runtime handle.
+      try { listener(handle); } catch {}
+    }
+  }
+  releaseRuntime(runtimeId) {
+    for (const [handle, entry] of this.handles) if (entry.runtimeId === runtimeId) this.release(handle);
+  }
+  releaseOwner(owner) {
+    for (const [handle, entry] of this.handles) if (entry.owner === owner) this.release(handle);
+  }
+  shutdown() { this.closed = true; for (const handle of [...this.handles.keys()]) this.release(handle); this.listeners.clear(); }
+}
+
+module.exports = { SelectedContextAuthority, SELECTED_CONTEXT_LIMITS: LIMITS };

electron/plugins/context/selectedContextAuthority.test.cjs

@@ -0,0 +1,87 @@
+"use strict";
+const assert = require("node:assert/strict");
+const test = require("node:test");
+const { SelectedContextAuthority } = require("./selectedContextAuthority.cjs");
+const deferred = () => { let resolve; const promise = new Promise((r) => { resolve = r; }); return { promise, resolve }; };
+function fixture() {
+  const owner = {}; let currentOwner = owner; let current = true; let version = "1"; let reads = 0; let now = 0;
+  const context = { pluginId: "same", pluginVersion: "1", runtimeId: "r1", assertActive: async () => {} };
+  const authority = new SelectedContextAuthority({ resolveDataOwner: async () => currentOwner, clock: () => now });
+  const selection = { scope: "note", title: "Synthetic Note", version: "1", isCurrent: () => current && version === "1",
+    readText: () => { reads++; return "synthetic content"; } };
+  return { owner, context, authority, selection, reads: () => reads, ownerTo: (v) => { currentOwner = v; },
+    versionTo: (v) => { version = v; }, currentTo: (v) => { current = v; }, timeTo: (v) => { now = v; } };
+}
+
+test("summaries enumerate only explicitly shared selection and never read content", async () => {
+  const f = fixture(); assert.deepEqual(await f.authority.summaries(f.context), []);
+  const note = await f.authority.share(f.selection, f.context);
+  assert.deepEqual(await f.authority.summaries(f.context), [note]); assert.equal(f.reads(), 0);
+  assert.equal((await f.authority.read(note, f.context)).text, "synthetic content");
+  assert.equal(f.reads(), 1);
+  assert.deepEqual(await f.authority.summaries({ ...f.context, runtimeId: "r2" }), []);
+  f.ownerTo({}); assert.deepEqual(await f.authority.summaries(f.context), []);
+  await assert.rejects(f.authority.read(note, f.context));
+  f.ownerTo(f.owner); assert.equal((await f.authority.read(note, f.context)).text, "synthetic content");
+});
+
+test("scope, owner, expected version and resource lifetime are enforced before read", async () => {
+  for (const action of ["version", "owner", "closed", "expiry", "release", "runtime"]) {
+    const f = fixture(); const note = await f.authority.share(f.selection, f.context);
+    if (action === "version") f.versionTo("2");
+    if (action === "owner") f.currentTo(false);
+    if (action === "closed") f.authority.shutdown();
+    if (action === "expiry") f.timeTo(300_001);
+    if (action === "release") f.authority.release(note.handle);
+    if (action === "runtime") f.authority.releaseRuntime("r1");
+    await assert.rejects(f.authority.read(note, f.context)); assert.equal(f.reads(), 0);
+  }
+  const f = fixture(); const note = await f.authority.share(f.selection, f.context);
+  await assert.rejects(f.authority.read({ ...note, version: "2" }, f.context));
+  await assert.rejects(f.authority.read({ handle: "menu-id", version: "1" }, f.context));
+  await assert.rejects(f.authority.share({ ...f.selection, scope: "all" }, f.context));
+});
+
+test("revocation/version/owner replacement while reader awaits suppresses late content", async () => {
+  for (const action of ["release", "version", "owner", "source"]) {
+    const f = fixture(); const pending = deferred(); let started;
+    const entered = new Promise((r) => { started = r; });
+    const note = await f.authority.share({ ...f.selection, readText: () => { started(); return pending.promise; } }, f.context);
+    const reading = f.authority.read(note, f.context); await entered;
+    if (action === "release") f.authority.release(note.handle);
+    if (action === "version") f.versionTo("2");
+    if (action === "owner") f.currentTo(false);
+    if (action === "source") f.ownerTo({});
+    pending.resolve("late secret text"); await assert.rejects(reading);
+  }
+});
+
+test("unknown Host fields, private keys and oversized content are excluded", async () => {
+  const f = fixture();
+  await assert.rejects(f.authority.share({ ...f.selection, password: "forbidden" }, f.context));
+  await assert.rejects(f.authority.share({ ...f.selection, privateKey: "forbidden" }, f.context));
+  for (const text of ["-----BEGIN OPENSSH PRIVATE KEY-----\nforbidden", "X".repeat(16_385)]) {
+    const note = await f.authority.share({ ...f.selection, readText: () => text }, f.context);
+    await assert.rejects(f.authority.read(note, f.context));
+  }
+});
+
+test("content reader has a deadline and respects request cancellation", async () => {
+  const owner = {}; const context = { runtimeId: "r", pluginVersion: "1", assertActive: async () => {} };
+  const authority = new SelectedContextAuthority({ resolveDataOwner: () => owner, limits: { readMs: 20 } });
+  const note = await authority.share({ scope: "note", title: "Synthetic", version: "1", isCurrent: () => true,
+    readText: () => new Promise(() => {}) }, context);
+  await assert.rejects(authority.read(note, context), /timed out/);
+  const controller = new AbortController(); const reading = authority.read(note, { ...context, signal: controller.signal });
+  controller.abort(); await assert.rejects(reading); authority.shutdown();
+});
+
+test("one failing invalidation listener cannot interrupt other listeners or handle cleanup", async () => {
+  const f = fixture(); const first = await f.authority.share(f.selection, f.context); const second = await f.authority.share(f.selection, f.context);
+  const notified = [];
+  f.authority.onDidInvalidate(() => { throw Error("listener failed"); });
+  f.authority.onDidInvalidate((handle) => notified.push(handle));
+  assert.doesNotThrow(() => f.authority.releaseRuntime("r1"));
+  assert.equal(f.authority.handles.size, 0);
+  assert.deepEqual(notified, [first.handle, second.handle]);
+});

electron/plugins/modelInvocation.test.cjs

@@ -0,0 +1,4 @@
+"use strict";
+// Include the isolated domain suites in npm run test:plugin-runtime.
+require("./ai/modelInvocationBroker.test.cjs");
+require("./context/selectedContextAuthority.test.cjs");