From 1ed57bac03b58669e81f23676aa1e30249c72327 Mon Sep 17 00:00:00 2001 From: witbrock Date: Wed, 16 Sep 2026 10:12:17 -0400 Subject: [PATCH 1/2] Bind conversation batch planning to its original source roots --- ...ternal_conversation_bulk_import_service.py | 16 ++++++++++++++ ...ternal_conversation_bulk_import_service.py | 22 +++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/src/backend/services/external_conversation_bulk_import_service.py b/src/backend/services/external_conversation_bulk_import_service.py index 88681264..5504704a 100644 --- a/src/backend/services/external_conversation_bulk_import_service.py +++ b/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, diff --git a/tests/backend/test_external_conversation_bulk_import_service.py b/tests/backend/test_external_conversation_bulk_import_service.py index 899b5aed..3dee91fd 100644 --- a/tests/backend/test_external_conversation_bulk_import_service.py +++ b/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") From e394cac4d8993f4d3c34ceedbb41599e8874aed5 Mon Sep 17 00:00:00 2001 From: witbrock Date: Wed, 16 Sep 2026 10:13:58 -0400 Subject: [PATCH 2/2] Reconcile retained raw reference on unchanged source reimport --- src/backend/services/chat_history_service.py | 2 ++ .../test_external_conversation_timestamp_fallback.py | 6 ++++++ 2 files changed, 8 insertions(+) diff --git a/src/backend/services/chat_history_service.py b/src/backend/services/chat_history_service.py index a3345c91..9210c274 100644 --- a/src/backend/services/chat_history_service.py +++ b/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 = ( diff --git a/tests/backend/test_external_conversation_timestamp_fallback.py b/tests/backend/test_external_conversation_timestamp_fallback.py index 65b53b6e..916f6e0e 100644 --- a/tests/backend/test_external_conversation_timestamp_fallback.py +++ b/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( [