diff --git a/routes/skills_routes.py b/routes/skills_routes.py
index 33b55e3c2..63d723071 100644
--- a/routes/skills_routes.py
+++ b/routes/skills_routes.py
@@ -9,6 +9,7 @@ The on-disk format is SKILL.md (frontmatter + structured body) under
 
 import logging
 import re
+from contextlib import aclosing
 from typing import List, Optional
 
 import httpx
@@ -558,70 +559,74 @@ async def _run_skill_test_job(
 
     messages = list(messages) if isinstance(messages, list) else _skill_test_messages(md, task)
     try:
-        async for chunk in stream_agent_loop(
+        # aclosing: pausing for an approval leaves this loop early, and the
+        # stream's ContextVar cleanup must run in this task, not in asyncio's
+        # finalizer (#6635).
+        async with aclosing(stream_agent_loop(
             url, model, messages, headers=headers,
             temperature=0.3, max_tokens=0, max_rounds=8, owner=owner,
             exact_approval=exact_approval,
             request_authority=(request_authority or (
                 exact_approval.pending.request_authority if exact_approval is not None else None
             ) or RequestAuthority.empty(owner=owner)),
-        ):
-            if not chunk.startswith("data: ") or chunk.strip() == "data: [DONE]":
-                continue
-            try:
-                d = _json.loads(chunk[6:])
-            except Exception:
-                continue
-            if d.get("delta"):
-                say_buf.append(d["delta"]); transcript.append(d["delta"])
-            elif d.get("type") == "tool_start":
-                _flush_say()
-                skill_stats["tool_calls"] += 1
-                cmd = str(d.get("command") or d.get("args") or "")[:300]
-                log.append({"type": "tool_start", "tool": d.get("tool"), "command": cmd})
-                transcript.append(f"\n[tool {d.get('tool')}] {cmd}\n")
-            elif d.get("type") == "tool_output":
-                _flush_say()
-                out = str(d.get("output") or "")[:600]
-                tool_log = {"type": "tool_output", "output": out}
-                approval = d.get("ask_user")
-                if isinstance(approval, dict):
-                    tool_log["ask_user"] = approval
-                log.append(tool_log)
-                transcript.append(f"[output] {out}\n")
-                if (
-                    isinstance(approval, dict)
-                    and approval.get("kind") == "tool_approval"
-                    and approval.get("approval_id")
-                ):
-                    # Manual skill tests have their own polling UI instead of a
-                    # chat session. Pause the run and retain only server-side
-                    # continuation state until the same owner approves/denies
-                    # this exact sealed action.
-                    job["status"] = "awaiting_approval"
-                    job["approval"] = approval
-                    job["_transcript"] = transcript
-                    return
-            elif d.get("type") == "agent_step":
-                _flush_say()
+        )) as stream:
+            async for chunk in stream:
+                if not chunk.startswith("data: ") or chunk.strip() == "data: [DONE]":
+                    continue
                 try:
-                    skill_stats["turns"] = max(skill_stats["turns"], int(d.get("round") or 0))
-                except (TypeError, ValueError):
-                    pass
-                log.append({"type": "agent_step", "round": d.get("round")})
-                transcript.append(f"\n--- round {d.get('round')} ---\n")
-            elif d.get("type") == "metrics":
-                data = d.get("data") or {}
-                try:
-                    skill_stats["turns"] = max(skill_stats["turns"], int(data.get("agent_rounds") or 0))
-                except (TypeError, ValueError):
-                    pass
-                try:
-                    skill_stats["tool_calls"] = max(skill_stats["tool_calls"], int(data.get("tool_calls") or 0))
-                except (TypeError, ValueError):
-                    pass
-            if len(log) > 600:
-                del log[0:len(log) - 600]
+                    d = _json.loads(chunk[6:])
+                except Exception:
+                    continue
+                if d.get("delta"):
+                    say_buf.append(d["delta"]); transcript.append(d["delta"])
+                elif d.get("type") == "tool_start":
+                    _flush_say()
+                    skill_stats["tool_calls"] += 1
+                    cmd = str(d.get("command") or d.get("args") or "")[:300]
+                    log.append({"type": "tool_start", "tool": d.get("tool"), "command": cmd})
+                    transcript.append(f"\n[tool {d.get('tool')}] {cmd}\n")
+                elif d.get("type") == "tool_output":
+                    _flush_say()
+                    out = str(d.get("output") or "")[:600]
+                    tool_log = {"type": "tool_output", "output": out}
+                    approval = d.get("ask_user")
+                    if isinstance(approval, dict):
+                        tool_log["ask_user"] = approval
+                    log.append(tool_log)
+                    transcript.append(f"[output] {out}\n")
+                    if (
+                        isinstance(approval, dict)
+                        and approval.get("kind") == "tool_approval"
+                        and approval.get("approval_id")
+                    ):
+                        # Manual skill tests have their own polling UI instead of a
+                        # chat session. Pause the run and retain only server-side
+                        # continuation state until the same owner approves/denies
+                        # this exact sealed action.
+                        job["status"] = "awaiting_approval"
+                        job["approval"] = approval
+                        job["_transcript"] = transcript
+                        return
+                elif d.get("type") == "agent_step":
+                    _flush_say()
+                    try:
+                        skill_stats["turns"] = max(skill_stats["turns"], int(d.get("round") or 0))
+                    except (TypeError, ValueError):
+                        pass
+                    log.append({"type": "agent_step", "round": d.get("round")})
+                    transcript.append(f"\n--- round {d.get('round')} ---\n")
+                elif d.get("type") == "metrics":
+                    data = d.get("data") or {}
+                    try:
+                        skill_stats["turns"] = max(skill_stats["turns"], int(data.get("agent_rounds") or 0))
+                    except (TypeError, ValueError):
+                        pass
+                    try:
+                        skill_stats["tool_calls"] = max(skill_stats["tool_calls"], int(data.get("tool_calls") or 0))
+                    except (TypeError, ValueError):
+                        pass
+                if len(log) > 600:
+                    del log[0:len(log) - 600]
         _flush_say()
     except Exception as e:
         _flush_say()
@@ -1054,53 +1059,55 @@ async def _run_skill_audit_arm(messages: list[dict], url, model, headers, owner,
         # OpenAI-compat) generate an empty completion, which manifested as
         # the skill test returning nothing while chat (which carries its
         # preset's max_tokens) worked. 4096 matches the chat default.
-        async for chunk in stream_agent_loop(
+        # aclosing: stopping for an approval leaves this loop early (#6635).
+        async with aclosing(stream_agent_loop(
             url, model, messages, headers=headers,
             temperature=0.3, max_tokens=4096, max_rounds=8,
             owner=owner, workload=workload, suppress_skills=True,
             request_authority=(active_request_authority() or RequestAuthority.empty(owner=owner)),
-        ):
-            # Streams can include an SSE event line before the data line,
-            # notably `event: error`. Do not silently discard those failures.
-            payload = next((line[6:] for line in chunk.splitlines() if line.startswith("data: ")), None)
-            if payload is None or payload == "[DONE]":
-                continue
-            try:
-                d = _json.loads(payload)
-            except Exception:
-                continue
-            if d.get("error") or d.get("type") == "error":
-                raise SkillAuditUnavailable(str(d.get("error") or d.get("message") or "Audit stream failed"))
-            if d.get("delta"):
-                transcript.append(d["delta"])
-            elif d.get("type") == "tool_start":
-                stats["tool_calls"] += 1
-                transcript.append(f"\n[tool {d.get('tool')}] {str(d.get('command') or d.get('args') or '')[:300]}\n")
-            elif d.get("type") == "tool_output":
-                transcript.append(f"[output] {str(d.get('output') or '')[:600]}\n")
-                approval = d.get("ask_user")
-                if (
-                    isinstance(approval, dict)
-                    and approval.get("kind") == "tool_approval"
-                ):
-                    approval_required = approval
-                    break
-            elif d.get("type") == "agent_step":
+        )) as stream:
+            async for chunk in stream:
+                # Streams can include an SSE event line before the data line,
+                # notably `event: error`. Do not silently discard those failures.
+                payload = next((line[6:] for line in chunk.splitlines() if line.startswith("data: ")), None)
+                if payload is None or payload == "[DONE]":
+                    continue
                 try:
-                    stats["turns"] = max(stats["turns"], int(d.get("round") or 0))
-                except (TypeError, ValueError):
-                    pass
-                transcript.append(f"\n--- round {d.get('round')} ---\n")
-            elif d.get("type") == "metrics":
-                data = d.get("data") or {}
-                try:
-                    stats["turns"] = max(stats["turns"], int(data.get("agent_rounds") or 0))
-                except (TypeError, ValueError):
-                    pass
-                try:
-                    stats["tool_calls"] = max(stats["tool_calls"], int(data.get("tool_calls") or 0))
-                except (TypeError, ValueError):
-                    pass
+                    d = _json.loads(payload)
+                except Exception:
+                    continue
+                if d.get("error") or d.get("type") == "error":
+                    raise SkillAuditUnavailable(str(d.get("error") or d.get("message") or "Audit stream failed"))
+                if d.get("delta"):
+                    transcript.append(d["delta"])
+                elif d.get("type") == "tool_start":
+                    stats["tool_calls"] += 1
+                    transcript.append(f"\n[tool {d.get('tool')}] {str(d.get('command') or d.get('args') or '')[:300]}\n")
+                elif d.get("type") == "tool_output":
+                    transcript.append(f"[output] {str(d.get('output') or '')[:600]}\n")
+                    approval = d.get("ask_user")
+                    if (
+                        isinstance(approval, dict)
+                        and approval.get("kind") == "tool_approval"
+                    ):
+                        approval_required = approval
+                        break
+                elif d.get("type") == "agent_step":
+                    try:
+                        stats["turns"] = max(stats["turns"], int(d.get("round") or 0))
+                    except (TypeError, ValueError):
+                        pass
+                    transcript.append(f"\n--- round {d.get('round')} ---\n")
+                elif d.get("type") == "metrics":
+                    data = d.get("data") or {}
+                    try:
+                        stats["turns"] = max(stats["turns"], int(data.get("agent_rounds") or 0))
+                    except (TypeError, ValueError):
+                        pass
+                    try:
+                        stats["tool_calls"] = max(stats["tool_calls"], int(data.get("tool_calls") or 0))
+                    except (TypeError, ValueError):
+                        pass
     except SkillAuditUnavailable:
         raise
     except Exception as e:
diff --git a/src/tool_approvals.py b/src/tool_approvals.py
index bf8b7a66a..d68adb82a 100644
--- a/src/tool_approvals.py
+++ b/src/tool_approvals.py
@@ -415,9 +415,24 @@ class ToolApprovalStore:
             backend_operation = BoundBackendOperation(backend,
                 request_authority.request_id if request_authority is not None else "",
                 _normalized_owner(owner), str(session_id or ""), operation.transport_tool, operation.input)
-            from src.agent_runtime.process_resources import needs_process_binding, resolve_process_operation
+            from src.agent_runtime.process_resources import (
+                LAUNCH_TOOLS, needs_process_binding, resolve_process_operation, seal_launch_scope)
             if request_authority is not None and needs_process_binding(operation, backend):
-                process_operation = resolve_process_operation(request_authority, operation, backend)
+                process_authority = request_authority
+                if (not request_authority.inherited and operation.tool in LAUNCH_TOOLS
+                        and not any(s.backend == backend for s in request_authority.launch_scopes)):
+                    # Launch scopes exist only for granted bash/python, but the
+                    # approval exists to admit an ungranted one. Seal its launch
+                    # boundary now, in the request's workspace, the same way an
+                    # ungranted filesystem action gets its root below (#6635).
+                    roots = request_authority.resource_roots
+                    if not roots and workspace:
+                        roots = (FilesystemRoot.seal(workspace, owner=_normalized_owner(owner)),)
+                    if len(roots) == 1:
+                        from dataclasses import replace
+                        process_authority = replace(
+                            request_authority, launch_scopes=(seal_launch_scope(backend, roots[0]),))
+                process_operation = resolve_process_operation(process_authority, operation, backend)
             from src.browser_identity import native_browser, resolve_browser_operation
             if native_browser(operation, backend):
                 if request_authority is None:
diff --git a/tests/test_exact_approval_ungranted_process.py b/tests/test_exact_approval_ungranted_process.py
new file mode 100644
index 000000000..fffb5c011
--- /dev/null
+++ b/tests/test_exact_approval_ungranted_process.py
@@ -0,0 +1,81 @@
+"""An exact approval must be able to run a bash/python call the request did not grant (#6635).
+
+`ToolApprovalStore.create()` sealed the process identity only from the request
+authority's launch scopes, and those exist only for granted bash/python
+backends. For an ungranted call the sealing failed silently, so the approval
+card was created with no process identity and the executor refused it:
+"Approved action has no sealed process/job identity".
+"""
+import pytest
+
+from src.agent_runtime.authority import OperationGrant, RequestAuthority
+from src.tool_approvals import ToolApprovalStore
+from src.tool_capabilities import ToolRunSecurityContext, capabilities_for_action
+from src.tool_types import ToolBlock
+
+
+@pytest.fixture(autouse=True)
+def admin_owner(monkeypatch):
+    from src import tool_execution as execution
+    monkeypatch.setattr(execution, "_owner_is_admin", lambda owner: True)
+
+
+def _authority(workspace, *tools):
+    return RequestAuthority("approval-request", "alice", "s", str(workspace),
+                            tuple(OperationGrant(t) for t in tools))
+
+
+def _approve(authority, tool, content, store=None):
+    store = store or ToolApprovalStore()
+    pending = store.create(owner=authority.owner, session_id=authority.session_id, origin_run_id="run",
+        tool_name=tool, content=content, workspace=authority.workspace, request_authority=authority,
+        external_untrusted_context_seen=True, capabilities=capabilities_for_action(tool, content))
+    return pending, store.consume(pending.approval_id, decision="approve",
+                                  owner=authority.owner, session_id=authority.session_id)
+
+
+async def _dispatch(authority, tool, content, exact):
+    from src import tool_execution as execution
+    return await execution.execute_tool_block(ToolBlock(tool, content), owner=authority.owner,
+        session_id=authority.session_id, workspace=authority.workspace,
+        request_authority=authority, exact_approval=exact,
+        security_context=ToolRunSecurityContext(external_untrusted_context_seen=True))
+
+
+def test_approval_for_ungranted_bash_seals_a_launch_in_the_workspace(tmp_path):
+    authority = _authority(tmp_path, "get_workspace")
+    assert authority.launch_scopes == ()
+
+    pending, _ = _approve(authority, "bash", "printf approved-run")
+
+    bound = pending.process_operation
+    assert bound is not None
+    assert bound.launch is not None
+    assert bound.launch.scope.root.path == str(tmp_path.resolve())
+    assert bound.request_id == authority.request_id
+
+
+async def test_approved_ungranted_bash_runs_once(tmp_path):
+    authority = _authority(tmp_path, "get_workspace")
+    _, exact = _approve(authority, "bash", "printf approved-run")
+
+    description, result = await _dispatch(authority, "bash", "printf approved-run", exact)
+
+    assert result.get("failure_kind") != "resource_identity_denied", result
+    assert result.get("exit_code") == 0, result
+    assert "approved-run" in str(result.get("output", "")) + description
+
+    _, replay = await _dispatch(authority, "bash", "printf approved-run", exact)
+    assert replay.get("blocked") is True, replay
+
+
+async def test_inherited_authority_still_cannot_widen_to_ungranted_bash(tmp_path):
+    # A child (inherited) authority may only narrow its parent, so an approval
+    # it proposes for an ungranted launch must stay unsealed and unrunnable.
+    from dataclasses import replace
+    child = replace(_authority(tmp_path, "get_workspace"), inherited=True)
+    pending, exact = _approve(child, "bash", "printf approved-run")
+
+    assert pending.process_operation is None
+    _, result = await _dispatch(child, "bash", "printf approved-run", exact)
+    assert result.get("blocked") is True
diff --git a/tests/test_skill_test_stream_closed.py b/tests/test_skill_test_stream_closed.py
new file mode 100644
index 000000000..bdaa816c2
--- /dev/null
+++ b/tests/test_skill_test_stream_closed.py
@@ -0,0 +1,61 @@
+"""Skill test runs must close the agent stream in their own task (#6635).
+
+When the agent asks for an exact approval, the skill test pauses by leaving its
+`async for` over `stream_agent_loop`. The abandoned generator was later closed
+by asyncio's finalizer in another task, so the ContextVar resets in its cleanup
+(request_authority, turn_contract, ...) raised "Token ... was created in a
+different Context". The stream must be closed before the run returns.
+"""
+import asyncio
+import json
+from contextvars import ContextVar
+
+import routes.skills_routes as skills_routes
+
+_BOUND: ContextVar = ContextVar("skill_test_bound", default=None)
+
+_APPROVAL = {"kind": "tool_approval", "approval_id": "approval-1", "question": "Allow?"}
+
+
+def _fake_stream(events, closed):
+    async def stream_agent_loop(*args, **kwargs):
+        token = _BOUND.set("bound")
+        try:
+            for event in events:
+                yield "data: " + json.dumps(event) + "\n\n"
+            yield "data: [DONE]\n\n"
+        finally:
+            # Raises ValueError when run from another task's context.
+            _BOUND.reset(token)
+            closed.append(asyncio.current_task())
+    return stream_agent_loop
+
+
+async def test_paused_skill_test_closes_the_agent_stream_in_its_own_task(monkeypatch):
+    closed = []
+    events = [{"type": "tool_output", "output": "blocked", "ask_user": _APPROVAL},
+              {"delta": "never reached"}]
+    monkeypatch.setattr("src.agent_loop.stream_agent_loop", _fake_stream(events, closed))
+    key = ("alice", "demo")
+    skills_routes._skill_test_jobs[key] = {"log": [], "status": "running"}
+    try:
+        await skills_routes._run_skill_test_job(
+            key, "demo", "# skill", "task", "http://model.test", "m", {}, "alice")
+        job = skills_routes._skill_test_jobs[key]
+        assert job["status"] == "awaiting_approval"
+        assert closed == [asyncio.current_task()]
+    finally:
+        skills_routes._skill_test_jobs.pop(key, None)
+
+
+async def test_audit_arm_closes_the_agent_stream_when_it_stops_for_approval(monkeypatch):
+    closed = []
+    events = [{"type": "tool_output", "output": "blocked", "ask_user": _APPROVAL},
+              {"delta": "never reached"}]
+    monkeypatch.setattr("src.agent_loop.stream_agent_loop", _fake_stream(events, closed))
+
+    _, _, approval = await skills_routes._run_skill_audit_arm(
+        [{"role": "user", "content": "task"}], "http://model.test", "m", {}, "alice")
+
+    assert approval == _APPROVAL
+    assert closed == [asyncio.current_task()]