"""Module-level helpers, constants and the event-sink wrapper the agent uses."""
import asyncio
from navi.llm.base import Message
# Sentinel: placed in the event sink by the tool wrapper to signal completion.
_TOOL_DONE = object()
def _inline_images_notice(total: int, in_context: int, paths: list[str]) -> str:
"""The line the model reads next to the images it was sent.
`paths` is index-aligned with the images and holds an empty string for one that
could not be stored. The notice always says the files exist, because a model told
the session directory exists and given no path will go looking for one — that hunt
is exactly what this line is here to prevent.
"""
noun = "image" if total == 1 else "images"
if in_context < total:
notice = (
f"[{total} {noun} attached inline — {in_context} are in your context, "
"the rest were not loaded"
)
dropped = [path for path in paths[in_context:] if path]
if dropped:
notice += f"; open them with image_view: {', '.join(dropped)}"
else:
notice += " and could not be saved to the session directory"
else:
notice = (
f"[{total} {noun} attached inline — already in your context, "
"no extra loading needed"
)
stored = [path for path in paths if path]
if stored:
notice += f". On disk: {', '.join(stored)}"
return notice + "]"
class _TaggedSink:
"""Routes one tool's live events into the shared batch queue, tagged by index.
Mirrors the asyncio.Queue surface the sinks actually use (``put`` /
``put_nowait``) so it is a drop-in replacement for a per-tool queue.
"""
def __init__(self, queue: asyncio.Queue, idx: int) -> None:
self._queue = queue
self._idx = idx
async def put(self, item) -> None:
await self._queue.put((self._idx, item))
def put_nowait(self, item) -> None:
self._queue.put_nowait((self._idx, item))
async def _todo_progress_message(
session_id: str, *, first_iteration: bool = False
) -> "Message | None":
"""Build a compact system reminder with current todo state and update discipline."""
from navi.tools.todo import get_progress_message
return await get_progress_message(session_id, first_iteration=first_iteration)
async def _protocol_progress_message(session_id: str) -> "Message | None":
"""System reminder for an active protocol run, or None.
The run was started in some earlier turn; compression may drop that turn and
the todo lines no longer say which of them the user pre-agreed to. This
reminder keeps the commitment — and the way to close it — in view.
"""
from navi.tools.protocol import get_protocol_message
return await get_protocol_message(session_id)
# Final-turn intercept nudges — injected when the model ends a turn with bare
# text (no tool call) while the todo still has open steps (or a protocol run is
# still open), then the loop is continued instead of closed. Escalating tone:
# 1st is a soft pointer to the available recovery tools, 2nd is a hard "act
# now, no bare text" instruction.
# Indexed by turn_ctx.final_interceptions (clamped to the last/hardest).
_FINAL_INTERCEPT_NUDGES = (
"[Final-turn check] You ended the turn with text only, but the todo list still has open steps "
"(pending/in_progress). Do not stop with bare text while work remains. Either mark finished "
"steps done/skipped via the todo tool, reflect on what is blocking, re-plan with the `plan` "
"tool if the approach is dead, or call a tool to continue — then keep going.",
"[Final-turn check — second stop] You already stopped once without acting and the todo still "
"has open steps. You must act now: call a tool — todo (close steps as done/skipped/failed with "
"validation), reflect, `plan` (re-plan with a reason), or continue execution. Do not output "
"text without a tool call.",
)
# Same intercept, different move: the todo shows no open work, but a protocol
# run is still `running` because it was never closed.
_PROTOCOL_FINAL_NUDGE = (
"[Final-turn check] You ended the turn with text only while a protocol run is still "
"open. Close it now: protocol(op='finish', report=...) if its steps are done, or "
"protocol(op='cancel'/'fail', reason=...) if not. Do not leave a run running."
)