"""Turn lifecycle: run, ephemeral runs, compaction and prompt-cache invalidation."""
from typing import TYPE_CHECKING
from collections.abc import AsyncGenerator
import structlog
from navi.config import settings
from navi.exceptions import (
NothingToCompactError,
SessionNotFound,
)
from navi.llm.base import Message
from ..events import (
AgentEvent,
CompressionStarted,
StreamEnd,
)
from ..subagent_runner import SubAgentOutcome
log = structlog.get_logger()
class AgentRuntimeMixin:
async def run(
self,
session_id: str,
user_message: str,
images: list[str] | None = None,
files: list[dict] | None = None,
is_recall: bool = False,
) -> str:
"""Non-streaming: run the full tool-calling loop and return the final text."""
full_content = ""
async for event in self.run_stream(
session_id, user_message, images=images, files=files, is_recall=is_recall
):
if isinstance(event, StreamEnd):
full_content = event.full_content or ""
return full_content
async def _drain_task_notes(self, session_id: str, session) -> None:
"""Move finished-background-task notes into the session context.
Called once at the start of run_stream(), before the first LLM call:
the note becomes part of the LLM-visible context (and the compression
window) but is never displayed — clients learned about completions via
the live task_update event. Only the foreground run_stream mutates the
session, so there is no race with background jobs.
"""
from navi.core import task_notes
try:
note = await task_notes.drain(session_id)
except Exception:
log.exception("agent.task_notes_drain_failed", session_id=session_id)
return
if not note:
return
session.context.append(
Message(
role="system",
content=note,
metadata={"source": "task_note"},
is_display=False,
)
)
await self._sessions.save(session)
log.info("agent.task_notes_drained", session_id=session_id)
async def run_ephemeral(
self,
user_message: str,
profile_id: str,
max_iterations: int = 40,
exclude_tools: list[str] | None = None,
briefing: str | None = None,
custom_system_prompt: str | None = None,
inherit_system_prompt: bool = False,
context_transfer: str | None = None,
parent_session_id: str | None = None,
timeout_seconds: float = 300.0,
) -> SubAgentOutcome:
"""Delegate to SubAgentRunner."""
return await self._subagent.run(
user_message=user_message,
profile_id=profile_id,
max_iterations=max_iterations,
exclude_tools=exclude_tools,
briefing=briefing,
custom_system_prompt=custom_system_prompt,
inherit_system_prompt=inherit_system_prompt,
context_transfer=context_transfer,
parent_session_id=parent_session_id,
timeout_seconds=timeout_seconds,
)
async def compact_stream(self, session_id: str) -> AsyncGenerator[AgentEvent, None]:
"""Force context compression now, bypassing the token threshold.
Triggered by the TUI ``/compact`` command. Unlike the pre/mid-turn
auto-compression (gated by ``should_compress``), this always attempts a
summary pass as long as there is enough to summarize. Yields
``CompressionStarted`` then ``ContextCompressed`` on success; raises
``NothingToCompactError`` when the context is too small to compress
(the orchestrator surfaces it as an error event so the user gets
feedback).
Emits NO ``StreamStart``/``StreamEnd`` — the client distinguishes a
forced compact from an in-turn auto-compress by the absence of a
streaming turn (the TUI keys off its ``_streaming`` flag).
"""
from navi.tools._internal.base import current_session_id as _sid_var
session = await self._sessions.get(session_id)
if session is None:
raise SessionNotFound(session_id)
profile = self._profiles.get(session.profile_id)
self._set_active_profile(profile)
llm = self._get_backend(profile.llm_backend)
_sid_token = _sid_var.set(session_id)
try:
if not settings.context_compression_enabled:
raise NothingToCompactError("Context compression is disabled.")
window = self._window_for(profile)
yield CompressionStarted(
# Real-baseline estimate when available (see _compression_events_preturn);
# the calibrated estimate is only the fallback before the first LLM
# call of a session.
context_tokens=self._compressor.real_baseline_estimate(
session.context, session.context, profile.model
),
max_context_tokens=window,
)
event = await self._compressor.compress_and_save_session(
session=session,
session_store=self._sessions,
llm=llm,
model=profile.model,
temperature=settings.context_summary_temperature,
session_id=session_id,
reason="forced",
memory_store=self._memory,
keep_recent=settings.context_keep_recent,
max_tokens=settings.context_summary_max_tokens,
# Enable the intra-turn fallback so a single long autonomous turn
# (one user message + many tool iterations = one "turn", the
# common navi_code shape) can still be compressed — without this
# keep_recent_messages, partition_messages finds nothing to
# summarize when turns <= keep_recent and forced compact always
# reports "nothing to compact". Mirrors the midturn auto-compress.
keep_recent_messages=max(12, settings.context_keep_recent * 2),
window=window,
)
if event is None:
raise NothingToCompactError(
"Nothing to compact yet — the context is still small."
)
yield event
finally:
_sid_var.reset(_sid_token)
def _set_active_profile(self, profile) -> None:
"""Update cached profile and propagate to subsystems that depend on it."""
self._compressor.set_profile(profile)
self._subagent.set_profile(profile)
def invalidate_prompt_cache(self, profile_id: str | None = None) -> None:
"""Drop cached system prompts so a profile change reaches the next turn.
The cache lives on this agent's ContextBuilder, and the agent handed out
by the container is process-wide — so an admin toggling a profile's
availability must clear it, or every other prompt keeps advertising the
profile list as it was.
"""
self._ctx_builder.invalidate_system_prompt_cache(profile_id)
