Preserve distinct coding-agent conversation identities during import

docs/engineering/external_conversation_import.md

@@ -1,5 +1,13 @@
 # External conversation import
 
+Codex machine discovery includes both active and archived sessions. For forked
+Codex rollouts, the first session metadata owns the conversation identity;
+copied ancestor metadata does not replace it. Gemini A2A journals reuse the
+`a2a-server` identifier, so separate runs are distinguished by their stable
+`startTime`. The original identifier and the identity basis remain in the import
+report; raw source bytes are unchanged. A missing A2A start time falls back to
+the source hash, so append reconciliation cannot be guaranteed for that case.
+
 - **Kind:** Design and implementation boundary
 - **Lifecycle:** Active
 - **Authority:** Canonical reference within external transcript import scope,

src/backend/services/external_conversation_bulk_import_service.py

@@ -124,7 +124,10 @@ def _provider_roots(provider: str) -> tuple[tuple[str, Path, str], ...]:
         pattern = "**/*.json*" if provider == "copilot" else "**/*.jsonl"
         return ((f"{provider}:configured", Path(override).expanduser(), pattern),)
     if provider == "codex":
-        return (("codex:sessions", home / ".codex" / "sessions", "**/*.jsonl"),)
+        return (
+            ("codex:sessions", home / ".codex" / "sessions", "**/*.jsonl"),
+            ("codex:archived", home / ".codex" / "archived_sessions", "**/*.jsonl"),
+        )
     if provider == "claude_code":
         return (("claude:projects", home / ".claude" / "projects", "**/*.jsonl"),)
     if provider == "gemini":

src/backend/services/external_conversation_import_service.py

@@ -30,7 +30,7 @@
 
 
 EXTERNAL_CONVERSATION_PACKAGE_SCHEMA_VERSION = "external_conversation_package.v1"
-EXTERNAL_CONVERSATION_PARSER_VERSION = "2026-08-29.2"
+EXTERNAL_CONVERSATION_PARSER_VERSION = "2026-09-16.1"
 EXTERNAL_CONVERSATION_ORIGIN_KIND = "external_conversation_import"
 EXTERNAL_CONVERSATION_LINEAGE_SCHEMA_VERSION = "conversation_lineage.v1"
 MAX_EXTERNAL_CONVERSATION_SOURCE_BYTES = 32 * 1024 * 1024
@@ -504,6 +504,10 @@ def _parse_codex_records(
         locator = f"line:{record.get('__line_number')}"
         timestamp = _iso_timestamp(record.get("timestamp") or payload.get("timestamp"))
         if envelope_type == "session_meta":
+            # Forked rollouts start with their own metadata, then may include
+            # copied ancestor metadata. The ancestor is not this conversation.
+            if session_id is not None:
+                continue
             session_id = (
                 _clean_text(payload.get("id"), maximum=240)
                 or _clean_text(payload.get("session_id"), maximum=240)
@@ -835,6 +839,23 @@ def _parse_gemini_records(
     session_id = _clean_text(state.get("sessionId"), maximum=240) or (
         f"gemini-{source_sha256[:24]}"
     )
+    identity_report: dict[str, Any] = {}
+    if session_id == "a2a-server":
+        # Gemini's A2A server reuses this placeholder across distinct runs.
+        # startTime stays fixed as a journal grows; lastUpdated/content do not.
+        started_at = _iso_timestamp(state.get("startTime"))
+        identity_suffix = (
+            hashlib.sha256(started_at.encode("utf-8")).hexdigest()[:24]
+            if started_at
+            else source_sha256[:24]
+        )
+        identity_report = {
+            "source_session_id": session_id,
+            "session_identity_basis": (
+                "session_id_and_start_time" if started_at else "source_sha256"
+            ),
+        }
+        session_id = f"{session_id}-{identity_suffix}"
     messages = state.get("messages")
     messages = messages if isinstance(messages, list) else []
     events: list[ExternalConversationEvent] = []
@@ -899,6 +920,7 @@ def _parse_gemini_records(
             "assistant_message_count": assistant_count,
             "source_has_no_assistant_messages": assistant_count == 0,
             "branch_information_present": False,
+            **identity_report,
         },
     )
 

tests/backend/test_external_conversation_bulk_import_service.py

@@ -84,6 +84,24 @@ def test_discovery_is_allowlisted_and_preview_does_not_persist(
     assert "local_path" not in json.dumps(preview)
 
 
+def test_default_codex_discovery_includes_archived_conversations(
+    monkeypatch, tmp_path: Path
+):
+    monkeypatch.delenv("VON_CODEX_CONVERSATION_ROOT", raising=False)
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    for directory in ("sessions", "archived_sessions"):
+        root = tmp_path / ".codex" / directory
+        root.mkdir(parents=True)
+        _codex_source(root / "conversation.jsonl")
+    sources, warnings = bulk.discover_local_conversations(["codex"])
+    assert {source.root_id for source in sources} == {
+        "codex:sessions",
+        "codex:archived",
+    }
+    assert len(sources) == 2
+    assert not warnings
+
+
 def test_batch_pause_resume_and_expired_lease_recovery(monkeypatch, tmp_path: Path):
     batches, items = _collections(monkeypatch)
     root = tmp_path / "codex"

tests/backend/test_external_conversation_import_service.py

@@ -24,6 +24,94 @@ def _jsonl(*records: dict) -> bytes:
     return ("\n".join(json.dumps(record) for record in records) + "\n").encode()
 
 
+def test_codex_child_identity_survives_inherited_ancestor_metadata():
+    def parse(child_id: str, *, append: bool = False):
+        records = [
+            {
+                "type": "session_meta",
+                "payload": {
+                    "id": child_id,
+                    "forked_from_id": "parent",
+                    "cwd": "/child",
+                },
+            },
+            {
+                "type": "session_meta",
+                "payload": {
+                    "id": "parent",
+                    "forked_from_id": "grandparent",
+                    "cwd": "/parent",
+                },
+            },
+            {
+                "type": "response_item",
+                "payload": {
+                    "type": "message",
+                    "role": "user",
+                    "content": "Inherited question",
+                },
+            },
+        ]
+        if append:
+            records.append(
+                {
+                    "type": "response_item",
+                    "payload": {
+                        "type": "message",
+                        "role": "assistant",
+                        "content": "Child answer",
+                    },
+                }
+            )
+        return parse_external_conversation_bytes(
+            _jsonl(*records), provider_hint="codex"
+        )
+
+    first, sibling, grown = (
+        parse("child-1"),
+        parse("child-2"),
+        parse("child-1", append=True),
+    )
+    assert first.source_session_id == grown.source_session_id == "child-1"
+    assert sibling.source_session_id == "child-2"
+    assert first.parent_conversation_id == "parent"
+    assert first.source_workspace == "/child"
+    assert first.events[0].event_id == grown.events[0].event_id
+    assert first.events[0].event_id != sibling.events[0].event_id
+
+
+def test_gemini_server_runs_have_distinct_stable_identities():
+    def parse(start: str, *, append: bool = False, session_id: str = "a2a-server"):
+        messages = [{"id": "user-1", "type": "user", "content": "Inspect the code"}]
+        if append:
+            messages.append({"id": "assistant-1", "type": "gemini", "content": "Done"})
+        return parse_external_conversation_bytes(
+            _jsonl(
+                {
+                    "sessionId": session_id,
+                    "projectHash": "workspace",
+                    "startTime": start,
+                    "lastUpdated": "2026-09-16T12:01:00Z" if append else start,
+                    "messages": messages,
+                }
+            ),
+            provider_hint="gemini",
+        )
+
+    first = parse("2026-09-16T12:00:00Z")
+    other = parse("2026-09-16T12:00:01Z")
+    grown = parse("2026-09-16T12:00:00+00:00", append=True)
+    assert first.source_session_id != other.source_session_id
+    assert first.source_session_id == grown.source_session_id
+    assert first.events[0].event_id == grown.events[0].event_id
+    assert first.loss_report["source_session_id"] == "a2a-server"
+    assert first.loss_report["session_identity_basis"] == "session_id_and_start_time"
+    assert (
+        parse("2026-09-16T12:00:00Z", session_id="unique-session").source_session_id
+        == "unique-session"
+    )
+
+
 def test_codex_parser_projects_visible_messages_without_importing_instructions():
     package = parse_external_conversation_bytes(
         _jsonl(