fix(coding): enroll canonical legacy transcript store

src/loushang/coding/bootstrap.py

@@ -1927,6 +1927,11 @@ def _create_agent_session_runtime(
             owned_transcripts=sys.platform == "linux",
             store_state_root=(resolve_platform_home() / "state/session-stores"
                               if sys.platform == "linux" else None),
+            enroll_legacy_shared_store=(
+                sys.platform == "linux"
+                and Path(session_dir).expanduser().resolve(strict=False)
+                == (resolve_platform_home() / "data/sessions").resolve(strict=False)
+            ),
         ),
         fixed_services=fixed_services,
         build_session=lambda session_manager, session_services, start_event: (

src/loushang/coding/runtime/agent_session_runtime.py

@@ -52,15 +52,21 @@ def __init__(
         session_index_flush_delay: float = 0.25,
         owned_transcripts: bool = False,
         store_state_root: Path | None = None,
+        enroll_legacy_shared_store: bool = False,
     ) -> None:
-        if type(owned_transcripts) is not bool:
+        if type(owned_transcripts) is not bool or type(enroll_legacy_shared_store) is not bool:
             raise TypeError("invalid owned transcript activation")
         if owned_transcripts and current_session is not None:
             raise ValueError("owned runtime must construct its own initial Session")
         if store_state_root is not None and not owned_transcripts:
             raise ValueError("store admission requires owned transcripts")
+        if enroll_legacy_shared_store and store_state_root is None:
+            raise ValueError("legacy store enrollment requires store admission")
         self._owned_transcript_factory = (
-            _create_owned_session_factory(store_state_root=store_state_root) if owned_transcripts else None
+            _create_owned_session_factory(
+                store_state_root=store_state_root,
+                enroll_legacy_shared_store=enroll_legacy_shared_store,
+            ) if owned_transcripts else None
         )
         transcript_type = (
             _bind_owned_session_manager(self._owned_transcript_factory)

src/loushang/coding/session_manager.py

@@ -154,7 +154,10 @@ def get_runtime_capability(self, slot: str) -> object | tuple[object, ...]:
         return self._lifecycle_session.product_binding.value(slot)
 
 
-def _create_owned_session_factory(*, store_state_root: Path | None = None) -> AgentTranscriptSessionFactory[
+def _create_owned_session_factory(
+    *, store_state_root: Path | None = None,
+    enroll_legacy_shared_store: bool = False,
+) -> AgentTranscriptSessionFactory[
     ResolvedRuntimeProfile, RuntimeProfileBinding,
 ]:
     """Create an application-owned factory; never replace the legacy singleton."""
@@ -170,6 +173,7 @@ def _create_owned_session_factory(*, store_state_root: Path | None = None) -> Ag
         session_file_factory=lifecycle.default_jsonl_session_file,
         owned_product_id=CODING_PRODUCT_ID,
         store_state_root=store_state_root,
+        enroll_legacy_shared_store=enroll_legacy_shared_store,
     )
 
 

src/loushang/harness/transcript/lifecycle.py

@@ -303,6 +303,7 @@ def prepare_owned_writer(
         store_state_root: Path | None = None,
         initialize_store: bool = False,
         store_root_observed: Event | None = None,
+        enroll_legacy_shared_store: bool = False,
     ) -> TranscriptWriterPreparation[BindingInputT, ProductBindingT]:
         """Purely prepare acquisition; callers retain this before the first await.
 
@@ -323,6 +324,7 @@ def prepare_owned_writer(
             expected_root_identity=expected_root_identity, expected_parent_identity=expected_parent_identity,
             store_state_root=store_state_root, initialize_store=initialize_store,
             store_root_observed=store_root_observed,
+            enroll_legacy_shared_store=enroll_legacy_shared_store,
         )
 
     def prepare_writer(

src/loushang/harness/transcript/session_factory.py

@@ -109,6 +109,7 @@ def __init__(
         index_writable: bool = True,
         store_state_root: Path | None = None,
         store_root_observed: Event | None = None,
+        enroll_legacy_shared_store: bool = False,
     ) -> None:
         if type(index_writable) is not bool:
             raise TypeError("index_writable must be a built-in bool")
@@ -130,10 +131,15 @@ def __init__(
         self._clock = clock or _utc_now
         self._conversation_id_factory = conversation_id_factory or _default_id
         self._owned_product_id = owned_product_id
+        if type(enroll_legacy_shared_store) is not bool:
+            raise TypeError("legacy store enrollment must be a built-in bool")
         if store_state_root is not None and owned_product_id is None:
             raise ValueError("store admission requires an owned persistent factory")
+        if enroll_legacy_shared_store and store_state_root is None:
+            raise ValueError("legacy store enrollment requires store admission")
         self._store_state_root = store_state_root
         self._store_root_observed = store_root_observed
+        self._enroll_legacy_shared_store = enroll_legacy_shared_store
         self._pending: dict[TranscriptWriterPreparation[BindingInputT, ProductBindingT], None] = {}
         self._loop: asyncio.AbstractEventLoop | None = None
         self._closing = False
@@ -223,6 +229,7 @@ def _prepare_owned(
             store_state_root=self._store_state_root,
             initialize_store=initialize_store and self._store_state_root is not None,
             store_root_observed=self._store_root_observed,
+            enroll_legacy_shared_store=self._enroll_legacy_shared_store,
         )
         self._pending[owner] = None
         return owner

src/loushang/harness/transcript/store_admission.py

@@ -11,6 +11,7 @@
 import hashlib
 import json
 import os
+import stat
 from dataclasses import dataclass
 from pathlib import Path
 from secrets import token_hex
@@ -114,9 +115,12 @@ class TranscriptStoreAdmission:
     """
 
     def __init__(self, root: Path, *, state_root: Path, create_if_missing: bool = False,
-                 root_observed: Event | None = None) -> None:
+                 root_observed: Event | None = None,
+                 enroll_legacy_shared_store: bool = False) -> None:
         root, state_root = Path(root), Path(state_root)
-        if type(create_if_missing) is not bool or (root_observed is not None and type(root_observed) is not Event):
+        if (type(create_if_missing) is not bool
+                or type(enroll_legacy_shared_store) is not bool
+                or (root_observed is not None and type(root_observed) is not Event)):
             raise TranscriptWriterError("invalid")
         for path in (root, state_root):
             if (not path.is_absolute() or path == path.parent or ".." in path.parts
@@ -129,6 +133,7 @@ def __init__(self, root: Path, *, state_root: Path, create_if_missing: bool = Fa
         key = hashlib.sha256(f"{_VERSION}\0{os.geteuid()}\0{root}".encode()).hexdigest()
         self.root, self.witness_root = root, state_root / key
         self._key, self._create = key, create_if_missing
+        self._enroll_legacy_shared_store = enroll_legacy_shared_store
         self._root_observed = root_observed
         self._existing = _WitnessLease(self.witness_root, _VERSION, key, create_lock=False)
         self._fresh = _WitnessLease(self.witness_root, _VERSION, key,
@@ -211,7 +216,8 @@ def _open(self, *, inspect_only: bool = False) -> TranscriptStoreBinding | None:
                     self._reject_residue()
                     if not _missing(self.root / ".transcript-writers"):
                         raise TranscriptWriterError("conflict")
-                    self._family.inspect_state_evidence(allow_legacy=True)
+                    if not self._enroll_legacy_shared_store:
+                        self._family.inspect_state_evidence(allow_legacy=True)
                 self._witness = self._fresh
             self._witness.acquire()
             if self._witness._canonical != self.witness_root:
@@ -259,8 +265,33 @@ def _open(self, *, inspect_only: bool = False) -> TranscriptStoreBinding | None:
             raise TranscriptWriterError("unavailable") from None
 
     def _reject_residue(self) -> None:
-        if any(os.path.lexists(self.root.parent / name) for name in ("session-assets", ".session-blob-writers")):
+        assets = self.root.parent / "session-assets"
+        writers = self.root.parent / ".session-blob-writers"
+        if not self._enroll_legacy_shared_store:
+            if any(os.path.lexists(path) for path in (assets, writers)):
+                raise TranscriptWriterError("conflict")
+            return
+        # This explicit compatibility grant is only for the pre-family layout.
+        # A blob-writer root is evidence that family-era initialization ran.
+        if os.path.lexists(writers):
             raise TranscriptWriterError("conflict")
+        for path in (assets, assets / ".locks"):
+            if not os.path.lexists(path):
+                continue
+            fd = os.open(
+                path,
+                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
+            )
+            try:
+                info = os.fstat(fd)
+                current = os.stat(path, follow_symlinks=False)
+                if (not os.path.samestat(info, current)
+                        or not stat.S_ISDIR(info.st_mode)
+                        or info.st_uid != os.geteuid()
+                        or stat.S_IMODE(info.st_mode) != 0o700):
+                    raise TranscriptWriterError("conflict")
+            finally:
+                os.close(fd)
 
     def check(self) -> None:
         """Revalidate the original known observation, granting no write rights."""

src/loushang/harness/transcript/writer_lifecycle.py

@@ -72,6 +72,7 @@ def __init__(
         store_state_root: Path | None = None,
         initialize_store: bool = False,
         store_root_observed: Event | None = None,
+        enroll_legacy_shared_store: bool = False,
     ) -> None:
         if (type(context) is not AgentTranscriptLifecycleContext or not context.persist
                 or context.session_file is None or context.session_file.parent != context.session_dir
@@ -82,13 +83,17 @@ def __init__(
             raise TranscriptWriterError("invalid")
         if writer is not None and (expected_root_identity is not None or expected_parent_identity is not None):
             raise TranscriptWriterError("invalid")
-        if (type(initialize_store) is not bool or (initialize_store and store_state_root is None)
+        if (type(initialize_store) is not bool
+                or type(enroll_legacy_shared_store) is not bool
+                or (enroll_legacy_shared_store and store_state_root is None)
+                or (initialize_store and store_state_root is None)
                 or (store_state_root is not None and (writer is not None or create_root
                     or expected_root_identity is not None or expected_parent_identity is not None))):
             raise TranscriptWriterError("invalid")
         self._store_admission = (TranscriptStoreAdmission(
             context.session_dir, state_root=store_state_root, create_if_missing=initialize_store,
             root_observed=store_root_observed,
+            enroll_legacy_shared_store=enroll_legacy_shared_store,
         ) if store_state_root is not None else None)
         self._admission_cleanup_task: asyncio.Task[None] | None = None
         self._context = deepcopy(context)

tests/architecture/test_coding_wave_a_budget.py

@@ -96,10 +96,14 @@ def test_coding_package_stays_within_wave_a_budget() -> None:
     # LMUX reviewed default-owned wiring: bootstrap +7, runtime +5,
     # manager +1 versus da820585. Keep all three in core and its six-line margin.
     lmux_default_owned_allowance = 7 + 5 + 1
+    # Canonical legacy-store enrollment stays in the same three Product owners:
+    # bootstrap +5, runtime +6 and manager +4. Preserve the existing margin.
+    canonical_legacy_enrollment_allowance = 5 + 6 + 4
     assert (
         sum(groups["core"].values())
         <= 33_686 + g18_core_allowance + interactive_startup_allowance + lmux_owned_core_allowance
         + capability_projection_allowance + lmux_default_owned_allowance
+        + canonical_legacy_enrollment_allowance
     ), groups["core"]
     assert sum(groups["g10"].values()) <= 1_800, groups["g10"]
     # Preserve main's optional execution projection allowance.

tests/coding/test_hosted_legacy_evidence.py

@@ -54,7 +54,7 @@ def _private_environment(root):
     # Bootstrap model/config lookup needs an OS home even for cwd sessions.
     # Do not read the actual user's home via POSIX pwd or Windows USERPROFILE.
     home = root / "user-home"
-    home.mkdir(exist_ok=True)
+    home.mkdir(mode=0o700, exist_ok=True)
     environment.update(HOME=str(home), USERPROFILE=str(home))
     return environment
 
@@ -102,7 +102,7 @@ def test_G17_TERMINAL_LEGACY_installed_profiles_and_embedded_startup(
     record_testsuite_property("native_platform", sys.platform)
     roots = {name: tmp_path / name for name in ("local", "g14", "embedded")}
     for root in roots.values():
-        root.mkdir()
+        root.mkdir(mode=0o700)
     with patch.dict(os.environ, _private_environment(roots["local"]), clear=True):
         _local(roots["local"], record_testsuite_property)
     # G14's older helper inherits cwd/environment; confine it without changing

tests/coding/test_owned_session_runtime.py

@@ -94,6 +94,67 @@ async def scenario():
         restored.close()
 
 
+def test_default_runtime_enrolls_canonical_legacy_store_with_shared_assets(
+    tmp_path, monkeypatch,
+):
+    platform = tmp_path / "platform"
+    monkeypatch.setenv("LOUSHANG_HOME", str(platform))
+    tmp_path.chmod(0o700)
+    platform.mkdir(mode=0o700)
+    (platform / "data").mkdir(mode=0o700)
+    session_root = platform / "data/sessions"
+    session_root.mkdir(mode=0o700)
+    assets = platform / "data/session-assets"
+    assets.mkdir(mode=0o700)
+    legacy_asset = assets / "legacy-asset"
+    legacy_asset.write_bytes(b"preserve me")
+    (platform / "state").mkdir(mode=0o700)
+    stores = platform / "state/session-stores"
+    stores.mkdir(mode=0o700)
+    (stores / "unrelated-incomplete").mkdir(mode=0o700)
+
+    async def scenario():
+        selected = create_agent_session_runtime(
+            session_dir=session_root,
+            model=_model(),
+            persist=True,
+            no_tools=True,
+        )
+        try:
+            session = await selected.create_session(cwd=str(tmp_path))
+            await session.session_manager.append_message(
+                UserMessage(role="user", content="upgraded store", timestamp=1),
+            )
+            assert session.session_manager.session_file.parent == session_root
+            assert legacy_asset.read_bytes() == b"preserve me"
+        finally:
+            await selected.dispose_session_runtime()
+
+    asyncio.run(scenario())
+
+
+def test_default_runtime_does_not_enroll_custom_legacy_store(tmp_path, monkeypatch):
+    monkeypatch.setenv("LOUSHANG_HOME", str(tmp_path / "platform"))
+    session_root = tmp_path / "custom-data/sessions"
+    session_root.mkdir(parents=True, mode=0o700)
+    (session_root.parent / "session-assets").mkdir(mode=0o700)
+
+    async def scenario():
+        selected = create_agent_session_runtime(
+            session_dir=session_root,
+            model=_model(),
+            persist=True,
+            no_tools=True,
+        )
+        try:
+            with pytest.raises(TranscriptWriterError, match="conflict"):
+                await selected.create_session(cwd=str(tmp_path))
+        finally:
+            await selected.dispose_session_runtime()
+
+    asyncio.run(scenario())
+
+
 def test_graph_index_unknown_close_keeps_writer_until_original_cleanup_settles(tmp_path, monkeypatch):
     from loushang.harness.journal._rooted_io import RootedFile