memory.dashboard.backend.routes.sessions

memory.dashboard.backend.routes.sessions

Session listing + lifecycle routes.

Split out of the former flat routes.py; behaviour is unchanged.

update_session

update_session(session_id: str, updates: SessionUpdate)

Update user-editable session fields on a live sidecar or persisted session.

Source code in memory/dashboard/backend/routes/sessions.py

@router.patch("/sessions/{session_id}")
def update_session(session_id: str, updates: SessionUpdate):
    """Update user-editable session fields on a live sidecar or persisted session."""
    if federation.split_namespace(session_id) is not None:
        raise HTTPException(status_code=403, detail="Federated sessions are read-only")

    if updates.title is None and updates.summary is None and updates.notes is None:
        raise HTTPException(status_code=400, detail="No updates provided")

    active = _read_active_session(session_id)
    if active is not None:
        if updates.title is not None:
            active["title"] = updates.title
        if updates.summary is not None:
            active["summary"] = updates.summary
        if updates.notes is not None:
            active["notes"] = updates.notes
        active["updated_at"] = datetime.now(timezone.utc).isoformat()
        path = _active_session_path(session_id)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(active, indent=2), encoding="utf-8")
        return {"id": session_id, "source": "active", "status": "updated"}

    from memory import storage

    path = storage.SESSIONS_DIR / f"{session_id}.md"
    if not path.exists():
        raise HTTPException(status_code=404, detail=f"Session not found: {session_id}")
    data = storage.read_file(path)
    if "body" in data and "search_text" not in data:
        data["search_text"] = data.pop("body")
    if updates.title is not None:
        data["title"] = updates.title
    if updates.summary is not None:
        data["summary"] = updates.summary
    if updates.notes is not None:
        data["notes"] = updates.notes
    if not data.get("search_text"):
        data["search_text"] = "\n".join(
            str(part) for part in [data.get("title"), data.get("summary"), data.get("notes")] if part
        )
    storage.write_session(data)
    _write_session_node(data)
    return {"id": session_id, "source": "persisted", "status": "updated"}

complete_session

complete_session(session_id: str, updates: SessionUpdate)

Mark a live/stale session sidecar complete and persist it as a session.

Source code in memory/dashboard/backend/routes/sessions.py

@router.post("/sessions/{session_id}/complete")
def complete_session(session_id: str, updates: SessionUpdate):
    """Mark a live/stale session sidecar complete and persist it as a session."""
    if federation.split_namespace(session_id) is not None:
        raise HTTPException(status_code=403, detail="Federated sessions are read-only")

    active = _read_active_session(session_id)
    if active is None:
        raise HTTPException(status_code=404, detail=f"Active session not found: {session_id}")

    now = datetime.now(timezone.utc).isoformat()
    if updates.title is not None:
        active["title"] = updates.title
    if updates.summary is not None:
        active["summary"] = updates.summary
    if updates.notes is not None:
        active["notes"] = updates.notes

    session_data = {
        "id": session_id,
        "project": active.get("project") or _get_project()["id"],
        "started_at": active.get("started_at") or now,
        "ended_at": now,
        "title": active.get("title") or "",
        "summary": active.get("summary") or active.get("title") or "Session completed by user",
        "kind": active.get("kind") or "session",
        "entries_created": json.dumps(active.get("entries_created") or []),
        "entries_accessed": json.dumps(active.get("entries_accessed") or []),
        "entries_modified": json.dumps(active.get("entries_modified") or []),
        "notes": str(active.get("notes") or ""),
        "search_text": "\n".join(
            str(part) for part in [
                active.get("title") or "",
                active.get("summary") or "",
                active.get("notes") or "",
            ] if part
        ),
    }
    _persist_session_data(session_data)

    request_path = _active_session_completion_path(session_id)
    request_path.parent.mkdir(parents=True, exist_ok=True)
    request_path.write_text(json.dumps({
        "id": session_id,
        "title": session_data["title"],
        "summary": session_data["summary"],
        "notes": session_data["notes"],
        "requested_at": now,
    }, indent=2), encoding="utf-8")

    try:
        _active_session_path(session_id).unlink(missing_ok=True)
    except OSError:
        pass

    return {"id": session_id, "status": "completed"}

delete_session

delete_session(session_id: str)

Delete a session entirely — removes the persisted file, active sidecar, and clears session_id on entries.

Source code in memory/dashboard/backend/routes/sessions.py

@router.delete("/sessions/{session_id}")
def delete_session(session_id: str):
    """Delete a session entirely — removes the persisted file, active sidecar, and clears session_id on entries."""
    from memory import storage

    if federation.split_namespace(session_id) is not None:
        raise HTTPException(status_code=403, detail="Federated sessions are read-only")

    persisted_path = storage.SESSIONS_DIR / f"{session_id}.md"
    active_path = _active_session_path(session_id)
    completion_path = _active_session_completion_path(session_id)

    found = persisted_path.exists() or active_path.exists()
    if not found:
        try:
            graph = loader.graph
            rows = list(graph.cypher(
                "MATCH (s:session) WHERE s.id = $sid RETURN s.id AS id",
                params={"sid": session_id},
            ))
            found = len(rows) > 0
        except Exception:
            pass
    if not found:
        raise HTTPException(status_code=404, detail=f"Session not found: {session_id}")

    for p in (persisted_path, active_path, completion_path):
        try:
            p.unlink(missing_ok=True)
        except OSError:
            pass

    cleared_entries: list[str] = []
    try:
        graph = loader.graph
        entry_rows = list(graph.cypher(
            "MATCH (e:entry) WHERE e.session_id = $sid RETURN e.id AS id",
            params={"sid": session_id},
        ))
        for row in entry_rows:
            eid = row.get("id")
            if eid:
                storage.update_entry_field(eid, {"session_id": ""})
                cleared_entries.append(eid)
    except Exception:
        pass

    # Deleting the .md/sidecar files doesn't advance the newest source mtime, so
    # the cached graph would keep serving the deleted session node and it would
    # reappear on the next list. Force a rebuild from the (now-updated) files.
    loader.invalidate()

    return {"id": session_id, "status": "deleted", "cleared_entries": cleared_entries}

reopen_session

reopen_session(session_id: str)

Reopen a persisted session as the current live session.

Source code in memory/dashboard/backend/routes/sessions.py

@router.post("/sessions/{session_id}/reopen")
def reopen_session(session_id: str):
    """Reopen a persisted session as the current live session."""
    from memory import storage

    if federation.split_namespace(session_id) is not None:
        raise HTTPException(status_code=403, detail="Federated sessions are read-only")

    active = _read_active_session(session_id)
    if active is not None:
        return {"id": session_id, "status": "already_active"}

    path = storage.SESSIONS_DIR / f"{session_id}.md"
    if not path.exists():
        raise HTTPException(status_code=404, detail=f"Session not found: {session_id}")
    data = storage.read_file(path)
    if "body" in data and "search_text" not in data:
        data["search_text"] = data.pop("body")

    now = datetime.now(timezone.utc).isoformat()
    sidecar = {
        "schema_version": 1,
        "id": session_id,
        "project": data.get("project") or _get_project()["id"],
        "process_id": None,
        "started_at": data.get("started_at") or now,
        "ended_at": None,
        "title": data.get("title") or "",
        "summary": data.get("summary") or "",
        "kind": data.get("kind") or "session",
        "entries_created": _json_list(data.get("entries_created")),
        "entries_accessed": _json_list(data.get("entries_accessed")),
        "entries_modified": _json_list(data.get("entries_modified")),
        "notes": str(data.get("notes") or ""),
        "duration_minutes": None,
        "active": True,
        "reopened": True,
        "reopened_at": now,
        "last_activity": now,
        "heartbeat_at": now,
        "updated_at": now,
    }
    active_path = _active_session_path(session_id)
    active_path.parent.mkdir(parents=True, exist_ok=True)
    active_path.write_text(json.dumps(sidecar, indent=2), encoding="utf-8")

    request_path = _active_session_reopen_path()
    request_path.write_text(json.dumps({
        "id": session_id,
        "project": sidecar["project"],
        "requested_at": now,
    }, indent=2), encoding="utf-8")

    try:
        _active_session_completion_path(session_id).unlink(missing_ok=True)
    except OSError:
        pass

    return {"id": session_id, "status": "reopened"}