memory.server

memory.server

Memory MCP server — persistent research tree, skills library, and live code graph.

Storage: entries are markdown files with YAML frontmatter in .memory/. The KGLite .memory.kgl file is a disposable cache rebuilt from these files.

Three disconnected subgraphs

1. Research tree: typed entries in parent-child branches
2. Skills library: flat procedural/pattern knowledge store
3. Code graph (separate in-memory KG): auto-generated codebase structure (rebuilt on git changes)

Tool groups

Research tree: record, discard, move, link, recall, search, codebase, history (digest/reconstruct), suggest_relationships, set_relationship, resolve_pin, drift (create_project is only registered while the repo is un-bootstrapped) Skills: save_skill, get_skill, search_skills (empty query = list), delete_skill Documents: documents (register/list/deregister) Sessions: handoff, pickup, search_sessions Maintenance/visualization: health, launch_dashboard

Federated view: another repo's tree can be browsed alongside this one as a read-only overlay in the dashboard. It is configured via federated_repos in .memory/config.yaml and surfaced only through launch_dashboard (the recall/search tools stay local-repo only); see that tool's docstring for the step-by-step setup.

Artifact/experiment tools (DVC large files, S3, pipelines, experiment reruns) live in the separate memory-artifacts server (memory/artifacts_server.py) so that heavy, rarely-used block can be toggled independently in Cursor without consuming this server's tool slots.

sync, vacuum, attach_plan, and the individual skill/document CRUD helpers remain importable for tests and CLI use but are not registered as MCP tools (Cursor caps total MCP tools across servers, so the surface is kept deliberately lean).

startup

startup() -> None

One-time process startup: reap dead sidecars, publish a live session.

Called by the MCP entry point so freshly started servers appear as live session cards immediately instead of after the first tool call.

Source code in memory/server.py

def startup() -> None:
    """One-time process startup: reap dead sidecars, publish a live session.

    Called by the MCP entry point so freshly started servers appear as live
    session cards immediately instead of after the first tool call.
    """
    try:
        _reap_dead_sidecars()
    except Exception as e:
        logger.warning("Dead-sidecar reaping failed: %s", e)
    try:
        _ensure_session()
    except Exception as e:
        logger.warning("Could not initialize session at startup: %s", e)

create_project

create_project(name: str, description: str) -> str

Create the root research project for this repo. Only one project is allowed.

Parameters:

Name	Type	Description	Default
name	str	Unique project name (typically the repo name)	required
description	str	What this project is about	required

Source code in memory/server.py

def create_project(name: str, description: str) -> str:
    """Create the root research project for this repo. Only one project is allowed.

    Args:
        name: Unique project name (typically the repo name)
        description: What this project is about
    """
    _ensure_session()
    existing = list(graph.cypher("MATCH (p:project) RETURN p.id AS id"))
    if existing:
        return _err(f"Project already exists: {existing[0]['id']}. Only one root project is allowed.", "Conflict")

    now = _now()
    project_data = {
        "id": name,
        "name": name,
        "description": description,
        "status": "active",
        "created_at": now,
    }

    storage.write_project(project_data)
    storage.write_config({"format_version": storage.FORMAT_VERSION})

    with _cache_lock:
        df = pd.DataFrame([project_data])
        graph.add_nodes(df, node_type="project", unique_id_field="id", node_title_field="name")
    _save()
    _schedule_memory_commit(str(storage.MEMORY_DIR / "project.md"))
    _schedule_memory_commit(str(storage.MEMORY_DIR / "config.yaml"))
    return json.dumps({"project": name, "status": "active", "created_at": now})

overview

overview() -> str

Use this first when you're unsure how to work with memory: a compact orientation map.

Read-only. Returns the memory server's purpose, the handful of workflows you reach for most often, and a tool -> action index so you can pick the right tool without reading every docstring. Authoritative action values also live on each tool's own action schema enum.

Source code in memory/server.py

@mcp_server.tool()
def overview() -> str:
    """Use this first when you're unsure how to work with memory: a compact orientation map.

    Read-only. Returns the memory server's purpose, the handful of workflows you
    reach for most often, and a tool -> action index so you can pick the right
    tool without reading every docstring. Authoritative action values also live
    on each tool's own ``action`` schema enum.
    """
    return (
        "# Memory server — orientation\n\n"
        "Persistent research knowledge for THIS repo, stored as markdown in `.memory/`\n"
        "and surfaced as a tree: project > subproject > phase > entries.\n\n"
        "## Start here\n"
        "- New session: `session(action=\"pickup\")`, then `recall` to orient.\n"
        "- Before exploring code or repeating work: `search` / `recall` first.\n"
        "- Before every git commit: `record` what happened (decision/checkpoint/experiment/note).\n\n"
        "## Tools\n"
        "- `record` — save an entry (decision, checkpoint, experiment, note, plan, todo, annotation).\n"
        "- `recall` — read the tree (view=tree|outline|flat|brief; entry_id for a subtree).\n"
        "- `search` — find entries (mode=hybrid|keyword|semantic).\n"
        "- `codebase(action=lookup|blast|callers|callees|imports|context|health)` — "
        "find symbols, see what breaks (blast), and why code exists (context: memory tip+chain).\n"
        "- `relate(action=link|set|suggest|merge|bulk|export)` — entry relationships + hygiene.\n"
        "- `session(action=handoff|pickup|search)` — session lifecycle.\n"
        "- `skill(action=save|get|search|delete)` — reusable procedures.\n"
        "- `documents(action=register|list|deregister)` — markdown reports/plans linked to the tree.\n"
        "- `history(action=digest|reconstruct)` — brownfield onboarding from git history.\n"
        "- `discard`, `move` — retire or reparent entries.\n"
        "- `drift`, `health`, `resolve_pin`, `repair_projects` — maintenance.\n"
        "- `launch_dashboard` — open the visual tree dashboard.\n\n"
        "## Resources you can pull\n"
        "- `memory://overview` (this map) · `memory://entry-types` "
        "(valid `record` types).\n"
    )

record

record(project: str, title: str, body: str, tags: str = '', type: str = 'note', node_label: str = '', parent_id: str = '', success: bool | None = None, reason: str = '', files: str = '', invalidates: str = '', auto_invalidate: bool = False) -> str

Use this when you need to save a decision, checkpoint, experiment, note, or plan to the research tree so it survives across sessions.

Parameters:

Name	Type	Description	Default
project	str	Project name	required
title	str	Full descriptive title	required
body	str	Free-form markdown content	required
tags	str	Comma-separated keywords	''
type	str	Entry type (plan/checkpoint/experiment/decision/note/annotation)	'note'
node_label	str	Short label for visualization (auto-derived if empty)	''
parent_id	str	Parent entry id (defaults to project root if empty)	''
success	bool | None	Whether this succeeded (experiments only, null for others)	None
reason	str	Rationale for the has_child edge	''
files	str	Comma-separated file paths this entry references (auto-committed and git-pinned)	''
invalidates	str	Entry id this supersedes (creates invalidated_by edge)	''
auto_invalidate	bool	If true for decisions, semantically detect and apply invalidation edges	False

Source code in memory/server.py

@mcp_server.tool()
def record(
    project: str,
    title: str,
    body: str,
    tags: str = "",
    type: str = "note",
    node_label: str = "",
    parent_id: str = "",
    success: bool | None = None,
    reason: str = "",
    files: str = "",
    invalidates: str = "",
    auto_invalidate: bool = False,
) -> str:
    """Use this when you need to save a decision, checkpoint, experiment, note, or plan to the research tree so it survives across sessions.

    Args:
        project: Project name
        title: Full descriptive title
        body: Free-form markdown content
        tags: Comma-separated keywords
        type: Entry type (plan/checkpoint/experiment/decision/note/annotation)
        node_label: Short label for visualization (auto-derived if empty)
        parent_id: Parent entry id (defaults to project root if empty)
        success: Whether this succeeded (experiments only, null for others)
        reason: Rationale for the has_child edge
        files: Comma-separated file paths this entry references (auto-committed and git-pinned)
        invalidates: Entry id this supersedes (creates invalidated_by edge)
        auto_invalidate: If true for decisions, semantically detect and apply invalidation edges
    """
    _ensure_session()

    # Reject unknown projects: recording under an unrecognized project name
    # silently produces an orphan subtree that the (single-project) dashboard
    # never shows. The design allows exactly one root project per store, so a
    # mismatch is almost always a typo. Intentional new roots go via
    # create_project. Skip the check pre-bootstrap (no project node yet).
    known_projects = _known_project_ids()
    if known_projects and project not in known_projects:
        return json.dumps({
            "error": (
                f"Unknown project '{project}'. This store's project is "
                f"'{known_projects[0]}'. Entries recorded under an unrecognized "
                f"project are invisible in the dashboard tree. Use the existing "
                f"project name, or call create_project to start a new root."
            ),
            "type": "UnknownProjectError",
            "known_projects": known_projects,
        })

    if not parent_id:
        parent_id = project

    # Validate parent_id up front, before any file pinning or session side
    # effects, so a rejected call leaves no trace. _resolve_parent_type refuses a
    # parent that is neither an existing entry nor a known project root —
    # otherwise the edge code below would vivify a phantom project node from it
    # (the accident behind multi-project stores). It also hands us the edge
    # source type, reused below so we don't re-query the graph.
    parent_type, parent_err = _resolve_parent_type(parent_id, known_projects)
    if parent_err:
        return json.dumps(parent_err)

    if _should_split_for_topic(project, parent_id, type or "note"):
        logger.info(
            "Topic shift detected — closing session %s before recording under %s",
            _session_id,
            parent_id,
        )
        _auto_close_session("topic shift")

    # Semantic drift detection: if this entry drifts from current session topic,
    # auto-close and start a fresh session before recording.
    if _check_drift(title, body):
        logger.info("Semantic drift detected — closing session %s", _session_id)
        _auto_close_session("semantic drift")

    entry_id = _gen_id(type[:4] if type else "note")

    if not node_label:
        node_label = title[:50] if len(title) > 50 else title

    pinned_files = ""
    pin_status: dict = {}
    pin_warnings: list[str] = []
    if files:
        pin_result = _pin_files(files, title)
        if pin_result.error:
            return json.dumps({
                "error": pin_result.error,
                "type": "DirtyWorkingTreeError",
                "git_status": pin_result.git_status,
            })
        pinned_files = pin_result.files
        pin_status = pin_result.git_status
        pin_warnings = pin_result.warnings

    now = _now()
    entry_data: dict = {
        "id": entry_id,
        "type": type or "note",
        "project": project,
        "parent_id": parent_id,
        "title": title,
        "node_label": node_label,
        "tags": tags,
        "status": "active",
        "open_threads": 0,
        "body": body,
        "success": json.dumps(success),
        "files": pinned_files,
        "session_id": _session_id,
        "created_at": now,
        "updated_at": now,
    }
    if reason:
        entry_data["has_child_rationale"] = reason
    if invalidates:
        entry_data["invalidates"] = invalidates

    # parent_type was resolved up front by _resolve_parent_type (guaranteed
    # "entry" or a known project root — never a phantom-minting dangling id).
    with _cache_lock:
        storage.write_entry(entry_data)

        df = pd.DataFrame([entry_data])
        graph.add_nodes(df, node_type="entry", unique_id_field="id", node_title_field="title")

        # Add has_child edge from parent to this entry
        edge_data = {"src": parent_id, "tgt": entry_id, "created_at": now}
        if reason:
            edge_data["rationale"] = reason
        edge_df = pd.DataFrame([edge_data])
        graph.add_connections(
            edge_df,
            connection_type="has_child",
            source_type=parent_type,
            source_id_field="src",
            target_type="entry",
            target_id_field="tgt",
        )

        # Handle invalidation edge
        if invalidates:
            inv_df = pd.DataFrame([{
                "src": invalidates, "tgt": entry_id,
                "rationale": reason or "superseded", "created_at": now,
            }])
            graph.add_connections(
                inv_df,
                connection_type="invalidated_by",
                source_type="entry",
                source_id_field="src",
                target_type="entry",
                target_id_field="tgt",
            )

        # Update parent's open_threads count
        threads = _count_active_children(parent_id)
        if parent_type == "entry":
            _upsert_node("entry", parent_id, {"open_threads": threads, "updated_at": now})
            storage.update_entry_field(parent_id, {"open_threads": threads, "updated_at": now})
        else:
            _upsert_node("project", parent_id, {"open_threads": threads, "name": parent_id})

        # Incremental embedding
        search_text = _build_search_text(title, tags, body)
        _upsert_node("entry", entry_id, {"search_text": search_text})
        try:
            _embed_graph_texts("entry", "search_text")
        except Exception:
            pass

        # Auto-detect invalidation for new decisions
        auto_invalidated: list[dict] = []
        if type == "decision" and auto_invalidate:
            auto_invalidated = _detect_invalidated_decisions(
                project, entry_id, search_text
            )
            for old in auto_invalidated:
                inv_df = pd.DataFrame([{
                    "src": old["id"], "tgt": entry_id,
                    "rationale": "auto-detected: semantic similarity",
                    "created_at": now,
                }])
                graph.add_connections(
                    inv_df,
                    connection_type="invalidated_by",
                    source_type="entry",
                    source_id_field="src",
                    target_type="entry",
                    target_id_field="tgt",
                )
            if auto_invalidated:
                inv_ids = ",".join(old["id"] for old in auto_invalidated)
                storage.update_entry_field(entry_id, {"invalidated_by": inv_ids})

    _save()
    _schedule_memory_commit(str(storage.ENTRIES_DIR / f"{entry_id}.md"))

    _track(entry_id, "created")
    topic_id = entry_id if parent_id == project and (type or "note").lower() == "plan" else parent_id
    if topic_id != project:
        with _session_lock:
            if topic_id not in _session_topic_ids:
                _session_topic_ids.append(topic_id)

    # Track this entry's embedding for drift detection in future record() calls
    entry_text = f"{title}\n{body[:500]}"
    entry_vec = _embed_single(entry_text)
    if entry_vec is not None:
        _session_embeddings.append(entry_vec)

    result: dict = {
        "entry_id": entry_id,
        "type": type,
        "title": title,
        "parent_id": parent_id,
        "files": pinned_files,
    }
    if pin_status:
        result["git_status"] = pin_status
    if pin_warnings:
        result["warnings"] = pin_warnings
    if auto_invalidated:
        result["auto_invalidated"] = auto_invalidated

    return json.dumps(result)

discard

discard(entry_id: str, reason: str = '') -> str

Mark an entry as discarded (soft delete).

Parameters:

Name	Type	Description	Default
entry_id	str	The entry to discard	required
reason	str	Why it's being discarded	''

Source code in memory/server.py

@mcp_server.tool()
def discard(entry_id: str, reason: str = "") -> str:
    """Mark an entry as discarded (soft delete).

    Args:
        entry_id: The entry to discard
        reason: Why it's being discarded
    """
    rows = list(graph.cypher(
        "MATCH (e:entry) WHERE e.id = $eid RETURN e.parent_id AS pid, e.project AS proj",
        params={"eid": entry_id}
    ))
    if not rows:
        return _err(f"Entry '{entry_id}' not found", "NotFound")

    parent_id = rows[0]["pid"]
    now = _now()

    storage.update_entry_field(entry_id, {"status": "discarded", "updated_at": now})

    with _cache_lock:
        _upsert_node("entry", entry_id, {"status": "discarded", "updated_at": now})

        threads = _count_active_children(parent_id)
        parent_check = list(graph.cypher(
            "MATCH (e:entry) WHERE e.id = $pid RETURN e.id AS id",
            params={"pid": parent_id}
        ))
        if parent_check:
            _upsert_node("entry", parent_id, {"open_threads": threads, "updated_at": now})
            storage.update_entry_field(parent_id, {"open_threads": threads, "updated_at": now})

    _save()
    _schedule_memory_commit(str(storage.ENTRIES_DIR / f"{entry_id}.md"))
    _track(entry_id, "modified")
    return json.dumps({
        "discarded": entry_id,
        "reason": reason,
    })

move

move(entry_id: str, new_parent_id: str) -> str

Move an entry to a different parent in the tree.

Parameters:

Name	Type	Description	Default
entry_id	str	The entry to move	required
new_parent_id	str	New parent entry id (or project id for top-level)	required

Source code in memory/server.py

@mcp_server.tool()
def move(entry_id: str, new_parent_id: str) -> str:
    """Move an entry to a different parent in the tree.

    Args:
        entry_id: The entry to move
        new_parent_id: New parent entry id (or project id for top-level)
    """
    rows = list(graph.cypher(
        "MATCH (e:entry) WHERE e.id = $eid RETURN e.parent_id AS old_parent, e.project AS proj",
        params={"eid": entry_id}
    ))
    if not rows:
        return _err(f"Entry '{entry_id}' not found", "NotFound")

    old_parent = rows[0]["old_parent"]
    now = _now()

    # Reject a dangling new parent for the same reason record() does: a
    # new_parent_id that is neither an existing entry nor a known project root
    # would orphan the entry, and the source_type="project" edge created below
    # can vivify a phantom root node from it. Reparenting to the project root
    # (by passing its id) stays allowed. _resolve_parent_type also gives us the
    # edge source type, reused below.
    new_parent_type, parent_err = _resolve_parent_type(new_parent_id, _known_project_ids())
    if parent_err:
        return json.dumps(parent_err)

    storage.update_entry_field(entry_id, {"parent_id": new_parent_id, "updated_at": now})

    with _cache_lock:
        _upsert_node("entry", entry_id, {"parent_id": new_parent_id, "updated_at": now})

        old_threads = _count_active_children(old_parent)
        old_parent_check = list(graph.cypher(
            "MATCH (e:entry) WHERE e.id = $pid RETURN e.id AS id",
            params={"pid": old_parent}
        ))
        if old_parent_check:
            _upsert_node("entry", old_parent, {"open_threads": old_threads, "updated_at": now})
            storage.update_entry_field(old_parent, {"open_threads": old_threads, "updated_at": now})

        new_threads = _count_active_children(new_parent_id)
        if new_parent_type == "entry":
            _upsert_node("entry", new_parent_id, {"open_threads": new_threads, "updated_at": now})
            storage.update_entry_field(new_parent_id, {"open_threads": new_threads, "updated_at": now})

        edge_df = pd.DataFrame([{"src": new_parent_id, "tgt": entry_id, "created_at": now}])
        graph.add_connections(
            edge_df,
            connection_type="has_child",
            source_type=new_parent_type,
            source_id_field="src",
            target_type="entry",
            target_id_field="tgt",
        )

        # Remove the stale old-parent -> entry has_child edge. The entry now
        # records new_parent_id as its parent, so this edge is no longer owned
        # by any node; leaving it would make the live graph disagree with a
        # from-scratch rebuild (which sees only new_parent -> entry). has_child
        # is owned unambiguously by the child, so a direct delete is safe.
        if old_parent and old_parent != new_parent_id:
            _delete_edge("has_child", old_parent, entry_id)

    _save()
    _schedule_memory_commit(str(storage.ENTRIES_DIR / f"{entry_id}.md"))
    return json.dumps({
        "moved": entry_id,
        "from": old_parent,
        "to": new_parent_id,
    })

link

link(entry_a: str, entry_b: str, rationale: str) -> str

Create a bidirectional cross-reference between two entries.

Parameters:

Name	Type	Description	Default
entry_a	str	First entry id	required
entry_b	str	Second entry id	required
rationale	str	Why these are connected	required

Source code in memory/server.py

def link(entry_a: str, entry_b: str, rationale: str) -> str:
    """Create a bidirectional cross-reference between two entries.

    Args:
        entry_a: First entry id
        entry_b: Second entry id
        rationale: Why these are connected
    """
    now = _now()

    with _cache_lock:
        # Persist edge to files: append to related_to lists
        a_data = storage.read_file(storage.ENTRIES_DIR / f"{entry_a}.md") if (storage.ENTRIES_DIR / f"{entry_a}.md").exists() else {}
        existing_a = str(a_data.get("related_to", "") or "")
        new_a = f"{existing_a},{entry_b}".strip(",") if existing_a else entry_b
        storage.update_entry_field(entry_a, {"related_to": new_a, "updated_at": now})

        b_data = storage.read_file(storage.ENTRIES_DIR / f"{entry_b}.md") if (storage.ENTRIES_DIR / f"{entry_b}.md").exists() else {}
        existing_b = str(b_data.get("related_to", "") or "")
        new_b = f"{existing_b},{entry_a}".strip(",") if existing_b else entry_a
        storage.update_entry_field(entry_b, {"related_to": new_b, "updated_at": now})

        # Add BOTH directed related edges to match a rebuild: link() writes
        # related_to bidirectionally to both files, so a from-scratch rebuild
        # (via _iter_entry_edges over each entry's related_to) materializes
        # a -> b and b -> a. The two rows have distinct (src, tgt) so they are
        # not parallel duplicates, and any later sync dedups an identical edge
        # across calls, so this stays rebuild-equal.
        df = pd.DataFrame([
            {"src": entry_a, "tgt": entry_b, "rationale": rationale, "created_at": now},
            {"src": entry_b, "tgt": entry_a, "rationale": rationale, "created_at": now},
        ])
        graph.add_connections(
            df,
            connection_type="related",
            source_type="entry",
            source_id_field="src",
            target_type="entry",
            target_id_field="tgt",
        )
    _save()
    _schedule_memory_commit(str(storage.ENTRIES_DIR / f"{entry_a}.md"))
    _schedule_memory_commit(str(storage.ENTRIES_DIR / f"{entry_b}.md"))
    _track(entry_a, "modified")
    _track(entry_b, "modified")
    return json.dumps({"linked": [entry_a, entry_b], "rationale": rationale})

suggest_relationships

suggest_relationships(project: str, entry_id: str, relation: str = 'related', limit: int = 5) -> str

Suggest semantically similar entries before choosing a relationship.

This tool only previews candidates. Use set_relationship to apply a related or invalidation edge after deciding what the similarity means.

Parameters:

Name	Type	Description	Default
project	str	Project name	required
entry_id	str	Entry to compare against	required
relation	str	Intended relation context ("related" or "invalidates")	'related'
limit	int	Max candidates to return	5

Source code in memory/server.py

def suggest_relationships(
    project: str,
    entry_id: str,
    relation: str = "related",
    limit: int = 5,
) -> str:
    """Suggest semantically similar entries before choosing a relationship.

    This tool only previews candidates. Use ``set_relationship`` to apply a
    related or invalidation edge after deciding what the similarity means.

    Args:
        project: Project name
        entry_id: Entry to compare against
        relation: Intended relation context ("related" or "invalidates")
        limit: Max candidates to return
    """
    _ensure_session()
    rows = list(graph.cypher(
        "MATCH (e:entry) WHERE e.id = $eid RETURN e.title AS title, e.tags AS tags, "
        "e.body AS body, e.search_text AS search_text",
        params={"eid": entry_id},
    ))
    if not rows:
        return _err(f"Entry '{entry_id}' not found", "NotFound")

    row = rows[0]
    search_text = row.get("search_text") or _build_search_text(
        row.get("title", ""),
        row.get("tags", ""),
        row.get("body", ""),
    )
    try:
        hits = graph.select("entry").search_text("search_text", search_text, top_k=max(limit + 5, 10))
    except Exception as exc:
        return _err(f"Relationship suggestion failed: {exc}", "SearchError")

    # Batch-fetch all hit rows in one query (avoids an N+1 Cypher round-trip per
    # hit). The IN-query may return rows in arbitrary order, so we index them by
    # id and re-assemble candidates in the original ranked hit order below.
    hit_ids = [
        hid for hid in (hit.get("id", "") for hit in hits)
        if hid and hid != entry_id
    ]
    fetched: dict = {}
    if hit_ids:
        for candidate in graph.cypher(
            "MATCH (e:entry) WHERE e.id IN $ids AND e.project = $proj AND e.status = $st "
            "RETURN e.id AS id, e.title AS title, e.type AS type",
            params={"ids": hit_ids, "proj": project, "st": "active"},
        ):
            fetched[candidate["id"]] = candidate

    candidates = []
    for hit in hits:
        hit_id = hit.get("id", "")
        if not hit_id or hit_id == entry_id:
            continue
        candidate = fetched.get(hit_id)
        if candidate is None:
            continue
        candidates.append({
            "id": candidate["id"],
            "title": candidate["title"],
            "type": candidate["type"],
            "score": round(hit.get("score", 0.0), 4),
        })
        if len(candidates) >= limit:
            break

    return json.dumps({
        "entry_id": entry_id,
        "relation": relation,
        "candidates": candidates,
        "note": "Similarity means candidates are connected somehow; choose related unless one entry truly supersedes another.",
    })

set_relationship

set_relationship(entry_a: str, entry_b: str, relationship: str = 'related', rationale: str = '', action: str = 'add') -> str

Add or remove an explicit relationship between two entries.

Use relationship='related' for connected notes/decisions that should both remain valid. Use relationship='invalidated_by' only when entry_b supersedes entry_a.

Parameters:

Name	Type	Description	Default
entry_a	str	Source/older entry id	required
entry_b	str	Target/newer entry id	required
relationship	str	"related" or "invalidated_by"	'related'
rationale	str	Why these entries are connected	''
action	str	"add" or "remove"	'add'

Source code in memory/server.py

def set_relationship(
    entry_a: str,
    entry_b: str,
    relationship: str = "related",
    rationale: str = "",
    action: str = "add",
) -> str:
    """Add or remove an explicit relationship between two entries.

    Use ``relationship='related'`` for connected notes/decisions that should
    both remain valid. Use ``relationship='invalidated_by'`` only when
    ``entry_b`` supersedes ``entry_a``.

    Args:
        entry_a: Source/older entry id
        entry_b: Target/newer entry id
        relationship: "related" or "invalidated_by"
        rationale: Why these entries are connected
        action: "add" or "remove"
    """
    _ensure_session()
    if entry_a == entry_b:
        return _err("Cannot relate an entry to itself", "Conflict")
    if relationship not in {"related", "invalidated_by"}:
        return _err("relationship must be 'related' or 'invalidated_by'", "BadRequest")
    if action not in {"add", "remove"}:
        return _err("action must be 'add' or 'remove'", "BadRequest")

    try:
        a_data = _entry_file_data(entry_a)
        b_data = _entry_file_data(entry_b)
    except ValueError as exc:
        return _err(str(exc), "BadRequest")

    now = _now()
    changed_entries: set[str] = set()

    if relationship == "related":
        a_related = _csv_values(a_data.get("related_to"))
        b_related = _csv_values(b_data.get("related_to"))
        if action == "add":
            if entry_b not in a_related:
                a_related.append(entry_b)
            if entry_a not in b_related:
                b_related.append(entry_a)
        else:
            a_related = [item for item in a_related if item != entry_b]
            b_related = [item for item in b_related if item != entry_a]
        _set_csv_value(entry_a, "related_to", a_related, now)
        _set_csv_value(entry_b, "related_to", b_related, now)
        changed_entries.update({entry_a, entry_b})
    else:
        # Persist invalidation on the older/source entry so file rebuilds create
        # edge entry_a -[:invalidated_by]-> entry_b.
        invalidated_by = _csv_values(a_data.get("invalidated_by"))
        if action == "add":
            if entry_b not in invalidated_by:
                invalidated_by.append(entry_b)
        else:
            invalidated_by = [item for item in invalidated_by if item != entry_b]
            b_invalidated_by = _csv_values(b_data.get("invalidated_by"))
            b_invalidated_by = [item for item in b_invalidated_by if item != entry_a]
            b_invalidates = _csv_values(b_data.get("invalidates"))
            b_invalidates = [item for item in b_invalidates if item != entry_a]
            _set_csv_value(entry_b, "invalidated_by", b_invalidated_by, now)
            _set_csv_value(entry_b, "invalidates", b_invalidates, now)
            changed_entries.add(entry_b)
        _set_csv_value(entry_a, "invalidated_by", invalidated_by, now)
        changed_entries.add(entry_a)

    # Apply an incremental edge add/remove for just the two touched entries
    # (mirroring link()'s incremental path) instead of rebuilding the whole
    # graph. Fall back to a full rebuild only if the delta declines.
    changed_keys = [
        storage._index_key(storage.ENTRIES_DIR / f"{eid}.md") for eid in sorted(changed_entries)
    ]
    if not _apply_delta(changed_keys):
        _rebuild_runtime_cache()
    for eid in changed_entries:
        _schedule_memory_commit(str(storage.ENTRIES_DIR / f"{eid}.md"))
        _track(eid, "modified")

    return json.dumps({
        "relationship": relationship,
        "action": action,
        "entry_a": entry_a,
        "entry_b": entry_b,
        "rationale": rationale,
        "updated": sorted(changed_entries),
    })

recall

recall(project: str = '', entry_id: str = '', type: str = '', view: str = 'tree', depth: int = 0, include_body: bool = False, include_discarded: bool = False, limit: int = 0, offset: int = 0, prose: bool = False) -> str

Use this when you need to read the current research tree — nested hierarchy, structural outline, or flat list — to orient before working.

Designed to stay legible on large trees: the server nests children under their parents (no client-side reconstruction from parent_id), summarizes deep or collapsed branches as descendant counts, and always returns a summary block plus open_phases for fast orientation.

Scope is THIS repo only. To view another repo's tree alongside this one, configure federated_repos in .memory/config.yaml and use the dashboard (launch_dashboard) — federation is a read-only dashboard overlay and is not surfaced through this tool.

Parameters:

Name	Type	Description	Default
project	str	Project name. Optional — omit it and the sole root project is auto-detected (this repo has exactly one).	''
entry_id	str	If provided, root the tree/outline at this entry (its subtree)	''
type	str	Filter to entries of this type. Forces a flat list (nesting is dropped).	''
view	str	"tree" (default, nested with depth control), "outline" (plan-node scaffold only, leaf entries collapsed into counts, each node prefixed with a hierarchical section number like 1, 1.2, 3.2.1 — best first call on a big tree), "flat" (paginated flat list, like a classic dump), or "brief" (a deterministic rollup summary of entry_id's subtree — child titles/types, concluding checkpoint, descendant type histogram, most-central entries, most-recent activity, and a completeness score; cached off-file in the MCP process — a read never dirties the working tree — and auto-invalidated when a descendant or the node changes).	'tree'
depth	int	For view="tree", max generations below the root to expand (0 = unlimited). Branches deeper than depth are collapsed and summarized via descendant_count / descendant_types.	0
include_body	bool	If true, include full body text (default false to keep payloads small).	False
include_discarded	bool	If true, include discarded/superseded entries (default active only).	False
limit	int	For view="flat" only — max entries per page (0 = all).	0
offset	int	For view="flat" only — entries to skip for pagination.	0
prose	bool	For view="brief" only — if true AND a narrative generator is injected (_brief_summarize_fn), add a short prose narrative derived from the deterministic rollup. Default false (deterministic only, no LLM). Plan nodes across every view also carry an additive completeness score in [0, 1].	False

Source code in memory/server.py

@mcp_server.tool()
def recall(
    project: str = "",
    entry_id: str = "",
    type: str = "",
    view: str = "tree",
    depth: int = 0,
    include_body: bool = False,
    include_discarded: bool = False,
    limit: int = 0,
    offset: int = 0,
    prose: bool = False,
) -> str:
    """Use this when you need to read the current research tree — nested hierarchy, structural outline, or flat list — to orient before working.

    Designed to stay legible on large trees: the server nests children under their
    parents (no client-side reconstruction from parent_id), summarizes deep or
    collapsed branches as descendant counts, and always returns a `summary` block
    plus `open_phases` for fast orientation.

    Scope is THIS repo only. To view another repo's tree alongside this one,
    configure `federated_repos` in `.memory/config.yaml` and use the dashboard
    (`launch_dashboard`) — federation is a read-only dashboard overlay and is
    not surfaced through this tool.

    Args:
        project: Project name. Optional — omit it and the sole root project is
            auto-detected (this repo has exactly one).
        entry_id: If provided, root the tree/outline at this entry (its subtree)
        type: Filter to entries of this type. Forces a flat list (nesting is dropped).
        view: "tree" (default, nested with depth control), "outline" (plan-node
            scaffold only, leaf entries collapsed into counts, each node prefixed
            with a hierarchical section number like 1, 1.2, 3.2.1 — best first call
            on a big tree), "flat" (paginated flat list, like a classic dump), or
            "brief" (a deterministic rollup summary of `entry_id`'s subtree —
            child titles/types, concluding checkpoint, descendant type histogram,
            most-central entries, most-recent activity, and a completeness score;
            cached off-file in the MCP process — a read never dirties the working
            tree — and auto-invalidated when a descendant or the node changes).
        depth: For view="tree", max generations below the root to expand (0 = unlimited).
            Branches deeper than `depth` are collapsed and summarized via
            `descendant_count` / `descendant_types`.
        include_body: If true, include full body text (default false to keep payloads small).
        include_discarded: If true, include discarded/superseded entries (default active only).
        limit: For view="flat" only — max entries per page (0 = all).
        offset: For view="flat" only — entries to skip for pagination.
        prose: For view="brief" only — if true AND a narrative generator is
            injected (`_brief_summarize_fn`), add a short prose narrative derived
            from the deterministic rollup. Default false (deterministic only, no
            LLM). Plan nodes across every view also carry an additive
            `completeness` score in [0, 1].
    """
    _ensure_session()
    project = _resolve_project(project)

    # Active code-drift surfacing for the returned nodes (see search()); memoized
    # by pinned ref within this call. Off by default → fields simply not added.
    _drift_on = _drift_surfacing_active()
    _drift_gate_on = _drift_gate_enabled()
    _drift_hydrate = _drift_hydrate_enabled()
    _drift_cache: dict[str, bool | None] = {}
    _drift_sym_cache: dict[str, tuple[str, str] | None] = {}
    # Symbol refinement needs entry bodies; fetch them once for the whole project
    # (recall already loads every row) so both tree and flat views can refine.
    _drift_bodies: dict[str, str] = {}

    base_fields = (
        "e.id AS id, e.type AS type, e.title AS title, e.node_label AS node_label, "
        "e.tags AS tags, e.status AS status, e.open_threads AS open_threads, "
        "e.parent_id AS parent_id, e.success AS success, e.files AS files, "
        "e.invalidated_by AS invalidated_by, "
        "e.created_at AS created_at, e.updated_at AS updated_at"
    )
    if include_body:
        base_fields += ", e.body AS body"
    all_rows = list(graph.cypher(
        f"MATCH (e:entry) WHERE e.project = $proj RETURN {base_fields}",
        params={"proj": project},
    ))

    # Bodies for symbol-granularity drift refinement. Reuse the rows if they
    # already carry bodies (include_body); otherwise fetch them once for the
    # project so both tree and flat views can refine without per-node queries.
    if _drift_on and _symbol_drift_enabled():
        if include_body:
            _drift_bodies = {r["id"]: (r.get("body") or "") for r in all_rows}
        else:
            _drift_bodies = _bodies_for_ids([r["id"] for r in all_rows])

    proj_rows = list(graph.cypher(
        "MATCH (p:project) WHERE p.id = $proj RETURN p.status AS status, p.description AS description",
        params={"proj": project},
    ))
    project_status = proj_rows[0]["status"] if proj_rows else "unknown"

    # --- Summary over the entire project (all statuses) ---
    by_type: dict[str, int] = {}
    by_status: dict[str, int] = {}
    for r in all_rows:
        by_type[r["type"]] = by_type.get(r["type"], 0) + 1
        by_status[r["status"]] = by_status.get(r["status"], 0) + 1
    summary = {
        "total": len(all_rows),
        "active": by_status.get("active", 0),
        "by_type": dict(sorted(by_type.items(), key=lambda kv: kv[1], reverse=True)),
        "by_status": by_status,
    }

    # Working set: active-only unless discarded explicitly requested.
    rows = all_rows if include_discarded else [r for r in all_rows if r["status"] == "active"]

    # --- Adjacency (children grouped by parent, ordered chronologically) ---
    children_of: dict[str, list] = {}
    for r in rows:
        children_of.setdefault(r["parent_id"], []).append(r)
    for kids in children_of.values():
        kids.sort(key=lambda r: r.get("created_at") or "")

    # All-status adjacency (active AND discarded) — the completeness score's
    # resolved-thread ratio needs discarded children the active working set omits.
    children_all_of: dict[str, list] = {}
    for r in all_rows:
        children_all_of.setdefault(r["parent_id"], []).append(r)

    # Memoized descendant aggregates (count + per-type breakdown) for every node.
    _agg_cache: dict[str, dict] = {}

    def aggregate(node_id: str) -> dict:
        cached = _agg_cache.get(node_id)
        if cached is not None:
            return cached
        total = 0
        types: dict[str, int] = {}
        for c in children_of.get(node_id, []):
            total += 1
            types[c["type"]] = types.get(c["type"], 0) + 1
            sub = aggregate(c["id"])
            total += sub["count"]
            for t, n in sub["types"].items():
                types[t] = types.get(t, 0) + n
        res = {"count": total, "types": types}
        _agg_cache[node_id] = res
        return res

    def node_dict(r: dict) -> dict:
        d = {
            "id": r["id"],
            "type": r["type"],
            "title": r["title"],
            "node_label": r.get("node_label"),
            "status": r["status"],
        }
        if r.get("tags"):
            d["tags"] = r["tags"]
        if r.get("open_threads"):
            d["open_threads"] = r["open_threads"]
        if r.get("success") is not None:
            d["success"] = r["success"]
        if r.get("files"):
            d["files"] = r["files"]
        d["created_at"] = r.get("created_at")
        if include_body and "body" in r:
            d["body"] = r.get("body")
        agg = aggregate(r["id"])
        d["child_count"] = len(children_of.get(r["id"], []))
        d["descendant_count"] = agg["count"]
        if agg["count"]:
            d["descendant_types"] = dict(sorted(agg["types"].items(), key=lambda kv: kv[1], reverse=True))
        # Additive per-plan-node completeness score (rides along; nothing removed
        # or reordered). Only plan nodes are "phases/subprojects" to score.
        # Container plan nodes roll up (mean of descendant leaf phases); leaf
        # phases use the direct formula.
        if r["type"] == "plan":
            d["completeness"] = _rollup_completeness(r["id"], children_all_of)
        if _drift_on:
            _annotate_entry_drift(
                d, r.get("files") or "", _drift_cache, hydrate=_drift_hydrate,
                body=_drift_bodies.get(r["id"], ""), sym_cache=_drift_sym_cache,
            )
            # Tree/outline keep structure intact (removing a node would orphan its
            # children), so a gated stale node is clearly flagged in place rather
            # than dropped; the flat view withholds it from the list instead.
            if _drift_gate_on and d.get("stale"):
                d["quarantined"] = True
        return d

    root_id = entry_id or project

    # --- Open-phase orientation: leaf plan nodes (work containers) lacking a checkpoint ---
    open_phases = []
    for r in rows:
        if r["type"] != "plan":
            continue
        kids = children_of.get(r["id"], [])
        if not kids or any(k["type"] == "plan" for k in kids):
            continue  # empty, or a container subproject rather than a work phase
        if not any(k["type"] == "checkpoint" for k in kids):
            open_phases.append({
                "id": r["id"],
                "title": r["title"],
                "open_threads": r.get("open_threads") or 0,
                # Leaf phases by construction (guarded above), but route through
                # the rollup for a single, uniform completeness entry point.
                "completeness": _rollup_completeness(r["id"], children_all_of),
            })

    # --- Skill suggestions ---
    skill_context_parts = [project]
    if proj_rows and proj_rows[0].get("description"):
        skill_context_parts.append(proj_rows[0]["description"])
    if entry_id:
        root_row = next((r for r in all_rows if r["id"] == entry_id), None)
        if root_row and root_row.get("title"):
            skill_context_parts.append(root_row["title"])
    suggested = _suggest_skills(" ".join(skill_context_parts))

    out: dict = {
        "project": project,
        "project_status": project_status,
        "view": view,
        "summary": summary,
        "open_phases": open_phases,
    }
    if entry_id:
        out["root"] = entry_id

    # --- Brief view: deterministic rollup summary of a node's subtree ---
    if view == "brief":
        if not entry_id:
            out["error"] = (
                "view='brief' requires entry_id — the plan/subproject node whose "
                "subtree to summarize."
            )
            return json.dumps(out, default=str)
        by_id = {r["id"]: r for r in all_rows}
        node = by_id.get(entry_id)
        if node is None:
            out["error"] = f"entry '{entry_id}' not found in project '{project}'."
            return json.dumps(out, default=str)

        # Active-subtree descendant ids (memoized over children_of).
        _desc_cache: dict[str, list] = {}

        def descendant_ids(node_id: str) -> list:
            cached = _desc_cache.get(node_id)
            if cached is not None:
                return cached
            ids: list = []
            for c in children_of.get(node_id, []):
                ids.append(c["id"])
                ids.extend(descendant_ids(c["id"]))
            _desc_cache[node_id] = ids
            return ids

        desc_ids = descendant_ids(entry_id)
        updated_map = {
            r["id"]: (r.get("updated_at") or r.get("created_at") or "")
            for r in all_rows
        }
        # WS1 importance channel over the whole tree, keyed by bare id. Falls back
        # to {} (→ recency ordering in the brief) if the records are unavailable.
        try:
            importance_map = _retrieval.build_importance_map(
                _entry_records_for_rerank()
            )
        except Exception:
            importance_map = {}

        agg = aggregate(entry_id)
        completeness = _rollup_completeness(entry_id, children_all_of)
        brief = _deterministic_brief(
            node,
            children_of.get(entry_id, []),
            desc_ids,
            agg["types"],
            updated_map,
            importance_map,
            by_id,
            completeness,
        )

        # Cache: reuse the stored brief when its descendant signature matches, so
        # summarizing an unchanged subtree is recompute-free and auto-invalidates
        # when a descendant is added/removed or touched.
        cached = _read_cached_brief(project, entry_id)
        fresh = cached is not None and cached.get("brief_sig") == brief["brief_sig"]
        if fresh:
            brief = cached

        # Optional prose layer: opt-in and only when a generator is injected.
        # Derived solely from the deterministic rollup so it cannot invent
        # structure. Regenerated on a cache hit only if it is still missing.
        if prose and _brief_summarize_fn is not None and not brief.get("prose"):
            try:
                brief["prose"] = _brief_summarize_fn(brief)
            except Exception:
                brief["prose"] = None
            fresh = False  # prose added → persist the updated brief

        if not fresh:
            _write_cached_brief(project, entry_id, brief)

        # Deterministic-only callers must never receive LLM prose. One cache slot
        # per node holds whatever prose a prior prose=True call generated, so on
        # the prose=False path return a prose-stripped COPY (leave the cache — and
        # its prose, still valid for a later prose=True hit — untouched).
        if not prose and brief.get("prose") is not None:
            brief = {**brief, "prose": None}

        out["brief"] = brief
        out["cached"] = fresh
        return json.dumps(out, default=str)

    # --- Flat view (also used whenever a type filter is given) ---
    if view == "flat" or type:
        if entry_id:
            visible: list = []
            stack = [entry_id]
            seen: set[str] = set()
            while stack:
                nid = stack.pop()
                for c in children_of.get(nid, []):
                    if c["id"] in seen:
                        continue
                    seen.add(c["id"])
                    visible.append(c)
                    stack.append(c["id"])
        else:
            visible = list(rows)
        if type:
            visible = [r for r in visible if r["type"] == type]
        visible.sort(key=lambda r: r.get("created_at") or "")
        total_count = len(visible)
        page = visible[offset:offset + limit] if limit and limit > 0 else visible[offset:]
        out["view"] = "flat"
        if _drift_on:
            for r in page:
                _annotate_entry_drift(
                    r, r.get("files") or "", _drift_cache, hydrate=_drift_hydrate,
                    body=_drift_bodies.get(r["id"], ""), sym_cache=_drift_sym_cache,
                )
            _demote_stale(page)
            if _drift_gate_on:
                quarantined = [r for r in page if r.get("stale")]
                page = [r for r in page if not r.get("stale")]
                if quarantined:
                    out["quarantined_stale"] = quarantined
        out["entries"] = page
        out["count"] = len(page)
        out["total_count"] = total_count
        out["offset"] = offset
        out["limit"] = limit
        if suggested:
            out["suggested_skills"] = suggested
        return json.dumps(out, default=str)

    # --- Outline view: structural plan scaffold only, leaves collapsed to counts ---
    if view == "outline":
        def render_outline(node_id: str, prefix: str = "") -> list:
            result = []
            idx = 0
            for c in children_of.get(node_id, []):
                if c["type"] != "plan":
                    continue
                idx += 1
                number = f"{prefix}{idx}" if prefix else str(idx)
                nd = node_dict(c)
                nd["number"] = number
                nd["title"] = f"{number}. {nd['title']}"
                child_plans = render_outline(c["id"], prefix=f"{number}.")
                if child_plans:
                    nd["children"] = child_plans
                result.append(nd)
            return result

        out["tree"] = render_outline(root_id)
        if suggested:
            out["suggested_skills"] = suggested
        return json.dumps(out, default=str)

    # --- Tree view (default): nested hierarchy with depth control ---
    node_count = 0

    def render_children(node_id: str, level: int) -> list:
        nonlocal node_count
        result = []
        for c in children_of.get(node_id, []):
            nd = node_dict(c)
            node_count += 1
            if children_of.get(c["id"]):
                if depth and level + 1 >= depth:
                    nd["truncated"] = True
                else:
                    nd["children"] = render_children(c["id"], level + 1)
            result.append(nd)
        return result

    out["depth"] = depth
    out["tree"] = render_children(root_id, 0)
    out["count"] = node_count
    if suggested:
        out["suggested_skills"] = suggested
    return json.dumps(out, default=str)

search

search(query: str, project: str = '', mode: str = 'hybrid') -> str

Use this when you're looking for prior entries by keyword, semantic similarity, or both — search here before exploring the codebase.

Scope is THIS repo only. Federated (other-repo) trees are not searched — browse them via the dashboard repo dropdown (see launch_dashboard).

In "hybrid" mode an optional blended reranker can layer graph-importance (degree centrality), session-recency, and MMR diversification on top of the keyword+semantic recall. It is OFF by default (output byte-identical to plain hybrid RRF) and opt-in via env vars: MEMORY_RERANK_W_IMPORTANCE / MEMORY_RERANK_W_RECENCY (channel weights, 0 = off), MEMORY_RERANK_MMR_LAMBDA (1.0 = no MMR), MEMORY_RERANK_OVERFETCH (recall multiplier), and MEMORY_RECENCY_HALF_LIFE_DAYS. When on, each text match carries a rerank breakdown (fused/importance/recency).

Parameters:

Name	Type	Description	Default
query	str	Search query (file path, keyword, or natural language question)	required
project	str	Optional project filter	''
mode	str	"hybrid" (default) combines keyword + semantic, "keyword" for exact substring, "semantic" for vector only	'hybrid'

Source code in memory/server.py

@mcp_server.tool()
def search(query: str, project: str = "", mode: str = "hybrid") -> str:
    """Use this when you're looking for prior entries by keyword, semantic similarity, or both — search here before exploring the codebase.

    Scope is THIS repo only. Federated (other-repo) trees are not searched —
    browse them via the dashboard repo dropdown (see `launch_dashboard`).

    In "hybrid" mode an optional blended reranker can layer graph-importance
    (degree centrality), session-recency, and MMR diversification on top of the
    keyword+semantic recall. It is OFF by default (output byte-identical to plain
    hybrid RRF) and opt-in via env vars: ``MEMORY_RERANK_W_IMPORTANCE`` /
    ``MEMORY_RERANK_W_RECENCY`` (channel weights, 0 = off),
    ``MEMORY_RERANK_MMR_LAMBDA`` (1.0 = no MMR), ``MEMORY_RERANK_OVERFETCH``
    (recall multiplier), and ``MEMORY_RECENCY_HALF_LIFE_DAYS``. When on, each
    text match carries a ``rerank`` breakdown (fused/importance/recency).

    Args:
        query: Search query (file path, keyword, or natural language question)
        project: Optional project filter
        mode: "hybrid" (default) combines keyword + semantic, "keyword" for exact substring, "semantic" for vector only
    """
    _ensure_session()
    file_results = []
    keyword_results = []
    semantic_results = []

    # Blended recall-then-rerank policy (importance + session-recency + MMR on top
    # of the keyword+semantic recall). OFF by default (non-punitive identity: the
    # output is byte-identical to plain hybrid RRF); opt-in via MEMORY_RERANK_* env
    # vars. Only the hybrid path is affected. When ON, over-fetch the recall so the
    # reranker has candidates to reorder before truncating to the final 20.
    rerank_cfg = _retrieval.memory_rerank_config()
    rerank_on = mode == "hybrid" and not _retrieval.rerank_disabled(rerank_cfg)
    recall_k = rerank_cfg.overfetch * 20 if rerank_on else 20

    # --- File path search (always runs) ---
    file_query = "MATCH (e:entry) WHERE e.files IS NOT NULL"
    params: dict = {}
    if project:
        file_query += " AND e.project = $proj"
        params["proj"] = project
    file_query += " RETURN e.id AS id, e.type AS type, e.title AS title, e.files AS files, e.status AS status, e.parent_id AS parent_id, e.created_at AS created_at"

    for row in graph.cypher(file_query, params=params):
        if row["files"] and query.replace("\\", "/") in row["files"].replace("\\", "/"):
            file_results.append(row)
            if len(file_results) >= 20:
                break

    # --- Keyword search (over the cached corpus; no per-query table scan) ---
    if mode in ("hybrid", "keyword"):
        query_lower = query.lower()
        scored_keyword: list[tuple] = []
        for pos, e in enumerate(_entry_keyword_corpus()):
            if project and e["project"] != project:
                continue
            title_match = query_lower in e["_tl"]
            tag_match = query_lower in e["_gl"]
            body_match = query_lower in e["_bl"]
            if title_match or tag_match or body_match:
                # Rank tier: title (0) beats tag (1) beats body (2); ``pos`` keeps
                # it stable within a tier. Feeds a meaningful order into RRF.
                tier = 0 if title_match else (1 if tag_match else 2)
                scored_keyword.append((tier, pos, {
                    "id": e["id"], "type": e["type"], "title": e["title"],
                    "tags": e["tags"], "status": e["status"], "created_at": e["created_at"],
                }))
        scored_keyword.sort(key=lambda t: (t[0], t[1]))
        keyword_results = [r for _, _, r in scored_keyword[:recall_k]]

    # --- Semantic search ---
    if mode in ("hybrid", "semantic"):
        try:
            hits = graph.select("entry").search_text("search_text", query, top_k=recall_k)
            for hit in hits:
                entry = {
                    "id": hit.get("id", ""),
                    "type": hit.get("type", ""),
                    "title": hit.get("title", ""),
                    "status": hit.get("status", ""),
                    "score": round(hit.get("score", 0.0), 4),
                    "created_at": hit.get("created_at", ""),
                }
                if project and hit.get("project") != project:
                    continue
                if hit.get("status") != "active":
                    continue
                semantic_results.append(entry)
        except Exception as e:
            semantic_results = [{"error": str(e)}]

    # --- Merge for hybrid mode (Reciprocal Rank Fusion) ---
    if mode == "hybrid":
        by_id: dict = {}
        for r in keyword_results:
            by_id.setdefault(r["id"], r)
        for r in semantic_results:
            if "error" in r:
                continue
            by_id.setdefault(r["id"], r)
        semantic_rank = [r["id"] for r in semantic_results if "error" not in r]
        keyword_rank = [r["id"] for r in keyword_results]
        fused_ids = rrf_order([semantic_rank, keyword_rank])
        if not rerank_on:
            text_results = [by_id[i] for i in fused_ids if i in by_id][:20]
        else:
            # Recall-then-rerank: treat the hybrid-fused order as the SIMILARITY
            # channel (synthetic descending score preserves that rank), then blend
            # graph-importance + session-recency and diversify with MMR. Disabled
            # config would have short-circuited above, so the output can only
            # differ from plain RRF when the user opts in.
            recall_ids = [i for i in fused_ids if i in by_id][:recall_k]
            n = len(recall_ids)
            candidates = [
                Candidate(i, float(n - pos), _retrieval.MEMORY_SOURCE_GRAPH, None)
                for pos, i in enumerate(recall_ids)
            ]
            vectors = (
                storage.load_embeddings_sidecar()
                if rerank_cfg.mmr_lambda < 1.0
                else None
            )
            with _session_lock:
                touched = set(_session_accessed) | set(_session_modified)
            reranked = _retrieval.rerank_entries(
                candidates, _entry_records_for_rerank(), cfg=rerank_cfg,
                vectors=vectors, touched_ids=touched, top_k=20,
            )
            text_results = []
            for r in reranked:
                base = by_id.get(r["id"])
                if base is None:
                    continue
                enriched = dict(base)
                enriched["rerank"] = {
                    "fused": round(r["fused"], 6),
                    "importance": round(r["importance"], 4),
                    "recency": round(r["recency"], 4),
                }
                text_results.append(enriched)
    elif mode == "keyword":
        text_results = keyword_results
    else:
        text_results = semantic_results

    # --- Active code-drift surfacing (inline flag / hydrate / gate) ---
    # Only for the entries actually returned (the ~20 hits), memoized by pinned
    # ref within this call. Skipped entirely when both layers are off, so the
    # payload below is byte-identical to the pre-drift output (identity path).
    out_payload: dict = {"query": query, "mode": mode}
    if _drift_surfacing_active():
        drift_cache: dict[str, bool | None] = {}
        sym_cache: dict[str, tuple[str, str] | None] = {}
        hydrate = _drift_hydrate_enabled()
        # file_matches already carry ``files``; text/semantic hits omit pins, so
        # fetch them for just the returned ids in one lookup. Bodies are fetched
        # the same way, only when symbol refinement is on (it needs them).
        all_ids = [r.get("id") for r in file_results] + [r.get("id") for r in text_results]
        files_by_id = _files_for_ids([r.get("id") for r in text_results])
        bodies_by_id = _bodies_for_ids(all_ids) if _symbol_drift_enabled() else {}
        for r in file_results:
            _annotate_entry_drift(
                r, r.get("files") or "", drift_cache, hydrate=hydrate,
                body=bodies_by_id.get(r.get("id"), ""), sym_cache=sym_cache,
            )
        for r in text_results:
            _annotate_entry_drift(
                r, files_by_id.get(r.get("id"), ""), drift_cache, hydrate=hydrate,
                body=bodies_by_id.get(r.get("id"), ""), sym_cache=sym_cache,
            )
        _demote_stale(file_results)
        _demote_stale(text_results)
        if _drift_gate_enabled():
            quarantined = [r for r in file_results if r.get("stale")]
            quarantined += [r for r in text_results if r.get("stale")]
            file_results = [r for r in file_results if not r.get("stale")]
            text_results = [r for r in text_results if not r.get("stale")]
            if quarantined:
                out_payload["quarantined_stale"] = quarantined

    for r in text_results[:10]:
        _track(r.get("id", ""), "accessed")

    out_payload["file_matches"] = file_results
    out_payload["text_matches"] = text_results
    out_payload["total"] = len(file_results) + len(text_results)
    return json.dumps(out_payload, default=str)

codebase

codebase(action: Literal['lookup', 'blast', 'callers', 'callees', 'imports', 'context', 'ownership', 'health'] = 'lookup', query: str = '', scope: str = '', path: str = '', depth: int = 0, limit: int = 0, expand_synapse: bool = False) -> str

Use this when working with code — find a symbol, see what breaks if you change it, or learn why it exists — via the live code graph.

A structural map of the repo's functions, classes, and modules, parsed from the working tree and rebuilt when git HEAD changes. Single dispatch tool keyed on action (all return JSON). Fully back-compatible: omit action (or pass lookup) and only query / scope for the original search.

Actions — action(params): lookup(query?, scope?): default — census (no args) or name/scope CONTAINS search over functions/classes/modules; each hit carries type, file, line. blast(query|path, depth?): "what breaks if I change this?" reverse CALLS BFS from a symbol or file, depth/node-capped, graded breaks (direct) vs maybe (transitive). callers(query, limit?): 1-hop inbound call edges (with call_lines). callees(query, limit?): 1-hop outbound call edges (with call_lines). imports(query|path, limit?): file/module import neighborhood (imports + imported_by). context(query|path, expand_synapse?): "why does this exist?" the memory tip + provenance chain tethered to the anchor; expand_synapse hops to ZK. ownership(query|path): "who authored this?" git-blame attribution over the anchor's line span or whole file, ranked by lines (memory-mcp excluded). health(): graded drivers (god_object / coupling / cycle / churn / dead_code) with located file:line and a suggestion.

Parameters:

Name	Type	Description	Default
action	Literal['lookup', 'blast', 'callers', 'callees', 'imports', 'context', 'ownership', 'health']	One of lookup|blast|callers|callees|imports|context|ownership|health.	'lookup'
query	str	Symbol name/qualified_name (or search term for lookup).	''
scope	str	Directory subtree filter (lookup).	''
path	str	File-path anchor for blast/imports/context.	''
depth	int	Blast BFS depth (default 2, capped at 4).	0
limit	int	Result cap for callers/callees/imports.	0
expand_synapse	bool	For context, include the synapse-hop hint for the tip id.	False

Source code in memory/server.py

@mcp_server.tool()
def codebase(
    action: Literal[
        "lookup", "blast", "callers", "callees", "imports", "context", "ownership", "health"
    ] = "lookup",
    query: str = "",
    scope: str = "",
    path: str = "",
    depth: int = 0,
    limit: int = 0,
    expand_synapse: bool = False,
) -> str:
    """Use this when working with code — find a symbol, see what breaks if you change it, or learn why it exists — via the live code graph.

    A structural map of the repo's functions, classes, and modules, parsed from
    the working tree and rebuilt when git HEAD changes. Single dispatch tool
    keyed on ``action`` (all return JSON). Fully back-compatible: omit ``action``
    (or pass ``lookup``) and only ``query`` / ``scope`` for the original search.

    Actions — action(params):
      lookup(query?, scope?): default — census (no args) or name/scope CONTAINS
        search over functions/classes/modules; each hit carries type, file, line.
      blast(query|path, depth?): "what breaks if I change this?" reverse CALLS
        BFS from a symbol or file, depth/node-capped, graded breaks (direct) vs
        maybe (transitive).
      callers(query, limit?): 1-hop inbound call edges (with call_lines).
      callees(query, limit?): 1-hop outbound call edges (with call_lines).
      imports(query|path, limit?): file/module import neighborhood (imports +
        imported_by).
      context(query|path, expand_synapse?): "why does this exist?" the memory
        tip + provenance chain tethered to the anchor; expand_synapse hops to ZK.
      ownership(query|path): "who authored this?" git-blame attribution over the
        anchor's line span or whole file, ranked by lines (memory-mcp excluded).
      health(): graded drivers (god_object / coupling / cycle / churn /
        dead_code) with located file:line and a suggestion.

    Args:
        action: One of lookup|blast|callers|callees|imports|context|ownership|health.
        query: Symbol name/qualified_name (or search term for lookup).
        scope: Directory subtree filter (lookup).
        path: File-path anchor for blast/imports/context.
        depth: Blast BFS depth (default 2, capped at 4).
        limit: Result cap for callers/callees/imports.
        expand_synapse: For context, include the synapse-hop hint for the tip id.
    """
    cgraph, code_head, err = _ensure_code_graph()
    if err is not None:
        return err
    assert cgraph is not None

    act = (action or "lookup").lower().strip()

    if act == "lookup":
        return _codebase_lookup(cgraph, query, scope)

    if act == "health":
        root = _git_repo_root()
        return json.dumps(code_context.health(cgraph, root), default=str)

    # The remaining actions are anchored on a symbol or file. Pass the repo root
    # so an otherwise-absent bare name can fall back to a bounded source scan
    # (recovers unparsed-language symbols, e.g. Julia, by their defining file).
    root = _git_repo_root()
    anchor = code_context.resolve_anchor(cgraph, query=query, path=path, root=root)
    if anchor.get("kind") == "absent" and act != "context":
        return _err(
            f"Could not resolve anchor from query={query!r} path={path!r}.",
            "NotFound",
            hint="Pass a function/class name (query) or a file path (path); "
                 "use action='lookup' to search by name.",
        )

    if act == "blast":
        payload = code_context.blast(cgraph, anchor, depth=depth, node_cap=code_context.DEFAULT_BLAST_NODES)
        return json.dumps(code_context.with_ambiguity(payload, anchor), default=str)
    if act == "callers":
        payload = code_context.callers(cgraph, anchor, limit=limit or code_context.DEFAULT_HOP_LIMIT)
        return json.dumps(code_context.with_ambiguity(payload, anchor), default=str)
    if act == "callees":
        payload = code_context.callees(cgraph, anchor, limit=limit or code_context.DEFAULT_HOP_LIMIT)
        return json.dumps(code_context.with_ambiguity(payload, anchor), default=str)
    if act == "imports":
        payload = code_context.import_neighborhood(cgraph, anchor, limit=limit or code_context.DEFAULT_HOP_LIMIT)
        return json.dumps(code_context.with_ambiguity(payload, anchor), default=str)
    if act == "ownership":
        payload = code_context.ownership(root, anchor)
        return json.dumps(code_context.with_ambiguity(payload, anchor), default=str)
    if act == "context":
        entries = _collect_pinned_entries()
        payload = code_context.cached_anchor_context(cgraph, root, anchor, entries, code_head)
        # Attach ambiguity post-cache (anchor-derived, kept out of the disk cache).
        payload = code_context.with_ambiguity(dict(payload), anchor)
        if expand_synapse and payload.get("tip"):
            payload["expand_synapse"] = {
                "tip_id": payload["tip"].get("id"),
                "how": "call synapse(action=\"connections\", node_id=<tip_id>, expand=true) "
                       "to reach ZK canon (agent-side; memory has no zettelkasten dependency)",
            }
        return json.dumps(payload, default=str)

    return _bad_action(codebase, action)

history

history(action: Literal['digest', 'reconstruct'], granularity: str = 'auto', scope: str = '', max_eras: int = 12, max_commits: int = 5000, era: str = '', include_messages: bool = False, project: str = '', entries_json: str = '', dry_run: bool = False, force: bool = False, session_label: str = 'git-history reconstruction') -> str

Use this when onboarding an existing (brownfield) repo into memory: digest git history, then reconstruct a back-dated tree.

Bridges a codebase's git history into an initially-empty research tree so recall is useful on day one. See the onboard-existing-codebase skill.

Actions — name(required, optional?): digest(granularity?, scope?, max_eras?, max_commits?, era?, include_messages?): read-only, deterministic summary of how the repo evolved (eras, churn, representative commits, contributors, milestones), sized for an LLM to synthesize; memory-mcp's own commits excluded. granularity = auto|tag|month|quarter; drill into one era (+ include_messages). reconstruct(project, entries_json?, dry_run?, force?, session_label?): bulk-persist a back-dated tree synthesized from a digest — honours path@sha pins, auto-tags entries reconstructed/git-history, bypasses live-session drift. Refuses a tree already holding non-reconstructed entries unless force=true; dry_run previews.

Both return JSON.

Source code in memory/server.py

@mcp_server.tool()
def history(
    action: Literal["digest", "reconstruct"],
    # --- digest params ---
    granularity: str = "auto",
    scope: str = "",
    max_eras: int = 12,
    max_commits: int = 5000,
    era: str = "",
    include_messages: bool = False,
    # --- reconstruct params ---
    project: str = "",
    entries_json: str = "",
    dry_run: bool = False,
    force: bool = False,
    session_label: str = "git-history reconstruction",
) -> str:
    """Use this when onboarding an existing (brownfield) repo into memory: digest git history, then reconstruct a back-dated tree.

    Bridges a codebase's git history into an initially-empty research tree so
    ``recall`` is useful on day one. See the ``onboard-existing-codebase`` skill.

    Actions — name(required, optional?):
        digest(granularity?, scope?, max_eras?, max_commits?, era?, include_messages?):
            read-only, deterministic summary of how the repo evolved (eras, churn,
            representative commits, contributors, milestones), sized for an LLM to
            synthesize; memory-mcp's own commits excluded. ``granularity`` =
            auto|tag|month|quarter; drill into one ``era`` (+ ``include_messages``).
        reconstruct(project, entries_json?, dry_run?, force?, session_label?):
            bulk-persist a back-dated tree synthesized from a digest — honours
            ``path@sha`` pins, auto-tags entries ``reconstructed``/``git-history``,
            bypasses live-session drift. Refuses a tree already holding
            non-reconstructed entries unless ``force=true``; ``dry_run`` previews.

    Both return JSON.
    """
    act = action.lower().strip()
    if act == "digest":
        return _history_digest(
            granularity=granularity, scope=scope, max_eras=max_eras,
            max_commits=max_commits, era=era, include_messages=include_messages,
        )
    if act == "reconstruct":
        if not project:
            return json.dumps({
                "error": "reconstruct requires a project.",
                "type": "BadInputError",
            })
        return _reconstruct_history(
            project=project, entries_json=entries_json, dry_run=dry_run,
            force=force, session_label=session_label,
        )
    return _bad_action(history, action)

save_skill

save_skill(name: str, category: str, content: str, tags: str) -> str

Create or overwrite a skill.

Parameters:

Name	Type	Description	Default
name	str	Unique skill name (e.g. "connect-to-postgres")	required
category	str	"procedural" or "pattern"	required
content	str	The knowledge (markdown, code, steps)	required
tags	str	Comma-separated keywords	required

Source code in memory/server.py

def save_skill(name: str, category: str, content: str, tags: str) -> str:
    """Create or overwrite a skill.

    Args:
        name: Unique skill name (e.g. "connect-to-postgres")
        category: "procedural" or "pattern"
        content: The knowledge (markdown, code, steps)
        tags: Comma-separated keywords
    """
    now = _now()
    skill_data = {
        "id": name,
        "name": name,
        "category": category,
        "content": content,
        "tags": tags,
        "status": "active",
        "created_at": now,
        "updated_at": now,
    }

    storage.write_skill(skill_data)

    with _cache_lock:
        df = pd.DataFrame([skill_data])
        graph.add_nodes(df, node_type="skill", unique_id_field="id", node_title_field="name")

        search_text = _build_search_text(name, tags, content)
        _upsert_node("skill", name, {"search_text": search_text}, title_field="name")
        try:
            _embed_graph_texts("skill", "search_text")
        except Exception:
            pass

    _save()
    safe_name = storage._safe_filename(name)
    _schedule_memory_commit(str(storage.SKILLS_DIR / f"{safe_name}.md"))
    return json.dumps({"skill": name, "category": category, "saved": True})

get_skill

get_skill(name: str) -> str

Read full skill content by name.

Parameters:

Name	Type	Description	Default
name	str	Skill name to retrieve	required

Source code in memory/server.py

def get_skill(name: str) -> str:
    """Read full skill content by name.

    Args:
        name: Skill name to retrieve
    """
    results = list(graph.cypher(
        "MATCH (s:skill) WHERE s.id = $name AND s.status = $st RETURN s.name AS name, s.category AS category, s.content AS content, s.tags AS tags, s.updated_at AS updated_at",
        params={"name": name, "st": "active"}
    ))
    if not results:
        return _err(f"Skill '{name}' not found", "NotFound")
    return json.dumps(results[0], default=str)

search_skills

search_skills(query: str = '', mode: str = 'hybrid', category: str = '', tags: str = '') -> str

Search skills by keyword/semantic similarity, or list all when query is empty.

Parameters:

Name	Type	Description	Default
query	str	Search query (keyword or natural language). Empty = list all skills.	''
mode	str	"hybrid" (default), "keyword", or "semantic"	'hybrid'
category	str	List mode only: filter by category (procedural/pattern)	''
tags	str	List mode only: filter by tag (comma-separated, matches any)	''

Source code in memory/server.py

def search_skills(query: str = "", mode: str = "hybrid", category: str = "", tags: str = "") -> str:
    """Search skills by keyword/semantic similarity, or list all when query is empty.

    Args:
        query: Search query (keyword or natural language). Empty = list all skills.
        mode: "hybrid" (default), "keyword", or "semantic"
        category: List mode only: filter by category (procedural/pattern)
        tags: List mode only: filter by tag (comma-separated, matches any)
    """
    if not query.strip():
        return list_skills(category=category, tags=tags)

    keyword_results = []
    semantic_results = []

    # --- Keyword (over the cached corpus; no per-query table scan) ---
    if mode in ("hybrid", "keyword"):
        query_lower = query.lower()
        scored_keyword: list[tuple] = []
        for pos, s in enumerate(_skill_keyword_corpus()):
            name_match = query_lower in s["_nl"]
            tag_match = query_lower in s["_gl"]
            content_match = query_lower in s["_cl"]
            if name_match or tag_match or content_match:
                tier = 0 if name_match else (1 if tag_match else 2)
                scored_keyword.append((tier, pos, {"name": s["name"], "category": s["category"], "tags": s["tags"]}))
        scored_keyword.sort(key=lambda t: (t[0], t[1]))
        keyword_results = [r for _, _, r in scored_keyword]

    # --- Semantic ---
    if mode in ("hybrid", "semantic"):
        try:
            hits = graph.select("skill").search_text("search_text", query, top_k=10)
            for hit in hits:
                if hit.get("status") != "active":
                    continue
                semantic_results.append({
                    "name": hit.get("id", ""),
                    "category": hit.get("category", ""),
                    "tags": hit.get("tags", ""),
                    "score": round(hit.get("score", 0.0), 4),
                })
        except Exception:
            pass

    # --- Merge (Reciprocal Rank Fusion) ---
    if mode == "hybrid":
        by_name: dict = {}
        for r in keyword_results:
            by_name.setdefault(r["name"], r)
        for r in semantic_results:
            by_name.setdefault(r["name"], r)
        semantic_rank = [r["name"] for r in semantic_results]
        keyword_rank = [r["name"] for r in keyword_results]
        fused = rrf_order([semantic_rank, keyword_rank])
        results = [by_name[n] for n in fused if n in by_name][:10]
    elif mode == "keyword":
        results = keyword_results
    else:
        results = semantic_results

    return json.dumps({"query": query, "mode": mode, "results": results, "count": len(results)})

list_skills

list_skills(category: str = '', tags: str = '') -> str

List all skills, optionally filtered by category or tags.

Parameters:

Name	Type	Description	Default
category	str	Filter by category (procedural/pattern)	''
tags	str	Filter by tag (comma-separated, matches any)	''

Source code in memory/server.py

def list_skills(category: str = "", tags: str = "") -> str:
    """List all skills, optionally filtered by category or tags.

    Args:
        category: Filter by category (procedural/pattern)
        tags: Filter by tag (comma-separated, matches any)
    """
    query = "MATCH (s:skill) WHERE s.status = $st"
    params: dict = {"st": "active"}
    if category:
        query += " AND s.category = $cat"
        params["cat"] = category
    query += " RETURN s.id AS name, s.category AS category, s.tags AS tags, s.updated_at AS updated_at ORDER BY s.name"

    results = list(graph.cypher(query, params=params))

    if tags:
        tag_list = [t.strip().lower() for t in tags.split(",")]
        results = [r for r in results if any(t in (r.get("tags") or "").lower() for t in tag_list)]

    return json.dumps({"skills": results, "count": len(results)}, default=str)

delete_skill

delete_skill(name: str) -> str

Remove a skill by name (soft-delete).

Parameters:

Name	Type	Description	Default
name	str	Skill name to delete	required

Source code in memory/server.py

def delete_skill(name: str) -> str:
    """Remove a skill by name (soft-delete).

    Args:
        name: Skill name to delete
    """
    now = _now()
    storage.update_skill_field(name, {"status": "deleted", "updated_at": now})

    with _cache_lock:
        _upsert_node("skill", name, {"status": "deleted", "updated_at": now})
    _save()
    safe_name = storage._safe_filename(name)
    _schedule_memory_commit(str(storage.SKILLS_DIR / f"{safe_name}.md"))
    return json.dumps({"deleted": name})

register_document

register_document(path: str, title: str, description: str = '', project: str = '', links_to: str = '') -> str

Register a report/document and optionally link it to a tree entry.

Parameters:

Name	Type	Description	Default
path	str	Relative file path (e.g. "documents/260605_report_agent_memory_1.md")	required
title	str	Human-readable title	required
description	str	Brief description of what the document covers	''
project	str	Project name (auto-detected from graph if omitted)	''
links_to	str	Entry ID to cross-reference (creates a 'related' edge)	''

Source code in memory/server.py

def register_document(path: str, title: str, description: str = "", project: str = "", links_to: str = "") -> str:
    """Register a report/document and optionally link it to a tree entry.

    Args:
        path: Relative file path (e.g. "documents/260605_report_agent_memory_1.md")
        title: Human-readable title
        description: Brief description of what the document covers
        project: Project name (auto-detected from graph if omitted)
        links_to: Entry ID to cross-reference (creates a 'related' edge)
    """
    project = _resolve_project(project)
    doc_id = _gen_id("doc")
    now = _now()

    base_path = path.split("@")[0]
    existing = list(graph.cypher(
        "MATCH (d:document) WHERE d.project = $proj AND d.status = $st RETURN d.id AS id, d.path AS path",
        params={"proj": project, "st": "active"}
    ))
    for ex in existing:
        if ex["path"].split("@")[0] == base_path:
            storage.update_document_field(ex["id"], {"status": "discarded"})
            _upsert_node("document", ex["id"], {"status": "discarded"})

    pin_result = _pin_files(path, f"register doc: {title}")
    if pin_result.error:
        return json.dumps({
            "error": pin_result.error,
            "type": "DirtyWorkingTreeError",
            "git_status": pin_result.git_status,
        })
    pinned_path = pin_result.files

    doc_data: dict = {
        "id": doc_id,
        "path": pinned_path,
        "title": title,
        "description": description,
        "project": project,
        "status": "active",
        "created_at": now,
    }
    if links_to:
        doc_data["links_to"] = links_to

    with _cache_lock:
        storage.write_document(doc_data)

        df = pd.DataFrame([doc_data])
        graph.add_nodes(df, node_type="document", unique_id_field="id", node_title_field="title")

        if links_to:
            edge_df = pd.DataFrame([{
                "src": doc_id, "tgt": links_to,
                "rationale": f"documents {links_to}", "created_at": now,
            }])
            graph.add_connections(
                edge_df,
                connection_type="related",
                source_type="document",
                source_id_field="src",
                target_type="entry",
                target_id_field="tgt",
            )

    _save()
    _schedule_memory_commit(str(storage.DOCUMENTS_DIR / f"{doc_id}.md"))
    return json.dumps({
        "doc_id": doc_id,
        "path": pinned_path,
        "title": title,
        "linked_to": links_to or None,
        "git_status": pin_result.git_status,
        "warnings": pin_result.warnings,
    })

list_documents

list_documents(project: str = '') -> str

List all registered documents for a project.

Parameters:

Name	Type	Description	Default
project	str	Project name (auto-detected from graph if omitted)	''

Source code in memory/server.py

def list_documents(project: str = "") -> str:
    """List all registered documents for a project.

    Args:
        project: Project name (auto-detected from graph if omitted)
    """
    project = _resolve_project(project)
    results = list(graph.cypher(
        "MATCH (d:document) WHERE d.project = $proj AND d.status = $st RETURN d.id AS id, d.path AS path, d.title AS title, d.description AS description, d.created_at AS created_at ORDER BY d.created_at",
        params={"proj": project, "st": "active"}
    ))
    return json.dumps({"documents": results, "count": len(results)}, default=str)

deregister_document

deregister_document(doc_id: str) -> str

Remove a document from the tree (soft delete).

Parameters:

Name	Type	Description	Default
doc_id	str	The document ID to deregister	required

Source code in memory/server.py

def deregister_document(doc_id: str) -> str:
    """Remove a document from the tree (soft delete).

    Args:
        doc_id: The document ID to deregister
    """
    with _cache_lock:
        storage.update_document_field(doc_id, {"status": "discarded"})
        _upsert_node("document", doc_id, {"status": "discarded"})
    _save()
    _schedule_memory_commit(str(storage.DOCUMENTS_DIR / f"{doc_id}.md"))
    return json.dumps({"deregistered": doc_id})

documents

documents(action: Literal['register', 'list', 'deregister'], path: str = '', title: str = '', description: str = '', project: str = '', links_to: str = '', doc_id: str = '') -> str

Use this when you need to register, list, or deregister markdown documents (reports, plans) linked to the research tree.

Actions — name(required, optional?): register(path, title, description?, project?, links_to?): register a document; links_to is an entry ID for a 'related' edge. list(project?): list all registered documents, optionally for one project. deregister(doc_id): soft-delete a document.

project is auto-detected when omitted. All return JSON.

Source code in memory/server.py

@mcp_server.tool()
def documents(
    action: Literal["register", "list", "deregister"],
    path: str = "",
    title: str = "",
    description: str = "",
    project: str = "",
    links_to: str = "",
    doc_id: str = "",
) -> str:
    """Use this when you need to register, list, or deregister markdown documents (reports, plans) linked to the research tree.

    Actions — name(required, optional?):
        register(path, title, description?, project?, links_to?): register a
            document; ``links_to`` is an entry ID for a 'related' edge.
        list(project?): list all registered documents, optionally for one project.
        deregister(doc_id): soft-delete a document.

    ``project`` is auto-detected when omitted. All return JSON.
    """
    if action == "register":
        if not path or not title:
            return _err("action='register' requires path and title.", "BadRequest")
        return register_document(
            path=path, title=title, description=description,
            project=project, links_to=links_to,
        )
    if action == "list":
        return list_documents(project=project)
    if action == "deregister":
        if not doc_id:
            return _err("action='deregister' requires doc_id.", "BadRequest")
        return deregister_document(doc_id)
    return _bad_action(documents, action)

attach_plan

attach_plan(entry_id: str, plan_path: str) -> str

Attach an external plan file to a memory entry.

Reads the file content and stores both the path (for linking) and the full content (for semantic search portability). If the file is later deleted, the content remains searchable and the dashboard shows a 'missing' indicator.

Parameters:

Name	Type	Description	Default
entry_id	str	The memory entry to attach the plan to	required
plan_path	str	Absolute path to the .plan.md file	required

Source code in memory/server.py

def attach_plan(entry_id: str, plan_path: str) -> str:
    """Attach an external plan file to a memory entry.

    Reads the file content and stores both the path (for linking) and the full
    content (for semantic search portability). If the file is later deleted,
    the content remains searchable and the dashboard shows a 'missing' indicator.

    Args:
        entry_id: The memory entry to attach the plan to
        plan_path: Absolute path to the .plan.md file
    """
    path = Path(plan_path)
    if not path.exists():
        return _err(f"File not found: {plan_path}", "NotFound")

    content = path.read_text(encoding="utf-8")
    title = path.stem.replace("_", " ").rsplit(" ", 1)[0]

    # Use _upsert_node for reliable persistence (Cypher SET doesn't persist new properties in KGLite)
    rows = list(graph.cypher(
        "MATCH (e:entry {id: $eid}) RETURN e.title AS title, e.tags AS tags, e.body AS body",
        params={"eid": entry_id}
    ))
    if not rows:
        return _err(f"Entry not found: {entry_id}", "NotFound")

    r = rows[0]
    combined_body = f"{r.get('body', '')}\n{content}"
    search_text = _build_search_text(r.get("title", ""), r.get("tags", ""), combined_body)

    with _cache_lock:
        storage.update_entry_field(entry_id, {
            "external_plan": str(path),
            "plan_content": content,
        })
        _upsert_node("entry", entry_id, {
            "external_plan": str(path),
            "plan_content": content,
            "search_text": search_text,
        })

        try:
            _embed_graph_texts("entry", "search_text")
        except Exception:
            pass

    _save()
    _schedule_memory_commit(str(storage.ENTRIES_DIR / f"{entry_id}.md"))
    _track(entry_id, "modified")
    return json.dumps({
        "attached": entry_id,
        "plan_path": str(path),
        "plan_title": title,
        "content_length": len(content),
    })

handoff

handoff(project: str = '', summary: str = '', kind: str = 'session') -> str

Persist the current session's activity and prepare for handoff to a new agent.

Saves a session record listing which memory entries were created, accessed, and modified during this session. Call this at end of session or when handing off to a new chat window.

Parameters:

Name	Type	Description	Default
project	str	Project name. Optional — auto-detected (the sole root project) when omitted.	''
summary	str	First line is a short descriptive title; remaining lines are the body summary of what was accomplished and what's next	''
kind	str	Session classification — "session" (default), "pipeline" (coordinator run), or "task" (short single-purpose interaction)	'session'

Source code in memory/server.py

def handoff(project: str = "", summary: str = "", kind: str = "session") -> str:
    """Persist the current session's activity and prepare for handoff to a new agent.

    Saves a session record listing which memory entries were created, accessed,
    and modified during this session. Call this at end of session or when
    handing off to a new chat window.

    Args:
        project: Project name. Optional — auto-detected (the sole root project) when omitted.
        summary: First line is a short descriptive title; remaining lines are the body summary of what was accomplished and what's next
        kind: Session classification — "session" (default), "pipeline" (coordinator run), or "task" (short single-purpose interaction)
    """
    global _session_id, _session_start, _session_created, _session_accessed, _session_modified

    _ensure_session()
    project = _resolve_project(project)

    with _session_lock:
        session_id = _session_id
        session_start = _session_start
        session_created = list(_session_created)
        session_accessed = list(_session_accessed)
        session_modified = list(_session_modified)

    created_details = []
    for eid in session_created:
        rows = list(graph.cypher(
            "MATCH (e:entry) WHERE e.id = $eid RETURN e.title AS title, e.type AS type",
            params={"eid": eid}
        ))
        if rows:
            created_details.append({"id": eid, "title": rows[0].get("title", ""), "type": rows[0].get("type", "")})

    accessed_details = []
    for eid in session_accessed:
        rows = list(graph.cypher(
            "MATCH (e:entry) WHERE e.id = $eid RETURN e.title AS title, e.type AS type",
            params={"eid": eid}
        ))
        if rows:
            accessed_details.append({"id": eid, "title": rows[0].get("title", ""), "type": rows[0].get("type", "")})

    modified_details = []
    for eid in session_modified:
        rows = list(graph.cypher(
            "MATCH (e:entry) WHERE e.id = $eid RETURN e.title AS title, e.type AS type",
            params={"eid": eid}
        ))
        if rows:
            modified_details.append({"id": eid, "title": rows[0].get("title", ""), "type": rows[0].get("type", "")})

    entry_titles = [d["title"] for d in created_details + accessed_details + modified_details if d.get("title")]

    # Split summary into title (first line) + description (rest)
    summary_lines = summary.strip().split("\n", 1) if summary else [""]
    session_title = summary_lines[0].strip()
    if len(session_title) > 80:
        dot_pos = session_title.find(". ")
        if 0 < dot_pos <= 80:
            session_title = session_title[:dot_pos + 1]
        else:
            session_title = session_title[:60].rsplit(" ", 1)[0] + "..."

    # Build enriched search_text
    resolved_kind = kind if kind in ("session", "pipeline", "task") else "session"
    search_parts = [f"{resolved_kind} session", summary]
    type_counts: dict[str, int] = {}
    tags_set: set[str] = set()
    for d in created_details + accessed_details + modified_details:
        etype = d.get("type", "")
        if etype:
            type_counts[etype] = type_counts.get(etype, 0) + 1
    if type_counts:
        search_parts.append("Entries: " + ", ".join(f"{c} {t}" for t, c in type_counts.items()))
    if entry_titles:
        search_parts.append(" ".join(entry_titles))
    search_text = "\n".join(search_parts).strip()

    session_data = {
        "id": session_id,
        "project": project,
        "started_at": session_start,
        "ended_at": _now(),
        "title": session_title,
        "summary": summary,
        "kind": resolved_kind,
        "entries_created": json.dumps(created_details),
        "entries_accessed": json.dumps(accessed_details),
        "entries_modified": json.dumps(modified_details),
        "search_text": search_text,
    }

    storage.write_session(session_data)

    with _cache_lock:
        df = pd.DataFrame([session_data])
        graph.add_nodes(df, node_type="session", unique_id_field="id", node_title_field="title")

        try:
            _embed_graph_texts("session", "search_text")
        except Exception as e:
            logger.warning(f"Session embedding failed: {e}")

    _save()
    _schedule_memory_commit(str(storage.SESSIONS_DIR / f"{session_id}.md"))

    # Finalize old session and start fresh
    global _session_initialized, _last_activity
    old_id = session_id
    _session_initialized = False
    _stop_active_session_heartbeat()
    _remove_active_session(old_id)
    _init_session()
    _last_activity = time.time()

    return json.dumps({
        "session_id": old_id,
        "project": project,
        "summary": summary,
        "created": created_details,
        "accessed": accessed_details,
        "modified": modified_details,
    })

pickup

pickup(project: str = '', query: str = '') -> str

Load a previous session's context for a new agent picking up work.

With no query: returns the most recent session (default behavior). With a query: semantically searches all past sessions and returns the best matches — useful for resuming a workflow from days or months ago.

Parameters:

Name	Type	Description	Default
project	str	Project name. Optional — auto-detected (the sole root project) when omitted.	''
query	str	Optional search query to find sessions by topic (e.g. "auto-invalidation work")	''

Source code in memory/server.py

def pickup(project: str = "", query: str = "") -> str:
    """Load a previous session's context for a new agent picking up work.

    With no query: returns the most recent session (default behavior).
    With a query: semantically searches all past sessions and returns the
    best matches — useful for resuming a workflow from days or months ago.

    Args:
        project: Project name. Optional — auto-detected (the sole root project) when omitted.
        query: Optional search query to find sessions by topic (e.g. "auto-invalidation work")
    """
    _ensure_session()
    project = _resolve_project(project)
    if query:
        return _pickup_search(project, query)
    return _pickup_latest(project)

search_sessions

search_sessions(query: str = '', project: str = '', limit: int = 10) -> str

Search or list past sessions.

With a query: performs semantic search over session history to find relevant past sessions by topic (e.g. "dashboard work", "coordinator pipeline runs"). Without a query: lists the N most recent sessions, ordered newest first.

Returns lightweight summaries — use pickup(query=...) for full session context.

Parameters:

Name	Type	Description	Default
query	str	Optional search query (natural language or keywords)	''
project	str	Project name (auto-detected if empty)	''
limit	int	Max sessions to return (default 10)	10

Source code in memory/server.py

def search_sessions(query: str = "", project: str = "", limit: int = 10) -> str:
    """Search or list past sessions.

    With a query: performs semantic search over session history to find relevant
    past sessions by topic (e.g. "dashboard work", "coordinator pipeline runs").
    Without a query: lists the N most recent sessions, ordered newest first.

    Returns lightweight summaries — use pickup(query=...) for full session context.

    Args:
        query: Optional search query (natural language or keywords)
        project: Project name (auto-detected if empty)
        limit: Max sessions to return (default 10)
    """
    _ensure_session()
    project = _resolve_project(project)

    if query:
        try:
            hits = graph.select("session").search_text("search_text", query, top_k=limit)
        except Exception as e:
            return _err(f"Session search failed: {e}", "SearchError")

        results = []
        for hit in hits:
            sid = hit.get("id", "")
            score = round(hit.get("score", 0.0), 4)
            rows = list(graph.cypher(
                "MATCH (s:session) WHERE s.id = $sid AND s.project = $proj "
                "RETURN s.id AS id, s.started_at AS started_at, s.ended_at AS ended_at, "
                "s.summary AS summary, s.kind AS kind, s.title AS title, "
                "s.entries_created AS entries_created",
                params={"sid": sid, "proj": project}
            ))
            if not rows:
                continue
            row = rows[0]
            created = _safe_parse_json(row.get("entries_created"))
            summary_text = row.get("summary") or ""
            title = row.get("title") or summary_text.split("\n")[0][:80] or "Untitled"
            description = summary_text.split("\n", 1)[1].strip() if "\n" in summary_text else ""
            results.append({
                "session_id": row.get("id"),
                "title": title,
                "description": description[:200] if description else "",
                "started_at": row.get("started_at"),
                "ended_at": row.get("ended_at"),
                "kind": row.get("kind") or "session",
                "entry_count": len(created),
                "score": score,
            })

        if not results:
            return json.dumps({"query": query, "matches": [], "message": f"No sessions matched: '{query}'"})
        return json.dumps({"query": query, "matches": results}, default=str)

    # No query — list recent sessions
    sessions = list(graph.cypher(
        "MATCH (s:session) WHERE s.project = $proj "
        "RETURN s.id AS id, s.started_at AS started_at, s.ended_at AS ended_at, "
        "s.summary AS summary, s.kind AS kind, s.title AS title, "
        "s.entries_created AS entries_created "
        "ORDER BY s.ended_at DESC",
        params={"proj": project}
    ))

    results = []
    for row in sessions[:limit]:
        created = _safe_parse_json(row.get("entries_created"))
        summary_text = row.get("summary") or ""
        title = row.get("title") or summary_text.split("\n")[0][:80] or "Untitled"
        description = summary_text.split("\n", 1)[1].strip() if "\n" in summary_text else ""
        results.append({
            "session_id": row.get("id"),
            "title": title,
            "description": description[:200] if description else "",
            "started_at": row.get("started_at"),
            "ended_at": row.get("ended_at"),
            "kind": row.get("kind") or "session",
            "entry_count": len(created),
        })

    if not results:
        return json.dumps({"sessions": [], "message": "No sessions found for this project."})
    return json.dumps({"sessions": results, "total": len(sessions)}, default=str)

skill

skill(action: Literal['save', 'get', 'search', 'delete'], name: str = '', category: str = '', content: str = '', tags: str = '', query: str = '', mode: str = 'hybrid') -> str

Use this when you want to save, retrieve, search, or delete a reusable procedure/pattern in the skill library.

Actions — name(required, optional?): save(name, category, content, tags): create/overwrite a skill. get(name): retrieve full skill content. search(query?, mode?, category?, tags?): find skills; empty query lists the whole library. delete(name): remove a skill.

All return JSON.

Source code in memory/server.py

@mcp_server.tool()
def skill(
    action: Literal["save", "get", "search", "delete"],
    name: str = "",
    category: str = "",
    content: str = "",
    tags: str = "",
    query: str = "",
    mode: str = "hybrid",
) -> str:
    """Use this when you want to save, retrieve, search, or delete a reusable procedure/pattern in the skill library.

    Actions — name(required, optional?):
        save(name, category, content, tags): create/overwrite a skill.
        get(name): retrieve full skill content.
        search(query?, mode?, category?, tags?): find skills; empty ``query`` lists
            the whole library.
        delete(name): remove a skill.

    All return JSON.
    """
    if action == "save":
        return save_skill(name=name, category=category, content=content, tags=tags)
    if action == "get":
        return get_skill(name=name)
    if action == "search":
        return search_skills(query=query, mode=mode, category=category, tags=tags)
    if action == "delete":
        return delete_skill(name=name)
    return _bad_action(skill, action)

relate

relate(action: Literal['link', 'set', 'suggest', 'merge', 'bulk', 'export'], entry_a: str = '', entry_b: str = '', rationale: str = '', relationship: str = 'related', op: str = 'add', project: str = '', entry_id: str = '', relation: str = 'related', limit: int = 5, mode: str = '', ids: str = '', add_tags: str = '', remove_tags: str = '', set_type: str = '') -> str

Use this when you need to cross-reference or clean up entries: link/set edges, suggest duplicates, merge, bulk-edit tags/types, or export the graph.

Actions — name(required, optional?): link(entry_a, entry_b, rationale): create a bidirectional related cross-reference between two entries. set(entry_a, entry_b, relationship?, op?, rationale?): add/remove an explicit edge (relationship = related|invalidated_by; op = add|remove). Generalizes link. suggest(project, entry_id, relation?, limit?, mode?): preview semantically similar candidates before deciding on a relation; mode="duplicate" returns near-DUPLICATE merge targets (high cosine + lexical confirm). merge(entry_a, entry_b): losslessly collapse entry_a (loser) into entry_b (winner) — reparent children, union tags/files/relationships, redirect inbound refs, soft-discard the loser (merged_into:<winner>). bulk(ids, add_tags?, remove_tags?, set_type?): batch-edit entries (ids comma-separated) in one commit. export(): export the tree (entry nodes + has_child/related edges) as a GEXF 1.2 document for Gephi/networkx. Read-only.

All return JSON.

Source code in memory/server.py

@mcp_server.tool()
def relate(
    action: Literal["link", "set", "suggest", "merge", "bulk", "export"],
    entry_a: str = "",
    entry_b: str = "",
    rationale: str = "",
    relationship: str = "related",
    op: str = "add",
    project: str = "",
    entry_id: str = "",
    relation: str = "related",
    limit: int = 5,
    mode: str = "",
    ids: str = "",
    add_tags: str = "",
    remove_tags: str = "",
    set_type: str = "",
) -> str:
    """Use this when you need to cross-reference or clean up entries: link/set edges, suggest duplicates, merge, bulk-edit tags/types, or export the graph.

    Actions — name(required, optional?):
        link(entry_a, entry_b, rationale): create a bidirectional ``related``
            cross-reference between two entries.
        set(entry_a, entry_b, relationship?, op?, rationale?): add/remove an
            explicit edge (``relationship`` = related|invalidated_by; ``op`` =
            add|remove). Generalizes ``link``.
        suggest(project, entry_id, relation?, limit?, mode?): preview semantically
            similar candidates before deciding on a ``relation``; ``mode="duplicate"``
            returns near-DUPLICATE merge targets (high cosine + lexical confirm).
        merge(entry_a, entry_b): losslessly collapse ``entry_a`` (loser) into
            ``entry_b`` (winner) — reparent children, union tags/files/relationships,
            redirect inbound refs, soft-discard the loser (``merged_into:<winner>``).
        bulk(ids, add_tags?, remove_tags?, set_type?): batch-edit entries (``ids``
            comma-separated) in one commit.
        export(): export the tree (entry nodes + has_child/related edges) as a
            GEXF 1.2 document for Gephi/networkx. Read-only.

    All return JSON.
    """
    if action == "link":
        return link(entry_a=entry_a, entry_b=entry_b, rationale=rationale)
    if action == "set":
        return set_relationship(
            entry_a=entry_a, entry_b=entry_b, relationship=relationship,
            rationale=rationale, action=op,
        )
    if action == "suggest":
        if mode == "duplicate":
            return _suggest_duplicates(project=project, entry_id=entry_id, limit=limit)
        return suggest_relationships(
            project=project, entry_id=entry_id, relation=relation, limit=limit)
    if action == "merge":
        return _merge_entries(loser_id=entry_a, winner_id=entry_b)
    if action == "bulk":
        return _bulk_edit(
            ids=ids, add_tags=add_tags, remove_tags=remove_tags, set_type=set_type)
    if action == "export":
        from memory import export as _export

        # Pick up any external .memory/ changes before a read-only projection,
        # so the export reflects the on-disk tree (matches sibling branches,
        # which sync via the tool wrapper).
        try:
            _maybe_auto_sync()
        except Exception as exc:
            logger.warning("Auto-sync before export failed: %s", exc)
        try:
            return json.dumps({"format": "gexf", "gexf": _export.graph_to_gexf(graph)})
        except Exception as exc:
            return _err(f"export failed: {exc}", "ExportError")
    return _bad_action(relate, action)

session

session(action: Literal['handoff', 'pickup', 'search'], project: str = '', summary: str = '', kind: str = 'session', query: str = '', limit: int = 10) -> str

Use this when starting or ending a work session: hand off the current session, pick up a previous one, or search past sessions.

project is optional throughout — the sole root project is auto-detected when omitted.

Actions — name(required, optional?): handoff(project?, summary?, kind?): persist the current session's activity for the next agent (summary first line = title; kind = session|pipeline|task). Called by the session-end hook. pickup(project?, query?): load a previous session's focused context; with a query, semantically search past sessions by topic. Called by the session-start hook. search(project?, query?, limit?): search/list past sessions; empty query lists the most recent limit.

All return JSON.

Source code in memory/server.py

@mcp_server.tool()
def session(
    action: Literal["handoff", "pickup", "search"],
    project: str = "",
    summary: str = "",
    kind: str = "session",
    query: str = "",
    limit: int = 10,
) -> str:
    """Use this when starting or ending a work session: hand off the current session, pick up a previous one, or search past sessions.

    ``project`` is optional throughout — the sole root project is auto-detected
    when omitted.

    Actions — name(required, optional?):
        handoff(project?, summary?, kind?): persist the current session's activity
            for the next agent (``summary`` first line = title; ``kind`` =
            session|pipeline|task). Called by the session-end hook.
        pickup(project?, query?): load a previous session's focused context; with a
            ``query``, semantically search past sessions by topic. Called by the
            session-start hook.
        search(project?, query?, limit?): search/list past sessions; empty ``query``
            lists the most recent ``limit``.

    All return JSON.
    """
    if action == "handoff":
        return handoff(project=project, summary=summary, kind=kind)
    if action == "pickup":
        return pickup(project=project, query=query)
    if action == "search":
        return search_sessions(query=query, project=project, limit=limit)
    return _bad_action(session, action)

resolve_pin

resolve_pin(pinned_ref: str) -> str

Retrieve file content at a pinned git SHA.

Parameters:

Name	Type	Description	Default
pinned_ref	str	A pinned reference like "memory/server.py@a3f7c2e"	required

Source code in memory/server.py

@mcp_server.tool()
def resolve_pin(pinned_ref: str) -> str:
    """Retrieve file content at a pinned git SHA.

    Args:
        pinned_ref: A pinned reference like "memory/server.py@a3f7c2e"
    """
    if "@" not in pinned_ref:
        return _err("Invalid pin format — expected 'path@sha'", "BadRequest")

    path, sha = pinned_ref.rsplit("@", 1)
    repo = _get_repo()
    if repo is None:
        return _err("Not in a git repo", "NotFound")
    # Pins record short SHAs (e.g. path@a3f7c2e); resolve to the full hex SHA the
    # object store is keyed by (dulwich rejects raw-binary keys and odd-length
    # hex), then read the blob through the shared pin-resolution path.
    full_sha = _resolve_short_sha(repo, sha)
    if full_sha is None:
        return _err(f"Could not resolve pin '{pinned_ref}': unknown commit", "NotFound")
    content = _read_file_at_commit(repo, full_sha, path)
    if content is None:
        return _err(f"Could not resolve pin '{pinned_ref}'", "NotFound")
    return json.dumps({
        "path": path,
        "sha": sha,
        "lines": len(content.splitlines()),
        "content": content[:8000],
        "truncated": len(content) > 8000,
    })

drift

drift(project: str = '', action: str = 'report', apply: bool = False) -> str

Scan git-pinned entries for code drift, and optionally sweep stale ones.

Compares each pinned file (path@sha) against the current HEAD. When symbol refinement is on (MEMORY_SYMBOL_DRIFT, default), a file-level-drifted entry is further judged at symbol granularity: an entry is semantically stale only when a code symbol its body names actually changed; a file that drifted but left every named symbol untouched is a spurious flag. Entries are ranked by drift count, most drifted first.

Parameters:

Name	Type	Description	Default
project	str	Optional project filter. If empty, scans all entries.	''
action	str	"report" (default, read-only) or "sweep". A sweep tags the semantically-stale entries with a reversible stale tag when apply=True; with apply=False it is a dry run listing what would be tagged. A sweep never discards or edits bodies.	'report'
apply	bool	Only meaningful for action="sweep" — actually write the stale tag. Defaults to a dry run so the sweep is safe to call blind.	False

Source code in memory/server.py

@mcp_server.tool()
def drift(project: str = "", action: str = "report", apply: bool = False) -> str:
    """Scan git-pinned entries for code drift, and optionally sweep stale ones.

    Compares each pinned file (``path@sha``) against the current HEAD. When symbol
    refinement is on (``MEMORY_SYMBOL_DRIFT``, default), a file-level-drifted entry
    is further judged at symbol granularity: an entry is *semantically stale* only
    when a code symbol its body names actually changed; a file that drifted but
    left every named symbol untouched is a *spurious* flag. Entries are ranked by
    drift count, most drifted first.

    Args:
        project: Optional project filter. If empty, scans all entries.
        action: ``"report"`` (default, read-only) or ``"sweep"``. A sweep tags the
            semantically-stale entries with a reversible ``stale`` tag when
            ``apply=True``; with ``apply=False`` it is a dry run listing what would
            be tagged. A sweep never discards or edits bodies.
        apply: Only meaningful for ``action="sweep"`` — actually write the ``stale``
            tag. Defaults to a dry run so the sweep is safe to call blind.
    """
    action = (action or "report").strip().lower()
    if action not in ("report", "sweep"):
        return _err("drift action must be 'report' or 'sweep'", "BadRequest")

    match_clause = "MATCH (e:entry)" if not project else (
        "MATCH (e:entry) WHERE e.project = $proj"
    )
    params = {"proj": project} if project else {}
    query = (
        f"{match_clause} WHERE e.files IS NOT NULL AND e.files <> '' "
        "RETURN e.id AS id, e.title AS title, e.files AS files, e.type AS type, "
        "e.body AS body"
    )
    results = list(graph.cypher(query, params=params))

    # HEAD is constant across the scan, so memo per pinned ref (file drift) and
    # per ref blob-reads (symbol refinement).
    drift_cache: dict[str, bool | None] = {}
    sym_cache: dict[str, tuple[str, str] | None] = {}
    use_symbols = _symbol_drift_enabled()

    drifted_entries: list[dict] = []
    semantically_stale_ids: list[str] = []
    spurious_only = 0
    for row in results:
        files_val = row.get("files", "")
        if not files_val:
            continue
        cls = _classify_pins(files_val, drift_cache)
        drifted_refs = cls["drifted_refs"]
        if not drifted_refs:
            continue
        # Refine to the subset whose named symbols actually changed (or that we
        # cannot judge, kept conservatively). Empty ⇒ every flag was spurious.
        body = row.get("body") or ""
        if use_symbols and body:
            refined_refs, drifted_symbols = _refine_drifted_refs(
                body, drifted_refs, sym_cache
            )
        else:
            refined_refs, drifted_symbols = drifted_refs, []
        semantically_stale = bool(refined_refs)
        if semantically_stale:
            semantically_stale_ids.append(row["id"])
        else:
            spurious_only += 1
        drifted_entries.append({
            "id": row["id"],
            "title": row["title"],
            "type": row["type"],
            # Full ``path@sha`` refs (not bare paths) — the audit tool's historical
            # shape, kept for the dashboard.
            "drifted_files": drifted_refs,
            "unchanged_files": cls["unchanged_refs"],
            "drift_count": len(drifted_refs),
            "semantically_stale": semantically_stale,
            "drifted_symbols": drifted_symbols,
            "spurious": not semantically_stale,
        })

    drifted_entries.sort(key=lambda x: x["drift_count"], reverse=True)
    report: dict = {
        "action": action,
        "total_entries_scanned": len(results),
        "entries_with_drift": len(drifted_entries),
        "entries_semantically_stale": len(semantically_stale_ids),
        "entries_spurious_only": spurious_only,
        "symbol_refinement": use_symbols,
        "drifted": drifted_entries,
    }

    if action == "sweep":
        report["would_tag"] = semantically_stale_ids
        report["would_tag_count"] = len(semantically_stale_ids)
        if apply and semantically_stale_ids:
            tag_result = json.loads(_bulk_edit(",".join(semantically_stale_ids), add_tags="stale"))
            report["applied"] = True
            report["tagged"] = tag_result.get("updated", [])
            report["tag"] = "stale"
        else:
            report["applied"] = False
            if not apply:
                report["note"] = "dry run — pass apply=true to write the 'stale' tag"

    return json.dumps(report, indent=2)

sync

sync(force: bool = False) -> str

Reconcile .memory/ files with the in-memory cache.

Call after git pull or external edits to .memory/ files. Reads what changed and applies an incremental delta by default; pass force=True (or hit a schema mismatch / deletion) to rebuild the whole KGLite cache from scratch. Reports what changed either way.

Parameters:

Name	Type	Description	Default
force	bool	Rebuild the entire cache from files instead of applying a delta.	False

Source code in memory/server.py

def sync(force: bool = False) -> str:
    """Reconcile .memory/ files with the in-memory cache.

    Call after git pull or external edits to .memory/ files. Reads what
    changed and applies an incremental delta by default; pass ``force=True``
    (or hit a schema mismatch / deletion) to rebuild the whole KGLite cache
    from scratch. Reports what changed either way.

    Args:
        force: Rebuild the entire cache from files instead of applying a delta.
    """
    global graph, _auto_sync_index, _auto_sync_last_check

    old_index = storage.load_index()
    new_index = storage.build_index()
    changed = storage.get_changed_files(old_index, new_index)

    if force or not _apply_delta(changed):
        _rebuild_with_embeddings(skip_ids=_changed_entry_ids(changed))
        mode = "full rebuild"
    else:
        mode = "incremental delta"

    storage.save_index(new_index)
    with _auto_sync_lock:
        _auto_sync_index = new_index
        _auto_sync_last_check = time.monotonic()

    data = storage.read_all()
    return json.dumps({
        "entries": len(data["entries"]),
        "skills": len(data["skills"]),
        "sessions": len(data["sessions"]),
        "documents": len(data["documents"]),
        "changed_files": changed,
        "changes": len(changed),
        "message": f"Cache reconciled from .memory/ files ({mode}).",
    })

health

health() -> str

Report memory system health and diagnostics.

Returns cache freshness, unpushed commit count, entry/skill/session counts, embeddings coverage, format version, and install location.

Source code in memory/server.py

@mcp_server.tool()
def health() -> str:
    """Report memory system health and diagnostics.

    Returns cache freshness, unpushed commit count, entry/skill/session counts,
    embeddings coverage, format version, and install location.
    """
    config = storage.read_config()
    data = storage.read_all()

    old_index = storage.load_index()
    new_index = storage.build_index()
    changed = storage.get_changed_files(old_index, new_index)

    entry_count = len(data["entries"])
    skill_count = len(data["skills"])
    session_count = len(data["sessions"])
    document_count = len(data["documents"])

    active_entries = sum(1 for e in data["entries"] if e.get("status") == "active")
    discarded_entries = entry_count - active_entries

    # Project-root sanity: the store should have exactly one root. More than one
    # means phantom roots have accreted (see repair_projects); surface it as a
    # warning rather than letting it silently drift. Guard the graph read under
    # _cache_lock — health() is auto-sync-exempt, so a concurrent write could
    # otherwise trigger a KGLite PyBorrowError panic.
    try:
        with _cache_lock:
            project_roots = [r["id"] for r in graph.cypher("MATCH (p:project) RETURN p.id AS id")]
    except Exception:
        project_roots = []
    project_roots_warning = (
        f"Store has {len(project_roots)} project roots {project_roots} but should "
        f"have exactly 1. Run repair_projects to fold strays under the canonical root."
        if len(project_roots) > 1 else None
    )

    embeddings_info: dict = {}
    # health() is auto-sync-exempt and so skips the wrapper's serialization;
    # guard this graph read explicitly so a concurrent write can't trigger a
    # KGLite PyBorrowError panic (which would crash the process, not raise).
    try:
        with _cache_lock:
            existing = graph.list_embeddings() or []
        for emb in existing:
            embeddings_info[emb.get("node_type", "?")] = emb.get("text_column", "?")
    except Exception:
        pass

    unpushed = 0
    git_status: dict = {}
    repo = _get_repo()
    if repo is not None:
        try:
            local_head = repo.head()
            symrefs = repo.refs.get_symrefs()
            head_ref = symrefs.get(b"HEAD", b"")
            if head_ref.startswith(b"refs/heads/"):
                branch = head_ref[len(b"refs/heads/"):]
                remote_ref = b"refs/remotes/origin/" + branch
                remote_head = repo.refs.get(remote_ref)
                if remote_head and local_head != remote_head:
                    walker = repo.get_walker(include=[local_head], exclude=[remote_head])
                    unpushed = sum(1 for _ in walker)
        except Exception:
            pass
    root = _git_repo_root()
    if root is not None:
        try:
            git_status = git_state.snapshot(root).to_dict()
        except Exception:
            git_status = {}

    import memory
    install_location = str(Path(memory.__file__).parent)

    return json.dumps({
        "format_version": config.get("format_version", "unknown"),
        "data_version": storage.FORMAT_VERSION,
        "install_location": install_location,
        "memory_dir": str(storage.MEMORY_DIR.resolve()) if storage.MEMORY_DIR.exists() else None,
        "cache_stale_files": changed,
        "cache_stale": len(changed) > 0,
        "auto_sync": {
            "enabled": _auto_sync_enabled(),
            "interval_seconds": _AUTO_SYNC_INTERVAL_SEC,
        },
        "counts": {
            "entries": entry_count,
            "entries_active": active_entries,
            "entries_discarded": discarded_entries,
            "skills": skill_count,
            "sessions": session_count,
            "documents": document_count,
        },
        "project_roots": project_roots,
        "project_roots_warning": project_roots_warning,
        "embeddings": embeddings_info,
        "unpushed_commits": unpushed,
        "git_status": git_status,
    })

vacuum

vacuum(days_old: int = 30, dry_run: bool = True) -> str

Archive discarded entries older than N days and rebuild the cache.

Moves qualifying entry files to .memory/archive/entries/, then rebuilds the KGLite cache from the remaining .memory/ files.

Parameters:

Name	Type	Description	Default
days_old	int	Only remove entries discarded more than this many days ago (default 30)	30
dry_run	bool	If true (default), report what would be removed without changing anything	True

Source code in memory/server.py

def vacuum(days_old: int = 30, dry_run: bool = True) -> str:
    """Archive discarded entries older than N days and rebuild the cache.

    Moves qualifying entry files to .memory/archive/entries/, then rebuilds
    the KGLite cache from the remaining .memory/ files.

    Args:
        days_old: Only remove entries discarded more than this many days ago (default 30)
        dry_run: If true (default), report what would be removed without changing anything
    """
    global graph

    candidates = list(graph.cypher(
        "MATCH (e:entry) WHERE e.status = $st RETURN e.id AS id, e.title AS title, "
        "e.type AS type, e.updated_at AS updated_at",
        params={"st": "discarded"}
    ))

    to_purge = []
    for c in candidates:
        updated = c.get("updated_at", "")
        if not updated:
            to_purge.append(c)
            continue
        try:
            updated_dt = datetime.fromisoformat(updated)
            age_days = (datetime.now(timezone.utc) - updated_dt).days
            if age_days >= days_old:
                to_purge.append(c)
        except (ValueError, TypeError):
            to_purge.append(c)

    if dry_run or not to_purge:
        return json.dumps({
            "dry_run": True,
            "would_purge": len(to_purge),
            "entries": [{"id": e["id"], "title": e["title"], "type": e["type"]} for e in to_purge],
            "kept": len(candidates) - len(to_purge),
            "message": "Run with dry_run=False to execute." if to_purge else "Nothing to purge.",
        })

    archived_paths = []
    for e in to_purge:
        result = storage.archive_entry(e["id"])
        if result:
            archived_paths.append(str(result))
            src = str(storage.ENTRIES_DIR / f"{e['id']}.md")
            archived_paths.append(src)

    _rebuild_with_embeddings()

    for p in archived_paths:
        _schedule_memory_commit(p)

    data = storage.read_all()
    return json.dumps({
        "dry_run": False,
        "purged": len(to_purge),
        "entries_purged": [{"id": e["id"], "title": e["title"]} for e in to_purge],
        "entries_remaining": len(data["entries"]),
    })

repair_projects

repair_projects(dry_run: bool = True) -> str

Fold stray/phantom project roots back under the single canonical root.

A store must have exactly one project root (see create_project). Accidents — historically a typo'd project/parent_id in an old record() call — could mint extra "phantom" roots that hold real entries as an orphaned subtree, which the dashboard surfaces via its multi-project switcher. This finds entries not under the canonical root (a project tag other than the canonical one, or a parent_id that resolves to neither an existing entry nor the root) and, unless dry_run, retags them to the canonical project and reparents the dangling ones onto the root, then rebuilds the cache so the now-empty phantom project nodes disappear.

The canonical root is the one backed by .memory/project.md. Entries whose parent_id points at another entry are left in place (only their tag is fixed), so an entire mis-rooted subtree collapses under the root as a subproject with its internal shape intact.

Parameters:

Name	Type	Description	Default
dry_run	bool	If true (default), report what would change without writing.	True

Source code in memory/server.py

@mcp_server.tool()
def repair_projects(dry_run: bool = True) -> str:
    """Fold stray/phantom project roots back under the single canonical root.

    A store must have exactly one project root (see create_project). Accidents —
    historically a typo'd ``project``/``parent_id`` in an old record() call —
    could mint extra "phantom" roots that hold real entries as an orphaned
    subtree, which the dashboard surfaces via its multi-project switcher. This
    finds entries not under the canonical root (a ``project`` tag other than the
    canonical one, or a ``parent_id`` that resolves to neither an existing entry
    nor the root) and, unless ``dry_run``, retags them to the canonical project
    and reparents the dangling ones onto the root, then rebuilds the cache so the
    now-empty phantom project nodes disappear.

    The canonical root is the one backed by ``.memory/project.md``. Entries whose
    ``parent_id`` points at another entry are left in place (only their tag is
    fixed), so an entire mis-rooted subtree collapses under the root as a
    subproject with its internal shape intact.

    Args:
        dry_run: If true (default), report what would change without writing.
    """
    data = storage.read_all()
    if not data["project"] or not data["project"].get("id"):
        return json.dumps({
            "error": "No canonical project root (.memory/project.md missing). "
                     "Nothing to repair.",
            "type": "NoProjectRootError",
        })
    canonical = data["project"]["id"]

    entries = data["entries"]
    entry_ids = {e["id"] for e in entries if e.get("id")}
    phantom_roots = [p for p in _known_project_ids() if p != canonical]

    changed: list[dict] = []
    for e in entries:
        eid = e.get("id")
        if not eid:
            continue
        updates: dict = {}
        proj = e.get("project")
        if proj and proj != canonical:
            updates["project"] = canonical
        pid = e.get("parent_id")
        # Dangling parent: not the root and not an existing entry (e.g. it points
        # at a phantom project id). Reparent onto the canonical root. A parent
        # that is a real entry is left alone so subtree shape is preserved.
        if pid and pid != canonical and pid not in entry_ids:
            updates["parent_id"] = canonical
        if updates:
            changed.append({
                "id": eid,
                "title": e.get("title", ""),
                "from_project": proj,
                "from_parent": pid,
                "updates": updates,
            })

    if dry_run:
        return json.dumps({
            "dry_run": True,
            "canonical_project": canonical,
            "phantom_roots": phantom_roots,
            "entries_to_repair": len(changed),
            "changes": changed,
            "message": (
                "Run with dry_run=False to apply."
                if changed else "Nothing to repair — single clean root."
            ),
        })

    if not changed:
        return json.dumps({
            "dry_run": False,
            "canonical_project": canonical,
            "repaired": 0,
            "message": "Nothing to repair — single clean root.",
        })

    now = _now()
    committed: list[str] = []
    for c in changed:
        upd = dict(c["updates"])
        upd["updated_at"] = now
        path = storage.update_entry_field(c["id"], upd)
        if path:
            committed.append(str(path))

    # Rebuild from the now-fixed files: the canonical root is the only project.md,
    # every parent_id resolves to an entry or the root, so the phantom project
    # nodes (childless, file-less) are not recreated and vanish from the graph.
    _rebuild_with_embeddings()
    for p in committed:
        _schedule_memory_commit(p)

    return json.dumps({
        "dry_run": False,
        "canonical_project": canonical,
        "repaired": len(changed),
        "phantom_roots_before": phantom_roots,
        "projects_after": _known_project_ids(),
        "entries_repaired": [{"id": c["id"], "title": c["title"]} for c in changed],
    })

launch_dashboard

launch_dashboard(force_restart: bool = False) -> str

Launch the memory research-tree dashboard (backend + frontend).

Adapts to the install: the backend always starts and, in a plain (non-editable) install, serves the bundled UI itself on :8744. In a source checkout it also starts the Vite dev server on :5175 (live reload). The chat agent runs in-process in the backend (cursor-sdk's bundled bridge — no system Node.js needed). If already running, returns the URL immediately unless force_restart=True; handles port conflicts by killing stale processes.

This is the memory/research-tree visualizer — for browsing entries, sessions, skills, and the code graph. NOT the zettelkasten dashboard.

FEDERATED VIEW — viewing another repo's tree inside this tree: The dashboard is the ONLY place that can show a related repo's memory tree alongside this repo's (the recall/search tools are local-repo only). It is a read-only overlay — external entries are projected in, never imported or mutated. To guide a user through it: 1. Edit .memory/config.yaml in THIS repo and add a federated_repos list. Each entry needs path (absolute, or relative to the workspace root) pointing at the other repo, plus an id and name. The target repo must already have a .memory/project.md. Example: federated_repos: - id: other-repo name: Other Repo path: C:\path\to\other-repo enabled: true # optional, default true 2. Call launch_dashboard (or rerun with force_restart=True if it is already open) and open the URL. 3. In the toolbar, use the repo dropdown: "All repos" merges the federated subtrees under the local project root, or pick a single repo to isolate it. Federated nodes are badged read-only. Misconfigured repos surface non-fatal errors in the dashboard's federationErrors; the most common cause is a path that does not resolve or a target missing .memory/project.md.

Returns the dashboard URL and component status.

Source code in memory/server.py

@mcp_server.tool()
def launch_dashboard(force_restart: bool = False) -> str:
    """Launch the memory research-tree dashboard (backend + frontend).

    Adapts to the install: the backend always starts and, in a plain
    (non-editable) install, serves the bundled UI itself on :8744. In a source
    checkout it also starts the Vite dev server on :5175 (live reload). The chat
    agent runs in-process in the backend (cursor-sdk's bundled bridge — no
    system Node.js needed). If already running, returns the URL immediately
    unless force_restart=True; handles port conflicts by killing stale processes.

    This is the memory/research-tree visualizer — for browsing entries,
    sessions, skills, and the code graph. NOT the zettelkasten dashboard.

    FEDERATED VIEW — viewing another repo's tree inside this tree:
    The dashboard is the ONLY place that can show a related repo's memory tree
    alongside this repo's (the `recall`/`search` tools are local-repo only). It
    is a read-only overlay — external entries are projected in, never imported
    or mutated. To guide a user through it:
      1. Edit `.memory/config.yaml` in THIS repo and add a `federated_repos`
         list. Each entry needs `path` (absolute, or relative to the workspace
         root) pointing at the other repo, plus an `id` and `name`. The target
         repo must already have a `.memory/project.md`. Example:
             federated_repos:
               - id: other-repo
                 name: Other Repo
                 path: C:\\path\\to\\other-repo
                 enabled: true   # optional, default true
      2. Call `launch_dashboard` (or rerun with force_restart=True if it is
         already open) and open the URL.
      3. In the toolbar, use the repo dropdown: "All repos" merges the
         federated subtrees under the local project root, or pick a single repo
         to isolate it. Federated nodes are badged read-only.
    Misconfigured repos surface non-fatal errors in the dashboard's
    `federationErrors`; the most common cause is a `path` that does not resolve
    or a target missing `.memory/project.md`.

    Returns the dashboard URL and component status.
    """
    import shutil

    dashboard_dir = Path(__file__).resolve().parent / "dashboard"
    frontend_dir = dashboard_dir / "frontend"
    workspace = Path(os.getcwd()).resolve()

    from memory.dashboard.launcher_core import (
        build_id,
        fetch_health,
        health_url,
        resolved_ports,
    )

    # Resolve ports the SAME way the `python -m memory.dashboard` launcher does
    # (per-workspace offset unless an env var overrides), so the MCP tool and the
    # launcher never land on different ports for one workspace and spawn dupes.
    backend_port, frontend_port = resolved_ports(
        workspace, 8744, 5175, "DASHBOARD_BACKEND_PORT", "DASHBOARD_FRONTEND_PORT"
    )

    # Capability detection — mode-aware, never exits the server.
    have_built_ui = (frontend_dir / "dist" / "index.html").is_file()
    have_frontend_src = (frontend_dir / "package.json").is_file()

    status: dict = {"actions": []}

    if not have_built_ui and not have_frontend_src:
        status["status"] = "error"
        status["error"] = (
            "no dashboard UI found (no built bundle and no frontend source) — "
            "reinstall angelo"
        )
        return json.dumps(status)

    # --- Backend (always; serves bundled UI when Vite isn't running) ---
    reload_flag = ["--reload"] if have_frontend_src else []
    backend_alive = _port_listening(backend_port)
    # Recycle a backend that's alive but running stale code (old build after an
    # upgrade or source edit), so every launch_dashboard call self-heals — not
    # just explicit force_restart=True.
    backend_stale = False
    if backend_alive:
        payload = fetch_health(health_url(backend_port))
        if payload and payload.get("build") and payload.get("build") != build_id():
            backend_stale = True
    if backend_alive and (force_restart or backend_stale):
        _kill_port(backend_port)
        time.sleep(1)
        backend_alive = False
        status["actions"].append(
            f"recycled stale backend on :{backend_port}"
            if backend_stale and not force_restart
            else f"killed stale backend on :{backend_port}"
        )

    if not backend_alive:
        proc = subprocess.Popen(
            [sys.executable, "-m", "uvicorn",
             "memory.dashboard.backend.main:app",
             *reload_flag, "--port", str(backend_port)],
            cwd=str(workspace),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        _dashboard_procs.append(proc)
        if _wait_for_port(backend_port):
            status["actions"].append(f"started backend on :{backend_port}")
        else:
            status["actions"].append(f"backend failed to start on :{backend_port}")
            status["status"] = "error"
            return json.dumps(status)
    else:
        status["actions"].append(f"backend already running on :{backend_port}")

    # The chat agent runs in-process in the backend (cursor-sdk's bundled
    # bridge); the Cursor API key is read lazily on the first chat, so there is
    # no separate sidecar to launch and no Node.js requirement.

    # --- Frontend: live Vite only in a source checkout ---
    run_vite = have_frontend_src and _ensure_node_deps_quiet(frontend_dir)
    if run_vite:
        npm = shutil.which("npm") or "npm"
        frontend_alive = _port_listening(frontend_port)
        if frontend_alive and force_restart:
            _kill_port(frontend_port)
            time.sleep(1)
            frontend_alive = False
            status["actions"].append(f"killed stale frontend on :{frontend_port}")
        if not frontend_alive:
            proc = subprocess.Popen(
                [npm, "run", "dev"],
                cwd=str(frontend_dir),
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
            _dashboard_procs.append(proc)
            if _wait_for_port(frontend_port, timeout=15.0):
                status["actions"].append(f"started frontend on :{frontend_port}")
            else:
                status["actions"].append(f"frontend failed to start on :{frontend_port}")
                run_vite = False
        else:
            status["actions"].append(f"frontend already running on :{frontend_port}")
    elif have_frontend_src:
        status["actions"].append("frontend (Vite) skipped: serving bundled UI from backend")
    else:
        status["actions"].append(f"serving bundled UI from backend on :{backend_port}")

    from memory.dashboard.launcher_core import dashboard_url

    ui_port = frontend_port if run_vite else backend_port
    status["url"] = dashboard_url(ui_port)
    status["status"] = "running"
    return json.dumps(status)