fix: serialize concurrent JJ repository initialization

CHANGELOG.md

@@ -4,6 +4,9 @@ All notable changes to FAVA Trails are documented here.
 
 ## Unreleased
 
+### Fixed
+- Serialize monorepo initialization across processes so two MCP server processes sharing one data repository can start together. Concurrent `jj git init --colocate` racing could abort one server with `target repo already exists`, and unsynchronized `jj config set --repo` writers could leave the shared repository config file unparseable (`Configuration cannot be parsed as TOML document`), failing server startup. `JjBackend.init_monorepo` now takes a blocking cross-process lock on the repository directory for the whole init. Addresses issue #142.
+
 ## [0.8.0] — 2026-10-02
 
 **Jev at the boundary between agent output and shared memory.** FAVA Trails

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fava-trails"
-version = "0.8.2"
+version = "0.8.3"
 description = "FAVA Trails 🫛👣 — Federated Agents Versioned Audit Trail. VCS-backed memory for AI agents via MCP."
 readme = "README.md"
 requires-python = ">=3.11"

src/fava_trails/__init__.py

@@ -10,4 +10,4 @@
 try:
     __version__ = version("fava-trails")
 except PackageNotFoundError:  # pragma: no cover - source tree without install metadata
-    __version__ = "0.8.2"
+    __version__ = "0.8.3"

src/fava_trails/vcs/jj_backend.py

@@ -11,6 +11,7 @@
 import os
 import re
 import shutil
+from contextlib import contextmanager
 from pathlib import Path
 
 from .base import (
@@ -81,6 +82,55 @@ def classify_remote_fetch_error(message: str) -> str:
     return "unreachable"
 
 
+@contextmanager
+def _repo_init_lock(repo_root: Path):
+    """Blocking cross-process exclusive lock serializing monorepo initialization.
+
+    Server processes that share a data repository run ``init_monorepo`` at
+    startup. On a fresh repository they race on ``jj git init --colocate`` (jj
+    refuses the second creator), and on every startup they rewrite the shared
+    repository config through four ``jj config set --repo`` calls. Two
+    unsynchronized writers can leave that config file unparseable, which aborts
+    the second server before it can serve any request (issue #142).
+
+    The lock is taken on the repository directory itself so it is available
+    before ``.jj`` exists — jj refuses to initialize a repository whose ``.jj``
+    directory was pre-created, so a lock file inside ``.jj`` cannot guard the
+    fresh-init case. Platforms without directory locking fall back to a lock
+    file beside jj's metadata once that metadata exists.
+    """
+    if os.name == "nt" or not hasattr(os, "O_DIRECTORY"):
+        lock_file = repo_root / ".jj" / "fava-init.lock"
+        if not lock_file.parent.is_dir():
+            yield
+            return
+        import msvcrt
+
+        with lock_file.open("a+b") as stream:
+            stream.write(b"0")
+            stream.flush()
+            stream.seek(0)
+            msvcrt.locking(stream.fileno(), msvcrt.LK_LOCK, 1)
+            try:
+                yield
+            finally:
+                stream.seek(0)
+                msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
+        return
+
+    import fcntl
+
+    fd = os.open(repo_root, os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX)
+        try:
+            yield
+        finally:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+    finally:
+        os.close(fd)
+
+
 class JjError(Exception):
     """Error from jj subprocess."""
 
@@ -292,13 +342,23 @@ async def _executable_bit_changes(self, paths: list[str]) -> list[str]:
     async def init_monorepo(self) -> str:
         """Initialize the monorepo at repo_root with four-case detection.
 
+        Serialized across processes (see ``_repo_init_lock``): concurrent server
+        startup on one shared data repository must not race the JJ init or the
+        shared repository config writes.
+        """
+        self.repo_root.mkdir(parents=True, exist_ok=True)
+        with _repo_init_lock(self.repo_root):
+            return await self._init_monorepo_locked()
+
+    async def _init_monorepo_locked(self) -> str:
+        """Init body; caller holds the cross-process repository init lock.
+
         Cases:
         1. .git only (no .jj) → colocate JJ on top of existing git repo
         2. Both .jj and .git exist → already initialized, configure and skip
         3. .jj only (no .git) → non-colocated repo, raise error with fix instructions
         4. Neither exists → fresh init with jj git init --colocate
         """
-        self.repo_root.mkdir(parents=True, exist_ok=True)
         jj_dir = self.repo_root / ".jj"
         git_dir = self.repo_root / ".git"
 

tests/test_jj_backend.py

@@ -1,11 +1,26 @@
 """Tests for JjBackend VCS operations."""
 
+import asyncio
+import os
+import shutil
 import subprocess
+import sys
+import textwrap
+from pathlib import Path
 from unittest.mock import patch
 
 import pytest
 
 
+def _jj_bin() -> str:
+    jj = shutil.which("jj")
+    if jj:
+        return jj
+    fallback = Path.home() / ".local" / "bin" / "jj"
+    assert fallback.is_file(), "jj binary not found — install via: fava-trails install-jj"
+    return str(fallback)
+
+
 @pytest.mark.asyncio
 async def test_init_trail(jj_backend):
     """.jj and .git at repo_root (monorepo), trail_path is just a directory."""
@@ -632,3 +647,89 @@ async def test_init_monorepo_sets_default_description(tmp_path):
         cwd=str(repo), capture_output=True, text=True,
     )
     assert proc.stdout.strip() == "(auto-described)"
+
+
+# A real worker process for concurrent-init coverage. It signals readiness,
+# waits for a shared release file, then initializes the monorepo — so both
+# processes reach init_monorepo together instead of being scheduled serially.
+_CONCURRENT_INIT_WORKER = textwrap.dedent(
+    '''
+    import asyncio
+    import sys
+    from pathlib import Path
+
+    from fava_trails.vcs.jj_backend import JjBackend
+
+
+    async def main():
+        repo = Path(sys.argv[1])
+        barrier = Path(sys.argv[2])
+        (barrier / f"ready-{sys.argv[3]}").write_text("ready")
+        go = barrier / "go"
+        for _ in range(2000):
+            if go.exists():
+                break
+            await asyncio.sleep(0.01)
+        trail = repo / "trails" / "process-isolation"
+        backend = JjBackend(repo_root=repo, trail_path=trail)
+        await backend.init_monorepo()
+        await backend.init_trail()
+
+
+    asyncio.run(main())
+    '''
+)
+
+
+@pytest.mark.asyncio
+async def test_concurrent_init_monorepo_is_serialized(tmp_path):
+    """Issue #142: two processes initializing one shared data repo must not race.
+
+    Real two-process coverage — not a simulator. Two OS processes each run
+    ``JjBackend.init_monorepo()`` on the same fresh repository, released
+    together from a shared start barrier. Before the fix one process aborted on
+    jj's "target repo already exists" from the unsynchronized
+    ``jj git init --colocate``, and the four ``jj config set --repo`` calls
+    rewrote the shared repository config without serialization.
+    """
+    repo = tmp_path / "shared-repo"
+    repo.mkdir()
+    xdg = tmp_path / "xdg-config"
+    xdg.mkdir()
+    barrier = tmp_path / "barrier"
+    barrier.mkdir()
+    worker = tmp_path / "init_worker.py"
+    worker.write_text(_CONCURRENT_INIT_WORKER)
+
+    env = {key: value for key, value in os.environ.items() if not key.startswith("FAVA_TRAILS_")}
+    env["XDG_CONFIG_HOME"] = str(xdg)
+    env["PATH"] = os.pathsep.join(
+        [str(Path(sys.executable).parent), str(Path(_jj_bin()).parent), os.environ.get("PATH", "")]
+    )
+
+    procs = [
+        await asyncio.create_subprocess_exec(
+            sys.executable, str(worker), str(repo), str(barrier), str(i),
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            env=env,
+        )
+        for i in range(2)
+    ]
+    for _ in range(2000):
+        if len(list(barrier.glob("ready-*"))) == 2:
+            break
+        await asyncio.sleep(0.01)
+    assert len(list(barrier.glob("ready-*"))) == 2, "init workers did not reach the start barrier"
+    (barrier / "go").write_text("go")
+
+    results = [await proc.communicate() for proc in procs]
+    assert all(proc.returncode == 0 for proc in procs), [stderr.decode() for _, stderr in results]
+
+    assert (repo / ".jj").exists()
+    assert (repo / ".git").exists()
+    check = subprocess.run(
+        ["jj", "config", "list", "--repo"],
+        cwd=str(repo), env=env, capture_output=True, text=True,
+    )
+    assert check.returncode == 0, check.stderr