wehub-resource-sync
1823 lines
81 KiB
Python
1823 lines
81 KiB
Python
"""
|
|
Shared adapter that wraps any inner :class:`Executor` instance as
|
|
a :class:`HarnessApp` subclass.
|
|
|
|
The four per-harness wraps (claude-sdk, codex, pi,
|
|
openai-agents-sdk) all use this adapter — they only differ in the
|
|
inner ``Executor`` they construct. The adapter handles:
|
|
|
|
- Per-conversation lazy executor construction (Layer 1 state on
|
|
the adapter instance).
|
|
- Per-turn translation of Omnigent :class:`CreateResponseRequest` →
|
|
inner :class:`Message` list + :class:`ExecutorConfig`.
|
|
- Per-turn translation of inner :class:`ExecutorEvent`s →
|
|
typed Omnigent SSE events emitted via :meth:`TurnContext.emit`.
|
|
- Forwarding ``request.tools`` and ``request.instructions`` to
|
|
the inner Executor; wiring a ``_tool_executor`` callback so
|
|
the inner SDK round-trips spec-declared tools through
|
|
:meth:`TurnContext.dispatch_tool` (the scaffold's
|
|
action_required path).
|
|
- Cancellation propagation (POST /cancel → ctx.cancelled →
|
|
inner ``Executor.interrupt_session``).
|
|
- Per-conversation cleanup on shutdown.
|
|
|
|
V1 limitations (documented in §Autonomous decisions):
|
|
|
|
- **No per-conversation executor configuration via spec.** The
|
|
per-harness wrap's executor factory determines configuration
|
|
(cwd, model, sandbox, credentials) at construction time. A
|
|
follow-up should thread spec config through the request body
|
|
or via subprocess env vars.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
import logging
|
|
import secrets
|
|
import uuid
|
|
from collections import deque
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
from fastapi import Response
|
|
|
|
from omnigent.errors import ElicitationDeclinedError
|
|
from omnigent.inner.executor import (
|
|
CompactionComplete,
|
|
Executor,
|
|
ExecutorConfig,
|
|
ExecutorError,
|
|
ExecutorEvent,
|
|
Message,
|
|
ReasoningChunk,
|
|
TextChunk,
|
|
ToolCallComplete,
|
|
ToolCallRequest,
|
|
TurnCancelled,
|
|
TurnComplete,
|
|
)
|
|
from omnigent.inner.tracing import TracingContext, is_tracing_enabled
|
|
from omnigent.policies.types import FAIL_CLOSED_PHASES
|
|
from omnigent.runtime.harnesses._scaffold import HarnessApp, PolicyVerdictPayload, TurnContext
|
|
from omnigent.runtime.tool_output import cap_tool_output
|
|
from omnigent.server.schemas import (
|
|
CreateResponseRequest,
|
|
ElicitationRequestParams,
|
|
InjectionConsumedEvent,
|
|
OutputItemDoneEvent,
|
|
OutputTextDeltaEvent,
|
|
ReasoningStartedEvent,
|
|
ReasoningSummaryTextDeltaEvent,
|
|
ReasoningTextDeltaEvent,
|
|
)
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
# Status string the per-harness wraps emit on observed
|
|
# function_call items (i.e., tools the inner SDK already executed
|
|
# natively). Distinct from ``"action_required"`` which is what
|
|
# server-dispatched tool calls use — see §Sub-agent representation
|
|
# in designs/SERVER_HARNESS_CONTRACT.md.
|
|
_OBSERVED_TOOL_CALL_STATUS = "in_progress"
|
|
|
|
|
|
# Prefix the Claude SDK applies to MCP-registered tool names
|
|
# (e.g. ``mcp__omnigent__sys_terminal_launch``). Tools whose
|
|
# name starts with this prefix round-trip through the inner
|
|
# executor's ``_tool_executor`` callback -> :func:`_bridge_one_dispatch`
|
|
# -> ``ctx.dispatch_tool``, which emits an action_required
|
|
# function_call event POST-STREAM (the SDK's MCP-server handler
|
|
# fires after the assistant message finishes streaming).
|
|
#
|
|
# Adapter strategy for MCP tools:
|
|
#
|
|
# 1. Emit the observed function_call event INLINE (when the
|
|
# inner SDK yields ToolCallRequest). That's what gives the
|
|
# REPL the ``⏵ tool_name`` line interleaved with text rather
|
|
# than bunched at the end of the turn.
|
|
# 2. Queue the SDK's ``tool_use_id`` so the post-stream
|
|
# dispatch uses the SAME ``call_id`` (see
|
|
# ``self._pending_mcp_call_ids``). Correlated ids let the
|
|
# SDK client's ``BlockStream`` dedupe the action_required
|
|
# event against the inline observed event — single render,
|
|
# no duplicates.
|
|
# 3. Skip the observed function_call_output emission on
|
|
# ToolCallComplete — the dispatch's PATCH handler emits the
|
|
# paired output. Keeps the dedup story symmetric.
|
|
_MCP_TOOL_NAME_PREFIX = "mcp__"
|
|
|
|
|
|
def _strip_mcp_tool_prefix(name: str) -> str:
|
|
"""
|
|
Strip the Claude SDK MCP tool prefix from a tool name.
|
|
|
|
The Claude SDK names MCP tools as ``mcp__{server}__{tool}``
|
|
(e.g. ``mcp__omnigent__sys_terminal_launch``). The bare
|
|
name (``sys_terminal_launch``) is what the Omnigent wire shape
|
|
and persisted conversation items carry — kept in sync with
|
|
:func:`omnigent.runtime.workflow._observed_tool_call_sse_dicts`
|
|
and :func:`_build_observed_tool_items` so the SSE name, the
|
|
store-item name, and the adapter's emitted name all agree.
|
|
|
|
Mirrors :func:`omnigent.runtime.workflow._strip_mcp_tool_prefix`
|
|
— kept local here to avoid a cross-module import that
|
|
would tighten the dependency between the harness adapter
|
|
and the workflow package; both modules need this helper for
|
|
different consumers.
|
|
|
|
:param name: Tool name, possibly MCP-prefixed.
|
|
:returns: The bare tool name. Only ``mcp__<server>__<tool>``
|
|
shapes are stripped; bare names with ``__`` pass through.
|
|
"""
|
|
if name.startswith("mcp__"):
|
|
parts = name.split("__", 2)
|
|
if len(parts) == 3:
|
|
return parts[2]
|
|
return name
|
|
|
|
|
|
class ExecutorAdapter(HarnessApp):
|
|
"""
|
|
:class:`HarnessApp` subclass that drives any inner
|
|
:class:`Executor` instance.
|
|
|
|
The per-harness wrap supplies an ``executor_factory`` —
|
|
a zero-arg callable that constructs the inner executor. The
|
|
adapter calls it lazily on the first turn (so heavyweight
|
|
constructors like :class:`ClaudeSDKExecutor`'s eager
|
|
Databricks credential resolution don't fire at FastAPI
|
|
boot, only when a real conversation starts).
|
|
|
|
The factory's return value is cached as Layer 1 state on the
|
|
adapter instance — reused across turns for the conversation's
|
|
lifetime. ``shutdown()`` calls ``executor.close()`` on
|
|
teardown.
|
|
|
|
:param executor_factory: Zero-arg callable returning a fresh
|
|
:class:`Executor`. Tests pass a ``lambda:
|
|
MockExecutor()``; production wraps pass a lambda that
|
|
constructs the real per-harness Executor with config
|
|
appropriate to the spec, e.g. ``lambda:
|
|
ClaudeSDKExecutor(databricks=True, databricks_profile="<your-profile>")``.
|
|
:param session_key: Stable identifier the inner executor uses
|
|
to scope per-session state (clients, subprocesses). The
|
|
adapter passes this to ``executor.close_session()`` and
|
|
``executor.interrupt_session()``. Defaults to a uuid hex
|
|
— production wraps may want to set it from the
|
|
``conversation_id`` once the runner exposes it on
|
|
``app.state``.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
executor_factory: Callable[[], Executor],
|
|
session_key: str | None = None,
|
|
harness_label: str = "Claude",
|
|
) -> None:
|
|
super().__init__()
|
|
self._executor_factory = executor_factory
|
|
self._session_key = session_key or uuid.uuid4().hex
|
|
# Human-readable harness name used in elicitation card messages,
|
|
# e.g. "Claude" → "Claude wants to call **Bash**" or
|
|
# "Cursor" → "Cursor wants to use **Bash**".
|
|
self._harness_label = harness_label
|
|
# Layer 1: lazily-constructed inner executor, reused
|
|
# across turns for the conversation's lifetime.
|
|
self._executor: Executor | None = None
|
|
# Per-turn pointer to the active :class:`TurnContext`,
|
|
# rebound at the top of each :meth:`run_turn` and cleared
|
|
# in its finally. The stable ``_tool_executor`` bridge
|
|
# (installed once on first use) reads from this slot so
|
|
# the MCP handlers cached inside ``ClaudeSDKClient`` —
|
|
# which closure-capture whatever ``_tool_executor`` was
|
|
# set on the FIRST turn — always dispatch into the
|
|
# CURRENT turn's ctx. Without this, turn N>1's tool calls
|
|
# park a Future inside turn 1's already-dead ctx and the
|
|
# SDK hangs forever waiting for a result that nobody can
|
|
# deliver.
|
|
self._current_ctx: TurnContext | None = None
|
|
self._current_agent: str | None = None
|
|
# FIFO queue of inner-SDK tool-use ids, one entry per
|
|
# ToolCallRequest the executor parses. Populated by
|
|
# :meth:`_translate_event` whenever ``event.metadata``
|
|
# carries a ``call_id`` (the inner Claude SDK / OpenAI
|
|
# Agents SDK / Codex / Pi executors all stamp it for
|
|
# bridged tools). Drained by :meth:`_stable_tool_executor`
|
|
# so each :func:`_bridge_one_dispatch` call reuses the
|
|
# SAME ``call_id`` the observed function_call event
|
|
# already carried — that's what lets the Omnigent REPL dedupe
|
|
# the inline observed render with the post-stream
|
|
# action_required render. Without this correlation, the
|
|
# two events have different uuids, the SDK client sees
|
|
# them as separate tool calls, and the REPL renders
|
|
# ``⏵ tool_name`` twice plus an empty result panel for
|
|
# the orphan call (the 2026-04-29 user-reported
|
|
# duplicate-render + empty-box regression for
|
|
# openai-agents — same shape as the 2026-04-28 claude-
|
|
# sdk regression resolved for MCP-prefixed
|
|
# tools). See commit 989bfde for the prior
|
|
# suppress-observed mitigation that introduced the
|
|
# end-of-turn ordering regression this queue resolves.
|
|
self._pending_mcp_call_ids: deque[str] = deque()
|
|
# Per-session tracing context. Created lazily on the first
|
|
# turn when tracing is enabled; reused across turns so the
|
|
# span parent chain stays rooted on the session's executor.
|
|
self._tracing_ctx: TracingContext | None = None
|
|
# Call ids actually round-tripped through :meth:`_stable_tool_executor`
|
|
# -> ``ctx.dispatch_tool`` this turn. ``dispatch_tool`` emits the paired
|
|
# function_call_output itself, so a ToolCallComplete carrying one of
|
|
# these ids must be SUPPRESSED in :meth:`_translate_event` to avoid a
|
|
# duplicate output card. Completions whose id is NOT here come from
|
|
# tools the inner SDK ran entirely internally (e.g. the antigravity
|
|
# executor, which has no ``_stable_tool_executor`` dispatch) — those are
|
|
# the ONLY output source and MUST be emitted. Reset per turn alongside
|
|
# ``_pending_mcp_call_ids``.
|
|
self._dispatched_call_ids: set[str] = set()
|
|
|
|
async def run_turn(self, request: CreateResponseRequest, ctx: TurnContext) -> None:
|
|
"""
|
|
Drive the inner executor and translate its events.
|
|
|
|
Lazily constructs the inner executor on the first call.
|
|
Subsequent calls reuse the cached instance. Wires a
|
|
per-turn ``_tool_executor`` callback so the inner SDK
|
|
can round-trip spec-declared tools through Omnigent via the
|
|
scaffold's ``dispatch_tool`` (action_required) path.
|
|
The hook is reset to ``None`` after the turn so cross-
|
|
turn references can't accidentally fire stale dispatch
|
|
contexts.
|
|
|
|
:param request: Decoded :class:`CreateResponseRequest`
|
|
synthesized by the harness scaffold for this turn.
|
|
:param ctx: Per-turn :class:`TurnContext` from the
|
|
scaffold.
|
|
"""
|
|
executor = self._ensure_executor()
|
|
messages = _translate_input_to_messages(request.input)
|
|
# Stamp every message with the adapter's session_key so
|
|
# the inner :class:`ClaudeSDKExecutor` keys its cached
|
|
# SDK client under THAT key. Without this stamp, the
|
|
# executor's :meth:`_session_key` falls back to
|
|
# ``"default"`` (no ``session_id`` on any message), and
|
|
# subsequent :meth:`Executor.enqueue_session_message`
|
|
# calls — which use ``self._session_key`` from this
|
|
# adapter — return ``False`` because the cached client
|
|
# is under a DIFFERENT key. Symptom (the user reported
|
|
# this): mid-turn steering arrives in the harness, the
|
|
# adapter forwards to enqueue_session_message, but the
|
|
# SDK never sees the new prompt and Claude keeps going.
|
|
for message in messages:
|
|
message["session_id"] = self._session_key
|
|
extra: dict[str, Any] = {}
|
|
if request.reasoning is not None:
|
|
effort = request.reasoning.get("effort")
|
|
if effort:
|
|
extra["reasoning_effort"] = effort
|
|
if request.max_output_tokens is not None:
|
|
extra["max_tokens"] = int(request.max_output_tokens)
|
|
# request.model is the agent name; request.model_override carries
|
|
# the per-request /model override. Threaded into cfg.model so the
|
|
# inner executor's per-turn precedence picks it up over the spec
|
|
# default (HARNESS_*_MODEL).
|
|
config = ExecutorConfig(model=request.model_override, extra=extra)
|
|
tools = _normalize_tool_schemas(request.tools or [])
|
|
system_prompt = request.instructions or ""
|
|
# Install the stable bridge ONCE on the executor. The SDK
|
|
# caches its client on first turn and the MCP handlers
|
|
# closure-capture ``self._tool_executor`` then. A fresh
|
|
# bridge per turn would leave those handlers pointing at
|
|
# a turn-1 closure forever — every later tool call would
|
|
# dispatch into a dead ctx and the SDK would hang.
|
|
# Rebind ``_current_ctx`` / ``_current_agent`` per turn
|
|
# instead; the stable bridge reads from those slots at
|
|
# call time. ``_tool_executor`` is harness-specific (not
|
|
# declared on the inner :class:`Executor` ABC), so the
|
|
# attribute set is best-effort — executors that don't
|
|
# read it ignore the value, executors that DO read it
|
|
# (Claude SDK) honor the round-trip protocol.
|
|
if getattr(executor, "_tool_executor", None) is None: # type: ignore[attr-defined]
|
|
executor._tool_executor = self._stable_tool_executor # type: ignore[attr-defined]
|
|
# Install the elicitation handler once on first use, same
|
|
# stable-reference pattern as ``_tool_executor``. The SDK's
|
|
# ``can_use_tool`` callback is constructed per-turn inside
|
|
# ClaudeSDKExecutor.run_turn() from this attribute, so the
|
|
# closure always reads ``_current_ctx`` at call time.
|
|
if getattr(executor, "_elicitation_handler", None) is None: # type: ignore[attr-defined]
|
|
executor._elicitation_handler = self._stable_elicitation_handler # type: ignore[attr-defined]
|
|
# Install the policy evaluator bridge once on first use, same
|
|
# stable-reference pattern as ``_tool_executor``. The inner
|
|
# executor calls this before/after each LLM call to evaluate
|
|
# LLM_REQUEST / LLM_RESPONSE policies via a round-trip to the
|
|
# Omnigent server (routed through the runner).
|
|
if getattr(executor, "_policy_evaluator", None) is None: # type: ignore[attr-defined]
|
|
executor._policy_evaluator = self._stable_policy_evaluator # type: ignore[attr-defined]
|
|
self._current_ctx = ctx
|
|
self._current_agent = request.model
|
|
# Reset the MCP call-id queue at turn start. A prior turn
|
|
# that errored mid-stream (e.g. cancelled while a tool_use
|
|
# block had been parsed but its MCP-handler hadn't fired
|
|
# yet) could leave entries behind; carrying them into the
|
|
# next turn would mis-correlate the new turn's first
|
|
# MCP dispatches with stale observed events from the
|
|
# previous turn. Clearing makes each turn's correlation
|
|
# window self-contained.
|
|
self._pending_mcp_call_ids.clear()
|
|
self._dispatched_call_ids.clear()
|
|
|
|
# --- Tracing setup ------------------------------------------------
|
|
# Create a TracingContext per turn when tracing is enabled.
|
|
# The trace_context_for_response wrapper derives the W3C
|
|
# trace ID from the response_id so operators can look up
|
|
# traces by response ID without a mapping table.
|
|
tracing = is_tracing_enabled()
|
|
from omnigent.runtime.telemetry import current_session_id, session_scope
|
|
|
|
# Prefer the conversation id the request hook already bound
|
|
# (authoritative, from the /sessions/<conv>/events path); fall back to
|
|
# the adapter session key, which can be a random uuid for harnesses
|
|
# built without one.
|
|
turn_session_id = current_session_id() or self._session_key
|
|
if tracing and self._tracing_ctx is None:
|
|
self._tracing_ctx = TracingContext(session_id=turn_session_id)
|
|
tctx = self._tracing_ctx if tracing else None
|
|
agent_span = None
|
|
# Active tool span for correlating ToolCallRequest → ToolCallComplete.
|
|
_active_tool_span = None
|
|
_active_tool_parent = None
|
|
|
|
user_message = _extract_last_user_message(request.input)
|
|
# --- End tracing setup --------------------------------------------
|
|
|
|
# Watcher for mid-turn steering injections. The scaffold
|
|
# routes incoming steering events with
|
|
# ``previous_response_id == ctx.response_id`` onto
|
|
# ``ctx.next_injection`` (see _push_injection in the
|
|
# scaffold). The watcher converts each injection into an
|
|
# :meth:`Executor.enqueue_session_message` call so the
|
|
# inner SDK delivers the new user message into its
|
|
# in-flight session — without this hook, AP-forwarded
|
|
# steering would queue forever and the LLM would never
|
|
# see it.
|
|
injection_watcher = asyncio.create_task(
|
|
self._watch_injections(ctx, executor),
|
|
name=f"executor-adapter-injection-watch:{ctx.response_id}",
|
|
)
|
|
try:
|
|
# Wrap the executor loop in the trace context so all
|
|
# MLflow spans share the response-derived trace ID.
|
|
# The context manager is built outside the `with` so we
|
|
# can fall back to nullcontext if the response_id format
|
|
# doesn't match (e.g. 24-char hex vs expected 32).
|
|
trace_cm: contextlib.AbstractContextManager[None] = contextlib.nullcontext()
|
|
if tctx:
|
|
try:
|
|
from omnigent.runtime.telemetry import trace_context_for_response
|
|
|
|
trace_cm = trace_context_for_response(response_id=ctx.response_id)
|
|
except Exception:
|
|
_logger.debug("trace_context_for_response unavailable", exc_info=True)
|
|
# Bind the session for the whole turn so every span the harness
|
|
# creates (agent/LLM/tool, the native tmux inject, DB/httpx child
|
|
# spans) is tagged with session.id by the span processor — no
|
|
# per-harness code. (session_scope imported above with current_session_id.)
|
|
with session_scope(turn_session_id), trace_cm:
|
|
if tctx is not None:
|
|
agent_span = tctx.start_agent_span(
|
|
agent_name=request.model or "unknown",
|
|
user_message=user_message,
|
|
model=request.model_override or request.model,
|
|
)
|
|
|
|
response_text: str | None = None
|
|
async for event in executor.run_turn(
|
|
messages=messages,
|
|
tools=tools,
|
|
system_prompt=system_prompt,
|
|
config=config,
|
|
):
|
|
if ctx.cancelled.is_set():
|
|
if tctx is not None and agent_span is not None:
|
|
from omnigent.runtime.telemetry import record_cancellation
|
|
|
|
record_cancellation(agent_span)
|
|
tctx.end_agent_span(agent_span, response=None)
|
|
agent_span = None
|
|
await executor.interrupt_session(self._session_key)
|
|
return
|
|
# --- Tracing: emit spans per event ---
|
|
if tctx is not None:
|
|
if isinstance(event, ToolCallRequest):
|
|
_active_tool_parent = tctx._current_span
|
|
_active_tool_span = tctx.start_tool_span(
|
|
_strip_mcp_tool_prefix(event.name),
|
|
event.args or {},
|
|
)
|
|
elif isinstance(event, ToolCallComplete):
|
|
if _active_tool_span is not None:
|
|
tctx.end_tool_span(
|
|
_active_tool_span,
|
|
result=event.result,
|
|
error=event.error,
|
|
duration_ms=event.duration_ms,
|
|
parent_span=_active_tool_parent,
|
|
)
|
|
_active_tool_span = None
|
|
_active_tool_parent = None
|
|
elif isinstance(event, TurnComplete):
|
|
response_text = event.response
|
|
if event.usage is not None:
|
|
from omnigent.runtime.telemetry import record_llm_usage
|
|
|
|
# Record usage on the agent span for
|
|
# aggregate visibility.
|
|
record_llm_usage(agent_span, event.usage)
|
|
# --- End tracing ---
|
|
self._translate_event(event, ctx)
|
|
if isinstance(event, TurnComplete):
|
|
if tctx is not None and agent_span is not None:
|
|
tctx.end_agent_span(agent_span, response=response_text)
|
|
agent_span = None
|
|
return
|
|
if isinstance(event, TurnCancelled):
|
|
ctx.cancelled.set()
|
|
if tctx is not None and agent_span is not None:
|
|
from omnigent.runtime.telemetry import record_cancellation
|
|
|
|
record_cancellation(agent_span)
|
|
tctx.end_agent_span(agent_span, response=None)
|
|
return
|
|
if isinstance(event, ExecutorError):
|
|
if tctx is not None and agent_span is not None:
|
|
tctx.end_agent_span(
|
|
agent_span,
|
|
response=None,
|
|
error=event.message,
|
|
)
|
|
agent_span = None
|
|
raise RuntimeError(f"inner executor error: {event.message}")
|
|
except ElicitationDeclinedError:
|
|
# Fallback for executors that propagate the exception directly
|
|
# (non-SDK / non-spawned-task paths). SDK-based executors use
|
|
# ctx.cancelled.set() from _stable_elicitation_handler instead,
|
|
# because the SDK wraps its control-request callbacks in their
|
|
# own tasks and would swallow a raised exception before it
|
|
# reached this handler. Either way the turn ends as cancelled.
|
|
_logger.info(
|
|
"elicitation explicitly declined for response %s — aborting turn",
|
|
ctx.response_id,
|
|
)
|
|
if tctx is not None and agent_span is not None:
|
|
from omnigent.runtime.telemetry import record_cancellation
|
|
|
|
record_cancellation(agent_span)
|
|
tctx.end_agent_span(agent_span, response=None)
|
|
ctx.cancelled.set()
|
|
# Interrupt the inner executor session so the in-flight
|
|
# generation stops immediately, same as the normal
|
|
# cancellation path.
|
|
if self._executor is not None:
|
|
await self._executor.interrupt_session(self._session_key)
|
|
except BaseException:
|
|
# End agent span on unhandled exceptions so it's not
|
|
# left open (which would leak on the OTel provider).
|
|
if tctx is not None and agent_span is not None:
|
|
tctx.end_agent_span(agent_span, response=None, error="unhandled exception")
|
|
agent_span = None
|
|
raise
|
|
finally:
|
|
# Stop the injection watcher and let it drain so a
|
|
# late ``next_injection`` doesn't fire after we've
|
|
# moved on. ``injection_watcher`` is a long-poll
|
|
# against an ``asyncio.Queue.get`` — cancelling is
|
|
# the only way to break it.
|
|
injection_watcher.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await injection_watcher
|
|
# Flush the OTel provider and finalize the trace status
|
|
# on the MLflow server. Without the flush, the
|
|
# BatchSpanProcessor may not have exported the final
|
|
# spans. Without the PATCH, the OTLP-ingested trace
|
|
# stays "In progress" because the server has no
|
|
# signal that all spans have arrived.
|
|
if tctx is not None:
|
|
try:
|
|
from opentelemetry import trace as otel_trace
|
|
|
|
provider = otel_trace.get_tracer_provider()
|
|
if hasattr(provider, "force_flush"):
|
|
provider.force_flush(timeout_millis=5000)
|
|
except Exception:
|
|
pass
|
|
# Clear the per-turn pointers so a stray late callback
|
|
# (e.g. one fired after the SDK's stream closed) sees
|
|
# ``None`` and returns an explicit error rather than
|
|
# silently dispatching into the just-finished ctx.
|
|
self._current_ctx = None
|
|
self._current_agent = None
|
|
|
|
async def _handle_interrupt_event(self) -> Response:
|
|
"""Cancel the turn AND drop the inner executor session.
|
|
|
|
The base handler sets ``ctx.cancelled`` (terminal event becomes
|
|
``response.cancelled``) and clears the inject target. But the live
|
|
client — dropping which is what actually stops the in-flight
|
|
generation and forces the next turn to rebuild fresh — was only
|
|
dropped by the run loop's between-events ``interrupt_session`` call,
|
|
which is skipped when the turn is blocked awaiting the first token or
|
|
torn down via HTTP disconnect. The next turn then reuses the client
|
|
and flushes the abandoned generation: the post-cancel stream dump +
|
|
the off-by-one. Drop it here, synchronously on the interrupt.
|
|
|
|
:returns: The base handler's 204 response.
|
|
:raises OmnigentError: 404 (from the base handler) when no turn is
|
|
in flight.
|
|
"""
|
|
response = await super()._handle_interrupt_event()
|
|
# ``self._executor`` is set once a turn has run; an interrupt only
|
|
# reaches here when one is in flight, so it is non-None in practice.
|
|
# interrupt_session is best-effort and idempotent (a no-op if the run
|
|
# loop already dropped the session), so call it bare like that path.
|
|
if self._executor is not None:
|
|
await self._executor.interrupt_session(self._session_key)
|
|
return response
|
|
|
|
async def _watch_injections(self, ctx: TurnContext, executor: Executor) -> None:
|
|
"""
|
|
Loop forwarding ``ctx.next_injection`` to the inner SDK.
|
|
|
|
Polls :meth:`TurnContext.next_injection` (a queue
|
|
backed by the scaffold's ``_push_injection``) and
|
|
translates each :class:`CreateResponseRequest` into an
|
|
:meth:`Executor.enqueue_session_message` call so the
|
|
inner SDK delivers the new user message into its
|
|
in-flight session. Loops until cancelled by the
|
|
:meth:`run_turn` finally block.
|
|
|
|
Best-effort throughout — a failed injection logs and
|
|
continues; a malformed request body (missing user
|
|
text) is skipped; the watcher never raises out of the
|
|
loop body. Errors that escape would teardown the watch
|
|
task and surface as a confusing test/test-env failure
|
|
in the parent run_turn.
|
|
|
|
:param ctx: The active turn's context.
|
|
:param executor: The inner executor to inject into.
|
|
Must implement ``enqueue_session_message`` for the
|
|
forwarding to actually deliver — executors that
|
|
don't (legacy harnesses) silently no-op.
|
|
"""
|
|
while True:
|
|
try:
|
|
injection = await ctx.next_injection(timeout=None)
|
|
except asyncio.CancelledError:
|
|
return
|
|
if injection is None:
|
|
# ``next_injection(None)`` blocks forever on the
|
|
# queue, so a None return here means the queue
|
|
# protocol changed or the scaffold pushed a
|
|
# sentinel. Defensive — bail rather than spin.
|
|
return
|
|
text = _extract_user_text(injection.input)
|
|
if not text:
|
|
# Empty / malformed input — skip rather than
|
|
# call the SDK with garbage.
|
|
_logger.warning(
|
|
"skipping in-band injection with no text payload: %r",
|
|
injection.input,
|
|
)
|
|
continue
|
|
if ctx.cancelled.is_set():
|
|
# Turn interrupted — don't deliver into the dying session.
|
|
return
|
|
try:
|
|
accepted = await executor.enqueue_session_message(self._session_key, text)
|
|
except Exception:
|
|
_logger.exception(
|
|
"inner executor.enqueue_session_message failed; in-band injection lost"
|
|
)
|
|
continue
|
|
if not accepted:
|
|
_logger.warning(
|
|
"inner executor refused in-band injection "
|
|
"(supports_live_message_queue=False?); LLM will "
|
|
"not see the steered message until the next turn"
|
|
)
|
|
continue
|
|
# The executor consumed this injection into the running turn.
|
|
# Echo the runner's correlation id back as an
|
|
# ``injection.consumed`` marker so the runner drops the
|
|
# buffered copy and does not re-deliver it as a continuation
|
|
# turn (RUNNER_MESSAGE_INGEST.md Part B). Only meaningful when
|
|
# the runner stamped an injection_id; fresh-turn injections and
|
|
# legacy callers leave it unset.
|
|
injection_id = getattr(injection, "injection_id", None)
|
|
if injection_id:
|
|
ctx.emit(
|
|
InjectionConsumedEvent(
|
|
type="injection.consumed",
|
|
injection_id=injection_id,
|
|
)
|
|
)
|
|
|
|
async def _stable_tool_executor(
|
|
self,
|
|
tool_name: str,
|
|
args: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Cached bridge the inner SDK keeps over the lifetime of
|
|
the executor instance.
|
|
|
|
Reads :attr:`_current_ctx` and :attr:`_current_agent` at
|
|
call time so the dispatch always lands in the active
|
|
turn's :class:`TurnContext`. If neither is set (a
|
|
stale callback fired after the turn ended, or the SDK
|
|
called the executor without a wrapping ``run_turn``),
|
|
returns a clear error payload — letting the call park
|
|
on a dead ctx would just hang the SDK indefinitely.
|
|
|
|
For MCP-prefixed tools, drains the next ``tool_use_id``
|
|
from :attr:`_pending_mcp_call_ids` and passes it to
|
|
:func:`_bridge_one_dispatch` so the dispatch's
|
|
action_required event shares a ``call_id`` with the
|
|
inline observed event already emitted by
|
|
:meth:`_translate_event`. The deque is FIFO and the
|
|
inner SDK invokes MCP-server handlers in the same
|
|
order it parsed the corresponding tool_use blocks, so
|
|
positional pop is correct.
|
|
|
|
:param tool_name: Tool name from the LLM's call. Carries
|
|
the MCP prefix for SDK-registered tools (e.g.
|
|
``"mcp__omnigent__sys_terminal_launch"``).
|
|
:param args: Decoded argument dict.
|
|
:returns: A dict suitable as the MCP tool result.
|
|
"""
|
|
ctx = self._current_ctx
|
|
agent = self._current_agent
|
|
if ctx is None or agent is None:
|
|
_logger.warning(
|
|
"tool callback fired with no active turn context (tool=%s); returning error",
|
|
tool_name,
|
|
)
|
|
return {"error": "no active turn context for tool dispatch"}
|
|
# Pop the matching ``tool_use_id`` for MCP tools so the
|
|
# dispatch reuses the observed event's call_id. Non-MCP
|
|
# tools and out-of-order edge cases (queue empty when an
|
|
# MCP tool fires) fall through to a freshly-allocated id
|
|
# in :func:`_bridge_one_dispatch` — that loses the dedup
|
|
# but doesn't break the dispatch protocol.
|
|
# ``_stable_tool_executor`` IS the SDK's MCP-server tool
|
|
# callback — only invoked for MCP-routed tools. The
|
|
# callback receives the BARE tool name (the MCP wrapper
|
|
# strips the ``mcp__omnigent__`` prefix before
|
|
# dispatching), so a ``startswith("mcp__")`` guard here
|
|
# would always be False and the queue would never pop.
|
|
# Pop whenever there's a queued id; non-MCP paths don't
|
|
# populate the queue, so this can't accidentally drain
|
|
# an unrelated id.
|
|
correlated_call_id: str | None = None
|
|
if self._pending_mcp_call_ids:
|
|
correlated_call_id = self._pending_mcp_call_ids.popleft()
|
|
# Allocate the dispatch id here (rather than letting
|
|
# _bridge_one_dispatch default it) so we can record EXACTLY which id was
|
|
# round-tripped. _translate_event suppresses the inner
|
|
# ToolCallComplete for these ids — dispatch_tool already emits their
|
|
# function_call_output — while letting internally-run SDK tool
|
|
# completions (not in this set) through as the sole output source.
|
|
dispatch_call_id = correlated_call_id or f"call_{uuid.uuid4().hex[:12]}"
|
|
self._dispatched_call_ids.add(dispatch_call_id)
|
|
return await _bridge_one_dispatch(
|
|
ctx,
|
|
agent,
|
|
tool_name,
|
|
args,
|
|
call_id=dispatch_call_id,
|
|
)
|
|
|
|
async def _stable_elicitation_handler(
|
|
self,
|
|
tool_name: str,
|
|
tool_input: dict[str, Any],
|
|
) -> bool:
|
|
"""
|
|
Cached bridge the inner SDK keeps over the executor's lifetime.
|
|
|
|
Called by :class:`omnigent.inner.claude_sdk_executor.ClaudeSDKExecutor`
|
|
via ``options.can_use_tool`` when the Claude CLI requests
|
|
permission before executing a tool and ``permission_mode`` is not
|
|
``"bypassPermissions"``. Reads :attr:`_current_ctx` at call time
|
|
so the elicitation always lands in the active turn's context —
|
|
same rebind-per-turn pattern as :meth:`_stable_tool_executor`.
|
|
|
|
If no turn context is active (callback fired after the turn ended
|
|
or before a turn starts), returns ``False`` to deny by default
|
|
rather than granting unreviewed permission.
|
|
|
|
:param tool_name: Tool name the agent wants to call, e.g. ``"Bash"``.
|
|
:param tool_input: Arguments dict for the tool call,
|
|
e.g. ``{"command": "ls -la"}``.
|
|
:returns: ``True`` when the user approves the tool call;
|
|
``False`` when they deny it.
|
|
"""
|
|
ctx = self._current_ctx
|
|
if ctx is None:
|
|
# Elicitation callback fired with no active turn — this is a
|
|
# code-level invariant violation (run_turn clears _current_ctx
|
|
# in its finally block; a callback should never arrive after
|
|
# that). Deny by default: fail safe rather than grant permission
|
|
# that was never reviewed.
|
|
_logger.error(
|
|
"elicitation callback fired with no active turn context "
|
|
"(tool=%s); denying by default",
|
|
tool_name,
|
|
)
|
|
return False
|
|
|
|
elicitation_id = f"elicit_{secrets.token_hex(16)}"
|
|
# Build a concise preview: truncate long args so the UI widget
|
|
# stays readable. 300 chars matches AP's policy-engine preview.
|
|
try:
|
|
preview = json.dumps(tool_input, ensure_ascii=False)
|
|
except (TypeError, ValueError):
|
|
preview = repr(tool_input)
|
|
preview = preview[:300]
|
|
|
|
label = self._harness_label
|
|
policy_name = f"{label.lower()}_sdk_permission"
|
|
params = ElicitationRequestParams(
|
|
mode="form",
|
|
message=f"{label} wants to use **{tool_name}**",
|
|
requestedSchema=None,
|
|
url=None,
|
|
phase="tool_call",
|
|
policy_name=policy_name,
|
|
content_preview=f"{tool_name}({preview})",
|
|
)
|
|
result = await ctx.elicit(elicitation_id, params)
|
|
if result.action == "decline":
|
|
# The SDK invokes this callback from a spawned control-request
|
|
# task whose try/except would swallow a raised exception before
|
|
# it could reach run_turn's except block. Signal the cancellation
|
|
# via ctx.cancelled instead — the run_turn event loop checks this
|
|
# flag between events and takes the existing interrupt path.
|
|
ctx.cancelled.set()
|
|
return result.action == "accept"
|
|
|
|
async def _stable_policy_evaluator(
|
|
self,
|
|
phase: str,
|
|
data: dict[str, Any],
|
|
) -> PolicyVerdictPayload:
|
|
"""
|
|
Cached bridge the inner executor keeps over its lifetime.
|
|
|
|
Called by the inner executor before (``PHASE_LLM_REQUEST``)
|
|
and after (``PHASE_LLM_RESPONSE``) each LLM call. Routes
|
|
the evaluation through the scaffold's
|
|
:meth:`TurnContext.evaluate_policy` round-trip, which emits
|
|
a ``policy_evaluation.requested`` SSE event and parks on a
|
|
Future until the runner delivers the verdict.
|
|
|
|
The round-trip has a timeout (see ``_POLICY_EVAL_TIMEOUT_S``
|
|
in ``_scaffold.py``) so a stalled verdict defaults to ALLOW
|
|
instead of hanging the executor.
|
|
|
|
:param phase: Proto-style phase string, e.g.
|
|
``"PHASE_LLM_REQUEST"`` or ``"PHASE_LLM_RESPONSE"``.
|
|
:param data: Event data dict for the policy engine.
|
|
:returns: The policy verdict from the Omnigent server.
|
|
"""
|
|
ctx = self._current_ctx
|
|
if ctx is None:
|
|
# Orphaned callback after a turn-context desync (#1026). Blanket
|
|
# ALLOW here silently bypasses guardrails: for a PHASE_TOOL_CALL this
|
|
# adapter is the only enforcement point (the call is never re-checked
|
|
# server-side), so an unevaluable verdict must fail closed. Mirror
|
|
# the runner's phase-aware default in _evaluate_policy_via_omnigent —
|
|
# tool calls DENY; advisory LLM phases and the post-execution result
|
|
# phase ALLOW so a transient desync never needlessly wedges them.
|
|
fail_closed = phase in FAIL_CLOSED_PHASES
|
|
action = "POLICY_ACTION_DENY" if fail_closed else "POLICY_ACTION_ALLOW"
|
|
_logger.warning(
|
|
"policy evaluator fired with no active turn context (phase=%s); "
|
|
"returning %s by default",
|
|
phase,
|
|
"DENY" if fail_closed else "ALLOW",
|
|
)
|
|
return PolicyVerdictPayload(
|
|
action=action,
|
|
reason=(
|
|
f"No active turn context; failing closed for {phase}." if fail_closed else None
|
|
),
|
|
)
|
|
evaluation_id = f"poleval_{secrets.token_hex(16)}"
|
|
return await ctx.evaluate_policy(evaluation_id, phase, data)
|
|
|
|
def _ensure_executor(self) -> Executor:
|
|
"""
|
|
Construct the inner executor on first use; return cached
|
|
instance thereafter.
|
|
|
|
:returns: The per-conversation inner :class:`Executor`.
|
|
"""
|
|
if self._executor is None:
|
|
self._executor = self._executor_factory()
|
|
return self._executor
|
|
|
|
def _translate_event(self, event: ExecutorEvent, ctx: TurnContext) -> None:
|
|
"""
|
|
Translate one inner :class:`ExecutorEvent` into Omnigent SSE
|
|
events emitted via ``ctx.emit``.
|
|
|
|
Per the v1 limitations in the module docstring, all
|
|
:class:`ToolCallRequest` / :class:`ToolCallComplete`
|
|
events are treated as observed-only (the SDK already
|
|
executed the tool natively) — they surface as paired
|
|
function_call + function_call_output items with
|
|
``status: "completed"`` per §Sub-agent representation.
|
|
|
|
:param event: The inner executor event to translate.
|
|
:param ctx: The per-turn context; events are pushed onto
|
|
its queue.
|
|
"""
|
|
if isinstance(event, TextChunk):
|
|
ctx.emit(
|
|
OutputTextDeltaEvent(
|
|
type="response.output_text.delta",
|
|
delta=event.text,
|
|
)
|
|
)
|
|
elif isinstance(event, ReasoningChunk):
|
|
# Translate inner reasoning to the Omnigent wire shape so the
|
|
# workflow sees the same SSE events whether the executor
|
|
# runs inline or behind the harness scaffold.
|
|
if event.event_type == "reasoning_started":
|
|
ctx.emit(
|
|
ReasoningStartedEvent(type="response.reasoning.started"),
|
|
)
|
|
elif event.event_type == "reasoning_summary":
|
|
ctx.emit(
|
|
ReasoningSummaryTextDeltaEvent(
|
|
type="response.reasoning_summary_text.delta",
|
|
delta=event.delta,
|
|
),
|
|
)
|
|
else:
|
|
# ``reasoning_text`` and any future flavor land here.
|
|
ctx.emit(
|
|
ReasoningTextDeltaEvent(
|
|
type="response.reasoning_text.delta",
|
|
delta=event.delta,
|
|
),
|
|
)
|
|
elif isinstance(event, ToolCallRequest):
|
|
# Observed function_call item, emitted INLINE as the
|
|
# inner SDK parses each tool_use block — that's what
|
|
# gives the REPL a ``⏵ tool_name`` line interleaved
|
|
# with assistant text rather than bunched at the end
|
|
# of the response.
|
|
#
|
|
# For MCP-prefixed tool names we ALSO queue the
|
|
# ``tool_use_id`` so the matching :func:`_bridge_one_dispatch`
|
|
# call (fired post-stream when the SDK invokes the
|
|
# MCP-server handler) uses the same call_id. With the
|
|
# ids correlated, the SDK client's ``BlockStream``
|
|
# dedupes the post-stream action_required event
|
|
# against this observed event — keeping the inline
|
|
# render and avoiding the 2026-04-28 duplicate-render
|
|
# bug. Without the queue, the two events would have
|
|
# different uuids and the REPL would render
|
|
# ``⏵ tool_name`` twice (which is what 989bfde
|
|
# mitigated by suppressing this branch entirely —
|
|
# but at the cost of inline rendering, the regression
|
|
# this fix reverses).
|
|
#
|
|
# The emitted ``name`` is the bare tool name. The
|
|
# inner SDK passes the MCP-prefixed form
|
|
# (``mcp__omnigent__sys_terminal_launch``) but the
|
|
# Omnigent wire shape and persisted conversation items
|
|
# carry the bare form — kept consistent with
|
|
# ``omnigent/runtime/workflow.py``'s
|
|
# ``_observed_tool_call_sse_dicts`` /
|
|
# ``_build_observed_tool_items`` pair so SSE name and
|
|
# store-item name don't drift.
|
|
tool_use_id = _call_id_from_metadata(event.metadata)
|
|
# Push the tool_use_id onto the correlation queue
|
|
# whenever the executor stamped one. The original
|
|
# gate restricted this to MCP-prefixed names (the
|
|
# claude-sdk path), but every wrapped harness whose
|
|
# tools round-trip through :meth:`_stable_tool_executor`
|
|
# needs the same correlation — openai-agents
|
|
# included. Without this, the dispatch path's
|
|
# action_required emit gets a fresh uuid, the AP
|
|
# client can't dedupe against the inline observed
|
|
# event, and the REPL renders the call twice plus an
|
|
# empty result panel for the orphan call.
|
|
if tool_use_id is not None:
|
|
self._pending_mcp_call_ids.append(tool_use_id)
|
|
call_id = tool_use_id or f"call_{uuid.uuid4().hex[:12]}"
|
|
bare_name = _strip_mcp_tool_prefix(event.name)
|
|
ctx.emit(
|
|
OutputItemDoneEvent(
|
|
type="response.output_item.done",
|
|
item={
|
|
"id": f"fc_{uuid.uuid4().hex[:12]}",
|
|
"type": "function_call",
|
|
"status": _OBSERVED_TOOL_CALL_STATUS,
|
|
"name": bare_name,
|
|
"arguments": _serialize_args(event.args),
|
|
"call_id": call_id,
|
|
"agent": ctx.response_id,
|
|
},
|
|
)
|
|
)
|
|
elif isinstance(event, ToolCallComplete):
|
|
# Paired function_call_output. Downstream consumers (web
|
|
# blockStream, runner persistence) pair results to requests
|
|
# STRICTLY by call_id and discard empty ones — there is NO
|
|
# positional correlation, so a ToolCallComplete that reaches
|
|
# here without a real call_id orphans and leaves its request a
|
|
# perpetual in_progress card. Inner executors must stamp a real
|
|
# id (antigravity allocates one positionally for the SDK's
|
|
# id-less error hook); an id-less completion is DROPPED below
|
|
# (it cannot pair) rather than emitted as a ghost card.
|
|
#
|
|
# A tool routed through _stable_tool_executor →
|
|
# ctx.dispatch_tool already has its function_call_output
|
|
# emitted by dispatch_tool when the Future resolves.
|
|
# Emitting a second one here would duplicate it on the
|
|
# SSE stream and produce ghost "Waiting for output"
|
|
# cards in the Web UI. Suppress ONLY for those
|
|
# round-tripped call ids (tracked in
|
|
# ``_dispatched_call_ids``) — the scaffold is their single
|
|
# output source.
|
|
#
|
|
# Tools the inner SDK ran ENTIRELY INTERNALLY — the
|
|
# antigravity executor has no _stable_tool_executor
|
|
# dispatch at all, and its ToolCallComplete is bridged
|
|
# from the SDK's PostToolCall/OnToolError hooks — never go
|
|
# through dispatch_tool, so their completion here is the
|
|
# ONLY output event. Suppressing those (the prior
|
|
# ``_current_ctx is not None`` blanket rule) left every
|
|
# antigravity tool call rendering as perpetual in_progress
|
|
# with no paired function_call_output. Emit them.
|
|
call_id = _call_id_from_metadata(getattr(event, "metadata", None)) or ""
|
|
# Suppress two cases — each would otherwise put a useless or duplicate
|
|
# function_call_output on the SSE stream:
|
|
# 1. a DISPATCHED id — _stable_tool_executor already emitted the
|
|
# authoritative output via ctx.dispatch_tool (its single source);
|
|
# a second one here duplicates it and ghosts a "Waiting for
|
|
# output" card.
|
|
# 2. an EMPTY call_id — downstream consumers pair STRICTLY by call_id
|
|
# and discard empty ones, so an id-less completion cannot pair; it
|
|
# only renders a perpetual ghost card. This is a shared adapter
|
|
# used by every adapter-backed harness, so an inner executor that
|
|
# bridges an id-less ToolCallComplete mid-turn (the old blanket
|
|
# ``_current_ctx is not None`` rule swallowed these) must not leak
|
|
# an empty-id output. Internal-tool executors (antigravity) stamp a
|
|
# real positional id, so their legitimate completions still emit.
|
|
if not call_id or call_id in self._dispatched_call_ids:
|
|
return
|
|
item: dict[str, Any] = {
|
|
"id": f"fco_{uuid.uuid4().hex[:12]}",
|
|
"type": "function_call_output",
|
|
"call_id": call_id,
|
|
# Cap the mirror; the inner SDK already consumed the full result.
|
|
"output": cap_tool_output(_serialize_tool_result(event)),
|
|
}
|
|
raw_args = getattr(event, "metadata", {}).get("arguments")
|
|
if isinstance(raw_args, dict):
|
|
item["arguments"] = raw_args
|
|
ctx.emit(OutputItemDoneEvent(type="response.output_item.done", item=item))
|
|
elif isinstance(event, TurnComplete):
|
|
# If the executor produced a final assistant message
|
|
# via TurnComplete.response (rather than through
|
|
# streaming TextChunks), emit it now. Streamed
|
|
# TextChunks have already gone out via the branch
|
|
# above; in that case TurnComplete.response is
|
|
# typically None or a duplicate that the inner
|
|
# executor sets for non-streaming consumers.
|
|
if event.response is not None:
|
|
# Avoid double-emitting if the response text
|
|
# matches the streamed deltas — but that
|
|
# accumulator lives on Session today, not on the
|
|
# adapter. For v1 we trust streaming-mode
|
|
# executors to leave response=None, and only emit
|
|
# on non-streaming paths.
|
|
pass
|
|
# Capture provider-reported usage so _build_terminal_event
|
|
# can include it in the response.completed SSE payload.
|
|
# The harness HTTP client reads response["usage"] from that
|
|
# payload to populate TurnComplete.usage on the Omnigent side.
|
|
if event.usage is not None:
|
|
ctx.provider_usage = event.usage
|
|
elif isinstance(event, CompactionComplete):
|
|
from omnigent.server.schemas import (
|
|
CompactionCompletedEvent,
|
|
CompactionInProgressEvent,
|
|
)
|
|
|
|
ctx.emit(
|
|
CompactionInProgressEvent(
|
|
type="response.compaction.in_progress",
|
|
)
|
|
)
|
|
ctx.emit(
|
|
CompactionCompletedEvent(
|
|
type="response.compaction.completed",
|
|
total_tokens=event.token_count,
|
|
summary=event.summary,
|
|
summary_model=event.model,
|
|
compacted_messages=event.compacted_messages,
|
|
)
|
|
)
|
|
# ExecutorError handled by the caller (re-raises so the
|
|
# scaffold can build a response.failed terminal event).
|
|
|
|
def _build_error_detail(self, exception: BaseException) -> Any:
|
|
"""
|
|
Map an inner-executor exception onto a contract-recognized
|
|
error code.
|
|
|
|
Default :meth:`HarnessApp._build_error_detail` uses
|
|
``type(exception).__name__`` as the code, which never
|
|
matches AP's retryable allowlist
|
|
(:data:`omnigent.runtime.harnesses._client_executor._RETRYABLE_HARNESS_ERROR_CODES`).
|
|
Result before this override: a known transient failure
|
|
like ``anthropic.RateLimitError`` would surface as
|
|
``code="RateLimitError"``, AP's allowlist wouldn't match,
|
|
retry-classification would call it permanent, and the
|
|
workflow would never retry. This method closes that gap
|
|
for the harnesses the adapter wraps (claude-sdk, plus the
|
|
codex / openai-agents-sdk / pi wraps once they land).
|
|
|
|
Translation precedence:
|
|
|
|
1. :class:`omnigent.errors.OmnigentError` (incl.
|
|
:class:`RetryableLLMError` / :class:`PermanentLLMError`)
|
|
— already carries a semantic ``code`` string per the
|
|
project's own classification, so use it verbatim.
|
|
2. OpenAI SDK exceptions — surfaced by the inner OpenAI
|
|
Agents SDK / Open Responses SDK executors used by the
|
|
codex / openai-agents wraps. Maps onto the AP
|
|
allowlist's semantic codes
|
|
(``"rate_limit_exceeded"``, ``"server_error"``,
|
|
``"timeout"``, ``"connection_error"``).
|
|
3. ``claude_agent_sdk`` exceptions — the Claude Code CLI
|
|
wrapper used by the claude-sdk wrap. Currently only
|
|
``CLIConnectionError`` maps to a retryable code; the
|
|
other CLI errors (NotFound, JSONDecode, ProcessError)
|
|
are non-retryable and fall through to the base
|
|
implementation so operators see the class name.
|
|
4. ``httpx`` exceptions for SDKs that surface raw
|
|
``httpx.TimeoutException`` / ``httpx.ConnectError``
|
|
rather than wrapping them.
|
|
5. Fallback: base class implementation
|
|
(``type(exception).__name__``).
|
|
|
|
:param exception: The exception :meth:`run_turn` raised.
|
|
:returns: An :class:`ErrorDetail` whose ``code`` matches
|
|
the Omnigent allowlist for known retryable failures, and is
|
|
still informative (provider exception class) for the
|
|
rest.
|
|
"""
|
|
from omnigent.errors import OmnigentError
|
|
from omnigent.server.schemas import ErrorDetail
|
|
|
|
if isinstance(exception, OmnigentError):
|
|
# Project-internal structured errors already carry a
|
|
# semantic code (e.g. ``RetryableLLMError(code="timeout")``).
|
|
return ErrorDetail(code=exception.code, message=str(exception))
|
|
|
|
code = classify_inner_exception(exception)
|
|
if code is not None:
|
|
return ErrorDetail(code=code, message=str(exception))
|
|
|
|
# Unknown exception type — preserve the class name so
|
|
# operators can still grep for it in logs, even though
|
|
# AP's retry allowlist won't match.
|
|
return super()._build_error_detail(exception)
|
|
|
|
async def on_shutdown(self) -> None:
|
|
"""
|
|
Release the inner executor's resources on subprocess
|
|
shutdown.
|
|
|
|
Overrides :meth:`HarnessApp.on_shutdown`; called from
|
|
the scaffold's lifespan teardown path. Closes the
|
|
executor's session and the executor itself so child
|
|
processes (e.g. ``claude --output-format``) are reaped
|
|
rather than orphaned.
|
|
"""
|
|
if self._executor is not None:
|
|
await self._executor.close_session(self._session_key)
|
|
await self._executor.close()
|
|
self._executor = None
|
|
|
|
|
|
def _classify_openai_exception(exception: BaseException) -> str | None:
|
|
"""
|
|
Map an OpenAI SDK exception onto the Omnigent semantic code allowlist.
|
|
|
|
The :mod:`openai` package is the runtime SDK for the
|
|
openai-agents-sdk + codex wraps (and the open-responses
|
|
inner executor). Its exception hierarchy mirrors HTTP
|
|
semantics. We only translate the variants that match AP's
|
|
retryable allowlist verbatim — adding more codes here without
|
|
extending the allowlist would just produce strings Omnigent ignores.
|
|
|
|
Lazy-imports :mod:`openai` because the package isn't a hard
|
|
dependency of every wrap (the claude-sdk wrap goes through the
|
|
Claude Code CLI, not the OpenAI API). ``ImportError`` returns
|
|
``None`` so the caller falls through to other classifiers —
|
|
never crashes the error path on a missing optional dep.
|
|
|
|
:param exception: The exception :meth:`ExecutorAdapter.run_turn`
|
|
caught.
|
|
:returns: A semantic code from the Omnigent allowlist (e.g.
|
|
``"rate_limit_exceeded"``), or ``None`` when *exception*
|
|
is not an OpenAI SDK exception we recognize.
|
|
"""
|
|
try:
|
|
import openai
|
|
except ImportError:
|
|
return None
|
|
|
|
if isinstance(exception, openai.RateLimitError):
|
|
return "rate_limit_exceeded"
|
|
if isinstance(exception, openai.APITimeoutError):
|
|
return "timeout"
|
|
if isinstance(exception, openai.APIConnectionError):
|
|
return "connection_error"
|
|
if isinstance(exception, openai.InternalServerError):
|
|
# 500-class server-side failures from the gateway. Worth
|
|
# retrying — usually transient capacity issues. Matches
|
|
# AP's existing ``"server_error"`` allowlist entry.
|
|
return "server_error"
|
|
# Context-window overflow may arrive as a direct BadRequestError
|
|
# or wrapped inside an openai-agents SDK exception. The generic
|
|
# classifier walks the cause chain for all providers.
|
|
from omnigent.llms.errors import is_context_length_exceeded
|
|
|
|
if is_context_length_exceeded(exception):
|
|
return "context_length_exceeded"
|
|
return None
|
|
|
|
|
|
def _classify_claude_sdk_exception(exception: BaseException) -> str | None:
|
|
"""
|
|
Map a :mod:`claude_agent_sdk` exception onto the Omnigent semantic
|
|
code allowlist.
|
|
|
|
The Claude Code CLI wrapper surfaces a small exception set
|
|
(``CLIConnectionError``, ``CLIJSONDecodeError``,
|
|
``CLINotFoundError``, ``ProcessError``,
|
|
``ClaudeSDKError``). Only ``CLIConnectionError`` matches a
|
|
retryable allowlist entry — ``CLINotFoundError`` (binary
|
|
missing) and ``CLIJSONDecodeError`` (corrupt CLI output)
|
|
are non-retryable so they fall through to the base
|
|
implementation where operators see the class name.
|
|
|
|
:param exception: The exception :meth:`ExecutorAdapter.run_turn`
|
|
caught.
|
|
:returns: A semantic code from the Omnigent allowlist, or ``None``
|
|
when *exception* isn't a recognized retryable
|
|
:mod:`claude_agent_sdk` exception.
|
|
"""
|
|
try:
|
|
import claude_agent_sdk
|
|
except ImportError:
|
|
return None
|
|
|
|
# ``CLINotFoundError`` is a subclass of ``CLIConnectionError``
|
|
# in the SDK's hierarchy — match it FIRST and return ``None``
|
|
# so the missing-binary case falls through to the base
|
|
# implementation (non-retryable: a missing CLI won't appear
|
|
# spontaneously). Without this guard, ``isinstance(not_found,
|
|
# CLIConnectionError)`` matches and we'd loop on a permanent
|
|
# configuration failure.
|
|
if isinstance(exception, claude_agent_sdk.CLINotFoundError):
|
|
return None
|
|
if isinstance(exception, claude_agent_sdk.CLIConnectionError):
|
|
return "connection_error"
|
|
return None
|
|
|
|
|
|
def _classify_httpx_exception(exception: BaseException) -> str | None:
|
|
"""
|
|
Map an :mod:`httpx` exception onto the Omnigent semantic code allowlist.
|
|
|
|
Some inner executors (notably ``litellm``-backed paths for
|
|
non-Anthropic providers) surface raw httpx exceptions instead
|
|
of wrapping them in a provider-specific class. Translating
|
|
these here keeps retry classification accurate regardless of
|
|
which transport layer raised.
|
|
|
|
Lazy-imports :mod:`httpx` for symmetry with the Anthropic
|
|
classifier — though httpx is currently a hard project
|
|
dependency, the lazy form keeps this module's import graph
|
|
minimal for harnesses that never raise httpx errors.
|
|
|
|
:param exception: The exception :meth:`ExecutorAdapter.run_turn`
|
|
caught.
|
|
:returns: A semantic code from the Omnigent allowlist, or ``None``
|
|
when *exception* is not an httpx exception we recognize.
|
|
"""
|
|
try:
|
|
import httpx
|
|
except ImportError:
|
|
return None
|
|
|
|
if isinstance(exception, httpx.TimeoutException):
|
|
return "timeout"
|
|
if isinstance(exception, httpx.ConnectError):
|
|
return "connection_error"
|
|
# ``httpx.ReadError`` / ``WriteError`` / ``CloseError`` are
|
|
# transport-level network failures (peer closed the
|
|
# connection mid-stream, broken pipe, etc.). They're
|
|
# siblings of ``ConnectError`` under ``NetworkError`` and
|
|
# behave the same way for retry purposes — the connection
|
|
# is gone, a fresh attempt may succeed. Without this branch,
|
|
# users see the bare class name (``[llm] ReadError``) with
|
|
# no semantic code, and the Omnigent retry classifier treats them
|
|
# as permanent.
|
|
if isinstance(exception, httpx.NetworkError):
|
|
return "connection_error"
|
|
# ``httpx.RemoteProtocolError`` ("peer closed connection without
|
|
# sending complete message body" — the canonical symptom of a
|
|
# subprocess being SIGKILL'd mid-stream). Sits under
|
|
# ``ProtocolError`` → ``TransportError``, NOT under
|
|
# ``NetworkError``, so the prior branch doesn't catch it.
|
|
# Without this branch, a SIGKILL during streaming surfaces as
|
|
# an unrecognized exception and the retry classifier treats it
|
|
# as permanent — the L2 retry layer never fires for the exact
|
|
# case it was designed to recover from.
|
|
if isinstance(exception, httpx.RemoteProtocolError):
|
|
return "connection_error"
|
|
return None
|
|
|
|
|
|
def _classify_anthropic_exception(exception: BaseException) -> str | None:
|
|
"""
|
|
Map an :mod:`anthropic` SDK exception onto the Omnigent semantic
|
|
code allowlist.
|
|
|
|
The Anthropic Python SDK is the underlying transport for the
|
|
Claude CLI subprocess used by the claude-sdk wrap; raw SDK
|
|
exceptions sometimes surface upward when the CLI fails to
|
|
intercept them (e.g. mid-stream gateway errors that bypass
|
|
the CLI's framing layer). Without a dedicated classifier
|
|
these would fall through to ``[llm] RateLimitError`` /
|
|
``[llm] APIConnectionError`` etc., which the Omnigent retry
|
|
allowlist doesn't match — silent demotion of retryable
|
|
failures to permanent.
|
|
|
|
Lazy-imports :mod:`anthropic` because the package isn't a
|
|
hard dependency of every wrap (only the claude-sdk path
|
|
pulls it in transitively via the CLI). ``ImportError``
|
|
returns ``None`` so the caller falls through to other
|
|
classifiers — matches the established pattern of the
|
|
sibling classifiers in this module.
|
|
|
|
Phase 3 of ``designs/RETRY_ACROSS_HARNESSES.md`` —
|
|
consolidates the per-SDK fan-out at
|
|
:func:`classify_inner_exception`.
|
|
|
|
:param exception: The exception :meth:`ExecutorAdapter.run_turn`
|
|
caught.
|
|
:returns: A semantic code from the Omnigent allowlist, or ``None``
|
|
when *exception* is not an Anthropic SDK exception we
|
|
recognize.
|
|
"""
|
|
try:
|
|
import anthropic
|
|
except ImportError:
|
|
return None
|
|
|
|
if isinstance(exception, anthropic.RateLimitError):
|
|
return "rate_limit_exceeded"
|
|
if isinstance(exception, anthropic.APITimeoutError):
|
|
return "timeout"
|
|
if isinstance(exception, anthropic.APIConnectionError):
|
|
return "connection_error"
|
|
if isinstance(exception, anthropic.InternalServerError):
|
|
# 500-class server-side failures from Anthropic — same
|
|
# semantic as ``_classify_openai_exception``: transient
|
|
# capacity issues that retry typically resolves.
|
|
return "server_error"
|
|
return None
|
|
|
|
|
|
def classify_inner_exception(exception: BaseException) -> str | None:
|
|
"""
|
|
Map any inner-SDK exception onto the Omnigent semantic code allowlist.
|
|
|
|
Single entry point that fans out across the per-SDK
|
|
classifiers. First match wins. Returns ``None`` when no
|
|
classifier recognizes the exception — caller is expected
|
|
to fall back to ``type(exception).__name__`` so operators
|
|
can still grep.
|
|
|
|
Phase 3 of ``designs/RETRY_ACROSS_HARNESSES.md``: this
|
|
function replaces the per-call fan-out at
|
|
:meth:`ExecutorAdapter._build_error_detail`'s old code,
|
|
which inlined three separate classifier calls. New
|
|
classifiers (e.g. for a Pi-specific exception type)
|
|
plug in here once and benefit every consumer.
|
|
|
|
Order matters when SDK exception hierarchies overlap. Today
|
|
none do — each lazy import only matches its own SDK's
|
|
classes — but if Anthropic ever subclasses an httpx
|
|
exception (or vice versa), a more-specific classifier
|
|
must come first. Documented as a sequence rather than a
|
|
dict so the order is explicit.
|
|
|
|
:param exception: The exception
|
|
:meth:`ExecutorAdapter.run_turn` caught.
|
|
:returns: A semantic code from the Omnigent allowlist (e.g.
|
|
``"rate_limit_exceeded"``), or ``None`` when no
|
|
classifier matched.
|
|
"""
|
|
for classifier in (
|
|
_classify_openai_exception,
|
|
_classify_anthropic_exception,
|
|
_classify_claude_sdk_exception,
|
|
_classify_httpx_exception,
|
|
):
|
|
code = classifier(exception)
|
|
if code is not None:
|
|
return code
|
|
# Generic fallback: context-window overflow from any SDK that
|
|
# stamps a recognized code on the exception or its cause chain
|
|
# but wasn't caught by the per-SDK classifiers above (e.g.
|
|
# because the SDK isn't installed so its classifier short-
|
|
# circuited on ImportError).
|
|
from omnigent.llms.errors import is_context_length_exceeded
|
|
|
|
if is_context_length_exceeded(exception):
|
|
return "context_length_exceeded"
|
|
return None
|
|
|
|
|
|
def _normalize_tool_schemas(
|
|
schemas: list[dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Flatten OpenAI Chat-Completions tool schemas to inner shape.
|
|
|
|
Omnigent emits :meth:`omnigent.tools.base.Tool.get_schema` in
|
|
OpenAI Chat Completions shape: ``{"type": "function",
|
|
"function": {"name", "description", "parameters"}}``. The
|
|
inner :class:`Executor` ABC and its
|
|
:func:`omnigent.inner.claude_sdk_executor._build_mcp_tools`
|
|
consumer read ``schema.get("name")`` / ``schema.get(...)``
|
|
directly — they expect a flat shape. Without this
|
|
translation the inner SDK registers MCP tools with empty
|
|
names, which the LLM cannot call.
|
|
|
|
Each schema is converted to:
|
|
``{"name": ..., "description": ..., "parameters": ...}``.
|
|
Schemas already in flat shape (``"name"`` at top level) pass
|
|
through unchanged so this function is idempotent.
|
|
|
|
:param schemas: AP-emitted tool schemas in either shape.
|
|
:returns: Schemas in flat shape ready for the inner
|
|
Executor.
|
|
"""
|
|
flat: list[dict[str, Any]] = []
|
|
for schema in schemas:
|
|
# Flat-form: validate ``name`` is non-empty even on the
|
|
# pass-through branch. The inner SDK's
|
|
# ``_build_mcp_tools`` reads ``schema.get("name")``
|
|
# directly; an entry like ``{"name": ""}`` would slip
|
|
# through and produce the "Tool name cannot be empty"
|
|
# warning + render a phantom MCP tool the LLM can't
|
|
# call. Drop those with the same warning the chat-
|
|
# completions branch uses.
|
|
if "name" in schema:
|
|
name = schema.get("name")
|
|
if not isinstance(name, str) or not name:
|
|
_logger.warning(
|
|
"skipping flat-form tool schema with empty name: %r",
|
|
schema,
|
|
)
|
|
continue
|
|
flat.append(schema)
|
|
continue
|
|
# Chat-completions form — pull from the ``function``
|
|
# sub-dict. Same skip-on-missing-name rule.
|
|
function = schema.get("function") or {}
|
|
name = function.get("name")
|
|
if not isinstance(name, str) or not name:
|
|
_logger.warning("skipping tool schema with no name: %r", schema)
|
|
continue
|
|
translated: dict[str, Any] = {"name": name}
|
|
description = function.get("description")
|
|
if isinstance(description, str):
|
|
translated["description"] = description
|
|
parameters = function.get("parameters")
|
|
if isinstance(parameters, dict):
|
|
translated["parameters"] = parameters
|
|
flat.append(translated)
|
|
return flat
|
|
|
|
|
|
async def _bridge_one_dispatch(
|
|
ctx: TurnContext,
|
|
agent: str,
|
|
tool_name: str,
|
|
args: dict[str, Any],
|
|
*,
|
|
call_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Round-trip one tool call through ``ctx.dispatch_tool``.
|
|
|
|
JSON-encodes *args*, awaits the result, and shapes the
|
|
return so the inner Claude SDK's MCP-tool wrapper hands a
|
|
sensible payload back to the LLM. Dispatch failures become
|
|
``{"error": <str>}`` so the SDK marks the MCP response as
|
|
an error rather than hanging.
|
|
|
|
:param ctx: The current turn's context.
|
|
:param agent: Agent name (required on the function_call item).
|
|
:param tool_name: Tool name from the LLM's call.
|
|
:param args: Decoded arguments dict from the LLM.
|
|
:param call_id: Optional explicit ``call_id`` to use on the
|
|
emitted action_required event. When provided (typically
|
|
the SDK's ``tool_use_id`` threaded through
|
|
:meth:`ExecutorAdapter._stable_tool_executor`'s queue),
|
|
lets the SDK client dedupe this event against the inline
|
|
observed event that :meth:`_translate_event` already
|
|
emitted with the same id. ``None`` falls back to a
|
|
freshly-allocated uuid — the dispatch still works but the
|
|
observed and action_required events render as separate
|
|
``⏵ tool_name`` lines.
|
|
:returns: A dict suitable as the MCP tool result.
|
|
"""
|
|
import json
|
|
|
|
if call_id is None:
|
|
call_id = f"call_{uuid.uuid4().hex[:12]}"
|
|
try:
|
|
output = await ctx.dispatch_tool(
|
|
call_id=call_id,
|
|
name=tool_name,
|
|
arguments=json.dumps(args),
|
|
agent=agent,
|
|
)
|
|
except Exception as exc:
|
|
_logger.exception("dispatch_tool failed for %s", tool_name)
|
|
return {"error": str(exc)}
|
|
# Try to parse the output as JSON — if Omnigent serialized the
|
|
# tool result via ``ToolResult.content`` (a JSON string),
|
|
# parsing here gives the SDK a structured dict it can show
|
|
# the LLM cleanly. Falls back to raw string under a
|
|
# ``result`` key for non-JSON outputs.
|
|
try:
|
|
parsed = json.loads(output)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return {"result": output}
|
|
if isinstance(parsed, dict):
|
|
return parsed
|
|
return {"result": parsed}
|
|
|
|
|
|
def _extract_last_user_message(
|
|
input_value: str | list[dict[str, Any]],
|
|
) -> str:
|
|
"""Extract the last user message text from a request input.
|
|
|
|
Handles both conversation-history shape (list of message items
|
|
with ``role``/``content``) and single-turn shape (plain string
|
|
or content-block list). Used by tracing to populate the agent
|
|
span's ``user_message`` input.
|
|
|
|
:param input_value: The request's ``input`` field.
|
|
:returns: The text of the last user message, or empty string.
|
|
"""
|
|
if isinstance(input_value, str):
|
|
return input_value
|
|
# Conversation-history shape: find last user message
|
|
last_user_text = ""
|
|
for item in input_value:
|
|
role = item.get("role")
|
|
if role == "user":
|
|
content = item.get("content")
|
|
if isinstance(content, str):
|
|
last_user_text = content
|
|
elif isinstance(content, list):
|
|
parts = []
|
|
for block in content:
|
|
text = block.get("text")
|
|
if isinstance(text, str):
|
|
parts.append(text)
|
|
if parts:
|
|
last_user_text = "\n".join(parts)
|
|
# Single-turn content-block shape (no role key)
|
|
elif role is None:
|
|
text = item.get("text")
|
|
if isinstance(text, str):
|
|
last_user_text = text
|
|
return last_user_text
|
|
|
|
|
|
def _extract_user_text(
|
|
input_value: str | list[dict[str, Any]],
|
|
) -> str:
|
|
"""
|
|
Pull plain-text content from a steering injection's input.
|
|
|
|
Used by :meth:`ExecutorAdapter._watch_injections` to convert
|
|
a scaffold-side :class:`CreateResponseRequest` into a single
|
|
string the inner executor's
|
|
:meth:`Executor.enqueue_session_message` can deliver. Mirrors
|
|
the structure of :func:`_translate_input_to_messages` but
|
|
returns a plain string rather than a Message dict.
|
|
|
|
:param input_value: The injection's ``input`` field — either
|
|
a plain string (shorthand) or a list of content blocks.
|
|
:returns: Concatenated text content. Empty string when no
|
|
text blocks are present.
|
|
"""
|
|
if isinstance(input_value, str):
|
|
return input_value
|
|
parts: list[str] = []
|
|
for block in input_value:
|
|
text_value = block.get("text")
|
|
if isinstance(text_value, str):
|
|
parts.append(text_value)
|
|
return "\n".join(parts)
|
|
|
|
|
|
def _translate_input_to_messages(
|
|
input_value: str | list[dict[str, Any]],
|
|
) -> list[Message]:
|
|
"""
|
|
Convert :class:`CreateResponseRequest.input` into inner
|
|
:class:`Message` list.
|
|
|
|
Two shapes are accepted on the wire:
|
|
|
|
1. **Conversation-history shape** — a list whose entries are
|
|
``{"type": "message", "role": "...", "content": [...]}``
|
|
items, optionally interleaved with typed items
|
|
(``function_call``, ``function_call_output``,
|
|
``reasoning``, native tool calls). This is what
|
|
:func:`_translate_messages_to_input` produces for the
|
|
full Layer 2 history; one inner :class:`Message` is
|
|
emitted per role-keyed message item, and tool-related
|
|
items are dropped (the inner :class:`ClaudeSDKExecutor`'s
|
|
``_build_prompt`` only consumes role-keyed messages
|
|
when serializing the "Conversation so far:" prefix). The
|
|
contract here is "preserve user/assistant turns;
|
|
tool-call records are reconstructed from the SDK's own
|
|
Layer 1 state on resume," which is the right tradeoff
|
|
for the small inner-Message contract.
|
|
2. **Single-turn fallback shape** — a plain string or a
|
|
list of content-block dicts (``input_text``, etc.) with
|
|
no role wrappers. AP's older clients (and any direct
|
|
hand-off that hasn't been migrated yet) still send this.
|
|
Concatenated into a single user-role message — same
|
|
behavior the harness has always had for the
|
|
single-message case.
|
|
|
|
:param input_value: The request's ``input`` field.
|
|
:returns: One inner :class:`Message` per conversation turn
|
|
when history is present; a single user message
|
|
otherwise.
|
|
"""
|
|
if isinstance(input_value, str):
|
|
return [{"role": "user", "content": input_value}]
|
|
|
|
history_messages = _extract_role_keyed_messages(input_value)
|
|
if history_messages:
|
|
return history_messages
|
|
|
|
# Fallback: legacy single-turn path for inputs that don't
|
|
# carry role wrappers (e.g. a bare list of content blocks).
|
|
# Preserves multimodal blocks when present.
|
|
content = _normalize_message_content(input_value)
|
|
return [{"role": "user", "content": content}]
|
|
|
|
|
|
def _extract_role_keyed_messages(
|
|
input_value: list[dict[str, Any]],
|
|
) -> list[Message]:
|
|
"""
|
|
Pull role-keyed message items out of an Omnigent ``input`` list.
|
|
|
|
Looks for ``{"type": "message", "role": ..., "content": ...}``
|
|
entries (the shape :func:`_translate_messages_to_input`
|
|
produces from the workflow's history list). Each match
|
|
becomes one inner :class:`Message` whose content is either
|
|
a plain string (text-only messages) or the original list
|
|
of typed content blocks (when ``input_image`` /
|
|
``input_file`` blocks are present alongside text).
|
|
|
|
Tool-call items (``function_call``, ``function_call_output``,
|
|
``reasoning``, native tool items) are intentionally
|
|
skipped: the inner SDK reconstructs those from its own
|
|
Layer 1 state when resuming, so duplicating them in the
|
|
serialized prompt would just confuse the LLM.
|
|
|
|
:param input_value: The Omnigent ``input`` list.
|
|
:returns: One :class:`Message` per role-keyed entry, or
|
|
an empty list when *input_value* contains no
|
|
role-keyed items (caller falls back to the legacy
|
|
single-user-message path).
|
|
"""
|
|
messages: list[Message] = []
|
|
for item in input_value:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if item.get("type") != "message" or "role" not in item:
|
|
continue
|
|
role = item["role"]
|
|
content = _normalize_message_content(item.get("content"))
|
|
# Skip empty messages (e.g. assistant turns that only
|
|
# produced tool calls without text). Their absence
|
|
# doesn't break the prompt — the prior user turn still
|
|
# carries the question and the next user turn carries
|
|
# the follow-up.
|
|
if not content:
|
|
continue
|
|
messages.append({"role": role, "content": content})
|
|
return messages
|
|
|
|
|
|
_MULTIMODAL_BLOCK_TYPES: frozenset[str] = frozenset({"input_image", "input_file", "input_audio"})
|
|
|
|
|
|
def _normalize_message_content(
|
|
content: Any,
|
|
) -> str | list[dict[str, Any]]:
|
|
"""
|
|
Normalize Responses API message ``content`` for inner executors.
|
|
|
|
When the content list contains only text blocks
|
|
(``input_text`` / ``output_text``), collapses to a plain
|
|
string (the common text-only fast path). When multimodal
|
|
blocks (``input_image``, ``input_file``, ``input_audio``)
|
|
are present, returns the full block list so inner executors
|
|
can forward image/file content to the LLM.
|
|
|
|
:param content: The ``content`` field of a Responses API
|
|
message item, e.g. ``[{"type": "input_text",
|
|
"text": "..."}]``.
|
|
:returns: A plain string when all blocks are text-only, or
|
|
the original content block list when multimodal blocks
|
|
are present. Empty string for ``None`` or non-list
|
|
inputs.
|
|
"""
|
|
if content is None:
|
|
return ""
|
|
if isinstance(content, str):
|
|
return content
|
|
if not isinstance(content, list):
|
|
return ""
|
|
|
|
has_multimodal = any(
|
|
isinstance(block, dict) and block.get("type") in _MULTIMODAL_BLOCK_TYPES
|
|
for block in content
|
|
)
|
|
if has_multimodal:
|
|
return content
|
|
|
|
parts: list[str] = []
|
|
for block in content:
|
|
if not isinstance(block, dict):
|
|
continue
|
|
text_value = block.get("text")
|
|
if isinstance(text_value, str):
|
|
parts.append(text_value)
|
|
return "\n".join(parts)
|
|
|
|
|
|
def _serialize_args(args: dict[str, Any]) -> str:
|
|
"""
|
|
JSON-encode a tool-call arguments dict.
|
|
|
|
:param args: The arguments dict from
|
|
:class:`ToolCallRequest.args`.
|
|
:returns: A JSON string the AP-side function_call item
|
|
carries verbatim.
|
|
"""
|
|
import json
|
|
|
|
encoded: str = json.dumps(args)
|
|
return encoded
|
|
|
|
|
|
def _serialize_tool_result(event: ToolCallComplete) -> str:
|
|
"""
|
|
Stringify a :class:`ToolCallComplete` for the
|
|
function_call_output's ``output`` field.
|
|
|
|
The inner executor populates ``result`` with whatever shape
|
|
the wrapped SDK hands back. For the Claude SDK that's the
|
|
Anthropic ``ToolResultBlock.content`` value, which can be
|
|
one of:
|
|
|
|
- A plain ``str`` — pass through unchanged.
|
|
- A list of typed content blocks (``{"type": "text", "text":
|
|
...}``, image blocks, etc.) — join the ``text`` fields so
|
|
the LLM-rendered tool output reaches Omnigent intact. Without
|
|
this branch, ``str([{...}])`` would emit a Python repr
|
|
(literal ``[{'type': 'text', 'text': '...'}]``) and
|
|
function_call_output would carry garbage instead of the
|
|
actual command output.
|
|
- Anything else (dicts, ints, ``None``, …) — best-effort
|
|
JSON encode, falling back to ``repr`` so we never
|
|
silently drop the value.
|
|
|
|
Failure path:
|
|
|
|
- ``error`` is preferred when ``result`` is None and the
|
|
status carries one.
|
|
- Empty string is a last resort — the inner Executor
|
|
contract is that one of result/error is populated;
|
|
logged so a regression surfaces in the Omnigent logs.
|
|
|
|
:param event: The completion event.
|
|
:returns: A string suitable for the AP-side
|
|
function_call_output's ``output`` field.
|
|
"""
|
|
if event.result is not None:
|
|
return _stringify_tool_payload(event.result)
|
|
if event.error is not None:
|
|
return f"[error] {event.error}"
|
|
_logger.warning(
|
|
"ToolCallComplete for %s had neither result nor error",
|
|
event.name,
|
|
)
|
|
return ""
|
|
|
|
|
|
def _stringify_tool_payload(value: Any) -> str:
|
|
"""
|
|
Coerce a tool's result payload into a string.
|
|
|
|
See :func:`_serialize_tool_result` for the full rationale —
|
|
this helper is split out so the list-of-blocks branch is
|
|
independently testable without standing up a full
|
|
:class:`ToolCallComplete`.
|
|
|
|
:param value: Anything the inner Executor put in
|
|
``ToolCallComplete.result``.
|
|
:returns: A string suitable for serialization on the wire.
|
|
"""
|
|
import json
|
|
|
|
if isinstance(value, str):
|
|
return value
|
|
if isinstance(value, list):
|
|
# Anthropic-style content blocks: pull out every
|
|
# ``text`` field. Non-text blocks (image, tool_use, etc.)
|
|
# have no string projection here and are skipped — a
|
|
# follow-up could route image blocks into AP's file-
|
|
# store, but for the v1 stdout-of-Bash use case skipping
|
|
# them is correct.
|
|
text_parts: list[str] = []
|
|
for block in value:
|
|
if not isinstance(block, dict):
|
|
continue
|
|
block_text = block.get("text")
|
|
if isinstance(block_text, str):
|
|
text_parts.append(block_text)
|
|
if text_parts:
|
|
return "".join(text_parts)
|
|
# No usable text blocks — fall through to JSON encode so
|
|
# callers still get a structured representation rather
|
|
# than an empty string.
|
|
try:
|
|
return json.dumps(value)
|
|
except (TypeError, ValueError):
|
|
return repr(value)
|
|
|
|
|
|
def _call_id_from_metadata(metadata: dict[str, Any] | None) -> str | None:
|
|
"""
|
|
Extract a call_id from an executor's per-call metadata dict.
|
|
|
|
Different inner executors stash the call_id under different
|
|
keys (Claude SDK uses ``"call_id"``; Codex uses ``"call_id"``
|
|
on the protocol envelope; Pi uses an opaque id). For v1 we
|
|
look for the conventional ``"call_id"`` key first; harness
|
|
wraps that need other keys can subclass and override.
|
|
|
|
:param metadata: The metadata dict from
|
|
:class:`ToolCallRequest.metadata` (or ``None``).
|
|
:returns: The call_id string if present and stringy, else
|
|
``None``.
|
|
"""
|
|
if not metadata:
|
|
return None
|
|
value = metadata.get("call_id")
|
|
if isinstance(value, str):
|
|
return value
|
|
return None
|