emrg: one producer for the directory a session lives in

emrg/client/app.py

@@ -26,7 +26,7 @@
 from emrg.protocol import TaskResponse, ToolEnd, ToolStart, new_task_id
 from emrg.sandbox.escalation import APPROVAL_TIMEOUT_SECONDS
 from emrg.sandbox.policy import SANDBOX_MODES
-from emrg.session import generate_session_id
+from emrg.session import generate_session_id, session_dir
 # The credential rule is a stdlib-only leaf (`emrg/server/redaction.py`), deliberately
 # importable from outside the daemon: this is the client's own log surface, and a window
 # applied there before the rule sees the text is the defect that module exists to prevent.
@@ -2719,7 +2719,7 @@ async def handle_key(data: bytes) -> bool:
             # After paste, check clipboard for images
             has_image, label = _detect_clipboard_image()
             if has_image:
-                images_dir = Path(cwd) / ".emrg" / "sessions" / session_id / "images"
+                images_dir = session_dir(cwd, session_id) / "images"
                 images_dir.mkdir(parents=True, exist_ok=True)
                 counter = len(_pending_images) + 1
                 tmp_path = images_dir / f"_clipboard_tmp_{counter}.png"
@@ -3440,7 +3440,7 @@ def _is_image_token(s, i):
                         if _is_image_token(inp.text, pos):
                             has_image, label = _detect_clipboard_image()
                             if has_image:
-                                images_dir = Path(cwd) / ".emrg" / "sessions" / session_id / "images"
+                                images_dir = session_dir(cwd, session_id) / "images"
                                 images_dir.mkdir(parents=True, exist_ok=True)
                                 counter = len(_pending_images) + 1
                                 tmp_path = images_dir / f"_clipboard_tmp_{counter}.png"

emrg/server/daemon.py

@@ -98,6 +98,7 @@
     is_safe_session_id,
     last_n_messages,
     records_to_messages,
+    session_dir,
 )
 
 
@@ -3551,8 +3552,8 @@ async def _process_message(
                 })
                 return
 
-            session_dir = Path(cwd) / ".emrg" / "sessions" / session_id
-            if not session_dir.exists():
+            session_path = session_dir(cwd, session_id)
+            if not session_path.exists():
                 await self._send(ws, {
                     "type": "session_deleted",
                     "error": f"Session {session_id} not found",
@@ -3917,8 +3918,8 @@ async def _process_message(
 
     def _get_or_create_session(self, session_id: str, cwd: Path) -> Session:
         """Load an existing session or create a new one."""
-        session_dir = cwd / ".emrg" / "sessions" / session_id
-        if session_dir.exists() and (session_dir / "meta.json").exists():
+        session_path = session_dir(cwd, session_id)
+        if session_path.exists() and (session_path / "meta.json").exists():
             return Session.load(session_id, cwd)
         return Session.create_with_id(session_id, cwd)
 
@@ -7015,8 +7016,8 @@ async def _handle_resume_session(
         或 message_count==0 时，按全局 sessions_index.json 校验请求 cwd 是否为
         该会话真实项目路径；不匹配 → not_found + warning（幽灵目录不再掩盖错误）。
         """
-        session_dir = cwd / ".emrg" / "sessions" / session_id
-        meta_path = session_dir / "meta.json"
+        session_path = session_dir(cwd, session_id)
+        meta_path = session_path / "meta.json"
 
         # ── Ask the meta's kind before opening it ──
         #
@@ -7063,7 +7064,7 @@ async def _handle_resume_session(
         # this cwd can resume. A missing path is unchanged — `_non_regular_kind` answers
         # `None` for a subject whose `stat` fails.
         kind = _non_regular_kind(meta_path)
-        if kind or not session_dir.exists() or not meta_path.exists():
+        if kind or not session_path.exists() or not meta_path.exists():
             if kind:
                 logger.warning(
                     "resume rejected: %s is %s, not a regular file (session %s)",
@@ -7087,7 +7088,7 @@ async def _handle_resume_session(
                 logger.warning(
                     "resume rejected: ghost session %s (0 messages) at %s — "
                     "real session lives at %s (canonical cwd from sessions index)",
-                    session_id, session_dir, canonical or "unknown",
+                    session_id, session_path, canonical or "unknown",
                 )
                 await self._send(ws, {
                     "type": "resume_result",
@@ -7409,14 +7410,14 @@ async def _handle_list_memories(
             store = ProjectMemoryStore(cwd_path)
             index_path = store.index_path
         else:
-            session_dir = Path(cwd) / ".emrg" / "sessions" / session_id
-            if not session_dir.exists():
+            session_path = session_dir(cwd, session_id)
+            if not session_path.exists():
                 await self._send(ws, {
                     "type": "memories_list",
                     "error": f"Session {session_id} not found",
                 })
                 return
-            store = SessionMemoryStore(session_dir)
+            store = SessionMemoryStore(session_path)
             index_path = store.index_path
 
         memories = store.list()
@@ -7452,14 +7453,14 @@ async def _handle_read_memory(
             cwd_path = Path(cwd) if cwd else Path.cwd()
             store = ProjectMemoryStore(cwd_path)
         else:
-            session_dir = Path(cwd) / ".emrg" / "sessions" / session_id
-            if not session_dir.exists():
+            session_path = session_dir(cwd, session_id)
+            if not session_path.exists():
                 await self._send(ws, {
                     "type": "memory_content",
                     "error": f"Session {session_id} not found",
                 })
                 return
-            store = SessionMemoryStore(session_dir)
+            store = SessionMemoryStore(session_path)
 
         # `get_with_reason` rather than `get`: "there is no such memory" and "the file
         # could not be read" are different answers, and a bare `None` collapses them
@@ -7710,8 +7711,7 @@ async def _run_subagent_child(
         # project root, and it is why nothing here fights `is_safe_session_id`,
         # which constrains the id alone.
         slot_cwd = (
-            Path(parent_cwd) / ".emrg" / "sessions" / parent_session_id
-            / "sub-agent" / f"sub{slot}"
+            session_dir(parent_cwd, parent_session_id) / "sub-agent" / f"sub{slot}"
         )
 
         try:
@@ -7894,9 +7894,7 @@ async def _run_boss_child(
         # fixed `boss` would make two tasks' owners indistinguishable there.
         child_id = f"boss-{parent_session_id}"
         state["child"] = child_id
-        parent_session_dir = (
-            Path(parent_cwd) / ".emrg" / "sessions" / parent_session_id
-        )
+        parent_session_dir = session_dir(parent_cwd, parent_session_id)
         # The cwd is two levels below the parent's session directory, which is how
         # the owner derives the history it must read: `<parent>/sub-agent/sub-boss`.
         boss_cwd = parent_session_dir / "sub-agent" / self.BOSS_DIR

emrg/session.py

@@ -340,13 +340,30 @@ def is_safe_session_id(session_id: object) -> bool:
     return True
 
 
+def session_dir(cwd: Path, session_id: str) -> Path:
+    """The directory a session's state lives in, for a workspace and an id.
+
+    This is the one producer of ``<cwd>/.emrg/sessions/<session_id>``. The rule has
+    more than one reader: ``Session`` fills the directory it computes here, and the
+    daemon derives a *parent's* directory from it to place a child's cwd inside —
+    ``sub-agent/sub<N>`` for a delegation, ``sub-agent/sub-boss`` for an owner. A
+    caller that has an id and a cwd but no handle asks this, so a layout change is an
+    edit in one place rather than a search for every copy of the expression.
+
+    The id is not validated here — ``is_safe_session_id`` decides that, and its caller
+    decides whether an unsafe id is refused or merely not a directory (see its
+    docstring for what an unchecked id did).
+    """
+    return Path(cwd) / ".emrg" / "sessions" / session_id
+
+
 class Session:
     """Manages a single conversation session on disk."""
 
     def __init__(self, session_id: str, cwd: Path) -> None:
         self.session_id = session_id
         self.cwd = cwd
-        self._dir = cwd / ".emrg" / "sessions" / session_id
+        self._dir = session_dir(cwd, session_id)
         self._dir.mkdir(parents=True, exist_ok=True)
 
         # Initialize memory subdirectory
@@ -1063,9 +1080,9 @@ def delete(session_id: str, cwd: Path) -> bool:
             )
             return False
 
-        session_dir = cwd / ".emrg" / "sessions" / session_id
-        if session_dir.exists():
-            shutil.rmtree(session_dir)
+        target = session_dir(cwd, session_id)
+        if target.exists():
+            shutil.rmtree(target)
             logger.info("session deleted: %s", session_id)
             # Global index (rant 2026-08-13T16:42:22): drop the deleted session.
             remove_session_index(session_id)

tests/test_session.py

@@ -1,7 +1,10 @@
 """Tests for session module: Session class and _validate_tool_messages."""
 
+import ast
 import json
 import os
+import re
+import warnings
 from datetime import datetime
 from pathlib import Path
 
@@ -14,9 +17,175 @@
     _validate_tool_messages,
     generate_session_id,
     last_n_messages,
+    session_dir,
 )
 
 
+def test_session_dir_names_the_directory_a_session_actually_fills(tmp_path):
+    """The class and the rule cannot disagree about where a session lives.
+
+    The subject has to be non-empty for this to mean anything: two spellings of the
+    same string compare equal however wrong the rule is, so the assertion is about a
+    directory `Session` really made, on disk, and then about the rule naming that same
+    directory.
+    """
+    sid = "s_260101_0000_abcd"
+    session = Session.create_with_id(sid, tmp_path)
+
+    assert session.dir_path.is_dir(), (
+        "the subject is empty — nothing was created, so an equality check below "
+        "would pass without saying anything"
+    )
+    assert session.dir_path == session_dir(tmp_path, sid)
+    assert session.dir_path.resolve() == session_dir(tmp_path, sid).resolve()
+    assert session.dir_path.parent == tmp_path / ".emrg" / "sessions"
+
+
+def test_session_dir_keeps_the_workspace_it_was_given(tmp_path):
+    """A relative workspace stays relative, so the caller's cwd still decides."""
+    assert str(session_dir("project", "s_260101_0000_abcd")) == str(
+        Path("project") / ".emrg" / "sessions" / "s_260101_0000_abcd"
+    )
+
+
+def _slash_chain(node: ast.AST) -> list[ast.AST]:
+    """`a / b / c` flattened to `[a, b, c]`, outermost division first."""
+    parts: list[ast.AST] = []
+    while isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div):
+        parts.append(node.right)
+        node = node.left
+    parts.append(node)
+    parts.reverse()
+    return parts
+
+
+def _spells_a_session_directory(tree: ast.AST) -> list[int]:
+    """Line numbers of expressions that build `<cwd>/.emrg/sessions/<id>`.
+
+    Read from the syntax tree rather than the text, because the subject is *who
+    builds the path*, not how it is spelled: a scan for one spelling passes a
+    second author who writes `sessions / sid`, and a scan of the text counts the
+    producer's own docstring as a second producer. The operand after `"sessions"`
+    is what makes it a leaf — `cwd / ".emrg" / "sessions"` without one names the
+    directory sessions live *in*, which several callers may legitimately ask for.
+    """
+    found: list[int] = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.BinOp) or not isinstance(node.op, ast.Div):
+            continue
+        parts = _slash_chain(node)
+        for i in range(len(parts) - 2):
+            head, tail = parts[i], parts[i + 1]
+            if (
+                isinstance(head, ast.Constant) and head.value == ".emrg"
+                and isinstance(tail, ast.Constant) and tail.value == "sessions"
+                and not isinstance(parts[i + 2], ast.Constant)
+            ):
+                found.append(node.lineno)
+                break
+    return found
+
+
+def test_the_directory_a_session_lives_in_has_one_producer():
+    """The rule is `session_dir`, and no other module may build the path itself.
+
+    Dozens of call sites need this directory, and each one that spelled it out
+    was a copy that a change to the layout would have to find. They agreed, so
+    nothing was broken — which is exactly what makes a copy invisible to a
+    behavioural test: five of them survived a refactor that claimed to leave one
+    producer, because every copy still produced the right answer.
+    """
+    package_root = Path(__file__).resolve().parents[1] / "emrg"
+    producer, offenders = [], []
+    for path in sorted(package_root.rglob("*.py")):
+        rel = path.relative_to(package_root).as_posix()
+        # A module elsewhere in the package carries an invalid escape sequence in a
+        # string; parsing it warns, and that warning is about a file this test does
+        # not judge. Silenced here rather than fixed there, so the guard reads only
+        # what it asserts.
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", SyntaxWarning)
+            tree = ast.parse(path.read_bytes())
+        for lineno in _spells_a_session_directory(tree):
+            (producer if rel == "session.py" else offenders).append(f"{rel}:{lineno}")
+
+    assert offenders == [], (
+        "these build <cwd>/.emrg/sessions/<id> themselves instead of asking "
+        f"session_dir(): {offenders} — a second producer is a second rule"
+    )
+    assert len(producer) == 1, (
+        f"expected exactly one producer in session.py, found {producer}"
+    )
+
+
+def _javascript_without_comments(source: str) -> str:
+    """`source` with every comment blanked to spaces, newlines and strings kept.
+
+    Line numbers survive, so a failure names the line a reader will find. Two
+    cheaper spellings are wrong here and both were considered: a regex that strips
+    `//` to end of line also eats the `//` inside a URL or a string, and dropping
+    comment lines outright shifts every number below them. A scan that silently
+    removed a line of code would report a lower count as clean.
+    """
+    out: list[str] = []
+    i, n = 0, len(source)
+    while i < n:
+        ch = source[i]
+        if ch == "/" and i + 1 < n and source[i + 1] == "/":
+            j = source.find("\n", i)
+            j = n if j < 0 else j
+            out.append(" " * (j - i))
+            i = j
+        elif ch == "/" and i + 1 < n and source[i + 1] == "*":
+            j = source.find("*/", i + 2)
+            j = n if j < 0 else j + 2
+            out.append("".join(c if c == "\n" else " " for c in source[i:j]))
+            i = j
+        elif ch in "\"'`":
+            quote, j = ch, i + 1
+            while j < n and source[j] != quote:
+                j += 2 if source[j] == "\\" else 1
+            out.append(source[i:j + 1])
+            i = j + 1
+        else:
+            out.append(ch)
+            i += 1
+    return "".join(out)
+
+
+def test_the_electron_client_builds_the_same_path_in_one_place():
+    """The rule's other language: the GUI main process is its own single producer.
+
+    `main.js` writes a pasted image into `<project>/.emrg/sessions/<id>/images` so
+    the daemon can find it, and it cannot call the Python producer — the two
+    processes share a filesystem, not a runtime. So delegation is not available on
+    this half; what is available is that the copy stays **one** copy, and that is
+    what this pins. Measured when written: exactly one code site, in `main.js`.
+    The path is also spelled in comments in `preload.js` and `main.js`, which is
+    why the comments are blanked rather than counted.
+    """
+    gui = Path(__file__).resolve().parents[1] / "emrg" / "gui"
+    sites: list[str] = []
+    for path in sorted(gui.rglob("*.js")):
+        if any(part in {"node_modules", "dist"} for part in path.relative_to(gui).parts):
+            continue
+        if path.name.endswith(".test.js"):
+            continue
+        code = _javascript_without_comments(
+            path.read_text(encoding="utf-8", errors="replace")
+        )
+        for match in re.finditer(r"""["']\.emrg["']\s*,\s*["']sessions["']""", code):
+            line = code[: match.start()].count("\n") + 1
+            sites.append(f"{path.relative_to(gui).as_posix()}:{line}")
+
+    assert len(sites) == 1, (
+        "expected exactly one place in the Electron client to build "
+        f"<project>/.emrg/sessions/<id>, found {sites} — the Python package gets "
+        "this from one producer; this half of the rule has to stay one copy, "
+        "because nothing compares them at runtime"
+    )
+
+
 class TestValidateToolMessages:
     """Tests for the _validate_tool_messages safety-net function."""