From 409062093ccbb0e1f1cbbc90c8d55fa92752b444 Mon Sep 17 00:00:00 2001 From: Y1fe1Zh0u Date: Tue, 28 Jul 2026 19:44:52 +0800 Subject: [PATCH 1/2] Enforce L3 approval before Runtime workspace deletion Durable Runtime executed typed delete tools outside the legacy autonomy boundary, allowing Agent and Group Workspace files to be removed even when delete_files was L3. Route Runtime deletes through the existing autonomy service, preserve Group scope in approval details, and execute an approved Group deletion against the original workspace. Constraint: Reuse the existing approval_requests schema and resolution endpoints; no migration. Rejected: Call the legacy execute_tool wrapper from Durable Runtime | it returns untyped results and loses Group Workspace scope. Confidence: high Scope-risk: moderate Reversibility: clean Directive: Keep approved Group Workspace actions bound to their captured tenant, group, participant, session, and path scope. Tested: Backend pytest 2126 passed; targeted Runtime and Group suite 86 passed; Ruff; compileall; git diff --check. Not-tested: Live browser approval flow before try deployment. --- .../services/agent_runtime/tool_execution.py | 2 + .../agent_runtime/tool_step_service.py | 180 ++++++++++++++ backend/app/services/autonomy_service.py | 44 ++++ .../test_agent_runtime_tool_step_service.py | 234 ++++++++++++++++++ .../test_autonomy_service_runtime_delete.py | 93 +++++++ 5 files changed, 553 insertions(+) create mode 100644 backend/tests/test_autonomy_service_runtime_delete.py diff --git a/backend/app/services/agent_runtime/tool_execution.py b/backend/app/services/agent_runtime/tool_execution.py index 6f8d30235..4020a708e 100644 --- a/backend/app/services/agent_runtime/tool_execution.py +++ b/backend/app/services/agent_runtime/tool_execution.py @@ -60,6 +60,8 @@ "size", "archive_status", "archive_error_code", + "approval_id", + "autonomy_level", "message_id", "accepted_recipients", "refused_recipients", diff --git a/backend/app/services/agent_runtime/tool_step_service.py b/backend/app/services/agent_runtime/tool_step_service.py index 68e2583fe..495a70365 100644 --- a/backend/app/services/agent_runtime/tool_step_service.py +++ b/backend/app/services/agent_runtime/tool_step_service.py @@ -23,6 +23,7 @@ ) from app.services.agent_runtime.command_worker import RuntimeSessionFactory from app.services.agent_runtime.group_runtime_tools import ( + GROUP_DELETE_WORKSPACE_FILE, GROUP_READ_TOOL_NAMES, GROUP_SCOPED_WORKSPACE_TOOL_NAMES, GROUP_TOOL_NAMES, @@ -74,6 +75,7 @@ ToolResultReconciler, ToolResultStore, ) +from app.services.autonomy_service import autonomy_service from app.services.agent_tools import ( agentbay_run_scope_id, execute_builtin_tool_outcome, @@ -479,6 +481,89 @@ def _is_group_workspace_mutation_call( ) +def _delete_autonomy_details( + state: RuntimeGraphState, + context: RuntimeContext, + agent: Agent, + tool_name: str, + arguments: Mapping[str, object], +) -> dict | None: + if tool_name not in {"delete_file", GROUP_DELETE_WORKSPACE_FILE}: + return None + actor_user_id = context.actor_user_id or str(agent.creator_id) + runtime_scope = { + "tenant_id": context.tenant_id, + "run_id": context.run_id, + "session_id": context.session_id, + "workspace_scope": "agent", + } + is_group_delete = tool_name == GROUP_DELETE_WORKSPACE_FILE or ( + tool_name == "delete_file" + and _is_group_scoped_workspace_call( + state, + tool_name, + arguments, + ) + ) + if is_group_delete: + initial_input = state["snapshots"].initial_input + group_id = initial_input.get("group_id") + participant_id = initial_input.get("target_participant_id") + group_context = initial_input.get("group_context") + context_agent = ( + group_context.get("agent") + if isinstance(group_context, Mapping) + else None + ) + context_agent_id = ( + context_agent.get("agent_id") + if isinstance(context_agent, Mapping) + else None + ) + try: + uuid.UUID(str(group_id)) + uuid.UUID(str(participant_id)) + uuid.UUID(context.session_id or "") + except (TypeError, ValueError) as exc: + raise ToolExecutionError( + "group_tool_scope_invalid", + "Group delete approval scope is incomplete", + ) from exc + if context_agent_id != str(agent.id): + raise ToolExecutionError( + "group_tool_scope_invalid", + "Group delete approval Agent does not match the executing Agent", + ) + path = arguments.get("path") + if not isinstance(path, str) or not path.strip(): + raise ToolExecutionError( + "invalid_tool_call", + "delete_file requires a non-empty path", + ) + workspace_path = path.replace("\\", "/").strip() + if workspace_path in {"", ".", "workspace", "workspace/"}: + raise ToolExecutionError( + "invalid_tool_call", + "delete_file must identify an item inside Group Workspace", + ) + if workspace_path.startswith("workspace/"): + workspace_path = workspace_path.removeprefix("workspace/") + runtime_scope.update( + { + "workspace_scope": "group", + "group_id": str(group_id), + "actor_participant_id": str(participant_id), + "workspace_path": workspace_path, + } + ) + return { + "tool": tool_name, + "args": dict(arguments), + "requested_by": actor_user_id, + "runtime_scope": runtime_scope, + } + + def _heartbeat_blocked_summary( agent: Agent, tool_name: str, @@ -1073,6 +1158,76 @@ async def _mark_policy_blocked( ), ) + async def _delete_autonomy_outcome( + self, + *, + state: RuntimeGraphState, + context: RuntimeContext, + agent: Agent, + tool_name: str, + arguments: Mapping[str, object], + ) -> ToolExecutionOutcome | None: + details = _delete_autonomy_details( + state, + context, + agent, + tool_name, + arguments, + ) + if details is None: + return None + try: + async with self._session_factory() as db: + async with db.begin(): + decision = await autonomy_service.check_and_enforce( + db, + agent, + "delete_files", + details, + ) + except Exception as exc: + return ToolExecutionOutcome( + status="failed", + result_summary=( + "Workspace deletion was blocked because the autonomy " + "policy check could not be completed." + ), + result_ref=None, + error_code="tool_autonomy_check_failed", + retryable=False, + metadata={"error_class": type(exc).__name__}, + ) + if decision.get("allowed"): + return None + level = str(decision.get("level") or "unknown") + approval_id = decision.get("approval_id") + if level == "L3" and isinstance(approval_id, str) and approval_id: + return ToolExecutionOutcome( + status="failed", + result_summary=( + "Workspace deletion requires approval and was not " + f"executed. Approval ID: {approval_id}" + ), + result_ref=None, + error_code="tool_approval_required", + retryable=False, + metadata={ + "approval_id": approval_id, + "autonomy_level": level, + }, + ) + return ToolExecutionOutcome( + status="failed", + result_summary=str( + decision.get("message") + or "Workspace deletion was denied by the autonomy policy." + ), + result_ref=None, + error_code="tool_autonomy_denied", + retryable=False, + metadata={"autonomy_level": level}, + ) + async def execute_pending( self, state: RuntimeGraphState, @@ -1447,6 +1602,31 @@ async def execute_pending( pending_tool_calls=tool_calls[index:], ) + autonomy_outcome = await self._delete_autonomy_outcome( + state=state, + context=context, + agent=agent, + tool_name=tool_name, + arguments=arguments, + ) + if autonomy_outcome is not None: + outcome = await self._settle_outcome( + tenant_id=tenant_id, + reservation=reservation, + lease_owner=lease_owner, + policy=policy, + outcome=autonomy_outcome, + ) + messages.append( + _result_message( + run_id=run_id, + call_id=call_id, + tool_name=tool_name, + outcome=outcome, + ) + ) + continue + canonical_cross_space_action = builtin_cross_space_action(tool_name) if ( _is_group_agent_run(state) diff --git a/backend/app/services/autonomy_service.py b/backend/app/services/autonomy_service.py index 97f311ac8..bb8f8ad57 100644 --- a/backend/app/services/autonomy_service.py +++ b/backend/app/services/autonomy_service.py @@ -14,6 +14,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from app.database import async_session from app.models.agent import Agent from app.models.audit import ApprovalRequest, AuditLog from app.models.channel_config import ChannelConfig @@ -194,6 +195,49 @@ async def _execute_approved_action( else: arguments = args_raw + runtime_scope = details.get("runtime_scope") + if ( + action_type == "delete_files" + and tool_name in {"delete_file", "group_delete_workspace_file"} + and isinstance(runtime_scope, dict) + and runtime_scope.get("workspace_scope") == "group" + ): + from app.services import group_file_service + + tenant_id = uuid.UUID(str(runtime_scope["tenant_id"])) + group_id = uuid.UUID(str(runtime_scope["group_id"])) + participant_id = uuid.UUID( + str(runtime_scope["actor_participant_id"]) + ) + session_id_raw = runtime_scope.get("session_id") + session_id = ( + uuid.UUID(str(session_id_raw)) + if session_id_raw + else None + ) + path = runtime_scope.get("workspace_path") + if not isinstance(path, str) or not path.strip(): + raise ValueError( + "Approved Group Workspace delete is missing its path" + ) + expected_version_token = arguments.get( + "expected_version_token" + ) + if not isinstance(expected_version_token, str): + expected_version_token = None + async with async_session() as action_db: + await group_file_service.delete_workspace_file( + action_db, + tenant_id=tenant_id, + group_id=group_id, + actor_participant_id=participant_id, + path=path, + expected_version_token=expected_version_token, + session_id=session_id, + ) + await action_db.commit() + return f"✅ Deleted {path} from Group Workspace" + # Import and call the tool's direct executor (no autonomy re-check) from app.services.agent_tools import _execute_tool_direct result = await _execute_tool_direct(tool_name, arguments, agent_id) diff --git a/backend/tests/test_agent_runtime_tool_step_service.py b/backend/tests/test_agent_runtime_tool_step_service.py index 6b40d57cc..d793b486e 100644 --- a/backend/tests/test_agent_runtime_tool_step_service.py +++ b/backend/tests/test_agent_runtime_tool_step_service.py @@ -1489,6 +1489,228 @@ async def execute_scoped_workspace_tool( assert group_calls[0][5]["lease_owner"] +@pytest.mark.asyncio +async def test_l3_private_workspace_delete_requires_approval_before_execution( + monkeypatch, +) -> None: + tenant_id = uuid.uuid4() + agent = _agent(tenant_id) + agent.autonomy_policy = {"delete_files": "L3"} + call = { + "id": "call-private-delete", + "type": "function", + "function": { + "name": "delete_file", + "arguments": '{"path":"workspace/remove-me.md"}', + }, + } + state = _state(tenant_id, agent, (call,)) + context = _context(state) + execution = _execution( + tenant_id, + uuid.UUID(context.run_id), + "call-private-delete", + "delete_file", + ) + approval_id = uuid.uuid4() + approval_calls: list[dict] = [] + settled_metadata: list[dict] = [] + + async def tools(agent_id): + assert agent_id == agent.id + return [{"type": "function", "function": {"name": "delete_file"}}] + + async def reserve(db, **kwargs): + del db, kwargs + return _reservation(execution) + + async def check_and_enforce(db, agent_arg, action_type, details): + del db + approval_calls.append( + { + "agent": agent_arg, + "action_type": action_type, + "details": details, + } + ) + return { + "allowed": False, + "level": "L3", + "approval_id": str(approval_id), + "message": "Approval requested from creator", + } + + async def mark_failed(db, **kwargs): + del db + settled_metadata.append(kwargs["metadata"]) + execution.status = "failed" + execution.result_summary = kwargs["result_summary"] + execution.error_code = kwargs["error_code"] + execution.result_metadata = kwargs["metadata"] + return execution + + async def forbidden_executor(*args, **kwargs): + raise AssertionError( + f"L3 delete reached the workspace executor: {args}, {kwargs}" + ) + + monkeypatch.setattr(tool_step_service, "reserve_tool_execution", reserve) + monkeypatch.setattr(tool_step_service, "mark_tool_execution_failed", mark_failed) + monkeypatch.setattr( + tool_step_service.autonomy_service, + "check_and_enforce", + check_and_enforce, + ) + service = tool_step_service.RuntimeToolStepService( + session_factory=_session_factory(agent), + cancel_source=_CancelSource(None), + tool_provider=tools, + tool_executor=forbidden_executor, + ) + + result = await service.execute_pending(state, context, (call,)) + + assert result.error is None + assert result.waiting_request is None + assert result.messages[0]["execution_status"] == "failed" + assert result.messages[0]["error_code"] == "tool_approval_required" + assert settled_metadata[0]["approval_id"] == str(approval_id) + assert approval_calls == [ + { + "agent": agent, + "action_type": "delete_files", + "details": { + "tool": "delete_file", + "args": {"path": "workspace/remove-me.md"}, + "requested_by": context.actor_user_id, + "runtime_scope": { + "tenant_id": context.tenant_id, + "run_id": context.run_id, + "session_id": context.session_id, + "workspace_scope": "agent", + }, + }, + } + ] + + +@pytest.mark.asyncio +async def test_l3_group_workspace_delete_preserves_group_scope_for_approval( + monkeypatch, +) -> None: + tenant_id = uuid.uuid4() + group_id = uuid.uuid4() + participant_id = uuid.uuid4() + agent = _agent(tenant_id) + agent.autonomy_policy = {"delete_files": "L3"} + call = { + "id": "call-group-delete", + "type": "function", + "function": { + "name": "delete_file", + "arguments": '{"path":"workspace/remove-me.md"}', + }, + } + state = _state(tenant_id, agent, (call,)) + state["snapshots"] = RunInputSnapshots( + session_context={"version": 0}, + session_context_version=0, + recent_session_messages=(), + related_run_summaries=(), + initial_input={ + "group_id": str(group_id), + "target_participant_id": str(participant_id), + "group_context": {"agent": {"agent_id": str(agent.id)}}, + }, + ) + context = _context(state) + execution = _execution( + tenant_id, + uuid.UUID(context.run_id), + "call-group-delete", + "delete_file", + ) + approval_id = uuid.uuid4() + captured_details: list[dict] = [] + + async def tools(agent_id): + assert agent_id == agent.id + return [{"type": "function", "function": {"name": "delete_file"}}] + + async def reserve(db, **kwargs): + del db + assert kwargs["arguments"]["workspace_scope"] == "group" + return _reservation(execution) + + async def check_and_enforce(db, agent_arg, action_type, details): + del db + assert agent_arg is agent + assert action_type == "delete_files" + captured_details.append(details) + return { + "allowed": False, + "level": "L3", + "approval_id": str(approval_id), + "message": "Approval requested from creator", + } + + async def mark_failed(db, **kwargs): + del db + execution.status = "failed" + execution.result_summary = kwargs["result_summary"] + execution.error_code = kwargs["error_code"] + execution.result_metadata = kwargs["metadata"] + return execution + + class _ForbiddenGroupToolService: + async def execute_scoped_workspace_tool(self, *args, **kwargs): + raise AssertionError( + f"L3 delete reached Group Workspace: {args}, {kwargs}" + ) + + async def forbidden_executor(*args, **kwargs): + raise AssertionError(f"Group delete reached Agent Workspace: {args}, {kwargs}") + + monkeypatch.setattr(tool_step_service, "reserve_tool_execution", reserve) + monkeypatch.setattr(tool_step_service, "mark_tool_execution_failed", mark_failed) + monkeypatch.setattr( + tool_step_service.autonomy_service, + "check_and_enforce", + check_and_enforce, + ) + service = tool_step_service.RuntimeToolStepService( + session_factory=_session_factory(agent), + cancel_source=_CancelSource(None), + tool_provider=tools, + tool_executor=forbidden_executor, + group_tool_service=_ForbiddenGroupToolService(), # type: ignore[arg-type] + ) + + result = await service.execute_pending(state, context, (call,)) + + assert result.error is None + assert result.messages[0]["error_code"] == "tool_approval_required" + assert captured_details == [ + { + "tool": "delete_file", + "args": { + "path": "workspace/remove-me.md", + "workspace_scope": "group", + }, + "requested_by": context.actor_user_id, + "runtime_scope": { + "tenant_id": context.tenant_id, + "run_id": context.run_id, + "session_id": context.session_id, + "workspace_scope": "group", + "group_id": str(group_id), + "actor_participant_id": str(participant_id), + "workspace_path": "remove-me.md", + }, + } + ] + + @pytest.mark.asyncio async def test_group_workspace_write_uses_ledger_id_and_reconciles_without_reexecution( monkeypatch, @@ -1786,6 +2008,13 @@ async def reserve(db, **_kwargs): async def fail_settle(*_args, **_kwargs): raise OSError("database unavailable after storage success") + async def allow_delete(*_args, **_kwargs): + return { + "allowed": True, + "level": "L2", + "message": "Executed and creator notified", + } + class _GroupToolService: async def execute(self, *_args, operation_id, **_kwargs): return ToolExecutionOutcome( @@ -1800,6 +2029,11 @@ async def execute(self, *_args, operation_id, **_kwargs): ) monkeypatch.setattr(tool_step_service, "reserve_tool_execution", reserve) + monkeypatch.setattr( + tool_step_service.autonomy_service, + "check_and_enforce", + allow_delete, + ) monkeypatch.setattr( tool_step_service, "mark_tool_execution_succeeded", diff --git a/backend/tests/test_autonomy_service_runtime_delete.py b/backend/tests/test_autonomy_service_runtime_delete.py new file mode 100644 index 000000000..50690b209 --- /dev/null +++ b/backend/tests/test_autonomy_service_runtime_delete.py @@ -0,0 +1,93 @@ +"""Runtime-specific approval execution tests.""" + +from contextlib import asynccontextmanager +import uuid + +import pytest + +from app.services import autonomy_service as autonomy_module +from app.services import group_file_service + + +@pytest.mark.asyncio +async def test_approved_group_workspace_delete_keeps_original_scope( + monkeypatch, +) -> None: + tenant_id = uuid.uuid4() + group_id = uuid.uuid4() + participant_id = uuid.uuid4() + session_id = uuid.uuid4() + agent_id = uuid.uuid4() + delete_calls: list[dict] = [] + + class _DB: + def __init__(self) -> None: + self.committed = False + + async def commit(self) -> None: + self.committed = True + + db = _DB() + + @asynccontextmanager + async def session_factory(): + yield db + + async def delete_workspace_file(db_arg, **kwargs): + assert db_arg is db + delete_calls.append(kwargs) + + async def forbidden_direct_executor(*args, **kwargs): + raise AssertionError( + f"Group approval used Agent Workspace executor: {args}, {kwargs}" + ) + + monkeypatch.setattr( + autonomy_module, + "async_session", + session_factory, + raising=False, + ) + monkeypatch.setattr( + group_file_service, + "delete_workspace_file", + delete_workspace_file, + ) + monkeypatch.setattr( + "app.services.agent_tools._execute_tool_direct", + forbidden_direct_executor, + ) + + result = await autonomy_module.AutonomyService()._execute_approved_action( + agent_id, + "delete_files", + { + "tool": "delete_file", + "args": { + "path": "workspace/remove-me.md", + "workspace_scope": "group", + }, + "runtime_scope": { + "tenant_id": str(tenant_id), + "run_id": str(uuid.uuid4()), + "session_id": str(session_id), + "workspace_scope": "group", + "group_id": str(group_id), + "actor_participant_id": str(participant_id), + "workspace_path": "remove-me.md", + }, + }, + ) + + assert result == "✅ Deleted remove-me.md from Group Workspace" + assert delete_calls == [ + { + "tenant_id": tenant_id, + "group_id": group_id, + "actor_participant_id": participant_id, + "path": "remove-me.md", + "expected_version_token": None, + "session_id": session_id, + } + ] + assert db.committed is True From b8aa04981250444b2169d80f976099e05e9f66ae Mon Sep 17 00:00:00 2001 From: Y1fe1Zh0u Date: Tue, 28 Jul 2026 21:09:08 +0800 Subject: [PATCH 2/2] Resume the originating Run after deletion approval Runtime-scoped L3 deletion approvals now preserve the pending tool call and place the originating Run in waiting_user. Resolving the approval writes an idempotent resume command for that same Run; approval replays the original tool call while rejection returns a typed tool result without executing the delete. Legacy approvals retain their direct-execution fallback. Constraint: No schema migration or new dependency is allowed for the v1.11.3 regression fix. Rejected: Execute approved Runtime deletes inside the approval endpoint | ends the original Agent Run before it can report the outcome. Confidence: high Scope-risk: moderate Reversibility: clean Directive: Runtime-scoped approvals must resume their stored run_id and correlation_id; do not add a second out-of-band executor. Tested: Backend pytest 2130 passed; targeted approval and Runtime tool-step tests; Ruff; compileall; git diff --check. Not-tested: Browser E2E on try will be performed after deployment. --- .../agent_runtime/tool_step_service.py | 113 ++++++--- backend/app/services/autonomy_service.py | 140 ++++++++++- .../test_agent_runtime_tool_step_service.py | 218 ++++++++++++++--- .../test_autonomy_service_runtime_delete.py | 219 ++++++++++++++++++ 4 files changed, 615 insertions(+), 75 deletions(-) diff --git a/backend/app/services/agent_runtime/tool_step_service.py b/backend/app/services/agent_runtime/tool_step_service.py index 495a70365..75e64ec10 100644 --- a/backend/app/services/agent_runtime/tool_step_service.py +++ b/backend/app/services/agent_runtime/tool_step_service.py @@ -485,6 +485,7 @@ def _delete_autonomy_details( state: RuntimeGraphState, context: RuntimeContext, agent: Agent, + call_id: str, tool_name: str, arguments: Mapping[str, object], ) -> dict | None: @@ -496,6 +497,7 @@ def _delete_autonomy_details( "run_id": context.run_id, "session_id": context.session_id, "workspace_scope": "agent", + "tool_call_id": call_id, } is_group_delete = tool_name == GROUP_DELETE_WORKSPACE_FILE or ( tool_name == "delete_file" @@ -1158,24 +1160,26 @@ async def _mark_policy_blocked( ), ) - async def _delete_autonomy_outcome( + async def _delete_autonomy_gate( self, *, state: RuntimeGraphState, context: RuntimeContext, agent: Agent, + call_id: str, tool_name: str, arguments: Mapping[str, object], - ) -> ToolExecutionOutcome | None: + ) -> tuple[ToolExecutionOutcome | None, JsonObject | None]: details = _delete_autonomy_details( state, context, agent, + call_id, tool_name, arguments, ) if details is None: - return None + return None, None try: async with self._session_factory() as db: async with db.begin(): @@ -1186,46 +1190,78 @@ async def _delete_autonomy_outcome( details, ) except Exception as exc: - return ToolExecutionOutcome( - status="failed", - result_summary=( - "Workspace deletion was blocked because the autonomy " - "policy check could not be completed." + return ( + ToolExecutionOutcome( + status="failed", + result_summary=( + "Workspace deletion was blocked because the autonomy " + "policy check could not be completed." + ), + result_ref=None, + error_code="tool_autonomy_check_failed", + retryable=False, + metadata={"error_class": type(exc).__name__}, ), - result_ref=None, - error_code="tool_autonomy_check_failed", - retryable=False, - metadata={"error_class": type(exc).__name__}, + None, ) if decision.get("allowed"): - return None + return None, None level = str(decision.get("level") or "unknown") approval_id = decision.get("approval_id") - if level == "L3" and isinstance(approval_id, str) and approval_id: + approval_status = decision.get("approval_status") + correlation_id = decision.get("correlation_id") + if ( + level == "L3" + and approval_status == "pending" + and isinstance(approval_id, str) + and approval_id + and isinstance(correlation_id, str) + and correlation_id + ): + return None, { + "waiting_type": "user", + "correlation_id": correlation_id, + "reason": "tool_approval_required", + "question": ( + "Workspace deletion requires approval. " + f"Approval ID: {approval_id}" + ), + "tool_call_id": call_id, + "approval_id": approval_id, + } + if ( + level == "L3" + and approval_status == "rejected" + and isinstance(approval_id, str) + and approval_id + ): return ToolExecutionOutcome( status="failed", result_summary=( - "Workspace deletion requires approval and was not " - f"executed. Approval ID: {approval_id}" + "Workspace deletion was rejected and was not executed. " + f"Approval ID: {approval_id}" ), result_ref=None, - error_code="tool_approval_required", + error_code="tool_approval_rejected", retryable=False, metadata={ "approval_id": approval_id, "autonomy_level": level, }, - ) - return ToolExecutionOutcome( - status="failed", - result_summary=str( - decision.get("message") - or "Workspace deletion was denied by the autonomy policy." + ), None + return ( + ToolExecutionOutcome( + status="failed", + result_summary=str( + decision.get("message") + or "Workspace deletion was denied by the autonomy policy." + ), + result_ref=None, + error_code="tool_autonomy_denied", + retryable=False, + metadata={"autonomy_level": level}, ), - result_ref=None, - error_code="tool_autonomy_denied", - retryable=False, - metadata={"autonomy_level": level}, + None, ) async def execute_pending( @@ -1343,6 +1379,22 @@ async def execute_pending( "tool_not_enabled", f"tool {tool_name!r} is not enabled for this Agent", ) + autonomy_outcome, approval_wait = ( + await self._delete_autonomy_gate( + state=state, + context=context, + agent=agent, + call_id=call_id, + tool_name=tool_name, + arguments=arguments, + ) + ) + if approval_wait is not None: + return ToolStepResult( + messages=tuple(messages), + waiting_request=approval_wait, + pending_tool_calls=tool_calls[index:], + ) policy = _policy(tool_name) lease_owner = _tool_execution_lease_owner( context.command_id, @@ -1602,13 +1654,6 @@ async def execute_pending( pending_tool_calls=tool_calls[index:], ) - autonomy_outcome = await self._delete_autonomy_outcome( - state=state, - context=context, - agent=agent, - tool_name=tool_name, - arguments=arguments, - ) if autonomy_outcome is not None: outcome = await self._settle_outcome( tenant_id=tenant_id, diff --git a/backend/app/services/autonomy_service.py b/backend/app/services/autonomy_service.py index bb8f8ad57..3b3769b98 100644 --- a/backend/app/services/autonomy_service.py +++ b/backend/app/services/autonomy_service.py @@ -25,6 +25,28 @@ class AutonomyService: """Enforce autonomy boundaries for agent operations.""" + @staticmethod + def _runtime_approval_identity( + action_type: str, + details: dict, + ) -> tuple[uuid.UUID, str] | None: + runtime_scope = details.get("runtime_scope") + if not isinstance(runtime_scope, dict): + return None + run_id_raw = runtime_scope.get("run_id") + tool_call_id = runtime_scope.get("tool_call_id") + if not isinstance(tool_call_id, str) or not tool_call_id: + return None + try: + run_id = uuid.UUID(str(run_id_raw)) + except (TypeError, ValueError): + return None + approval_id = uuid.uuid5( + run_id, + f"runtime-approval:{action_type}:{tool_call_id}", + ) + return approval_id, f"approval:{approval_id}" + async def check_and_enforce( self, db: AsyncSession, agent: Agent, action_type: str, details: dict ) -> dict: @@ -40,6 +62,43 @@ async def check_and_enforce( """ policy = agent.autonomy_policy or {} level = policy.get(action_type, "L2") # Default to L2 + runtime_identity = self._runtime_approval_identity(action_type, details) + + if runtime_identity is not None: + approval_id, correlation_id = runtime_identity + existing_result = await db.execute( + select(ApprovalRequest).where(ApprovalRequest.id == approval_id) + ) + existing = existing_result.scalar_one_or_none() + if existing is not None: + if ( + existing.agent_id != agent.id + or existing.action_type != action_type + ): + raise ValueError( + "Runtime approval identity does not match the requested action" + ) + if existing.status == "approved": + return { + "allowed": True, + "level": "L3", + "approval_id": str(existing.id), + "approval_status": "approved", + "correlation_id": correlation_id, + "message": "Approval granted", + } + return { + "allowed": False, + "level": "L3", + "approval_id": str(existing.id), + "approval_status": existing.status, + "correlation_id": correlation_id, + "message": ( + "Approval requested from creator" + if existing.status == "pending" + else "Approval rejected" + ), + } # Log the action regardless of level audit = AuditLog( @@ -70,10 +129,20 @@ async def check_and_enforce( elif level == "L3": # Create approval request and block + approval_details = details + approval_id = None + correlation_id = None + if runtime_identity is not None: + approval_id, correlation_id = runtime_identity + approval_details = dict(details) + runtime_scope = dict(approval_details["runtime_scope"]) + runtime_scope["approval_correlation_id"] = correlation_id + approval_details["runtime_scope"] = runtime_scope approval = ApprovalRequest( + id=approval_id, agent_id=agent.id, action_type=action_type, - details=details, + details=approval_details, ) db.add(approval) await db.flush() @@ -85,6 +154,8 @@ async def check_and_enforce( "allowed": False, "level": "L3", "approval_id": str(approval.id), + "approval_status": "pending", + "correlation_id": correlation_id, "message": "Approval requested from creator", } @@ -122,9 +193,42 @@ async def resolve_approval( details={"approval_id": str(approval.id), "action_type": approval.action_type}, )) - # Post-processing: execute the approved action + # Runtime-scoped approvals resume the exact waiting Run. Legacy + # approvals keep their historical direct-execution behavior. execution_result = None - if approval.status == "approved" and approval.details: + runtime_resume = self._runtime_resume_details(approval) + if runtime_resume is not None: + from app.services.agent_runtime.adapter import RuntimeCommandIntake + from app.services.agent_runtime.contracts import ResumeRunCommand + + await db.flush() + await RuntimeCommandIntake(db).resume_run( + ResumeRunCommand( + tenant_id=runtime_resume["tenant_id"], + run_id=runtime_resume["run_id"], + idempotency_key=( + f"approval:{approval.id}:{approval.status}" + ), + payload={ + "resume_type": "user_input", + "correlation_id": runtime_resume["correlation_id"], + "payload": { + "content": ( + "Workspace deletion approved. Continue the " + "pending tool call." + if approval.status == "approved" + else "Workspace deletion rejected. Do not " + "execute the pending tool call." + ), + "approval_id": str(approval.id), + "decision": approval.status, + }, + }, + actor_user_id=user.id, + ) + ) + execution_result = "Original Agent Run queued to resume" + elif approval.status == "approved" and approval.details: execution_result = await self._execute_approved_action( approval.agent_id, approval.action_type, approval.details ) @@ -168,6 +272,36 @@ async def resolve_approval( await db.flush() return approval + @staticmethod + def _runtime_resume_details( + approval: ApprovalRequest, + ) -> dict | None: + details = approval.details + if not isinstance(details, dict): + return None + runtime_scope = details.get("runtime_scope") + if not isinstance(runtime_scope, dict): + return None + correlation_id = runtime_scope.get("approval_correlation_id") + tool_call_id = runtime_scope.get("tool_call_id") + if ( + not isinstance(correlation_id, str) + or not correlation_id + or not isinstance(tool_call_id, str) + or not tool_call_id + ): + return None + try: + tenant_id = uuid.UUID(str(runtime_scope.get("tenant_id"))) + run_id = uuid.UUID(str(runtime_scope.get("run_id"))) + except (TypeError, ValueError): + return None + return { + "tenant_id": tenant_id, + "run_id": run_id, + "correlation_id": correlation_id, + } + async def _execute_approved_action( self, agent_id: uuid.UUID, action_type: str, details: dict ) -> str | None: diff --git a/backend/tests/test_agent_runtime_tool_step_service.py b/backend/tests/test_agent_runtime_tool_step_service.py index d793b486e..a8416377d 100644 --- a/backend/tests/test_agent_runtime_tool_step_service.py +++ b/backend/tests/test_agent_runtime_tool_step_service.py @@ -1513,17 +1513,22 @@ async def test_l3_private_workspace_delete_requires_approval_before_execution( "delete_file", ) approval_id = uuid.uuid4() + correlation_id = f"approval:{approval_id}" approval_calls: list[dict] = [] - settled_metadata: list[dict] = [] + reservation_calls: list[dict] = [] + execution_calls: list[dict] = [] async def tools(agent_id): assert agent_id == agent.id return [{"type": "function", "function": {"name": "delete_file"}}] async def reserve(db, **kwargs): - del db, kwargs + del db + reservation_calls.append(kwargs) return _reservation(execution) + approval_status = "pending" + async def check_and_enforce(db, agent_arg, action_type, details): del db approval_calls.append( @@ -1534,28 +1539,50 @@ async def check_and_enforce(db, agent_arg, action_type, details): } ) return { - "allowed": False, + "allowed": approval_status == "approved", "level": "L3", "approval_id": str(approval_id), + "approval_status": approval_status, + "correlation_id": correlation_id, "message": "Approval requested from creator", } - async def mark_failed(db, **kwargs): + async def mark_succeeded(db, **kwargs): del db - settled_metadata.append(kwargs["metadata"]) - execution.status = "failed" + execution.status = "succeeded" execution.result_summary = kwargs["result_summary"] - execution.error_code = kwargs["error_code"] - execution.result_metadata = kwargs["metadata"] return execution - async def forbidden_executor(*args, **kwargs): - raise AssertionError( - f"L3 delete reached the workspace executor: {args}, {kwargs}" + async def executor( + name, + arguments, + agent_id, + user_id, + session_id="", + on_output=None, + ): + execution_calls.append( + { + "name": name, + "arguments": arguments, + "agent_id": agent_id, + "user_id": user_id, + "session_id": session_id, + "on_output": on_output, + } + ) + return ToolExecutionOutcome( + status="succeeded", + result_summary="✅ Deleted workspace/remove-me.md", + result_ref=None, ) monkeypatch.setattr(tool_step_service, "reserve_tool_execution", reserve) - monkeypatch.setattr(tool_step_service, "mark_tool_execution_failed", mark_failed) + monkeypatch.setattr( + tool_step_service, + "mark_tool_execution_succeeded", + mark_succeeded, + ) monkeypatch.setattr( tool_step_service.autonomy_service, "check_and_enforce", @@ -1565,16 +1592,27 @@ async def forbidden_executor(*args, **kwargs): session_factory=_session_factory(agent), cancel_source=_CancelSource(None), tool_provider=tools, - tool_executor=forbidden_executor, + tool_executor=executor, ) - result = await service.execute_pending(state, context, (call,)) + waiting = await service.execute_pending(state, context, (call,)) - assert result.error is None - assert result.waiting_request is None - assert result.messages[0]["execution_status"] == "failed" - assert result.messages[0]["error_code"] == "tool_approval_required" - assert settled_metadata[0]["approval_id"] == str(approval_id) + assert waiting.error is None + assert waiting.messages == () + assert waiting.pending_tool_calls == (call,) + assert waiting.waiting_request == { + "waiting_type": "user", + "correlation_id": correlation_id, + "reason": "tool_approval_required", + "question": ( + "Workspace deletion requires approval. " + f"Approval ID: {approval_id}" + ), + "tool_call_id": "call-private-delete", + "approval_id": str(approval_id), + } + assert reservation_calls == [] + assert execution_calls == [] assert approval_calls == [ { "agent": agent, @@ -1588,11 +1626,114 @@ async def forbidden_executor(*args, **kwargs): "run_id": context.run_id, "session_id": context.session_id, "workspace_scope": "agent", + "tool_call_id": "call-private-delete", }, }, } ] + approval_status = "approved" + resumed = await service.execute_pending(state, context, (call,)) + + assert resumed.error is None + assert resumed.waiting_request is None + assert resumed.pending_tool_calls == () + assert resumed.messages[0]["execution_status"] == "succeeded" + assert resumed.messages[0]["content"] == "✅ Deleted workspace/remove-me.md" + assert len(reservation_calls) == 1 + assert len(execution_calls) == 1 + assert execution_calls[0]["name"] == "delete_file" + assert execution_calls[0]["arguments"] == { + "path": "workspace/remove-me.md" + } + + +@pytest.mark.asyncio +async def test_l3_private_workspace_delete_rejection_resumes_with_failed_result( + monkeypatch, +) -> None: + tenant_id = uuid.uuid4() + agent = _agent(tenant_id) + agent.autonomy_policy = {"delete_files": "L3"} + call = { + "id": "call-rejected-delete", + "type": "function", + "function": { + "name": "delete_file", + "arguments": '{"path":"workspace/keep-me.md"}', + }, + } + state = _state(tenant_id, agent, (call,)) + context = _context(state) + execution = _execution( + tenant_id, + uuid.UUID(context.run_id), + "call-rejected-delete", + "delete_file", + ) + approval_id = uuid.uuid4() + + async def tools(agent_id): + assert agent_id == agent.id + return [{"type": "function", "function": {"name": "delete_file"}}] + + async def reject_delete(*_args, **_kwargs): + return { + "allowed": False, + "level": "L3", + "approval_id": str(approval_id), + "approval_status": "rejected", + "correlation_id": f"approval:{approval_id}", + "message": "Approval rejected", + } + + async def reserve(db, **kwargs): + del db, kwargs + return _reservation(execution) + + async def mark_failed(db, **kwargs): + del db + execution.status = "failed" + execution.result_summary = kwargs["result_summary"] + execution.error_code = kwargs["error_code"] + execution.result_metadata = kwargs["metadata"] + return execution + + async def forbidden_executor(*args, **kwargs): + raise AssertionError( + f"Rejected delete reached the executor: {args}, {kwargs}" + ) + + monkeypatch.setattr( + tool_step_service.autonomy_service, + "check_and_enforce", + reject_delete, + ) + monkeypatch.setattr(tool_step_service, "reserve_tool_execution", reserve) + monkeypatch.setattr( + tool_step_service, + "mark_tool_execution_failed", + mark_failed, + ) + service = tool_step_service.RuntimeToolStepService( + session_factory=_session_factory(agent), + cancel_source=_CancelSource(None), + tool_provider=tools, + tool_executor=forbidden_executor, + ) + + result = await service.execute_pending(state, context, (call,)) + + assert result.error is None + assert result.waiting_request is None + assert result.pending_tool_calls == () + assert result.messages[0]["execution_status"] == "failed" + assert result.messages[0]["error_code"] == "tool_approval_rejected" + assert result.messages[0]["content"] == ( + "Workspace deletion was rejected and was not executed. " + f"Approval ID: {approval_id}" + ) + @pytest.mark.asyncio async def test_l3_group_workspace_delete_preserves_group_scope_for_approval( @@ -1624,13 +1765,8 @@ async def test_l3_group_workspace_delete_preserves_group_scope_for_approval( }, ) context = _context(state) - execution = _execution( - tenant_id, - uuid.UUID(context.run_id), - "call-group-delete", - "delete_file", - ) approval_id = uuid.uuid4() + correlation_id = f"approval:{approval_id}" captured_details: list[dict] = [] async def tools(agent_id): @@ -1638,9 +1774,9 @@ async def tools(agent_id): return [{"type": "function", "function": {"name": "delete_file"}}] async def reserve(db, **kwargs): - del db - assert kwargs["arguments"]["workspace_scope"] == "group" - return _reservation(execution) + raise AssertionError( + f"Waiting Group delete created a tool receipt: {db}, {kwargs}" + ) async def check_and_enforce(db, agent_arg, action_type, details): del db @@ -1651,17 +1787,11 @@ async def check_and_enforce(db, agent_arg, action_type, details): "allowed": False, "level": "L3", "approval_id": str(approval_id), + "approval_status": "pending", + "correlation_id": correlation_id, "message": "Approval requested from creator", } - async def mark_failed(db, **kwargs): - del db - execution.status = "failed" - execution.result_summary = kwargs["result_summary"] - execution.error_code = kwargs["error_code"] - execution.result_metadata = kwargs["metadata"] - return execution - class _ForbiddenGroupToolService: async def execute_scoped_workspace_tool(self, *args, **kwargs): raise AssertionError( @@ -1672,7 +1802,6 @@ async def forbidden_executor(*args, **kwargs): raise AssertionError(f"Group delete reached Agent Workspace: {args}, {kwargs}") monkeypatch.setattr(tool_step_service, "reserve_tool_execution", reserve) - monkeypatch.setattr(tool_step_service, "mark_tool_execution_failed", mark_failed) monkeypatch.setattr( tool_step_service.autonomy_service, "check_and_enforce", @@ -1689,7 +1818,19 @@ async def forbidden_executor(*args, **kwargs): result = await service.execute_pending(state, context, (call,)) assert result.error is None - assert result.messages[0]["error_code"] == "tool_approval_required" + assert result.messages == () + assert result.pending_tool_calls == (call,) + assert result.waiting_request == { + "waiting_type": "user", + "correlation_id": correlation_id, + "reason": "tool_approval_required", + "question": ( + "Workspace deletion requires approval. " + f"Approval ID: {approval_id}" + ), + "tool_call_id": "call-group-delete", + "approval_id": str(approval_id), + } assert captured_details == [ { "tool": "delete_file", @@ -1703,6 +1844,7 @@ async def forbidden_executor(*args, **kwargs): "run_id": context.run_id, "session_id": context.session_id, "workspace_scope": "group", + "tool_call_id": "call-group-delete", "group_id": str(group_id), "actor_participant_id": str(participant_id), "workspace_path": "remove-me.md", diff --git a/backend/tests/test_autonomy_service_runtime_delete.py b/backend/tests/test_autonomy_service_runtime_delete.py index 50690b209..b2b222f84 100644 --- a/backend/tests/test_autonomy_service_runtime_delete.py +++ b/backend/tests/test_autonomy_service_runtime_delete.py @@ -5,6 +5,9 @@ import pytest +from app.models.agent import Agent +from app.models.audit import ApprovalRequest +from app.models.user import User from app.services import autonomy_service as autonomy_module from app.services import group_file_service @@ -91,3 +94,219 @@ async def forbidden_direct_executor(*args, **kwargs): } ] assert db.committed is True + + +class _ScalarResult: + def __init__(self, value) -> None: + self.value = value + + def scalar_one_or_none(self): + return self.value + + +@pytest.mark.asyncio +async def test_runtime_l3_approval_is_reused_as_the_tool_call_decision( + monkeypatch, +) -> None: + tenant_id = uuid.uuid4() + run_id = uuid.uuid4() + agent = Agent( + id=uuid.uuid4(), + tenant_id=tenant_id, + creator_id=uuid.uuid4(), + name="Approval Agent", + status="idle", + is_expired=False, + access_mode="company", + autonomy_policy={"delete_files": "L3"}, + ) + + class _DB: + def __init__(self) -> None: + self.approval = None + self.added = [] + + async def execute(self, _statement): + return _ScalarResult(self.approval) + + def add(self, value) -> None: + self.added.append(value) + if isinstance(value, ApprovalRequest): + self.approval = value + + async def flush(self) -> None: + return None + + db = _DB() + requested = [] + + async def request_approval(_self, _db, _agent, approval): + requested.append(approval) + + monkeypatch.setattr( + autonomy_module.AutonomyService, + "_request_approval", + request_approval, + ) + details = { + "tool": "delete_file", + "args": {"path": "workspace/remove-me.md"}, + "runtime_scope": { + "tenant_id": str(tenant_id), + "run_id": str(run_id), + "session_id": str(uuid.uuid4()), + "workspace_scope": "agent", + "tool_call_id": "call-delete", + }, + } + service = autonomy_module.AutonomyService() + + pending = await service.check_and_enforce( + db, agent, "delete_files", details # type: ignore[arg-type] + ) + + expected_id = uuid.uuid5( + run_id, + "runtime-approval:delete_files:call-delete", + ) + assert pending == { + "allowed": False, + "level": "L3", + "approval_id": str(expected_id), + "approval_status": "pending", + "correlation_id": f"approval:{expected_id}", + "message": "Approval requested from creator", + } + assert db.approval.id == expected_id + assert db.approval.details["runtime_scope"]["approval_correlation_id"] == ( + f"approval:{expected_id}" + ) + assert requested == [db.approval] + + db.approval.status = "approved" + approved = await service.check_and_enforce( + db, agent, "delete_files", details # type: ignore[arg-type] + ) + + assert approved["allowed"] is True + assert approved["approval_status"] == "approved" + assert approved["approval_id"] == str(expected_id) + assert requested == [db.approval] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("action", ["approve", "reject"]) +async def test_runtime_approval_resolution_resumes_the_original_run( + monkeypatch, + action, +) -> None: + tenant_id = uuid.uuid4() + run_id = uuid.uuid4() + creator_id = uuid.uuid4() + approval_id = uuid.uuid4() + correlation_id = f"approval:{approval_id}" + approval = ApprovalRequest( + id=approval_id, + agent_id=uuid.uuid4(), + action_type="delete_files", + status="pending", + details={ + "tool": "delete_file", + "args": {"path": "workspace/remove-me.md"}, + "runtime_scope": { + "tenant_id": str(tenant_id), + "run_id": str(run_id), + "session_id": str(uuid.uuid4()), + "workspace_scope": "agent", + "tool_call_id": "call-delete", + "approval_correlation_id": correlation_id, + }, + }, + ) + agent = Agent( + id=approval.agent_id, + tenant_id=tenant_id, + creator_id=creator_id, + name="Approval Agent", + status="idle", + is_expired=False, + access_mode="company", + ) + user = User( + id=creator_id, + tenant_id=tenant_id, + display_name="Creator", + role="member", + is_active=True, + ) + + class _DB: + def __init__(self) -> None: + self.results = iter((approval, agent)) + self.added = [] + self.flush_count = 0 + + async def execute(self, _statement): + return _ScalarResult(next(self.results)) + + def add(self, value) -> None: + self.added.append(value) + + async def flush(self) -> None: + self.flush_count += 1 + + db = _DB() + resumed = [] + notifications = [] + + class _RuntimeCommandIntake: + def __init__(self, db_arg) -> None: + assert db_arg is db + + async def resume_run(self, command): + resumed.append(command) + + async def send_notification(db_arg, **kwargs): + assert db_arg is db + notifications.append(kwargs) + + async def forbidden_direct_execution(*args, **kwargs): + raise AssertionError( + f"Runtime approval executed out of band: {args}, {kwargs}" + ) + + monkeypatch.setattr( + "app.services.agent_runtime.adapter.RuntimeCommandIntake", + _RuntimeCommandIntake, + ) + monkeypatch.setattr( + "app.services.notification_service.send_notification", + send_notification, + ) + monkeypatch.setattr( + autonomy_module.AutonomyService, + "_execute_approved_action", + forbidden_direct_execution, + ) + + resolved = await autonomy_module.AutonomyService().resolve_approval( + db, approval_id, user, action # type: ignore[arg-type] + ) + + expected_status = "approved" if action == "approve" else "rejected" + assert resolved.status == expected_status + assert len(resumed) == 1 + command = resumed[0] + assert command.tenant_id == tenant_id + assert command.run_id == run_id + assert command.idempotency_key == ( + f"approval:{approval_id}:{expected_status}" + ) + assert command.payload["resume_type"] == "user_input" + assert command.payload["correlation_id"] == correlation_id + assert command.payload["payload"]["decision"] == expected_status + assert command.actor_user_id == creator_id + assert db.flush_count == 2 + assert notifications[0]["body"] == ( + "Result: Original Agent Run queued to resume" + )