diff --git a/marm-mcp-server/marm_mcp_server/core/stdio_logging.py b/marm-mcp-server/marm_mcp_server/core/stdio_logging.py new file mode 100644 index 00000000..70656767 --- /dev/null +++ b/marm-mcp-server/marm_mcp_server/core/stdio_logging.py @@ -0,0 +1,35 @@ +"""STDIO transport logging setup for MARM MCP Server.""" + +import logging +import os +import pathlib +import sys + + +_log_dir_env = os.environ.get("MARM_STDIO_LOG_DIR") +_log_dir = ( + pathlib.Path(_log_dir_env) + if _log_dir_env + else pathlib.Path.home() / ".marm" / "logs" +) +_log_level_name = os.environ.get("MARM_STDIO_LOG_LEVEL", "INFO").upper() +_log_level = getattr(logging, _log_level_name, logging.INFO) +_debug = _log_level <= logging.DEBUG + +_stdio_log = logging.getLogger("marm.stdio") +_stdio_log.setLevel(_log_level) +_stdio_log.propagate = False + +_fmt = logging.Formatter("%(asctime)s [MARM] %(levelname)s %(message)s") + +_sh = logging.StreamHandler(sys.stderr) +_sh.setFormatter(_fmt) +_stdio_log.addHandler(_sh) + +try: + _log_dir.mkdir(parents=True, exist_ok=True) + _fh = logging.FileHandler(_log_dir / "marm-stdio.log", encoding="utf-8") + _fh.setFormatter(_fmt) + _stdio_log.addHandler(_fh) +except Exception: + pass diff --git a/marm-mcp-server/marm_mcp_server/core/stdio_tool_lifecycle.py b/marm-mcp-server/marm_mcp_server/core/stdio_tool_lifecycle.py new file mode 100644 index 00000000..23a61121 --- /dev/null +++ b/marm-mcp-server/marm_mcp_server/core/stdio_tool_lifecycle.py @@ -0,0 +1,129 @@ +"""Per-call lifecycle wrapper for MARM STDIO tools: logging, session +init, protocol/compaction injection.""" + +import asyncio +import functools +import json + +from .compaction import claim_pending_compaction_prompt +from .memory import memory +from ..services.documentation import ensure_marm_started, maybe_auto_refresh +from ..utils.helpers import read_protocol_file, read_protocol_lite_file +from .stdio_logging import _debug, _stdio_log + + +_protocol_delivered = False +_protocol_call_count = 0 +_STDIO_LITE_INTERVAL = 30 +_protocol_delivery_lock = asyncio.Lock() + + +def _log_tool_call(fn): + @functools.wraps(fn) + async def wrapper(*args, **kwargs): + name = fn.__name__ + if _debug: + safe = [] + for k, v in kwargs.items(): + if k == "session_name": + safe.append(f"session={v}") + elif k == "query": + safe.append(f"query_len={len(v) if v else 0}") + elif k in ("limit", "search_all"): + safe.append(f"{k}={v}") + _stdio_log.debug("CALL %s %s", name, " ".join(safe)) + else: + _stdio_log.info("CALL %s", name) + + global _protocol_delivered, _protocol_call_count + session_name = kwargs.get("session_name", "default") + try: + await ensure_marm_started(session_name) + except Exception as e: + _stdio_log.warning("session init failed: %s", e) + + _protocol_call_count += 1 + call_count = _protocol_call_count + + try: + result = await fn(*args, **kwargs) + except Exception as e: + _stdio_log.error("EXCEPTION %s %s: %s", name, type(e).__name__, e) + raise + if isinstance(result, dict): + status = result.get("status", "ok") + if status == "error": + _stdio_log.error("FAIL %s: %s", name, result.get("message", "")) + elif _debug: + count = next( + ( + result[k] + for k in ("results_count", "total_entries", "total_count") + if k in result + ), + None, + ) + _stdio_log.debug( + "OK %s status=%s%s", + name, + status, + f" count={count}" if count is not None else "", + ) + else: + _stdio_log.info("OK %s", name) + + protocol_injected = False + async with _protocol_delivery_lock: + if not _protocol_delivered: + try: + result["marm_protocol"] = await read_protocol_file() + _protocol_delivered = True + protocol_injected = True + except Exception as e: + _stdio_log.warning("protocol injection failed: %s", e) + elif call_count % _STDIO_LITE_INTERVAL == 0: + try: + lite_content = await read_protocol_lite_file() + if lite_content: + result["marm_protocol_lite"] = lite_content + # Lite does not block compaction — protocol_injected stays False + except Exception as e: + _stdio_log.warning("lite protocol injection failed: %s", e) + + if not protocol_injected: + # fn may have switched or created the active session (e.g. + # create_log_entry_stdio without an explicit session_name + # writes to memory.active_log_session), which makes the + # pre-call session_name snapshot above stale. Only override + # it when the caller didn't explicitly pass one -- explicit + # caller intent is always respected as-is. + compaction_session = session_name + if not kwargs.get("session_name"): + compaction_session = memory.active_log_session + try: + compaction_block = await asyncio.to_thread( + claim_pending_compaction_prompt, memory, compaction_session + ) + if compaction_block: + serialized_result = json.dumps(result, ensure_ascii=False) + result = { + **result, + "content": [ + compaction_block, + { + "type": "text", + "text": serialized_result, + }, + ], + } + except Exception as e: + _stdio_log.warning("compaction injection failed: %s", e) + + try: + await maybe_auto_refresh() + except Exception as e: + _stdio_log.warning("auto-refresh failed: %s", e) + + return result + + return wrapper diff --git a/marm-mcp-server/marm_mcp_server/server_stdio.py b/marm-mcp-server/marm_mcp_server/server_stdio.py index 1fd0c2a2..bedf6e87 100644 --- a/marm-mcp-server/marm_mcp_server/server_stdio.py +++ b/marm-mcp-server/marm_mcp_server/server_stdio.py @@ -12,7 +12,6 @@ import asyncio import builtins -import json import sys _real_print = builtins.print @@ -20,13 +19,7 @@ *args, **{**kwargs, "file": sys.stderr} ) -import functools # noqa: E402 -import logging # noqa: E402 import os # noqa: E402 -import pathlib # noqa: E402 -import re # noqa: E402 -import uuid # noqa: E402 -from datetime import datetime, timezone # noqa: E402 from typing import Literal, Optional # noqa: E402 from anyio import BrokenResourceError, ClosedResourceError, EndOfStream # noqa: E402 @@ -34,151 +27,20 @@ os.environ["SERVER_HOST"] = "127.0.0.1" -_log_dir_env = os.environ.get("MARM_STDIO_LOG_DIR") -_log_dir = ( - pathlib.Path(_log_dir_env) - if _log_dir_env - else pathlib.Path.home() / ".marm" / "logs" -) -_log_level_name = os.environ.get("MARM_STDIO_LOG_LEVEL", "INFO").upper() -_log_level = getattr(logging, _log_level_name, logging.INFO) -_debug = _log_level <= logging.DEBUG - -_stdio_log = logging.getLogger("marm.stdio") -_stdio_log.setLevel(_log_level) -_stdio_log.propagate = False - -_fmt = logging.Formatter("%(asctime)s [MARM] %(levelname)s %(message)s") - -_sh = logging.StreamHandler(sys.stderr) -_sh.setFormatter(_fmt) -_stdio_log.addHandler(_sh) - -try: - _log_dir.mkdir(parents=True, exist_ok=True) - _fh = logging.FileHandler(_log_dir / "marm-stdio.log", encoding="utf-8") - _fh.setFormatter(_fmt) - _stdio_log.addHandler(_fh) -except Exception: - pass - -_protocol_delivered = False -_protocol_call_count = 0 -_STDIO_LITE_INTERVAL = 30 - - -def _log_tool_call(fn): - @functools.wraps(fn) - async def wrapper(*args, **kwargs): - name = fn.__name__ - if _debug: - safe = [] - for k, v in kwargs.items(): - if k == "session_name": - safe.append(f"session={v}") - elif k == "query": - safe.append(f"query_len={len(v) if v else 0}") - elif k in ("limit", "search_all"): - safe.append(f"{k}={v}") - _stdio_log.debug("CALL %s %s", name, " ".join(safe)) - else: - _stdio_log.info("CALL %s", name) - - global _protocol_delivered, _protocol_call_count - session_name = kwargs.get("session_name", "default") - try: - await ensure_marm_started(session_name) - except Exception as e: - _stdio_log.warning("session init failed: %s", e) - - _protocol_call_count += 1 - call_count = _protocol_call_count - - try: - result = await fn(*args, **kwargs) - except Exception as e: - _stdio_log.error("EXCEPTION %s %s: %s", name, type(e).__name__, e) - raise - if isinstance(result, dict): - status = result.get("status", "ok") - if status == "error": - _stdio_log.error("FAIL %s: %s", name, result.get("message", "")) - elif _debug: - count = next( - ( - result[k] - for k in ("results_count", "total_entries", "total_count") - if k in result - ), - None, - ) - _stdio_log.debug( - "OK %s status=%s%s", - name, - status, - f" count={count}" if count is not None else "", - ) - else: - _stdio_log.info("OK %s", name) - - protocol_injected = False - if not _protocol_delivered: - try: - result["marm_protocol"] = await read_protocol_file() - _protocol_delivered = True - protocol_injected = True - except Exception as e: - _stdio_log.warning("protocol injection failed: %s", e) - elif call_count % _STDIO_LITE_INTERVAL == 0: - try: - lite_content = await read_protocol_lite_file() - if lite_content: - result["marm_protocol_lite"] = lite_content - # Lite does not block compaction — protocol_injected stays False - except Exception as e: - _stdio_log.warning("lite protocol injection failed: %s", e) - - if not protocol_injected: - try: - compaction_block = await asyncio.to_thread( - claim_pending_compaction_prompt, memory, session_name - ) - if compaction_block: - serialized_result = json.dumps(result, ensure_ascii=False) - result = { - **result, - "content": [ - compaction_block, - { - "type": "text", - "text": serialized_result, - }, - ], - } - except Exception as e: - _stdio_log.warning("compaction injection failed: %s", e) - - try: - await maybe_auto_refresh() - except Exception as e: - _stdio_log.warning("auto-refresh failed: %s", e) - - return result - - return wrapper +from .core.stdio_logging import _stdio_log # noqa: E402 + +from .core.stdio_tool_lifecycle import _log_tool_call # noqa: E402 from mcp.server.fastmcp import FastMCP # noqa: E402 from marm_mcp_server.core.memory import memory # noqa: E402 -from marm_mcp_server.core.compaction import claim_pending_compaction_prompt # noqa: E402 -from marm_mcp_server.core.events import events # noqa: E402 from marm_mcp_server.services.notebook import notebook_dispatch # noqa: E402 -from marm_mcp_server.services.documentation import ( # noqa: E402 - ensure_marm_started, - maybe_auto_refresh, +from marm_mcp_server.services.stdio_entry_tools import ( # noqa: E402 + create_log_entry_stdio, + delete_entry_stdio, + list_log_entries_stdio, ) -from marm_mcp_server.utils.helpers import read_protocol_file, read_protocol_lite_file # noqa: E402 from marm_mcp_server.services.summary import generate_session_summary # noqa: E402 from marm_mcp_server.services.recall import smart_recall # noqa: E402 from marm_mcp_server.endpoints.concepts import ( # noqa: E402 @@ -193,8 +55,6 @@ async def wrapper(*args, **kwargs): SERVER_VERSION, DEFAULT_DB_PATH, SEMANTIC_SEARCH_AVAILABLE, - MARM_PROJECT, - MARM_PLATFORM, ) from marm_mcp_server.core.graph_supervisor import graph_supervisor # noqa: E402 from marm_graph.core import tool_router as graph_router # noqa: E402 @@ -272,10 +132,6 @@ async def marm_smart_recall( ) -_SESSION_PREFIXES = ("Session: ", "Topic: ") -_SESSION_INACTIVITY_NOTICE_SECONDS = 3600 - - @mcp.tool() @_log_tool_call async def marm_log_entry( @@ -298,181 +154,7 @@ async def marm_log_entry( Returns: status, message confirming the entry or session switch, entry_id, memory_id """ - try: - formatted_entry = entry.strip() - - # Session-switch detection - for prefix in _SESSION_PREFIXES: - if formatted_entry.startswith(prefix): - base_name = formatted_entry[len(prefix) :].strip() - if not base_name: - return { - "status": "error", - "message": "Session name cannot be empty.", - } - date_tag = datetime.now(timezone.utc).strftime("%Y-%m-%d") - new_session = f"{base_name}-{date_tag}" - marker_id = str(uuid.uuid4()) - with memory.get_connection() as conn: - conn.execute("UPDATE sessions SET marm_active = FALSE") - conn.execute( - """ - INSERT INTO sessions (session_name, last_accessed, marm_active) - VALUES (?, ?, TRUE) - ON CONFLICT(session_name) DO UPDATE SET - last_accessed = excluded.last_accessed, - marm_active = TRUE - """, - (new_session, datetime.now(timezone.utc).isoformat()), - ) - conn.execute( - """ - INSERT INTO log_entries - (id, session_name, entry_date, topic, summary, full_entry, project, platform) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - marker_id, - new_session, - date_tag, - "session_start", - base_name, - formatted_entry, - MARM_PROJECT or None, - MARM_PLATFORM or None, - ), - ) - try: - conn.execute( - "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", - (datetime.now(timezone.utc).isoformat(), new_session), - ) - except Exception: - pass - conn.commit() - memory.active_log_session = new_session - await events.emit("session_created", {"session": new_session}) - return { - "status": "session_switched", - "message": f"📂 Session switched to '{new_session}'", - "session_name": new_session, - } - - # Resolve session — explicit > active > dated fallback - if session_name: - session = session_name - elif memory.active_log_session != "main": - session = memory.active_log_session - else: - date_tag = datetime.now(timezone.utc).strftime("%Y-%m-%d") - session = f"session-{date_tag}" - with memory.get_connection() as conn: - conn.execute("UPDATE sessions SET marm_active = FALSE") - conn.execute( - """ - INSERT INTO sessions (session_name, last_accessed, marm_active) - VALUES (?, ?, TRUE) - ON CONFLICT(session_name) DO UPDATE SET - last_accessed = excluded.last_accessed, - marm_active = TRUE - """, - (session, datetime.now(timezone.utc).isoformat()), - ) - conn.commit() - memory.active_log_session = session - - # Chunk boundary check - with memory.get_connection() as conn: - row = conn.execute( - "SELECT last_accessed FROM sessions WHERE session_name = ?", (session,) - ).fetchone() - if row and row[0]: - try: - last_dt = datetime.fromisoformat(row[0]) - if last_dt.tzinfo is None: - last_dt = last_dt.replace(tzinfo=timezone.utc) - gap = (datetime.now(timezone.utc) - last_dt).total_seconds() - if gap > _SESSION_INACTIVITY_NOTICE_SECONDS: - print( - f"[MARM] Chunk boundary detected for '{session}' — {gap:.0f}s since last write" - ) - except Exception: - pass - - entry_pattern = r"^(\d{4}-\d{2}-\d{2})-(.*?)-(.*?)$" - match = re.match(entry_pattern, formatted_entry) - - if match: - entry_date, topic, summary = match.groups() - else: - entry_date = datetime.now(timezone.utc).strftime("%Y-%m-%d") - topic = "general" - summary = formatted_entry - - entry_id = str(uuid.uuid4()) - now_iso = datetime.now(timezone.utc).isoformat() - with memory.get_connection() as conn: - conn.execute( - """ - INSERT INTO log_entries (id, session_name, entry_date, topic, summary, full_entry, project, platform) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - entry_id, - session, - entry_date, - topic, - summary, - formatted_entry, - MARM_PROJECT or None, - MARM_PLATFORM or None, - ), - ) - conn.execute( - """ - INSERT INTO sessions (session_name, last_accessed) - VALUES (?, ?) - ON CONFLICT(session_name) DO UPDATE SET last_accessed = excluded.last_accessed - """, - (session, now_iso), - ) - try: - conn.execute( - "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", - (now_iso, session), - ) - except Exception: - pass - conn.commit() - - # Dual-write into semantic memory so marm_smart_recall can find it; - # a store failure must never fail the log write itself. - memory_id = None - try: - memory_id = await memory.store_memory_queued( - formatted_entry, - session, - metadata={"source": "log_entry", "log_entry_id": entry_id}, - ) - except Exception as store_error: - _stdio_log.warning( - "semantic store failed for log entry %s: %s", entry_id, store_error - ) - - await events.emit( - "log_entry_created", - {"entry_id": entry_id, "session": session, "content": formatted_entry}, - ) - - return { - "status": "success", - "message": f"📝 Log entry added: {formatted_entry}", - "entry_id": entry_id, - "memory_id": memory_id, - "formatted_entry": formatted_entry, - } - except Exception as e: - return {"status": "error", "message": f"Error creating log entry: {e!s}"} + return await create_log_entry_stdio(entry, session_name) @mcp.tool() @@ -493,48 +175,7 @@ async def marm_log_show( Returns (no session_name): status, sessions list with session_name/entry_count, total_sessions Returns (with session_name): status, session_name, entries list with id/entry_date/topic/summary/full_entry, total_entries """ - try: - with memory.get_connection() as conn: - if session_name: - cursor = conn.execute( - """ - SELECT id, entry_date, topic, summary, full_entry - FROM log_entries WHERE session_name = ? - ORDER BY entry_date DESC - """, - (session_name,), - ) - entries = [ - { - "id": r[0], - "entry_date": r[1], - "topic": r[2], - "summary": r[3], - "full_entry": r[4], - } - for r in cursor.fetchall() - ] - return { - "status": "success", - "session_name": session_name, - "entries": entries, - "total_entries": len(entries), - } - else: - cursor = conn.execute( - "SELECT session_name, COUNT(*) FROM log_entries GROUP BY session_name" - ) - sessions = [ - {"session_name": r[0], "entry_count": r[1]} - for r in cursor.fetchall() - ] - return { - "status": "success", - "sessions": sessions, - "total_sessions": len(sessions), - } - except Exception as e: - return {"status": "error", "message": f"Error retrieving log entries: {e!s}"} + return await list_log_entries_stdio(session_name) @mcp.tool() @@ -551,88 +192,7 @@ async def marm_delete( type="log" (no session_name): delete entire session and all its entries type="notebook": delete notebook entry by name """ - try: - with memory.get_connection() as conn: - if type == "log": - memories_deleted = 0 - if session_name: - # Dual-written semantic memories must not outlive their log entries - rows = conn.execute( - "SELECT id FROM log_entries WHERE session_name = ? AND (id = ? OR topic = ?)", - (session_name, target, target), - ).fetchall() - entry_ids = [r[0] for r in rows] - cursor = conn.execute( - "DELETE FROM log_entries WHERE session_name = ? AND (id = ? OR topic = ?)", - (session_name, target, target), - ) - deleted = cursor.rowcount - if entry_ids: - placeholders = ",".join("?" * len(entry_ids)) - memories_deleted = conn.execute( - "DELETE FROM memories WHERE json_extract(metadata, '$.source') = 'log_entry' " - f"AND json_extract(metadata, '$.log_entry_id') IN ({placeholders})", - entry_ids, - ).rowcount - if deleted: - try: - conn.execute( - "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", - (datetime.now(timezone.utc).isoformat(), session_name), - ) - except Exception: - pass - else: - conn.execute( - "DELETE FROM sessions WHERE session_name = ?", (target,) - ) - cursor = conn.execute( - "DELETE FROM log_entries WHERE session_name = ?", (target,) - ) - deleted = cursor.rowcount - try: - conn.execute( - "DELETE FROM session_summary_cache WHERE session_name = ?", - (target,), - ) - except Exception: - pass - memories_deleted = conn.execute( - "DELETE FROM memories WHERE session_name = ? " - "AND json_extract(metadata, '$.source') = 'log_entry'", - (target,), - ).rowcount - if memory.active_log_session == target: - memory.active_log_session = "main" - conn.commit() - return { - "status": "success", - "message": f"🗑️ Deleted {deleted} items", - "deleted_count": deleted, - "memories_deleted": memories_deleted, - } - elif type == "notebook": - cursor = conn.execute( - "DELETE FROM notebook_entries WHERE name = ?", (target,) - ) - deleted = cursor.rowcount - conn.commit() - if deleted > 0: - memory.remove_active_notebook_entry(target) - return { - "status": "success" if deleted > 0 else "not_found", - "message": f"🗑️ Deleted notebook entry '{target}'" - if deleted > 0 - else f"Entry '{target}' not found", - "deleted": deleted > 0, - } - else: - return { - "status": "error", - "message": f"Invalid type '{type}'. Must be 'log' or 'notebook'.", - } - except Exception as e: - return {"status": "error", "message": f"Error deleting: {e!s}"} + return await delete_entry_stdio(type, target, session_name) @mcp.tool() diff --git a/marm-mcp-server/marm_mcp_server/services/stdio_entry_tools.py b/marm-mcp-server/marm_mcp_server/services/stdio_entry_tools.py new file mode 100644 index 00000000..690a9fdf --- /dev/null +++ b/marm-mcp-server/marm_mcp_server/services/stdio_entry_tools.py @@ -0,0 +1,328 @@ +"""STDIO log-entry/notebook data operations with real inline SQL (not +thin service-call wrappers).""" + +import re +import uuid +from datetime import datetime, timezone +from typing import Optional + +from ..config.settings import MARM_PLATFORM, MARM_PROJECT +from ..core.events import events +from ..core.memory import memory +from ..core.stdio_logging import _stdio_log + + +_SESSION_PREFIXES = ("Session: ", "Topic: ") +_SESSION_INACTIVITY_NOTICE_SECONDS = 3600 + + +async def create_log_entry_stdio(entry: str, session_name: Optional[str]) -> dict: + try: + formatted_entry = entry.strip() + + # Session-switch detection + for prefix in _SESSION_PREFIXES: + if formatted_entry.startswith(prefix): + base_name = formatted_entry[len(prefix) :].strip() + if not base_name: + return { + "status": "error", + "message": "Session name cannot be empty.", + } + date_tag = datetime.now(timezone.utc).strftime("%Y-%m-%d") + new_session = f"{base_name}-{date_tag}" + marker_id = str(uuid.uuid4()) + with memory.get_connection() as conn: + conn.execute("UPDATE sessions SET marm_active = FALSE") + conn.execute( + """ + INSERT INTO sessions (session_name, last_accessed, marm_active) + VALUES (?, ?, TRUE) + ON CONFLICT(session_name) DO UPDATE SET + last_accessed = excluded.last_accessed, + marm_active = TRUE + """, + (new_session, datetime.now(timezone.utc).isoformat()), + ) + conn.execute( + """ + INSERT INTO log_entries + (id, session_name, entry_date, topic, summary, full_entry, project, platform) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + marker_id, + new_session, + date_tag, + "session_start", + base_name, + formatted_entry, + MARM_PROJECT or None, + MARM_PLATFORM or None, + ), + ) + try: + conn.execute( + "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", + (datetime.now(timezone.utc).isoformat(), new_session), + ) + except Exception: + pass + conn.commit() + memory.active_log_session = new_session + await events.emit("session_created", {"session": new_session}) + return { + "status": "session_switched", + "message": f"📂 Session switched to '{new_session}'", + "session_name": new_session, + } + + # Resolve session — explicit > active > dated fallback + if session_name: + session = session_name + elif memory.active_log_session != "main": + session = memory.active_log_session + else: + date_tag = datetime.now(timezone.utc).strftime("%Y-%m-%d") + session = f"session-{date_tag}" + with memory.get_connection() as conn: + conn.execute("UPDATE sessions SET marm_active = FALSE") + conn.execute( + """ + INSERT INTO sessions (session_name, last_accessed, marm_active) + VALUES (?, ?, TRUE) + ON CONFLICT(session_name) DO UPDATE SET + last_accessed = excluded.last_accessed, + marm_active = TRUE + """, + (session, datetime.now(timezone.utc).isoformat()), + ) + conn.commit() + memory.active_log_session = session + + # Chunk boundary check + with memory.get_connection() as conn: + row = conn.execute( + "SELECT last_accessed FROM sessions WHERE session_name = ?", (session,) + ).fetchone() + if row and row[0]: + try: + last_dt = datetime.fromisoformat(row[0]) + if last_dt.tzinfo is None: + last_dt = last_dt.replace(tzinfo=timezone.utc) + gap = (datetime.now(timezone.utc) - last_dt).total_seconds() + if gap > _SESSION_INACTIVITY_NOTICE_SECONDS: + _stdio_log.info( + "Chunk boundary detected for '%s' — %.0fs since last write", + session, + gap, + ) + except Exception: + pass + + entry_pattern = r"^(\d{4}-\d{2}-\d{2})-(.*?)-(.*?)$" + match = re.match(entry_pattern, formatted_entry) + + if match: + entry_date, topic, summary = match.groups() + else: + entry_date = datetime.now(timezone.utc).strftime("%Y-%m-%d") + topic = "general" + summary = formatted_entry + + entry_id = str(uuid.uuid4()) + now_iso = datetime.now(timezone.utc).isoformat() + with memory.get_connection() as conn: + conn.execute( + """ + INSERT INTO log_entries (id, session_name, entry_date, topic, summary, full_entry, project, platform) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + entry_id, + session, + entry_date, + topic, + summary, + formatted_entry, + MARM_PROJECT or None, + MARM_PLATFORM or None, + ), + ) + conn.execute( + """ + INSERT INTO sessions (session_name, last_accessed) + VALUES (?, ?) + ON CONFLICT(session_name) DO UPDATE SET last_accessed = excluded.last_accessed + """, + (session, now_iso), + ) + try: + conn.execute( + "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", + (now_iso, session), + ) + except Exception: + pass + conn.commit() + + # Dual-write into semantic memory so marm_smart_recall can find it; + # a store failure must never fail the log write itself. + memory_id = None + try: + memory_id = await memory.store_memory_queued( + formatted_entry, + session, + metadata={"source": "log_entry", "log_entry_id": entry_id}, + ) + except Exception as store_error: + _stdio_log.warning( + "semantic store failed for log entry %s: %s", entry_id, store_error + ) + + await events.emit( + "log_entry_created", + {"entry_id": entry_id, "session": session, "content": formatted_entry}, + ) + + return { + "status": "success", + "message": f"📝 Log entry added: {formatted_entry}", + "entry_id": entry_id, + "memory_id": memory_id, + "formatted_entry": formatted_entry, + } + except Exception as e: + return {"status": "error", "message": f"Error creating log entry: {e!s}"} + + +async def list_log_entries_stdio(session_name: Optional[str]) -> dict: + try: + with memory.get_connection() as conn: + if session_name: + cursor = conn.execute( + """ + SELECT id, entry_date, topic, summary, full_entry + FROM log_entries WHERE session_name = ? + ORDER BY entry_date DESC + """, + (session_name,), + ) + entries = [ + { + "id": r[0], + "entry_date": r[1], + "topic": r[2], + "summary": r[3], + "full_entry": r[4], + } + for r in cursor.fetchall() + ] + return { + "status": "success", + "session_name": session_name, + "entries": entries, + "total_entries": len(entries), + } + else: + cursor = conn.execute( + "SELECT session_name, COUNT(*) FROM log_entries GROUP BY session_name" + ) + sessions = [ + {"session_name": r[0], "entry_count": r[1]} + for r in cursor.fetchall() + ] + return { + "status": "success", + "sessions": sessions, + "total_sessions": len(sessions), + } + except Exception as e: + return {"status": "error", "message": f"Error retrieving log entries: {e!s}"} + + +async def delete_entry_stdio( + type: str, target: str, session_name: Optional[str] +) -> dict: + try: + with memory.get_connection() as conn: + if type == "log": + memories_deleted = 0 + if session_name: + # Dual-written semantic memories must not outlive their log entries + rows = conn.execute( + "SELECT id FROM log_entries WHERE session_name = ? AND (id = ? OR topic = ?)", + (session_name, target, target), + ).fetchall() + entry_ids = [r[0] for r in rows] + cursor = conn.execute( + "DELETE FROM log_entries WHERE session_name = ? AND (id = ? OR topic = ?)", + (session_name, target, target), + ) + deleted = cursor.rowcount + if entry_ids: + placeholders = ",".join("?" * len(entry_ids)) + memories_deleted = conn.execute( + "DELETE FROM memories WHERE json_extract(metadata, '$.source') = 'log_entry' " + f"AND json_extract(metadata, '$.log_entry_id') IN ({placeholders})", + entry_ids, + ).rowcount + if deleted: + try: + conn.execute( + "UPDATE session_summary_cache SET dirty = TRUE, updated_at = ? WHERE session_name = ?", + (datetime.now(timezone.utc).isoformat(), session_name), + ) + except Exception: + pass + else: + conn.execute( + "DELETE FROM sessions WHERE session_name = ?", (target,) + ) + cursor = conn.execute( + "DELETE FROM log_entries WHERE session_name = ?", (target,) + ) + deleted = cursor.rowcount + try: + conn.execute( + "DELETE FROM session_summary_cache WHERE session_name = ?", + (target,), + ) + except Exception: + pass + memories_deleted = conn.execute( + "DELETE FROM memories WHERE session_name = ? " + "AND json_extract(metadata, '$.source') = 'log_entry'", + (target,), + ).rowcount + if memory.active_log_session == target: + memory.active_log_session = "main" + conn.commit() + return { + "status": "success", + "message": f"🗑️ Deleted {deleted} items", + "deleted_count": deleted, + "memories_deleted": memories_deleted, + } + elif type == "notebook": + cursor = conn.execute( + "DELETE FROM notebook_entries WHERE name = ?", (target,) + ) + deleted = cursor.rowcount + conn.commit() + if deleted > 0: + memory.remove_active_notebook_entry(target) + return { + "status": "success" if deleted > 0 else "not_found", + "message": f"🗑️ Deleted notebook entry '{target}'" + if deleted > 0 + else f"Entry '{target}' not found", + "deleted": deleted > 0, + } + else: + return { + "status": "error", + "message": f"Invalid type '{type}'. Must be 'log' or 'notebook'.", + } + except Exception as e: + return {"status": "error", "message": f"Error deleting: {e!s}"} diff --git a/marm-mcp-server/tests/test_docker_transports.py b/marm-mcp-server/tests/test_docker_transports.py index 8fbfedb3..131cc064 100644 --- a/marm-mcp-server/tests/test_docker_transports.py +++ b/marm-mcp-server/tests/test_docker_transports.py @@ -259,9 +259,9 @@ def test_docker_healthcheck_status_becomes_healthy(docker_image, marm_data_dir): break time.sleep(2) - assert ( - status == "healthy" - ), f"container health status never became healthy (last: {status})" + assert status == "healthy", ( + f"container health status never became healthy (last: {status})" + ) finally: _run_docker(["rm", "-f", container], timeout=30) diff --git a/marm-mcp-server/tests/test_protocol_lite.py b/marm-mcp-server/tests/test_protocol_lite.py index 2e5c6c91..b93fc07c 100644 --- a/marm-mcp-server/tests/test_protocol_lite.py +++ b/marm-mcp-server/tests/test_protocol_lite.py @@ -229,6 +229,7 @@ def test_hard_cap_limits_call_counts(monkeypatch, tmp_path): def test_stdio_lite_injected_on_interval(monkeypatch, tmp_path): """STDIO transport injects lite every 30 calls.""" import marm_mcp_server.server_stdio as stdio + import marm_mcp_server.core.stdio_tool_lifecycle as lifecycle import marm_mcp_server.services.notebook as notebook_service from marm_mcp_server.core.memory import MARMMemory @@ -237,8 +238,8 @@ def test_stdio_lite_injected_on_interval(monkeypatch, tmp_path): monkeypatch.setattr(stdio, "memory", mem) monkeypatch.setattr(notebook_service, "memory", mem) - stdio._protocol_delivered = False - stdio._protocol_call_count = 0 + lifecycle._protocol_delivered = False + lifecycle._protocol_call_count = 0 async def dummy_tool(*args, **kwargs): return {"status": "success"} @@ -246,9 +247,11 @@ async def dummy_tool(*args, **kwargs): async def noop(*args, **kwargs): return None - monkeypatch.setattr(stdio, "ensure_marm_started", noop) - monkeypatch.setattr(stdio, "maybe_auto_refresh", noop) - monkeypatch.setattr(stdio, "claim_pending_compaction_prompt", lambda *a, **kw: None) + monkeypatch.setattr(lifecycle, "ensure_marm_started", noop) + monkeypatch.setattr(lifecycle, "maybe_auto_refresh", noop) + monkeypatch.setattr( + lifecycle, "claim_pending_compaction_prompt", lambda *a, **kw: None + ) wrapped = stdio._log_tool_call(dummy_tool) @@ -272,6 +275,7 @@ async def noop(*args, **kwargs): def test_stdio_lite_and_compaction_coexist(monkeypatch, tmp_path): """STDIO: on call 30, lite AND compaction both appear in result.""" import marm_mcp_server.server_stdio as stdio + import marm_mcp_server.core.stdio_tool_lifecycle as lifecycle import marm_mcp_server.services.notebook as notebook_service from marm_mcp_server.core.memory import MARMMemory @@ -280,8 +284,8 @@ def test_stdio_lite_and_compaction_coexist(monkeypatch, tmp_path): monkeypatch.setattr(stdio, "memory", mem) monkeypatch.setattr(notebook_service, "memory", mem) - stdio._protocol_delivered = False - stdio._protocol_call_count = 0 + lifecycle._protocol_delivered = False + lifecycle._protocol_call_count = 0 async def dummy_tool(*args, **kwargs): return {"status": "success"} @@ -289,12 +293,12 @@ async def dummy_tool(*args, **kwargs): async def noop(*args, **kwargs): return None - monkeypatch.setattr(stdio, "ensure_marm_started", noop) - monkeypatch.setattr(stdio, "maybe_auto_refresh", noop) + monkeypatch.setattr(lifecycle, "ensure_marm_started", noop) + monkeypatch.setattr(lifecycle, "maybe_auto_refresh", noop) # Return a known compaction string instead of None monkeypatch.setattr( - stdio, + lifecycle, "claim_pending_compaction_prompt", lambda *a, **kw: {"type": "text", "text": "COMPACTION_NUDGE_CONTENT"}, ) diff --git a/marm-mcp-server/tests/test_stdio_transport.py b/marm-mcp-server/tests/test_stdio_transport.py index 68553fb0..4f8a9929 100644 --- a/marm-mcp-server/tests/test_stdio_transport.py +++ b/marm-mcp-server/tests/test_stdio_transport.py @@ -12,23 +12,47 @@ def _isolated_stdio(monkeypatch, tmp_path): import marm_mcp_server.server_stdio as stdio + import marm_mcp_server.core.stdio_tool_lifecycle as lifecycle import marm_mcp_server.services.notebook as notebook_service + import marm_mcp_server.services.stdio_entry_tools as stdio_entry_tools from marm_mcp_server.core.memory import MARMMemory mem = MARMMemory(str(tmp_path / "stdio-inprocess.db")) mem._encoder_failed = True monkeypatch.setattr(stdio, "memory", mem) monkeypatch.setattr(notebook_service, "memory", mem) + # marm_log_entry's body now lives in services.stdio_entry_tools + # (server-stdio-module-split.md Task 3) with its own `memory` binding -- + # without this, its writes (including the queued semantic-memory store) + # silently hit the real production singleton instead of this test's + # isolated instance, which can also hang waiting on a write queue that + # was never started in-process. + monkeypatch.setattr(stdio_entry_tools, "memory", mem) + # _log_tool_call re-resolves the compaction-claim session from + # memory.active_log_session when the caller omits session_name (PR #88 + # CodeRabbit finding) -- without patching lifecycle's own binding too, + # that read hits the real production singleton's state instead of this + # test's isolated instance. + monkeypatch.setattr(lifecycle, "memory", mem) async def _noop(*args, **kwargs): return None - monkeypatch.setattr(stdio, "ensure_marm_started", _noop) - monkeypatch.setattr(stdio, "maybe_auto_refresh", _noop) + # ensure_marm_started/maybe_auto_refresh/claim_pending_compaction_prompt + # and the protocol-delivery state now live in core.stdio_tool_lifecycle + # (server-stdio-module-split.md Task 2) -- _log_tool_call resolves these + # from its own module's globals, not server_stdio's. + monkeypatch.setattr(lifecycle, "ensure_marm_started", _noop) + monkeypatch.setattr(lifecycle, "maybe_auto_refresh", _noop) monkeypatch.setattr( - stdio, "claim_pending_compaction_prompt", lambda *args, **kwargs: None + lifecycle, "claim_pending_compaction_prompt", lambda *args, **kwargs: None ) - stdio._protocol_delivered = True + lifecycle._protocol_delivered = True + # _protocol_call_count is also module-global and leaks across tests -- + # if it lands on a multiple of _STDIO_LITE_INTERVAL, _log_tool_call + # injects marm_protocol_lite unexpectedly, breaking exact-dict + # assertions in tests that don't expect it. + lifecycle._protocol_call_count = 0 return stdio @@ -152,6 +176,44 @@ def message(msg): assert len(tools) == 14 +def test_stdio_compaction_claimed_against_resolved_session_not_literal_default( + monkeypatch, tmp_path +): + """CodeRabbit finding on PR #88: _log_tool_call snapshots session_name = + kwargs.get("session_name", "default") BEFORE calling fn. create_log_entry_stdio + can resolve/create a totally different session internally (memory.active_log_session) + when the caller omits session_name -- the pre-call snapshot then claims + compaction against the literal string "default", which almost never has + any log entries, silently missing the session that actually received the + write. + """ + import marm_mcp_server.core.stdio_tool_lifecycle as lifecycle + + stdio = _isolated_stdio(monkeypatch, tmp_path) + + claimed_sessions = [] + + def _spy_claim(memory, session_name): + claimed_sessions.append(session_name) + return None + + monkeypatch.setattr(lifecycle, "claim_pending_compaction_prompt", _spy_claim) + + result = asyncio.run(stdio.marm_log_entry(entry="regression test entry")) + assert result["status"] == "success" + + resolved_session = stdio.memory.active_log_session + assert resolved_session != "main", ( + "test setup assumption broken: expected create_log_entry_stdio to " + "resolve a fresh dated session when none was supplied" + ) + assert claimed_sessions == [resolved_session], ( + f"compaction claimed against {claimed_sessions}, expected the " + f"actually-resolved session {resolved_session!r}, not the literal " + f"string 'default'" + ) + + def test_stdio_delete_notebook_removes_entry_from_active_state(monkeypatch, tmp_path): stdio = _isolated_stdio(monkeypatch, tmp_path) @@ -282,17 +344,45 @@ async def run(): "marm_log_entry", {"entry": "Session: envelope-session"}, ) + resolved_session = json.loads(session_result.content[0].text)[ + "session_name" + ] entry_result = await client.call_tool( "marm_log_entry", {"entry": "2026-06-03-envelope-routing verified"}, ) - return add_result, use_result, delete_result, session_result, entry_result - - add_result, use_result, delete_result, session_result, entry_result = asyncio.run( - run() - ) + log_show_result = await client.call_tool( + "marm_log_show", + {"session_name": resolved_session}, + ) + return ( + add_result, + use_result, + delete_result, + session_result, + entry_result, + log_show_result, + resolved_session, + ) - for result in (add_result, use_result, delete_result, session_result, entry_result): + ( + add_result, + use_result, + delete_result, + session_result, + entry_result, + log_show_result, + resolved_session, + ) = asyncio.run(run()) + + for result in ( + add_result, + use_result, + delete_result, + session_result, + entry_result, + log_show_result, + ): assert result.content assert result.content[0].type == "text" @@ -308,6 +398,18 @@ async def run(): f"stdio log entry did not dual-write a semantic memory: {entry_body}" ) + # marm_log_show through the real MCP tool surface -- not exercised + # anywhere else in this file (PR #88 review finding). Two entries are + # expected: the "Session: envelope-session" switch itself writes a + # session_start marker entry, plus the envelope-routing entry below. + log_show_body = json.loads(log_show_result.content[0].text) + assert log_show_body["status"] == "success" + assert log_show_body["session_name"] == resolved_session + assert log_show_body["total_entries"] == 2 + assert "2026-06-03-envelope-routing verified" in [ + e["full_entry"] for e in log_show_body["entries"] + ] + def test_stdio_graph_tool_returns_unavailable_when_backend_down(monkeypatch, tmp_path): stdio = _isolated_stdio(monkeypatch, tmp_path) @@ -329,8 +431,10 @@ def test_stdio_graph_unavailable_response_is_not_shared_mutable_state( real first-call path (_protocol_delivered = False, unlike the other tests in this module) and call twice to prove no state survives between calls. """ + import marm_mcp_server.core.stdio_tool_lifecycle as lifecycle + stdio = _isolated_stdio(monkeypatch, tmp_path) - stdio._protocol_delivered = False + lifecycle._protocol_delivered = False monkeypatch.setattr(stdio.graph_supervisor, "is_available", lambda: False) first = asyncio.run(stdio.marm_graph_index(repo_path="/tmp/some-repo")) @@ -908,6 +1012,7 @@ def message(msg): def test_stdio_protocol_injected_on_first_tool_call_not_on_second(monkeypatch): import marm_mcp_server.server_stdio as stdio + import marm_mcp_server.core.stdio_tool_lifecycle as lifecycle async def _noop(*args, **kwargs): return None @@ -918,11 +1023,11 @@ async def _protocol(): def _claim(memory): return None - monkeypatch.setattr(stdio, "ensure_marm_started", _noop) - monkeypatch.setattr(stdio, "maybe_auto_refresh", _noop) - monkeypatch.setattr(stdio, "read_protocol_file", _protocol) - monkeypatch.setattr(stdio, "claim_pending_compaction_prompt", _claim) - stdio._protocol_delivered = False + monkeypatch.setattr(lifecycle, "ensure_marm_started", _noop) + monkeypatch.setattr(lifecycle, "maybe_auto_refresh", _noop) + monkeypatch.setattr(lifecycle, "read_protocol_file", _protocol) + monkeypatch.setattr(lifecycle, "claim_pending_compaction_prompt", _claim) + lifecycle._protocol_delivered = False @stdio._log_tool_call async def fake_tool(): @@ -937,21 +1042,22 @@ async def fake_tool(): def test_stdio_compaction_injection_wraps_tool_result(monkeypatch, tmp_path): import marm_mcp_server.server_stdio as stdio + import marm_mcp_server.core.stdio_tool_lifecycle as lifecycle async def _noop(*args, **kwargs): return None - monkeypatch.setattr(stdio, "ensure_marm_started", _noop) - monkeypatch.setattr(stdio, "maybe_auto_refresh", _noop) + monkeypatch.setattr(lifecycle, "ensure_marm_started", _noop) + monkeypatch.setattr(lifecycle, "maybe_auto_refresh", _noop) monkeypatch.setattr( - stdio, + lifecycle, "claim_pending_compaction_prompt", lambda memory, session_name: { "type": "text", "text": "[MARM COMPACTION REQUEST]\nabc", }, ) - stdio._protocol_delivered = True + lifecycle._protocol_delivered = True @stdio._log_tool_call async def fake_tool(): @@ -967,6 +1073,7 @@ async def fake_tool(): def test_stdio_protocol_call_suppresses_same_call_compaction(monkeypatch, tmp_path): import marm_mcp_server.server_stdio as stdio + import marm_mcp_server.core.stdio_tool_lifecycle as lifecycle calls = {"claim": 0} @@ -980,11 +1087,11 @@ def _claim(memory): calls["claim"] += 1 return {"type": "text", "text": "[MARM COMPACTION REQUEST]\nabc"} - monkeypatch.setattr(stdio, "ensure_marm_started", _noop) - monkeypatch.setattr(stdio, "maybe_auto_refresh", _noop) - monkeypatch.setattr(stdio, "read_protocol_file", _protocol) - monkeypatch.setattr(stdio, "claim_pending_compaction_prompt", _claim) - stdio._protocol_delivered = False + monkeypatch.setattr(lifecycle, "ensure_marm_started", _noop) + monkeypatch.setattr(lifecycle, "maybe_auto_refresh", _noop) + monkeypatch.setattr(lifecycle, "read_protocol_file", _protocol) + monkeypatch.setattr(lifecycle, "claim_pending_compaction_prompt", _claim) + lifecycle._protocol_delivered = False @stdio._log_tool_call async def fake_tool():