diff --git a/src/langbot/pkg/agent/runner/context_builder.py b/src/langbot/pkg/agent/runner/context_builder.py index 0f9d73729..45c92e2f1 100644 --- a/src/langbot/pkg/agent/runner/context_builder.py +++ b/src/langbot/pkg/agent/runner/context_builder.py @@ -114,6 +114,7 @@ class AgentResources(typing.TypedDict): skills: list[SkillResource] storage: StorageResource platform_capabilities: dict[str, typing.Any] + workspace_operations: typing.NotRequired[list[dict[str, typing.Any]]] class AgentRuntimeContext(typing.TypedDict): diff --git a/src/langbot/pkg/agent/runner/host_models.py b/src/langbot/pkg/agent/runner/host_models.py index 2fb855e29..7b4ebb4ee 100644 --- a/src/langbot/pkg/agent/runner/host_models.py +++ b/src/langbot/pkg/agent/runner/host_models.py @@ -129,6 +129,9 @@ class ResourcePolicy(pydantic.BaseModel): allow_workspace_storage: bool = False """Whether workspace storage is allowed.""" + allowed_workspace_operations: list[str] = pydantic.Field(default_factory=list) + """Workspace operations explicitly authorized for this processor.""" + class StatePolicy(pydantic.BaseModel): """State policy for agent binding. diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index 0e04cc67a..56874e761 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -294,6 +294,9 @@ class AgentRunOrchestrator: execution_query=execution_query, platform_context=freeze_platform_context(event), reply_streams=reply_streams, + processor_type=binding.processor_type, + processor_id=binding.processor_id, + trigger_actor=event.actor.actor_id if event.actor else None, ) event_log_id = await self.journal.write_event_log( diff --git a/src/langbot/pkg/agent/runner/query_entry_adapter.py b/src/langbot/pkg/agent/runner/query_entry_adapter.py index a3ecbfe16..021f12793 100644 --- a/src/langbot/pkg/agent/runner/query_entry_adapter.py +++ b/src/langbot/pkg/agent/runner/query_entry_adapter.py @@ -47,7 +47,7 @@ class QueryEntryAdapter: INTERNAL_PREFIX = '_' SENSITIVE_PATTERNS = ('secret', 'token', 'key', 'password', 'credential', 'api_key', 'apikey') - PERMISSION_VARS = ('_pipeline_bound_plugins', '_authorized', '_permission') + PERMISSION_VARS = ('_pipeline_bound_plugins', '_pipeline_workspace_operations', '_authorized', '_permission') EVENT_DATA_MAX_STRING_BYTES = 512 @classmethod @@ -139,6 +139,7 @@ class QueryEntryAdapter: resolved_tool_sources=cls._extract_allowed_tool_sources(query), resolved_kb_uuids=cls._extract_allowed_kbs(query), resolved_skill_names=cls._extract_allowed_skills(query), + allowed_workspace_operations=cls._extract_allowed_workspace_operations(query), ) # Build state policy @@ -749,3 +750,14 @@ class QueryEntryAdapter: if not isinstance(bound_skills, list): return [] return [str(skill_name) for skill_name in bound_skills if skill_name] + + @classmethod + def _extract_allowed_workspace_operations( + cls, + query: pipeline_query.Query, + ) -> list[str] | None: + """Extract pipeline-visible workspace operation IDs from query.""" + variables = getattr(query, 'variables', None) + if not variables or '_pipeline_workspace_operations' not in variables: + return None + return ResourcePolicyProjector.normalize_names(variables.get('_pipeline_workspace_operations')) diff --git a/src/langbot/pkg/agent/runner/resource_builder.py b/src/langbot/pkg/agent/runner/resource_builder.py index 6cec4ba6a..30c59d031 100644 --- a/src/langbot/pkg/agent/runner/resource_builder.py +++ b/src/langbot/pkg/agent/runner/resource_builder.py @@ -116,8 +116,14 @@ class AgentResourceBuilder: descriptor, ) storage = self._build_storage_from_binding(manifest_perms, binding) + workspace_operations = await self._build_workspace_operations_from_binding( + execution_context, + event, + binding, + descriptor, + ) - return { + resources: AgentResources = { 'models': models, 'tools': tools, 'knowledge_bases': knowledge_bases, @@ -125,6 +131,101 @@ class AgentResourceBuilder: 'storage': storage, 'platform_capabilities': platform_capabilities, } + if workspace_operations is not None: + resources['workspace_operations'] = workspace_operations + + return resources + + async def _build_workspace_operations_from_binding( + self, + execution_context: ExecutionContext, + event: AgentEventEnvelope, + binding: AgentBinding, + descriptor: RunnerDescriptor, + ) -> list[dict[str, typing.Any]] | None: + """Build authorized workspace operations for this run. + + Delivered as an independent 'workspace_operations' resource field (never added to tools). + Delivery conditions: + - Runner manifest declares workspace_operations permission, OR + - Builtin LocalAgent with user-configured operations. + """ + is_local_agent = ( + getattr(descriptor, 'plugin_author', None) == 'langbot-team' + and getattr(descriptor, 'plugin_name', None) == 'LocalAgent' + ) or getattr(descriptor, 'id', '') == 'plugin:langbot-team/LocalAgent/default' + + manifest_declares = False + perms = getattr(descriptor, 'permissions', None) + if perms is not None and getattr(perms, 'workspace_operations', None): + manifest_declares = True + else: + raw_manifest = getattr(descriptor, 'raw_manifest', None) or {} + manifest_perms = raw_manifest.get('permissions') or {} + if manifest_perms.get('workspace_operations'): + manifest_declares = True + + if not (is_local_agent or manifest_declares): + return None + + allowed_ops = binding.resource_policy.allowed_workspace_operations + if not allowed_ops: + return [] + + from ...api.application.processor_context import build_processor_request_context + from ...api.application.granted_operations import list_granted + + workspace_uuid = execution_context.workspace_uuid or event.workspace_id or '' + processor_kind = binding.processor_type or 'agent' + processor_uuid = binding.processor_id or event.event_id + trigger_actor = event.actor.actor_id if event.actor else None + + try: + if getattr(self.ap, 'workspace_service', None) is not None: + processor_context = await build_processor_request_context( + self.ap, + workspace_uuid=workspace_uuid, + processor_kind=processor_kind, + processor_uuid=processor_uuid, + run_id=event.event_id, + operation_ids=allowed_ops, + trigger_actor=trigger_actor, + ) + else: + from ...api.context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext + + principal = PrincipalContext( + principal_type=PrincipalType.PROCESSOR, + processor_kind=processor_kind, + processor_uuid=processor_uuid, + run_id=event.event_id, + trigger_actor=trigger_actor, + ) + workspace = WorkspaceContext( + workspace_uuid=workspace_uuid, + permissions=frozenset(), + ) + processor_context = RequestContext( + instance_uuid='test-instance', + placement_generation=1, + request_id=event.event_id, + auth_type='processor', + principal=principal, + workspace=workspace, + allowed_operation_ids=frozenset(allowed_ops), + ) + granted_ops = list_granted( + self.ap, + processor_context, + consumer='processor', + granted=allowed_ops, + ) + return [op.as_dict() for op in granted_ops] + except Exception as e: + getattr(self.ap, 'logger', None) and self.ap.logger.warning( + f'Failed to resolve workspace operations for run: {e}' + ) + return [] async def _build_models_from_binding( self, diff --git a/src/langbot/pkg/agent/runner/resource_policy.py b/src/langbot/pkg/agent/runner/resource_policy.py index afd767e6c..1d774781a 100644 --- a/src/langbot/pkg/agent/runner/resource_policy.py +++ b/src/langbot/pkg/agent/runner/resource_policy.py @@ -24,6 +24,7 @@ class ResourcePolicyProjector: allowed_platform_tool_names: collections.abc.Iterable[typing.Any] | None = None, allowed_host_tool_names: collections.abc.Iterable[typing.Any] | None = None, override_runner_tools: bool = False, + allowed_workspace_operations: collections.abc.Iterable[typing.Any] | None = None, ) -> ResourcePolicy: """Project standard resource fields without depending on a runner ID. @@ -74,6 +75,7 @@ class ResourcePolicyProjector: allow_all_tools=allow_all_tools, allowed_kb_uuids=allowed_kb_uuids, allowed_skill_names=cls.normalize_optional_names(resolved_skill_names), + allowed_workspace_operations=cls.normalize_names(allowed_workspace_operations), ) @staticmethod diff --git a/src/langbot/pkg/agent/runner/session_registry.py b/src/langbot/pkg/agent/runner/session_registry.py index e7c12716b..bcf7e4ddc 100644 --- a/src/langbot/pkg/agent/runner/session_registry.py +++ b/src/langbot/pkg/agent/runner/session_registry.py @@ -50,6 +50,10 @@ class RunAuthorizationSnapshot(typing.TypedDict): platform_context: dict[str, typing.Any] authorized_ids: dict[str, set[str]] authorized_operations: dict[str, dict[str, set[str]]] + processor_type: typing.NotRequired[str | None] + processor_id: typing.NotRequired[str | None] + trigger_actor: typing.NotRequired[str | None] + workspace_operations: typing.NotRequired[list[dict[str, typing.Any]]] SteeringQueueItem = dict[str, typing.Any] @@ -79,6 +83,11 @@ class AgentRunSession(typing.TypedDict): status: AgentRunSessionStatus reply_streams: typing.Any steering_queue: list[SteeringQueueItem] + processor_type: typing.NotRequired[str | None] + processor_id: typing.NotRequired[str | None] + trigger_actor: typing.NotRequired[str | None] + workspace_id: typing.NotRequired[str | None] + workspace_operations: typing.NotRequired[list[dict[str, typing.Any]]] class AgentRunSessionRegistry: @@ -117,6 +126,9 @@ class AgentRunSessionRegistry: execution_query: pipeline_query.Query | None = None, platform_context: dict[str, typing.Any] | None = None, reply_streams: typing.Any = None, + processor_type: str | None = None, + processor_id: str | None = None, + trigger_actor: str | None = None, ) -> None: """Register a new agent run session. @@ -162,6 +174,10 @@ class AgentRunSessionRegistry: 'platform_context': copy.deepcopy(platform_context or {}), 'authorized_ids': self._build_authorized_ids(resources_snapshot), 'authorized_operations': self._build_authorized_operations(resources_snapshot), + 'processor_type': processor_type, + 'processor_id': processor_id, + 'trigger_actor': trigger_actor, + 'workspace_operations': copy.deepcopy(resources_snapshot.get('workspace_operations', [])), } session: AgentRunSession = { @@ -177,6 +193,11 @@ class AgentRunSessionRegistry: 'last_activity_at': now, }, 'steering_queue': [], + 'processor_type': processor_type, + 'processor_id': processor_id, + 'trigger_actor': trigger_actor, + 'workspace_id': workspace_id, + 'workspace_operations': copy.deepcopy(resources_snapshot.get('workspace_operations', [])), } async with self._lock: diff --git a/src/langbot/pkg/api/application/execution.py b/src/langbot/pkg/api/application/execution.py index 144d6a25c..b0aae0df9 100644 --- a/src/langbot/pkg/api/application/execution.py +++ b/src/langbot/pkg/api/application/execution.py @@ -653,6 +653,33 @@ EXECUTION_OPERATIONS = ( 'runner_admin', True, ), + ExecutionOperation( + 'list_application_operations', + Empty, + _output( + 'ExecutionListApplicationOperationsOutput', + operations=(list[JsonValue], ...), + domains=(list[JsonValue], ...), + ), + 'run', + False, + ), + ExecutionOperation( + 'call_application_operation', + _input( + 'ExecutionCallApplicationOperationInput', + operation_id=(Text, ...), + arguments=(Object, {}), + ), + _output( + 'ExecutionCallApplicationOperationOutput', + ok=(bool, ...), + result=(JsonValue | None, None), + error=(Object | None, None), + ), + 'run', + True, + ), ) _BY_ID = {operation.operation_id: operation for operation in EXECUTION_OPERATIONS} diff --git a/src/langbot/pkg/api/http/service/agent.py b/src/langbot/pkg/api/http/service/agent.py index 4c8e7d85a..1ce3d4a1f 100644 --- a/src/langbot/pkg/api/http/service/agent.py +++ b/src/langbot/pkg/api/http/service/agent.py @@ -299,6 +299,7 @@ class AgentService: allowed_platform_tool_names=resolve_agent_platform_tool_names(config, event_type), allowed_host_tool_names=config.get('allowed_tools'), override_runner_tools='allowed_tools' in config, + allowed_workspace_operations=config.get('allowed_workspace_operations'), ), state_policy=StatePolicy( state_scopes=['conversation', 'actor', 'subject', 'runner'], diff --git a/src/langbot/pkg/api/http/service/pipeline_migration.py b/src/langbot/pkg/api/http/service/pipeline_migration.py index b4ec9b502..0418ec133 100644 --- a/src/langbot/pkg/api/http/service/pipeline_migration.py +++ b/src/langbot/pkg/api/http/service/pipeline_migration.py @@ -804,7 +804,10 @@ class PipelineMigrationService: if tool_grant: if not config_schema.uses_host_tools(descriptor) or 'call' not in getattr(permissions, 'tools', []): raise MigrationError('runner_resource_unavailable') - policy = ResourcePolicyProjector.from_runner_config(runner_config) + policy = ResourcePolicyProjector.from_runner_config( + runner_config, + allowed_workspace_operations=[], + ) tools = await AgentResourceBuilder(self.ap)._build_tools_from_binding( execution, permissions, diff --git a/src/langbot/pkg/pipeline/controller.py b/src/langbot/pkg/pipeline/controller.py index bb042b591..4778d16d6 100644 --- a/src/langbot/pkg/pipeline/controller.py +++ b/src/langbot/pkg/pipeline/controller.py @@ -60,6 +60,7 @@ class Controller: query.pipeline_config = pipeline.pipeline_entity.config query.variables['_pipeline_bound_plugins'] = pipeline.bound_plugins query.variables['_pipeline_bound_mcp_servers'] = pipeline.bound_mcp_servers + query.variables['_pipeline_workspace_operations'] = getattr(pipeline, 'workspace_operations', []) if getattr(getattr(query, 'launcher_type', None), 'value', None) == 'group': from .resprule.resprule import GroupRespondRuleCheckStage from .entities import ResultType diff --git a/src/langbot/pkg/pipeline/pipelinemgr.py b/src/langbot/pkg/pipeline/pipelinemgr.py index 7383032b7..9aab3722a 100644 --- a/src/langbot/pkg/pipeline/pipelinemgr.py +++ b/src/langbot/pkg/pipeline/pipelinemgr.py @@ -128,6 +128,7 @@ class RuntimePipeline: extensions_prefs = normalize_extension_preferences(pipeline_entity.extensions_preferences) self.enable_all_plugins = extensions_prefs['enable_all_plugins'] is True self.enable_all_mcp_servers = extensions_prefs['enable_all_mcp_servers'] is True + self.workspace_operations = extensions_prefs.get('workspace_operations', []) pipeline_config = pipeline_entity.config or {} runner_config: dict[str, typing.Any] = {} runner_id = RunnerConfigResolver.resolve_runner_id(pipeline_config) @@ -196,6 +197,7 @@ class RuntimePipeline: query.variables['_pipeline_bound_mcp_servers'] = self.bound_mcp_servers query.variables['_pipeline_mcp_resource_attachments'] = self.mcp_resource_attachments query.variables['_pipeline_mcp_resource_agent_read_enabled'] = self.mcp_resource_agent_read_enabled + query.variables['_pipeline_workspace_operations'] = self.workspace_operations # Telemetry: the bindings this lane runs with. None means "all enabled", # which the *_all flags make explicit in the chain record. try: diff --git a/src/langbot/pkg/pipeline/preproc/preproc.py b/src/langbot/pkg/pipeline/preproc/preproc.py index 37c5d28b8..c96d03e79 100644 --- a/src/langbot/pkg/pipeline/preproc/preproc.py +++ b/src/langbot/pkg/pipeline/preproc/preproc.py @@ -105,6 +105,9 @@ class PreProcessor(stage.PipelineStage): policy = ResourcePolicyProjector.from_runner_config( runner_config, resolved_tool_names=ResourcePolicyProjector.extract_tool_names(catalog), + allowed_workspace_operations=(query.variables or {}).get('_pipeline_workspace_operations') + if isinstance(getattr(query, 'variables', None), dict) + else None, ) catalog = ResourcePolicyProjector.filter_tools(catalog, policy) ToolManager.bind_query_tool_sources(query, catalog) diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index cd3e150fe..eb49c51f3 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -894,6 +894,7 @@ class RuntimeBot: allowed_platform_tool_names=resolve_agent_platform_tool_names(config, event_type), allowed_host_tool_names=config.get('allowed_tools'), override_runner_tools='allowed_tools' in config, + allowed_workspace_operations=config.get('allowed_workspace_operations'), ), state_policy=StatePolicy(state_scopes=['conversation', 'actor', 'subject', 'runner']), delivery_policy=DeliveryPolicy( diff --git a/src/langbot/pkg/plugin/handler.py b/src/langbot/pkg/plugin/handler.py index aead3867e..da8d5c63d 100644 --- a/src/langbot/pkg/plugin/handler.py +++ b/src/langbot/pkg/plugin/handler.py @@ -66,6 +66,15 @@ from .agent_run_support import ( ) +class _RuntimeActionName: + def __init__(self, value: str): + self.value = value + + +def _plugin_runtime_action(name: str, value: str) -> Any: + return getattr(PluginToRuntimeAction, name, _RuntimeActionName(value)) + + _HOST_RESERVED_QUERY_VAR_PREFIXES = ( '_host_', '_pipeline_bound_', @@ -2819,6 +2828,118 @@ class RuntimeConnectionHandler(handler.Handler): }, ) + @self.action(_plugin_runtime_action('LIST_APPLICATION_OPERATIONS', 'list_application_operations')) + async def list_application_operations(data: dict[str, Any]) -> handler.ActionResponse: + run_id = data.get('run_id') + if not run_id: + return handler.ActionResponse.error(message='run_id is required') + session_registry = get_session_registry() + session = await session_registry.get(run_id) + if not session: + return handler.ActionResponse.error(message=f'Run session {run_id} not found or expired') + caller_plugin_identity = data.get('caller_plugin_identity') + if caller_plugin_identity and caller_plugin_identity != session.get('plugin_identity'): + return handler.ActionResponse.error(message=f'Plugin identity mismatch for run_id {run_id}') + + resources = session.get('authorization', {}).get('resources', {}) + operations = resources.get('workspace_operations') or session.get('workspace_operations', []) + from ..api.application.granted_operations import domains + + serialized_ops = [op.as_dict() if hasattr(op, 'as_dict') else op for op in operations] + domains_summary = domains(serialized_ops) + return handler.ActionResponse.success( + data={ + 'operations': serialized_ops, + 'domains': domains_summary, + } + ) + + @self.action(_plugin_runtime_action('CALL_APPLICATION_OPERATION', 'call_application_operation')) + async def call_application_operation(data: dict[str, Any]) -> handler.ActionResponse: + run_id = data.get('run_id') + if not run_id: + return handler.ActionResponse.error(message='run_id is required') + session_registry = get_session_registry() + session = await session_registry.get(run_id) + if not session: + return handler.ActionResponse.error(message=f'Run session {run_id} not found or expired') + caller_plugin_identity = data.get('caller_plugin_identity') + if caller_plugin_identity and caller_plugin_identity != session.get('plugin_identity'): + return handler.ActionResponse.error(message=f'Plugin identity mismatch for run_id {run_id}') + + operation_id = data.get('operation_id') + if not operation_id or not isinstance(operation_id, str): + return handler.ActionResponse.error(message='operation_id is required') + arguments = data.get('arguments', {}) + if arguments is None: + arguments = {} + if not isinstance(arguments, dict): + return handler.ActionResponse.error(message='arguments must be an object') + + authorization = session.get('authorization', {}) + workspace_id = authorization.get('workspace_id') or session.get('workspace_id') + if not workspace_id: + return handler.ActionResponse.error(message='workspace_id not found in run session') + + processor_type = authorization.get('processor_type') or session.get('processor_type', 'agent') + processor_id = authorization.get('processor_id') or session.get('processor_id') or run_id + trigger_actor = authorization.get('trigger_actor') or session.get('trigger_actor') + + # Frozen granted operations from session + resources = authorization.get('resources', {}) + frozen_ops = resources.get('workspace_operations') or session.get('workspace_operations', []) + granted_ids = [op.get('id') if isinstance(op, dict) else getattr(op, 'id', str(op)) for op in frozen_ops] + + from ..api.application.processor_context import build_processor_request_context + import langbot.pkg.api.application.granted_operations as granted_operations + + try: + if getattr(self.ap, 'workspace_service', None) is not None: + processor_context = await build_processor_request_context( + self.ap, + workspace_uuid=workspace_id, + processor_kind=processor_type, + processor_uuid=processor_id, + run_id=run_id, + operation_ids=granted_ids, + trigger_actor=trigger_actor, + ) + else: + from ..api.context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext + + principal = PrincipalContext( + principal_type=PrincipalType.PROCESSOR, + processor_kind=processor_type, + processor_uuid=processor_id, + run_id=run_id, + trigger_actor=trigger_actor, + ) + workspace = WorkspaceContext( + workspace_uuid=workspace_id, + permissions=frozenset(), + ) + processor_context = RequestContext( + instance_uuid='test-instance', + placement_generation=1, + request_id=str(uuid.uuid4()), + auth_type='processor', + principal=principal, + workspace=workspace, + allowed_operation_ids=frozenset(granted_ids), + ) + + result = await granted_operations.call( + self.ap, + processor_context, + operation_id=operation_id, + arguments=arguments, + consumer='processor', + granted=granted_ids, + ) + return handler.ActionResponse.success(data=result) + except Exception as e: + return handler.ActionResponse.error(message=f'Application operation execution failed: {e}') + self._secure_plugin_actions() @contextlib.contextmanager