wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
419 lines
16 KiB
Python
419 lines
16 KiB
Python
"""High-level facade for the three-layer memory subsystem.
|
|
|
|
All callers — API routers, LLM tools, surface event hooks — go through
|
|
:class:`MemoryStore`. The store is stateless; per-user isolation is
|
|
inherited from :func:`paths.memory_root` which resolves :class:`PathService`
|
|
lazily via context variables.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import logging
|
|
from pathlib import Path
|
|
import shutil
|
|
from typing import Literal
|
|
|
|
from deeptutor.services.memory import consolidator, paths, trace
|
|
from deeptutor.services.memory.consolidator import ConsolidateResult, OnEvent
|
|
from deeptutor.services.memory.document import Document, parse, serialize
|
|
from deeptutor.services.memory.ops import AddOp, ApplyReport, EditOp
|
|
from deeptutor.services.memory.ops import apply as ops_apply
|
|
from deeptutor.services.memory.paths import L3Slot, Surface
|
|
from deeptutor.services.memory.trace import TraceEvent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
Layer = Literal["L2", "L3"]
|
|
|
|
_V1_FILES = ("PROFILE.md", "SUMMARY.md")
|
|
_NO_MEMORY = (
|
|
"(No memory available — interact with DeepTutor and update from the Memory page to build one.)"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class DocOverview:
|
|
layer: Layer
|
|
key: str # surface name (L2) or slot name (L3)
|
|
exists: bool
|
|
updated_at: str | None
|
|
entry_count: int
|
|
backlog: int # L1 events since last update (L2 only; 0 for L3)
|
|
|
|
|
|
class MemoryStore:
|
|
"""Stateless facade. Safe to call as a process-wide singleton."""
|
|
|
|
def __init__(self) -> None:
|
|
self._write_locks: dict[str, asyncio.Lock] = {}
|
|
|
|
# ── L1 ────────────────────────────────────────────────────────────────
|
|
|
|
async def emit(self, event: TraceEvent) -> None:
|
|
await trace.append(event)
|
|
|
|
# ── L2 / L3 read ──────────────────────────────────────────────────────
|
|
|
|
def read_doc(self, layer: Layer, key: str) -> Document:
|
|
path = self._path(layer, key)
|
|
if not path.exists():
|
|
return Document(title=_default_title(layer, key))
|
|
return parse(path.read_text(encoding="utf-8"))
|
|
|
|
def read_raw(self, layer: Layer, key: str) -> str:
|
|
path = self._path(layer, key)
|
|
if not path.exists():
|
|
return ""
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
def read_l3_concat(self) -> str:
|
|
"""Concatenate all four L3 docs for the ``read_memory`` tool."""
|
|
parts: list[str] = []
|
|
for slot in paths.L3_SLOTS:
|
|
body = self.read_raw("L3", slot).strip()
|
|
if body:
|
|
parts.append(body)
|
|
if not parts:
|
|
return _NO_MEMORY
|
|
return "\n\n---\n\n".join(parts) + "\n"
|
|
|
|
# ── L2 / L3 write (manual paths) ──────────────────────────────────────
|
|
|
|
async def overwrite_doc(self, layer: Layer, key: str, md: str) -> None:
|
|
"""Direct user-driven save from the workbench editor."""
|
|
path = self._path(layer, key)
|
|
async with self._lock_for(path):
|
|
await asyncio.to_thread(_atomic_write, path, md)
|
|
|
|
async def delete_entry(self, layer: Layer, key: str, entry_id: str) -> bool:
|
|
path = self._path(layer, key)
|
|
if not path.exists():
|
|
return False
|
|
async with self._lock_for(path):
|
|
doc = parse(path.read_text(encoding="utf-8"))
|
|
if not doc.remove(entry_id):
|
|
return False
|
|
await asyncio.to_thread(_atomic_write, path, serialize(doc))
|
|
return True
|
|
|
|
# ── L2 / L3 write (consolidator paths) ────────────────────────────────
|
|
|
|
async def update_l2(
|
|
self,
|
|
surface: Surface,
|
|
*,
|
|
language: str = "en",
|
|
user_label: str = "anonymous",
|
|
on_event: OnEvent | None = None,
|
|
apply_ops: bool = True,
|
|
) -> ConsolidateResult:
|
|
path = paths.l2_file(surface)
|
|
async with self._lock_for(path):
|
|
return await consolidator.consolidate_l2(
|
|
surface,
|
|
language=language,
|
|
user_label=user_label,
|
|
on_event=on_event,
|
|
apply_ops=apply_ops,
|
|
)
|
|
|
|
async def update_l3(
|
|
self,
|
|
slot: L3Slot,
|
|
*,
|
|
language: str = "en",
|
|
user_label: str = "anonymous",
|
|
on_event: OnEvent | None = None,
|
|
apply_ops: bool = True,
|
|
) -> ConsolidateResult:
|
|
if slot == "preferences":
|
|
raise ValueError("preferences.md is not auto-consolidated")
|
|
path = paths.l3_file(slot)
|
|
async with self._lock_for(path):
|
|
return await consolidator.consolidate_l3(
|
|
slot,
|
|
language=language,
|
|
user_label=user_label,
|
|
on_event=on_event,
|
|
apply_ops=apply_ops,
|
|
)
|
|
|
|
async def apply_ops_payload(
|
|
self, layer: Layer, key: str, ops_payload: list[dict]
|
|
) -> ApplyReport:
|
|
"""Apply a list of ops-as-JSON to a layer doc atomically.
|
|
|
|
Used by the workbench's preview → apply two-step flow. The
|
|
payload typically comes from a previous ``apply_ops=False``
|
|
consolidate call surfaced to the user for review.
|
|
"""
|
|
from deeptutor.services.memory.consolidator import _parse_ops_response
|
|
|
|
path = self._path(layer, key)
|
|
json_like = {"ops": ops_payload}
|
|
import json as _json
|
|
|
|
ops = _parse_ops_response(_json.dumps(json_like, ensure_ascii=False))
|
|
async with self._lock_for(path):
|
|
default_title = _default_title(layer, key)
|
|
doc = (
|
|
parse(path.read_text(encoding="utf-8"))
|
|
if path.exists()
|
|
else Document(title=default_title)
|
|
)
|
|
report = ops_apply(doc, ops)
|
|
if report.accepted and ops:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
await asyncio.to_thread(_atomic_write, path, serialize(doc))
|
|
return report
|
|
|
|
async def write_preference(
|
|
self,
|
|
*,
|
|
op: Literal["add", "edit"],
|
|
text: str,
|
|
target_id: str | None = None,
|
|
reason: str | None = None,
|
|
trace_id: str,
|
|
) -> ApplyReport:
|
|
"""Write the chat-mode preference signal. The ``write_memory`` tool
|
|
is the only caller; ``trace_id`` is the current chat turn's L1 id
|
|
injected by runtime."""
|
|
path = paths.l3_file("preferences")
|
|
async with self._lock_for(path):
|
|
doc = (
|
|
parse(path.read_text(encoding="utf-8"))
|
|
if path.exists()
|
|
else Document(title=_default_title("L3", "preferences"))
|
|
)
|
|
section = "Preferences"
|
|
if op == "add":
|
|
report = ops_apply(
|
|
doc,
|
|
[AddOp(section=section, text=text, refs=[trace_id])],
|
|
)
|
|
else:
|
|
if not target_id:
|
|
return ApplyReport(accepted=False, reason="edit requires target_id")
|
|
report = ops_apply(
|
|
doc,
|
|
[
|
|
EditOp(
|
|
target_id=target_id,
|
|
new_text=text,
|
|
new_refs=[trace_id],
|
|
)
|
|
],
|
|
)
|
|
if report.accepted:
|
|
await asyncio.to_thread(_atomic_write, path, serialize(doc))
|
|
if reason:
|
|
# Surface the reason in logs for workbench observability.
|
|
logger.info("write_memory %s id=%s reason=%s", op, target_id or "new", reason)
|
|
return report
|
|
|
|
# ── Workbench overview ────────────────────────────────────────────────
|
|
|
|
def overview(self) -> list[DocOverview]:
|
|
rows: list[DocOverview] = []
|
|
for surface in paths.SURFACES:
|
|
rows.append(self._overview_for("L2", surface))
|
|
for slot in paths.L3_SLOTS:
|
|
rows.append(self._overview_for("L3", slot))
|
|
return rows
|
|
|
|
def _overview_for(self, layer: Layer, key: str) -> DocOverview:
|
|
path = self._path(layer, key)
|
|
if not path.exists():
|
|
backlog = trace.count_since(key) if layer == "L2" else 0 # type: ignore[arg-type]
|
|
return DocOverview(
|
|
layer=layer,
|
|
key=key,
|
|
exists=False,
|
|
updated_at=None,
|
|
entry_count=0,
|
|
backlog=backlog,
|
|
)
|
|
|
|
stat = path.stat()
|
|
updated_at = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat()
|
|
try:
|
|
doc = parse(path.read_text(encoding="utf-8"))
|
|
entry_count = len(doc.all_entries())
|
|
except Exception:
|
|
entry_count = 0
|
|
|
|
backlog = 0
|
|
if layer == "L2":
|
|
try:
|
|
cutoff = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)
|
|
backlog = trace.count_since(key, since=cutoff) # type: ignore[arg-type]
|
|
except Exception:
|
|
backlog = 0
|
|
|
|
return DocOverview(
|
|
layer=layer,
|
|
key=key,
|
|
exists=True,
|
|
updated_at=updated_at,
|
|
entry_count=entry_count,
|
|
backlog=backlog,
|
|
)
|
|
|
|
# ── Internals ─────────────────────────────────────────────────────────
|
|
|
|
def _path(self, layer: Layer, key: str) -> Path:
|
|
if layer == "L2":
|
|
if key not in paths.SURFACES:
|
|
raise ValueError(f"unknown surface {key!r}")
|
|
return paths.l2_file(key) # type: ignore[arg-type]
|
|
if key not in paths.L3_SLOTS:
|
|
raise ValueError(f"unknown L3 slot {key!r}")
|
|
return paths.l3_file(key) # type: ignore[arg-type]
|
|
|
|
def _lock_for(self, path: Path) -> asyncio.Lock:
|
|
key = str(path)
|
|
lock = self._write_locks.get(key)
|
|
if lock is None:
|
|
lock = asyncio.Lock()
|
|
self._write_locks[key] = lock
|
|
return lock
|
|
|
|
|
|
# ── v1 → v2 startup migration ─────────────────────────────────────────────
|
|
|
|
|
|
def migrate_v1_if_needed() -> Path | None:
|
|
"""If any v1 memory files are present under the memory root, move the
|
|
whole memory directory's loose files into ``memory/backup/<ts>/``.
|
|
|
|
Idempotent: if there's nothing v1-shaped at the root, this is a no-op.
|
|
|
|
Returns the backup directory path on migration, or ``None`` otherwise.
|
|
"""
|
|
root = paths.memory_root()
|
|
if not root.exists():
|
|
return None
|
|
v1_present = [name for name in _V1_FILES if (root / name).exists()]
|
|
if not v1_present:
|
|
return None
|
|
|
|
ts = datetime.now(tz=timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
backup_dir = paths.backup_root() / ts
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
for item in list(root.iterdir()):
|
|
if item.name in {"trace", "L2", "L3", "backup"}:
|
|
continue
|
|
try:
|
|
shutil.move(str(item), str(backup_dir / item.name))
|
|
except OSError:
|
|
logger.warning("v1 memory migration: failed to move %s", item, exc_info=True)
|
|
logger.info("v1 memory migrated to %s", backup_dir)
|
|
return backup_dir
|
|
|
|
|
|
def migrate_partner_surface_if_needed() -> bool:
|
|
"""Rename the legacy ``tutorbot`` memory surface to ``partner``.
|
|
|
|
The partner surface key used to be ``tutorbot`` — so footnote refs read
|
|
``tutorbot:<id>`` and the consolidator even wrote "tutorbot" into L2
|
|
prose. It is now ``partner``. This moves any on-disk artifacts (L2 doc +
|
|
meta, snapshot dir, trace dir) to the new name, rewrites the ``tutorbot``
|
|
token to ``partner`` inside the L2 doc/meta (both the ``tutorbot:`` ref
|
|
prefix and the bare prose word), and renames the per-surface key inside
|
|
every L3 meta.
|
|
|
|
Idempotent: skips any target that already exists; a no-op when nothing
|
|
tutorbot-shaped lives under the memory root.
|
|
"""
|
|
import json
|
|
import re
|
|
|
|
root = paths.memory_root()
|
|
if not root.exists():
|
|
return False
|
|
|
|
moved = False
|
|
|
|
l2 = paths.l2_dir()
|
|
old_md, new_md = l2 / "tutorbot.md", l2 / "partner.md"
|
|
if old_md.exists() and not new_md.exists():
|
|
text = old_md.read_text(encoding="utf-8")
|
|
text = text.replace("tutorbot:", "partner:") # footnote/inline refs
|
|
text = re.sub(r"\btutorbot\b", "partner", text) # bare prose word
|
|
text = re.sub(r"\bTutorbot\b", "Partner", text)
|
|
new_md.write_text(text, encoding="utf-8")
|
|
old_md.unlink()
|
|
moved = True
|
|
old_meta, new_meta = l2 / "tutorbot.meta.json", l2 / "partner.meta.json"
|
|
if old_meta.exists() and not new_meta.exists():
|
|
text = old_meta.read_text(encoding="utf-8").replace("tutorbot:", "partner:")
|
|
new_meta.write_text(text, encoding="utf-8")
|
|
old_meta.unlink()
|
|
moved = True
|
|
|
|
# snapshot/<surface>/ and trace/<surface>/ — plain directory moves
|
|
# (entity ids carry no surface prefix, so no content rewrite needed).
|
|
for sub in ("snapshot", "trace"):
|
|
old_dir, new_dir = root / sub / "tutorbot", root / sub / "partner"
|
|
if old_dir.is_dir() and not new_dir.exists():
|
|
shutil.move(str(old_dir), str(new_dir))
|
|
moved = True
|
|
|
|
# L3 metas track seen L2 entry ids per surface — rename that key.
|
|
l3 = paths.l3_dir()
|
|
if l3.is_dir():
|
|
for meta_path in l3.glob("*.meta.json"):
|
|
try:
|
|
data = json.loads(meta_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
continue
|
|
seen = data.get("seen_l2_entry_ids")
|
|
if isinstance(seen, dict) and "tutorbot" in seen and "partner" not in seen:
|
|
seen["partner"] = seen.pop("tutorbot")
|
|
meta_path.write_text(
|
|
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
moved = True
|
|
|
|
if moved:
|
|
logger.info("migrated legacy 'tutorbot' memory surface to 'partner'")
|
|
return moved
|
|
|
|
|
|
# ── Singleton accessor ────────────────────────────────────────────────────
|
|
|
|
|
|
_singleton: MemoryStore | None = None
|
|
|
|
|
|
def get_memory_store() -> MemoryStore:
|
|
global _singleton
|
|
if _singleton is None:
|
|
_singleton = MemoryStore()
|
|
return _singleton
|
|
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _default_title(layer: Layer, key: str) -> str:
|
|
if layer == "L2":
|
|
return f"{key} memory"
|
|
return {
|
|
"recent": "Recent summary",
|
|
"profile": "User profile",
|
|
"scope": "Knowledge scope",
|
|
"preferences": "Preferences",
|
|
}.get(key, key)
|
|
|
|
|
|
def _atomic_write(path: Path, content: str) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
tmp.write_text(content, encoding="utf-8")
|
|
tmp.replace(path)
|