fix(sync): retire the incomplete-run markers of runs a full rebuild supersedes

codebase_rag/cli.py

@@ -81,6 +81,8 @@
 from .sync_guard import clear_sync_incomplete as _clear_sync_incomplete
 from .sync_guard import partial_graph_refusal as _partial_graph_refusal
 from .sync_guard import project_owner_refusal as _project_owner_refusal
+from .sync_guard import retire_stranded_runs as _retire_stranded_runs
+from .sync_guard import stranded_runs as _stranded_runs
 from .sync_guard import sync_marker_params as _sync_marker_params
 from .sync_guard import sync_owner as _sync_owner
 from .sync_lock import SyncLockError, repo_sync_lock
@@ -1107,16 +1109,17 @@ def _sweep_glosses_after_failed_delete(
 
 
 def _run_updater_deferring_interrupt(
-    updater: "GraphUpdater",
+    updater: "GraphUpdater", *, rebuild: bool = False
 ) -> KeyboardInterrupt | None:
     """Run the sync's graph update; return a Ctrl+C that landed after its commit.
 
     The run stamps the sync time as part of its commit (#2444), and the
     caller clears its marker before re-raising what this returns; a Ctrl+C
-    before the commit raises `SyncInterrupted` instead.
+    before the commit raises `SyncInterrupted` instead. `rebuild` forces a
+    full re-index over runs another writer left part-way.
     """
     try:
-        updater.run()
+        updater.run(force=rebuild)
     except ex.EmbeddingsInterrupted as stop:
         # Raised only after the run committed, so the graph is whole and
         # the sync is recorded like any other; the interrupt then ends
@@ -1186,6 +1189,15 @@ def _run_graph_sync(
             # hits and can map onto unrelated nodes in the rebuilt graph.
             clear_all_embeddings()
 
+        # Read before this sync marks: the runs whose markers its rebuild
+        # supersedes (review of PR #3493).
+        stranded = _stranded_runs(ingestor, project_name, cs.CLI_SYNC_RUN_ID)
+        if stranded:
+            logger.warning(
+                ls.REBUILD_OVER_STRANDED_RUNS.format(
+                    project=project_name, count=len(stranded)
+                )
+            )
         # After the wipe, which would delete the marker with everything else,
         # and before `ensure_constraints`, whose migration can already purge.
         _mark_sync_incomplete(ingestor, project_name, owner, owned_roots)
@@ -1205,8 +1217,9 @@ def _run_graph_sync(
             capture=_capture_selection(capture),
             skip_embeddings=skip_embeddings,
         )
-        interrupted = _run_updater_deferring_interrupt(updater)
+        interrupted = _run_updater_deferring_interrupt(updater, rebuild=bool(stranded))
         _clear_sync_incomplete(ingestor, project_name)
+        _retire_stranded_runs(ingestor, project_name, stranded)
         # Taken before the export: counting it made a no-op sync of a small
         # repo report the time to dump the whole shared graph (issue #2440).
         elapsed = time.monotonic() - elapsed

codebase_rag/constants/graph.py

@@ -283,7 +283,9 @@ class GoImplementersUnknown(StrEnum):
 # Who wrote the CLI sync's incomplete-run marker (PR #2532 review): the host
 # (machine and pid namespace) and pid of its process, and a token unique to
 # that one sync, so `delete-project` clears it only once that sync provably
-# stopped, and only if no newer sync has re-marked it since.
+# stopped, and only if no newer sync has re-marked it since. The MCP
+# server's and `cgr check`'s markers name the host and pid too, so a full
+# rebuild retires one only once its process exited (review of PR #3496).
 KEY_OWNER_HOST = "owner_host"
 KEY_OWNER_PID = "owner_pid"
 KEY_OWNER_TOKEN = "owner_token"

codebase_rag/cypher_queries.py

@@ -198,6 +198,13 @@
 CYPHER_MARK_CLI_SYNC_INCOMPLETE = (
     CYPHER_MARK_PROJECT_INCOMPLETE + ", " + _CLI_SYNC_MARKER_OWNER_SET
 )
+# The MCP server's runs and `cgr check` name their process the same way
+# (review of PR #3496), so a full rebuild can tell a run whose process exited
+# from one still writing. No token: each of these runs has an id of its own.
+CYPHER_MARK_RUN_INCOMPLETE = (
+    CYPHER_MARK_PROJECT_INCOMPLETE
+    + ", m.owner_host = $owner_host, m.owner_pid = $owner_pid"
+)
 CYPHER_CLI_SYNC_MARKER_OWNER = (
     "MATCH (m:IncompleteRun {project: $project_name, run_id: $run_id}) "
     "RETURN m.owner_host AS owner_host, m.owner_pid AS owner_pid, "
@@ -260,6 +267,29 @@
     "RETURN coalesce(m.run_incomplete, false) AS run_incomplete, "
     "coalesce(m.writing, true) AS writing"
 )
+# The runs a full rebuild of the project supersedes (review of PR #3493): no
+# other path removes a marker its run left after its first write, so one
+# whose process died stayed down for good. Read before the rebuild marks,
+# with the process each marker names: `sync_guard.stranded_runs` keeps only
+# the runs whose process provably exited, since the checkout's sync lock
+# keeps out no writer of another checkout or CGR_HOME under the same project
+# name (review of PR #3496). A marker from before run ids existed is left
+# alone, as `CYPHER_CLEAR_LEGACY_PROJECT_INCOMPLETE` leaves it.
+CYPHER_STRANDED_RUN_IDS = (
+    "MATCH (m:IncompleteRun {project: $project_name}) "
+    "WHERE m.run_id IS NOT NULL AND m.run_id <> $run_id "
+    "AND coalesce(m.writing, true) "
+    "RETURN m.run_id AS run_id, m.owner_host AS owner_host, "
+    "m.owner_pid AS owner_pid"
+)
+# Deletes one run `CYPHER_STRANDED_RUN_IDS` returned, once the forced rebuild
+# that supersedes it has committed and cleared its own marker, and only while
+# the marker still names the exited process: the CLI sync's run id is shared,
+# so a CLI writer that started since has marked it under its own.
+CYPHER_RETIRE_STRANDED_RUN = (
+    "MATCH (m:IncompleteRun {project: $project_name, run_id: $run_id}) "
+    "WHERE m.owner_host = $owner_host AND m.owner_pid = $owner_pid DELETE m"
+)
 
 # Every project with an outstanding marker, whichever path wrote it: `cgr
 # status` reports them so a sync that was interrupted is visible there, not

codebase_rag/logs.py

@@ -1206,6 +1206,23 @@
     "The sync of {project} finished, but its incomplete-run marker could not "
     "be cleared: {error}. The graph is complete; the next sync clears it."
 )
+REBUILD_OVER_STRANDED_RUNS = (
+    "{count} earlier run(s) that wrote to project {project} stopped part-way, "
+    "so its graph may be partial. Re-indexing the whole repository instead of "
+    "only what changed."
+)
+STRANDED_RUNS_RETIRED = (
+    "Cleared {count} incomplete-run marker(s) the rebuild of {project} superseded."
+)
+STRANDED_RUNS_UNREADABLE = (
+    "Could not read which earlier runs left project {project} partial "
+    "({error}); their markers stay until a later full sync."
+)
+STRANDED_RUNS_NOT_RETIRED = (
+    "The rebuild of {project} finished, but the incomplete-run markers it "
+    "superseded could not be cleared: {error}. The graph is complete; the next "
+    "full sync clears them."
+)
 CLI_DELETE_MARKER_NOT_CLEARED = (
     "Project {project} was deleted, but its incomplete-run marker could not be "
     "cleared: {error}. cgr status may list it as interrupted until it is synced "

codebase_rag/mcp/tools.py

@@ -32,6 +32,7 @@
     run_to_text_answer,
 )
 from codebase_rag.services.provenance import head_commit
+from codebase_rag.sync_guard import retire_stranded_runs, run_owner, stranded_runs
 from codebase_rag.sync_lock import repo_sync_lock
 from codebase_rag.tools import tool_descriptions as td
 from codebase_rag.tools.ast_grep_service import AstGrepService
@@ -1505,6 +1506,11 @@ def _index_repository_sync(self) -> str:
         # a marker saying so lets a fresh process clear it and carry on rather
         # than refuse until a full update (#1705 review). The phase advances
         # to writing immediately before `delete_project`.
+        #
+        # The runs whose markers this re-index supersedes are read first
+        # (review of PR #3493). It deletes the project before rebuilding it,
+        # so it is already the full rebuild they need.
+        stranded = stranded_runs(self.ingestor, project_name, self._run_id)
         if (refusal := self._require_marker(project_name, writing=False)) is not None:
             raise RuntimeError(refusal)
         self._cleanup_embeddings_then_begin_writing(project_name)
@@ -1536,6 +1542,7 @@ def _index_repository_sync(self) -> str:
         # (#1705 review).
         if (stuck := self._require_marker_cleared(project_name)) is not None:
             raise RuntimeError(stuck)
+        retire_stranded_runs(self.ingestor, project_name, stranded)
         self._live_updater = updater
 
         return cs.MCP_INDEX_SUCCESS_PROJECT.format(
@@ -1604,6 +1611,10 @@ def _update_repository_sync(self) -> str:
         # Aborts on failure because the marker is the only protection that
         # survives a restart; nothing has been touched yet, so stopping costs
         # nothing.
+        #
+        # The runs whose markers this update supersedes are read first, and
+        # force a full re-index (review of PR #3493).
+        stranded = stranded_runs(self.ingestor, project_name, self._run_id)
         if (refusal := self._require_marker(project_name)) is not None:
             raise RuntimeError(refusal)
         self.ingestor.ensure_constraints()
@@ -1619,7 +1630,13 @@ def _update_repository_sync(self) -> str:
             exclude_paths=exclude_paths,
             project_name=project_name,
         )
-        updater.run()
+        if stranded:
+            logger.warning(
+                lg.REBUILD_OVER_STRANDED_RUNS.format(
+                    project=project_name, count=len(stranded)
+                )
+            )
+        updater.run(force=bool(stranded))
         self.ingestor.flush_all()
         # Cleared only after the flush: the marker must outlive any failure
         # that leaves batches uncommitted.
@@ -1632,6 +1649,7 @@ def _update_repository_sync(self) -> str:
         # (#1705 review).
         if (stuck := self._require_marker_cleared(project_name)) is not None:
             raise RuntimeError(stuck)
+        retire_stranded_runs(self.ingestor, project_name, stranded)
         self._live_updater = updater
         return cs.MCP_UPDATE_SUCCESS.format(path=self.project_root)
 
@@ -1696,7 +1714,10 @@ def _persist_incomplete(
             cs.KEY_RUN_ID: self._run_id,
         }
         if incomplete:
-            query = cq.CYPHER_MARK_PROJECT_INCOMPLETE
+            # Naming this process lets a full rebuild retire the marker once
+            # it has exited without clearing it (review of PR #3496).
+            query = cq.CYPHER_MARK_RUN_INCOMPLETE
+            params |= run_owner()
             params[cs.KEY_WRITING] = writing
         else:
             query = cq.CYPHER_CLEAR_PROJECT_INCOMPLETE

codebase_rag/structural_check.py

@@ -38,6 +38,7 @@
 from .graph_updater import GraphUpdater, _load_exclusion_state, _load_project_stamps
 from .services import QueryingIngestorProtocol
 from .structural_delta import StructuralDelta, normalise_paths, observe
+from .sync_guard import run_owner
 from .types_defs import LanguageQueries, PropertyDict, ReingestReport
 from .utils.path_utils import derive_project_name, should_skip_path
 
@@ -393,7 +394,8 @@ def _measure_then_restore(
         cs.KEY_RUN_ID: uuid.uuid4().hex,
     }
     store.execute_write(
-        cq.CYPHER_MARK_PROJECT_INCOMPLETE, {**marker, cs.KEY_WRITING: True}
+        cq.CYPHER_MARK_RUN_INCOMPLETE,
+        {**marker, **run_owner(), cs.KEY_WRITING: True},
     )
     try:
         return measure(lambda: apply(lambda: guard.capture(updater.reingest_scope)))

codebase_rag/sync_guard.py

@@ -24,8 +24,12 @@
 from . import cypher_queries as cq
 from . import logs as ls
 from .services import QueryProtocol
-from .types_defs import PropertyValue
-from .utils.process_owner import process_host
+from .types_defs import PropertyValue, ResultRow
+from .utils.process_owner import (
+    pid_has_exited,
+    process_host,
+    process_table_identified,
+)
 
 
 class SyncRefusedError(RuntimeError):
@@ -282,3 +286,98 @@ def partial_graph_refusal(ingestor: QueryProtocol, project_name: str) -> str | N
     ):
         return cs.CLI_ERR_EDIT_GRAPH_PARTIAL.format(project=project_name)
     return None
+
+
+class StrandedRun(NamedTuple):
+    """A run whose marker a full rebuild retires, and the exited process it names."""
+
+    run_id: str
+    owner_host: str
+    owner_pid: int
+
+
+def run_owner() -> dict[str, PropertyValue]:
+    """The process an MCP or `cgr check` run's marker names (review of PR #3496)."""
+    return {cs.KEY_OWNER_HOST: process_host(), cs.KEY_OWNER_PID: os.getpid()}
+
+
+def _stopped_run(row: ResultRow) -> StrandedRun | None:
+    """The run a marker names if its process provably exited.
+
+    Only a process in this process table can be checked, and only where the
+    host's name tells this table from every other (`process_table_identified`):
+    a hostname alone may be another machine's, whose pid is no process here.
+    A marker from another host, from a live process (this one included), or
+    from before markers named their process is kept: it may guard a graph its
+    run is still writing, as `cli._stopped_sync_token` reasons.
+    """
+    run_id = row.get(cs.KEY_RUN_ID)
+    pid = row.get(cs.KEY_OWNER_PID)
+    here = process_host()
+    if (
+        not isinstance(run_id, str)
+        or not process_table_identified()
+        or row.get(cs.KEY_OWNER_HOST) != here
+        or not isinstance(pid, int)
+        or pid == os.getpid()
+        or not pid_has_exited(pid)
+    ):
+        return None
+    return StrandedRun(run_id, here, pid)
+
+
+def stranded_runs(
+    ingestor: QueryProtocol, project_name: str, own_run_id: str
+) -> list[StrandedRun]:
+    """The other runs whose markers a full rebuild of the project supersedes.
+
+    Asked before the rebuild marks. A run counts only once its process has
+    exited, whichever checkout it wrote from: the checkout's sync lock keeps
+    out only the writers of this checkout under this CGR_HOME, and a run of
+    the same project elsewhere may still be writing (review of PR #3496). The
+    rebuild must then be forced: a stopped re-ingest may have deleted an
+    unchanged dependent's definitions, which an incremental run skips by its
+    saved hash, as `cgr watch`'s `_marked_run` says. A store that cannot
+    answer yields none, so the rebuild runs as before and those markers stay
+    for the next one.
+    """
+    try:
+        rows = ingestor.fetch_all(
+            cq.CYPHER_STRANDED_RUN_IDS,
+            {cs.KEY_PROJECT_NAME: project_name, cs.KEY_RUN_ID: own_run_id},
+        )
+    except Exception as exc:
+        logger.warning(
+            ls.STRANDED_RUNS_UNREADABLE.format(project=project_name, error=exc)
+        )
+        return []
+    return [run for row in rows if (run := _stopped_run(row)) is not None]
+
+
+def retire_stranded_runs(
+    ingestor: QueryProtocol, project_name: str, runs: list[StrandedRun]
+) -> None:
+    """Take off the markers of `runs` once a forced rebuild committed.
+
+    Never fails the rebuild, which is complete by now: a marker that stays
+    only refuses until the next full sync retires it.
+    """
+    if not runs:
+        return
+    try:
+        for run in runs:
+            ingestor.execute_write(
+                cq.CYPHER_RETIRE_STRANDED_RUN,
+                {
+                    cs.KEY_PROJECT_NAME: project_name,
+                    cs.KEY_RUN_ID: run.run_id,
+                    cs.KEY_OWNER_HOST: run.owner_host,
+                    cs.KEY_OWNER_PID: run.owner_pid,
+                },
+            )
+    except Exception as exc:
+        logger.warning(
+            ls.STRANDED_RUNS_NOT_RETIRED.format(project=project_name, error=exc)
+        )
+        return
+    logger.info(ls.STRANDED_RUNS_RETIRED.format(project=project_name, count=len(runs)))

codebase_rag/tests/test_check_isolation_marker.py

@@ -72,7 +72,7 @@ def test_the_marker_is_cleared_once_graph_and_cache_are_back(tmp_path: Path) ->
     _run(tmp_path, store)
 
     assert store.writes == [
-        cq.CYPHER_MARK_PROJECT_INCOMPLETE,
+        cq.CYPHER_MARK_RUN_INCOMPLETE,
         cq.CYPHER_CLEAR_PROJECT_INCOMPLETE,
     ]
 
@@ -90,4 +90,4 @@ def block_the_cache() -> None:
     with pytest.raises(OSError):
         _run(tmp_path, store, during=block_the_cache)
 
-    assert store.writes == [cq.CYPHER_MARK_PROJECT_INCOMPLETE]
+    assert store.writes == [cq.CYPHER_MARK_RUN_INCOMPLETE]