"""Observers for the native stateful ReAct agent."""
from __future__ import annotations
import asyncio
import datetime as _dt
import logging
import re
import time
from typing import Any
from rich.console import Console
from frontier_agent.components.finalization import (
has_malformed_tool_protocol,
remaining_phase_budget_s,
)
from frontier_agent.components.observers.console import (
RichConsoleObserver as RichConsoleObserver,
)
from frontier_agent.core.llm import LLMClient
from frontier_agent.core.loop_types import (
AgentLoopResult,
BaseObserver,
)
from frontier_agent.core.messages import Message, assistant_msg_with_reasoning, user_msg
from frontier_agent.infra.nonblocking_stream import nonblocking_stderr
from workflows._shared.sdk_shim import (
ReporterDeltaEmitter,
record_reporter_usage,
)
from workflows.stateful_react_agent._runtime import (
_build_recovery_messages,
_forced_final_stop_reasons,
_minimal_best_effort_answer,
_strip_leaked_tool_calls,
force_final_answer,
scrub_leaked_tool_calls,
unwrap_fenced_images,
)
from workflows.stateful_react_agent.prompts import get_report_prompt
logger = logging.getLogger(__name__)
# NonBlockingStream implements only the part of the file protocol Rich uses
# (write / flush / isatty); see its docstring.
_console = Console(
file=nonblocking_stderr(), # pyright: ignore[reportArgumentType]
width=200,
force_terminal=True,
)
_CONTENT_PREVIEW = 2000
_ARGS_PREVIEW = 400
_RESULT_PREVIEW = 400
class FinalAnswerSalvageObserver(BaseObserver):
"""Synthesise a reporter-disabled final answer before streaming/terminal.
On bounded/abnormal stops (``max_turns`` / ``context_limit_reached`` / infra
errors) the agent produced no clean no-tool final turn, so the answer is
synthesised by :func:`force_final_answer`. Historically that ran in the node
tail — AFTER the loop's terminal ``final`` and the reporter stream — so the
wire carried the pre-salvage draft while the node returned a different,
synthesised answer.
Reporter-enabled runs use :class:`ReportSynthesisObserver` directly and do
not mount this observer. In reporter-disabled runs, running it here
(``critical``, ordered before :class:`ReporterStreamObserver` and the
serve chain's protocol stream observer) sets ``result.final_content`` and
``result.metadata["final_answer"]`` before either reads them, so the raw
stream, terminal ``final``, and node return agree. No-op on normal
``no_tool`` completion and on cancel/pause.
"""
critical: bool = True
def __init__(
self,
*,
llm: LLMClient,
timeout: float,
task_description: str,
thinking_format: str,
salvage_infra_errors: bool,
final_prompt: str | None = None,
language: str = "",
phase_deadline_monotonic: float | None = None,
) -> None:
self._llm = llm
self._timeout = timeout
self._task_description = task_description
self._thinking_format = thinking_format
self._salvage_infra_errors = salvage_infra_errors
self._final_prompt = final_prompt
self._language = language
self._phase_deadline_monotonic = phase_deadline_monotonic
async def on_loop_end(self, result: AgentLoopResult) -> None:
# D5b: cancel/pause leaves no terminal — don't synthesise a salvage
# answer for a run that was stopped.
if result.stopped_by == "paused":
return
# force_final_answer self-gates: it returns early unless stopped_by is a
# forced reason and no final_answer exists yet. It never raises (falls
# back to an explicit best-effort status), so no guard is needed here.
await force_final_answer(
result,
self._llm,
# Never ask for more time than the external ceiling still allows —
# being cancelled mid-rescue would leave the run with no answer at
# all, which is exactly what the rescue exists to prevent.
remaining_phase_budget_s(
self._timeout, self._phase_deadline_monotonic,
),
task_description=self._task_description,
thinking_format=self._thinking_format,
salvage_infra_errors=self._salvage_infra_errors,
final_prompt=self._final_prompt,
language=self._language,
)
class ReporterStreamObserver(BaseObserver):
"""Re-stream the final answer as reporter ``llm_delta`` frames, pre-terminal.
The stateful agent's final answer IS the report, but the frontend renders
the report from ``response.swarm.llm_delta`` frames with
``agent_id="reporter"``. The agent's own turns stream under
``agent_id="stateful_react"`` and ``top_terminal_tool=None`` for this
pipeline, so nothing reporter-attributed otherwise reaches the wire.
This observer emits the reporter run envelope + the resolved answer as
``output_text`` deltas at ``on_loop_end``. It is ``critical`` and MUST be
ordered BEFORE the serve chain's protocol stream observer in the observer
list (``notify_observers`` awaits critical hooks inline in list order): the
protocol observer emits the terminal ``final`` in ITS ``on_loop_end``, so
the reporter stream lands immediately before the terminal.
Cancellation/pause emits nothing — a hard cancel fires ``on_loop_cancelled``
(not ``on_loop_end``), and the ``paused`` stop is skipped here, matching the
D5b "no terminal on cancel" convention.
On salvage stops (``max_turns`` / ``context_limit_reached`` / infra errors)
the resolved answer comes from :class:`FinalAnswerSalvageObserver`, which
MUST be ordered before this one so ``result`` already holds the synthesised
answer here — otherwise the stream would carry the pre-salvage draft.
"""
critical: bool = True
def __init__(self, emitter: object) -> None:
self._stream = ReporterDeltaEmitter(emitter)
async def on_loop_end(self, result: AgentLoopResult) -> None:
# D5b: cancellation / pause leaves no terminal — emit nothing.
if result.stopped_by == "paused":
return
text = _strip_leaked_tool_calls(str(
result.metadata.get("final_answer") or result.final_content or ""
))
if not text:
return
self._stream.start()
self._stream.stream_output(text)
self._stream.finish(final_content=text)
_THINK_OPEN_RE = re.compile(r"<\s*think\s*>", re.IGNORECASE)
_THINK_CLOSE_RE = re.compile(r"<\s*/\s*think\s*>", re.IGNORECASE)
# Leaked qwen text-mode markup, split into openers + their closers so a partially
# streamed block can be held back instead of shipped raw (the final-text
# equivalents live in ``_runtime._LEAKED_*``).
_LEAK_OPEN_RE = re.compile(r"<\s*(tool_call|tool_response|function)\b", re.IGNORECASE)
_LEAK_CLOSE_RE = {
"tool_call": re.compile(r"<\s*/\s*tool_call\s*>", re.IGNORECASE),
"tool_response": re.compile(r"<\s*/\s*tool_response\s*>", re.IGNORECASE),
"function": re.compile(r"<\s*/\s*function\s*>", re.IGNORECASE),
}
# A trailing fragment that could still grow into any of these must be withheld.
_PARTIAL_TAGS = ("", "", "",)
def _partial_tail_len(text: str, literals: tuple[str, ...]) -> int:
"""Length of the trailing suffix that could still become one of ``literals``."""
limit = min(len(text), max(len(literal) for literal in literals) - 1)
for size in range(limit, 0, -1):
tail = text[-size:].lower()
if any(literal.startswith(tail) for literal in literals):
return size
return 0
def _drop_think_prefix(text: str) -> str:
"""Everything after the LAST ```` (mirrors mtv2's ``_strip_thinking``).
Belt-and-braces for the persisted answer: a closing tag that survives the
streaming filter (duplicated tags, a tag nested inside held markup) still
must not reach ``final_answer`` / the trace.
"""
if "" in text.lower():
return re.split(_THINK_CLOSE_RE, text)[-1]
return text
class _ReportStreamFilter:
"""Incrementally sanitise a streamed report body.
The delta stream and the persisted answer must agree (a consumer may
assemble the report from ``output_text`` deltas — the documented contract in
``reporter_stream.py``), so this applies BOTH of the final text's removals
per chunk:
* ``…`` blocks — including the SGLang / Qwen quirk where the
OPENING tag is missing but the closing tag is always emitted. For an
OPENER-LESS close the **last** ```` wins, matching ``_strip_thinking``
in mtv2 (``rsplit("", 1)[-1]``): a close arriving after visible text
was already streamed retroactively drops that text from the persisted report
(a live stream cannot be retracted). A PAIRED ``…`` after
visible text is a block embedded in the report — it is excised and the
surrounding text kept.
* leaked qwen text-mode tool-call markup
(:func:`~workflows.stateful_react_agent._runtime.scrub_leaked_tool_calls`).
Tags and whole markup blocks may straddle provider chunks, so any suffix
that could still become a delimiter — and any unclosed markup block — is
withheld until the following ``feed``. Visible leading/trailing whitespace is
deferred so the result matches a final ``.strip()`` without buffering the
body.
``assume_think_prefix`` (tag-mode thinking with ``enable_thinking``, i.e. the
Apodex profiles' live configuration) starts the filter INSIDE an implicit
think block: with an absent opener that is the only leak-free reading of the
stream. Two escapes keep it from swallowing a report: endpoints that do
separate reasoning (SGLang ``--reasoning-parser``) release the hold on the
first ``reasoning_content`` delta via :meth:`note_reasoning_channel`, and a
stream that produced neither a tag nor a reasoning delta is flushed verbatim
by :meth:`finish` (that call loses live streaming, never the report).
"""
def __init__(self, *, assume_think_prefix: bool = False) -> None:
# ``_assumed_hold``: inside an *implicit* (opener-less) think block —
# everything is withheld rather than trimmed, so the hold can be
# released later with the report body intact.
self._assumed_hold = bool(assume_think_prefix)
self._in_think = bool(assume_think_prefix)
self._pending = ""
self._visible = ""
self._after_think = False
self._visible_started = False
self._pending_whitespace = ""
self._saw_tag = False
# ``_paired_open``: the think block currently open was entered via a real
# ```` opener (not the implicit hold / an opener-less close), so
# its close is a block EMBEDDED in the report — the surrounding visible
# text must survive rather than be retracted by last-close-wins.
self._paired_open = False
@property
def visible_text(self) -> str:
"""Sanitised report text so far — authoritative for the persisted answer."""
return self._visible
def _emit(self, text: str) -> str:
text = scrub_leaked_tool_calls(text)
if not text:
return ""
if self._after_think:
text = text.lstrip()
self._after_think = False
if not self._visible_started:
text = text.lstrip()
if not text:
return ""
trailing = len(text) - len(text.rstrip())
if trailing:
core = text[:-trailing]
whitespace = text[-trailing:]
else:
core = text
whitespace = ""
if not core:
self._pending_whitespace += whitespace
return ""
out = self._pending_whitespace + core
self._pending_whitespace = whitespace
self._visible_started = True
self._visible += out
return out
def _close_think(self, end: int) -> None:
"""Leave a think block whose ```` ends at ``end`` in ``_pending``."""
self._pending = self._pending[end:]
self._in_think = False
self._assumed_hold = False
self._after_think = True
self._saw_tag = True
paired = self._paired_open
self._paired_open = False
if self._visible and not paired:
# Last close wins for an OPENER-LESS close (missing-opener SGLang/Qwen
# quirk, duplicated tags): what looked like the report was reasoning
# after all. The wire already has it; drop it from the persisted text
# so surfaces B and C stay correct. A PAIRED … after
# visible text is instead a block embedded in the report — excise it
# and keep the surrounding text (``_after_think`` trims its seam).
logger.warning(
"reporter: late after %d streamed chars — "
"dropping them from the persisted report", len(self._visible),
)
self._visible = ""
self._visible_started = False
self._pending_whitespace = ""
def _safe_end(self) -> int:
"""End of the portion of ``_pending`` that can be emitted now."""
text = self._pending
for match in _LEAK_OPEN_RE.finditer(text):
closer = _LEAK_CLOSE_RE[match.group(1).lower()]
if closer.search(text, match.end()) is None:
return match.start()
return len(text) - _partial_tail_len(text, _PARTIAL_TAGS)
def feed(self, text: str) -> str:
"""Consume one raw content delta and return its safe visible portion."""
if not text:
return ""
self._pending += text
emitted: list[str] = []
while self._pending:
if self._in_think:
close = _THINK_CLOSE_RE.search(self._pending)
if close is None:
if not self._assumed_hold:
# Confirmed reasoning: discard all but a partial close tag.
keep = _partial_tail_len(self._pending, _PARTIAL_CLOSE_TAGS)
self._pending = self._pending[len(self._pending) - keep:] if keep else ""
break
self._close_think(close.end())
continue
open_m = _THINK_OPEN_RE.search(self._pending)
close_m = _THINK_CLOSE_RE.search(self._pending)
if close_m is not None and (open_m is None or close_m.start() < open_m.start()):
# Stray close with no opener → everything before it is reasoning.
self._close_think(close_m.end())
continue
if open_m is not None:
visible = self._emit(self._pending[:open_m.start()])
if visible:
emitted.append(visible)
self._pending = self._pending[open_m.end():]
self._in_think = True
self._assumed_hold = False
self._saw_tag = True
self._paired_open = True
continue
end = self._safe_end()
safe, self._pending = self._pending[:end], self._pending[end:]
visible = self._emit(safe)
if visible:
emitted.append(visible)
break
return "".join(emitted)
def note_reasoning_channel(self) -> str:
"""Release an ``assume_think_prefix`` hold: the endpoint separates reasoning.
A native ``reasoning_content`` delta proves thinking is NOT inlined in
``content``, so the withheld text is report body. Returns whatever
becomes emittable (usually ``""`` — reasoning deltas precede content).
"""
if not self._assumed_hold:
return ""
self._assumed_hold = False
self._in_think = False
pending, self._pending = self._pending, ""
return self.feed(pending)
def finish(self) -> str:
"""Flush the settled tail; discard thinking and trailing whitespace."""
tail = ""
if self._in_think:
if self._assumed_hold and not self._saw_tag:
# Neither tag ever arrived and the endpoint never used the
# reasoning channel: the implicit-think hold was unnecessary, so
# flush verbatim rather than swallow the report.
self._in_think = False
self._assumed_hold = False
tail = self._emit(self._pending)
else:
tail = self._emit(self._pending)
self._pending = ""
self._pending_whitespace = ""
return tail
class ReportSynthesisObserver(BaseObserver):
"""Lightweight single-LLM reporter (opt-in via ``agent.reporter: true``).
On a natural loop end this runs ONE tool-free, streaming LLM call over the
whole conversation to synthesise a structured, cited report — the harness
analogue of a *standard-mode* reporter: append a summarize prompt to the
agent message history and stream one summary call. The LLM call itself is
streamed (for the ``reasoning`` channel + the incremental think-tag filter),
but the report BODY is buffered until the call finishes, cleaned (think-tag
strip + :func:`~workflows.stateful_react_agent._runtime.unwrap_fenced_images`,
which needs the whole document to match an image against its citation), and
only then re-chunked onto serve stdout as ``response.swarm.llm_delta`` frames
(``agent_id="reporter"``, ``channel="output_text"``) via
:meth:`~workflows._shared.sdk_shim.ReporterDeltaEmitter.stream_output`.
This trades live token-by-token streaming for a guarantee that ``output_text``
and ``final.answer`` are byte-identical (a fenced-then-unwrapped image can
never render differently on the wire than in the persisted report). This
observer REPLACES :class:`ReporterStreamObserver` (which merely re-emits the
raw last turn) — wire one XOR the other, never both.
Ordering: ``critical`` and placed after
:class:`FinalAnswerSalvageObserver` (so bounded exits already have a real
clean-context baseline) and before the serve chain's protocol stream
observer, so the reporter stream lands before the terminal ``final``.
Fail-open: any error keeps the pre-reporter salvage/last-turn answer, or
installs an explicit deterministic best-effort baseline when none exists.
The reporter is never a hard dependency. Cancellation/pause emits nothing
(D5b: no terminal on a stopped run).
"""
critical: bool = True
def __init__(
self,
*,
llm: LLMClient,
timeout: float | None,
task_description: str,
emitter: object = None,
usage_aggregator: object = None,
language: str = "English",
thinking_format: str = "tag",
inline_thinking: bool = False,
extra_observers: object = None,
context_max_tokens: int = 220_000,
phase_timeout: float | None = None,
phase_deadline_monotonic: float | None = None,
) -> None:
self._llm = llm
self._timeout = timeout
self._task_description = task_description
self._language = language or "English"
self._thinking_format = thinking_format
# Tag-mode thinking that the endpoint may NOT split onto its own channel
# (``enable_thinking`` + ``thinking_format: tag``): the ```` opener
# can be missing while ```` is always emitted, so the stream
# filter must hold the leading body back until the close proves where
# reasoning ended. See :class:`_ReportStreamFilter`.
self._inline_thinking = bool(inline_thinking)
self._emitter = emitter
self._usage_aggregator = usage_aggregator
# A worker-trace observer lives among the serve chain's extra
# observers; used to append an ``agent_type="reporter"`` timing row
# so the trace's llm_call_timings stays consistent with
# message_history + usage.
self._extra_observers = extra_observers
self._context_max_tokens = max(1_024, int(context_max_tokens))
self._phase_timeout = phase_timeout
# Absolute instant the whole task must finish by, when an external
# ceiling is known. ``phase_timeout`` is the *planned* budget; on a
# short wall the research loop's reserve does not survive intact, so
# the phase clamps itself to whatever is really left at start.
self._phase_deadline_monotonic = phase_deadline_monotonic
def _effective_phase_timeout(self) -> float | None:
"""Planned phase budget, clamped to the time the hard wall still allows."""
if self._phase_deadline_monotonic is None:
return self._phase_timeout
if self._phase_timeout is None:
return max(self._phase_deadline_monotonic - time.monotonic(), 1.0)
return remaining_phase_budget_s(
self._phase_timeout, self._phase_deadline_monotonic,
)
async def on_loop_end(self, result: AgentLoopResult) -> None:
# D5b: cancel/pause leaves no terminal — synthesise nothing.
if result.stopped_by == "paused":
return
stream = ReporterDeltaEmitter(self._emitter)
# Baseline answer to keep if the reporter fails / returns empty.
metadata_answer = result.metadata.get("final_answer")
baseline = _strip_leaked_tool_calls(str(
metadata_answer or result.final_content or "",
))
if baseline and not result.metadata.get("final_answer_source"):
source = (
"existing_partial"
if not metadata_answer
and result.stopped_by in _forced_final_stop_reasons(True)
else "agent"
)
result.metadata["final_answer"] = baseline
result.metadata["final_answer_source"] = source
if not baseline:
baseline = _minimal_best_effort_answer(
self._task_description,
result.stopped_by,
language=self._language,
)
result.metadata["final_answer"] = baseline
result.metadata["final_answer_source"] = "deterministic_fallback"
result.final_content = baseline
report_prompt = get_report_prompt(self._task_description, self._language)
history = list(result.messages)
if has_malformed_tool_protocol(history):
# A malformed/orphan tool call can make every reporter fallback leg
# fail with the same provider-side 400. Recover only in that case;
# healthy runs preserve their full structured conversation.
messages = _build_recovery_messages(
history,
task_description=self._task_description,
final_prompt=report_prompt,
context_max_tokens=self._context_max_tokens,
)
else:
messages = history
if (
messages
and isinstance(messages[-1], dict)
and messages[-1].get("role") == "user"
):
messages.pop()
messages.append(user_msg(report_prompt))
stream.start()
cancelled = False
# Stream through an incremental filter: reasoning models may inline
# ``…`` in ``delta.content`` (opener sometimes absent),
# with tags split across chunks, and may leak ```` markup.
# The filter's per-chunk output is accumulated into ``visible_text``
# (below) but deliberately NOT sent to ``output_text`` live: the report
# body is only put on the wire once, fully cleaned, after the call
# finishes (see the class docstring) — that is what keeps the delta
# stream and ``final.answer`` byte-identical. Native
# ``reasoning_content`` still streams live on its own channel; it never
# reaches the persisted report so it has no consistency requirement.
think_filter = _ReportStreamFilter(assume_think_prefix=self._inline_thinking)
raw_parts: list[str] = []
# Capture terminal stream metadata (usage on the late include_usage
# chunk, model/provider stamps) — last non-empty wins, matching the
# loop's own streaming extraction (llm_client.py).
rep_usage: dict = {}
rep_model = rep_provider = ""
started_ts = _dt.datetime.now(_dt.UTC)
status = "success"
error_message = ""
try:
async with asyncio.timeout(self._effective_phase_timeout()):
async for delta in self._llm.stream(
messages, timeout=self._timeout,
):
chunk = getattr(delta, "content", None)
if chunk:
raw_parts.append(chunk)
think_filter.feed(chunk)
rc = getattr(delta, "reasoning_content", "")
if rc:
# A native reasoning delta proves thinking is NOT
# inlined in ``content`` → release any implicit-think
# hold (the report text it frees up is picked up from
# ``visible_text`` below; nothing goes out live here).
think_filter.note_reasoning_channel()
stream.reasoning(rc)
if getattr(delta, "usage", None):
rep_usage = delta.usage
if getattr(delta, "model", ""):
rep_model = delta.model
if getattr(delta, "provider", ""):
rep_provider = delta.provider
except asyncio.CancelledError:
cancelled = True
raise
except Exception as exc:
status = "error"
error_message = f"{type(exc).__name__}: {exc}"
logger.warning(
"ReportSynthesisObserver: synthesis failed (%s) — "
"falling back to baseline answer", error_message,
)
if cancelled:
return
think_filter.finish()
# The LLM call happened whenever ANY terminal metadata arrived (usage /
# model / a content chunk), even if it then errored or returned empty —
# record it on the aggregator so billing never under-counts a real call.
raw_response = "".join(raw_parts)
# Persisted answer = exactly what the filter let through, plus a final
# ````-tail drop + strip as belt-and-braces (``final_answer`` /
# the trace must never carry leaked reasoning) and the citation-gated
# image unwrap. This is also what gets chunked onto the wire below, so
# the delta stream and ``final_answer`` are the same text by construction.
report = unwrap_fenced_images(
_strip_leaked_tool_calls(_drop_think_prefix(think_filter.visible_text)),
)
call_happened = bool(rep_usage or rep_model or raw_parts)
if call_happened:
record_reporter_usage(
self._usage_aggregator,
usage=rep_usage, provider=rep_provider, model=rep_model,
)
if status == "success" and report:
# Append a compact report-prompt + response pair to the user-visible
# history. The exact protocol-clean reporter request is captured in
# the reporter timing row below.
if (
result.messages
and isinstance(result.messages[-1], dict)
and result.messages[-1].get("role") == "user"
):
result.messages.pop()
result.messages.append(user_msg(report_prompt))
result.messages.append(
assistant_msg_with_reasoning(
report, "", thinking_format=self._thinking_format,
),
)
result.metadata["final_answer"] = report
result.metadata["final_answer_source"] = "reporter_llm"
result.metadata["final_answer_rescued"] = False
result.metadata["final_answer_rescue_mode"] = ""
result.final_content = report
# Put the fully-cleaned report on the wire in one shot — deltas are
# literal slices of ``report``, so they and ``final.answer`` agree
# by construction (see the class docstring).
stream.stream_output(report)
else:
# Failed OR empty synthesis: nothing was streamed live (the body is
# always buffered now), so fall back to the baseline answer as the
# thing put on the wire.
if status == "success":
status = "error"
error_message = "reporter returned empty content"
logger.warning(
"ReportSynthesisObserver: %s — keeping baseline answer",
error_message,
)
if baseline:
stream.stream_output(baseline)
# Worker-trace reporter timing row (agent_type="reporter") — keeps
# llm_call_timings consistent with the appended message + usage. Runs
# BEFORE the worker-trace observer's ``on_loop_end`` (this observer
# precedes the serve chain), so the row is present when the trace
# serialises.
self._append_reporter_timing(
started_ts=started_ts, status=status, error=error_message,
model=rep_model, provider=rep_provider, usage=rep_usage,
input_messages=messages, raw_response=raw_response,
)
stream.finish(
final_content=str(
result.metadata.get("final_answer") or result.final_content or "",
),
status=status,
stop_reason="final_answer" if status == "success" else "error",
error_message=error_message,
)
def _append_reporter_timing(
self,
*,
started_ts: _dt.datetime,
status: str,
error: str,
model: str,
provider: str,
usage: dict[str, Any],
input_messages: list[Message],
raw_response: Any,
) -> None:
"""Best-effort: append an ``agent_type="reporter"`` row to the worker trace."""
obs = self._extra_observers
if not obs:
return
try:
from workflows._shared.sdk_shim import find_trace_observer
trace_obs = find_trace_observer(obs)
except Exception:
trace_obs = None
if trace_obs is None:
return
end_ts = _dt.datetime.now(_dt.UTC)
try:
trace_obs.append_reporter_llm_call_timing(
purpose="Reporter | Synthesize Report",
model=model,
provider=provider,
status="success" if status == "success" else "failed",
error=error,
start_time=started_ts.isoformat(), end_time=end_ts.isoformat(),
duration_ms=int((end_ts - started_ts).total_seconds() * 1000),
ttft_ms=None, usage=dict(usage) if usage else None,
input_messages=[
dict(message)
for message in input_messages
if isinstance(message, dict)
],
raw_response=raw_response,
)
except Exception as exc:
logger.debug("ReportSynthesisObserver: timing append failed: %s", exc)