Skip to content

refactor(memory): remove flush_memories entirely #15696

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
teknium1 merged 1 commit into from
Apr 25, 2026
Merged

refactor(memory): remove flush_memories entirely #15696

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
6 changes: 3 additions & 3 deletions agent/anthropic_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1680,9 +1680,9 @@ def build_anthropic_kwargs(

# ── Strip sampling params on 4.7+ ─────────────────────────────────
# Opus 4.7 rejects any non-default temperature/top_p/top_k with a 400.
# Callers (auxiliary_client, flush_memories, etc.) may set these for
# older models; drop them here as a safety net so upstream 4.6 → 4.7
# migrations don't require coordinated edits everywhere.
# Callers (auxiliary_client, etc.) may set these for older models;
# drop them here as a safety net so upstream 4.6 → 4.7 migrations
# don't require coordinated edits everywhere.
if _forbids_sampling_params(model):
for _sampling_key in ("temperature", "top_p", "top_k"):
kwargs.pop(_sampling_key, None)
Expand Down
8 changes: 4 additions & 4 deletions agent/auxiliary_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,7 @@ def create(self, **kwargs) -> Any:
# Note: the Codex endpoint (chatgpt.com/backend-api/codex) does NOT
# support max_output_tokens or temperature — omit to avoid 400 errors.

# Tools support for flush_memories and similar callers
# Tools support for auxiliary callers (e.g. skills_hub) that pass function schemas
tools = kwargs.get("tools")
if tools:
converted = []
Expand Down Expand Up @@ -2803,8 +2803,8 @@ def _build_call_kwargs(
temperature = fixed_temperature

# Opus 4.7+ rejects any non-default temperature/top_p/top_k — silently
# drop here so auxiliary callers that hardcode temperature (e.g. 0.3 on
# flush_memories, 0 on structured-JSON extraction) don't 400 the moment
# drop here so auxiliary callers that hardcode temperature (e.g. 0 on
# structured-JSON extraction) don't 400 the moment
# the aux model is flipped to 4.7.
if temperature is not None:
from agent.anthropic_adapter import _forbids_sampling_params
Expand Down Expand Up @@ -2892,7 +2892,7 @@ def call_llm(

Args:
task: Auxiliary task name ("compression", "vision", "web_extract",
"session_search", "skills_hub", "mcp", "flush_memories").
"session_search", "skills_hub", "mcp", "title_generation").
Reads provider:model from config/env. Ignored if provider is set.
provider: Explicit provider override.
model: Explicit model override.
Expand Down
10 changes: 0 additions & 10 deletions cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4668,10 +4668,6 @@ def _notify_session_boundary(self, event_type: str) -> None:
def new_session(self, silent=False):
"""Start a fresh session with a new session ID and cleared agent state."""
if self.agent and self.conversation_history:
try:
self.agent.flush_memories(self.conversation_history)
except (Exception, KeyboardInterrupt):
pass
# Trigger memory extraction on the old session before session_id rotates.
self.agent.commit_memory_session(self.conversation_history)
self._notify_session_boundary("on_session_finalize")
Expand Down Expand Up @@ -10788,12 +10784,6 @@ def _suppress_closed_loop_errors(loop, context):
self.agent.interrupt()
except Exception:
pass
# Flush memories before exit (only for substantial conversations)
if self.agent and self.conversation_history:
try:
self.agent.flush_memories(self.conversation_history)
except (Exception, KeyboardInterrupt):
pass
# Shut down voice recorder (release persistent audio stream)
if hasattr(self, '_voice_recorder') and self._voice_recorder:
try:
Expand Down
215 changes: 36 additions & 179 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -524,7 +524,7 @@ def _load_gateway_config() -> dict:
def _resolve_gateway_model(config: dict | None = None) -> str:
"""Read model from config.yaml — single source of truth.

Without this, temporary AIAgent instances (memory flush, /compress) fall
Without this, temporary AIAgent instances (e.g. /compress) fall
back to the hardcoded default which fails when the active provider is
openai-codex.
"""
Expand Down Expand Up @@ -915,129 +915,6 @@ async def _safe_adapter_disconnect(self, adapter, platform) -> None:
e,
)

# -----------------------------------------------------------------

def _flush_memories_for_session(
self,
old_session_id: str,
session_key: Optional[str] = None,
):
"""Prompt the agent to save memories/skills before context is lost.

Synchronous worker — meant to be called via run_in_executor from
an async context so it doesn't block the event loop.
"""
# Skip cron sessions — they run headless with no meaningful user
# conversation to extract memories from.
if old_session_id and old_session_id.startswith("cron_"):
logger.debug("Skipping memory flush for cron session: %s", old_session_id)
return

try:
history = self.session_store.load_transcript(old_session_id)
if not history or len(history) < 4:
return

from run_agent import AIAgent
model, runtime_kwargs = self._resolve_session_agent_runtime(
session_key=session_key,
)
if not runtime_kwargs.get("api_key"):
return

tmp_agent = AIAgent(
**runtime_kwargs,
model=model,
max_iterations=8,
quiet_mode=True,
skip_memory=True, # Flush agent — no memory provider
enabled_toolsets=["memory", "skills"],
session_id=old_session_id,
)
try:
# Fully silence the flush agent — quiet_mode only suppresses init
# messages; tool call output still leaks to the terminal through
# _safe_print → _print_fn. Set a no-op to prevent that.
tmp_agent._print_fn = lambda *a, **kw: None

# Build conversation history from transcript
msgs = [
{"role": m.get("role"), "content": m.get("content")}
for m in history
if m.get("role") in ("user", "assistant") and m.get("content")
]

# Read live memory state from disk so the flush agent can see
# what's already saved and avoid overwriting newer entries.
_current_memory = ""
try:
from tools.memory_tool import get_memory_dir
_mem_dir = get_memory_dir()
for fname, label in [
("MEMORY.md", "MEMORY (your personal notes)"),
("USER.md", "USER PROFILE (who the user is)"),
]:
fpath = _mem_dir / fname
if fpath.exists():
content = fpath.read_text(encoding="utf-8").strip()
if content:
_current_memory += f"\n\n## Current {label}:\n{content}"
except Exception:
pass # Non-fatal — flush still works, just without the guard

# Give the agent a real turn to think about what to save
flush_prompt = (
"[System: This session is about to be automatically reset due to "
"inactivity or a scheduled daily reset. The conversation context "
"will be cleared after this turn.\n\n"
"Review the conversation above and:\n"
"1. Save any important facts, preferences, or decisions to memory "
"(user profile or your notes) that would be useful in future sessions.\n"
"2. If you discovered a reusable workflow or solved a non-trivial "
"problem, consider saving it as a skill.\n"
"3. If nothing is worth saving, that's fine — just skip.\n\n"
)

if _current_memory:
flush_prompt += (
"IMPORTANT — here is the current live state of memory. Other "
"sessions, cron jobs, or the user may have updated it since this "
"conversation ended. Do NOT overwrite or remove entries unless "
"the conversation above reveals something that genuinely "
"supersedes them. Only add new information that is not already "
"captured below."
f"{_current_memory}\n\n"
)

flush_prompt += (
"Do NOT respond to the user. Just use the memory and skill_manage "
"tools if needed, then stop.]"
)

tmp_agent.run_conversation(
user_message=flush_prompt,
conversation_history=msgs,
)
finally:
self._cleanup_agent_resources(tmp_agent)
logger.info("Pre-reset memory flush completed for session %s", old_session_id)
except Exception as e:
logger.debug("Pre-reset memory flush failed for session %s: %s", old_session_id, e)

async def _async_flush_memories(
self,
old_session_id: str,
session_key: Optional[str] = None,
):
"""Run the sync memory flush in a thread pool so it won't block the event loop."""
loop = asyncio.get_running_loop()
await loop.run_in_executor(
None,
self._flush_memories_for_session,
old_session_id,
session_key,
)

@property
def should_exit_cleanly(self) -> bool:
return self._exit_cleanly
Expand Down Expand Up @@ -2272,7 +2149,7 @@ async def start(self) -> bool:
except Exception as e:
logger.error("Recovered watcher setup error: %s", e)

# Start background session expiry watcher for proactive memory flushing
# Start background session expiry watcher to finalize expired sessions
asyncio.create_task(self._session_expiry_watcher())

# Start background reconnection watcher for platforms that failed at startup
Expand All @@ -2289,25 +2166,24 @@ async def start(self) -> bool:
return True

async def _session_expiry_watcher(self, interval: int = 300):
"""Background task that proactively flushes memories for expired sessions.

Runs every `interval` seconds (default 5 min). For each session that
has expired according to its reset policy, flushes memories in a thread
pool and marks the session so it won't be flushed again.
"""Background task that finalizes expired sessions.

This means memories are already saved by the time the user sends their
next message, so there's no blocking delay.
Runs every ``interval`` seconds (default 5 min). For each session
whose reset policy has expired, invokes ``on_session_finalize``
hooks, cleans up the cached AIAgent's tool resources, evicts the
cache entry so it can be garbage-collected, and marks the session
so it won't be finalized again.
"""
await asyncio.sleep(60) # initial delay — let the gateway fully start
_flush_failures: dict[str, int] = {} # session_id -> consecutive failure count
_MAX_FLUSH_RETRIES = 3
_finalize_failures: dict[str, int] = {} # session_id -> consecutive failure count
_MAX_FINALIZE_RETRIES = 3
while self._running:
try:
self.session_store._ensure_loaded()
# Collect expired sessions first, then log a single summary.
_expired_entries = []
for key, entry in list(self.session_store._entries.items()):
if entry.memory_flushed:
if entry.expiry_finalized:
continue
if not self.session_store._is_session_expired(entry):
continue
Expand All @@ -2325,13 +2201,12 @@ async def _session_expiry_watcher(self, interval: int = 300):
f"{p}:{c}" for p, c in sorted(_platforms.items())
)
logger.info(
"Session expiry: %d sessions to flush (%s)",
"Session expiry: %d sessions to finalize (%s)",
len(_expired_entries), _plat_summary,
)

for key, entry in _expired_entries:
try:
await self._async_flush_memories(entry.session_id, key)
try:
from hermes_cli.plugins import invoke_hook as _invoke_hook
_parts = key.split(":")
Expand Down Expand Up @@ -2363,48 +2238,48 @@ async def _session_expiry_watcher(self, interval: int = 300):
# be garbage-collected. Otherwise the cache grows
# unbounded across the gateway's lifetime.
self._evict_cached_agent(key)
# Mark as flushed and persist to disk so the flag
# Mark as finalized and persist to disk so the flag
# survives gateway restarts.
with self.session_store._lock:
entry.memory_flushed = True
entry.expiry_finalized = True
self.session_store._save()
logger.debug(
"Memory flush completed for session %s",
"Session expiry finalized for %s",
entry.session_id,
)
_flush_failures.pop(entry.session_id, None)
_finalize_failures.pop(entry.session_id, None)
except Exception as e:
failures = _flush_failures.get(entry.session_id, 0) + 1
_flush_failures[entry.session_id] = failures
if failures >= _MAX_FLUSH_RETRIES:
failures = _finalize_failures.get(entry.session_id, 0) + 1
_finalize_failures[entry.session_id] = failures
if failures >= _MAX_FINALIZE_RETRIES:
logger.warning(
"Memory flush gave up after %d attempts for %s: %s. "
"Marking as flushed to prevent infinite retry loop.",
"Session finalize gave up after %d attempts for %s: %s. "
"Marking as finalized to prevent infinite retry loop.",
failures, entry.session_id, e,
)
with self.session_store._lock:
entry.memory_flushed = True
entry.expiry_finalized = True
self.session_store._save()
_flush_failures.pop(entry.session_id, None)
_finalize_failures.pop(entry.session_id, None)
else:
logger.debug(
"Memory flush failed (%d/%d) for %s: %s",
failures, _MAX_FLUSH_RETRIES, entry.session_id, e,
"Session finalize failed (%d/%d) for %s: %s",
failures, _MAX_FINALIZE_RETRIES, entry.session_id, e,
)

if _expired_entries:
_flushed = sum(
1 for _, e in _expired_entries if e.memory_flushed
_done = sum(
1 for _, e in _expired_entries if e.expiry_finalized
)
_failed = len(_expired_entries) - _flushed
_failed = len(_expired_entries) - _done
if _failed:
logger.info(
"Session expiry done: %d flushed, %d pending retry",
_flushed, _failed,
"Session expiry done: %d finalized, %d pending retry",
_done, _failed,
)
else:
logger.info(
"Session expiry done: %d flushed", _flushed,
"Session expiry done: %d finalized", _done,
)

# Sweep agents that have been idle beyond the TTL regardless
Expand Down Expand Up @@ -5021,19 +4896,11 @@ async def _handle_reset_command(self, event: MessageEvent) -> str:
# Get existing session key
session_key = self._session_key_for_source(source)
self._invalidate_session_run_generation(session_key, reason="session_reset")

# Flush memories in the background (fire-and-forget) so the user
# gets the "Session reset!" response immediately.
try:
old_entry = self.session_store._entries.get(session_key)
if old_entry:
_flush_task = asyncio.create_task(
self._async_flush_memories(old_entry.session_id, session_key)
)
self._background_tasks.add(_flush_task)
_flush_task.add_done_callback(self._background_tasks.discard)
except Exception as e:
logger.debug("Gateway memory flush on reset failed: %s", e)

# Snapshot the old entry so on_session_finalize can report the
# expiring session id before reset_session() rotates it.
old_entry = self.session_store._entries.get(session_key)

# Close tool resources on the old agent (terminal sandboxes, browser
# daemons, background processes) before evicting from cache.
# Guard with getattr because test fixtures may skip __init__.
Expand Down Expand Up @@ -7252,16 +7119,6 @@ async def _handle_resume_command(self, event: MessageEvent) -> str:
if current_entry.session_id == target_id:
return f"📌 Already on session **{name}**."

# Flush memories for current session before switching
try:
_flush_task = asyncio.create_task(
self._async_flush_memories(current_entry.session_id, session_key)
)
self._background_tasks.add(_flush_task)
_flush_task.add_done_callback(self._background_tasks.discard)
except Exception as e:
logger.debug("Memory flush on resume failed: %s", e)

# Clear any running agent for this session key
self._release_running_agent_state(session_key)

Expand Down
Loading
Loading