feat(context): model-aware conversation compaction for long scans (#881)

strix/core/execution.py

@@ -22,6 +22,7 @@ from strix.core.sessions import (
     open_agent_session,
     strip_all_images_from_session,
 )
+from strix.llm.compaction import is_context_overflow, maybe_compact
 
 
 if TYPE_CHECKING:
@@ -40,6 +41,41 @@ logger = logging.getLogger(__name__)
 StreamEventSink = Callable[[str, Any], None]
 
 _INPUT_REJECTION_CODES = frozenset({400, 404, 422})
+_MAX_COMPACTIONS_PER_CYCLE = 2
+
+
+def _run_config_model(run_config: RunConfig) -> str | None:
+    return run_config.model if isinstance(run_config.model, str) else None
+
+
+def _agent_instructions(agent: Any) -> str:
+    instructions = getattr(agent, "instructions", None)
+    return instructions if isinstance(instructions, str) else ""
+
+
+def _agent_tools_text(agent: Any) -> str:
+    parts: list[str] = []
+    for tool in getattr(agent, "tools", []) or []:
+        name = getattr(tool, "name", "")
+        description = getattr(tool, "description", "") or ""
+        schema = getattr(tool, "params_json_schema", "") or ""
+        parts.append(f"{name} {description} {schema}")
+    return "\n".join(parts)
+
+
+async def _compact_session(
+    agent: Any, session: Session, run_config: RunConfig, *, force: bool
+) -> bool:
+    model = _run_config_model(run_config)
+    if session is None or model is None:
+        return False
+    return await maybe_compact(
+        session,
+        model=model,
+        instructions=_agent_instructions(agent),
+        tools_text=_agent_tools_text(agent),
+        force=force,
+    )
 
 
 async def run_agent_loop(
@@ -350,6 +386,7 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
     hooks: RunHooks[dict[str, Any]] | None,
 ) -> RunResultBase | None:
     image_strips = 0
+    compactions = 0
     while True:
         try:
             await coordinator.mark_running(agent_id)
@@ -360,6 +397,10 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                         await enforce_image_budget(session, max_images)
                     except Exception:
                         logger.exception("image-budget enforcement failed for %s", agent_id)
+                try:
+                    await _compact_session(agent, session, run_config, force=False)
+                except Exception:
+                    logger.exception("proactive compaction failed for %s", agent_id)
             stream = Runner.run_streamed(
                 agent,
                 input=input_data,
@@ -428,6 +469,25 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                     )
                     input_data = []
                     continue
+            if (
+                compactions < _MAX_COMPACTIONS_PER_CYCLE
+                and session is not None
+                and is_context_overflow(exc)
+            ):
+                try:
+                    compacted = await _compact_session(agent, session, run_config, force=True)
+                except Exception:
+                    logger.exception("overflow compaction recovery failed for %s", agent_id)
+                    compacted = False
+                if compacted:
+                    compactions += 1
+                    logger.info(
+                        "Compacted %s session after context overflow; retrying (%d)",
+                        agent_id,
+                        compactions,
+                    )
+                    input_data = []
+                    continue
             if not interactive:
                 raise
             if isinstance(exc, MaxTurnsExceeded):

strix/core/sessions.py

@@ -92,6 +92,39 @@ async def _rewrite_session(
         return True
 
 
+async def replace_session_items(
+    session: Session,
+    new_items: list[Any],
+    *,
+    expected_len: int | None = None,
+) -> bool:
+    """Overwrite the session's items, restoring the originals on failure.
+
+    When ``expected_len`` is given, the rewrite is skipped if the session no
+    longer has that many items (a concurrent writer changed it), so a slow
+    compaction summary can't clobber newer turns.
+    """
+    async with session_write_lock(session):
+        original = list(await session.get_items())
+        if expected_len is not None and len(original) != expected_len:
+            logger.warning(
+                "skipping session rewrite: expected %d items, found %d",
+                expected_len,
+                len(original),
+            )
+            return False
+        rebuilt = cast("list[TResponseInputItem]", new_items)
+        await session.clear_session()
+        try:
+            await session.add_items(rebuilt)
+        except Exception:
+            logger.exception("session rewrite failed; restoring original items")
+            await session.clear_session()
+            await session.add_items(original)
+            raise
+        return True
+
+
 async def strip_all_images_from_session(session: Session) -> bool:
     """Replace every image tool output with a text placeholder (rejection recovery)."""