From b965a8ede51589df275e670512963e16198f710a Mon Sep 17 00:00:00 2001
From: JasonDeng <15301580353@163.com>
Date: Tue, 20 Oct 2026 10:56:34 +0800
Subject: [PATCH 1/3] fix(code): name a generation before its commit, so a lost
 directory stays reclaimable

---
 src/powercontext/builtin/code/cache.py   |  87 ++++++++++++++-
 src/powercontext/builtin/code/graph.py   |  21 +++-
 src/powercontext/builtin/code/seekdb.py  |  33 +++++-
 src/powercontext/builtin/code/service.py |  10 +-
 src/powercontext/builtin/code/store.py   |  13 ++-
 tests/builtin/test_code_publication.py   | 130 +++++++++++++++++++++++
 6 files changed, 286 insertions(+), 8 deletions(-)
 create mode 100644 tests/builtin/test_code_publication.py

diff --git a/src/powercontext/builtin/code/cache.py b/src/powercontext/builtin/code/cache.py
index fb179ccfe0..e58b90d4d7 100644
--- a/src/powercontext/builtin/code/cache.py
+++ b/src/powercontext/builtin/code/cache.py
@@ -125,6 +125,10 @@ def source(self, path: str, deadline: float) -> bytes:
 
 
 class GenerationCache:
+    #: Publication records are bookkeeping, not data: keeping a short window is
+    #: enough to reclaim an interrupted build and bounded on a long-lived binding.
+    PUBLICATION_WINDOW = 64
+
     def __init__(self, directory: Path, store: GraphStore | None = None) -> None:
         self.directory = directory
         self.store = store if store is not None else SQLiteGraphStore()
@@ -198,7 +202,83 @@ def staging(self) -> Path:
             (path / child).mkdir(mode=0o700)
         return path
 
-    def publish(self, staging: Path, manifest: dict[str, Any], deadline: float, max_bytes: int) -> Generation:
+    def publications(self) -> dict[str, dict[str, Any]]:
+        path = self.directory / "publication.json"
+        if not path.exists():
+            return {}
+        records = read_json(path, maximum=64 * 1024).get("attempts")
+        if not isinstance(records, dict):
+            raise CodeError("index_integrity_failed")
+        return {str(attempt): record for attempt, record in records.items() if isinstance(record, dict)}
+
+    def begin_publication(self, staging: Path) -> str:
+        """Name the attempt before the store writes any row under it.
+
+        The identifier has to come from here rather than the store: a commit that
+        has to be recalled later must be nameable even when the directory that
+        carried its name is gone.
+        """
+        attempt = uuid.uuid4().hex
+        records = self.publications()
+        records[attempt] = {"directory": staging.name, "state": "pending", "started_at": time.time()}
+        self._write_publications(records)
+        return attempt
+
+    def complete_publication(self, attempt: str, directory: Path) -> None:
+        records = self.publications()
+        # The generation is already reachable at this point, so a missing record is
+        # repaired rather than raised: losing a name must not fail a live publication.
+        record = records.setdefault(
+            attempt, {"started_at": time.time(), "state": "pending", "directory": directory.name}
+        )
+        record.update({"directory": directory.name, "state": "published"})
+        self._write_publications(records)
+
+    def recover(self, deadline: float) -> int:
+        """Reclaim commits whose generation directory no longer exists.
+
+        ``collect()`` starts from directory names, so it cannot see a commit whose
+        directory was removed out of band — a wiped cache, a pruned staging area.
+        The records are the only place that still knows the identifier.
+        """
+        records = self.publications()
+        abandoned = [
+            attempt
+            for attempt, record in records.items()
+            if not (self.directory / str(record.get("directory", ""))).exists()
+        ]
+        for attempt in abandoned:
+            self.store.remove_generation(attempt, self.directory.name, deadline)
+            records.pop(attempt)
+        if abandoned:
+            self._write_publications(records)
+        return len(abandoned)
+
+    def _write_publications(self, records: dict[str, dict[str, Any]]) -> None:
+        if len(records) > self.PUBLICATION_WINDOW:
+            ordered = sorted(records.items(), key=lambda item: item[1].get("started_at", 0))
+            records = dict(ordered[-self.PUBLICATION_WINDOW :])
+        self.replace_json("publication.json", {"attempts": records})
+
+    def _forget_collected(self) -> None:
+        records = self.publications()
+        live = {
+            attempt: record
+            for attempt, record in records.items()
+            if (self.directory / str(record.get("directory", ""))).exists()
+        }
+        if len(live) != len(records):
+            self._write_publications(live)
+
+    def publish(
+        self,
+        staging: Path,
+        manifest: dict[str, Any],
+        deadline: float,
+        max_bytes: int,
+        *,
+        attempt: str | None = None,
+    ) -> Generation:
         manifest_content = json_bytes(manifest)
         write_private(staging / "manifest.json", manifest_content)
         write_private(staging / "reader.lock", b"")
@@ -226,6 +306,10 @@ def publish(self, staging: Path, manifest: dict[str, Any], deadline: float, max_
             self.replace_json(
                 "current.json", {"current": reference, "previous": pointer["current"] if pointer else None}
             )
+            if attempt is not None:
+                # The rename is the moment the generation becomes reachable, so the
+                # record has to point at the final name before anything reads it.
+                self.complete_publication(attempt, directory)
             self.collect(self.pointer(), deadline)
         return Generation(directory, manifest, self.store)
 
@@ -284,3 +368,4 @@ def collect(self, pointer: dict[str, Any] | None, deadline: float) -> None:
             except CodeError as error:
                 if error.code != "code_cache_busy":
                     raise
+        self._forget_collected()
diff --git a/src/powercontext/builtin/code/graph.py b/src/powercontext/builtin/code/graph.py
index f0de02b6de..db3b33e03c 100644
--- a/src/powercontext/builtin/code/graph.py
+++ b/src/powercontext/builtin/code/graph.py
@@ -37,8 +37,16 @@ class GraphStore(Protocol):
     identity: str
 
     def create(
-        self, directory: Path, nodes: list[dict[str, Any]], edges: list[dict[str, Any]], deadline: float
-    ) -> dict[str, Any]: ...
+        self,
+        directory: Path,
+        nodes: list[dict[str, Any]],
+        edges: list[dict[str, Any]],
+        deadline: float,
+        *,
+        generation_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Write one generation; the caller may name it so a recall can address it later."""
+        ...
 
     def verify(self, directory: Path, manifest: dict[str, Any], deadline: float) -> None: ...
 
@@ -48,4 +56,13 @@ def open(
 
     def remove(self, directory: Path, deadline: float) -> None: ...
 
+    def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None:
+        """Drop a commit by identifier, for directories that no longer exist.
+
+        ``remove()`` takes a directory and reads the identifier out of it. When the
+        directory is already gone, this is the only entry point that can still
+        reach the rows.
+        """
+        ...
+
     def size(self, directory: Path, deadline: float) -> int: ...
diff --git a/src/powercontext/builtin/code/seekdb.py b/src/powercontext/builtin/code/seekdb.py
index 298e031e29..7938a670ff 100644
--- a/src/powercontext/builtin/code/seekdb.py
+++ b/src/powercontext/builtin/code/seekdb.py
@@ -241,10 +241,24 @@ def _reference(self, directory: Path) -> dict[str, Any]:
         return reference
 
     def create(
-        self, directory: Path, nodes: list[dict[str, Any]], edges: list[dict[str, Any]], deadline: float
+        self,
+        directory: Path,
+        nodes: list[dict[str, Any]],
+        edges: list[dict[str, Any]],
+        deadline: float,
+        *,
+        generation_id: str | None = None,
     ) -> dict[str, Any]:
         self.initialize(deadline)
-        reference = {"backend": self.identity, "binding": directory.parent.name, "id": uuid.uuid4().hex}
+        if generation_id is not None and (
+            len(generation_id) != 32 or any(character not in "0123456789abcdef" for character in generation_id)
+        ):
+            raise CodeError("index_integrity_failed")
+        reference = {
+            "backend": self.identity,
+            "binding": directory.parent.name,
+            "id": generation_id or uuid.uuid4().hex,
+        }
         # Persist the cleanup handle before any database commit can leave orphan rows.
         # A killed writer may leave the temporary file incomplete, but remove() must
         # only see a complete descriptor or no committed database work at all.
@@ -404,6 +418,21 @@ def remove(self, directory: Path, deadline: float) -> None:
             )
             connection.commit()
 
+    def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None:
+        """Delete a commit whose directory is gone; the identifier is the only handle left."""
+        try:
+            with self.connection(deadline) as connection, connection.cursor() as cursor:
+                for table in ("pc_code_edges", "pc_code_nodes"):
+                    check_deadline(deadline)
+                    cursor.execute(f"DELETE FROM {table} WHERE generation_id = %s", (generation_id,))  # noqa: S608 - constant tables.
+                cursor.execute(
+                    "DELETE FROM pc_code_generations WHERE id = %s AND binding = %s", (generation_id, binding)
+                )
+                connection.commit()
+        except pymysql.MySQLError as error:
+            if error.args[0] != 1146:  # This binding never created its tables.
+                raise
+
     def _incomplete_staging(self, directory: Path, deadline: float) -> bool:
         # Older builders wrote graph.json in place before opening the transaction.
         # Recover their interrupted writes without accepting corrupt published graphs
diff --git a/src/powercontext/builtin/code/service.py b/src/powercontext/builtin/code/service.py
index ac9bb6ea72..ce981a5e4f 100644
--- a/src/powercontext/builtin/code/service.py
+++ b/src/powercontext/builtin/code/service.py
@@ -90,6 +90,9 @@ def index(self, scope_id: str, *, full: bool = False) -> dict[str, Any]:
                 if abandoned.is_dir() and not abandoned.is_symlink():
                     self.store.remove(abandoned, deadline)
                     shutil.rmtree(abandoned)
+            # A staged build is not the only way to lose a directory: reclaim what the
+            # records still name, so a wiped generation does not keep paying for rows.
+            cache.recover(deadline)
             try:
                 result = self._build(repository, cache, binding, deadline, full=full)
                 result.update(status="ready", elapsed_seconds=time.monotonic() - started, finished_at=_now())
@@ -153,7 +156,8 @@ def _build(
                     attributes.update(edge_count=len(edges), diagnostic_count=sum(map(len, diagnostics.values())))
                 nodes = [node for fact in facts for node in fact["nodes"]]
                 with stage("store"):
-                    storage = self.store.create(staging, nodes, edges, deadline)
+                    publication = cache.begin_publication(staging)
+                    storage = self.store.create(staging, nodes, edges, deadline, generation_id=publication)
                 for path, issues in diagnostics.items():
                     key = entries[path]["extraction_key"]
                     content = json_bytes({"items": issues})
@@ -169,7 +173,9 @@ def _build(
                 manifest.update(storage)
                 manifest["previous_fingerprint"] = previous.fingerprint if previous else None
                 with stage("publish") as attributes:
-                    generation = cache.publish(staging, manifest, deadline, self.config.limits.max_cache_bytes)
+                    generation = cache.publish(
+                        staging, manifest, deadline, self.config.limits.max_cache_bytes, attempt=publication
+                    )
                     attributes.update(manifest["coverage"])
                     attributes.update(extracted_files=extracted, reused_files=len(entries) - extracted)
                 return {
diff --git a/src/powercontext/builtin/code/store.py b/src/powercontext/builtin/code/store.py
index 4a2079356d..8da270b62d 100644
--- a/src/powercontext/builtin/code/store.py
+++ b/src/powercontext/builtin/code/store.py
@@ -200,14 +200,25 @@ class SQLiteGraphStore:
     identity = "sqlite"
 
     def create(
-        self, directory: Path, nodes: list[dict[str, Any]], edges: list[dict[str, Any]], deadline: float
+        self,
+        directory: Path,
+        nodes: list[dict[str, Any]],
+        edges: list[dict[str, Any]],
+        deadline: float,
+        *,
+        generation_id: str | None = None,
     ) -> dict[str, Any]:
         from powercontext.builtin.code.cache import hash_file
 
+        del generation_id  # Rows live inside the generation directory, so no name is needed.
         path = directory / "graph.sqlite"
         create_graph(path, nodes, edges, deadline)
         return {"database_sha256": hash_file(path, deadline)}
 
+    def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None:
+        # A directory that no longer exists takes its SQLite rows with it.
+        del generation_id, binding, deadline
+
     def verify(self, directory: Path, manifest: dict[str, Any], deadline: float) -> None:
         from powercontext.builtin.code.cache import hash_file
 
diff --git a/tests/builtin/test_code_publication.py b/tests/builtin/test_code_publication.py
new file mode 100644
index 0000000000..cf552f9020
--- /dev/null
+++ b/tests/builtin/test_code_publication.py
@@ -0,0 +1,130 @@
+# Copyright (c) 2026 OceanBase.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+
+"""Publication records: the commit that outlives the directory which named it."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import time
+from pathlib import Path
+
+import pytest
+
+from powercontext.builtin.code import CodeConfig, CodeRepositoryConfig, CodeService
+from powercontext.builtin.code.cache import GenerationCache
+from powercontext.builtin.code.store import SQLiteGraphStore
+
+pytest.importorskip("tree_sitter")
+pytest.importorskip("tree_sitter_python")
+
+pytestmark = pytest.mark.skipif(
+    os.name != "posix", reason="native code indexing requires POSIX file locks and resource limits"
+)
+
+
+class RecordingStore(SQLiteGraphStore):
+    """A store that remembers which identifiers were recalled by name."""
+
+    def __init__(self) -> None:
+        self.reclaimed: list[str] = []
+
+    def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None:
+        self.reclaimed.append(generation_id)
+
+
+def git(root: Path, *arguments: str) -> None:
+    executable = shutil.which("git")
+    assert executable
+    subprocess.run([executable, "-C", str(root), *arguments], check=True, capture_output=True)
+
+
+def cache_of(service: CodeService) -> GenerationCache:
+    directories = [path for path in (service.config.cache_dir).iterdir() if path.is_dir()]
+    assert len(directories) == 1
+    return GenerationCache(directories[0], service.store)
+
+
+@pytest.fixture
+def repository(tmp_path):
+    root = tmp_path / "repository"
+    (root / "src").mkdir(parents=True)
+    (root / "src/app.py").write_text("def run():\n    return 1\n")
+    git(root, "init")
+    git(root, "add", ".")
+    git(root, "-c", "user.name=Code Test", "-c", "user.email=test@example.invalid", "commit", "-m", "fixture")
+    store = RecordingStore()
+    service = CodeService(
+        CodeConfig(
+            enabled=True,
+            repositories={"scope": CodeRepositoryConfig(root=root)},
+            cache_dir=tmp_path / "cache",
+        ),
+        store=store,
+    )
+    service.index("scope", full=True)
+    return service, root
+
+
+def test_a_published_generation_is_recorded_under_its_final_directory(repository):
+    service, _ = repository
+    cache = cache_of(service)
+
+    records = cache.publications()
+    pointer = cache.pointer()
+    assert pointer is not None
+    assert len(records) == 1
+    _, record = records.popitem()
+    assert record["state"] == "published"
+    assert record["directory"] == pointer["current"]["directory"]
+
+
+def test_reindexing_keeps_a_live_generation_out_of_the_sweep(repository):
+    service, root = repository
+    (root / "src/app.py").write_text("def run():\n    return 2\n")
+    git(root, "add", ".")
+
+    service.index("scope")
+
+    assert service.store.reclaimed == []
+    assert [record["state"] for record in cache_of(service).publications().values()] == ["published", "published"]
+
+
+def test_a_generation_directory_removed_out_of_band_is_reclaimed_by_name(repository):
+    service, root = repository
+    cache = cache_of(service)
+    attempt = next(iter(cache.publications()))
+    pointer = cache.pointer()
+    assert pointer is not None
+    shutil.rmtree(cache.directory / pointer["current"]["directory"])
+    (root / "src/app.py").write_text("def run():\n    return 3\n")
+    git(root, "add", ".")
+
+    service.index("scope")
+
+    assert service.store.reclaimed == [attempt]
+    assert attempt not in cache.publications()
+
+
+def test_a_pending_attempt_without_its_staging_directory_is_reclaimed(repository):
+    service, _ = repository
+    cache = cache_of(service)
+    attempt = "0" * 32
+    journal = json.loads((cache.directory / "publication.json").read_text())
+    journal["attempts"][attempt] = {"directory": "staging-nowhere", "state": "pending", "started_at": 0}
+    (cache.directory / "publication.json").write_text(json.dumps(journal))
+
+    reclaimed = cache.recover(time.monotonic() + 30)
+
+    assert reclaimed == 1
+    assert service.store.reclaimed == [attempt]
+    assert attempt not in cache.publications()

From 5d26c4c966e320f9abf4f9a305bd7f931a022bdd Mon Sep 17 00:00:00 2001
From: JasonDeng <15301580353@163.com>
Date: Tue, 20 Oct 2026 11:21:56 +0800
Subject: [PATCH 2/3] test(code): forward the caller-supplied generation id in
 the interrupted-build harness

---
 tests/e2e/test_native_code_seekdb.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/e2e/test_native_code_seekdb.py b/tests/e2e/test_native_code_seekdb.py
index 4aa874f491..1836171348 100644
--- a/tests/e2e/test_native_code_seekdb.py
+++ b/tests/e2e/test_native_code_seekdb.py
@@ -294,8 +294,8 @@ def interrupted_write(path, content):
     seekdb.write_private = interrupted_write
 else:
     create = seekdb.SeekDBGraphStore.create
-    def interrupted_create(self, *args):
-        result = create(self, *args)
+    def interrupted_create(self, *args, **kwargs):
+        result = create(self, *args, **kwargs)
         pause()
         return result
     seekdb.SeekDBGraphStore.create = interrupted_create

From 1072176e99b52901e5561875fe3ef7d0e15360e0 Mon Sep 17 00:00:00 2001
From: JasonDeng <15301580353@163.com>
Date: Tue, 20 Oct 2026 21:35:55 +0800
Subject: [PATCH 3/3] fix(code): decide a recall by the rows, not by the
 directory

Review found four ways the publication record still lost or destroyed data.
All four are the same mistake in different places: the record was treated as
bookkeeping about directories, when it is the only handle to rows that can
outlive the directory which carried them.

Recovery deleted a live generation. The record was updated after
``os.replace``, so a crash in between left it pointing at a staging directory
that had already been renamed away. ``recover()`` read "directory missing" and
deleted the rows of the generation ``current.json`` was serving, including one
pinned by an active reader. The record now names the final directory before
the rename, which makes the window one-sided: a record can only refer to a
directory that was never created, whose rows are the ones this attempt
committed. Recovery additionally refuses to reclaim an identifier that a
reachable generation's manifest still names, so a record left by an older
writer is not trusted either.

``clear()`` dropped the handle instead of settling it. Records for missing
directories were simply forgotten, so the rows stayed charged to the binding
with nothing left that could name them. Collection now starts no reclamation
at all; ``recover()`` reclaims and only then forgets, and ``clear()`` runs it
after unpublishing.

The record window evicted live handles. Capping at 64 by age discarded the
only handle to a generation that ``collect()`` could not retire because a
reader held it. The cap now retires commits instead of records: entries whose
directory still exists are never dropped, and the excess is allowed to stand.

A commit could not be recalled before its tables existed. ``connection()``
turns MySQL errors into ``CodeError`` on exit, so the outer handler never saw
1146 and every retry failed during recovery, before ``create()`` could create
the tables it was waiting for. The missing-table case is handled inside the
connection context.

The tests assert the name-to-row correspondence they were missing: the store
double records the identifier handed to ``create()`` and the manifest is read
back, and the SeekDB e2e checks that the committed row is the recorded one.
Each of the four new cases fails on the previous commit.
---
 src/powercontext/builtin/code/cache.py   | 120 ++++++++++-----
 src/powercontext/builtin/code/seekdb.py  |  14 +-
 src/powercontext/builtin/code/service.py |   2 +-
 tests/builtin/test_code_publication.py   | 188 ++++++++++++++++++++++-
 tests/e2e/test_native_code_seekdb.py     |  18 +++
 5 files changed, 297 insertions(+), 45 deletions(-)

diff --git a/src/powercontext/builtin/code/cache.py b/src/powercontext/builtin/code/cache.py
index e58b90d4d7..6b3da96b8e 100644
--- a/src/powercontext/builtin/code/cache.py
+++ b/src/powercontext/builtin/code/cache.py
@@ -125,8 +125,9 @@ def source(self, path: str, deadline: float) -> bytes:
 
 
 class GenerationCache:
-    #: Publication records are bookkeeping, not data: keeping a short window is
-    #: enough to reclaim an interrupted build and bounded on a long-lived binding.
+    #: How many publication records a binding normally carries. The bound is
+    #: enforced by recalling the commits it retires, never by forgetting a handle,
+    #: so a binding that keeps many generations pinned may stand above it.
     PUBLICATION_WINDOW = 64
 
     def __init__(self, directory: Path, store: GraphStore | None = None) -> None:
@@ -211,7 +212,7 @@ def publications(self) -> dict[str, dict[str, Any]]:
             raise CodeError("index_integrity_failed")
         return {str(attempt): record for attempt, record in records.items() if isinstance(record, dict)}
 
-    def begin_publication(self, staging: Path) -> str:
+    def begin_publication(self, staging: Path, deadline: float) -> str:
         """Name the attempt before the store writes any row under it.
 
         The identifier has to come from here rather than the store: a commit that
@@ -221,54 +222,87 @@ def begin_publication(self, staging: Path) -> str:
         attempt = uuid.uuid4().hex
         records = self.publications()
         records[attempt] = {"directory": staging.name, "state": "pending", "started_at": time.time()}
-        self._write_publications(records)
+        self._write_publications(records, deadline)
         return attempt
 
-    def complete_publication(self, attempt: str, directory: Path) -> None:
-        records = self.publications()
-        # The generation is already reachable at this point, so a missing record is
-        # repaired rather than raised: losing a name must not fail a live publication.
-        record = records.setdefault(
-            attempt, {"started_at": time.time(), "state": "pending", "directory": directory.name}
-        )
-        record.update({"directory": directory.name, "state": "published"})
-        self._write_publications(records)
-
     def recover(self, deadline: float) -> int:
         """Reclaim commits whose generation directory no longer exists.
 
         ``collect()`` starts from directory names, so it cannot see a commit whose
         directory was removed out of band — a wiped cache, a pruned staging area.
         The records are the only place that still knows the identifier.
+
+        A missing directory is not proof that the commit is unwanted. The pointer
+        can still reach a generation whose manifest names the identifier these rows
+        were committed under, and a reader may be holding it. Reclaiming on
+        directory absence alone deletes live data, so that name is checked first.
         """
         records = self.publications()
+        retained = self._retained_identifiers()
         abandoned = [
             attempt
             for attempt, record in records.items()
-            if not (self.directory / str(record.get("directory", ""))).exists()
+            if attempt not in retained and not (self.directory / self._recorded_directory(record)).exists()
         ]
         for attempt in abandoned:
             self.store.remove_generation(attempt, self.directory.name, deadline)
             records.pop(attempt)
-        if abandoned:
-            self._write_publications(records)
+        if abandoned or len(records) > self.PUBLICATION_WINDOW:
+            self._write_publications(records, deadline)
         return len(abandoned)
 
-    def _write_publications(self, records: dict[str, dict[str, Any]]) -> None:
-        if len(records) > self.PUBLICATION_WINDOW:
-            ordered = sorted(records.items(), key=lambda item: item[1].get("started_at", 0))
-            records = dict(ordered[-self.PUBLICATION_WINDOW :])
-        self.replace_json("publication.json", {"attempts": records})
+    def _retained_identifiers(self) -> set[str]:
+        """Identifiers the pointer still reaches, read from the generations themselves.
 
-    def _forget_collected(self) -> None:
-        records = self.publications()
-        live = {
-            attempt: record
-            for attempt, record in records.items()
-            if (self.directory / str(record.get("directory", ""))).exists()
-        }
-        if len(live) != len(records):
-            self._write_publications(live)
+        A row is addressed by its identifier, not by the directory that carries it,
+        so the manifest inside each reachable generation is the only place that can
+        say which rows a live generation owns. A record whose directory has gone can
+        still point at one of them — that is the crash this guard exists for.
+        """
+        try:
+            pointer = self.pointer()
+        except CodeError:
+            return set()
+        retained: set[str] = set()
+        for reference in (pointer or {}).values():
+            if not isinstance(reference, dict):
+                continue
+            directory = reference.get("directory")
+            if not isinstance(directory, str):
+                continue
+            try:
+                manifest = read_json(self.directory / directory / "manifest.json")
+            except (CodeError, OSError):
+                continue
+            graph = manifest.get("graph")
+            identifier = graph.get("id") if isinstance(graph, dict) else None
+            if isinstance(identifier, str):
+                retained.add(identifier)
+        return retained
+
+    @staticmethod
+    def _recorded_directory(record: dict[str, Any]) -> str:
+        return str(record.get("directory", ""))
+
+    def _write_publications(self, records: dict[str, dict[str, Any]], deadline: float) -> None:
+        excess = len(records) - self.PUBLICATION_WINDOW
+        if excess > 0:
+            # A record is the only handle to rows that outlive their directory, so the
+            # window recalls those commits instead of forgetting them. A record whose
+            # directory still exists is a generation a reader may hold; it is never
+            # dropped, and the excess is allowed to stand.
+            stale = sorted(
+                (
+                    (attempt, record)
+                    for attempt, record in records.items()
+                    if not (self.directory / self._recorded_directory(record)).exists()
+                ),
+                key=lambda item: item[1].get("started_at", 0),
+            )
+            for attempt, _ in stale[:excess]:
+                self.store.remove_generation(attempt, self.directory.name, deadline)
+                records.pop(attempt)
+        self.replace_json("publication.json", {"attempts": records})
 
     def publish(
         self,
@@ -301,15 +335,26 @@ def publish(
                 if size > max_bytes:
                     raise CodeError("code_cache_limit_exceeded")
             directory = self.directory / ("generation-" + uuid.uuid4().hex)
+            if attempt is not None:
+                # Name the generation before it exists. Recording the final name only
+                # after the rename leaves a window where the record still points at a
+                # staging directory that has already gone: recovery would then read
+                # "directory missing" and delete the rows of a live generation.
+                # Written first, the record can only be wrong in the harmless
+                # direction — it names a directory that was never created, and the
+                # rows it recalls are the ones this attempt committed.
+                records = self.publications()
+                records[attempt] = {
+                    "directory": directory.name,
+                    "state": "published",
+                    "started_at": records.get(attempt, {}).get("started_at", time.time()),
+                }
+                self._write_publications(records, deadline)
             os.replace(staging, directory)
             reference = {"directory": directory.name, "manifest_sha256": digest_bytes(manifest_content)}
             self.replace_json(
                 "current.json", {"current": reference, "previous": pointer["current"] if pointer else None}
             )
-            if attempt is not None:
-                # The rename is the moment the generation becomes reachable, so the
-                # record has to point at the final name before anything reads it.
-                self.complete_publication(attempt, directory)
             self.collect(self.pointer(), deadline)
         return Generation(directory, manifest, self.store)
 
@@ -353,6 +398,10 @@ def clear(self, deadline: float) -> dict[str, Any]:
             finally:
                 os.close(descriptor)
             self.collect(None, deadline)
+            # A directory that was removed out of band leaves nothing for collect() to
+            # start from; clearing is the last moment the binding is still named, so
+            # the records have to be settled here or their rows are charged forever.
+            self.recover(deadline)
             retained = sum(path.is_dir() and not path.is_symlink() for path in self.directory.glob("generation-*"))
             return {"status": "cleared", "retained_reader_generations": retained}
 
@@ -368,4 +417,3 @@ def collect(self, pointer: dict[str, Any] | None, deadline: float) -> None:
             except CodeError as error:
                 if error.code != "code_cache_busy":
                     raise
-        self._forget_collected()
diff --git a/src/powercontext/builtin/code/seekdb.py b/src/powercontext/builtin/code/seekdb.py
index 7938a670ff..c46b391011 100644
--- a/src/powercontext/builtin/code/seekdb.py
+++ b/src/powercontext/builtin/code/seekdb.py
@@ -420,8 +420,8 @@ def remove(self, directory: Path, deadline: float) -> None:
 
     def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None:
         """Delete a commit whose directory is gone; the identifier is the only handle left."""
-        try:
-            with self.connection(deadline) as connection, connection.cursor() as cursor:
+        with self.connection(deadline) as connection, connection.cursor() as cursor:
+            try:
                 for table in ("pc_code_edges", "pc_code_nodes"):
                     check_deadline(deadline)
                     cursor.execute(f"DELETE FROM {table} WHERE generation_id = %s", (generation_id,))  # noqa: S608 - constant tables.
@@ -429,9 +429,13 @@ def remove_generation(self, generation_id: str, binding: str, deadline: float) -
                     "DELETE FROM pc_code_generations WHERE id = %s AND binding = %s", (generation_id, binding)
                 )
                 connection.commit()
-        except pymysql.MySQLError as error:
-            if error.args[0] != 1146:  # This binding never created its tables.
-                raise
+            except pymysql.MySQLError as error:
+                # A binding whose first build was interrupted before initialize() has no
+                # tables yet, so nothing can be committed under this identifier. Handled
+                # inside the context because connection() turns MySQL errors into
+                # CodeError on exit, which would fail the retry this call exists to allow.
+                if error.args[0] != 1146:
+                    raise
 
     def _incomplete_staging(self, directory: Path, deadline: float) -> bool:
         # Older builders wrote graph.json in place before opening the transaction.
diff --git a/src/powercontext/builtin/code/service.py b/src/powercontext/builtin/code/service.py
index ce981a5e4f..afa9f8997e 100644
--- a/src/powercontext/builtin/code/service.py
+++ b/src/powercontext/builtin/code/service.py
@@ -156,7 +156,7 @@ def _build(
                     attributes.update(edge_count=len(edges), diagnostic_count=sum(map(len, diagnostics.values())))
                 nodes = [node for fact in facts for node in fact["nodes"]]
                 with stage("store"):
-                    publication = cache.begin_publication(staging)
+                    publication = cache.begin_publication(staging, deadline)
                     storage = self.store.create(staging, nodes, edges, deadline, generation_id=publication)
                 for path, issues in diagnostics.items():
                     key = entries[path]["extraction_key"]
diff --git a/tests/builtin/test_code_publication.py b/tests/builtin/test_code_publication.py
index cf552f9020..fa34e3773c 100644
--- a/tests/builtin/test_code_publication.py
+++ b/tests/builtin/test_code_publication.py
@@ -33,15 +33,61 @@
 
 
 class RecordingStore(SQLiteGraphStore):
-    """A store that remembers which identifiers were recalled by name."""
+    """A store that remembers which identifiers were committed and which were recalled.
+
+    Both lists matter: recalling a name only reclaims the rows if they were
+    committed under that same name. A test that checks `reclaimed` alone passes
+    even when the store ignores the name it was given.
+    """
 
     def __init__(self) -> None:
+        self.created: list[str | None] = []
         self.reclaimed: list[str] = []
 
+    def create(
+        self,
+        directory: Path,
+        nodes: list[dict[str, object]],
+        edges: list[dict[str, object]],
+        deadline: float,
+        *,
+        generation_id: str | None = None,
+    ) -> dict[str, object]:
+        self.created.append(generation_id)
+        return super().create(directory, nodes, edges, deadline, generation_id=generation_id)
+
     def remove_generation(self, generation_id: str, binding: str, deadline: float) -> None:
         self.reclaimed.append(generation_id)
 
 
+class NamedStore(RecordingStore):
+    """Behaves like the SeekDB backend where it matters: the manifest carries a name.
+
+    SQLite rows live inside the generation directory, so losing the directory loses
+    the rows and there is nothing to protect. The invariant this file is about only
+    has consequences when the rows outlive the directory, which is the shape here.
+    """
+
+    def create(
+        self,
+        directory: Path,
+        nodes: list[dict[str, object]],
+        edges: list[dict[str, object]],
+        deadline: float,
+        *,
+        generation_id: str | None = None,
+    ) -> dict[str, object]:
+        reference = super().create(directory, nodes, edges, deadline, generation_id=generation_id)
+        return {
+            "graph": {
+                "backend": "named",
+                "binding": directory.parent.name,
+                "id": generation_id,
+                **{key: value for key, value in reference.items() if key == "database_sha256"},
+            }
+        }
+
+
 def git(root: Path, *arguments: str) -> None:
     executable = shutil.which("git")
     assert executable
@@ -55,14 +101,14 @@ def cache_of(service: CodeService) -> GenerationCache:
 
 
 @pytest.fixture
-def repository(tmp_path):
+def repository(tmp_path, request):
     root = tmp_path / "repository"
     (root / "src").mkdir(parents=True)
     (root / "src/app.py").write_text("def run():\n    return 1\n")
     git(root, "init")
     git(root, "add", ".")
     git(root, "-c", "user.name=Code Test", "-c", "user.email=test@example.invalid", "commit", "-m", "fixture")
-    store = RecordingStore()
+    store = getattr(request, "param", RecordingStore)()
     service = CodeService(
         CodeConfig(
             enabled=True,
@@ -128,3 +174,139 @@ def test_a_pending_attempt_without_its_staging_directory_is_reclaimed(repository
     assert reclaimed == 1
     assert service.store.reclaimed == [attempt]
     assert attempt not in cache.publications()
+
+
+@pytest.mark.parametrize("repository", [NamedStore], indirect=True)
+def test_the_recorded_name_is_the_identifier_the_store_committed(repository):
+    """The name in the record has to be the name the rows were written under.
+
+    Recalling a name the store never used deletes nothing and raises nothing, so
+    the leak this change exists to fix would survive every other assertion here.
+    """
+
+    service, _ = repository
+    cache = cache_of(service)
+    attempt = next(iter(cache.publications()))
+    pointer = cache.pointer()
+    assert pointer is not None
+    manifest = json.loads(
+        (cache.directory / pointer["current"]["directory"] / "manifest.json").read_text(encoding="utf-8")
+    )
+
+    assert service.store.created == [attempt]
+    assert manifest["graph"]["id"] == attempt
+
+
+@pytest.mark.parametrize("repository", [NamedStore], indirect=True)
+def test_recovery_cannot_reclaim_rows_a_reachable_generation_still_owns(repository):
+    """A record naming a moved-away staging directory must not decide a live commit.
+
+    This is the state a crash between the rename and the journal update used to
+    leave behind: `current.json` reaches a generation whose manifest owns the rows,
+    while the record still names the staging directory that no longer exists.
+    """
+
+    service, _ = repository
+    cache = cache_of(service)
+    attempt = next(iter(cache.publications()))
+    pointer = cache.pointer()
+    assert pointer is not None
+    journal = json.loads((cache.directory / "publication.json").read_text())
+    journal["attempts"][attempt] = {"directory": "staging-renamed-away", "state": "pending", "started_at": 0}
+    (cache.directory / "publication.json").write_text(json.dumps(journal))
+
+    reclaimed = cache.recover(time.monotonic() + 30)
+
+    assert reclaimed == 0
+    assert service.store.reclaimed == []
+    assert attempt in cache.publications()
+
+
+def test_clear_settles_a_record_whose_directory_went_missing(repository):
+    """Clearing used to forget the handle and leave the rows charged to the binding.
+
+    Directory absence is not evidence that collection reclaimed the commit: the
+    record is the only thing that still names the rows, so clearing has to settle
+    it rather than drop it.
+    """
+
+    service, _ = repository
+    cache = cache_of(service)
+    attempt = next(iter(cache.publications()))
+    pointer = cache.pointer()
+    assert pointer is not None
+    shutil.rmtree(cache.directory / pointer["current"]["directory"])
+
+    service.clear("scope")
+
+    assert service.store.reclaimed == [attempt]
+    assert cache.publications() == {}
+
+
+def test_the_record_window_never_drops_a_generation_that_still_exists(repository):
+    """The bound may not discard the only handle to rows that can still be recalled.
+
+    A generation pinned by a reader keeps its directory, so `collect()` cannot
+    retire it. If the window evicted its record anyway, the rows would be
+    unreachable the moment the directory went away.
+    """
+
+    service, _ = repository
+    cache = cache_of(service)
+    kept = [f"{index:032x}" for index in range(70)]
+    abandoned = [f"{index:032x}" for index in range(1000, 1006)]
+    for name in kept:
+        (cache.directory / ("generation-" + name)).mkdir()
+    attempts = {
+        **{name: {"directory": "generation-" + name, "state": "published", "started_at": 1} for name in kept},
+        **{name: {"directory": "generation-" + name, "state": "published", "started_at": 0} for name in abandoned},
+    }
+    (cache.directory / "publication.json").write_text(json.dumps({"attempts": attempts}))
+
+    cache.begin_publication(cache.staging(), time.monotonic() + 30)
+
+    records = cache.publications()
+    assert sorted(service.store.reclaimed) == sorted(abandoned)
+    assert all(name in records for name in kept)
+
+
+def test_a_commit_can_be_recalled_before_its_tables_exist(tmp_path, monkeypatch):
+    """The first build can name its commit and die before initialize() creates tables.
+
+    ``connection()`` converts every MySQL error into ``CodeError`` on exit, so the
+    outer handler never saw 1146 and every retry failed during recovery — before
+    ``create()`` could create the tables it was waiting for. A missing table means
+    there is nothing to recall, not that the recall failed.
+    """
+
+    import pymysql
+
+    from powercontext.builtin.code.seekdb import SeekDBGraphStore
+
+    class Cursor:
+        def __enter__(self) -> Cursor:
+            return self
+
+        def __exit__(self, *arguments: object) -> bool:
+            return False
+
+        def execute(self, *arguments: object) -> None:
+            raise pymysql.err.ProgrammingError(1146, "Table 'test.pc_code_edges' doesn't exist")
+
+    class Connection:
+        def __enter__(self) -> Connection:
+            return self
+
+        def __exit__(self, *arguments: object) -> bool:
+            return False
+
+        def begin(self) -> None: ...
+        def rollback(self) -> None: ...
+        def commit(self) -> None: ...
+        def cursor(self) -> Cursor:
+            return Cursor()
+
+    monkeypatch.setattr(pymysql, "connect", lambda **options: Connection())
+    store = SeekDBGraphStore(tmp_path / "seekdb", {})
+
+    store.remove_generation("a" * 32, "binding", time.monotonic() + 30)
diff --git a/tests/e2e/test_native_code_seekdb.py b/tests/e2e/test_native_code_seekdb.py
index 1836171348..b61d442575 100644
--- a/tests/e2e/test_native_code_seekdb.py
+++ b/tests/e2e/test_native_code_seekdb.py
@@ -339,6 +339,24 @@ def interrupted_create(self, *args, **kwargs):
             assert cursor.fetchone()[0] == 0
 
 
+def test_a_record_names_the_row_the_commit_wrote(repository):
+    """The journal key has to be the identifier the rows were committed under.
+
+    Recall deletes by name. If the store ignored the name it was given and minted
+    its own, the delete would match no rows and raise nothing, so the leak would be
+    back with every other test here still passing.
+    """
+
+    _, service = repository
+    directory = next(service.config.cache_dir.iterdir())
+    cache = GenerationCache(directory, service.store)
+    pointer = cache.pointer()
+    assert pointer is not None
+    manifest = json.loads((directory / pointer["current"]["directory"] / "manifest.json").read_text())
+
+    assert manifest["graph"]["id"] in cache.publications()
+
+
 @pytest.mark.parametrize("content", (b"", b'{"backend":'))
 @pytest.mark.parametrize("recovery", ("sync", "index", "clear"))
 def test_legacy_incomplete_staging_descriptor_recovers(repository, content, recovery):