async def complete_terminal(
plan: Any,
ctx: Any,
transcript: Any,
request: TurnRequest,
kernel: Any,
last_turn: ModelTurn,
*,
stream_model: bool,
observed: Any,
emitter: Any,
publish: Any,
before_effect: Any = None,
) -> tuple[list[KernelEvent], ModelTurn | None]:
"""Return events and final turn; provider failure preserves the original."""
if (
stream_model
or last_turn.text
or plan.limits.max_terminal_synthesis_calls != 1
or plan.bindings.terminal_model is None
):
return [], None
from symfonic.kernel._effect_boundary import check_effect
adapter = plan.bindings.terminal_model
await check_effect(before_effect, lifecycle=ctx.lifecycle)
messages = adapter.messages_for(transcript)
pre_traces = await ctx.lifecycle.bounded(
kernel.dispatch_pre_model(
plan,
request,
resolved=ctx.resolved,
transcript=replace(transcript, wire=messages, typed=list(transcript.typed)),
),
doing="governing a terminal model request",
)
events = [await publish(event) for event in observed.stage_events(pre_traces)]
await check_effect(before_effect, lifecycle=ctx.lifecycle)
try:
turn = await ctx.lifecycle.bounded(
adapter.invoke(transcript, messages),
doing="synthesizing an empty final answer",
)
except asyncio.CancelledError:
raise
except Exception as exc:
if ctx.lifecycle.owns_timeout(exc):
raise
# The original empty turn remains the truthful fallback. The model
# adapter records the attempted provider call for private telemetry.
return events, None
# Extraction directives are meaningful even when their visible text was
# filtered to empty. Preserve that turn for the configured extraction and
# finalizers without turning a hidden directive into a text delta.
has_final_content = bool(turn.text or turn.extraction_seen or turn.extraction_ops)
# An empty reply still consumed a provider call, so post-model observes it
# exactly once. It has nothing to keep, so stages with a memory effect do
# not see it: they would file an empty answer the turn never gave.
traces = await ctx.lifecycle.bounded(
kernel.dispatch_post_model(
plan if has_final_content else _memory_free_post_model(plan),
request,
turn,
resolved=ctx.resolved,
transcript=transcript,
),
doing="governing a terminal answer",
)
for staged in observed.stage_events(traces):
events.append(await publish(staged))
if has_final_content:
await check_effect(before_effect, lifecycle=ctx.lifecycle)
plan.bindings.conversation.close_round(transcript, turn, (), ())
if turn.text:
events.append(await publish(emitter.make("text_delta", text=turn.text)))
ctx.absorb(turn.text, turn.usage, ctx.stop_reason or turn.stop_reason)
return events, turn if has_final_content else None