From e102b2118073c7051d8449568d28c5c568cbf200 Mon Sep 17 00:00:00 2001 From: Deshraj Yadav Date: Tue, 28 Jul 2026 01:45:10 -0700 Subject: [PATCH] feat(mem0-agent): WS0+WS1 foundation - verified contract, config, client core Encodes the platform contract validated against the live API: - project_id/org_id pinned in request body (query params are ignored) - latest_only=True enforced on every read - metadata.type as the read-time taxonomy (categorization lags ~4h) - NOT filters take a list; delete_all takes query params; fields repeated Includes project config (write-gate instructions, 6 custom categories, decay), verified filter recipes, circuit breaker, session-scoped state, keychain auth, and identity/project resolution. Co-Authored-By: Claude Fable 5 --- integrations/mem0-agent/pyproject.toml | 29 +++ .../mem0-agent/src/mem0_agent/__init__.py | 6 + integrations/mem0-agent/src/mem0_agent/api.py | 174 +++++++++++++ .../mem0-agent/src/mem0_agent/breaker.py | 71 ++++++ .../src/mem0_agent/config/__init__.py | 23 ++ .../src/mem0_agent/config/filters.py | 102 ++++++++ .../src/mem0_agent/config/project_config.py | 138 ++++++++++ integrations/mem0-agent/src/mem0_agent/ctx.py | 85 ++++++ .../mem0-agent/src/mem0_agent/settings.py | 241 ++++++++++++++++++ 9 files changed, 869 insertions(+) create mode 100644 integrations/mem0-agent/pyproject.toml create mode 100644 integrations/mem0-agent/src/mem0_agent/__init__.py create mode 100644 integrations/mem0-agent/src/mem0_agent/api.py create mode 100644 integrations/mem0-agent/src/mem0_agent/breaker.py create mode 100644 integrations/mem0-agent/src/mem0_agent/config/__init__.py create mode 100644 integrations/mem0-agent/src/mem0_agent/config/filters.py create mode 100644 integrations/mem0-agent/src/mem0_agent/config/project_config.py create mode 100644 integrations/mem0-agent/src/mem0_agent/ctx.py create mode 100644 integrations/mem0-agent/src/mem0_agent/settings.py diff --git a/integrations/mem0-agent/pyproject.toml b/integrations/mem0-agent/pyproject.toml new file mode 100644 index 000000000..d2e0291d7 --- /dev/null +++ b/integrations/mem0-agent/pyproject.toml @@ -0,0 +1,29 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "mem0-agent" +version = "0.1.0" +description = "Coding-agent memory for Claude Code and friends, built on the Mem0 platform" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "Apache-2.0" } +dependencies = [] + +[project.optional-dependencies] +keyring = ["keyring>=24"] +dev = ["pytest>=8", "keyring>=24"] + +[project.scripts] +mem0-agent = "mem0_agent.cli:main" + +[tool.hatch.build.targets.wheel] +packages = ["src/mem0_agent"] + +[tool.ruff] +line-length = 120 +target-version = "py310" + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/integrations/mem0-agent/src/mem0_agent/__init__.py b/integrations/mem0-agent/src/mem0_agent/__init__.py new file mode 100644 index 000000000..1ee7cd3a9 --- /dev/null +++ b/integrations/mem0-agent/src/mem0_agent/__init__.py @@ -0,0 +1,6 @@ +"""mem0-agent: coding-agent memory built on existing Mem0 platform APIs.""" + +from .config.project_config import DURABLE_TYPES, POLICY_VERSION, TYPES + +__all__ = ["TYPES", "DURABLE_TYPES", "POLICY_VERSION", "__version__"] +__version__ = "0.1.0" diff --git a/integrations/mem0-agent/src/mem0_agent/api.py b/integrations/mem0-agent/src/mem0_agent/api.py new file mode 100644 index 000000000..33437736c --- /dev/null +++ b/integrations/mem0-agent/src/mem0_agent/api.py @@ -0,0 +1,174 @@ +"""REST client that MECHANICALLY ENFORCES the verified platform contract. + +Four rules no caller can forget, because the wrapper applies them: + +1. project_id + org_id go in the request BODY. As query params they are silently + ignored and the call lands in whatever project the API key defaults to -- the + most likely cause of v1's benchmark data polluting the production project. +2. Every read carries latest_only=True, or superseded memories come back beside + the memories that replaced them. +3. DELETE /v1/memories/ takes QUERY params, not a body. +4. project.get `fields` must be repeated params, not comma-joined. + +Everything fails open: hooks must never block a developer's session. +""" + +from __future__ import annotations + +import json +import time +import urllib.error +import urllib.parse +import urllib.request +from typing import Any + +from .breaker import Breaker + +DEFAULT_BASE = "https://api.mem0.ai" +READ_TIMEOUT = 8.0 +WRITE_TIMEOUT = 15.0 + + +class ContractError(RuntimeError): + """Raised when an internal caller bypasses an enforced rule (tests use this).""" + + +class Api: + def __init__(self, api_key: str, org_id: str | None = None, project_id: str | None = None, + *, base: str = DEFAULT_BASE, breaker: Breaker | None = None, + strict: bool = False, opener=None): + self.api_key = api_key + self.org_id = org_id + self.project_id = project_id + self.base = base.rstrip("/") + self.breaker = breaker or Breaker() + self.strict = strict + self._opener = opener or urllib.request.urlopen + self.last_error: str | None = None + + # ---------- plumbing ---------- + def _pin(self, project_id: str | None = None) -> dict[str, str]: + pid = project_id or self.project_id + oid = self.org_id + if not pid or not oid: + if self.strict: + raise ContractError("project_id and org_id must be set; unpinned calls leak into the key's default project") + return {} + return {"project_id": pid, "org_id": oid} + + def call(self, method: str, path: str, body: dict | None = None, + params: dict | None = None, timeout: float = READ_TIMEOUT) -> tuple[int, Any]: + """Returns (status, parsed_body). Never raises on HTTP or network error.""" + if not self.breaker.allow(): + return 0, {"error": "circuit open"} + url = self.base + path + if params: + url += "?" + urllib.parse.urlencode(params, doseq=True) + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request(url, data=data, method=method) + req.add_header("Authorization", f"Token {self.api_key}") + req.add_header("Content-Type", "application/json") + try: + with self._opener(req, timeout=timeout) as r: + raw = r.read().decode() + self.breaker.record_success() + self.last_error = None + return r.status, (json.loads(raw) if raw else None) + except urllib.error.HTTPError as e: + raw = e.read().decode() if hasattr(e, "read") else "" + # 4xx is a contract problem, not an availability problem: don't trip the breaker. + if e.code >= 500: + self.breaker.record_failure() + self.last_error = f"HTTP {e.code}" + try: + return e.code, json.loads(raw) + except Exception: + return e.code, {"error": raw[:400]} + except Exception as e: # timeouts, DNS, connection reset + self.breaker.record_failure() + self.last_error = str(e)[:200] + return 0, {"error": self.last_error} + + # ---------- identity ---------- + def ping(self) -> tuple[int, Any]: + """Returns org_id, project_id and user_email -- the stable cross-machine identity.""" + return self.call("GET", "/v1/ping/") + + # ---------- memory: writes ---------- + def add(self, messages: list[dict], *, project_id: str | None = None, **kw) -> tuple[int, Any]: + """Fire-and-forget. With infer=True the response is only {event_id, status:PENDING}; + extraction lands 20s-5min later, so never read back within a session.""" + body = {"messages": messages, **self._pin(project_id), **kw} + return self.call("POST", "/v3/memories/add/", body, timeout=WRITE_TIMEOUT) + + def update(self, memory_id: str, *, project_id: str | None = None, **kw) -> tuple[int, Any]: + body = {**self._pin(project_id), **kw} + return self.call("PUT", f"/v1/memories/{urllib.parse.quote(memory_id)}/", body, + timeout=WRITE_TIMEOUT) + + def delete(self, memory_id: str, *, project_id: str | None = None) -> tuple[int, Any]: + return self.call("DELETE", f"/v1/memories/{urllib.parse.quote(memory_id)}/", + self._pin(project_id) or None, timeout=WRITE_TIMEOUT) + + def delete_all(self, *, project_id: str | None = None, **entity) -> tuple[int, Any]: + """Rule 3: this endpoint reads QUERY params; a body yields 400.""" + return self.call("DELETE", "/v1/memories/", None, + params={**self._pin(project_id), **entity}, timeout=WRITE_TIMEOUT) + + def feedback(self, memory_id: str, feedback: str, reason: str | None = None, + *, project_id: str | None = None) -> tuple[int, Any]: + """404s without the project pin.""" + body = {"memory_id": memory_id, "feedback": feedback, + "feedback_reason": reason, **self._pin(project_id)} + return self.call("POST", "/v1/feedback/", body, timeout=WRITE_TIMEOUT) + + # ---------- memory: reads (latest_only enforced) ---------- + def get_all(self, filters: dict, *, page: int = 1, page_size: int = 50, + project_id: str | None = None, latest_only: bool = True, **kw) -> tuple[int, Any]: + if latest_only is not True and self.strict and not kw.pop("_allow_superseded", False): + raise ContractError("reads must use latest_only=True unless explicitly auditing superseded records") + body = {"filters": filters, "latest_only": latest_only, **self._pin(project_id), **kw} + return self.call("POST", "/v3/memories/", body, + params={"page": page, "page_size": page_size}) + + def search(self, query: str, filters: dict, *, project_id: str | None = None, + latest_only: bool = True, **kw) -> tuple[int, Any]: + if latest_only is not True and self.strict and not kw.pop("_allow_superseded", False): + raise ContractError("reads must use latest_only=True unless explicitly auditing superseded records") + body = {"query": query, "filters": filters, "latest_only": latest_only, + **self._pin(project_id), **kw} + return self.call("POST", "/v3/memories/search/", body) + + def get_one(self, memory_id: str) -> tuple[int, Any]: + """Fetch by ID ignores expiration -- expired memories are still returned here.""" + return self.call("GET", f"/v1/memories/{urllib.parse.quote(memory_id)}/") + + # ---------- project config ---------- + def project_get(self, *, project_id: str | None = None, + fields: list[str] | None = None) -> tuple[int, Any]: + pid = project_id or self.project_id + params = {"fields": fields} if fields else None # rule 4: repeated params + return self.call("GET", f"/api/v1/orgs/organizations/{self.org_id}/projects/{pid}/", + None, params=params) + + def project_update(self, *, project_id: str | None = None, **kw) -> tuple[int, Any]: + pid = project_id or self.project_id + return self.call("PATCH", f"/api/v1/orgs/organizations/{self.org_id}/projects/{pid}/", + kw, timeout=WRITE_TIMEOUT) + + def project_create(self, name: str, description: str = "") -> tuple[int, Any]: + return self.call("POST", f"/api/v1/orgs/organizations/{self.org_id}/projects/", + {"name": name, "description": description}, timeout=WRITE_TIMEOUT) + + +def results_of(body: Any) -> list[dict]: + """Normalize the paginated / bare-list shapes the API returns.""" + if isinstance(body, dict): + got = body.get("results") + return got if isinstance(got, list) else [] + return body if isinstance(body, list) else [] + + +def expiry_date(days: int, now: float | None = None) -> str: + """YYYY-MM-DD in UTC, the only format the platform accepts.""" + return time.strftime("%Y-%m-%d", time.gmtime((now or time.time()) + days * 86400)) diff --git a/integrations/mem0-agent/src/mem0_agent/breaker.py b/integrations/mem0-agent/src/mem0_agent/breaker.py new file mode 100644 index 000000000..935e26b6f --- /dev/null +++ b/integrations/mem0-agent/src/mem0_agent/breaker.py @@ -0,0 +1,71 @@ +"""Circuit breaker. When the API is unhealthy the plugin must get out of the way +fast and say so once -- v1 silently burned a full timeout on every prompt instead. +""" + +from __future__ import annotations + +import json +import time +from pathlib import Path + +FAILURE_THRESHOLD = 3 +COOLDOWN_SECONDS = 600 + + +class Breaker: + def __init__(self, path: Path | None = None, *, threshold: int = FAILURE_THRESHOLD, + cooldown: int = COOLDOWN_SECONDS, clock=time.time): + self.path = path + self.threshold = threshold + self.cooldown = cooldown + self._clock = clock + self._state = {"failures": 0, "open_until": 0.0, "notified": False} + self._load() + + # --- persistence (best effort; never raises) --- + def _load(self) -> None: + if not self.path or not self.path.exists(): + return + try: + self._state.update(json.loads(self.path.read_text())) + except Exception: + pass + + def _save(self) -> None: + if not self.path: + return + try: + self.path.parent.mkdir(parents=True, exist_ok=True) + tmp = self.path.with_suffix(".tmp") + tmp.write_text(json.dumps(self._state)) + tmp.replace(self.path) + except Exception: + pass + + # --- API --- + def allow(self) -> bool: + return self._clock() >= float(self._state.get("open_until", 0)) + + @property + def is_open(self) -> bool: + return not self.allow() + + def record_success(self) -> None: + if self._state["failures"] or self._state["open_until"]: + self._state = {"failures": 0, "open_until": 0.0, "notified": False} + self._save() + + def record_failure(self) -> None: + self._state["failures"] = int(self._state.get("failures", 0)) + 1 + if self._state["failures"] >= self.threshold: + self._state["open_until"] = self._clock() + self.cooldown + self._save() + + def take_notice(self) -> str | None: + """Returns a user-facing message exactly once per open period.""" + if self.is_open and not self._state.get("notified"): + self._state["notified"] = True + self._save() + mins = max(1, int((float(self._state["open_until"]) - self._clock()) / 60)) + return f"mem0 is unreachable; memory paused for ~{mins} min (your session is unaffected)" + return None diff --git a/integrations/mem0-agent/src/mem0_agent/config/__init__.py b/integrations/mem0-agent/src/mem0_agent/config/__init__.py new file mode 100644 index 000000000..8c667b77a --- /dev/null +++ b/integrations/mem0-agent/src/mem0_agent/config/__init__.py @@ -0,0 +1,23 @@ +from . import filters +from .project_config import ( + CATEGORIES, + DURABLE_TYPES, + INSTRUCTIONS, + POLICY_VERSION, + SESSION_STATE_TTL_DAYS, + TYPES, + USER_SCOPED_TYPES, + apply_project_config, +) + +__all__ = [ + "filters", + "CATEGORIES", + "INSTRUCTIONS", + "TYPES", + "DURABLE_TYPES", + "USER_SCOPED_TYPES", + "POLICY_VERSION", + "SESSION_STATE_TTL_DAYS", + "apply_project_config", +] diff --git a/integrations/mem0-agent/src/mem0_agent/config/filters.py b/integrations/mem0-agent/src/mem0_agent/config/filters.py new file mode 100644 index 000000000..87eb4b2ae --- /dev/null +++ b/integrations/mem0-agent/src/mem0_agent/config/filters.py @@ -0,0 +1,102 @@ +"""Filter recipes, every one verified against the live v2 API. + +Rules encoded here (each learned the hard way -- see docs/CONTRACT.md): + +* Implicit null scoping does NOT work. `{"user_id": u}` alone also returns + project-scoped records, so user-scope reads need an explicit NOT clause. +* `NOT` takes a LIST, not an object. The object form returns HTTP 400. +* Type is matched on `metadata.type` (available immediately) OR on `categories` + (assigned by a background job ~4h later). Reads union both so fresh and old + memories are equally retrievable. +* Metadata filters support only equality/contains/ne -- multi-value needs OR. +""" + +from __future__ import annotations + +from typing import Any + +from .project_config import DURABLE_TYPES + +Filter = dict[str, Any] + + +def _type_clauses(types: tuple[str, ...] | list[str]) -> list[Filter]: + """Match a set of types by metadata (immediate) or categories (eventual).""" + clauses: list[Filter] = [{"metadata": {"type": t}} for t in types] + clauses.append({"categories": {"in": list(types)}}) + return clauses + + +def _null_app() -> Filter: + """Records with no app_id, i.e. user-scoped ones.""" + return {"NOT": [{"app_id": "*"}]} + + +def context_pack(user_id: str, app_id: str, types: tuple[str, ...] = DURABLE_TYPES) -> Filter: + """Everything the session-start pack needs, in ONE call (~310ms measured). + + Spans both scopes: project-scoped records for this repo, plus the user's + global preferences which carry no app_id. + """ + return { + "AND": [ + {"user_id": user_id}, + {"OR": [{"app_id": app_id}, _null_app()]}, + {"OR": _type_clauses(types)}, + ] + } + + +def user_prefs(user_id: str) -> Filter: + """User-scope only. Without the NOT clause this also returns every project record.""" + return {"AND": [{"user_id": user_id}, _null_app()]} + + +def project_scope(user_id: str, app_id: str, types: tuple[str, ...] | None = None) -> Filter: + f: list[Filter] = [{"user_id": user_id}, {"app_id": app_id}] + if types: + f.append({"OR": _type_clauses(types)}) + return {"AND": f} + + +def session_state(user_id: str, app_id: str, session_id: str | None = None) -> Filter: + """The open-thread record. One per session, found by metadata.""" + f: list[Filter] = [ + {"user_id": user_id}, + {"app_id": app_id}, + {"metadata": {"type": "session_state"}}, + ] + if session_id: + f.append({"metadata": {"session_id": session_id}}) + return {"AND": f} + + +def error_assist(user_id: str, app_id: str) -> Filter: + """Past gotchas and procedures -- the only semantic search on the hot path.""" + return { + "AND": [ + {"user_id": user_id}, + {"app_id": app_id}, + {"OR": _type_clauses(("insight", "runbook"))}, + ] + } + + +def by_session(user_id: str, app_id: str, session_id: str) -> Filter: + return { + "AND": [ + {"user_id": user_id}, + {"app_id": app_id}, + {"metadata": {"session_id": session_id}}, + ] + } + + +def all_in_scope(user_id: str, app_id: str) -> Filter: + """Maintenance / stats: everything for this user+project.""" + return {"AND": [{"user_id": user_id}, {"app_id": app_id}]} + + +def team_scope(app_id: str) -> Filter: + """Planned for the team fast-follow; not wired into v1 of the client.""" + return {"AND": [{"app_id": app_id}, {"user_id": "*"}]} diff --git a/integrations/mem0-agent/src/mem0_agent/config/project_config.py b/integrations/mem0-agent/src/mem0_agent/config/project_config.py new file mode 100644 index 000000000..05822d4f4 --- /dev/null +++ b/integrations/mem0-agent/src/mem0_agent/config/project_config.py @@ -0,0 +1,138 @@ +"""Project-level configuration: the write gate, the taxonomy, and the lifecycle toggle. + +This module is the single source of truth for how the platform is configured. The +custom instructions ARE the write gate -- they were validated against the real +polluted v1 corpus (see eval/fixtures.py) and suppress every mechanical-noise class. + +Bump POLICY_VERSION whenever INSTRUCTIONS or CATEGORIES change; it is stamped onto +every memory's metadata so quality regressions can be traced to a policy revision. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +POLICY_VERSION = "v2.0" + +# The six memory types. This enum is authoritative for both metadata.type (used at +# read time, available immediately) and the platform's custom categories (assigned +# by the classifier hours later, used as a secondary signal). +TYPES: tuple[str, ...] = ( + "preference", + "decision", + "convention", + "insight", + "runbook", + "session_state", +) + +# Types that make up the durable knowledge base. session_state is excluded: it is +# session-scoped, short-lived, and retrieved by its own recipe. +DURABLE_TYPES: tuple[str, ...] = ( + "preference", + "decision", + "convention", + "insight", + "runbook", +) + +# Types stored at user scope (null app_id) rather than project scope. +USER_SCOPED_TYPES: frozenset[str] = frozenset({"preference"}) + +# Days until a session_state record stops surfacing. Expiration hides, never deletes. +SESSION_STATE_TTL_DAYS = 14 + + +INSTRUCTIONS = """Your Task: Extract durable knowledge from a developer's session with a coding assistant. +A fact qualifies ONLY if it would change how an assistant behaves in a future session. + +Information to Extract: +1. preference - how the developer wants work done: style, workflow, tools, communication, review habits. +2. decision - a resolved technical choice AND the reasoning behind it. +3. convention - a project or team rule that is not written in the repository docs. +4. insight - a root-caused gotcha, constraint, or non-obvious behavior of the system. +5. runbook - a multi-step procedure that was verified to work end to end. + +Guidelines: +- One self-contained fact per memory, understandable without the conversation. +- State the general lesson, not the incident that revealed it. +- Attribute correctly: the assistant's own observations are NOT the user's preferences. +- Include the reasoning for decisions when it was stated. + +Exclude (never store): +- Progress updates, status heartbeats, ETAs, percentages, epochs, task notifications, monitoring output, training or job metrics +- Anything the assistant merely did, said, asked, or planned in the middle of a task +- Lists of files modified, commits made, or PRs opened, and other activity derivable from git +- Contents of repository files such as CLAUDE.md, README, or configs +- Anything true only for the current session or the current run +- One-off instructions that apply only to the task at hand ("you do it", "run it yourself this time", "skip tests for now"); store a preference only when it is stated as a general or recurring rule +- Secrets, API keys, tokens, credentials, connection strings +""" + + +CATEGORIES: list[dict[str, str]] = [ + {"preference": "How this developer wants work done: coding style, workflow, tooling, communication and review habits"}, + {"decision": "A resolved technical choice and the reasoning behind it; superseded when the choice is reversed"}, + {"convention": "A project or team rule that is not documented in the repository itself"}, + {"insight": "A root-caused gotcha, constraint, or non-obvious behavior of the system or its tooling"}, + {"runbook": "A verified multi-step procedure such as deploying, debugging, setting up, or releasing"}, + {"session_state": "Open-thread snapshot for a single session: current goal, status, blockers, next step"}, +] + + +@dataclass +class ConfigReport: + """Outcome of applying project configuration, with a verified round-trip.""" + + applied: list[str] = field(default_factory=list) + failed: list[tuple[str, Any]] = field(default_factory=list) + verified: dict[str, Any] = field(default_factory=dict) + + @property + def ok(self) -> bool: + if self.failed: + return False + cats = self.verified.get("custom_categories") or [] + names = [next(iter(c)) for c in cats if isinstance(c, dict)] + return ( + bool(self.verified.get("custom_instructions")) + and set(names) == set(TYPES) + and self.verified.get("decay") is True + ) + + def summary(self) -> str: + state = "ok" if self.ok else "incomplete" + return ( + f"project config {state}: applied={','.join(self.applied) or 'none'}" + + (f" failed={self.failed}" if self.failed else "") + ) + + +def apply_project_config(api, *, project_id: str | None = None) -> ConfigReport: + """Idempotently push instructions, categories and decay, then verify by reading back. + + Safe to call on every onboard. `api` is a mem0_agent.api.Api instance. + """ + report = ConfigReport() + updates = { + "custom_instructions": {"custom_instructions": INSTRUCTIONS}, + "custom_categories": {"custom_categories": CATEGORIES}, + "decay": {"decay": True}, + } + for name, payload in updates.items(): + status, body = api.project_update(project_id=project_id, **payload) + if status == 200: + report.applied.append(name) + else: + report.failed.append((name, body)) + + # Verified behavior: `fields` must be sent as repeated query params, not comma-joined. + status, body = api.project_get( + project_id=project_id, fields=["custom_instructions", "custom_categories", "decay"] + ) + if status == 200 and isinstance(body, dict): + report.verified = body + else: + report.failed.append(("verify", body)) + return report diff --git a/integrations/mem0-agent/src/mem0_agent/ctx.py b/integrations/mem0-agent/src/mem0_agent/ctx.py new file mode 100644 index 000000000..85dc8eb3a --- /dev/null +++ b/integrations/mem0-agent/src/mem0_agent/ctx.py @@ -0,0 +1,85 @@ +"""One object every hook builds: credentials, identity, scope, API, state.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Optional + +from .api import Api +from .breaker import Breaker +from .settings import ( + SessionState, + Settings, + get_api_key, + resolve_app_id, + resolve_branch, + resolve_user_id, +) + + +@dataclass +class Ctx: + api: Optional[Api] + settings: Settings + state: SessionState + user_id: str + app_id: str + session_id: str + branch: str | None + ready: bool + reason: str = "" + + @property + def editor(self) -> str: + return os.environ.get("MEM0_EDITOR", "claude-code") + + def provenance(self, mtype: str) -> dict: + """Metadata stamped on every write. `type` is what reads filter on -- platform + categories arrive hours later and cannot be relied on at read time.""" + from .config.project_config import POLICY_VERSION + + meta = { + "type": mtype, + "session_id": self.session_id, + "editor": self.editor, + "policy": POLICY_VERSION, + } + if self.branch: + meta["branch"] = self.branch + return meta + + def log(self, event: str, **fields) -> None: + self.state.append("events.jsonl", {"event": event, **fields}) + + +def build(session_id: str | None = None, cwd: str | None = None, *, strict: bool = False) -> Ctx: + """Never raises. If anything is missing, returns ready=False and the caller no-ops.""" + session_id = session_id or os.environ.get("MEM0_SESSION_ID") or "no-session" + settings = Settings.load() + state = SessionState(session_id) + key = get_api_key() + if not key: + return Ctx(None, settings, state, "", "", session_id, None, False, "no API key") + + org = settings.get("org_id") or os.environ.get("MEM0_ORG_ID") + project = settings.get("memory_project_id") or settings.get("default_project_id") \ + or os.environ.get("MEM0_PLATFORM_PROJECT_ID") + api = Api(key, org_id=org, project_id=project, + breaker=Breaker(state.breaker_path), strict=strict) + + # First run: learn identity and the key's home project from the API itself. + if not org or not project: + status, body = api.ping() + if status == 200 and isinstance(body, dict): + api.org_id = org = body.get("org_id") + api.project_id = project = settings.get("memory_project_id") or body.get("project_id") + settings.set("org_id", org) + settings.set("default_project_id", body.get("project_id")) + else: + return Ctx(None, settings, state, "", "", session_id, None, False, "identity unavailable") + + user_id = resolve_user_id(api, settings) + app_id = resolve_app_id(cwd, settings) + branch = resolve_branch(cwd) + return Ctx(api, settings, state, user_id, app_id, session_id, branch, True) diff --git a/integrations/mem0-agent/src/mem0_agent/settings.py b/integrations/mem0-agent/src/mem0_agent/settings.py new file mode 100644 index 000000000..9678ff0a8 --- /dev/null +++ b/integrations/mem0-agent/src/mem0_agent/settings.py @@ -0,0 +1,241 @@ +"""Settings, identity, project resolution and per-session state. + +Two deliberate departures from v1: + +* State lives under ~/.mem0/v2/sessions//, never in /tmp keyed by + $USER. v1's counters and stats collided between concurrent sessions, so nudges + fired at the wrong time and one session could disarm another's safety net. +* The API key is read from the OS keychain or the environment. v1 grepped shell + rc files for it and re-exported it in plaintext. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import subprocess +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +HOME = Path(os.path.expanduser("~")) / ".mem0" / "v2" +SETTINGS_PATH = HOME / "settings.json" +KEYRING_SERVICE = "mem0-agent" + +# Capture: how eager the trigger detector is. Retrieval: how much gets injected. +CAPTURE_LEVELS = ("conservative", "balanced", "aggressive") +RETRIEVAL_LEVELS = ("conservative", "balanced", "aggressive") +MEMORY_MODES = ("dual", "full") + +RETRIEVAL_BUDGETS = {"conservative": 600, "balanced": 1500, "aggressive": 2500} +ERROR_ASSIST_THRESHOLD = {"conservative": None, "balanced": 0.55, "aggressive": 0.35} + +DEFAULTS: dict[str, Any] = { + "capture": "balanced", + "retrieval": "balanced", + "memory_mode": "dual", + "telemetry": False, + "projects": {}, +} + + +def _read_json(path: Path) -> dict: + try: + return json.loads(path.read_text()) + except Exception: + return {} + + +def _write_json(path: Path, data: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(data, indent=2)) + tmp.replace(path) + + +@dataclass +class Settings: + data: dict = field(default_factory=lambda: dict(DEFAULTS)) + path: Path = SETTINGS_PATH + + @classmethod + def load(cls, path: Path = SETTINGS_PATH) -> "Settings": + merged = dict(DEFAULTS) + merged.update(_read_json(path)) + return cls(data=merged, path=path) + + def save(self) -> None: + _write_json(self.path, self.data) + + def get(self, key: str, default: Any = None) -> Any: + return self.data.get(key, default) + + def set(self, key: str, value: Any) -> None: + self.data[key] = value + self.save() + + # -- per-project overrides (memory mode is chosen per repo at onboard) -- + def project_setting(self, app_id: str, key: str, default: Any = None) -> Any: + return (self.data.get("projects", {}).get(app_id, {}) or {}).get(key, self.data.get(key, default)) + + def set_project_setting(self, app_id: str, key: str, value: Any) -> None: + self.data.setdefault("projects", {}).setdefault(app_id, {})[key] = value + self.save() + + @property + def retrieval_budget(self) -> int: + return RETRIEVAL_BUDGETS.get(self.get("retrieval", "balanced"), 1500) + + @property + def error_assist_threshold(self) -> float | None: + return ERROR_ASSIST_THRESHOLD.get(self.get("retrieval", "balanced"), 0.55) + + +# --------------------------- credentials --------------------------- +def get_api_key() -> str | None: + """Env first (explicit beats implicit), then the OS keychain. Never rc files.""" + key = os.environ.get("MEM0_API_KEY") + if key: + return key.strip() + try: + import keyring # optional dependency + + return keyring.get_password(KEYRING_SERVICE, "api_key") + except Exception: + return None + + +def store_api_key(key: str) -> bool: + try: + import keyring + + keyring.set_password(KEYRING_SERVICE, "api_key", key) + return True + except Exception: + return False + + +# --------------------------- identity --------------------------- +def resolve_user_id(api, settings: Settings | None = None) -> str: + """Stable across machines: the mem0 account behind the key (verified via /v1/ping/). + + Order: explicit override -> cached -> account email local-part -> $USER. + """ + override = os.environ.get("MEM0_USER_ID") + if override: + return override.strip() + settings = settings or Settings.load() + cached = settings.get("user_id") + if cached: + return cached + status, body = api.ping() + if status == 200 and isinstance(body, dict): + email = body.get("user_email") or "" + uid = email.split("@")[0] if email else "" + if uid: + settings.set("user_id", uid) + if body.get("org_id"): + settings.set("org_id", body["org_id"]) + if body.get("project_id"): + settings.set("default_project_id", body["project_id"]) + return uid + return os.environ.get("USER", "default") + + +# --------------------------- project scope --------------------------- +def _git(args: list[str], cwd: str | None = None) -> str | None: + try: + out = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, timeout=3) + return out.stdout.strip() if out.returncode == 0 else None + except Exception: + return None + + +def _slug_from_remote(url: str) -> str | None: + u = url.strip().removesuffix(".git") + for sep in ("://", "@"): + if sep in u: + u = u.split(sep, 1)[1] + u = u.replace(":", "/") + parts = [p for p in u.split("/") if p] + if len(parts) >= 2: + return f"{parts[-2]}-{parts[-1]}".lower() + return None + + +def resolve_app_id(cwd: str | None = None, settings: Settings | None = None) -> str: + """Repo identity, stable across clones, worktrees and folder moves. + + env override -> cwd map -> remote-hash map (self-healing after a move) + -> owner-repo slug -> directory name. + """ + override = os.environ.get("MEM0_PROJECT_ID") + if override: + return override.strip() + cwd = cwd or os.getcwd() + settings = settings or Settings.load() + pmap = settings.get("project_map", {}) or {} + + if cwd in pmap: + return pmap[cwd] + + remote = _git(["config", "--get", "remote.origin.url"], cwd) + if remote: + rkey = "remote:" + hashlib.sha256(remote.encode()).hexdigest()[:16] + if rkey in pmap: # folder moved; heal the cwd entry + app = pmap[rkey] + pmap[cwd] = app + settings.set("project_map", pmap) + return app + slug = _slug_from_remote(remote) + if slug: + pmap[cwd] = slug + pmap[rkey] = slug + settings.set("project_map", pmap) + return slug + + toplevel = _git(["rev-parse", "--show-toplevel"], cwd) or cwd + return Path(toplevel).name.lower() + + +def resolve_branch(cwd: str | None = None) -> str | None: + return _git(["branch", "--show-current"], cwd or os.getcwd()) or None + + +# --------------------------- session state --------------------------- +class SessionState: + """Per-session scratch dir. Keyed by session_id so concurrent sessions never collide.""" + + def __init__(self, session_id: str, root: Path = HOME / "sessions"): + self.session_id = session_id or "unknown" + self.dir = root / self.session_id + self.dir.mkdir(parents=True, exist_ok=True) + + def _p(self, name: str) -> Path: + return self.dir / name + + def read(self, name: str, default: Any = None) -> Any: + data = _read_json(self._p(name)) + return data if data else (default if default is not None else {}) + + def write(self, name: str, data: Any) -> None: + _write_json(self._p(name), data) + + def append(self, name: str, record: dict) -> None: + try: + with self._p(name).open("a") as fh: + fh.write(json.dumps(record) + "\n") + except Exception: + pass + + def read_lines(self, name: str) -> list[dict]: + try: + return [json.loads(l) for l in self._p(name).read_text().splitlines() if l.strip()] + except Exception: + return [] + + @property + def breaker_path(self) -> Path: + # Breaker state is global (the API is up or down for everyone), not per session. + return HOME / "breaker.json"