feat: close AgentScope D2+D4 thinking/transport gaps

dare_framework/agent/base_agent.py

@@ -5,6 +5,7 @@
 from abc import ABC, abstractmethod
 import asyncio
 import contextlib
+from contextvars import ContextVar
 from dataclasses import replace
 import logging
 from typing import TYPE_CHECKING, Any
@@ -237,10 +238,20 @@ async def _execute_polled_message(
         envelope_id: str | None,
     ) -> None:
         """Execute one polled message and send response envelope through channel."""
-        result = await self.execute(task, transport=channel)
+        # Keep transport-loop execution state task-local so concurrent execute() calls on
+        # the same agent instance do not leak loop state across tasks.
+        transport_loop_token = _TRANSPORT_LOOP_EXECUTION_CTX.set(True)
+        try:
+            result = await self.execute(task, transport=channel)
+        finally:
+            _TRANSPORT_LOOP_EXECUTION_CTX.reset(transport_loop_token)
         result = self._with_normalized_output_text(result)
         await self._send_transport_result(result, task=task, transport=channel, reply_to=envelope_id)
 
+    def _is_transport_loop_execution(self, *, transport: AgentChannel | None) -> bool:
+        """Return whether execute() is currently running under the transport loop."""
+        return bool(transport is not None and _TRANSPORT_LOOP_EXECUTION_CTX.get())
+
     def _with_normalized_output_text(self, result: RunResult) -> RunResult:
         """Ensure RunResult.output_text is filled for downstream consumers."""
         if result.output_text is not None:
@@ -355,6 +366,7 @@ def get_agent_control_handler(self) -> None:
 
 
 _NO_OP_AGENT_CHANNEL = _NoOpAgentChannel()
+_TRANSPORT_LOOP_EXECUTION_CTX: ContextVar[bool] = ContextVar("_transport_loop_execution", default=False)
 
 
 def _coerce_polled_envelopes(polled: Any) -> list[TransportEnvelope]:

dare_framework/agent/react_agent.py

@@ -91,7 +91,9 @@ def _print_context_list(
 from dare_framework.plan.types import RunResult
 from dare_framework.plan.types import Task
 from dare_framework.tool import IToolGateway, IToolProvider
+from dare_framework.transport.interaction.payloads import build_error_payload, build_success_payload
 from dare_framework.transport.kernel import AgentChannel
+from dare_framework.transport.types import EnvelopeKind, TransportEnvelope, TransportEventType, new_envelope_id
 
 
 class ReactAgent(BaseAgent):
@@ -151,7 +153,6 @@ async def _execute_basic(
         transport: AgentChannel | None = None,
     ) -> RunResult:
         """原始基础 ReAct 循环实现。"""
-        _ = transport
         task_description = task.description if isinstance(task, Task) else task
         user_message = Message(role="user", content=task_description)
         self._context.stm_add(user_message)
@@ -204,6 +205,14 @@ async def _execute_basic(
                     latest_usage = usage
             n_tools = len(response.tool_calls) if response.tool_calls else 0
             print(f"[{self.name}] 模型返回, tool_calls={n_tools}", flush=True)
+            thinking_content = (response.thinking_content or "").strip()
+            if thinking_content:
+                await self._emit_transport_success(
+                    transport=transport,
+                    event_type=TransportEventType.THINKING.value,
+                    target="model",
+                    resp={"output": thinking_content},
+                )
 
             if usage is not None:
                 tokens = _usage_total_tokens(usage)
@@ -217,6 +226,10 @@ async def _execute_basic(
                     final_text = "模型未返回可显示的文本回复。请重试，或明确要求先调用 ask_user 再继续。"
                 assistant_message = Message(role="assistant", content=final_text)
                 self._context.stm_add(assistant_message)
+                await self._emit_terminal_transport_message(
+                    transport=transport,
+                    output=final_text,
+                )
                 output = build_output_envelope(
                     final_text,
                     usage=latest_usage,
@@ -237,6 +250,10 @@ async def _execute_basic(
             if repeated_tool_rounds >= 3:
                 loop_guard = "模型连续重复调用相同工具，已停止自动循环。请换一种描述，或明确要求先调用 ask_user 再继续。"
                 self._context.stm_add(Message(role="assistant", content=loop_guard))
+                await self._emit_terminal_transport_message(
+                    transport=transport,
+                    output=loop_guard,
+                )
                 output = build_output_envelope(loop_guard, usage=latest_usage)
                 return RunResult(success=True, output=output, output_text=output["content"])
 
@@ -252,6 +269,16 @@ async def _execute_basic(
                 name = tool_call.get("name", "")
                 tool_call_id = tool_call.get("id", "")
                 params = _normalize_tool_args(tool_call.get("arguments", {}))
+                await self._emit_transport_success(
+                    transport=transport,
+                    event_type=TransportEventType.TOOL_CALL.value,
+                    target=name or "tool_call",
+                    resp={
+                        "id": tool_call_id,
+                        "name": name,
+                        "arguments": params,
+                    },
+                )
                 # 打印工具调用信息：名称、参数
                 params_str = json.dumps(params, ensure_ascii=False)
                 params_preview = params_str[:300] + ("..." if len(params_str) > 300 else "")
@@ -260,6 +287,12 @@ async def _execute_basic(
                 try:
                     result = await gateway.invoke(name, envelope=envelope, **params)
                 except Exception as exc:
+                    await self._emit_transport_error(
+                        transport=transport,
+                        target=name or "tool_call",
+                        code="TOOL_CALL_FAILED",
+                        reason=str(exc).strip() or exc.__class__.__name__,
+                    )
                     result = type("R", (), {"success": False, "output": {}, "error": str(exc)})()
 
                 success = getattr(result, "success", False)
@@ -268,6 +301,18 @@ async def _execute_basic(
                 out_preview = _preview_output(output)
                 print(f"[{self.name}] 工具结果: {name} | success={success} | {out_preview}", flush=True)
                 error = getattr(result, "error", "") or ""
+                await self._emit_transport_success(
+                    transport=transport,
+                    event_type=TransportEventType.TOOL_RESULT.value,
+                    target=name or "tool_result",
+                    resp={
+                        "id": tool_call_id,
+                        "name": name,
+                        "success": bool(success),
+                        "output": output,
+                        "error": error,
+                    },
+                )
 
                 tool_content = json.dumps(
                     {"success": success, "output": output, "error": error} if not success
@@ -278,6 +323,10 @@ async def _execute_basic(
                 self._context.stm_add(tool_msg)
 
         final_message = "模型在工具循环中未收敛（达到最大轮次）。请缩小范围，或明确要求先调用 ask_user 再继续。"
+        await self._emit_terminal_transport_message(
+            transport=transport,
+            output=final_message,
+        )
         output = build_output_envelope(final_message, usage=latest_usage)
         return RunResult(
             success=True,
@@ -522,6 +571,73 @@ async def _execute_with_smart_context(
             output_text=final_message,
         )
 
+    async def _emit_terminal_transport_message(
+        self,
+        *,
+        transport: AgentChannel | None,
+        output: str,
+    ) -> None:
+        """Emit terminal MESSAGE for direct execute() transport calls."""
+        # BaseAgent emits terminal RESULT envelopes in transport-loop execution.
+        # Avoid emitting duplicate terminal MESSAGE events in that path.
+        if self._is_transport_loop_execution(transport=transport):
+            return
+        await self._emit_transport_success(
+            transport=transport,
+            event_type=TransportEventType.MESSAGE.value,
+            target="prompt",
+            resp={"output": output},
+        )
+
+    async def _emit_transport_success(
+        self,
+        *,
+        transport: AgentChannel | None,
+        event_type: str,
+        target: str,
+        resp: dict[str, Any],
+    ) -> None:
+        """Emit a canonical success payload to transport when available."""
+        if transport is None:
+            return
+        envelope = TransportEnvelope(
+            id=new_envelope_id(),
+            kind=EnvelopeKind.MESSAGE,
+            event_type=event_type,
+            payload=build_success_payload(kind="message", target=target, resp=resp),
+        )
+        try:
+            await transport.send(envelope)
+        except Exception:
+            self._logger.exception("react agent transport success emission failed")
+
+    async def _emit_transport_error(
+        self,
+        *,
+        transport: AgentChannel | None,
+        target: str,
+        code: str,
+        reason: str,
+    ) -> None:
+        """Emit a canonical error payload to transport when available."""
+        if transport is None:
+            return
+        envelope = TransportEnvelope(
+            id=new_envelope_id(),
+            kind=EnvelopeKind.MESSAGE,
+            event_type=TransportEventType.ERROR.value,
+            payload=build_error_payload(
+                kind="message",
+                target=target,
+                code=code,
+                reason=reason,
+            ),
+        )
+        try:
+            await transport.send(envelope)
+        except Exception:
+            self._logger.exception("react agent transport error emission failed")
+
 
 def _preview_output(output: Any, max_len: int = 120) -> str:
     """生成 output 的简短预览，用于日志打印。"""

dare_framework/model/adapters/openai_adapter.py

@@ -103,11 +103,13 @@ async def generate(
 
         tool_calls = self._extract_tool_calls(response)
         usage = self._extract_usage(response)
+        thinking_content = self._extract_thinking_content(response)
 
         return ModelResponse(
             content=response.content or "",
             tool_calls=tool_calls,
             usage=usage,
+            thinking_content=thinking_content,
         )
 
     def _ensure_client(self) -> Any:
@@ -231,11 +233,55 @@ def _extract_usage(self, response: Any) -> dict[str, Any] | None:
         """Extract usage information from the response."""
         usage = getattr(response, "response_metadata", {}).get("token_usage")
         if usage:
-            return {
+            normalized = {
                 "prompt_tokens": usage.get("prompt_tokens", 0),
                 "completion_tokens": usage.get("completion_tokens", 0),
                 "total_tokens": usage.get("total_tokens", 0),
             }
+            reasoning_tokens = self._extract_reasoning_tokens(usage)
+            if reasoning_tokens is not None:
+                normalized["reasoning_tokens"] = reasoning_tokens
+            return normalized
+        return None
+
+    def _extract_reasoning_tokens(self, usage: dict[str, Any]) -> int | None:
+        """Extract reasoning token count from provider-specific usage payloads."""
+        candidates: list[Any] = [
+            usage.get("reasoning_tokens"),
+            usage.get("output_tokens_details", {}).get("reasoning_tokens")
+            if isinstance(usage.get("output_tokens_details"), dict)
+            else None,
+            usage.get("output_tokens_details", {}).get("reasoning")
+            if isinstance(usage.get("output_tokens_details"), dict)
+            else None,
+            usage.get("completion_tokens_details", {}).get("reasoning_tokens")
+            if isinstance(usage.get("completion_tokens_details"), dict)
+            else None,
+        ]
+        for candidate in candidates:
+            try:
+                if candidate is None:
+                    continue
+                return int(candidate)
+            except (TypeError, ValueError):
+                continue
+        return None
+
+    def _extract_thinking_content(self, response: Any) -> str | None:
+        """Extract provider reasoning text into framework-level thinking content."""
+        additional_kwargs = getattr(response, "additional_kwargs", {})
+        if isinstance(additional_kwargs, dict):
+            for key in ("reasoning_content", "reasoning", "thinking"):
+                content = _coerce_text(additional_kwargs.get(key))
+                if content:
+                    return content
+
+        response_metadata = getattr(response, "response_metadata", {})
+        if isinstance(response_metadata, dict):
+            for key in ("reasoning_content", "reasoning", "thinking"):
+                content = _coerce_text(response_metadata.get(key))
+                if content:
+                    return content
         return None
 
     def _log_client_config(self, client: Any) -> None:
@@ -277,3 +323,25 @@ def _build_http_clients(self) -> tuple[Any | None, Any | None]:
 
 
 __all__ = ["OpenAIModelAdapter"]
+
+
+def _coerce_text(value: Any) -> str | None:
+    """Coerce heterogenous provider reasoning payloads into a non-empty string."""
+    if isinstance(value, str):
+        text = value.strip()
+        return text or None
+    if isinstance(value, dict):
+        for key in ("text", "content", "reasoning", "thinking"):
+            text = _coerce_text(value.get(key))
+            if text:
+                return text
+        return None
+    if isinstance(value, list):
+        parts: list[str] = []
+        for item in value:
+            text = _coerce_text(item)
+            if text:
+                parts.append(text)
+        if parts:
+            return "\n".join(parts)
+    return None