fix(approvals): let "approve once" run ungranted bash/python, close skill-test streams (#6635) (#6683)

ToolApprovalStore.create() sealed a process identity only from the request
authority's launch scopes, which exist only for granted bash/python. For an
ungranted call the sealing failed silently, so the approval was created
without a process identity and the executor always refused it ("Approved
action has no sealed process/job identity"). When the authority is not
inherited and has no launch scope for the backend, seal one in the
request's single workspace root at approval time, as filesystem approvals
already do. Inherited authorities still cannot widen.

The skill test run and audit arm left their `async for` over
stream_agent_loop early when an approval was needed, without closing the
generator. asyncio closed it later from another task, so its ContextVar
resets raised "Token ... was created in a different Context". Close the
stream with aclosing, like the scheduler and teacher escalation do.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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

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()]