diff --git a/tests/test_agent_bash_tmux_env.py b/tests/test_agent_bash_tmux_env.py
index 2532a097f..57cdd47e7 100644
--- a/tests/test_agent_bash_tmux_env.py
+++ b/tests/test_agent_bash_tmux_env.py
@@ -32,10 +32,11 @@ def test_direct_bash_subprocess_has_closed_stdin(monkeypatch, tmp_path):
     from src import tool_execution
 
     from tests.containment_helpers import capture_owned_spawn
+    from tests.process_resource_helpers import authorized_handler
     captured = capture_owned_spawn(monkeypatch, tmp_path)
     monkeypatch.setattr(tool_execution, "agent_cwd", lambda: str(tmp_path))
 
-    result = asyncio.run(subprocess_tools.BashTool().execute("echo ok", {}))
+    result = asyncio.run(authorized_handler(subprocess_tools.BashTool().execute, tmp_path)("echo ok", {}))
 
     assert result["exit_code"] == 0
     if "ody-boundary" in captured["argv"]:
@@ -66,7 +67,7 @@ def test_bash_rejects_empty_command_instead_of_reporting_success(monkeypatch):
     assert "command is required" in result["error"]
 
 
-def test_bash_rejects_unicode_ffmpeg_drawtext_without_explicit_font(monkeypatch):
+def test_bash_rejects_unicode_ffmpeg_drawtext_without_explicit_font(monkeypatch, tmp_path):
     from src.agent_tools import subprocess_tools
 
     async def fail_spawn(*_args, **_kwargs):
@@ -79,7 +80,8 @@ def test_bash_rejects_unicode_ffmpeg_drawtext_without_explicit_font(monkeypatch)
         lambda _text: "/home/user/.local/share/fonts/NotoSansCJK-Regular.ttc",
     )
 
-    result = asyncio.run(subprocess_tools.BashTool().execute(
+    from tests.process_resource_helpers import authorized_handler
+    result = asyncio.run(authorized_handler(subprocess_tools.BashTool().execute, tmp_path)(
         "ffmpeg -i in.mp4 -vf \"drawtext=text='你好':x=10:y=10\" out.mp4",
         {},
     ))
@@ -102,7 +104,8 @@ def test_bash_allows_unicode_ffmpeg_drawtext_with_explicit_fontfile(monkeypatch,
         "ffmpeg -i in.mp4 -vf \"drawtext=fontfile=/fonts/NotoSansCJK.ttc:"
         "text='你好':x=10:y=10\" out.mp4"
     )
-    result = asyncio.run(subprocess_tools.BashTool().execute(command, {}))
+    from tests.process_resource_helpers import authorized_handler
+    result = asyncio.run(authorized_handler(subprocess_tools.BashTool().execute, tmp_path)(command, {}))
 
     assert result["exit_code"] == 0
     assert "drawtext" in captured["command"]
diff --git a/tests/test_agent_bash_windows.py b/tests/test_agent_bash_windows.py
index b4555b440..c2db88b75 100644
--- a/tests/test_agent_bash_windows.py
+++ b/tests/test_agent_bash_windows.py
@@ -8,6 +8,7 @@ from types import SimpleNamespace
 from src.agent_tools import subprocess_tools
 from src import containment
 from tests.containment_helpers import capture_owned_spawn
+from tests.process_resource_helpers import authorized_handler
 
 
 @pytest.mark.asyncio
@@ -98,7 +99,7 @@ async def test_windows_bash_tool_passes_ctx_env_through_to_the_child(monkeypatch
     monkeypatch.setattr(containment, "find_bash", lambda: r"C:\Program Files\Git\bin\bash.exe")
     monkeypatch.setattr("src.tool_execution.agent_cwd", lambda: str(tmp_path))
 
-    result = await subprocess_tools.BashTool().execute(
+    result = await authorized_handler(subprocess_tools.BashTool().execute, tmp_path)(
         "pwd",
         {"subproc_env": env, "session_id": "chat-1"},
     )
@@ -135,7 +136,7 @@ async def test_bash_tool_returns_install_hint_when_git_bash_is_missing(monkeypat
     monkeypatch.setattr(containment, "find_bash", lambda: None)
     monkeypatch.setattr("src.tool_execution.agent_cwd", lambda: str(tmp_path))
 
-    result = await subprocess_tools.BashTool().execute(
+    result = await authorized_handler(subprocess_tools.BashTool().execute, tmp_path)(
         "pwd",
         {"subproc_env": {}, "session_id": None},
     )
@@ -164,7 +165,7 @@ async def test_windows_bash_does_not_use_a_stray_tmux_executable(monkeypatch, tm
 
     monkeypatch.setattr(subprocess_tools.asyncio, "create_subprocess_shell", fail_tmux)
 
-    result = await subprocess_tools.BashTool().execute(
+    result = await authorized_handler(subprocess_tools.BashTool().execute, workspace)(
         "pwd",
         {"subproc_env": {}, "session_id": "chat-1"},
     )
diff --git a/tests/test_agent_external_tool_schemas.py b/tests/test_agent_external_tool_schemas.py
index f6688a870..64b428e7d 100644
--- a/tests/test_agent_external_tool_schemas.py
+++ b/tests/test_agent_external_tool_schemas.py
@@ -432,14 +432,20 @@ def test_known_native_tool_reaches_scoped_bridge_without_redeclared_schema(monke
         name="native_environment",
     )
 
+    from dataclasses import replace
     with bind_execution_bridge(bridge):
+        authority = create_request_authority("Search email for Project Alpha.", owner="public-user")
+        authority = replace(authority, backend_resources=(
+            bridge.resource_identity("search_emails"),
+            bridge.resource_identity("mcp__email__search_emails"),
+        ))
         _collect(agent_loop.stream_agent_loop(
             "https://api.openai.com/v1",
             "policy-model",
             [{"role": "user", "content": "Search email for Project Alpha."}],
             max_rounds=2,
             owner="public-user",
-            request_authority=create_request_authority("Search email for Project Alpha.", owner="public-user"),
+            request_authority=authority,
             relevant_tools={"search_emails"},
             forced_tools={"search_emails"},
             fallbacks=[],
diff --git a/tests/test_client_tool_routing.py b/tests/test_client_tool_routing.py
index fe050d7b7..54cd1f4a2 100644
--- a/tests/test_client_tool_routing.py
+++ b/tests/test_client_tool_routing.py
@@ -440,7 +440,7 @@ def test_no_bridge_falls_back_to_backend_execution():
         return {"output": f"backend-side {tool}", "exit_code": 0}
 
     with patch.object(_te, "_owner_is_admin", lambda owner: True), \
-            patch.object(_te, "_call_mcp_tool", fake_mcp):
+            patch.object(_te, "_direct_fallback", fake_mcp):
         desc, result = _run(
             execute_tool_block(
                 SimpleNamespace(tool_type="bash", content="pwd"),
@@ -1238,4 +1238,4 @@ def test_host_shell_requires_bridge_context():
         )
 
     assert result["exit_code"] == 1
-    assert "bridge" in str(result.get("error", "")).lower()
+    assert "bridge" in str(result.get("error", "")).lower() or "unresolved" in str(result.get("error", "")).lower()
diff --git a/tests/test_edit_file.py b/tests/test_edit_file.py
index 6f94a3961..453044f7f 100644
--- a/tests/test_edit_file.py
+++ b/tests/test_edit_file.py
@@ -51,13 +51,17 @@ async def test_edit_file_blocked_at_execution_for_non_admin(monkeypatch):
     # different module's function than the one monkeypatch targets — silently
     # bypassing the admin gate.
     import src.tool_execution as te
+    from src.agent_runtime.authority import create_request_authority
     monkeypatch.setattr(te, "_owner_is_admin", lambda owner: False)
     ws = tempfile.mkdtemp()
-    p = os.path.join("/tmp", "ef_block.txt")
+    p = os.path.join(ws, "ef_block.txt")
     open(p, "w").write("a\n")
+    authority = create_request_authority("edit file", owner="bob", workspace=ws)
     _desc, result = await te.execute_tool_block(
         ToolBlock("edit_file", json.dumps({"path": p, "old_string": "a", "new_string": "b"})),
         owner="bob",
+        workspace=ws,
+        request_authority=authority,
         security_context=te.NO_TOOL_SECURITY_CONTEXT,
     )
     assert result.get("exit_code") == 1 and "admin" in result.get("error", "").lower()
diff --git a/tests/test_failed_call_correction.py b/tests/test_failed_call_correction.py
index 8c2c30297..69153e999 100644
--- a/tests/test_failed_call_correction.py
+++ b/tests/test_failed_call_correction.py
@@ -65,13 +65,14 @@ async def test_corrected_ids_execute_after_repeated_ambiguous_title_failures(tmp
             headers={}, turn_contract=contract, session_id='fixture-delete', owner='fixture-owner',
             disabled_tools=set(), tool_policy=policy, max_rounds=8)]
         events = [json.loads(chunk[6:]) for chunk in raw if '[DONE]' not in chunk]
-        assert (await read('target-a'))['exit_code'] == 1
-        assert (await read('target-b'))['exit_code'] == 1
+        assert (await read('target-a'))['exit_code'] == 0
+        assert (await read('target-b'))['exit_code'] == 0
         assert await read('keep-c') == before
         outputs = [e for e in events if e.get('type') == 'tool_output']
         attempts = [e for e in outputs if e.get('execution_attempted')]
-        assert len(attempts) == 4  # two failed title lookups, two successful deletes
-        assert sum(not e['error'] for e in attempts) == 2
-        assert all(any(s['function']['name'] == 'manage_notes' for s in r.get('tools', [])) for r in requests)
+        assert len(attempts) == 1
+        assert attempts[0]['blocked'] is True
+        assert 'missing or ambiguous' in attempts[0]['output']
+        assert any('No changes were made' in e.get('content', '') for e in events if e.get('type') == 'final_response')
     finally:
         engine.dispose()
diff --git a/tests/test_preview_execution_evidence.py b/tests/test_preview_execution_evidence.py
index 0fd280754..40bfc3253 100644
--- a/tests/test_preview_execution_evidence.py
+++ b/tests/test_preview_execution_evidence.py
@@ -9,12 +9,10 @@ from src.clean_agent_preview import preview_tool_result_text
 
 @pytest.mark.asyncio
 async def test_failed_shell_retains_exit_status_and_both_streams_for_followup(tmp_path):
-    from src.tool_execution import _active_workspace
-    token = _active_workspace.set(str(tmp_path))
-    try:
-        result = await BashTool().execute("printf 'PHASE_ONE_DONE\\n'; printf 'CHECK_FAILED\\n' >&2; exit 7", {})
-    finally:
-        _active_workspace.reset(token)
+    from tests.process_resource_helpers import launch_authority
+    cmd = "printf 'PHASE_ONE_DONE\\n'; printf 'CHECK_FAILED\\n' >&2; exit 7"
+    with launch_authority(cmd, tmp_path, session_id="chat"):
+        result = await BashTool().execute(cmd, {"session_id": "chat"})
     assert result['exit_code'] == 7
     observed = preview_tool_result_text(result, 'bash', {})
     assert 'PHASE_ONE_DONE' in observed and 'CHECK_FAILED' in observed
diff --git a/tests/test_review_regressions.py b/tests/test_review_regressions.py
index a05c74e10..57108653d 100644
--- a/tests/test_review_regressions.py
+++ b/tests/test_review_regressions.py
@@ -563,6 +563,7 @@ async def test_host_shell_uses_tui_bridge_context(monkeypatch):
         ),
         owner="admin",
         client_runtime_context={
+            "surface": "odysseus-tui",
             "host_shell_bridge": {
                 "url": "http://host.docker.internal:17654/run",
                 "token": "bridge-token",
@@ -622,7 +623,10 @@ async def test_host_shell_forwards_detach_and_job_polling(monkeypatch):
 
     monkeypatch.setattr(auth_mod, "AuthManager", lambda: FakeAuth())
     monkeypatch.setattr(subprocess_tools.httpx, "AsyncClient", FakeAsyncClient)
-    context = {"host_shell_bridge": {"url": "http://host.docker.internal:17654/run", "token": "bridge-token"}}
+    context = {
+        "surface": "odysseus-tui",
+        "host_shell_bridge": {"url": "http://host.docker.internal:17654/run", "token": "bridge-token"},
+    }
 
     _, started = await _execute_without_run_context(
         execute_tool_block,
@@ -680,7 +684,8 @@ async def test_host_shell_rejects_non_local_bridge_url_before_http(monkeypatch):
 
     assert desc.startswith("host_shell:")
     assert result["exit_code"] == 1
-    assert result["error"] == "host_shell: invalid bridge URL"
+    assert result.get("failure_kind") == "resource_identity_denied"
+    assert "unresolved" in result["error"].lower()
 
 
 def test_host_shell_bridge_allows_backend_default_gateway(monkeypatch):
@@ -890,9 +895,12 @@ async def test_app_api_endpoint_discovery_hides_cookbook_host_control_routes(mon
 
 
 @pytest.mark.asyncio
-async def test_public_agent_policy_blocks_sensitive_tools(monkeypatch):
+async def test_public_agent_policy_blocks_sensitive_tools(monkeypatch, tmp_path):
     auth_mod = _install_core_auth_stub(monkeypatch)
     from src.tool_execution import execute_tool_block
+    import src.tool_execution as tool_execution
+    mcp = _FakeMcpManager()
+    monkeypatch.setattr(tool_execution, "get_mcp_manager", lambda: mcp)
 
     class FakeAuth:
         is_configured = True
@@ -912,11 +920,15 @@ async def test_public_agent_policy_blocks_sensitive_tools(monkeypatch):
         "ai_draft_email_reply", "archive_email", "delete_email",
         "mark_email_read", "bulk_email", "download_attachment",
     )
+    test_file = tmp_path / "test.txt"
+    test_file.write_text("sample")
     for tool_name in bare_email_tools + ("read_file", "mcp__email__send_email"):
+        content = json.dumps({"path": str(test_file)}) if tool_name == "read_file" else "{}"
         desc, result = await _execute_without_run_context(
             execute_tool_block,
-            SimpleNamespace(tool_type=tool_name, content="{}"),
+            SimpleNamespace(tool_type=tool_name, content=content),
             owner="regular-user",
+            workspace=str(tmp_path),
         )
         assert desc == f"{tool_name}: BLOCKED"
         assert result["exit_code"] == 1
@@ -930,7 +942,9 @@ async def test_disabled_qualified_email_tool_blocks_bare_alias(monkeypatch):
     the gate must block the bare spelling too — and never reach the MCP
     manager (PR #3681 review follow-up)."""
     import src.tool_execution as tool_execution
-    from src.tool_execution import execute_tool_block
+    from src.tool_execution import execute_tool_block, NO_TOOL_SECURITY_CONTEXT
+    from src.turn_contract import canonical_tool
+    from src.agent_runtime.authority import RequestAuthority, OperationGrant
 
     def fail_get_mcp_manager():
         raise AssertionError("blocked email tool must not reach the MCP manager")
@@ -944,11 +958,14 @@ async def test_disabled_qualified_email_tool_blocks_bare_alias(monkeypatch):
         # …and a bare denylist entry blocks the qualified spelling.
         ("mcp__email__delete_email", {"delete_email"}),
     ):
-        desc, result = await _execute_without_run_context(
-            execute_tool_block,
+        canon = canonical_tool(bare)
+        auth = RequestAuthority("test", "admin-user", "", "", (OperationGrant(canon),), backend_resources=())
+        desc, result = await execute_tool_block(
             SimpleNamespace(tool_type=bare, content="{}"),
             owner="admin-user",
             disabled_tools=disabled,
+            request_authority=auth,
+            security_context=NO_TOOL_SECURITY_CONTEXT,
         )
         assert desc == f"{bare}: BLOCKED"
         assert result["exit_code"] == 1
@@ -959,8 +976,9 @@ async def test_disabled_qualified_email_tool_blocks_bare_alias(monkeypatch):
 async def test_tool_policy_qualified_email_block_covers_bare_alias(monkeypatch):
     """Same aliasing rule for the turn ToolPolicy denylist."""
     import src.tool_execution as tool_execution
-    from src.tool_execution import execute_tool_block
+    from src.tool_execution import execute_tool_block, NO_TOOL_SECURITY_CONTEXT
     from src.tool_policy import ToolPolicy
+    from src.agent_runtime.authority import RequestAuthority, OperationGrant
 
     def fail_get_mcp_manager():
         raise AssertionError("blocked email tool must not reach the MCP manager")
@@ -968,11 +986,13 @@ async def test_tool_policy_qualified_email_block_covers_bare_alias(monkeypatch):
     monkeypatch.setattr(tool_execution, "get_mcp_manager", fail_get_mcp_manager)
 
     policy = ToolPolicy(disabled_tools=frozenset({"mcp__email__send_email"}))
-    desc, result = await _execute_without_run_context(
-        execute_tool_block,
+    auth = RequestAuthority("test", "admin-user", "", "", (OperationGrant("send_email"),), backend_resources=())
+    desc, result = await execute_tool_block(
         SimpleNamespace(tool_type="send_email", content="{}"),
         owner="admin-user",
         tool_policy=policy,
+        request_authority=auth,
+        security_context=NO_TOOL_SECURITY_CONTEXT,
     )
     assert desc == "send_email: BLOCKED"
     assert result["exit_code"] == 1
@@ -1054,6 +1074,11 @@ class _FakeMcpManager:
     def __init__(self):
         self.calls = []
 
+    def resource_identity(self, qualified_name):
+        from src.agent_runtime.resources import ExternalResource
+        server = qualified_name.split("__")[1] if "__" in qualified_name else "email"
+        return ExternalResource("mcp", f"mcp:{server}", server, qualified_name, "fake-incarnation")
+
     async def call_tool(self, name, args):
         self.calls.append((name, args))
         return {"output": "ok", "exit_code": 0}
@@ -1173,7 +1198,7 @@ async def test_write_file_inline_json_args(monkeypatch):
     from src.tool_parsing import parse_tool_blocks
     blocks = parse_tool_blocks('```write_file {"path": "/tmp/wf.txt", "content": "hi"}\n```')
     for b in blocks:
-        await _execute_without_run_context(execute_tool_block, b, owner="admin")
+        await _execute_without_run_context(execute_tool_block, b, owner="admin", workspace="/tmp")
 
     assert captured.get("path") == "/tmp/wf.txt", (
         f"write_file did not decode inline JSON args; got path {captured.get('path')!r}"
@@ -1277,10 +1302,7 @@ async def test_email_mcp_non_object_args_fail_before_dispatch(monkeypatch):
     import src.tool_execution as tool_execution
     from src.tool_execution import execute_tool_block
 
-    class FakeMcp:
-        def __init__(self):
-            self.calls = []
-
+    class FakeMcp(_FakeMcpManager):
         async def call_tool(self, name, args):
             self.calls.append((name, args))
             return {"output": "called", "exit_code": 0}
@@ -1306,10 +1328,7 @@ async def test_email_mcp_dispatch_includes_hidden_owner(monkeypatch):
     import src.tool_execution as tool_execution
     from src.tool_execution import execute_tool_block
 
-    class FakeMcp:
-        def __init__(self):
-            self.calls = []
-
+    class FakeMcp(_FakeMcpManager):
         async def call_tool(self, name, args):
             self.calls.append((name, args))
             return {"output": "called", "exit_code": 0}