chore-shared-helper-duplication-hooks-services-2026-06-10: extract shared helpers

better_memory/_common.py

@@ -0,0 +1,64 @@
+"""Tiny shared helpers used across hooks, services, storage, and the MCP layer.
+
+Pure stdlib with no intra-package imports, so hooks can import this module
+without paying for SQLite / sqlite-vec / boto3 / config at invocation time
+(hooks must start fast and never fail). Everything heavier belongs in
+:mod:`better_memory.config` or the relevant service module.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime
+from pathlib import Path
+from uuid import uuid4
+
+_DEFAULT_HOME = "~/.better-memory"
+
+
+def env_session_id() -> str | None:
+    """Return the Claude session id from the environment, or ``None``.
+
+    Reads ``CLAUDE_SESSION_ID`` first (kept as the primary name for
+    back-compat), then ``CLAUDE_CODE_SESSION_ID`` (the name Claude Code
+    actually exports). The shared resolution order makes hook-written
+    events and MCP-written observations agree on the same session id.
+    """
+    return (
+        os.environ.get("CLAUDE_SESSION_ID")
+        or os.environ.get("CLAUDE_CODE_SESSION_ID")
+    )
+
+
+def get_session_id() -> str:
+    """Return the current Claude session id, generating one if no env var
+    is set. Falls back to a fresh ``uuid4().hex`` (32 chars)."""
+    return env_session_id() or uuid4().hex
+
+
+def default_clock() -> datetime:
+    """UTC-aware ``now``. The conventional default for injectable ``clock``
+    parameters across services."""
+    return datetime.now(UTC)
+
+
+def resolve_home() -> Path:
+    """Return ``BETTER_MEMORY_HOME`` (or its default) with ``~`` expanded."""
+    raw = os.environ.get("BETTER_MEMORY_HOME", _DEFAULT_HOME)
+    return Path(raw).expanduser()
+
+
+def default_spool_dir() -> Path:
+    """Return ``$BETTER_MEMORY_HOME/spool``, defaulting to ``~/.better-memory``."""
+    return resolve_home() / "spool"
+
+
+def safe_timestamp(raw: str | None) -> str:
+    """Return a filesystem-safe timestamp component.
+
+    Replaces ``:`` (illegal on NTFS) with ``-``. Falls back to current UTC
+    time if ``raw`` is missing or empty.
+    """
+    if not raw:
+        raw = datetime.now(UTC).isoformat()
+    return raw.replace(":", "-")

better_memory/config.py

@@ -23,8 +23,8 @@
 from typing import Literal
 
 from better_memory import _diag
+from better_memory._common import resolve_home
 
-_DEFAULT_HOME = "~/.better-memory"
 _DEFAULT_OLLAMA_HOST = "http://localhost:11434"
 _DEFAULT_EMBED_MODEL = "nomic-embed-text"
 _DEFAULT_EMBEDDINGS_BACKEND = "ollama"
@@ -34,12 +34,6 @@
 _DEFAULT_AGENTCORE_REGION = "eu-west-2"
 
 
-def resolve_home() -> Path:
-    """Return ``BETTER_MEMORY_HOME`` (or its default) with ``~`` expanded."""
-    raw = os.environ.get("BETTER_MEMORY_HOME", _DEFAULT_HOME)
-    return Path(raw).expanduser()
-
-
 # Maps absolute cwd string → resolved project name (or None when no git tree
 # is reachable). Both successes and ``None`` are cached for the process
 # lifetime: the walk is deterministic given filesystem state, so caching the

better_memory/hooks/observer.py

@@ -16,37 +16,15 @@
 import sys
 import time
 from datetime import UTC, datetime
-from pathlib import Path
+
+from better_memory._common import default_spool_dir, safe_timestamp
 
 # Cap stdin reads so a malicious or accidentally huge payload can't starve the
 # hook process of memory. 1 MiB is far larger than anything Claude Code emits
 # in practice but small enough to be trivially bounded.
 _MAX_STDIN_BYTES = 1_048_576
 
 
-def _default_spool_dir() -> Path:
-    """Return ``$BETTER_MEMORY_HOME/spool``, defaulting to ``~/.better-memory``.
-
-    Kept separate from :func:`better_memory.config.get_config` so hooks do not
-    import SQLite / sqlite-vec / anything heavyweight at invocation time.
-    """
-    home = os.environ.get("BETTER_MEMORY_HOME")
-    if home:
-        return Path(home).expanduser() / "spool"
-    return Path.home() / ".better-memory" / "spool"
-
-
-def _safe_timestamp(raw: str | None) -> str:
-    """Return a filesystem-safe timestamp component.
-
-    Replaces ``:`` (illegal on NTFS) with ``-``. Falls back to current UTC
-    time if ``raw`` is missing or empty.
-    """
-    if not raw:
-        raw = datetime.now(UTC).isoformat()
-    return raw.replace(":", "-")
-
-
 def _safe_tool(raw: object) -> str:
     """Return a filesystem-safe tool component."""
     if not raw or not isinstance(raw, str):
@@ -75,10 +53,10 @@ def main() -> None:
         if "timestamp" not in data or not data["timestamp"]:
             data["timestamp"] = datetime.now(UTC).isoformat()
 
-        spool_dir = _default_spool_dir()
+        spool_dir = default_spool_dir()
         spool_dir.mkdir(parents=True, exist_ok=True)
 
-        ts_component = _safe_timestamp(data.get("timestamp"))
+        ts_component = safe_timestamp(data.get("timestamp"))
         tool_component = _safe_tool(data.get("tool"))
         # SHA-256 prefix of the serialised payload — cheap collision avoidance
         # for two events that happen in the same second on the same tool. The

better_memory/hooks/post_commit.py

@@ -29,9 +29,13 @@
 import time
 from datetime import UTC, datetime
 from pathlib import Path
-from uuid import uuid4
 
-from better_memory.config import project_name, resolve_home
+from better_memory._common import (
+    default_spool_dir,
+    get_session_id,
+    safe_timestamp,
+)
+from better_memory.config import project_name
 
 _MAX_STDIN_BYTES = 1_048_576
 
@@ -42,17 +46,6 @@
 _TRAILER_KEY = "closes-episode"
 
 
-def _default_spool_dir() -> Path:
-    """Return ``$BETTER_MEMORY_HOME/spool``, defaulting to ``~/.better-memory``."""
-    return resolve_home() / "spool"
-
-
-def _safe_timestamp(raw: str | None) -> str:
-    if not raw:
-        raw = datetime.now(UTC).isoformat()
-    return raw.replace(":", "-")
-
-
 def _read_head_commit_message() -> tuple[str, str]:
     """Return ``(subject_plus_body, commit_sha)`` for HEAD.
 
@@ -146,11 +139,7 @@ def main() -> None:
             sys.exit(0)
 
         now_iso = datetime.now(UTC).isoformat()
-        session_id = (
-            os.environ.get("CLAUDE_SESSION_ID")
-            or os.environ.get("CLAUDE_CODE_SESSION_ID")
-            or uuid4().hex
-        )
+        session_id = get_session_id()
         cwd = _resolve_cwd()
 
         data: dict[str, object] = {
@@ -162,10 +151,10 @@ def main() -> None:
             "commit_sha": commit_sha,
         }
 
-        spool_dir = _default_spool_dir()
+        spool_dir = default_spool_dir()
         spool_dir.mkdir(parents=True, exist_ok=True)
 
-        ts_component = _safe_timestamp(now_iso)
+        ts_component = safe_timestamp(now_iso)
         serialised = json.dumps(data, sort_keys=True).encode("utf-8")
         salt = f"{time.time_ns()}:{os.getpid()}".encode()
         hash_hex = hashlib.sha256(serialised + salt).hexdigest()[:12]

better_memory/hooks/session_bootstrap.py

@@ -13,8 +13,8 @@
 import sys
 from contextlib import closing
 from pathlib import Path
-from uuid import uuid4
 
+from better_memory._common import get_session_id
 from better_memory.config import get_config
 from better_memory.db.connection import connect
 from better_memory.hooks._error_log import record_hook_error
@@ -62,11 +62,7 @@ def main() -> None:
     session_id = (
         str(payload.get("session_id"))
         if payload.get("session_id")
-        else (
-            os.environ.get("CLAUDE_SESSION_ID")
-            or os.environ.get("CLAUDE_CODE_SESSION_ID")
-            or uuid4().hex
-        )
+        else get_session_id()
     )
     cwd_str = str(payload.get("cwd")) if payload.get("cwd") else os.getcwd()
 

better_memory/hooks/session_close.py

@@ -16,45 +16,28 @@
 import sys
 import time
 from datetime import UTC, datetime
-from pathlib import Path
-from uuid import uuid4
 
+from better_memory._common import (
+    default_spool_dir,
+    env_session_id,
+    get_session_id,
+    resolve_home,
+    safe_timestamp,
+)
 from better_memory.config import get_config
 
 # Mirror the observer cap: reject any stdin payload above 1 MiB without
 # raising. Hooks must never fail.
 _MAX_STDIN_BYTES = 1_048_576
 
 
-def _default_spool_dir() -> Path:
-    """Return ``$BETTER_MEMORY_HOME/spool``, defaulting to ``~/.better-memory``.
-
-    Mirrors the observer hook. Kept duplicated to avoid a cross-module import
-    that would slow hook startup.
-    """
-    home = os.environ.get("BETTER_MEMORY_HOME")
-    if home:
-        return Path(home).expanduser() / "spool"
-    return Path.home() / ".better-memory" / "spool"
-
-
-def _safe_timestamp(raw: str | None) -> str:
-    if not raw:
-        raw = datetime.now(UTC).isoformat()
-    return raw.replace(":", "-")
-
-
 def _synthesise_marker() -> dict[str, str]:
     """Build a minimal ``session_end`` payload from env + clock."""
     return {
         "event_type": "session_end",
         "timestamp": datetime.now(UTC).isoformat(),
         "cwd": os.environ.get("PWD") or os.getcwd(),
-        "session_id": (
-            os.environ.get("CLAUDE_SESSION_ID")
-            or os.environ.get("CLAUDE_CODE_SESSION_ID")
-            or uuid4().hex
-        ),
+        "session_id": get_session_id(),
     }
 
 
@@ -103,12 +86,7 @@ def _fire_agentcore_closure(*, session_id: str, project: str) -> bool:
             resolve_actor_id,
         )
 
-        home_env = os.environ.get("BETTER_MEMORY_HOME")
-        home = (
-            Path(home_env).expanduser()
-            if home_env
-            else Path.home() / ".better-memory"
-        )
+        home = resolve_home()
         cfg = load_agentcore_config(home)
         if cfg is None:
             return False
@@ -267,19 +245,12 @@ def main() -> None:
         if "timestamp" not in data or not data["timestamp"]:
             data["timestamp"] = datetime.now(UTC).isoformat()
         if "session_id" not in data or not data["session_id"]:
-            data["session_id"] = (
-                os.environ.get("CLAUDE_SESSION_ID")
-                or os.environ.get("CLAUDE_CODE_SESSION_ID")
-                or uuid4().hex
-            )
+            data["session_id"] = get_session_id()
         if "cwd" not in data or not data["cwd"]:
             data["cwd"] = os.environ.get("PWD") or os.getcwd()
 
         session_id_str = (
-            os.environ.get("CLAUDE_SESSION_ID")
-            or os.environ.get("CLAUDE_CODE_SESSION_ID")
-            or data.get("session_id")
-            or ""
+            env_session_id() or data.get("session_id") or ""
         )
         if session_id_str and _emit_rating_directive_if_unrated(
             str(session_id_str)
@@ -307,10 +278,10 @@ def main() -> None:
             project=str(project_for_closure),
         )
 
-        spool_dir = _default_spool_dir()
+        spool_dir = default_spool_dir()
         spool_dir.mkdir(parents=True, exist_ok=True)
 
-        ts_component = _safe_timestamp(str(data.get("timestamp")))
+        ts_component = safe_timestamp(str(data.get("timestamp")))
         # Salt the hash with monotonic-nanosecond clock + PID so two
         # byte-identical payloads in the same second can't collide on
         # filename. The salt does NOT appear in the written body.

better_memory/mcp/_util.py

@@ -8,13 +8,13 @@
 from __future__ import annotations
 
 import logging
-import os
 import time
 from collections.abc import Callable
 from pathlib import Path
 from typing import Any
 
 from better_memory import _diag
+from better_memory._common import env_session_id
 from better_memory.runtime.session_marker import read_session_id
 
 logger = logging.getLogger(__name__)
@@ -64,8 +64,4 @@ def resolve_session_id(home: Path) -> str | None:
     propagate the session id into the spawned stdio MCP server's env, so
     the marker file is the fallback for every rating call.
     """
-    return (
-        os.environ.get("CLAUDE_SESSION_ID")
-        or os.environ.get("CLAUDE_CODE_SESSION_ID")
-        or read_session_id(home)
-    )
+    return env_session_id() or read_session_id(home)

better_memory/mcp/handlers/sessions.py

@@ -8,12 +8,12 @@
 
 import json
 import os
-import uuid
 from pathlib import Path
 from typing import Any
 
 from mcp.types import TextContent
 
+from better_memory._common import get_session_id
 from better_memory.mcp._util import resolve_session_id
 from better_memory.services import ui_launcher
 from better_memory.services.memory_rating import MemoryRatingService
@@ -45,12 +45,7 @@ def tools(self) -> dict[str, Any]:
 
     async def session_bootstrap(self, args: dict[str, Any]) -> list[TextContent]:
         cwd_arg = args.get("cwd") or os.getcwd()
-        session_id_arg = (
-            args.get("session_id")
-            or os.environ.get("CLAUDE_SESSION_ID")
-            or os.environ.get("CLAUDE_CODE_SESSION_ID")
-            or uuid.uuid4().hex
-        )
+        session_id_arg = args.get("session_id") or get_session_id()
         result = self._session_bootstrap.bootstrap(
             source=args.get("source"),
             session_id=session_id_arg,