feat(api-tools): runner workspace operations delivery and execution API (LANG-227)

Co-authored-by: multica-agent <[email protected]>

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):

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.

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(

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'))

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,

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

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:

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}
 

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'],

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,

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

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:

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)

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(

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