LangBot/src/langbot/pkg/api/http/service/agent.py

from __future__ import annotations

from ....i18n import t

from contextlib import aclosing

import datetime
import copy
import fnmatch
import time
import uuid
import typing

import sqlalchemy
from langbot_plugin.api.entities.builtin.runner.delivery import DeliveryContext
from langbot_plugin.api.entities.builtin.runner.event import (
    ActorContext,
    RawEventRef,
    SubjectContext,
)
from langbot_plugin.api.entities.builtin.runner.input import AgentInput
from langbot_plugin.api.entities.builtin.provider import message as provider_message
from langbot_plugin.api.entities.builtin.provider import prompt as provider_prompt

from ....core import app
from ....agent.runner.config_resolver import RunnerConfigResolver
from ....agent.runner.config_schema import extract_prompt_config
from ....agent.runner.execution_context import build_execution_query, project_mcp_resource_config
from ....agent.runner.host_models import (
    AgentBinding,
    AgentEventEnvelope,
    BindingScope,
    DeliveryPolicy,
    StatePolicy,
)
from ....agent.runner.resource_policy import ResourcePolicyProjector
from ....agent.runner.platform_tools import (
    PLATFORM_TOOL_DEFINITIONS,
    platform_tool_catalog,
    resolve_agent_platform_tool_names,
    validate_debug_mock_options,
)
from ....entity.persistence import agent as persistence_agent
from ....workspace.errors import WorkspaceNotFoundError
from ..context import ExecutionContext, RequestContext
from .tenant import TenantContext, require_workspace_uuid, scope_statement


AGENT_KIND_AGENT = 'agent'
AGENT_KIND_PIPELINE = 'pipeline'
AGENT_KIND_EVENT_PROCESSOR = 'event_processor'
PIPELINE_EVENT_PATTERNS = ['message.received']
AGENT_DEFAULT_EVENT_PATTERNS = ['*']


class AgentService:
    """Unified processor facade for the peer Agent and Pipeline types."""

    ap: app.Application

    def __init__(self, ap: app.Application) -> None:
        self.ap = ap

    async def get_agent_metadata(self, context: TenantContext) -> dict[str, typing.Any]:
        """Return metadata needed by Agent forms."""
        pipeline_metadata = await self.ap.pipeline_service.get_pipeline_metadata(context)
        ai_metadata = next((item for item in pipeline_metadata if item.get('name') == 'ai'), None)
        host_tools: list[dict[str, typing.Any]] | None = None
        get_tool_catalog = getattr(getattr(self.ap, 'tool_mgr', None), 'get_resolved_tool_catalog', None)
        if get_tool_catalog is not None:
            try:
                host_tools = await get_tool_catalog(
                    context,
                    include_skill_authoring=True,
                    include_mcp_resource_tools=True,
                )
            except Exception as exc:
                self.ap.logger.warning(f'Failed to load Agent Host tool catalog: {exc}')
        event_processors = []
        registry = getattr(self.ap, 'runner_registry', None)
        if registry is not None:
            event_processors = [
                item.model_dump(mode='json')
                for item in await registry.list_runners(
                    context,
                    usage='event',
                    use_cache=False,
                )
            ]
        return {
            'event_processors': event_processors,
            'runner_config': ai_metadata,
            'platform_tools': platform_tool_catalog(),
            'host_tools': host_tools,
            'kinds': [
                {
                    'name': AGENT_KIND_AGENT,
                    'supported_event_patterns': AGENT_DEFAULT_EVENT_PATTERNS,
                    'message_only': False,
                },
                {
                    'name': AGENT_KIND_PIPELINE,
                    'supported_event_patterns': PIPELINE_EVENT_PATTERNS,
                    'message_only': True,
                },
                {'name': AGENT_KIND_EVENT_PROCESSOR, 'supported_event_patterns': ['*'], 'message_only': False},
            ],
        }

    async def get_agents(
        self,
        context: TenantContext,
        sort_by: str = 'updated_at',
        sort_order: str = 'DESC',
    ) -> list[dict]:
        agents = await self._get_agent_rows(context)
        pipelines = await self.ap.pipeline_service.get_pipelines(
            context,
            sort_by='updated_at',
            sort_order='DESC',
        )

        items = [self._agent_to_product_item(agent) for agent in agents]
        items.extend(self._pipeline_to_product_item(pipeline) for pipeline in pipelines)

        reverse = sort_order == 'DESC'
        sort_key = sort_by if sort_by in {'created_at', 'updated_at'} else 'updated_at'
        return sorted(items, key=lambda item: self._parse_sort_time(item.get(sort_key)), reverse=reverse)

    async def get_agent(self, context: TenantContext, agent_uuid: str) -> dict | None:
        agent = await self._get_agent_row(context, agent_uuid)
        if agent is not None:
            return self._agent_to_product_item(agent, include_config=True)

        pipeline = await self.ap.pipeline_service.get_pipeline(context, agent_uuid)
        if pipeline is not None:
            return self._pipeline_to_product_item(pipeline, include_config=True)

        return None

    async def debug_agent(
        self,
        context: RequestContext,
        agent_uuid: str,
        payload: dict[str, typing.Any],
        *,
        on_result: typing.Callable[[dict[str, typing.Any]], typing.Awaitable[None]] | None = None,
    ) -> dict[str, typing.Any]:
        """Execute one synthetic event against a configured Agent.

        The debug surface uses a trusted Workspace execution context, never
        delivers outputs to a real platform, and supports both message and
        non-message event envelopes.
        """
        agent = await self.get_agent(context, agent_uuid)
        if agent is None or agent.get('kind') not in {AGENT_KIND_AGENT, AGENT_KIND_EVENT_PROCESSOR}:
            raise ValueError('Agent not found')

        event_type = str(payload.get('event_type', 'message.received')).strip()
        if not event_type or len(event_type) > 128:
            raise ValueError('Invalid event_type')
        if not self._supports_event_type(
            agent.get('supported_event_patterns'),
            event_type,
        ):
            raise ValueError('Agent does not support this event type')

        text = str(payload.get('text') or '').strip()
        if len(text) > 20_000:
            raise ValueError('Debug input is too long')
        event_data = payload.get('data', {})
        if not isinstance(event_data, dict):
            raise ValueError('Debug event data must be an object')
        mock_options = validate_debug_mock_options(payload.get('mock', {}))

        config = agent.get('config')
        if not isinstance(config, dict):
            raise ValueError('Agent configuration is invalid')
        _, runner_id, runner_config = RunnerConfigResolver.resolve_agent_config(config)
        if not runner_id:
            raise ValueError('Agent has no configured runner')

        conversation_id = str(payload.get('conversation_id') or f'debug:{agent_uuid}').strip()
        if not conversation_id or len(conversation_id) > 256:
            raise ValueError('Invalid debug conversation_id')

        actor_payload = payload.get(
            'actor',
            {
                'actor_type': 'user',
                'actor_id': str(
                    event_data.get('member_id')
                    or event_data.get('requester_id')
                    or event_data.get('user_id')
                    or 'debug-user'
                ),
                'actor_name': str(
                    event_data.get('member_name')
                    or event_data.get('requester_name')
                    or event_data.get('user_name')
                    or 'Debug User'
                ),
            },
        )
        subject_payload = payload.get(
            'subject',
            {
                'subject_type': 'message' if event_type.startswith('message.') else event_type.split('.', 1)[0],
                'subject_id': 'debug-subject',
                'data': event_data,
            },
        )
        if not isinstance(actor_payload, dict) or not isinstance(subject_payload, dict):
            raise ValueError('Debug actor and subject must be objects')

        event_id = f'debug:{agent_uuid}:{uuid.uuid4()}'
        is_group = event_type.startswith(('group.', 'bot.')) or bool(event_data.get('group_id'))
        actor = ActorContext.model_validate(actor_payload)
        target_id = str(event_data.get('group_id') or 'debug-group') if is_group else actor.actor_id
        event = AgentEventEnvelope(
            event_id=event_id,
            event_type=event_type,
            event_time=int(time.time()),
            source='webui',
            source_event_type=event_type,
            workspace_id=context.workspace_uuid,
            conversation_id=conversation_id,
            actor=actor,
            subject=SubjectContext.model_validate(subject_payload),
            input=AgentInput.model_validate(
                {
                    'text': text or event_type,
                    'contents': [
                        {'type': 'text', 'text': text or event_type},
                    ],
                    'attachments': [],
                }
            ),
            delivery=DeliveryContext(
                surface='webui',
                reply_target={
                    'target_type': 'group' if is_group else 'person',
                    'target_id': target_id,
                    **({'group_id': target_id} if is_group else {}),
                    **(
                        {'message_id': str(event_data.get('message_id') or 'debug-message')}
                        if event_type.startswith('message.')
                        else {}
                    ),
                },
                supports_streaming=on_result is not None,
                supports_edit=False,
                supports_reaction=False,
                platform_capabilities={
                    'event_type': event_type,
                    'debug': True,
                    'debug_mock': True,
                    'supported_apis': sorted(
                        {tool.api for tool in PLATFORM_TOOL_DEFINITIONS} - set(mock_options.get('unsupported_apis', []))
                    ),
                    'mock_options': mock_options,
                },
            ),
            raw_ref=RawEventRef(ref_id=event_id, storage_key=None),
            data=event_data,
        )
        if agent.get('kind') == AGENT_KIND_EVENT_PROCESSOR:
            from langbot_plugin.api.entities.builtin.platform.events import parse_eba_event

            typed_event = parse_eba_event({**event_data, 'type': event_type})
            from ....platform.botmgr import RuntimeBot

            event.actor = RuntimeBot._infer_actor_context(typed_event)
            event.subject = RuntimeBot._infer_subject_context(typed_event)
            target_type, target_id, target_metadata = RuntimeBot._infer_reply_target(typed_event)
            if target_id is not None:
                event.delivery.reply_target = {
                    'target_type': target_type,
                    'target_id': str(target_id),
                    **target_metadata,
                }
            event.data = typed_event.model_dump(mode='json', exclude={'source_platform_object', 'legacy_event'})
        binding = AgentBinding(
            binding_id=(
                f'event_processor:{agent_uuid}'
                if agent.get('kind') == AGENT_KIND_EVENT_PROCESSOR
                else f'debug:{agent_uuid}:{runner_id}'
            ),
            scope=BindingScope(scope_type='agent', scope_id=agent_uuid),
            event_types=[event_type],
            runner_id=runner_id,
            runner_config=runner_config,
            resource_policy=ResourcePolicyProjector.from_runner_config(
                runner_config,
                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,
            ),
            state_policy=StatePolicy(
                state_scopes=['conversation', 'actor', 'subject', 'runner'],
            ),
            delivery_policy=DeliveryPolicy(
                enable_streaming=on_result is not None,
                enable_reply=False,
                enable_interactions=False,
            ),
            agent_id=agent_uuid,
            processor_type=agent.get('kind', 'agent'),
            processor_id=agent_uuid,
        )
        execution_context = ExecutionContext.from_request(
            context,
            query_uuid=event_id,
        )
        # Debug bypasses pipeline preprocessing, so supply the selected Runner's
        # prompt before advertising the Host prompt API. Keep explicit [] intact.
        descriptor = await self.ap.runner_registry.get(execution_context, runner_id)
        prompt_config = extract_prompt_config(descriptor, runner_config, [])
        execution_query = build_execution_query(event, [])
        execution_query.prompt = provider_prompt.Prompt(
            name='default',
            messages=[provider_message.Message(**message) for message in copy.deepcopy(prompt_config)],
        )
        for field_name in ('instance_uuid', 'workspace_uuid', 'placement_generation', 'query_uuid'):
            object.__setattr__(execution_query, field_name, getattr(execution_context, field_name))
        project_mcp_resource_config(execution_query, runner_config)

        output_items: list[dict[str, typing.Any]] = []
        execution_events: list[dict[str, typing.Any]] = []

        async def observe_result(result: dict[str, typing.Any]) -> None:
            if result.get('type') not in {
                'message.delta',
                'message.completed',
                'tool.call.started',
                'tool.call.completed',
                'run.completed',
                'run.failed',
                'processor.log',
            }:
                return
            visible_result = copy.deepcopy(
                {
                    key: result[key]
                    for key in ('type', 'data', 'sequence', 'timestamp', 'run_id', 'usage')
                    if key in result
                }
            )
            if on_result is not None:
                await on_result(visible_result)
            elif len(execution_events) < 1000:
                execution_events.append(visible_result)

        final_text = ''
        # A debug run has no platform event: the synthetic event envelope is the
        # execution's inbound boundary, so the chain is opened here with the real
        # event id and marked as a debug WebUI/API execution. The Runner inside
        # then joins this chain instead of becoming a trace origin of its own.
        from ....telemetry.execution import ingress as execution_ingress

        principal_type = str(getattr(getattr(context, 'principal', None), 'principal_type', '') or '')
        origin = 'api' if principal_type == 'api_key' else 'webui'
        with execution_ingress(
            self.ap,
            'event_done',
            execution_context,
            event_id,
            debug=True,
            origin=origin,
            synthetic_event=event_type,
        ):
            async with aclosing(
                self.ap.agent_run_orchestrator.run(
                    event,
                    binding,
                    adapter_context={
                        '_query': execution_query,
                        '_execution_context': execution_context,
                        '_result_observer': observe_result,
                    },
                )
            ) as owned_stream:
                async for output in owned_stream:
                    output_text = self._provider_output_to_text(output)
                    if output_text:
                        final_text = output_text
                    output_items.append(
                        {
                            'kind': output.__class__.__name__,
                            'role': str(getattr(output, 'role', '') or ''),
                            'text': output_text,
                        }
                    )

        return {
            'event_id': event_id,
            'event_type': event_type,
            'conversation_id': conversation_id,
            'final_text': final_text,
            'outputs': output_items,
            'execution_events': execution_events,
        }

    @staticmethod
    def _supports_event_type(patterns: typing.Any, event_type: str) -> bool:
        normalized = patterns if isinstance(patterns, list) else AGENT_DEFAULT_EVENT_PATTERNS
        return any(isinstance(pattern, str) and fnmatch.fnmatchcase(event_type, pattern) for pattern in normalized)

    @staticmethod
    def _provider_output_to_text(output: typing.Any) -> str:
        all_content = getattr(output, 'all_content', None)
        if all_content:
            return str(all_content)
        content = getattr(output, 'content', None)
        if content is None:
            return ''
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            parts: list[str] = []
            for item in content:
                item_data = item.model_dump(mode='json') if hasattr(item, 'model_dump') else item
                if isinstance(item_data, dict) and item_data.get('text') is not None:
                    parts.append(str(item_data['text']))
                elif item_data is not None and not isinstance(item_data, dict):
                    parts.append(str(item_data))
            return ''.join(parts)
        return str(content)

    async def create_agent(self, context: TenantContext, agent_data: dict) -> dict[str, str]:
        workspace_uuid = require_workspace_uuid(context)
        kind = agent_data.get('kind') or AGENT_KIND_AGENT
        if kind == AGENT_KIND_PIPELINE:
            pipeline_uuid = await self.ap.pipeline_service.create_pipeline(
                context,
                {
                    'name': agent_data.get('name') or t('New Pipeline'),
                    'description': agent_data.get('description') or '',
                    'emoji': agent_data.get('emoji') or '⚙️',
                },
            )
            return {'uuid': pipeline_uuid, 'kind': AGENT_KIND_PIPELINE}

        if kind not in {AGENT_KIND_AGENT, AGENT_KIND_EVENT_PROCESSOR}:
            raise ValueError(f'Unsupported agent kind: {kind}')

        if kind == AGENT_KIND_EVENT_PROCESSOR:
            config, runner_id, patterns = await self._prepare_event_processor(context, agent_data)
        else:
            config = agent_data['config'] if 'config' in agent_data else await self._get_default_agent_config(context)
            config, runner_id, _ = RunnerConfigResolver.resolve_agent_config(config)
            await self._validate_runner_for_agent(context, runner_id)
            if runner_id:
                descriptor = await self.ap.runner_registry.get(context, runner_id)
                config['runner_config'][runner_id] = await self.ap.llm_model_service.apply_default_model(
                    context,
                    descriptor.config_schema,
                    config['runner_config'][runner_id],
                )
            patterns = agent_data.get('supported_event_patterns', AGENT_DEFAULT_EVENT_PATTERNS)
        new_uuid = str(uuid.uuid4())
        values = {
            'workspace_uuid': workspace_uuid,
            'uuid': new_uuid,
            'name': agent_data.get('name') or t('New Agent'),
            'description': agent_data.get('description') or '',
            'emoji': agent_data.get('emoji') or ('🧩' if kind == AGENT_KIND_EVENT_PROCESSOR else '🤖'),
            'kind': kind,
            'component_ref': runner_id,
            'config': config,
            'supported_event_patterns': patterns,
        }
        await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_agent.Agent).values(**values))
        return {'uuid': new_uuid, 'kind': kind}

    async def update_agent(self, context: TenantContext, agent_uuid: str, agent_data: dict) -> None:
        existing_agent = await self._get_agent_row(context, agent_uuid)
        if existing_agent is None:
            pipeline = await self.ap.pipeline_service.get_pipeline(context, agent_uuid)
            if pipeline is None:
                raise ValueError(f'Agent {agent_uuid} not found')
            await self.ap.pipeline_service.update_pipeline(context, agent_uuid, agent_data)
            return

        update_data = {
            field: agent_data[field]
            for field in ('name', 'description', 'emoji', 'config', 'supported_event_patterns')
            if field in agent_data
        }
        if existing_agent.kind == AGENT_KIND_EVENT_PROCESSOR and any(
            field in agent_data for field in ('config', 'component_ref', 'parameters', 'supported_event_patterns')
        ):
            config, runner_id, patterns = await self._prepare_event_processor(context, agent_data, existing_agent)
            update_data.update(config=config, component_ref=runner_id, supported_event_patterns=patterns)
        if 'config' in update_data:
            config, runner_id, _ = RunnerConfigResolver.resolve_agent_config(update_data['config'])
            update_data['config'] = config
        else:
            _, runner_id, _ = RunnerConfigResolver.resolve_agent_config(existing_agent.config)
        update_data['component_ref'] = runner_id
        if existing_agent.kind == AGENT_KIND_AGENT:
            await self._validate_runner_for_agent(context, runner_id)
        result = await self.ap.persistence_mgr.execute_async(
            scope_statement(
                sqlalchemy.update(persistence_agent.Agent)
                .where(persistence_agent.Agent.uuid == agent_uuid)
                .values(**update_data),
                persistence_agent.Agent,
                context,
            )
        )
        if getattr(result, 'rowcount', None) == 0:
            raise WorkspaceNotFoundError('Agent not found')

    async def delete_agent(self, context: TenantContext, agent_uuid: str) -> None:
        existing_agent = await self._get_agent_row(context, agent_uuid)
        if existing_agent is not None:
            result = await self.ap.persistence_mgr.execute_async(
                scope_statement(
                    sqlalchemy.delete(persistence_agent.Agent).where(persistence_agent.Agent.uuid == agent_uuid),
                    persistence_agent.Agent,
                    context,
                )
            )
            if getattr(result, 'rowcount', None) == 0:
                raise WorkspaceNotFoundError('Agent not found')
            return

        pipeline = await self.ap.pipeline_service.get_pipeline(context, agent_uuid)
        if pipeline is None:
            raise ValueError(f'Agent {agent_uuid} not found')
        await self.ap.pipeline_service.delete_pipeline(context, agent_uuid)

    async def _validate_runner_for_agent(self, context, runner_id):
        if runner_id:
            descriptor = await self.ap.runner_registry.get(context, runner_id)
            if 'agent' not in descriptor.usages:
                raise ValueError('The selected Runner does not support agent usage')

    async def _prepare_event_processor(self, context, data, existing=None):
        """Resolve an installed component and keep its capability declaration authoritative."""
        draft = existing is None and 'config' not in data and 'parameters' not in data
        config = copy.deepcopy(data.get('config', existing.config if existing is not None else {}))
        if not isinstance(config, dict):
            raise ValueError('Processor configuration must be an object')
        component_ref = data.get('component_ref') or (existing.component_ref if existing is not None else None)
        if component_ref is None and not config and not data.get('parameters'):
            # An unconfigured instance cannot subscribe to or execute any events.
            return {}, None, []
        if not isinstance(component_ref, str) or not component_ref.startswith('plugin:'):
            raise ValueError('Select an installed Runner component')
        try:
            descriptor = await self.ap.runner_registry.get(context, component_ref)
        except Exception as exc:
            from ....agent.runner.errors import RunnerNotFoundError

            if isinstance(exc, RunnerNotFoundError):
                raise ValueError('Runner component is unavailable') from exc
            raise
        if 'event' not in descriptor.usages or not descriptor.supported_event_patterns:
            raise ValueError('The component does not declare supported events')
        config['runner'] = {'id': component_ref}
        parameters = data.get('parameters')
        if parameters is None:
            runner_config = config.get('runner_config', {})
            if not isinstance(runner_config, dict):
                raise ValueError('Runner configuration must be an object')
            parameters = runner_config.get(component_ref)
        if parameters is None:
            parameters = self.ap.pipeline_service._get_default_values_from_schema(descriptor.config_schema)
        if draft:
            parameters = await self.ap.llm_model_service.apply_default_model(
                context, descriptor.config_schema, parameters
            )
        if not isinstance(parameters, dict):
            raise ValueError('Processor parameters must be an object')
        for field in descriptor.config_schema:
            if not draft and field.get('required') and parameters.get(field['name']) in (None, ''):
                raise ValueError(f'Required processor parameter: {field["name"]}')
        config['runner_config'] = {component_ref: parameters}
        # Component-only creation keeps defaults for the detail form, but does
        # not subscribe to events until configuration is saved and validated.
        return config, component_ref, [] if draft else descriptor.supported_event_patterns

    async def get_processor_runs(self, context, processor_id, *, before_id=None):
        """Read only this Workspace's Agent or plugin processor runs."""
        from ....agent.runner.run_ledger_store import RunLedgerStore

        processor = await self.get_agent(context, processor_id)
        if processor is None or processor.get('kind') not in {'agent', AGENT_KIND_EVENT_PROCESSOR}:
            raise ValueError('Processor not found')
        store = RunLedgerStore(self.ap.persistence_mgr.get_db_engine())
        items, cursor, has_more, total = await store.list_runs(
            workspace_id=require_workspace_uuid(context),
            **(
                {'binding_id': f'event_processor:{processor_id}'}
                if processor.get('kind') == AGENT_KIND_EVENT_PROCESSOR
                else {'agent_id': processor_id}
            ),
            before_id=before_id,
        )
        return {'items': items, 'next_cursor': cursor, 'has_more': has_more, 'total': total}

    async def get_processor_run_events(self, context, processor_id, run_id, *, after_sequence=None):
        """Authorize the parent run before exposing any trace events."""
        from ....agent.runner.run_ledger_store import RunLedgerStore

        processor = await self.get_agent(context, processor_id)
        if processor is None or processor.get('kind') not in {'agent', AGENT_KIND_EVENT_PROCESSOR}:
            raise ValueError('Processor not found')
        store = RunLedgerStore(self.ap.persistence_mgr.get_db_engine())
        run = await store.get_run(run_id)
        if (
            run is None
            or run.get('workspace_id') != require_workspace_uuid(context)
            or not (
                run.get('binding_id') == f'event_processor:{processor_id}'
                if processor.get('kind') == AGENT_KIND_EVENT_PROCESSOR
                else run.get('agent_id') == processor_id
                or (
                    run.get('agent_id') is None
                    and str(run.get('binding_id', '')).startswith((f'agent_{processor_id}_', f'debug:{processor_id}:'))
                )
            )
        ):
            raise ValueError('Processor run not found')
        items, next_cursor, _, has_more = await store.page_run_events(
            run_id=run_id,
            after_sequence=after_sequence,
            limit=100,
        )
        return {'run': run, 'items': items, 'next_cursor': next_cursor, 'has_more': has_more}

    async def _get_agent_rows(self, context: TenantContext) -> list[persistence_agent.Agent]:
        result = await self.ap.persistence_mgr.execute_async(
            scope_statement(
                sqlalchemy.select(persistence_agent.Agent),
                persistence_agent.Agent,
                context,
            )
        )
        return list(result.all())

    async def _get_agent_row(
        self,
        context: TenantContext,
        agent_uuid: str,
    ) -> persistence_agent.Agent | None:
        result = await self.ap.persistence_mgr.execute_async(
            scope_statement(
                sqlalchemy.select(persistence_agent.Agent).where(persistence_agent.Agent.uuid == agent_uuid),
                persistence_agent.Agent,
                context,
            )
        )
        return result.first()

    async def _get_default_agent_config(self, context: TenantContext) -> dict[str, typing.Any]:
        runners = []
        if getattr(self.ap, 'runner_registry', None) is not None:
            try:
                runners = await self.ap.runner_registry.list_runners(context, bound_plugins=None)
            except Exception as e:
                if getattr(self.ap, 'logger', None):
                    self.ap.logger.warning(f'Failed to load plugin agent runners for default agent config: {e}')

        if not runners:
            return {'runner': {'id': '', 'expire-time': 0}, 'runner_config': {}}

        selected_runner = runners[0]
        return {
            'runner': {'id': selected_runner.id, 'expire-time': 0},
            'runner_config': {
                selected_runner.id: self.ap.pipeline_service._get_default_values_from_schema(
                    selected_runner.config_schema
                )
            },
        }

    def _agent_to_product_item(
        self,
        agent: persistence_agent.Agent,
        include_config: bool = False,
    ) -> dict[str, typing.Any]:
        item = self.ap.persistence_mgr.serialize_model(persistence_agent.Agent, agent)
        item['kind'] = item.get('kind') or AGENT_KIND_AGENT
        supported_event_patterns = item.get('supported_event_patterns')
        item['capability'] = {
            'supported_event_patterns': (
                supported_event_patterns if isinstance(supported_event_patterns, list) else AGENT_DEFAULT_EVENT_PATTERNS
            ),
            'message_only': False,
        }
        if not include_config:
            item.pop('config', None)
        return item

    @staticmethod
    def _pipeline_to_product_item(pipeline: dict, include_config: bool = False) -> dict[str, typing.Any]:
        item = pipeline.copy()
        item['kind'] = AGENT_KIND_PIPELINE
        item['component_ref'] = 'pipeline'
        item['supported_event_patterns'] = PIPELINE_EVENT_PATTERNS
        item['capability'] = {
            'supported_event_patterns': PIPELINE_EVENT_PATTERNS,
            'message_only': True,
        }
        if not include_config:
            item.pop('config', None)
        return item

    @staticmethod
    def _parse_sort_time(value: typing.Any) -> datetime.datetime:
        if isinstance(value, datetime.datetime):
            return value
        if isinstance(value, str):
            try:
                return datetime.datetime.fromisoformat(value)
            except ValueError:
                return datetime.datetime.min
        return datetime.datetime.min