Keep conversation import recovery within its original source folders

src/backend/services/chat_history_service.py

@@ -4831,6 +4831,8 @@ def _event_fingerprint(entry: Mapping[str, Any]) -> str:
         == stored_manifest.get("package_sha256")
         and existing_manifest.get("parser_version")
         == stored_manifest.get("parser_version")
+        and existing_manifest.get("raw_file_copy_concept_id")
+        == stored_manifest.get("raw_file_copy_concept_id")
     ):
         existing_synchronisation = existing_manifest.get("synchronisation")
         existing_divergence = (

src/backend/services/external_conversation_bulk_import_service.py

@@ -234,9 +234,24 @@ def preview_local_conversation_import(
     }
 
 
+def _discovery_scope(providers: Sequence[Any] | None) -> str:
+    """Bind recovery to the creating process's server-configured directories."""
+    roots = [
+        (provider, root_id, str(path.expanduser().resolve()), pattern)
+        for provider in _normalise_providers(providers)
+        for root_id, path, pattern in _provider_roots(provider)
+    ]
+    return hashlib.sha256(repr(roots).encode("utf-8")).hexdigest()
+
+
 def _populate_batch(batch: Mapping[str, Any]) -> None:
     batches, items = _collections()
     batch_id = str(batch["batch_id"])
+    scope = batch.get("discovery_scope")
+    if scope and scope != _discovery_scope(batch.get("providers")):
+        # A worker sharing this database may have different local roots. It
+        # may process already-planned paths, but cannot substitute its files.
+        return
     sources, warnings = discover_local_conversations(batch.get("providers"))
     now = _now()
     for offset in range(0, len(sources), 500):
@@ -313,6 +328,7 @@ def start_local_conversation_import(
         "organisation_concept_id": organisation_concept_id,
         "role_in_org": role_in_org,
         "providers": list(_normalise_providers(providers)),
+        "discovery_scope": _discovery_scope(providers),
         "status": "planning",
         "source_count": 0,
         "source_bytes": 0,

tests/backend/test_external_conversation_bulk_import_service.py

@@ -40,6 +40,28 @@ def _codex_source(path: Path) -> None:
     path.write_text("\n".join(json.dumps(record) for record in records) + "\n")
 
 
+def test_planning_recovery_cannot_substitute_another_workers_roots(monkeypatch, tmp_path):
+    batches, items = _collections(monkeypatch)
+    roots = [tmp_path / "mac_snapshot", tmp_path / "dgx_sessions"]
+    for root in roots:
+        root.mkdir()
+        _codex_source(root / "conversation.jsonl")
+    monkeypatch.setenv("VON_CODEX_CONVERSATION_ROOT", str(roots[0]))
+    batch = {"batch_id": "recover", "custodian_user_id": "#V#user",
+             "providers": ["codex"], "status": "planning",
+             "discovery_scope": bulk._discovery_scope(["codex"])}
+    batches.insert_one(batch)
+    monkeypatch.setenv("VON_CODEX_CONVERSATION_ROOT", str(roots[1]))
+    bulk._populate_batch(batch)
+    assert items.count_documents({}) == 0
+    assert batches.find_one({})["status"] == "planning"
+    monkeypatch.setenv("VON_CODEX_CONVERSATION_ROOT", str(roots[0]))
+    bulk._populate_batch(batch)
+    assert items.count_documents({}) == 1
+    assert items.find_one({})["local_path"].startswith(str(roots[0]))
+    assert batches.find_one({})["status"] == "ready"
+
+
 def test_local_blob_store_streams_a_file_without_changing_its_bytes(tmp_path: Path):
     source = tmp_path / "source.bin"
     source.write_bytes((b"bounded-stream" * 1024) + b"end")

tests/backend/test_external_conversation_timestamp_fallback.py

@@ -129,6 +129,12 @@ def write(entries, revision):
     ]
     assert stored["updated_at"] == datetime(2026, 1, 3)
     assert write([dated, repaired, extra], "new")["action"] == "unchanged"
+    # A valid raw-source receipt repairs a missing pointer even with the same
+    # source bytes/parser; this does not duplicate transcript events.
+    manifest["raw_file_copy_concept_id"] = "#V#retained_file"
+    raw_repair = write([dated, repaired, extra], "new")
+    assert raw_repair["action"] == "refreshed"
+    assert collection.find_one({})["external_conversation_import"]["raw_file_copy_concept_id"] == "#V#retained_file"
     # Changed historical content remains a divergence, never a timestamp repair.
     divergent = write(
         [