Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -88,8 +88,7 @@ playwright-browsers/

hf-cache/

# Mini Browser's persistent Chromium profile (cookies, signed-in sessions)
# and its cross-process lock file. Its password vault lives in .credentials/
# (ignored above). Never commit.
**/mini_browser_profile/
**/mini_browser_profile.lock
# Mini Browser's persistent Chromium profiles, one per chat (cookies,
# signed-in sessions), and their cross-process lock files. Its password
# vault lives in .credentials/ (ignored above). Never commit.
**/mini_browser_profile*
44 changes: 40 additions & 4 deletions agent_core/core/impl/action/router.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
from agent_core.core.impl.llm.interface import LLMContextOverflowError
from agent_core import get_event_stream_manager
from agent_core.utils.logger import logger
from agent_core.utils.token import count_tokens


def _is_visible_in_mode(action, GUI_mode: bool) -> bool:
Expand Down Expand Up @@ -212,6 +213,36 @@ def render_prompt() -> str:
# Internal helpers
# ------------------------------------------------------------------

def _fold(self, stream, system_prompt: str, static_prompt: Optional[str]) -> None:
"""Fold ``stream``, keeping as much recent history as the budget allows.

A fold keeps context.keep_recent_tokens of recent events, but never
more than half of the room the model's input budget leaves after
what a fold does not shrink: the system prompt, the prompt around
the event stream (``static_prompt``) and the running summary. So
every fold frees at least half of that room for new events, whatever
the model's window.
"""
from app.config import get_keep_recent_tokens

room = (
self.llm_interface.input_budget()
- count_tokens(system_prompt)
- count_tokens(static_prompt or "")
- count_tokens(stream.head_summary or "")
)
stream.summarize_by_LLM(
keep_tokens=min(get_keep_recent_tokens(), max(room, 0) // 2)
)

def _checkpoint(self, stream, task_id: str, call_type: str) -> None:
"""Tell the LLM interface how far the session history now covers the
stream (it saves the history as of this point, so a restart can
continue the session on the provider's cached prefix)."""
self.llm_interface.checkpoint_session(
task_id, call_type, stream.sync_marker(call_type) if stream else None
)

async def _prompt_for_decision(
self,
prompt: str,
Expand All @@ -237,7 +268,7 @@ async def _prompt_for_decision(
max_retries = 3
last_error: Optional[Exception] = None
current_prompt = prompt
# One fold per decision: a fold keeps at most keep_recent_tokens of what
# One fold per decision: a fold keeps at most its keep target of what
# the model has read (new events count against it), so a request that
# still does not fit is carrying new events too large to send in full.
# Those are collapsed to pointers once; if it still does not fit, that
Expand Down Expand Up @@ -315,7 +346,7 @@ async def _prompt_for_decision(
f"[SESSION CACHE] Next request exceeds the context "
f"budget; folding the event stream for {call_type}"
)
stream.summarize_by_LLM()
self._fold(stream, system_prompt, static_prompt)
folded = True
current_prompt = render_prompt()
self.context_engine.reset_event_stream_sync(
Expand All @@ -338,6 +369,7 @@ async def _prompt_for_decision(
self.context_engine.mark_event_stream_synced(
call_type, session_id=effective_session_id
)
self._checkpoint(stream, current_task_id, call_type)
else:
# No new events - this could mean summarization happened
logger.info(
Expand Down Expand Up @@ -373,10 +405,14 @@ async def _prompt_for_decision(
self.context_engine.mark_event_stream_synced(
call_type, session_id=effective_session_id
)
self._checkpoint(stream, current_task_id, call_type)
else:
# No session registered (simple task) - use prefix cache / regular response
raw_response = await self.llm_interface.generate_response_async(
system_prompt, current_prompt, prompt_name=prompt_name
system_prompt,
current_prompt,
prompt_name=prompt_name,
usage_key=f"{current_task_id}:{call_type}",
)
# The prompt carried the whole stream; a later fold may
# collapse any of it now.
Expand Down Expand Up @@ -442,7 +478,7 @@ async def _prompt_for_decision(
f"[SESSION CACHE] Request exceeded the context budget; folding "
f"the event stream and retrying once for {call_type}"
)
stream.summarize_by_LLM()
self._fold(stream, system_prompt, static_prompt)
folded = True
elif not stream.collapse_undelivered_oversize():
# Folded, and the new events have nothing left to shrink.
Expand Down
124 changes: 76 additions & 48 deletions agent_core/core/impl/event_stream/event_stream.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,15 +10,16 @@
log(kind, message, severity="INFO") -> int (event index)
to_prompt_snapshot(max_events=60, include_summary=True) -> str
summarize_by_rule() # force summarization of oldest chunk
summarize_by_LLM() # force summarization of oldest chunk
summarize_by_LLM(keep_tokens) # force summarization of oldest chunk
"""

from __future__ import annotations
from datetime import datetime, timezone
import hashlib
import re
import time
from pathlib import Path
from typing import List, Optional, Set, Tuple
from typing import Any, Callable, Dict, List, Optional, Set, Tuple
from agent_core.core.event_stream.event import Event, EventRecord, EventType
from agent_core.core.protocols.llm import LLMInterfaceProtocol
from agent_core.core.prompts import EVENT_STREAM_SUMMARIZATION_PROMPT
Expand All @@ -31,27 +32,6 @@
SEVERITIES = ("DEBUG", "INFO", "WARN", "ERROR")


def _configured_keep_recent_tokens() -> int:
"""Read context.keep_recent_tokens from settings.json.

app.config owns the defaults and already absorbs a missing file, bad JSON
and out-of-range values, so there is nothing left to guard here — a raised
ImportError means the app package is genuinely gone, which is a broken
install, not a case to paper over with a second copy of the numbers.

The import is deferred because app.config imports agent_core at module
scope (``from agent_core import get_credential``), and agent_core's
__init__ loads this module — a module-scope import would close that cycle.
By stream-construction time every module is loaded and the import is fine.

Read once per stream, so a settings.json edit applies to sessions created
after it; the main session's stream needs a restart.
"""
from app.config import get_keep_recent_tokens

return get_keep_recent_tokens()


# Messages longer than this are externalized to a temp file and replaced with a
# pointer (+keywords) so a single large action output (e.g. get_notion, read_pdf,
# an http_request body) can't bloat the prompt. ~8000 chars ≈ ~2000 tokens; the
Expand Down Expand Up @@ -118,15 +98,13 @@ def __init__(
llm: LLMInterfaceProtocol,
temp_dir: Path | None = None,
) -> None:
# The stream never decides to fold on its own. The router asks for a
# fold (summarize_by_LLM) when the NEXT REQUEST would not fit the
# context budget; the only stream-side setting is how much recent
# history a fold keeps. Tests pin it by patching
# _configured_keep_recent_tokens (see the event_stream_limits fixture).
# The stream never decides to fold on its own, nor how much a fold
# keeps. The router asks for a fold (summarize_by_LLM) when the NEXT
# REQUEST would not fit the model's context budget, and says how many
# recent tokens to keep.
self.head_summary: Optional[str] = None
self.llm = llm
self.tail_events: List[EventRecord] = []
self.tail_keep_after_summarize_tokens = _configured_keep_recent_tokens()
self.temp_dir = temp_dir

self._lock = threading.RLock()
Expand Down Expand Up @@ -442,18 +420,17 @@ def _find_token_cutoff(self, events: List[EventRecord], keep_tokens: int) -> int
)
return cutoff

def _fold_cutoff(self) -> int:
def _fold_cutoff(self, keep_tokens: int) -> int:
"""How many leading tail events a fold may collapse.

The keep-recent window decides, except that a fold never reaches an
event the model has not received. The router asks for a fold right
after actions return, before their results are sent; folding those
results then hands the agent a summary or a pointer of output it never
saw, and it runs the same read again, which overflows again.
The keep-recent window (``keep_tokens``) decides, except that a fold
never reaches an event the model has not received. The router asks
for a fold right after actions return, before their results are sent;
folding those results then hands the agent a summary or a pointer of
output it never saw, and it runs the same read again, which overflows
again.
"""
cutoff = self._find_token_cutoff(
self.tail_events, self.tail_keep_after_summarize_tokens
)
cutoff = self._find_token_cutoff(self.tail_events, keep_tokens)
first_undelivered = next(
(i for i, r in enumerate(self.tail_events) if not r.delivered),
len(self.tail_events),
Expand Down Expand Up @@ -527,7 +504,7 @@ def collapse_undelivered_oversize(self) -> int:
"""Externalize oversized events the model has not received yet.

For the router only, when a request still does not fit after a fold:
the fold keeps at most keep_recent_tokens of what the model had read
the fold keeps at most its keep target of what the model had read
(new events count against it), so the events it has not received are
too large to send in full. Pointers are the only way to hand them over.

Expand All @@ -545,28 +522,27 @@ def collapse_undelivered_oversize(self) -> int:
)
return reclaimed

def summarize_by_LLM(self) -> None:
def summarize_by_LLM(self, keep_tokens: int) -> None:
"""
Summarize the oldest tail events using the language model.
Summarize the oldest tail events using the language model, keeping
about ``keep_tokens`` of the most recent events verbatim.

This is a SYNCHRONOUS blocking call that holds the lock for the entire
duration, including the LLM call. This ensures no events can be added
while summarization is in progress.

Called from log() which already holds the lock (RLock allows reentry).
"""
if not self.tail_events:
return

# Find cutoff based on tokens to keep (never past what the model has read)
cutoff = self._fold_cutoff()
cutoff = self._fold_cutoff(keep_tokens)

# Collapse anything oversized that the recent-event pin is holding
# verbatim BEFORE deciding whether an LLM call is warranted — that alone
# often drops the stream back under the threshold for free.
if self._shrink_pinned_oversize(cutoff):
# Budget changed; the fold boundary moves with it.
cutoff = self._fold_cutoff()
cutoff = self._fold_cutoff(keep_tokens)

if cutoff <= 0:
# Nothing old enough to summarize
Expand Down Expand Up @@ -811,21 +787,73 @@ def clear(self) -> None:
self.tail_events.clear()
self._total_tokens = 0
self._session_sync_points.clear()
# A new stream: its first event gets a fresh datetime marker.
self._last_datetime_ts = None

# ───────────────────── Session Cache Delta Tracking ─────────────────────

def mark_delivered(self) -> None:
"""Record that the model has received every event now in the tail.
def mark_delivered(self, up_to: Optional[int] = None) -> None:
"""Record that the model has received every event now in the tail
(or the first ``up_to`` of them).

Called once a request carrying the stream has been answered: through
mark_session_synced on the session-cache paths, and by the router
after a stateless request. Events logged later stay undelivered until
the next answered request, so a fold cannot collapse them first.
"""
with self._lock:
for rec in self.tail_events:
for rec in self.tail_events[:up_to]:
rec.delivered = True

def sync_marker(self, call_type: str) -> Optional[Dict[str, Any]]:
"""A session cache's sync point in a form that can be checked after
a restart: the point, the timestamps of the first and last event
before it, and a digest of the head summary. None when the call type
has no sync point."""
with self._lock:
point = self._session_sync_points.get(call_type)
return None if point is None else self._marker_at(point)

def restore_sync_point(self, call_type: str, marker: Dict[str, Any]) -> bool:
"""Put back a sync point saved with sync_marker, but only if this
stream still holds the same events before it (same summary, same
first and last event). Otherwise there is none, and the next request
sends the whole stream."""
with self._lock:
point = marker.get("point")
if not isinstance(point, int) or not 0 <= point <= len(self.tail_events):
return False
if self._marker_at(point) != marker:
return False
self._session_sync_points[call_type] = point
return True

def _marker_at(self, point: int) -> Dict[str, Any]:
"""Caller holds the lock."""
return {
"point": point,
"first": self.tail_events[0].ts.isoformat() if point else None,
"last": self.tail_events[point - 1].ts.isoformat() if point else None,
"summary": hashlib.sha256(
(self.head_summary or "").encode("utf-8")
).hexdigest(),
}

def replace_ui_artifact(
self, artifact_id: str, replace: Callable[[Dict[str, Any]], Dict[str, Any]]
) -> int:
"""Swap the UI artifact payload of every event that carries artifact
``artifact_id`` for ``replace(payload)`` (the user closed it). Returns
how many events carried it."""
count = 0
with self._lock:
for rec in self.tail_events:
artifact = rec.event.ui_artifact
if artifact and artifact.get("id") == artifact_id:
rec.event.ui_artifact = replace(artifact)
count += 1
return count

def mark_session_synced(self, call_type: str) -> None:
"""
Mark that all current events have been synced to the session cache.
Expand Down
19 changes: 19 additions & 0 deletions agent_core/core/impl/event_stream/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,25 @@ def _should_skip_event_type(self, kind: str) -> bool:
"""
return kind in SKIP_UNPROCESSED_EVENT_TYPES

def clear_event_log(self, session_id: str) -> None:
"""Empty a session's EVENT.md (the session is starting over).

The agent greps EVENT.md to recover detail its context no longer
holds, so a cleared conversation must not stay there. Its
EVENT_UNPROCESSED.md is kept: those events still go to long-term
memory at the next memory run.
"""
stream = self._streams.get(session_id)
if stream is None or stream.temp_dir is None:
return
with self._file_lock:
event_file = Path(stream.temp_dir) / "EVENT.md"
try:
if event_file.exists():
event_file.write_text("", encoding="utf-8")
except OSError as e:
logger.warning(f"[EventStreamManager] Failed to clear EVENT.md: {e}")

def _log_to_files(self, kind: str, message: str, temp_dir: Optional[Path]) -> None:
"""
Append an event to the writing session's EVENT.md and (optionally)
Expand Down
Loading