diff --git a/DESIGN/live-mem/MCP_TOOLS_SPEC.md b/DESIGN/live-mem/MCP_TOOLS_SPEC.md index 0386b08..f36bcfd 100644 --- a/DESIGN/live-mem/MCP_TOOLS_SPEC.md +++ b/DESIGN/live-mem/MCP_TOOLS_SPEC.md @@ -452,15 +452,21 @@ Synchronizes the bank into Graph Memory. Deletes old documents and re-ingests up ```python @mcp.tool() -async def graph_push(space_id: str) -> dict: +async def graph_push(space_id: str, include_volatile: bool = False) -> dict: ``` **Behavior**: - The space must first be connected via `graph_connect` +- Skips volatile bank files by default (`GRAPH_PUSH_VOLATILE_FILES`, default: + `activeContext.md,progress.md`) +- `include_volatile=True` re-includes volatile files for explicit debug / + migration only, requires `manage`/`admin`, and emits an audit log - Intelligent delete + re-ingestion (graph recalculation) - Orphan cleanup (files removed from the bank) - ~10-30s per file (LLM entity/relation extraction + embeddings) - Updates metrics in `_meta.json` +- Response keeps `pushed` as the numeric counter and adds `pushed_files`, + `skipped_volatile`, and a `warning` when files are skipped --- diff --git a/README.md b/README.md index 7333274..992ee7e 100644 --- a/README.md +++ b/README.md @@ -333,7 +333,7 @@ docker compose logs -f live-mem-service --tail 50 # Logs | Tool | Parameters | Description | | ------------------ | ---------------------------------------------------- | --------------------------------------------------------------------------------------------------------- | | `graph_connect` | `space_id`, `url`, `token`, `memory_id`, `ontology?` | Connects a space to Graph Memory. Tests connection, creates memory if needed. Default ontology: `general` | -| `graph_push` | `space_id` | Synchronizes bank → graph. Smart delete + re-ingest, orphan cleanup. ~30s/file | +| `graph_push` | `space_id`, `include_volatile?=false` | Synchronizes non-volatile bank files → graph. Smart delete + re-ingest, orphan cleanup. ~30s/file | | `graph_status` | `space_id` | Connection status + graph stats (documents, entities, relations, top entities, documents list) | | `graph_disconnect` | `space_id` | Disconnects (data remains in graph) | @@ -374,7 +374,7 @@ docker compose logs -f live-mem-service --tail 50 # Logs > > Therefore, **`graph_push` is NOT a routine action**: pushing the full bank into the graph teaches it transient content that a later compaction strands as stale. Routine flows should ingest **canonical repository documents** directly into Graph Memory from the agent / tooling layer, using stable `source_path` keys. `graph_push` remains available for one-off bootstrap and explicit debug / migration only. > -> In particular, `activeContext.md` and `progress.md` **must never** end up in Graph Memory. A future revision (tracked in [`DESIGN/live-mem/EVOLUTION_LIVE_GRAPH_INTEGRATION.md`](DESIGN/live-mem/EVOLUTION_LIVE_GRAPH_INTEGRATION.md)) will turn this into a server-side guardrail. See [`WORKSPACE_CLINE_ADVANCE_RULES.md`](WORKSPACE_CLINE_ADVANCE_RULES.md) for the agent-side template. +> By default, `graph_push` skips volatile bank files listed by `GRAPH_PUSH_VOLATILE_FILES` (`activeContext.md,progress.md`). An operator can pass `include_volatile=true` for explicit debug / migration pushes only; this requires `manage`/`admin` permission and emits an audit log. See [`WORKSPACE_CLINE_ADVANCE_RULES.md`](WORKSPACE_CLINE_ADVANCE_RULES.md) for the agent-side template. Live Memory can push its Memory Bank into a [Graph Memory](https://github.com/Cloud-Temple/graph-memory) instance for long-term memory. The knowledge graph extracts entities, relations, and embeddings from bank files. diff --git a/src/live_mem/config.py b/src/live_mem/config.py index 2547950..e6db7cd 100644 --- a/src/live_mem/config.py +++ b/src/live_mem/config.py @@ -150,6 +150,12 @@ def _normalize_proxy_url(cls, v: str | None) -> str | None: # ─── Response limits ────────────────────────────────────── response_max_bytes: int = 512 * 1024 # Max response body size (512 KB) + # ─── Graph push guardrails ──────────────────────────────── + # CSV of Memory Bank filenames excluded from graph_push by default. + # Operators can override with GRAPH_PUSH_VOLATILE_FILES, or opt in per + # call with include_volatile=True (manage/admin only). + graph_push_volatile_files: str = "activeContext.md,progress.md" + # ─── Admin console /api/tool (ADM-05 fix) ───────────── api_tool_max_body_bytes: int = 1_048_576 # Max request body for /api/tool (1 MB) diff --git a/src/live_mem/core/graph_bridge.py b/src/live_mem/core/graph_bridge.py index 1a3cbca..e68e223 100644 --- a/src/live_mem/core/graph_bridge.py +++ b/src/live_mem/core/graph_bridge.py @@ -33,15 +33,55 @@ import base64 import asyncio import logging +import unicodedata from datetime import datetime, timezone from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client +from ..config import get_settings from .storage import get_storage, bank_relpath from .models import GraphMemoryConfig logger = logging.getLogger("live_mem.graph_bridge") +audit_logger = logging.getLogger("live_mem.audit") + + +def _parse_graph_push_volatile_files(value: str | None = None) -> set[str]: + """Parse the CSV volatile-file policy into normalized basenames.""" + if value is None: + value = get_settings().graph_push_volatile_files + + files: set[str] = set() + for raw in value.split(","): + normalized = _volatile_policy_key(raw) + if normalized: + files.add(normalized) + return files + + +def _volatile_policy_key(filename: str) -> str: + """Normalize a bank relpath for volatile-policy matching.""" + clean = unicodedata.normalize("NFC", str(filename).strip()).replace("\\", "/") + basename = clean.rsplit("/", 1)[-1].strip() + return basename + + +def _split_volatile_bank_files( + bank_files: dict[str, str], + volatile_files: set[str], +) -> tuple[dict[str, str], list[str]]: + """Return pushable bank files and skipped volatile filenames.""" + pushable: dict[str, str] = {} + skipped: list[str] = [] + + for filename, content in bank_files.items(): + if _volatile_policy_key(filename) in volatile_files: + skipped.append(filename) + else: + pushable[filename] = content + + return pushable, skipped # ───────────────────────────────────────────────────────────── @@ -378,7 +418,12 @@ async def connect( # PUSH — Pousser la bank dans graph-memory # ───────────────────────────────────────────────────────── - async def push(self, space_id: str) -> dict: + async def push( + self, + space_id: str, + include_volatile: bool = False, + audit_caller: str | None = None, + ) -> dict: """ Pousse les fichiers bank du space dans graph-memory. @@ -396,6 +441,8 @@ async def push(self, space_id: str) -> dict: Args: space_id: Identifiant du space live-memory + include_volatile: Inclure explicitement les fichiers bank volatiles + audit_caller: Nom de l'appelant à tracer si include_volatile=True Returns: {"status": "ok", "pushed": N, "deleted": N, "errors": N, ...} @@ -429,6 +476,12 @@ async def push(self, space_id: str) -> dict: bank_files = { bank_relpath(item["key"], space_id): item["content"] for item in bank_data } + volatile_files = _parse_graph_push_volatile_files() + push_bank_files, skipped_volatile = ( + (bank_files, []) + if include_volatile + else _split_volatile_bank_files(bank_files, volatile_files) + ) if not bank_files: return { @@ -436,10 +489,30 @@ async def push(self, space_id: str) -> dict: "space_id": space_id, "message": "Aucun fichier bank à pousser", "pushed": 0, + "pushed_files": [], + "skipped_volatile": [], "deleted": 0, "errors": 0, } + if include_volatile: + volatile_in_scope = [ + filename + for filename in bank_files + if _volatile_policy_key(filename) in volatile_files + ] + audit_logger.info( + json.dumps( + { + "event": "graph_push_volatile_optin", + "caller": audit_caller or "unknown", + "space_id": space_id, + "files": volatile_in_scope, + }, + ensure_ascii=False, + ) + ) + # Connexion à graph-memory gm = GraphMemoryClient(config.url, config.token, timeout=180.0) @@ -457,10 +530,12 @@ async def push(self, space_id: str) -> dict: existing_docs.add(doc.get("filename", "")) logger.info( - "Push '%s' → '%s' : %d fichiers bank, %d docs existants", + "Push '%s' → '%s' : %d fichiers bank, %d poussables, %d volatiles sautés, %d docs existants", space_id, memory_id, len(bank_files), + len(push_bank_files), + len(skipped_volatile), len(existing_docs), ) @@ -468,7 +543,7 @@ async def push(self, space_id: str) -> dict: calls = [] call_metadata = [] # Pour tracker quel appel fait quoi - for filename, content in bank_files.items(): + for filename, content in push_bank_files.items(): # Si le document existe → supprimer d'abord if filename in existing_docs: calls.append( @@ -498,6 +573,9 @@ async def push(self, space_id: str) -> dict: call_metadata.append(("ingest", filename)) # 3. Nettoyage des orphelins + # Important: orphan cleanup stays scoped to the full bank, not the + # filtered push set. Otherwise a default push would delete already + # ingested volatile docs solely because this guard skipped re-ingest. orphan_docs = existing_docs - set(bank_files.keys()) for orphan in orphan_docs: calls.append( @@ -512,10 +590,11 @@ async def push(self, space_id: str) -> dict: call_metadata.append(("clean", orphan)) # 4. Exécuter tout le batch dans une seule session - results = await gm.call_tools_batch(calls) + results = await gm.call_tools_batch(calls) if calls else [] # 5. Analyser les résultats pushed = 0 + pushed_files = [] deleted_before_reingest = 0 cleaned = 0 errors = 0 @@ -547,6 +626,7 @@ async def push(self, space_id: str) -> dict: else: if action == "ingest": pushed += 1 + pushed_files.append(filename) logger.info("Ingéré '%s'", filename) elif action == "delete": deleted_before_reingest += 1 @@ -580,12 +660,23 @@ async def push(self, space_id: str) -> dict: "space_id": space_id, "memory_id": memory_id, "pushed": pushed, + "pushed_files": pushed_files, + "skipped_volatile": skipped_volatile, "deleted_before_reingest": deleted_before_reingest, "cleaned_orphans": cleaned, "errors": errors, "duration_seconds": duration, } + if skipped_volatile: + result["warning"] = ( + "Volatile bank files were skipped by default; pass " + "include_volatile=True with manage/admin permission for explicit " + "debug or migration pushes." + ) + if not pushed_files and skipped_volatile and errors == 0: + result["message"] = "No non-volatile bank files to push" + if error_details: result["error_details"] = error_details diff --git a/src/live_mem/tools/graph.py b/src/live_mem/tools/graph.py index 792f85b..7f8e226 100644 --- a/src/live_mem/tools/graph.py +++ b/src/live_mem/tools/graph.py @@ -20,6 +20,8 @@ """ import ipaddress +import json +import logging from typing import Annotated, Optional from urllib.parse import urlparse @@ -34,6 +36,7 @@ # metadata cloud 169.254.169.254, ...). L'URL persiste ensuite dans # `_meta.json` et est ré-utilisée à chaque graph_push. _ALLOWED_GM_SCHEMES = ("http", "https") +_audit_logger = logging.getLogger("live_mem.audit") def _validate_gm_url(url: str) -> Optional[str]: @@ -203,6 +206,16 @@ async def graph_push( space_id: Annotated[ str, Field(description="Identifiant du space live-memory à synchroniser") ], + include_volatile: Annotated[ + bool, + Field( + default=False, + description=( + "Inclure explicitement les fichiers bank volatiles " + "(activeContext.md, progress.md par défaut). Requiert manage/admin." + ), + ), + ] = False, ) -> dict: """ ⚠️ **Advanced / debug tool — NOT for routine flows.** @@ -254,11 +267,11 @@ async def graph_push( stabilised bank, - explicit debug / migration scenarios under operator control. - In particular, `activeContext.md` and `progress.md` **must not** - end up in Graph Memory. A future revision (tracked in - `DESIGN/live-mem/EVOLUTION_LIVE_GRAPH_INTEGRATION.md`) will turn this - into a server-side guardrail. For now, the contract is doctrinal: - do not invoke this tool as part of session-end consolidation. + By default, volatile bank files (`activeContext.md`, `progress.md`, + configurable via `GRAPH_PUSH_VOLATILE_FILES`) are skipped and returned + in `skipped_volatile`. `include_volatile=True` is an explicit + debug / migration opt-in, requires manage/admin permission, and emits + an audit log. See `WORKSPACE_CLINE_ADVANCE_RULES.md`, README "Live ↔ Graph Memory" section, and `DESIGN/live-mem/ARCHITECTURE.md` for the @@ -266,11 +279,18 @@ async def graph_push( Args: space_id: Identifiant du space live-memory + include_volatile: Inclure explicitement les fichiers volatiles + (manage/admin uniquement) Returns: Métriques de push : fichiers poussés, nettoyés, erreurs, durée """ - from ..auth.context import check_access, check_write_permission + from ..auth.context import ( + check_access, + check_manage_permission, + check_write_permission, + current_token_info, + ) from ..core.graph_bridge import get_graph_bridge try: @@ -282,7 +302,31 @@ async def graph_push( if write_err: return write_err - return await get_graph_bridge().push(space_id) + caller = "unknown" + token_info = current_token_info.get() or {} + if token_info.get("client_name"): + caller = token_info["client_name"] + + if include_volatile: + manage_err = check_manage_permission() + if manage_err: + return manage_err + _audit_logger.info( + json.dumps( + { + "event": "graph_push_include_volatile_requested", + "caller": caller, + "space_id": space_id, + }, + ensure_ascii=False, + ) + ) + + return await get_graph_bridge().push( + space_id, + include_volatile=include_volatile, + audit_caller=caller, + ) except Exception as e: from ..auth.context import safe_error diff --git a/tests/test_graph_volatile_guard.py b/tests/test_graph_volatile_guard.py new file mode 100644 index 0000000..f3b2968 --- /dev/null +++ b/tests/test_graph_volatile_guard.py @@ -0,0 +1,276 @@ +# -*- coding: utf-8 -*- +"""Tests for graph_push volatile bank-file guardrails.""" + +from __future__ import annotations + +import base64 +import json +import logging +from contextlib import contextmanager +from unittest.mock import AsyncMock, patch + +import pytest + +from live_mem.auth.context import current_token_info +from live_mem.config import get_settings +from live_mem.core.graph_bridge import GraphBridgeService +from live_mem.tools.graph import register as register_graph_tools + + +class FakeStorage: + def __init__(self, bank_files: dict[str, str]): + self.meta = { + "space_id": "alpha", + "graph_memory": { + "url": "https://graph.example.com/mcp", + "token": "gm-token", + "memory_id": "mem-alpha", + "ontology": "general", + "push_count": 0, + "files_pushed": 0, + }, + } + self.bank_files = bank_files + self.saved_meta = None + + async def get_json(self, key: str) -> dict: + assert key == "alpha/_meta.json" + return self.meta + + async def list_and_get(self, prefix: str) -> list[dict]: + assert prefix == "alpha/bank/" + return [ + { + "key": f"alpha/bank/{filename}", + "content": content, + "size": len(content), + "last_modified": "", + } + for filename, content in self.bank_files.items() + ] + + async def put_json(self, key: str, data: dict) -> None: + assert key == "alpha/_meta.json" + self.saved_meta = data + + +class FakeGraphMemoryClient: + existing_documents: list[str] = [] + batch_calls: list[tuple[str, dict]] = [] + + def __init__(self, url: str, token: str, timeout: float = 120.0): + self.url = url + self.token = token + self.timeout = timeout + + async def call_tool(self, tool_name: str, arguments: dict) -> dict: + assert tool_name == "document_list" + return { + "status": "ok", + "documents": [ + {"filename": filename} for filename in self.existing_documents + ], + } + + async def call_tools_batch(self, calls: list[tuple[str, dict]]) -> list[dict]: + self.__class__.batch_calls.extend(calls) + return [{"status": "ok"} for _ in calls] + + +@contextmanager +def _patched_graph_dependencies(storage: FakeStorage): + FakeGraphMemoryClient.batch_calls = [] + FakeGraphMemoryClient.existing_documents = [] + with patch("live_mem.core.graph_bridge.get_storage", return_value=storage), patch( + "live_mem.core.graph_bridge.GraphMemoryClient", + new=FakeGraphMemoryClient, + ): + yield + + +def _ingested_filenames() -> list[str]: + return [ + args["filename"] + for tool, args in FakeGraphMemoryClient.batch_calls + if tool == "memory_ingest" + ] + + +def _decoded_ingests() -> dict[str, str]: + decoded = {} + for tool, args in FakeGraphMemoryClient.batch_calls: + if tool != "memory_ingest": + continue + decoded[args["filename"]] = base64.b64decode( + args["content_base64"] + ).decode("utf-8") + return decoded + + +@pytest.fixture(autouse=True) +def clear_settings_cache(): + get_settings.cache_clear() + yield + get_settings.cache_clear() + + +@pytest.mark.asyncio +async def test_push_skips_volatile_by_default(): + storage = FakeStorage( + { + "activeContext.md": "# active", + "progress.md": "# progress", + "productContext.md": "# product", + } + ) + + with _patched_graph_dependencies(storage): + result = await GraphBridgeService().push("alpha") + + assert result["status"] == "ok" + assert result["pushed"] == 1 + assert result["pushed_files"] == ["productContext.md"] + assert sorted(result["skipped_volatile"]) == [ + "activeContext.md", + "progress.md", + ] + assert _ingested_filenames() == ["productContext.md"] + assert _decoded_ingests() == {"productContext.md": "# product"} + assert "warning" in result + + +@pytest.mark.asyncio +async def test_push_with_include_volatile_pushes_everything(caplog): + storage = FakeStorage( + { + "activeContext.md": "# active", + "progress.md": "# progress", + "productContext.md": "# product", + } + ) + + with caplog.at_level(logging.INFO, logger="live_mem.audit"): + with _patched_graph_dependencies(storage): + result = await GraphBridgeService().push( + "alpha", + include_volatile=True, + audit_caller="manager", + ) + + assert result["status"] == "ok" + assert result["pushed"] == 3 + assert result["skipped_volatile"] == [] + assert _ingested_filenames() == [ + "activeContext.md", + "progress.md", + "productContext.md", + ] + + audit_payloads = [ + json.loads(r.message) + for r in caplog.records + if r.name == "live_mem.audit" and "graph_push_volatile_optin" in r.message + ] + assert audit_payloads + assert audit_payloads[-1]["event"] == "graph_push_volatile_optin" + assert audit_payloads[-1]["caller"] == "manager" + assert audit_payloads[-1]["space_id"] == "alpha" + assert sorted(audit_payloads[-1]["files"]) == [ + "activeContext.md", + "progress.md", + ] + + +@pytest.mark.asyncio +async def test_only_volatile_files_returns_empty_pushed_not_error(): + storage = FakeStorage( + { + "activeContext.md": "# active", + "progress.md": "# progress", + } + ) + + with _patched_graph_dependencies(storage): + result = await GraphBridgeService().push("alpha") + + assert result["status"] == "ok" + assert result["pushed"] == 0 + assert result["pushed_files"] == [] + assert sorted(result["skipped_volatile"]) == [ + "activeContext.md", + "progress.md", + ] + assert result["message"] == "No non-volatile bank files to push" + assert FakeGraphMemoryClient.batch_calls == [] + + +@pytest.mark.asyncio +async def test_volatile_filter_is_configurable(monkeypatch): + monkeypatch.setenv("GRAPH_PUSH_VOLATILE_FILES", "productContext.md") + get_settings.cache_clear() + storage = FakeStorage( + { + "activeContext.md": "# active", + "productContext.md": "# product", + "techContext.md": "# tech", + } + ) + + with _patched_graph_dependencies(storage): + result = await GraphBridgeService().push("alpha") + + assert result["status"] == "ok" + assert result["skipped_volatile"] == ["productContext.md"] + assert _ingested_filenames() == ["activeContext.md", "techContext.md"] + + +@pytest.mark.asyncio +async def test_nested_volatile_basename_is_skipped_by_default(): + storage = FakeStorage( + { + "1.MEMORY_BANK/activeContext.md": "# active", + "stable/runbook.md": "# runbook", + } + ) + + with _patched_graph_dependencies(storage): + result = await GraphBridgeService().push("alpha") + + assert result["skipped_volatile"] == ["1.MEMORY_BANK/activeContext.md"] + assert _ingested_filenames() == ["stable/runbook.md"] + + +@pytest.mark.asyncio +async def test_graph_push_include_volatile_requires_manage_permission(): + class FakeMCP: + def __init__(self): + self.tools = {} + + def tool(self, annotations=None): + def decorator(func): + self.tools[func.__name__] = func + return func + + return decorator + + mcp = FakeMCP() + register_graph_tools(mcp) + push = mcp.tools["graph_push"] + bridge = AsyncMock() + + tok = current_token_info.set( + { + "client_name": "writer", + "permissions": ["read", "write"], + "allowed_resources": ["alpha"], + } + ) + try: + with patch("live_mem.core.graph_bridge.get_graph_bridge", return_value=bridge): + result = await push("alpha", include_volatile=True) + finally: + current_token_info.reset(tok) + + assert result["status"] == "error" + assert "manage" in result["message"] + bridge.push.assert_not_called()