diff --git a/docs/engineering/external_conversation_import.md b/docs/engineering/external_conversation_import.md index b4cc4123..19728df9 100644 --- a/docs/engineering/external_conversation_import.md +++ b/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, diff --git a/src/backend/services/external_conversation_bulk_import_service.py b/src/backend/services/external_conversation_bulk_import_service.py index a37b27fd..88681264 100644 --- a/src/backend/services/external_conversation_bulk_import_service.py +++ b/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": diff --git a/src/backend/services/external_conversation_import_service.py b/src/backend/services/external_conversation_import_service.py index 85a3e1a9..5fe5ec08 100644 --- a/src/backend/services/external_conversation_import_service.py +++ b/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, }, ) diff --git a/tests/backend/test_external_conversation_bulk_import_service.py b/tests/backend/test_external_conversation_bulk_import_service.py index 8d77ea5f..899b5aed 100644 --- a/tests/backend/test_external_conversation_bulk_import_service.py +++ b/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" diff --git a/tests/backend/test_external_conversation_import_service.py b/tests/backend/test_external_conversation_import_service.py index 195d625c..bec39e77 100644 --- a/tests/backend/test_external_conversation_import_service.py +++ b/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(