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..75e64ec10 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,91 @@ def _is_group_workspace_mutation_call( ) +def _delete_autonomy_details( + state: RuntimeGraphState, + context: RuntimeContext, + agent: Agent, + call_id: str, + 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", + "tool_call_id": call_id, + } + 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 +1160,110 @@ async def _mark_policy_blocked( ), ) + async def _delete_autonomy_gate( + self, + *, + state: RuntimeGraphState, + context: RuntimeContext, + agent: Agent, + call_id: str, + tool_name: str, + arguments: Mapping[str, object], + ) -> tuple[ToolExecutionOutcome | None, JsonObject | None]: + details = _delete_autonomy_details( + state, + context, + agent, + call_id, + tool_name, + arguments, + ) + if details is None: + return None, 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__}, + ), + None, + ) + if decision.get("allowed"): + return None, None + level = str(decision.get("level") or "unknown") + approval_id = decision.get("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 was rejected and was not executed. " + f"Approval ID: {approval_id}" + ), + result_ref=None, + error_code="tool_approval_rejected", + retryable=False, + metadata={ + "approval_id": approval_id, + "autonomy_level": level, + }, + ), 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}, + ), + None, + ) + async def execute_pending( self, state: RuntimeGraphState, @@ -1188,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, @@ -1447,6 +1654,24 @@ async def execute_pending( pending_tool_calls=tool_calls[index:], ) + 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..3b3769b98 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 @@ -24,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: @@ -39,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( @@ -69,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() @@ -84,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", } @@ -121,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 ) @@ -167,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: @@ -194,6 +329,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..a8416377d 100644 --- a/backend/tests/test_agent_runtime_tool_step_service.py +++ b/backend/tests/test_agent_runtime_tool_step_service.py @@ -1489,6 +1489,370 @@ 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() + correlation_id = f"approval:{approval_id}" + approval_calls: 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 + 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( + { + "agent": agent_arg, + "action_type": action_type, + "details": details, + } + ) + return { + "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_succeeded(db, **kwargs): + del db + execution.status = "succeeded" + execution.result_summary = kwargs["result_summary"] + return execution + + 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_succeeded", + mark_succeeded, + ) + 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=executor, + ) + + waiting = await service.execute_pending(state, context, (call,)) + + 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, + "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", + "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( + 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) + approval_id = uuid.uuid4() + correlation_id = f"approval:{approval_id}" + 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): + 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 + assert agent_arg is agent + assert action_type == "delete_files" + captured_details.append(details) + return { + "allowed": False, + "level": "L3", + "approval_id": str(approval_id), + "approval_status": "pending", + "correlation_id": correlation_id, + "message": "Approval requested from creator", + } + + 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.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 == () + 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", + "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", + "tool_call_id": "call-group-delete", + "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 +2150,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 +2171,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..b2b222f84 --- /dev/null +++ b/backend/tests/test_autonomy_service_runtime_delete.py @@ -0,0 +1,312 @@ +"""Runtime-specific approval execution tests.""" + +from contextlib import asynccontextmanager +import uuid + +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 + + +@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 + + +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" + )