fix(hermes): retain managed permissions through native restore

Cover native edit approval within the authorized invocation and record bounded,
versioned tool grants per conversation. Recheck policy and cancellation before
each operation. Restore named custom connections through their configured
identity and reject native route fallback. Bind assigned skill contents across
temporary lease replacements while preserving their write protection.

Qualify real native once/deny/session file effects, provider restart, distinct
same-name tool identities, conversation isolation and unanswered permission
cancellation with the credential-free production ACPX fixture. Update the
reviewed platform closure pins and document the remaining release gates.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/plans/2026-10-06-hermes-native-runner.md

@@ -1629,3 +1629,47 @@ The correction passes 259 focused image/account/catalog/report tests, 128
 native source checks and Product E2E typechecking. These credential-free checks
 include plausible wrong image bytes and forbidden-tool evidence; a browser
 campaign and fresh cloud CI remain separate requirements.
+
+### 2026-10-08 native permission and route restoration corrections
+
+The restrictive-mode browser exposed two approval boundaries for one native
+edit. The managed middleware now covers the native edit requester only within
+the same authorized tool invocation. Nested native edits still require their
+own policy and permission check. Planning and protected-path checks run before
+every operation, including operations with a recorded session grant.
+
+Allow-for-session grants use the closed `paperclip.hermes.permissions.v1`
+metadata record. It contains only a scope digest and native tool names. It
+preserves grants across provider stops and strict restoration, including native
+compaction heads. Changed conversation, model, route, workspace or permission
+policy clears the grant. Unknown or unreadable records fail restoration.
+Concurrent calls to one tool share a confirmed session grant. Allow once and
+deny never create a grant, and cancellation racing a reply prevents execution.
+
+Native Hermes records a named custom provider as `custom`. Restoring through
+that bare alias loses its identity. The bridge validates the recorded route,
+resolves through the selected configuration's provider name and rejects a
+native model or route fallback before starting a turn. Assigned-skill identity
+uses bounded, regular-file content hashes and relative names. Replacing an
+unchanged disposable skill lease therefore preserves the grant; content changes
+invalidate it. Assigned skills remain protected from writes.
+
+The correction passes 50 checks on the pinned Python runtime, 27 focused
+TypeScript integrity/recovery/extension checks and seven packaging checks.
+The native ACPX fixture verifies actual once/deny/session file effects,
+unchanged saved metadata around a provider restart, two same-name tool call
+identities, conversation isolation and Stop while approval is unanswered.
+The selected scripted model has no authentication or paid calls. This is native
+transport coverage, not browser or live-provider acceptance. Earlier failed
+fixture attempts retain their verdicts, including a message-separator oracle
+error, a lost session grant and its bounded timeout.
+The complete local ACPX runtime suite passes all three tests in 106.346 seconds,
+including the 50 pinned Python checks and restoration through a newly copied
+assigned-skill lease. The native file tool refuses an assigned-skill write even
+when it has a session grant. The independently checked skill bytes stay intact.
+
+The reviewed Mac and Linux closure pins change only for the bridge source.
+Fresh cloud materialization, current-head native checks and PR verification are
+required. No local Docker or Rust build is used. Shared configuration adoption,
+routine completion and transcript UI patches remain unapplied outside the fixed
+PR file set. Hermes remains pending the full release qualification matrix.

packages/paperclip-runner/scripts/qualify-hermes-runtime.test.mjs

@@ -4,7 +4,7 @@
 import assert from "node:assert/strict";
 import { execFile } from "node:child_process";
 import { once } from "node:events";
-import { mkdir, mkdtemp, rm, readFile, readdir } from "node:fs/promises";
+import { mkdir, mkdtemp, rm, readFile, readdir, writeFile } from "node:fs/promises";
 import { createServer } from "node:http";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -39,6 +39,212 @@ test("pinned Python qualifies Hermes bridge and wire billing without credentials
   t.diagnostic(output.stderr.trim());
 });
 
+test("native Hermes permissions cover edits once, survive process restore, and stop unanswered writes", {
+  skip: process.env.PAPERCLIP_HERMES_QUALIFY !== "1", timeout: 180_000,
+}, async t => {
+  const root = await mkdtemp(join(tmpdir(), "paperclip-hermes-permissions-"));
+  const workspace = join(root, "workspace");
+  const agentFiles = join(root, "agent-files");
+  const skillSource = join(root, "assigned-source");
+  const assignedText = "---\nname: assigned\ndescription: Native permission fixture\n---\nRead only assigned instructions.\n";
+  await mkdir(workspace); await mkdir(agentFiles); await mkdir(skillSource); await mkdir(join(root, "runtime"));
+  await writeFile(join(skillSource, "SKILL.md"), assignedText);
+  let host;
+  const requests = [];
+  const decisions = [];
+  const failures = [];
+  let stage = "opening";
+  const scopeInputs = [];
+  const permissionRecords = [];
+  let qualified = false;
+  const markers = {
+    ONCE: [{ path: "once.txt", content: "allowed-once" }],
+    DENY: [{ path: "denied.txt", content: "MUST_NOT_EXIST" }],
+    GRANT: [{ path: "granted.txt", content: "session-granted" }],
+    REUSE: [{ path: "reuse-a.txt", content: "session-reuse-a" }, { path: "reuse-b.txt", content: "session-reuse-b" }],
+    PROTECT: [{ path: "set-from-current-assigned-lease", content: "MUST_NOT_REPLACE_ASSIGNED_SKILL" }],
+    ISOLATE: [{ path: "isolated.txt", content: "MUST_NOT_EXIST" }],
+    STOP: [{ path: "stopped.txt", content: "MUST_NOT_EXIST" }],
+  };
+  const server = createServer(async (req, res) => {
+    try {
+      const chunks = [];
+      for await (const chunk of req) chunks.push(chunk);
+      const body = JSON.parse(Buffer.concat(chunks).toString() || "{}");
+      requests.push({ path: req.url, authorizationPresent: req.headers.authorization !== undefined, body });
+      if (req.url !== "/v1/chat/completions") { res.writeHead(404).end(); return; }
+      assert.equal(body.stream, true, "An undeclared auxiliary model request is not permission proof");
+      assert.equal(body.model, "hermes-permission-fixture");
+      const index = body.messages.findLastIndex(message => message.role === "user");
+      const text = JSON.stringify(body.messages[index].content);
+      const marker = Object.keys(markers).find(name => text.includes(`PERMISSION_${name}`));
+      assert.ok(marker, "Model request has no exact permission fixture marker");
+      const prior = requests.filter(row => row.body.stream && JSON.stringify(row.body.messages[row.body.messages.findLastIndex(message => message.role === "user")]?.content).includes(`PERMISSION_${marker}`));
+      assert.ok(prior.length <= 2, "Native fixture attempted an undeclared model retry");
+      const results = body.messages.slice(index + 1).filter(message => message.role === "tool");
+      res.writeHead(200, { "content-type": "text/event-stream" });
+      if (!results.length) {
+        if (marker === "PROTECT") markers.PROTECT[0].path = join(scopeInputs.at(-1).assignedSkills[0], "assigned/SKILL.md");
+        const tool = body.tools.find(tool => tool.function.name === "write_file");
+        assert.ok(tool, "The actual native file tool is missing");
+        const calls = markers[marker].map((args, i) => ({ index: i, id: `permission-${marker}-${i}`, type: "function",
+          function: { name: tool.function.name, arguments: JSON.stringify(args) } }));
+        res.write(`data: ${JSON.stringify({ id: "permission-fixture", choices: [{ index: 0, delta: { role: "assistant", tool_calls: calls }, finish_reason: "tool_calls" }] })}\n\n`);
+      } else {
+        assert.equal(results.length, markers[marker].length);
+        for (const [i, args] of markers[marker].entries()) {
+          assert.equal(results[i].tool_call_id, `permission-${marker}-${i}`);
+          const content = String(results[i].content);
+          if (marker === "PROTECT") {
+            assert.match(content, /Paperclip protects/);
+            assert.equal(await readFile(args.path, "utf8"), assignedText);
+            continue;
+          }
+          if (marker === "DENY" || marker === "ISOLATE") {
+            assert.match(content, /Paperclip denied/);
+            assert.equal(await readFile(join(workspace, args.path)).catch(() => null), null);
+          } else {
+            assert.doesNotMatch(content, /"error"\s*:/);
+            assert.equal(await readFile(join(workspace, args.path), "utf8"), args.content);
+          }
+        }
+        res.write(`data: ${JSON.stringify({ id: "permission-fixture", choices: [{ index: 0, delta: { content: `PERMISSION_${marker}_COMPLETE` }, finish_reason: "stop" }] })}\n\n`);
+      }
+      res.end('data: {"choices":[],"usage":{"prompt_tokens":10,"completion_tokens":5,"total_tokens":15}}\n\ndata: [DONE]\n\n');
+    } catch (error) {
+      failures.push(String(error));
+      // A fixture assertion must settle the native turn rather than trigger
+      // SDK retries or conceal its cause behind the outer test timeout.
+      if (!res.headersSent) res.writeHead(200, { "content-type": "text/event-stream" });
+      res.end('data: {"choices":[{"index":0,"delta":{"content":"PERMISSION_FIXTURE_REJECTED"},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n');
+    }
+  });
+  t.after(async () => {
+    t.diagnostic(JSON.stringify(qualified ? { qualified, permissionRequests: decisions.length,
+      nativeModelCalls: requests.filter(row => row.body.stream).length, restoredPermissionRecords: permissionRecords.length }
+      : { stage, failures, scopeInputs, permissionRecords, decisions: decisions.map(row => ({ marker: row.marker,
+        title: row.request.toolCall?.title, outcome: row.outcome })), modelCalls: requests.filter(row => row.body.stream).length }));
+    await host?.close({ reason: "native permission fixture cleanup" });
+    server.closeAllConnections(); await new Promise(resolve => server.close(resolve));
+    await rm(root, { recursive: true, force: true });
+  });
+  server.listen(0, "127.0.0.1"); await once(server, "listening");
+  const endpoint = `http://127.0.0.1:${server.address().port}/v1`;
+  const options = {
+    runtimeDirectory: join(root, "runtime"), normalizedSessionId: "hermes-native-permission-conversation",
+    workingDirectory: workspace, agent: "hermes", model: "hermes-permission-fixture", permissionMode: "approve-reads",
+    providerPolicy: { readOnly: false }, clientCapabilities: acpxProfileClientCapabilities("hermes"),
+    systemInstructions: "Exercise only the supplied native permission fixture.",
+    runtimeContext: { instructions: { workingCopy: { kind: "agent_files", rootPath: agentFiles, entryPath: "AGENTS.md" } },
+      skills: [{ key: "assigned", runtimeName: "assigned", versionId: "fixture-v1", bundle: {
+        schema: "paperclip.runtime-asset.v1", rootPath: skillSource, digest: "3".repeat(64), manifestDigest: "3".repeat(64),
+        fileCount: 1, totalBytes: Buffer.byteLength(assignedText),
+      } }], mcp: {} },
+    environment: { PATH: process.env.PATH, PAPERCLIP_HERMES_CONNECTION_FINGERPRINT: "2".repeat(64),
+      PAPERCLIP_HERMES_CONFIG_JSON: JSON.stringify({ model: { provider: "custom:paperclip", default: "hermes-permission-fixture" },
+        providers: { paperclip: { base_url: endpoint, transport: "chat_completions", default_model: "hermes-permission-fixture", api_key: "no-key-required" } },
+        paperclip_auth: { protocol: "chat", style: "none" } }),
+    }, signal: AbortSignal.timeout(150_000),
+  };
+  const dependencies = { openRuntime: input => {
+    scopeInputs.push({ policy: JSON.parse(input.launchEnvironment.PAPERCLIP_HERMES_POLICY),
+      assignedSkills: JSON.parse(input.launchEnvironment.PAPERCLIP_HERMES_ASSIGNED_SKILLS) });
+    return openCodexAcpxRuntime(input);
+  } };
+  async function permissionRecord(label) {
+    const runtime = fileURLToPath(new URL(`../provider-assets/hermes/${process.platform}-${process.arch}/`, import.meta.url));
+    const code = "import json, sqlite3, sys\n" +
+      "db=sqlite3.connect('file:'+sys.argv[1]+'?mode=ro',uri=True)\n" +
+      "row=db.execute('select model,model_config from sessions where id=?',(sys.argv[2],)).fetchone()\n" +
+      "print(json.dumps({'model':row[0],'meta':json.loads(row[1])}))\n";
+    const output = await promisify(execFile)(join(runtime, "python/bin/python3.12"), ["-I", "-B", "-c", code,
+      join(host.runtimeRoot(), "hermes-home/state.db"), host.identity().backendSessionId], {
+      timeout: 10_000, maxBuffer: 64 * 1024, env: { PATH: process.env.PATH, HOME: root },
+    });
+    const record = JSON.parse(output.stdout);
+    permissionRecords.push({ label, ...record });
+    return record;
+  }
+  host = await AcpxRuntimeHost.open(options, dependencies);
+  async function run(marker, outcome, expectedDecisions) {
+    stage = marker;
+    const before = decisions.length;
+    const events = [];
+    const turn = host.startTurn({ text: `PERMISSION_${marker}`, requestId: `permission-${marker.toLowerCase()}`,
+      onPermissionRequest: async request => {
+        decisions.push({ marker, request: request.raw, outcome });
+        try {
+          assert.ok(decisions.length - before <= expectedDecisions, "One operation asked for a second permission or lost its session grant");
+          assert.match(request.raw.toolCall.title, /write_file/);
+          if (outcome === "allow_always") assert.ok(request.raw.options.some(option => option.optionId === "allow_session" && option.kind === "allow_always"));
+          return { outcome };
+        } catch (error) {
+          failures.push(`${marker}: ${error}`);
+          return { outcome: "reject_once" };
+        }
+      },
+    });
+    for await (const event of turn.events) events.push(event);
+    assert.equal((await turn.result).status, "completed");
+    assert.deepEqual(failures, []);
+    assert.equal(decisions.length - before, expectedDecisions);
+    // Native tool boundaries emit blank message separators. Preserve the
+    // complete text while comparing its one scripted answer without that space.
+    assert.equal(events.filter(event => event.type === "text_delta" && event.stream !== "thought").map(event => event.text).join("").trim(), `PERMISSION_${marker}_COMPLETE`);
+    assert.deepEqual(failures, []);
+    return events;
+  }
+  await run("ONCE", "allow_once", 1);
+  await run("DENY", "reject_once", 1);
+  await run("GRANT", "allow_always", 1);
+  const granted = await permissionRecord("granted");
+  assert.deepEqual(granted.meta.paperclip_permissions.tools, ["write_file"]);
+  const identity = acpxProviderSessionIdentity(host.identity(), host.binding());
+  stage = "restoring";
+  await host.close({ reason: "permission per-turn process restoration" });
+  assert.deepEqual(await permissionRecord("provider-stopped"), granted);
+  host = await AcpxRuntimeHost.open({ ...options, expectedIdentity: identity }, dependencies);
+  assert.deepEqual(await permissionRecord("provider-restored"), granted);
+  const reused = await run("REUSE", "reject_once", 0);
+  assert.notDeepEqual(scopeInputs[0].assignedSkills, scopeInputs[1].assignedSkills, "Restore did not replace the disposable assigned-skill lease");
+  const ids = markers.REUSE.map((_args, i) => `permission-REUSE-${i}`);
+  for (const id of ids) {
+    assert.equal(reused.filter(event => event.type === "tool_call" && event.tag === "tool_call" && event.toolCallId === id).length, 1);
+    assert.equal(reused.filter(event => event.type === "tool_call" && event.tag === "tool_call_update" && event.status === "completed" && event.toolCallId === id).length, 1);
+  }
+  await run("PROTECT", "reject_once", 0);
+  await host.close({ reason: "permission conversation isolation" });
+  host = await AcpxRuntimeHost.open({ ...options, normalizedSessionId: "hermes-native-permission-other-conversation" }, dependencies);
+  await run("ISOLATE", "reject_once", 1);
+  stage = "STOP";
+  let entered;
+  const waiting = new Promise(resolve => { entered = resolve; });
+  const stop = host.startTurn({ text: "PERMISSION_STOP", requestId: "permission-stop",
+    onPermissionRequest: async (request, context) => {
+      assert.match(request.raw.toolCall.title, /stopped.txt/);
+      decisions.push({ marker: "STOP", request: request.raw, outcome: "cancel" });
+      entered();
+      return await new Promise(resolve => {
+        if (context.signal.aborted) resolve({ outcome: "cancel" });
+        else context.signal.addEventListener("abort", () => resolve({ outcome: "cancel" }), { once: true });
+      });
+    },
+  });
+  const drained = (async () => { for await (const _event of stop.events) { /* drain */ } })();
+  await Promise.race([waiting, stop.result.then(result => { throw new Error(`Stop fixture ended without its permission request: ${JSON.stringify(result)}`); })]);
+  await host.interruptActiveTurn("stop unanswered native permission");
+  await drained;
+  assert.equal((await stop.result).status, "cancelled");
+  assert.equal(await readFile(join(workspace, "stopped.txt")).catch(() => null), null);
+  assert.ok(requests.every(row => !row.authorizationPresent));
+  assert.deepEqual(failures, []);
+  assert.deepEqual(Object.fromEntries(Object.keys(markers).map(marker => [marker, requests.filter(row => row.body.stream &&
+    JSON.stringify(row.body.messages[row.body.messages.findLastIndex(message => message.role === "user")]?.content).includes(`PERMISSION_${marker}`)).length])),
+    { ONCE: 2, DENY: 2, GRANT: 2, REUSE: 2, PROTECT: 2, ISOLATE: 2, STOP: 1 });
+  qualified = true;
+  t.diagnostic("Native once/deny/session, provider process restoration with a new assigned-skill lease, protected skill writes, two same-name tool identities, conversation isolation and unanswered permission Stop passed. No paid model or browser proof.");
+});
+
 test("pinned Hermes streams through the production ACPX host using a no-auth local connection", {
   skip: process.env.PAPERCLIP_HERMES_QUALIFY !== "1", timeout: 180_000,
 }, async (t) => {

packages/paperclip-runner/src/drivers/acpx/hermes-distributions.ts

@@ -1,5 +1,5 @@
 /** Reviewed native closures. Setup cannot adopt a digest from downloaded files. */
 export const HERMES_CLOSURES: Readonly<Record<string, string>> = Object.freeze({
-  "darwin-arm64": "2230a296b80cb79079f0e6223449e18affb0e9543a476fb5901fb0524f18a0f5",
-  "linux-x64": "5c49a3018d6f0875bdbc318f934d05814db9c40add9d91b5e834566b3b8b32b4",
+  "darwin-arm64": "91ece73cc9aa57ccf43847d5177a57c3820f8016972bb52d7c39b839c7edd31b",
+  "linux-x64": "f423779a11a2f2340b8f69088b9e1f13f5624084bd14bcc662ec5fbde63c1e95",
 });

packages/paperclip-runner/src/providers/hermes/bridge.py

@@ -8,11 +8,16 @@ from __future__ import annotations
 import asyncio
 from urllib.parse import urlsplit, urlunsplit
 import concurrent.futures
+import hashlib
 import json
 import math
 import os
+import re
+import stat
+import threading
 import uuid
 from collections import deque
+from contextvars import ContextVar
 from pathlib import Path
 
 import acp
@@ -30,6 +35,135 @@ ANSWER_MAX_LENGTH = 65536
 USAGE_COUNTERS = ("input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens", "reasoning_tokens", "total_tokens", "estimated_cost_usd")
 
 
+class SessionPermissions:
+    """Tool grants for one recorded conversation, never global Hermes approvals."""
+    SCHEMA = "paperclip.hermes.permissions.v1"
+    MAX_TOOLS = 256
+
+    def __init__(self, record=None):
+        self._lock = threading.RLock()
+        self._operations = {}
+        self.scope = None
+        self.tools = set()
+        if record is not None:
+            if (not isinstance(record, dict) or set(record) != {"schema", "scope", "tools"}
+                    or record["schema"] != self.SCHEMA
+                    or not isinstance(record["scope"], str)
+                    or not re.fullmatch(r"[a-f0-9]{64}", record["scope"])
+                    or not isinstance(record["tools"], list) or len(record["tools"]) > self.MAX_TOOLS
+                    or any(not self.valid_name(name) for name in record["tools"])
+                    or len(set(record["tools"])) != len(record["tools"])):
+                raise ValueError("Hermes saved permission state is unreadable")
+            self.scope, self.tools = record["scope"], set(record["tools"])
+
+    @staticmethod
+    def valid_name(name):
+        return isinstance(name, str) and re.fullmatch(r"[A-Za-z0-9_.:-]{1,240}", name) is not None
+
+    def bind(self, scope):
+        with self._lock:
+            if self.scope != scope:
+                self.scope, self.tools = scope, set()
+                self._operations = {}
+
+    def allow(self, name, ask, ensure_active):
+        if not self.valid_name(name):
+            raise PermissionError("Invalid native permission operation")
+        with self._lock:
+            if name not in self._operations:
+                if len(self._operations) >= self.MAX_TOOLS:
+                    raise PermissionError("Too many native permission operations")
+                self._operations[name] = threading.Lock()
+            operation = self._operations[name]
+        # Concurrent calls to this tool see the first confirmed session grant.
+        # Other tools can still ask independently. A once/deny never becomes a grant.
+        with operation:
+            ensure_active()
+            with self._lock:
+                granted = name in self.tools
+            if not granted:
+                answer = ask()
+                ensure_active()  # Stop can race delivery of an allowed reply.
+                if answer not in {"once", "session"}:
+                    raise PermissionError("Paperclip denied this operation")
+                if answer == "session":
+                    with self._lock:
+                        if len(self.tools) >= self.MAX_TOOLS:
+                            raise PermissionError("Too many native session permission grants")
+                        self.tools.add(name)
+
+    def record(self):
+        with self._lock:
+            return None if self.scope is None else {
+                "schema": self.SCHEMA, "scope": self.scope, "tools": sorted(self.tools),
+            }
+
+
+def assigned_skill_identity(roots):
+    """Bind sealed assigned contents, not the disposable lease directory name."""
+    if not isinstance(roots, list) or len(roots) > 128:
+        raise ValueError("Hermes assigned skill roots are unreadable")
+    identities, count, total = [], 0, 0
+    def unreadable(error):
+        raise ValueError("Hermes assigned skill inventory is unreadable") from error
+    for root in roots:
+        if not isinstance(root, str) or not Path(root).is_absolute():
+            raise ValueError("Hermes assigned skill root is invalid")
+        root_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+        entries = []
+        try:
+            for directory, dirs, files, directory_fd in os.fwalk(".", dir_fd=root_fd, follow_symlinks=False, onerror=unreadable):
+                for name in sorted(dirs + files):
+                    count += 1
+                    if count > 100000:
+                        raise ValueError("Hermes assigned skills exceed the managed file limit")
+                    before = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
+                    relative = (Path(directory) / name).as_posix()
+                    if stat.S_ISDIR(before.st_mode):
+                        entries.append(["directory", relative])
+                        continue
+                    if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1 or before.st_size > 256 * 1024 * 1024:
+                        raise ValueError("Hermes assigned skills require bounded regular files")
+                    total += before.st_size
+                    if total > 2 * 1024 * 1024 * 1024:
+                        raise ValueError("Hermes assigned skills exceed managed storage limits")
+                    descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory_fd)
+                    with os.fdopen(descriptor, "rb") as stream:
+                        opened = os.fstat(stream.fileno())
+                        if (opened.st_dev, opened.st_ino, opened.st_ctime_ns) != (before.st_dev, before.st_ino, before.st_ctime_ns):
+                            raise ValueError("Hermes assigned skill changed while binding permissions")
+                        digest, size = hashlib.sha256(), 0
+                        while chunk := stream.read(65536):
+                            size += len(chunk)
+                            if size > before.st_size:
+                                raise ValueError("Hermes assigned skill changed while binding permissions")
+                            digest.update(chunk)
+                        after = os.fstat(stream.fileno())
+                        if size != before.st_size or (after.st_ctime_ns, after.st_mtime_ns) != (before.st_ctime_ns, before.st_mtime_ns):
+                            raise ValueError("Hermes assigned skill changed while binding permissions")
+                    entries.append(["file", relative, size, bool(before.st_mode & 0o111), digest.hexdigest()])
+            identities.append(hashlib.sha256(json.dumps(sorted(entries), separators=(",", ":")).encode()).hexdigest())
+        finally:
+            os.close(root_fd)
+    return identities
+
+
+def permission_scope(state, session_id, policy, assigned_identity):
+    # The runner's recovery binding already pins the connection/account and tool
+    # profile. Bind this private metadata to the native conversation and policy too.
+    route = {}
+    for key in ("provider", "base_url", "api_mode"):
+        value = getattr(state.agent, key, None)
+        if isinstance(value, str):
+            if key == "base_url":
+                parts = urlsplit(value)
+                value = urlunsplit(parts._replace(path=parts.path.rstrip("/")))
+            route[key] = value
+    value = {"sessionId": session_id, "cwd": state.cwd, "model": str(state.model),
+             "route": route, "policy": policy, "assignedSkills": assigned_identity}
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
 def committed_human_input_wait(name, result):
     # Only the authenticated assigned Paperclip operation can end native work
     # at this boundary. Other tool output and model text cannot signal a wait.
@@ -174,6 +308,11 @@ class ManagedSessionManager(SessionManager):
                 db.create_session(session_id=head, source="acp", model=state.model, cwd=state.cwd)
             db.replace_messages(head, state.history, active_only=True)
         meta = {"cwd": state.cwd, "paperclip_head": head}
+        permissions = getattr(state, "_paperclip_permissions", None)
+        if permissions is not None:
+            record = permissions.record()
+            if record is not None:
+                meta["paperclip_permissions"] = record
         for key in ("provider", "base_url", "api_mode"):
             value = getattr(agent, key, None)
             if isinstance(value, str) and value.strip():
@@ -211,15 +350,21 @@ class ManagedSessionManager(SessionManager):
         meta = _parse_model_config(row.get("model_config"))
         if not meta.get("cwd"):
             raise ValueError("Hermes session workspace identity is missing")
+        # Existing recordings have no permission metadata and start with no
+        # grants. Unknown or malformed versions must never acquire authority.
+        permissions = SessionPermissions(meta.get("paperclip_permissions"))
         agent = self._make_agent(session_id=head, cwd=meta["cwd"], model=row.get("model"),
                                  api_mode=meta.get("api_mode"), requested_provider=meta.get("provider"),
                                  base_url=meta.get("base_url"))
-        return self._install_state(session_id, agent, meta["cwd"], row.get("model") or agent.model,
-                                   history, persist=False)
+        state = self._install_state(session_id, agent, meta["cwd"], row.get("model") or agent.model,
+                                    history, persist=False)
+        state._paperclip_permissions = permissions
+        return state
 
     def _make_agent(self, **kwargs):
         # Validate the configured route before native construction can try its
         # permissive default-provider fallback.
+        runtime = None
         if self._agent_factory is None:
             from hermes_cli.config import load_config
             from hermes_cli.runtime_provider import resolve_runtime_provider
@@ -228,19 +373,35 @@ class ManagedSessionManager(SessionManager):
             runtime = resolve_runtime_provider(requested=model["provider"], target_model=model["default"])
             if kwargs.get("model") not in (None, model["default"]):
                 raise ValueError("Hermes restored model differs from the selected connection")
+            def normalize(url):
+                parts = urlsplit(url)
+                return urlunsplit(parts._replace(path=parts.path.rstrip("/")))
             for argument, field in (("requested_provider", "provider"), ("base_url", "base_url"), ("api_mode", "api_mode")):
                 actual, expected = kwargs.get(argument), runtime.get(field)
                 if argument == "base_url" and isinstance(actual, str) and isinstance(expected, str):
                     # SDK clients normalize a base URL by appending a slash.
                     # Compare only that normalization; query, host and protocol
                     # remain exact and Paperclip pins the connection fingerprint.
-                    def normalize(url):
-                        parts = urlsplit(url)
-                        return urlunsplit(parts._replace(path=parts.path.rstrip("/")))
                     actual, expected = normalize(actual), normalize(expected)
                 if actual is not None and actual != expected:
                     raise ValueError("Hermes restored route differs from the selected connection")
+            # Native recordings flatten a named custom provider to "custom".
+            # Resolving that bare alias during restore loses its configured
+            # identity and can enter the native default-provider fallback.
+            # Resolve through the selected configuration after validating the
+            # recording; its exact endpoint and protocol still must agree.
+            kwargs["requested_provider"] = model["provider"]
         agent = super()._make_agent(**kwargs)
+        if runtime is not None:
+            if getattr(agent, "model", None) != model["default"]:
+                raise ValueError("Hermes native model differs from the selected connection")
+            for field in ("provider", "base_url", "api_mode"):
+                actual, expected = getattr(agent, field, None), runtime.get(field)
+                if isinstance(expected, str):
+                    if field == "base_url" and isinstance(actual, str):
+                        actual, expected = normalize(actual), normalize(expected)
+                    if actual != expected:
+                        raise ValueError("Hermes native route differs from the selected connection")
         agent.ephemeral_system_prompt = os.environ.get("PAPERCLIP_HERMES_SYSTEM_INSTRUCTIONS", "")
         # Authorized attachments belong to the selected model. Never route an
         # unknown vision model through Hermes's auxiliary image-description API.
@@ -259,6 +420,7 @@ class ManagedHermesACPAgent(HermesACPAgent):
         self._billing = None
         self._billing_negotiated = False
         self._committed_wait_updates = []
+        self._assigned_skill_identity = assigned_skill_identity(json.loads(os.environ.get("PAPERCLIP_HERMES_ASSIGNED_SKILLS", "[]")))
 
     def _build_model_state(self, state):
         from acp.schema import ModelInfo, SessionModelState
@@ -516,6 +678,47 @@ class ManagedHermesACPAgent(HermesACPAgent):
                 self._dispatch_original = middleware.run_tool_execution_middleware
             policy = json.loads(os.environ["PAPERCLIP_HERMES_POLICY"])
             assigned = json.loads(os.environ.get("PAPERCLIP_HERMES_ASSIGNED_SKILLS", "[]"))
+            permissions = getattr(state, "_paperclip_permissions", None)
+            if permissions is None:
+                permissions = state._paperclip_permissions = SessionPermissions()
+            permissions.bind(permission_scope(state, session_id, policy, self._assigned_skill_identity))
+            approved_operation = ContextVar("paperclip.hermes.approved_operation", default=None)
+
+            def ensure_active():
+                if self._active != (session_id, token) or state.cancel_event.is_set():
+                    raise PermissionError("Paperclip turn has stopped")
+
+            def operation_key(name, args):
+                return name, json.dumps(args, sort_keys=True, separators=(",", ":"))
+
+            def check_permission(name, args):
+                read = authorize_tool(name, args, policy=policy, cwd=state.cwd, assigned_skills=assigned)
+                ensure_active()
+                automatic = (name in REPORTING_TOOLS or name in {"clarify", "todo_list"}
+                             or name in policy.get("paperclipAutomaticTools", []))
+                if policy["permissionMode"] != "approve-all" and not automatic and not (read and policy["permissionMode"] == "approve-reads"):
+                    permissions.allow(name, lambda: callbacks.approval_cb(
+                        name, json.dumps(args), allow_permanent=False, allow_session=True,
+                    ), ensure_active)
+                ensure_active()
+
+            def edit_approval(proposal):
+                # Native Hermes validates and prepares its diff before calling
+                # this requester. Its own approval belongs to the same approved
+                # invocation, not a second UI prompt. Nested/direct native edits
+                # still cross managed policy and permission checks separately.
+                try:
+                    if approved_operation.get() == operation_key(proposal.tool_name, proposal.arguments):
+                        authorize_tool(proposal.tool_name, proposal.arguments, policy=policy,
+                                       cwd=state.cwd, assigned_skills=assigned)
+                        ensure_active()
+                    else:
+                        check_permission(proposal.tool_name, proposal.arguments)
+                    return True
+                except PermissionError:
+                    return False
+
+            callbacks.edit_approval_requester = edit_approval
 
             def dispatch(name, args, execute, **kwargs):
                 # Both sequential inline tools and concurrent registry tools
@@ -526,19 +729,13 @@ class ManagedHermesACPAgent(HermesACPAgent):
 
             def authorized(name, args, execute, call_id):
                 try:
-                    read = authorize_tool(name, args, policy=policy, cwd=state.cwd, assigned_skills=assigned)
-                    if self._active != (session_id, token) or state.cancel_event.is_set():
-                        raise PermissionError("Paperclip turn has stopped")
-                    automatic = (name in REPORTING_TOOLS or name in {"clarify", "todo_list"}
-                                 or name in policy.get("paperclipAutomaticTools", []))
-                    if policy["permissionMode"] != "approve-all" and not automatic and not (read and policy["permissionMode"] == "approve-reads"):
-                        allowed = callbacks.approval_cb(name, json.dumps(args), allow_permanent=False, allow_session=True)
-                        if allowed not in {"once", "session"}:
-                            raise PermissionError("Paperclip denied this operation")
+                    check_permission(name, args)
                     scope = tool_policy.set((policy, assigned))
+                    approval = approved_operation.set(operation_key(name, args))
                     try:
                         return execute(args)
                     finally:
+                        approved_operation.reset(approval)
                         tool_policy.reset(scope)
                 except PermissionError as error:
                     result = json.dumps({"error": str(error)})

packages/paperclip-runner/src/providers/hermes/test_bridge.py

@@ -1,12 +1,15 @@
 import asyncio
 import json
 import threading
+import tempfile
 import unittest
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock, patch
 
 from acp.schema import ClientCapabilities
-from bridge import ManagedHermesACPAgent, ManagedSessionManager, ChannelClient, native_answers, question_set, install_no_auth_transport, disable_background_title_inference, turn_usage, USAGE_COUNTERS
+from bridge import ManagedHermesACPAgent, ManagedSessionManager, SessionPermissions, permission_scope, assigned_skill_identity, ChannelClient, native_answers, question_set, install_no_auth_transport, disable_background_title_inference, turn_usage, USAGE_COUNTERS
 from policy import authorize_tool
 from billing import TurnBilling
 
@@ -110,7 +113,161 @@ class Questions(unittest.TestCase):
         self.assertIsNone(native_answers(form, {"outcome": "cancelled"}, False))
 
 
+class PermissionGrants(unittest.TestCase):
+    def test_concurrent_same_tool_uses_one_confirmed_session_grant(self):
+        permissions = SessionPermissions()
+        permissions.bind("a" * 64)
+        started, release = threading.Event(), threading.Event()
+        calls = []
+        def ask():
+            calls.append("permission")
+            started.set()
+            self.assertTrue(release.wait(5))
+            return "session"
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            first = executor.submit(permissions.allow, "write_file", ask, lambda: None)
+            self.assertTrue(started.wait(5))
+            second = executor.submit(permissions.allow, "write_file", ask, lambda: None)
+            release.set()
+            first.result(timeout=5)
+            second.result(timeout=5)
+        self.assertEqual(calls, ["permission"])
+        self.assertEqual(permissions.record()["tools"], ["write_file"])
+
+    def test_other_tools_do_not_wait_on_an_unanswered_tool_and_once_or_deny_is_not_saved(self):
+        permissions = SessionPermissions()
+        permissions.bind("a" * 64)
+        waiting, other, release = threading.Event(), threading.Event(), threading.Event()
+        def blocked():
+            waiting.set()
+            self.assertTrue(release.wait(5))
+            return "once"
+        def independent():
+            other.set()
+            return "deny"
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            first = executor.submit(permissions.allow, "write_file", blocked, lambda: None)
+            self.assertTrue(waiting.wait(5))
+            second = executor.submit(permissions.allow, "terminal", independent, lambda: None)
+            self.assertTrue(other.wait(5))
+            release.set()
+            first.result(timeout=5)
+            with self.assertRaises(PermissionError):
+                second.result(timeout=5)
+        self.assertEqual(permissions.record()["tools"], [])
+
+    def test_private_record_is_closed_versioned_and_bounded(self):
+        valid = {"schema": SessionPermissions.SCHEMA, "scope": "a" * 64, "tools": ["write_file"]}
+        self.assertEqual(SessionPermissions(valid).record(), valid)
+        for record in [[], {**valid, "schema": "future"}, {**valid, "scope": "unknown"},
+                       {**valid, "tools": ["write_file", "write_file"]}, {**valid, "tools": [None]},
+                       {**valid, "tools": ["tool\ncommand"]}, {**valid, "tools": ["tool" + str(i) for i in range(257)]},
+                       {**valid, "arguments": {"apiKey": "MUST_NOT_BE_STORED"}}]:
+            with self.subTest(record=record), self.assertRaises(ValueError):
+                SessionPermissions(record)
+        full = SessionPermissions({**valid, "tools": ["tool" + str(i) for i in range(256)]})
+        with self.assertRaises(PermissionError):
+            full.allow("new_tool", lambda: "session", lambda: None)
+        self.assertEqual(len(full.record()["tools"]), 256)
+
+    def test_restore_keeps_compatible_grants_and_invalidates_changed_scope_without_token_material(self):
+        state = SimpleNamespace(cwd="/workspace", model="model-A", agent=SimpleNamespace(
+            provider="custom:paperclip", base_url="https://model.test/v1", api_mode="chat_completions", api_key="FIRST_SECRET"))
+        policy = {"permissionMode": "approve-reads", "readOnly": False, "protectedPaths": ["/runtime"]}
+        scope = permission_scope(state, "session-A", policy, [])
+        permissions = SessionPermissions()
+        permissions.bind(scope)
+        permissions.allow("write_file", lambda: "session", lambda: None)
+        restored = SessionPermissions(permissions.record())
+        state.agent.api_key = "REFRESHED_SECRET"
+        state.agent.base_url = "https://model.test/v1/"
+        restored.bind(permission_scope(state, "session-A", policy, []))
+        ask = Mock()
+        restored.allow("write_file", ask, lambda: None)
+        ask.assert_not_called()
+        self.assertNotIn("SECRET", json.dumps(restored.record()))
+        changed = [permission_scope(state, "session-B", policy, []),
+                   permission_scope(state, "session-A", {**policy, "permissionMode": "approve-paperclip"}, []),
+                   permission_scope(state, "session-A", {**policy, "readOnly": True}, []),
+                   permission_scope(state, "session-A", policy, ["/assigned/skill"])]
+        state.cwd = "/other-workspace"
+        changed.append(permission_scope(state, "session-A", policy, []))
+        state.cwd, state.model = "/workspace", "model-B"
+        changed.append(permission_scope(state, "session-A", policy, []))
+        state.model, state.agent.base_url = "model-A", "https://other-model.test/v1"
+        changed.append(permission_scope(state, "session-A", policy, []))
+        for changed_scope in changed:
+            restored = SessionPermissions(permissions.record())
+            restored.bind(changed_scope)
+            self.assertEqual(restored.record()["tools"], [])
+
+
 class Restore(unittest.TestCase):
+    def test_assigned_skill_identity_survives_new_lease_paths_and_detects_changes(self):
+        with tempfile.TemporaryDirectory() as root:
+            first, second = Path(root) / "lease-A", Path(root) / "lease-B"
+            for path in (first, second):
+                (path / "assigned" / "empty").mkdir(parents=True)
+                (path / "assigned" / "SKILL.md").write_text("Original assigned instructions")
+            identity = assigned_skill_identity([str(first)])
+            self.assertEqual(assigned_skill_identity([str(second)]), identity)
+            (second / "assigned" / "SKILL.md").write_text("Changed assigned instructions")
+            self.assertNotEqual(assigned_skill_identity([str(second)]), identity)
+            (second / "assigned" / "SKILL.md").write_text("Original assigned instructions")
+            (second / "assigned" / "empty").rename(second / "assigned" / "renamed")
+            self.assertNotEqual(assigned_skill_identity([str(second)]), identity)
+
+    def test_assigned_skill_identity_rejects_links_and_unsupported_roots(self):
+        with tempfile.TemporaryDirectory() as root:
+            root = Path(root)
+            original = root / "original"; original.write_text("Assigned content")
+            assigned = root / "assigned"; assigned.mkdir()
+            (assigned / "link").symlink_to(original)
+            with self.assertRaises(ValueError):
+                assigned_skill_identity([str(assigned)])
+            (assigned / "link").unlink()
+            (assigned / "link").hardlink_to(original)
+            with self.assertRaises(ValueError):
+                assigned_skill_identity([str(assigned)])
+            for roots in [None, ["relative"], [None], [str(assigned)] * 129]:
+                with self.subTest(roots=roots), self.assertRaises(ValueError):
+                    assigned_skill_identity(roots)
+
+    def test_versioned_permission_metadata_follows_native_compaction_and_legacy_history(self):
+        db = Mock()
+        manager = ManagedSessionManager(db=db)
+        manager._get_db = lambda: db
+        agent = SimpleNamespace(model="test", _session_db=db, session_id="head",
+                                _flush_messages_to_session_db=Mock(return_value=True))
+        permissions = SessionPermissions()
+        permissions.bind("a" * 64)
+        permissions.allow("write_file", lambda: "session", lambda: None)
+        state = SimpleNamespace(session_id="parent", model="test", cwd="/workspace", agent=agent,
+            history=[{"role": "user", "content": "Saved history"}], _paperclip_permissions=permissions)
+        db.get_session.return_value = {"source": "acp"}
+        db.get_messages_as_conversation.return_value = state.history
+        manager._persist(state)
+        saved = [json.loads(c.args[1]) for c in db.update_session_meta.call_args_list]
+        self.assertEqual(len(saved), 2)
+        self.assertEqual(saved[0], saved[1])
+        self.assertEqual(saved[0]["paperclip_permissions"], permissions.record())
+        self.assertEqual(saved[0]["paperclip_head"], "head")
+        db.get_compression_tip.return_value = "head"
+        db.get_compression_chain.return_value = ["parent", "head"]
+        db.get_session.side_effect = lambda identity: {"source": "acp", "model": "test", "model_config": json.dumps(saved[0])}
+        manager._make_agent = Mock(return_value=SimpleNamespace(model="test"))
+        restored = manager._restore("parent")
+        self.assertEqual(restored._paperclip_permissions.record(), permissions.record())
+        legacy = {k: v for k, v in saved[0].items() if k != "paperclip_permissions"}
+        db.get_session.side_effect = lambda identity: {"source": "acp", "model": "test", "model_config": json.dumps(legacy)}
+        self.assertIsNone(manager._restore("parent")._paperclip_permissions.record())
+        for broken in [{"schema": "future"}, {**permissions.record(), "tools": ["write_file", "write_file"]}]:
+            db.get_session.side_effect = lambda identity: {"source": "acp", "model": "test", "model_config": json.dumps({**saved[0], "paperclip_permissions": broken})}
+            manager._make_agent.reset_mock()
+            with self.assertRaisesRegex(ValueError, "saved permission state"):
+                manager._restore("parent")
+            manager._make_agent.assert_not_called()
+
     def test_missing_or_failed_history_never_becomes_an_empty_success(self):
         manager = ManagedSessionManager(agent_factory=lambda: SimpleNamespace())
         db = Mock()
@@ -148,17 +305,42 @@ class Restore(unittest.TestCase):
     def test_restore_accepts_sdk_base_url_normalization_but_rejects_routing_changes(self):
         manager = ManagedSessionManager()
         config = {"model": {"provider": "custom:paperclip", "default": "exact-model"}}
-        route = {"provider": "custom:paperclip", "base_url": "https://example.test/v1?route=a", "api_mode": "chat_completions"}
-        agent = SimpleNamespace()
+        route = {"provider": "custom", "base_url": "https://example.test/v1?route=a", "api_mode": "chat_completions"}
+        agent = SimpleNamespace(**route, model="exact-model")
         with patch("hermes_cli.config.load_config", return_value=config), patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value=route), patch("acp_adapter.session.SessionManager._make_agent", return_value=agent) as native:
-            manager._make_agent(model="exact-model", base_url="https://example.test/v1/?route=a", api_mode="chat_completions")
+            manager._make_agent(model="exact-model", requested_provider="custom", base_url="https://example.test/v1/?route=a", api_mode="chat_completions")
             native.assert_called_once()
+            self.assertEqual(native.call_args.kwargs["requested_provider"], "custom:paperclip")
             for change in [{"model": "other"}, {"base_url": "https://example.test/v2?route=a"}, {"base_url": "https://example.test/v1?route=b"}, {"requested_provider": "other"}, {"api_mode": "responses"}]:
                 native.reset_mock()
                 with self.assertRaises(ValueError):
                     manager._make_agent(**{"model": "exact-model", "base_url": route["base_url"], "api_mode": "chat_completions", **change})
                 native.assert_not_called()
 
+    def test_native_route_fallback_is_rejected_before_a_turn(self):
+        manager = ManagedSessionManager()
+        config = {"model": {"provider": "custom:paperclip", "default": "exact-model"}}
+        route = {"provider": "custom", "base_url": "https://example.test/v1", "api_mode": "chat_completions"}
+        for change in [{"provider": ""}, {"provider": "openrouter"}, {"base_url": "https://other.test/v1"},
+                       {"base_url": "https://example.test/v2"}, {"api_mode": "responses"}, {"model": "other-model"}]:
+            agent = SimpleNamespace(**{**route, "model": "exact-model", **change})
+            with self.subTest(change=change), patch("hermes_cli.config.load_config", return_value=config), \
+                    patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value=route), \
+                    patch("acp_adapter.session.SessionManager._make_agent", return_value=agent):
+                with self.assertRaisesRegex(ValueError, "Hermes native (route|model) differs"):
+                    manager._make_agent(model="exact-model", requested_provider="custom", base_url=route["base_url"], api_mode=route["api_mode"])
+
+    def test_new_native_route_accepts_sdk_url_normalization_without_recorded_arguments(self):
+        manager = ManagedSessionManager()
+        config = {"model": {"provider": "custom:paperclip", "default": "exact-model"}}
+        route = {"provider": "custom", "base_url": "https://example.test/v1", "api_mode": "chat_completions"}
+        agent = SimpleNamespace(**{**route, "base_url": "https://example.test/v1/", "model": "exact-model"})
+        with patch("hermes_cli.config.load_config", return_value=config), \
+                patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value=route), \
+                patch("acp_adapter.session.SessionManager._make_agent", return_value=agent) as native:
+            manager._make_agent()
+        self.assertEqual(native.call_args.kwargs["requested_provider"], "custom:paperclip")
+
     def test_failed_native_flush_cannot_be_reported_as_durable(self):
         db = Mock()
         manager = ManagedSessionManager(db=db)
@@ -172,7 +354,7 @@ class Restore(unittest.TestCase):
 class Controls(unittest.IsolatedAsyncioTestCase):
     async def asyncSetUp(self):
         self.state = SimpleNamespace(runtime_lock=threading.Lock(), is_running=True,
-                                     cancel_event=threading.Event(), agent=Mock(), cwd="/workspace")
+                                     cancel_event=threading.Event(), agent=Mock(), cwd="/workspace", model="test")
         self.manager = Mock()
         self.manager.get_session.return_value = self.state
         self.bridge = ManagedHermesACPAgent(self.manager)
@@ -349,6 +531,64 @@ class Controls(unittest.IsolatedAsyncioTestCase):
             self.assertIn("planning mode permits", update.raw_output)
         self.assertEqual(callbacks.tool_call_meta, {})
 
+    async def test_one_native_write_approval_covers_its_prepared_edit_and_session_reuse(self):
+        from hermes_cli import middleware
+        from acp_adapter.edit_approval import maybe_require_edit_approval, set_edit_approval_requester, reset_edit_approval_requester
+        callbacks = SimpleNamespace(tool_call_ids={}, tool_call_meta={}, approval_cb=Mock(side_effect=["once", "session"]),
+                                    edit_approval_requester=Mock(return_value=False))
+        with tempfile.TemporaryDirectory() as root:
+            self.state.cwd = root
+            written = []
+            def native_edit(args):
+                binding = set_edit_approval_requester(callbacks.edit_approval_requester)
+                try:
+                    refusal = maybe_require_edit_approval("write_file", args)
+                    self.assertIsNone(refusal)
+                    Path(args["path"]).write_text(args["content"])
+                    written.append(args["path"])
+                    return "written"
+                finally:
+                    reset_edit_approval_requester(binding)
+            policy = {"permissionMode": "approve-reads", "readOnly": False, "protectedPaths": [root + "/protected"]}
+            with patch.dict("os.environ", {"PAPERCLIP_HERMES_POLICY": json.dumps(policy)}), patch("hermes_cli.middleware.run_tool_execution_middleware", side_effect=lambda name, args, run, **kwargs: run(args)), patch("bridge.HermesACPAgent._wire_turn_callbacks", return_value=callbacks), patch("bridge._send_update"):
+                self.bridge._wire_turn_callbacks(self.state, "session", Mock(), asyncio.get_running_loop())
+                for index in range(3):
+                    path = str(Path(root, f"proof-{index}.txt"))
+                    args = {"path": path, "content": "native-edit-proof"}
+                    self.assertEqual(middleware.run_tool_execution_middleware("write_file", args, native_edit), "written")
+                    self.assertEqual(Path(path).read_text(), args["content"])
+                self.assertEqual(callbacks.approval_cb.call_count, 2)
+                self.assertEqual(self.state._paperclip_permissions.record()["tools"], ["write_file"])
+                denied = middleware.run_tool_execution_middleware("write_file", {"path": root + "/protected/secret", "content": "x"}, native_edit)
+                self.assertIn("protects", json.loads(denied)["error"])
+                self.assertEqual(callbacks.approval_cb.call_count, 2)
+                self.assertEqual(len(written), 3)
+
+    async def test_native_edit_approval_does_not_inherit_an_unrelated_tool_invocation(self):
+        from hermes_cli import middleware
+        from acp_adapter.edit_approval import EditProposal
+        callbacks = SimpleNamespace(tool_call_ids={}, tool_call_meta={}, approval_cb=Mock(side_effect=["once", "deny"]))
+        with patch.dict("os.environ", {"PAPERCLIP_HERMES_POLICY": json.dumps({"permissionMode": "approve-reads", "readOnly": False})}), patch("hermes_cli.middleware.run_tool_execution_middleware", side_effect=lambda name, args, run, **kwargs: run(args)), patch("bridge.HermesACPAgent._wire_turn_callbacks", return_value=callbacks), patch("bridge._send_update"):
+            self.bridge._wire_turn_callbacks(self.state, "session", Mock(), asyncio.get_running_loop())
+            result = middleware.run_tool_execution_middleware("execute_code", {"code": "native nested edit"}, lambda args: callbacks.edit_approval_requester(EditProposal("write_file", "/workspace/a.txt", None, "x", {"path": "a.txt", "content": "x"})))
+            self.assertFalse(result)
+            self.assertEqual(callbacks.approval_cb.call_count, 2)
+            self.assertEqual(self.state._paperclip_permissions.record()["tools"], [])
+
+    async def test_stop_racing_a_session_approval_cannot_grant_or_execute(self):
+        from hermes_cli import middleware
+        def stop_then_allow(*args, **kwargs):
+            self.state.cancel_event.set()
+            return "session"
+        callbacks = SimpleNamespace(tool_call_ids={}, tool_call_meta={}, approval_cb=Mock(side_effect=stop_then_allow))
+        execute = Mock()
+        with patch.dict("os.environ", {"PAPERCLIP_HERMES_POLICY": json.dumps({"permissionMode": "approve-reads", "readOnly": False})}), patch("hermes_cli.middleware.run_tool_execution_middleware", side_effect=lambda name, args, run, **kwargs: run(args)), patch("bridge.HermesACPAgent._wire_turn_callbacks", return_value=callbacks), patch("bridge._send_update"):
+            self.bridge._wire_turn_callbacks(self.state, "session", Mock(), asyncio.get_running_loop())
+            result = middleware.run_tool_execution_middleware("write_file", {"path": "a.txt", "content": "x"}, execute, tool_call_id="cancelled-write")
+        execute.assert_not_called()
+        self.assertIn("turn has stopped", json.loads(result)["error"])
+        self.assertEqual(self.state._paperclip_permissions.record()["tools"], [])
+
     async def test_managed_restore_retains_history_without_replaying_it_as_live_output(self):
         self.bridge._conn = SimpleNamespace(session_update=AsyncMock())
         history = [{"role": "assistant", "content": "Previous reply"}]