fix: make consolidation API async with background task + increased timeout
Three fixes for consolidation reliability:
1. Fire-and-forget API: POST /memory/consolidate now launches consolidation
as an asyncio background task and returns immediately. The old approach
blocked until Cat's WS response, which could take 5+ minutes (LLM
extraction calls), exceeding both the WS timeout and browser fetch
timeout. Web UI now polls /memory/status to track completion.
2. Increased timeout: cat_client.trigger_consolidation() timeout raised
from 300s to 600s (configurable via parameter). Logs unexpected WS
message types for debugging.
3. Better logging: Consolidation log messages prefixed with 🌙 for
grep-friendliness. cat_client errors include exc_info=True for
traceback visibility. Web UI shows elapsed time while polling.
This commit is contained in:
@@ -577,46 +577,51 @@ class CatAdapter:
|
||||
logger.error(f"Error clearing conversation history: {e}")
|
||||
return False
|
||||
|
||||
async def trigger_consolidation(self) -> Optional[str]:
|
||||
async def trigger_consolidation(self, timeout: int = 600) -> Optional[str]:
|
||||
"""
|
||||
Trigger memory consolidation by sending a special message via WebSocket.
|
||||
The memory_consolidation plugin's tool 'consolidate_memories' is
|
||||
triggered when it sees 'consolidate now' in the text.
|
||||
The memory_consolidation plugin's agent_prompt_prefix hook detects
|
||||
'consolidate now' in the text and runs the consolidation synchronously.
|
||||
Uses WebSocket with a system user ID for proper context.
|
||||
|
||||
Args:
|
||||
timeout: Max seconds to wait for the consolidation response.
|
||||
Default 600 (10 min) as consolidation + LLM call can be slow.
|
||||
"""
|
||||
try:
|
||||
ws_base = self._base_url.replace("http://", "ws://").replace("https://", "wss://")
|
||||
ws_url = f"{ws_base}/ws/system_consolidation"
|
||||
|
||||
logger.info("🌙 Triggering memory consolidation via WS...")
|
||||
logger.info(f"🌙 Triggering memory consolidation via WS (timeout={timeout}s)...")
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.ws_connect(
|
||||
ws_url,
|
||||
timeout=300, # Consolidation can be very slow
|
||||
timeout=timeout,
|
||||
) as ws:
|
||||
await ws.send_json({"text": "consolidate now"})
|
||||
|
||||
# Wait for the final chat response
|
||||
deadline = asyncio.get_event_loop().time() + 300
|
||||
deadline = asyncio.get_event_loop().time() + timeout
|
||||
last_type = ""
|
||||
|
||||
while True:
|
||||
remaining = deadline - asyncio.get_event_loop().time()
|
||||
if remaining <= 0:
|
||||
logger.error("Consolidation timed out (>300s)")
|
||||
logger.error(f"🌙 Consolidation timed out (>{timeout}s)")
|
||||
return "Consolidation timed out"
|
||||
|
||||
try:
|
||||
ws_msg = await asyncio.wait_for(
|
||||
ws.receive(),
|
||||
timeout=remaining
|
||||
timeout=max(1.0, remaining)
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
logger.error("Consolidation WS receive timeout")
|
||||
logger.error("🌙 Consolidation WS receive timeout")
|
||||
return "Consolidation timed out waiting for response"
|
||||
|
||||
if ws_msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSING, aiohttp.WSMsgType.CLOSED):
|
||||
logger.warning("Consolidation WS closed by server")
|
||||
logger.warning("🌙 Consolidation WS closed by server")
|
||||
return "Connection closed during consolidation"
|
||||
if ws_msg.type == aiohttp.WSMsgType.ERROR:
|
||||
return f"WebSocket error: {ws.exception()}"
|
||||
@@ -631,20 +636,24 @@ class CatAdapter:
|
||||
msg_type = msg.get("type", "")
|
||||
if msg_type == "chat":
|
||||
reply = msg.get("content") or msg.get("text", "")
|
||||
logger.info(f"Consolidation result: {reply[:200]}")
|
||||
logger.info(f"🌙 Consolidation result: {reply[:200]}")
|
||||
return reply
|
||||
elif msg_type == "error":
|
||||
error_desc = msg.get("description", "Unknown error")
|
||||
logger.error(f"Consolidation error: {error_desc}")
|
||||
logger.error(f"🌙 Consolidation error: {error_desc}")
|
||||
return f"Consolidation error: {error_desc}"
|
||||
else:
|
||||
# Log unexpected message types for debugging
|
||||
if msg_type != last_type:
|
||||
logger.debug(f"🌙 Consolidation WS msg type: {msg_type}")
|
||||
last_type = msg_type
|
||||
continue
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
logger.error("Consolidation WS connection timed out")
|
||||
logger.error("🌙 Consolidation WS connection timed out")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Consolidation error: {e}")
|
||||
logger.error(f"🌙 Consolidation error: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
# ====================================================================
|
||||
|
||||
Reference in New Issue
Block a user