diff --git a/app/runtime/event/contracts.py b/app/runtime/event/contracts.py index 71f6eeda0..bab81eabe 100644 --- a/app/runtime/event/contracts.py +++ b/app/runtime/event/contracts.py @@ -143,13 +143,13 @@ _SNAPSHOT_EVENTS = { ChainEventType.SubscribeCompletionCheck, } -_INPUT_MODELS = { +_INPUT_MODELS: dict[EventType | ChainEventType, type[BaseModel]] = { ChainEventType.ResourceSelection: event_schemas.ResourceSelectionInputContractData, ChainEventType.ResourceDownload: event_schemas.ResourceDownloadInputContractData, ChainEventType.SubscribeCompletionCheck: event_schemas.SubscribeCompletionCheckInputContractData, } -_OUTPUT_MODELS = { +_OUTPUT_MODELS: dict[EventType | ChainEventType, type[BaseModel]] = { ChainEventType.ResourceSelection: event_schemas.ResourceSelectionOutputContractData, ChainEventType.ResourceDownload: event_schemas.ResourceDownloadOutputContractData, ChainEventType.SubscribeCompletionCheck: event_schemas.SubscribeCompletionCheckOutputContractData, @@ -213,9 +213,14 @@ def _build_contract(event_type: EventType | ChainEventType) -> EventContract: ) -EVENT_CONTRACTS = { +_ALL_EVENT_TYPES: tuple[EventType | ChainEventType, ...] = ( + *tuple(EventType), + *tuple(ChainEventType), +) + +EVENT_CONTRACTS: dict[EventType | ChainEventType, EventContract] = { event_type: _build_contract(event_type) - for event_type in (*tuple(EventType), *tuple(ChainEventType)) + for event_type in _ALL_EVENT_TYPES } diff --git a/app/runtime/events.py b/app/runtime/events.py index f7ff05645..4548c7460 100644 --- a/app/runtime/events.py +++ b/app/runtime/events.py @@ -357,6 +357,8 @@ class EventManager(metaclass=Singleton): return False try: owner = object() + completion: concurrent.futures.Future[Any] = concurrent.futures.Future() + self.__sync_handles[owner] = completion def _tracked_sync() -> Any: """在同步 handler 调用栈中发布当前事件 owner。""" @@ -368,14 +370,25 @@ class EventManager(metaclass=Singleton): handle = self.__executor.submit(_tracked_sync) except RuntimeError: + self.__sync_handles.pop(owner, None) logger.warning("同步事件处理器无法投递，线程池已停止") return False - self.__sync_handles[owner] = handle - handle.add_done_callback( - lambda _completed, current_owner=owner: ( - self.__remove_sync_handle(current_owner) - ) - ) + + def _complete_sync_handle( + completed: concurrent.futures.Future[Any], + ) -> None: + """把真实线程句柄的结果转移到已登记的结算句柄。""" + if completed.cancelled(): + completion.cancel() + else: + error = completed.exception() + if error is not None: + completion.set_exception(error) + else: + completion.set_result(completed.result()) + self.__remove_sync_handle(owner) + + handle.add_done_callback(_complete_sync_handle) return True def __remove_sync_handle(self, owner: object) -> None: diff --git a/app/schemas/event.py b/app/schemas/event.py index ae7a2b2c2..9583dfc39 100644 --- a/app/schemas/event.py +++ b/app/schemas/event.py @@ -63,14 +63,14 @@ class ContextSnapshot(_ContextSnapshotBase): arbitrary_types_allowed=True, ) - @model_validator(mode="before") + @model_validator(mode="before") # type: ignore[misc] @classmethod def coerce_runtime_context(cls, value: Any) -> Any: """允许旧 Context/dataclass 进入校验，同时保持原对象继续投递。""" return _coerce_event_snapshot(value) -class FileContextSnapshot(BaseModel): +class FileContextSnapshot(BaseModel): # type: ignore[misc] """音乐批次中单个文件的元数据上下文快照。""" path: str @@ -79,7 +79,7 @@ class FileContextSnapshot(BaseModel): model_config = ConfigDict(extra="allow", from_attributes=True) - @model_validator(mode="before") + @model_validator(mode="before") # type: ignore[misc] @classmethod def coerce_runtime_context(cls, value: Any) -> Any: """把旧文件上下文对象转换成可验证的字典。""" diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index c4670d056..8ee9a8936 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -11,30 +11,26 @@ from app import schemas from app.agent.contracts import ReplyMode from app.agent.orchestrator import agent_manager from app.api.endpoints.agent import ( - _WebAgentEventPublisher, _WEB_AGENT_FILE_REGISTRY, _apply_web_agent_display_event, + _build_web_agent_command_items, + _build_web_agent_display_message_from_events, _build_web_agent_input_attachments, _build_web_agent_message_events, - _build_web_agent_command_items, _build_web_agent_session_id, _build_web_agent_session_id_async, _build_web_agent_traditional_callback_payload, - _build_web_agent_display_message_from_events, _collect_web_agent_traditional_events, _get_web_agent_type, _has_web_agent_traditional_interaction, _prepare_web_agent_audio_attachment_path_async, _resolve_web_agent_audio_refs, - _transcribe_web_agent_audio_files, - web_agent_stream, _resolve_web_agent_choice_payload, _split_web_agent_output, + _transcribe_web_agent_audio_files, + _WebAgentEventPublisher, + web_agent_stream, ) -from app.runtime.events import Event -from app.db.oper.agentchat import AgentChatOper -from app.db.models.agentchat import AgentChat -from app.application.messaging.chat import AgentChatService, configure_agent_chat_service from app.application.messaging.agent import ( AgentInteractionOption, agent_interaction_manager, @@ -45,10 +41,13 @@ from app.application.messaging.agent import ( extract_web_agent_message_from_event_data, wait_web_agent_background_tasks, ) +from app.application.messaging.chat import AgentChatService, configure_agent_chat_service from app.application.messaging.skill import skill_interaction_manager from app.chain.message import MessageChain +from app.db.oper.agentchat import AgentChatOper +from app.runtime.events import Event from app.schemas.notification import ChannelCapability, ChannelCapabilityManager -from app.schemas.types import EventType, NotificationChannel, MessageType +from app.schemas.types import EventType, MessageType, NotificationChannel @pytest.fixture(autouse=True) @@ -764,7 +763,7 @@ def test_prepare_web_agent_audio_attachment_async_cancellation_reaps_process(tmp conversion_task = asyncio.create_task( _prepare_web_agent_audio_attachment_path_async(str(source_path)) ) - await asyncio.wait_for(started.wait(), timeout=1) + await asyncio.wait_for(started.wait(), timeout=5) conversion_task.cancel() with pytest.raises(asyncio.CancelledError): await conversion_task