refactor(gateway): split TurnRunner.run_sync into 11 helper methods (1229 -> 211 LOC)
This commit is contained in:
Teknium
+372
-309
@@ -5015,66 +5015,8 @@ class TurnRunner:
|
||||
ctx._cleanup_msg_ids.append(str(mid))
|
||||
_fut.add_done_callback(_track_status_id)
|
||||
|
||||
def run_sync(self):
|
||||
def _setup_stream_consumer(self, platform_key):
|
||||
ctx = self._ctx
|
||||
# As a method the turn message lives on the shared TurnContext: every rebind writes
|
||||
# `ctx.message`, so the outer `_run_agent_inner` body sees the update as via the closure cell.
|
||||
|
||||
# session_key propagates via contextvars (_set_session_env / set_current_session_key):
|
||||
# concurrency-safe and inherited by tool worker threads. Deliberately do NOT write
|
||||
# os.environ["HERMES_SESSION_KEY"]: it is process-global, so concurrent sessions would clobber
|
||||
# each other and a tool thread with an unset contextvar would read the wrong key, misrouting
|
||||
# approvals. Only the TUI slash-worker subprocess exports the env var (from its own argv).
|
||||
|
||||
# Map platform enum to the platform hint key the agent understands.
|
||||
# Platform.LOCAL ("local") maps to "cli"; others pass through as-is.
|
||||
platform_key = "cli" if ctx.source.platform == Platform.LOCAL else ctx.source.platform.value
|
||||
|
||||
# Combine platform context, YAML channel_prompts hint for this chat, channel_overrides
|
||||
# system_prompt (or global ephemeral), and the gateway ephemeral prompt.
|
||||
combined_ephemeral = ctx.context_prompt or ""
|
||||
event_channel_prompt = (ctx.channel_prompt or "").strip()
|
||||
if event_channel_prompt:
|
||||
combined_ephemeral = (combined_ephemeral + "\n\n" + event_channel_prompt).strip()
|
||||
cfg_channel_prompt = self._runner._get_system_prompt_for_channel(
|
||||
ctx.source.platform,
|
||||
ctx.source.chat_id or "",
|
||||
thread_id=getattr(ctx.source, "thread_id", None),
|
||||
parent_id=getattr(ctx.source, "parent_chat_id", None),
|
||||
)
|
||||
if cfg_channel_prompt:
|
||||
combined_ephemeral = (combined_ephemeral + "\n\n" + cfg_channel_prompt).strip()
|
||||
|
||||
max_iterations = _current_max_iterations()
|
||||
|
||||
try:
|
||||
model, runtime_kwargs = self._runner._resolve_session_agent_runtime(
|
||||
source=ctx.source,
|
||||
session_key=ctx.session_key,
|
||||
user_config=ctx.user_config,
|
||||
)
|
||||
logger.debug(
|
||||
"run_agent resolved: model=%s provider=%s session=%s",
|
||||
model, runtime_kwargs.get("provider"), ctx.session_key or "",
|
||||
)
|
||||
except Exception as exc:
|
||||
return {
|
||||
"final_response": f"⚠️ Provider authentication failed: {exc}",
|
||||
"messages": [],
|
||||
"api_calls": 0,
|
||||
"tools": [],
|
||||
}
|
||||
|
||||
pr = self._runner._provider_routing
|
||||
reasoning_config = self._runner._resolve_session_reasoning_config(
|
||||
source=ctx.source,
|
||||
session_key=ctx.session_key,
|
||||
model=model,
|
||||
)
|
||||
self._runner._reasoning_config = reasoning_config
|
||||
self._runner._service_tier = self._runner._resolve_session_service_tier(
|
||||
source=ctx.source, session_key=ctx.session_key
|
||||
)
|
||||
# Set up stream consumer for token streaming or interim commentary.
|
||||
_stream_consumer = None
|
||||
_stream_delta_cb = None
|
||||
@@ -5165,9 +5107,12 @@ class TurnRunner:
|
||||
logger=logger,
|
||||
log_message="interim_assistant_callback scheduling error",
|
||||
)
|
||||
return _stream_consumer, _stream_delta_cb, _interim_assistant_cb, _want_interim_messages
|
||||
|
||||
turn_route = self._runner._resolve_turn_agent_config(ctx.message, model, runtime_kwargs)
|
||||
|
||||
def _resolve_turn_agent(
|
||||
self, turn_route, platform_key, combined_ephemeral, max_iterations, reasoning_config, pr,
|
||||
):
|
||||
ctx = self._ctx
|
||||
# Per-platform skip_context_files — messaging platforms can opt out of filesystem-heavy
|
||||
# context-file discovery (SOUL.md, AGENTS.md, .cursorrules) to cut AIAgent build latency.
|
||||
_platforms_gw_cfg = (ctx.user_config.get("gateway") or {}).get("platforms") or {}
|
||||
@@ -5387,7 +5332,13 @@ class TurnRunner:
|
||||
)
|
||||
self._runner._enforce_agent_cache_cap()
|
||||
logger.debug("Created new agent for session %s (sig=%s)", ctx.session_key, _sig)
|
||||
return agent, reused_cached_agent
|
||||
|
||||
def _wire_turn_agent_callbacks(
|
||||
self, agent, turn_route, reasoning_config,
|
||||
_stream_delta_cb, _interim_assistant_cb, _want_interim_messages,
|
||||
):
|
||||
ctx = self._ctx
|
||||
# Per-message state — callbacks and reasoning config change every turn, so they aren't baked
|
||||
# into the cached agent. The progress callback is ALWAYS attached (never gated to None): its
|
||||
# body gates each event class, and subagent-failure notices must fire even with
|
||||
@@ -5520,144 +5471,7 @@ class TurnRunner:
|
||||
_mem_notif = "on" if _mem_notif else "off"
|
||||
agent.memory_notifications = str(_mem_notif).lower() if _mem_notif else "on"
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Shared native-stream boundary close: for native-streaming platforms (e.g. WeCom), an
|
||||
# interrupting interaction (approval or clarify prompt) must finalize the current stream
|
||||
# and disable native streaming first, or post-interaction output keeps updating the OLD
|
||||
# bubble above the prompt. Runs on the agent thread; the consumer serializes via its queue.
|
||||
def _close_native_stream_boundary(
|
||||
_reason: str, _placeholder: str | None = None, _reopen: bool = False,
|
||||
) -> bool:
|
||||
_sc = ctx.stream_consumer_holder[0] if ctx.stream_consumer_holder else None
|
||||
if not (_sc and getattr(_sc, "_use_native_streaming", False)):
|
||||
return True
|
||||
_cancelled_flag = None
|
||||
try:
|
||||
_boundary_result = _sc.close_for_approval_prompt(
|
||||
_placeholder, reason=_reason, reopen=_reopen,
|
||||
)
|
||||
# Returns (future, cancelled_flag) or just a future.
|
||||
if isinstance(_boundary_result, tuple):
|
||||
_boundary_future, _cancelled_flag = _boundary_result
|
||||
else:
|
||||
_boundary_future = _boundary_result
|
||||
if hasattr(_boundary_future, "result"):
|
||||
_ok = _boundary_future.result(timeout=10)
|
||||
if not _ok:
|
||||
logger.warning(
|
||||
"%s boundary failed to close stream properly — "
|
||||
"prompt may still appear in typing bubble", _reason,
|
||||
)
|
||||
return bool(_ok)
|
||||
return True
|
||||
except (TimeoutError, Exception) as _boundary_err:
|
||||
if _cancelled_flag is not None:
|
||||
_cancelled_flag["cancelled"] = True
|
||||
logger.warning(
|
||||
"%s boundary timed out or failed: %s", _reason, _boundary_err,
|
||||
)
|
||||
return False
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Clarify callback: present a clarify prompt and block on a response. Runs on the agent's
|
||||
# worker thread (clarify_tool's synchronous contract): schedules the adapter's send_clarify
|
||||
# on the gateway loop, then blocks on the primitive's threading.Event with a timeout.
|
||||
# Returns the response string, or a sentinel explaining no response arrived.
|
||||
# ------------------------------------------------------------------
|
||||
def _clarify_callback_sync(question: str, choices, multi_select: bool = False) -> str:
|
||||
from tools import clarify_gateway as _clarify_mod
|
||||
import uuid as _uuid
|
||||
|
||||
if not ctx._status_adapter:
|
||||
return ""
|
||||
|
||||
clarify_id = _uuid.uuid4().hex[:10]
|
||||
_clarify_mod.register(
|
||||
clarify_id=clarify_id,
|
||||
session_key=ctx.session_key or "",
|
||||
question=question,
|
||||
choices=list(choices) if choices else None,
|
||||
multi_select=bool(multi_select),
|
||||
)
|
||||
|
||||
# WeCom native streaming: finalize the current stream before the clarify prompt so the
|
||||
# post-answer output opens a fresh bubble below the question ("气泡割裂" otherwise). Unlike
|
||||
# approval, clarify passes reopen=True so the continuation re-opens a native stream; if
|
||||
# the re-seed fails the consumer degrades to send() automatically.
|
||||
_close_native_stream_boundary(
|
||||
"Clarify", "💬 等待你的选择...", _reopen=True,
|
||||
)
|
||||
|
||||
# Pause typing — as with approval, a "thinking..." status must not obscure the prompt or
|
||||
# block an "Other" reply on platforms that disable input while typing (Slack Assistant).
|
||||
with suppress(Exception):
|
||||
ctx._status_adapter.pause_typing_for_chat(ctx._status_chat_id)
|
||||
|
||||
# Ordering barrier: flush buffered assistant prose to the platform BEFORE sending the
|
||||
# poll, which goes out on a separate agent-thread-blocking path and would otherwise
|
||||
# render ABOVE its own explanation. Best-effort + short timeout so the agent thread
|
||||
# never hangs if the consumer task isn't running.
|
||||
try:
|
||||
_sc = ctx.stream_consumer_holder[0] if ctx.stream_consumer_holder else None
|
||||
_flush = getattr(_sc, "flush_pending_sync", None)
|
||||
if callable(_flush):
|
||||
_flush(timeout=3.0)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Stream-consumer flush before clarify prompt failed",
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
fut = safe_schedule_threadsafe(
|
||||
ctx._status_adapter.send_clarify(
|
||||
chat_id=ctx._status_chat_id,
|
||||
question=question,
|
||||
choices=list(choices) if choices else None,
|
||||
clarify_id=clarify_id,
|
||||
session_key=ctx.session_key or "",
|
||||
metadata=ctx._status_thread_metadata,
|
||||
),
|
||||
ctx._loop_for_step,
|
||||
logger=logger,
|
||||
log_message="Clarify send failed to schedule",
|
||||
)
|
||||
# Boundary rule (see _approval_send_outcome): a send timeout is AMBIGUOUS — the card may
|
||||
# have posted with a late ack. Only a definitive failure tears down the registration;
|
||||
# ambiguous falls through to the bounded wait so a late reply resolves.
|
||||
_clarify_response = _clarify_send_then_wait(
|
||||
fut,
|
||||
clarify_id=clarify_id,
|
||||
session_key=ctx.session_key or "",
|
||||
clarify_mod=_clarify_mod,
|
||||
)
|
||||
# Only re-arm typing when the user actually answered — the undeliverable sentinel and the
|
||||
# timeout/cancellation strings start with '[' and must pass through untouched.
|
||||
if not (
|
||||
isinstance(_clarify_response, str)
|
||||
and _clarify_response.startswith("[")
|
||||
):
|
||||
# User answered: reopen typing IMMEDIATELY, not on the LLM's first post-answer token
|
||||
# (native streaming otherwise re-seeds lazily on the first delta: ~48s of dead air).
|
||||
# request_reopen_seed is a no-op outside the reopen-pending native state; always safe.
|
||||
_sc_reopen = ctx.stream_consumer_holder[0] if ctx.stream_consumer_holder else None
|
||||
if _sc_reopen is not None:
|
||||
try:
|
||||
_sc_reopen.request_reopen_seed()
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"request_reopen_seed after clarify answer failed",
|
||||
exc_info=True,
|
||||
)
|
||||
try:
|
||||
ctx._status_adapter.resume_typing_for_chat(ctx._status_chat_id)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"resume_typing_for_chat after clarify answer failed",
|
||||
exc_info=True,
|
||||
)
|
||||
return _clarify_response
|
||||
|
||||
agent.clarify_callback = _clarify_callback_sync
|
||||
agent.clarify_callback = self._clarify_callback_sync
|
||||
|
||||
# Show assistant thinking between tool calls — independent of tool_progress mode. Mattermost
|
||||
# needs an explicit per-platform opt-in so global scratch-text doesn't leak into threads.
|
||||
@@ -5674,6 +5488,147 @@ class TurnRunner:
|
||||
# Capture the full tool definitions for transcript logging
|
||||
ctx.tools_holder[0] = agent.tools if hasattr(agent, 'tools') else None
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Shared native-stream boundary close: for native-streaming platforms (e.g. WeCom), an
|
||||
# interrupting interaction (approval or clarify prompt) must finalize the current stream
|
||||
# and disable native streaming first, or post-interaction output keeps updating the OLD
|
||||
# bubble above the prompt. Runs on the agent thread; the consumer serializes via its queue.
|
||||
def _close_native_stream_boundary(
|
||||
self, _reason: str, _placeholder: str | None = None, _reopen: bool = False,
|
||||
) -> bool:
|
||||
ctx = self._ctx
|
||||
_sc = ctx.stream_consumer_holder[0] if ctx.stream_consumer_holder else None
|
||||
if not (_sc and getattr(_sc, "_use_native_streaming", False)):
|
||||
return True
|
||||
_cancelled_flag = None
|
||||
try:
|
||||
_boundary_result = _sc.close_for_approval_prompt(
|
||||
_placeholder, reason=_reason, reopen=_reopen,
|
||||
)
|
||||
# Returns (future, cancelled_flag) or just a future.
|
||||
if isinstance(_boundary_result, tuple):
|
||||
_boundary_future, _cancelled_flag = _boundary_result
|
||||
else:
|
||||
_boundary_future = _boundary_result
|
||||
if hasattr(_boundary_future, "result"):
|
||||
_ok = _boundary_future.result(timeout=10)
|
||||
if not _ok:
|
||||
logger.warning(
|
||||
"%s boundary failed to close stream properly — "
|
||||
"prompt may still appear in typing bubble", _reason,
|
||||
)
|
||||
return bool(_ok)
|
||||
return True
|
||||
except (TimeoutError, Exception) as _boundary_err:
|
||||
if _cancelled_flag is not None:
|
||||
_cancelled_flag["cancelled"] = True
|
||||
logger.warning(
|
||||
"%s boundary timed out or failed: %s", _reason, _boundary_err,
|
||||
)
|
||||
return False
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Clarify callback: present a clarify prompt and block on a response. Runs on the agent's
|
||||
# worker thread (clarify_tool's synchronous contract): schedules the adapter's send_clarify
|
||||
# on the gateway loop, then blocks on the primitive's threading.Event with a timeout.
|
||||
# Returns the response string, or a sentinel explaining no response arrived.
|
||||
# ------------------------------------------------------------------
|
||||
def _clarify_callback_sync(self, question: str, choices, multi_select: bool = False) -> str:
|
||||
ctx = self._ctx
|
||||
from tools import clarify_gateway as _clarify_mod
|
||||
import uuid as _uuid
|
||||
|
||||
if not ctx._status_adapter:
|
||||
return ""
|
||||
|
||||
clarify_id = _uuid.uuid4().hex[:10]
|
||||
_clarify_mod.register(
|
||||
clarify_id=clarify_id,
|
||||
session_key=ctx.session_key or "",
|
||||
question=question,
|
||||
choices=list(choices) if choices else None,
|
||||
multi_select=bool(multi_select),
|
||||
)
|
||||
|
||||
# WeCom native streaming: finalize the current stream before the clarify prompt so the
|
||||
# post-answer output opens a fresh bubble below the question ("气泡割裂" otherwise). Unlike
|
||||
# approval, clarify passes reopen=True so the continuation re-opens a native stream; if
|
||||
# the re-seed fails the consumer degrades to send() automatically.
|
||||
self._close_native_stream_boundary(
|
||||
"Clarify", "💬 等待你的选择...", _reopen=True,
|
||||
)
|
||||
|
||||
# Pause typing — as with approval, a "thinking..." status must not obscure the prompt or
|
||||
# block an "Other" reply on platforms that disable input while typing (Slack Assistant).
|
||||
with suppress(Exception):
|
||||
ctx._status_adapter.pause_typing_for_chat(ctx._status_chat_id)
|
||||
|
||||
# Ordering barrier: flush buffered assistant prose to the platform BEFORE sending the
|
||||
# poll, which goes out on a separate agent-thread-blocking path and would otherwise
|
||||
# render ABOVE its own explanation. Best-effort + short timeout so the agent thread
|
||||
# never hangs if the consumer task isn't running.
|
||||
try:
|
||||
_sc = ctx.stream_consumer_holder[0] if ctx.stream_consumer_holder else None
|
||||
_flush = getattr(_sc, "flush_pending_sync", None)
|
||||
if callable(_flush):
|
||||
_flush(timeout=3.0)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Stream-consumer flush before clarify prompt failed",
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
fut = safe_schedule_threadsafe(
|
||||
ctx._status_adapter.send_clarify(
|
||||
chat_id=ctx._status_chat_id,
|
||||
question=question,
|
||||
choices=list(choices) if choices else None,
|
||||
clarify_id=clarify_id,
|
||||
session_key=ctx.session_key or "",
|
||||
metadata=ctx._status_thread_metadata,
|
||||
),
|
||||
ctx._loop_for_step,
|
||||
logger=logger,
|
||||
log_message="Clarify send failed to schedule",
|
||||
)
|
||||
# Boundary rule (see _approval_send_outcome): a send timeout is AMBIGUOUS — the card may
|
||||
# have posted with a late ack. Only a definitive failure tears down the registration;
|
||||
# ambiguous falls through to the bounded wait so a late reply resolves.
|
||||
_clarify_response = _clarify_send_then_wait(
|
||||
fut,
|
||||
clarify_id=clarify_id,
|
||||
session_key=ctx.session_key or "",
|
||||
clarify_mod=_clarify_mod,
|
||||
)
|
||||
# Only re-arm typing when the user actually answered — the undeliverable sentinel and the
|
||||
# timeout/cancellation strings start with '[' and must pass through untouched.
|
||||
if not (
|
||||
isinstance(_clarify_response, str)
|
||||
and _clarify_response.startswith("[")
|
||||
):
|
||||
# User answered: reopen typing IMMEDIATELY, not on the LLM's first post-answer token
|
||||
# (native streaming otherwise re-seeds lazily on the first delta: ~48s of dead air).
|
||||
# request_reopen_seed is a no-op outside the reopen-pending native state; always safe.
|
||||
_sc_reopen = ctx.stream_consumer_holder[0] if ctx.stream_consumer_holder else None
|
||||
if _sc_reopen is not None:
|
||||
try:
|
||||
_sc_reopen.request_reopen_seed()
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"request_reopen_seed after clarify answer failed",
|
||||
exc_info=True,
|
||||
)
|
||||
try:
|
||||
ctx._status_adapter.resume_typing_for_chat(ctx._status_chat_id)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"resume_typing_for_chat after clarify answer failed",
|
||||
exc_info=True,
|
||||
)
|
||||
return _clarify_response
|
||||
|
||||
def _load_turn_history(self, agent, reused_cached_agent):
|
||||
ctx = self._ctx
|
||||
# Convert history to agent format. Transcript path: {role, content, timestamp} dicts — strip
|
||||
# timestamps. Interrupt path (agent result["messages"]): full agent messages with
|
||||
# tool_calls/tool_call_id/reasoning — pass through intact so the API sees valid assistant→tool
|
||||
@@ -5710,112 +5665,107 @@ class TurnRunner:
|
||||
# Collect MEDIA paths already in history to exclude them from this turn's extraction.
|
||||
# Compression-safe: even if the message list shrinks, we know which paths are old.
|
||||
_history_media_paths: set = _collect_history_media_paths(agent_history)
|
||||
return agent_history, observed_group_context, _history_media_paths
|
||||
|
||||
# Per-session gateway approval callback: dangerous-command approval blocks the agent thread
|
||||
# (mirrors CLI input()); the callback bridges sync→async to send the request immediately.
|
||||
from tools.approval import (
|
||||
register_gateway_notify,
|
||||
reset_current_session_key,
|
||||
set_current_session_key,
|
||||
unregister_gateway_notify,
|
||||
)
|
||||
|
||||
def _approval_notify_sync(approval_data: dict) -> None:
|
||||
"""Send the approval request to the user from the agent thread.
|
||||
def _approval_notify_sync(self, approval_data: dict) -> None:
|
||||
"""Send the approval request to the user from the agent thread.
|
||||
|
||||
Uses the adapter's interactive button approvals (e.g. ``send_exec_approval``) when
|
||||
available, else a plain text message with ``/approve`` instructions.
|
||||
"""
|
||||
# Pause typing while awaiting approval: Slack's assistant_threads_setStatus disables the
|
||||
# compose box, so the user can't type /approve while "is thinking..." shows. The approval
|
||||
# send auto-clears it; pausing stops _keep_typing re-setting it. Resumed in approve/deny.
|
||||
ctx._status_adapter.pause_typing_for_chat(ctx._status_chat_id)
|
||||
ctx = self._ctx
|
||||
# Pause typing while awaiting approval: Slack's assistant_threads_setStatus disables the
|
||||
# compose box, so the user can't type /approve while "is thinking..." shows. The approval
|
||||
# send auto-clears it; pausing stops _keep_typing re-setting it. Resumed in approve/deny.
|
||||
ctx._status_adapter.pause_typing_for_chat(ctx._status_chat_id)
|
||||
|
||||
# WeCom native streaming: ask the stream consumer to close the current stream before the
|
||||
# approval prompt — via the consumer's queue, so it serializes with pending deltas.
|
||||
_close_native_stream_boundary("Approval")
|
||||
# WeCom native streaming: ask the stream consumer to close the current stream before the
|
||||
# approval prompt — via the consumer's queue, so it serializes with pending deltas.
|
||||
self._close_native_stream_boundary("Approval")
|
||||
|
||||
cmd = approval_data.get("command", "")
|
||||
desc = approval_data.get("description", "dangerous command")
|
||||
cmd = approval_data.get("command", "")
|
||||
desc = approval_data.get("description", "dangerous command")
|
||||
|
||||
# Redact credentials from the command before display — Tirith's findings are already
|
||||
# redacted, but the raw command string still leaks secrets to the chat platform. Done
|
||||
# here so BOTH the button-based and plain-text fallback paths use the redacted value.
|
||||
cmd = _redact_approval_command(cmd)
|
||||
# Redact credentials from the command before display — Tirith's findings are already
|
||||
# redacted, but the raw command string still leaks secrets to the chat platform. Done
|
||||
# here so BOTH the button-based and plain-text fallback paths use the redacted value.
|
||||
cmd = _redact_approval_command(cmd)
|
||||
|
||||
# Prefer button-based approval when the adapter supports it. Check the *class*, not the
|
||||
# instance — avoids false positives from MagicMock auto-attribute creation in tests.
|
||||
if getattr(type(ctx._status_adapter), "send_exec_approval", None) is not None:
|
||||
try:
|
||||
_approval_fut = safe_schedule_threadsafe(
|
||||
ctx._status_adapter.send_exec_approval(
|
||||
chat_id=ctx._status_chat_id,
|
||||
command=cmd,
|
||||
session_key=_approval_session_key,
|
||||
description=desc,
|
||||
metadata=ctx._status_thread_metadata,
|
||||
allow_permanent=approval_data.get("allow_permanent", True),
|
||||
allow_session=approval_data.get("allow_session", True),
|
||||
smart_denied=approval_data.get("smart_denied", False),
|
||||
),
|
||||
ctx._loop_for_step,
|
||||
logger=logger,
|
||||
log_message="send_exec_approval scheduling error",
|
||||
)
|
||||
if _approval_fut is None:
|
||||
raise RuntimeError("send_exec_approval: loop unavailable")
|
||||
_outcome = _approval_send_outcome(_approval_fut, timeout=15)
|
||||
if _outcome == "sent":
|
||||
return
|
||||
if _outcome == "ambiguous":
|
||||
# Timeout ≠ failure: the card may have posted with a late ack (slow API or
|
||||
# backpressure). The prompt registration stays alive so a tap still resolves;
|
||||
# re-sending made duplicate cards + orphaned "/approve: nothing pending". Skip.
|
||||
logger.warning(
|
||||
"Button-based approval send timed out — treating "
|
||||
"as possibly-delivered (no re-send; the prompt "
|
||||
"stays armed for a late tap)"
|
||||
)
|
||||
return
|
||||
logger.warning(
|
||||
"Button-based approval failed (send returned error), falling back to text"
|
||||
)
|
||||
except Exception as _e:
|
||||
logger.warning(
|
||||
"Button-based approval failed, falling back to text: %s", _e
|
||||
)
|
||||
|
||||
# Fallback: plain-text approval prompt with the adapter's typed prefix (e.g. `!approve`) —
|
||||
# typed "/" is blocked in Slack threads and reserved by Matrix clients.
|
||||
_p = getattr(ctx._status_adapter, "typed_command_prefix", "/")
|
||||
msg = _format_exec_approval_fallback(
|
||||
cmd,
|
||||
desc,
|
||||
_p,
|
||||
allow_permanent=approval_data.get("allow_permanent", True),
|
||||
allow_session=approval_data.get("allow_session", True),
|
||||
smart_denied=approval_data.get("smart_denied", False),
|
||||
)
|
||||
# Prefer button-based approval when the adapter supports it. Check the *class*, not the
|
||||
# instance — avoids false positives from MagicMock auto-attribute creation in tests.
|
||||
if getattr(type(ctx._status_adapter), "send_exec_approval", None) is not None:
|
||||
try:
|
||||
# Mark as approval prompt so WeCom routes through control lane
|
||||
_approval_metadata = dict(ctx._status_thread_metadata or {})
|
||||
_approval_metadata["is_approval_prompt"] = True
|
||||
|
||||
_approval_send_fut = safe_schedule_threadsafe(
|
||||
ctx._status_adapter.send(
|
||||
ctx._status_chat_id,
|
||||
msg,
|
||||
metadata=_interim_metadata(_approval_metadata),
|
||||
_approval_fut = safe_schedule_threadsafe(
|
||||
ctx._status_adapter.send_exec_approval(
|
||||
chat_id=ctx._status_chat_id,
|
||||
command=cmd,
|
||||
session_key=ctx.session_key or "",
|
||||
description=desc,
|
||||
metadata=ctx._status_thread_metadata,
|
||||
allow_permanent=approval_data.get("allow_permanent", True),
|
||||
allow_session=approval_data.get("allow_session", True),
|
||||
smart_denied=approval_data.get("smart_denied", False),
|
||||
),
|
||||
ctx._loop_for_step,
|
||||
logger=logger,
|
||||
log_message="Approval text-send scheduling error",
|
||||
log_message="send_exec_approval scheduling error",
|
||||
)
|
||||
if _approval_fut is None:
|
||||
raise RuntimeError("send_exec_approval: loop unavailable")
|
||||
_outcome = _approval_send_outcome(_approval_fut, timeout=15)
|
||||
if _outcome == "sent":
|
||||
return
|
||||
if _outcome == "ambiguous":
|
||||
# Timeout ≠ failure: the card may have posted with a late ack (slow API or
|
||||
# backpressure). The prompt registration stays alive so a tap still resolves;
|
||||
# re-sending made duplicate cards + orphaned "/approve: nothing pending". Skip.
|
||||
logger.warning(
|
||||
"Button-based approval send timed out — treating "
|
||||
"as possibly-delivered (no re-send; the prompt "
|
||||
"stays armed for a late tap)"
|
||||
)
|
||||
return
|
||||
logger.warning(
|
||||
"Button-based approval failed (send returned error), falling back to text"
|
||||
)
|
||||
if _approval_send_fut is not None:
|
||||
_approval_send_fut.result(timeout=15)
|
||||
except Exception as _e:
|
||||
logger.error("Failed to send approval request: %s", _e)
|
||||
logger.warning(
|
||||
"Button-based approval failed, falling back to text: %s", _e
|
||||
)
|
||||
|
||||
# Fallback: plain-text approval prompt with the adapter's typed prefix (e.g. `!approve`) —
|
||||
# typed "/" is blocked in Slack threads and reserved by Matrix clients.
|
||||
_p = getattr(ctx._status_adapter, "typed_command_prefix", "/")
|
||||
msg = _format_exec_approval_fallback(
|
||||
cmd,
|
||||
desc,
|
||||
_p,
|
||||
allow_permanent=approval_data.get("allow_permanent", True),
|
||||
allow_session=approval_data.get("allow_session", True),
|
||||
smart_denied=approval_data.get("smart_denied", False),
|
||||
)
|
||||
try:
|
||||
# Mark as approval prompt so WeCom routes through control lane
|
||||
_approval_metadata = dict(ctx._status_thread_metadata or {})
|
||||
_approval_metadata["is_approval_prompt"] = True
|
||||
|
||||
_approval_send_fut = safe_schedule_threadsafe(
|
||||
ctx._status_adapter.send(
|
||||
ctx._status_chat_id,
|
||||
msg,
|
||||
metadata=_interim_metadata(_approval_metadata),
|
||||
),
|
||||
ctx._loop_for_step,
|
||||
logger=logger,
|
||||
log_message="Approval text-send scheduling error",
|
||||
)
|
||||
if _approval_send_fut is not None:
|
||||
_approval_send_fut.result(timeout=15)
|
||||
except Exception as _e:
|
||||
logger.error("Failed to send approval request: %s", _e)
|
||||
|
||||
def _prepare_turn_message(self, agent_history):
|
||||
ctx = self._ctx
|
||||
# Keep real user text separate from API-only recovery guidance: if an auto-continue note is
|
||||
# prepended below, persist the original so stale guidance never replays as user text.
|
||||
_persist_user_message_override: Optional[Any] = ctx.persist_user_message
|
||||
@@ -5918,10 +5868,25 @@ class TurnRunner:
|
||||
getattr(_sn_adapter, "interactive_resume", True)
|
||||
),
|
||||
)
|
||||
return _persist_user_message_override, _persist_user_timestamp_override
|
||||
|
||||
def _run_conversation_with_approval(
|
||||
self, agent, agent_history, observed_group_context,
|
||||
_persist_user_message_override, _persist_user_timestamp_override,
|
||||
):
|
||||
ctx = self._ctx
|
||||
# Per-session gateway approval callback: dangerous-command approval blocks the agent thread
|
||||
# (mirrors CLI input()); the callback bridges sync→async to send the request immediately.
|
||||
from tools.approval import (
|
||||
register_gateway_notify,
|
||||
reset_current_session_key,
|
||||
set_current_session_key,
|
||||
unregister_gateway_notify,
|
||||
)
|
||||
|
||||
_approval_session_key = ctx.session_key or ""
|
||||
_approval_session_token = set_current_session_key(_approval_session_key)
|
||||
register_gateway_notify(_approval_session_key, _approval_notify_sync)
|
||||
register_gateway_notify(_approval_session_key, self._approval_notify_sync)
|
||||
try:
|
||||
# If _prepare_inbound_message_text buffered image paths for native attachment, wrap the
|
||||
# user turn as an OpenAI-style multimodal content list. Consume-and-clear so subsequent
|
||||
@@ -5992,6 +5957,10 @@ class TurnRunner:
|
||||
except Exception:
|
||||
pass
|
||||
reset_current_session_key(_approval_session_token)
|
||||
return result
|
||||
|
||||
def _finish_stream_consumer(self, result, agent_history, _stream_consumer):
|
||||
ctx = self._ctx
|
||||
# Canonicalize a model-emitted computer-use screenshot path at the common result boundary: the
|
||||
# streaming finalizer below and the non-streaming delivery path must see the same response;
|
||||
# repairing only in later media scanning leaves streaming a mangled path + rejected attachment.
|
||||
@@ -6035,25 +6004,8 @@ class TurnRunner:
|
||||
else:
|
||||
_stream_consumer.finish()
|
||||
|
||||
# Signal the streaming-TTS consumer that the agent is done. finish() runs on the outer
|
||||
# event-loop thread after the executor returns, so early run_sync returns are also finalised.
|
||||
|
||||
# Return final response, or a message if something went wrong
|
||||
final_response = result.get("final_response")
|
||||
|
||||
# Extract actual token counts from the agent instance used for this run
|
||||
_last_prompt_toks = 0
|
||||
_input_toks = 0
|
||||
_output_toks = 0
|
||||
_context_length = 0
|
||||
_agent = ctx.agent_holder[0]
|
||||
if _agent and hasattr(_agent, "context_compressor"):
|
||||
_last_prompt_toks = getattr(_agent.context_compressor, "last_prompt_tokens", 0)
|
||||
_input_toks = getattr(_agent, "session_prompt_tokens", 0)
|
||||
_output_toks = getattr(_agent, "session_completion_tokens", 0)
|
||||
_context_length = getattr(_agent.context_compressor, "context_length", 0) or 0
|
||||
_resolved_model = getattr(_agent, "model", None) if _agent else None
|
||||
|
||||
def _sync_session_after_run(self, agent_history):
|
||||
ctx = self._ctx
|
||||
# Sync session_id right after run_conversation(): compression can rotate before a follow-up
|
||||
# model call fails, and the failure return below must still point at the compressed child.
|
||||
agent = ctx.agent_holder[0]
|
||||
@@ -6142,6 +6094,117 @@ class TurnRunner:
|
||||
_effective_history_offset = (
|
||||
0 if (_session_was_split or _compacted_in_place) else len(agent_history)
|
||||
)
|
||||
return _compacted_in_place, effective_session_id, _effective_history_offset
|
||||
|
||||
def run_sync(self):
|
||||
ctx = self._ctx
|
||||
# As a method the turn message lives on the shared TurnContext: every rebind writes
|
||||
# `ctx.message`, so the outer `_run_agent_inner` body sees the update as via the closure cell.
|
||||
|
||||
# session_key propagates via contextvars (_set_session_env / set_current_session_key):
|
||||
# concurrency-safe and inherited by tool worker threads. Deliberately do NOT write
|
||||
# os.environ["HERMES_SESSION_KEY"]: it is process-global, so concurrent sessions would clobber
|
||||
# each other and a tool thread with an unset contextvar would read the wrong key, misrouting
|
||||
# approvals. Only the TUI slash-worker subprocess exports the env var (from its own argv).
|
||||
|
||||
# Map platform enum to the platform hint key the agent understands.
|
||||
# Platform.LOCAL ("local") maps to "cli"; others pass through as-is.
|
||||
platform_key = "cli" if ctx.source.platform == Platform.LOCAL else ctx.source.platform.value
|
||||
|
||||
# Combine platform context, YAML channel_prompts hint for this chat, channel_overrides
|
||||
# system_prompt (or global ephemeral), and the gateway ephemeral prompt.
|
||||
combined_ephemeral = ctx.context_prompt or ""
|
||||
event_channel_prompt = (ctx.channel_prompt or "").strip()
|
||||
if event_channel_prompt:
|
||||
combined_ephemeral = (combined_ephemeral + "\n\n" + event_channel_prompt).strip()
|
||||
cfg_channel_prompt = self._runner._get_system_prompt_for_channel(
|
||||
ctx.source.platform,
|
||||
ctx.source.chat_id or "",
|
||||
thread_id=getattr(ctx.source, "thread_id", None),
|
||||
parent_id=getattr(ctx.source, "parent_chat_id", None),
|
||||
)
|
||||
if cfg_channel_prompt:
|
||||
combined_ephemeral = (combined_ephemeral + "\n\n" + cfg_channel_prompt).strip()
|
||||
|
||||
max_iterations = _current_max_iterations()
|
||||
|
||||
try:
|
||||
model, runtime_kwargs = self._runner._resolve_session_agent_runtime(
|
||||
source=ctx.source,
|
||||
session_key=ctx.session_key,
|
||||
user_config=ctx.user_config,
|
||||
)
|
||||
logger.debug(
|
||||
"run_agent resolved: model=%s provider=%s session=%s",
|
||||
model, runtime_kwargs.get("provider"), ctx.session_key or "",
|
||||
)
|
||||
except Exception as exc:
|
||||
return {
|
||||
"final_response": f"⚠️ Provider authentication failed: {exc}",
|
||||
"messages": [],
|
||||
"api_calls": 0,
|
||||
"tools": [],
|
||||
}
|
||||
|
||||
pr = self._runner._provider_routing
|
||||
reasoning_config = self._runner._resolve_session_reasoning_config(
|
||||
source=ctx.source,
|
||||
session_key=ctx.session_key,
|
||||
model=model,
|
||||
)
|
||||
self._runner._reasoning_config = reasoning_config
|
||||
self._runner._service_tier = self._runner._resolve_session_service_tier(
|
||||
source=ctx.source, session_key=ctx.session_key
|
||||
)
|
||||
(
|
||||
_stream_consumer,
|
||||
_stream_delta_cb,
|
||||
_interim_assistant_cb,
|
||||
_want_interim_messages,
|
||||
) = self._setup_stream_consumer(platform_key)
|
||||
|
||||
turn_route = self._runner._resolve_turn_agent_config(ctx.message, model, runtime_kwargs)
|
||||
agent, reused_cached_agent = self._resolve_turn_agent(
|
||||
turn_route, platform_key, combined_ephemeral, max_iterations, reasoning_config, pr,
|
||||
)
|
||||
self._wire_turn_agent_callbacks(
|
||||
agent, turn_route, reasoning_config,
|
||||
_stream_delta_cb, _interim_assistant_cb, _want_interim_messages,
|
||||
)
|
||||
agent_history, observed_group_context, _history_media_paths = (
|
||||
self._load_turn_history(agent, reused_cached_agent)
|
||||
)
|
||||
_persist_user_message_override, _persist_user_timestamp_override = (
|
||||
self._prepare_turn_message(agent_history)
|
||||
)
|
||||
result = self._run_conversation_with_approval(
|
||||
agent, agent_history, observed_group_context,
|
||||
_persist_user_message_override, _persist_user_timestamp_override,
|
||||
)
|
||||
self._finish_stream_consumer(result, agent_history, _stream_consumer)
|
||||
|
||||
# Signal the streaming-TTS consumer that the agent is done. finish() runs on the outer
|
||||
# event-loop thread after the executor returns, so early run_sync returns are also finalised.
|
||||
|
||||
# Return final response, or a message if something went wrong
|
||||
final_response = result.get("final_response")
|
||||
|
||||
# Extract actual token counts from the agent instance used for this run
|
||||
_last_prompt_toks = 0
|
||||
_input_toks = 0
|
||||
_output_toks = 0
|
||||
_context_length = 0
|
||||
_agent = ctx.agent_holder[0]
|
||||
if _agent and hasattr(_agent, "context_compressor"):
|
||||
_last_prompt_toks = getattr(_agent.context_compressor, "last_prompt_tokens", 0)
|
||||
_input_toks = getattr(_agent, "session_prompt_tokens", 0)
|
||||
_output_toks = getattr(_agent, "session_completion_tokens", 0)
|
||||
_context_length = getattr(_agent.context_compressor, "context_length", 0) or 0
|
||||
_resolved_model = getattr(_agent, "model", None) if _agent else None
|
||||
|
||||
_compacted_in_place, effective_session_id, _effective_history_offset = (
|
||||
self._sync_session_after_run(agent_history)
|
||||
)
|
||||
|
||||
if not final_response:
|
||||
final_response = _normalize_empty_agent_response(
|
||||
|
||||
Reference in New Issue
Block a user