Debug-only model exchanges: prompt + response text per model round.
The metrics pipeline is counts-only by design (see
StageRecord: how many memories a turn recalled is publishable; the
memories are not). This module is the single, explicit hole in that
posture, and it behaves like one:
- Capture runs only when
SYMFONIC_DEBUG_MODEL_EXCHANGES is truthy
(1/true/yes). Default off: zero storage, zero behavior
change, one environment read per model round.
- Storage is a bounded in-memory ring (no durable sink, no log line).
A restart clears it; nothing here survives the process.
- Reads merge onto telemetry snapshots the caller is already scoped
to see, so an exchange surfaces only beside a row the admin
endpoint already authorized. There is deliberately no unkeyed dump.
Enable only on a local dev server with non-sensitive data.
capture_exchange
capture_exchange(*, run_id: str, prompt: Any, turn: Any, input_tokens: int, model: str) -> None
Record one round's exchange when the debug flag is on.
The per-run ordinal is the merge key the admin detail view joins
on (position within the run's snapshot group); input_tokens
lets the join verify instead of trusting order alone.
Source code in src/symfonic/core/observability/debug_exchanges.py
| def capture_exchange(
*,
run_id: str,
prompt: Any,
turn: Any,
input_tokens: int,
model: str,
) -> None:
"""Record one round's exchange when the debug flag is on.
The per-run ordinal is the merge key the admin detail view joins
on (position within the run's snapshot group); ``input_tokens``
lets the join verify instead of trusting order alone.
"""
if not debug_exchanges_enabled():
return
ordinal = _run_counts.get(run_id, 0)
_run_counts[run_id] = ordinal + 1
_ring.append(
{
"run_id": run_id,
"ordinal": ordinal,
"model": model,
"input_tokens": int(input_tokens or 0),
"prompt": prompt_text(prompt),
"response": response_text(turn),
}
)
|
clear_exchanges
clear_exchanges() -> None
Drop all captured exchanges (tests, and manual reset).
Source code in src/symfonic/core/observability/debug_exchanges.py
| def clear_exchanges() -> None:
"""Drop all captured exchanges (tests, and manual reset)."""
_ring.clear()
_run_counts.clear()
|
debug_exchanges_enabled
debug_exchanges_enabled() -> bool
True when the process opted into prompt/response capture.
Source code in src/symfonic/core/observability/debug_exchanges.py
| def debug_exchanges_enabled() -> bool:
"""True when the process opted into prompt/response capture."""
return os.environ.get(FLAG_ENV_VAR, "").strip().lower() in (
"1",
"true",
"yes",
)
|
exchanges_for_rows
exchanges_for_rows(rows: list[dict[str, Any]]) -> list[tuple[str | None, str | None]]
Align (prompt, response) pairs onto one run's snapshot rows.
rows must be the timestamp-ordered snapshots of a single run
group (as build_conversation_detail groups them). Position is
the join, input_tokens the verification: a mismatch attaches
nothing rather than a neighboring call's text (fail-closed on ring
skew, e.g. the flag toggled mid-process). All-None when the
flag is off or nothing was captured.
Source code in src/symfonic/core/observability/debug_exchanges.py
| def exchanges_for_rows(
rows: list[dict[str, Any]],
) -> list[tuple[str | None, str | None]]:
"""Align (prompt, response) pairs onto one run's snapshot rows.
``rows`` must be the timestamp-ordered snapshots of a single run
group (as ``build_conversation_detail`` groups them). Position is
the join, ``input_tokens`` the verification: a mismatch attaches
nothing rather than a neighboring call's text (fail-closed on ring
skew, e.g. the flag toggled mid-process). All-``None`` when the
flag is off or nothing was captured.
"""
empty: list[tuple[str | None, str | None]] = [(None, None)] * len(rows)
if not rows or not debug_exchanges_enabled():
return empty
run_ids = {str(r.get("run_id") or r.get("call_id") or "") for r in rows}
if len(run_ids) != 1:
return empty
(run_id,) = run_ids
if not run_id:
return empty
by_ordinal = {e["ordinal"]: e for e in get_exchanges(run_id)}
aligned: list[tuple[str | None, str | None]] = []
for ordinal, row in enumerate(rows):
entry = by_ordinal.get(ordinal)
if entry is None:
aligned.append((None, None))
continue
if int(row.get("input_tokens", 0) or 0) != entry["input_tokens"]:
aligned.append((None, None))
continue
aligned.append((entry["prompt"], entry["response"]))
return aligned
|
get_exchanges
get_exchanges(run_id: str) -> list[dict[str, Any]]
Captured exchanges for one run, oldest first (debug flag only).
Source code in src/symfonic/core/observability/debug_exchanges.py
| def get_exchanges(run_id: str) -> list[dict[str, Any]]:
"""Captured exchanges for one run, oldest first (debug flag only)."""
if not debug_exchanges_enabled():
return []
return [entry for entry in _ring if entry["run_id"] == run_id]
|
prompt_text
prompt_text(transcript: Any) -> str
Best-effort transcript rendering; never raises, never empty.
Source code in src/symfonic/core/observability/debug_exchanges.py
| def prompt_text(transcript: Any) -> str:
"""Best-effort transcript rendering; never raises, never empty."""
if transcript is None:
return "(no transcript captured)"
try:
messages = _transcript_messages(transcript)
if isinstance(messages, str):
text = messages
else:
text = "\n\n".join(_message_text(m) for m in messages)
except Exception: # noqa: BLE001 - debug rendering must not fail turns
text = f"(unrenderable transcript: {type(transcript).__name__})"
if len(text) > MAX_PROMPT_CHARS:
text = text[:MAX_PROMPT_CHARS] + "\n...[prompt truncated]"
return text or "(empty transcript)"
|
response_text
response_text(turn: Any) -> str
Best-effort model-turn rendering; never raises, never empty.
A tool-only round has no visible text — the tool requests are
the response — so they render as a section rather than vanishing.
Source code in src/symfonic/core/observability/debug_exchanges.py
| def response_text(turn: Any) -> str:
"""Best-effort model-turn rendering; never raises, never empty.
A tool-only round has no visible text — the tool requests *are*
the response — so they render as a section rather than vanishing.
"""
if turn is None:
return "(no turn captured)"
try:
text = str(getattr(turn, "text", "") or "")
requests = getattr(turn, "tool_requests", None) or ()
if requests:
lines = []
for req in requests:
name = getattr(req, "name", "?")
args = getattr(req, "arguments", {})
try:
rendered_args = json.dumps(dict(args), ensure_ascii=False)
except (TypeError, ValueError):
rendered_args = str(args)
lines.append(f"- {name}({rendered_args})")
text = f"{text}\n\n[tool calls]\n" + "\n".join(lines)
text = text.strip()
stop = getattr(turn, "stop_reason", None)
if stop and not text:
text = f"(no text; stop reason: {stop})"
except Exception: # noqa: BLE001 - debug rendering must not fail turns
text = f"(unrenderable turn: {type(turn).__name__})"
if len(text) > MAX_RESPONSE_CHARS:
text = text[:MAX_RESPONSE_CHARS] + "\n...[response truncated]"
return text or "(empty response)"
|