Preserve Codex context on workbench open

Avoid replacing a live transcript with context-control or prefix-only data so the workbench reflects the active Codex input without dropping prior turns.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: HaShiShark

codex_context.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from typing import Any
+
+CONTEXT_CONTROL_COMMANDS = {"ctx", "/ctx", "context", "/context"}
+
+
+def text_value(value: Any) -> str:
+    return str(value or "")
+
+
+def is_context_control_command_text(text: Any) -> bool:
+    return text_value(text).strip().lower() in CONTEXT_CONTROL_COMMANDS
+
+
+def is_contextual_user_text(text: Any) -> bool:
+    trimmed = text_value(text).lstrip()
+    lowered = trimmed.lower()
+    return (
+        trimmed.startswith("# AGENTS.md instructions for ")
+        or lowered.startswith("<environment_context>")
+        or lowered.startswith("<skills>")
+        or lowered.startswith("<user_shell_command>")
+        or lowered.startswith("<turn_aborted>")
+        or lowered.startswith("<subagent_notification>")
+    )
+
+
+def is_conversation_record(record: dict[str, Any]) -> bool:
+    role = text_value(record.get("role")).strip()
+    if role == "assistant":
+        return True
+    if role != "user":
+        return False
+    text = text_value(record.get("text"))
+    return (
+        bool(text.strip())
+        and not is_contextual_user_text(text)
+        and not is_context_control_command_text(text)
+    )
+
+
+def conversation_record_count(transcript: list[dict[str, Any]]) -> int:
+    return sum(1 for record in transcript if is_conversation_record(record))

proxy_server.py

@@ -18,6 +18,12 @@
 from pathlib import Path
 from typing import Any
 
+from codex_context import (
+    conversation_record_count,
+    is_context_control_command_text,
+    is_contextual_user_text,
+)
+
 
 HOST = os.environ.get("HASH_CONTEXT_PROXY_HOST", "127.0.0.1")
 PORT = int(os.environ.get("HASH_CONTEXT_PROXY_PORT", "8787"))
@@ -36,7 +42,6 @@
 CODEX_PROXY_BASE_URL = f"http://{HOST}:{PORT}/v1"
 INTERNAL_CONTEXT_HEADER = "x-hash-context-internal"
 INTERNAL_CONTEXT_VALUE = "context-workbench"
-CONTEXT_CONTROL_COMMANDS = {"ctx", "/ctx", "context", "/context"}
 CONTEXT_CONTROL_NOTICE_TEXT = "Hash Context: opened workbench."
 CONTROL_PORT = int(os.environ.get("HASH_CONTEXT_CONTROL_PORT", "8790"))
 LOCAL_COMPACT_PROMPT_PREFIX = "You are performing a CONTEXT CHECKPOINT COMPACTION."
@@ -507,10 +512,6 @@ def flush_assistant() -> None:
     return records
 
 
-def is_context_control_command_text(text: str) -> bool:
-    return str(text or "").strip().lower() in CONTEXT_CONTROL_COMMANDS
-
-
 def context_control_command_from_input(input_items: Any) -> str:
     transcript = input_items_to_transcript(input_items)
     for record in reversed(transcript):
@@ -1115,19 +1116,6 @@ def is_local_compact_summary_text(text: str) -> bool:
     return str(text or "").startswith(f"{LOCAL_COMPACT_SUMMARY_PREFIX}\n\n")
 
 
-def is_contextual_user_text(text: str) -> bool:
-    trimmed = str(text or "").lstrip()
-    lowered = trimmed.lower()
-    return (
-        trimmed.startswith("# AGENTS.md instructions for ")
-        or lowered.startswith("<environment_context>")
-        or lowered.startswith("<skills>")
-        or lowered.startswith("<user_shell_command>")
-        or lowered.startswith("<turn_aborted>")
-        or lowered.startswith("<subagent_notification>")
-    )
-
-
 def is_initial_context_prefix_record(record: dict[str, Any]) -> bool:
     role = str(record.get("role") or "").strip()
     if role in {"system", "developer"}:
@@ -1161,6 +1149,13 @@ def with_fresh_initial_context_prefix(
     return clean_transcript([*prefix, *body])
 
 
+def should_replace_transcript_from_control_intercept(
+    existing_transcript: list[dict[str, Any]],
+    candidate_transcript: list[dict[str, Any]],
+) -> bool:
+    return conversation_record_count(candidate_transcript) >= conversation_record_count(existing_transcript)
+
+
 def local_compact_source_from_transcript(transcript: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
     records = clean_transcript(transcript)
     if not records:
@@ -1410,8 +1405,12 @@ def record_control_intercept(self, session_id: str, body: dict[str, Any], header
             if session is None:
                 session = ProxySession(id=session_id, title=f"Codex {session_id[:8]}")
                 self.sessions[session_id] = session
-            source_transcript = strip_context_edit_notice_records(input_items_to_transcript(body.get("input")))
-            session.transcript = clean_transcript(source_transcript)
+            source_transcript = clean_transcript(
+                strip_context_edit_notice_records(input_items_to_transcript(body.get("input")))
+            )
+            existing_transcript = clean_transcript(session.transcript)
+            if should_replace_transcript_from_control_intercept(existing_transcript, source_transcript):
+                session.transcript = source_transcript
             session.pending_transcript = None
             session.local_compact_source_transcript = None
             session.status = "override" if session.edited_transcript is not None else "mirror"

scripts/test_compact_proxy.py

@@ -28,6 +28,16 @@
 PRE_TOOL_TEXT = "I will inspect the root directory."
 FINAL_TEXT = "The root contains README.md and proxy_server.py."
 TOOL_CALL_ID = "call_root_ls"
+DEVELOPER_INSTRUCTIONS = "developer instructions"
+ENVIRONMENT_CONTEXT = "<environment_context><cwd>/repo</cwd></environment_context>"
+
+
+def record(role: str, text: str) -> dict[str, Any]:
+    return proxy_server.transcript_record(role, text, [proxy_server.provider_message(role, text)])
+
+
+def provider_texts(records: list[dict[str, Any]]) -> list[str]:
+    return [str(record.get("text") or "") for record in records]
 
 
 class MockCompactUpstream(BaseHTTPRequestHandler):
@@ -341,6 +351,114 @@ def test_request_without_override_preserves_codex_body() -> None:
         assert request_log[-1]["forwarded_body"] == original_body
 
 
+def test_clean_transcript_preserves_repeated_input_context_records() -> None:
+    input_mapped_transcript = [
+        record("developer", DEVELOPER_INSTRUCTIONS),
+        record("user", ENVIRONMENT_CONTEXT),
+        record("user", "first real question"),
+        record("assistant", "first answer"),
+        record("developer", DEVELOPER_INSTRUCTIONS),
+        record("user", ENVIRONMENT_CONTEXT),
+        record("user", "second real question"),
+    ]
+
+    cleaned = proxy_server.clean_transcript(input_mapped_transcript)
+
+    assert provider_texts(cleaned) == [
+        DEVELOPER_INSTRUCTIONS,
+        ENVIRONMENT_CONTEXT,
+        "first real question",
+        "first answer",
+        DEVELOPER_INSTRUCTIONS,
+        ENVIRONMENT_CONTEXT,
+        "second real question",
+    ]
+
+
+def test_control_intercept_preserves_existing_transcript_when_input_is_prefix_only() -> None:
+    with tempfile.TemporaryDirectory() as temp_dir:
+        store = proxy_server.ProxyStore(Path(temp_dir) / "proxy_state.json")
+        existing = proxy_server.clean_transcript(
+            [
+                record("developer", DEVELOPER_INSTRUCTIONS),
+                record("user", ENVIRONMENT_CONTEXT),
+                record("user", "first real question"),
+                record("assistant", "first answer"),
+            ]
+        )
+        store.sessions[SESSION_ID] = proxy_server.ProxySession(
+            id=SESSION_ID,
+            title="Codex fake",
+            transcript=existing,
+            status="mirror",
+        )
+        body = {
+            "input": [
+                proxy_server.provider_message("developer", DEVELOPER_INSTRUCTIONS),
+                proxy_server.provider_message("user", ENVIRONMENT_CONTEXT),
+                proxy_server.provider_message("user", "ctx"),
+            ]
+        }
+
+        store.record_control_intercept(SESSION_ID, body, {"x-codex-session-id": SESSION_ID}, "ctx")
+
+        session = store.get_session(SESSION_ID)
+        assert session is not None
+        assert provider_texts(session["transcript"]) == provider_texts(existing)
+        assert store.sessions[SESSION_ID].request_log[-1]["kind"] == "context_control_intercept"
+
+
+def test_codex_local_session_sync_does_not_fallback_to_prefix_only_transcript() -> None:
+    with tempfile.TemporaryDirectory() as temp_dir:
+        previous_sessions_dir = web_server.CODEX_LOCAL_SESSIONS_DIR
+        previous_proxy_state_file = web_server.PROXY_STATE_FILE
+        try:
+            temp_path = Path(temp_dir)
+            web_server.CODEX_LOCAL_SESSIONS_DIR = temp_path / "missing-sessions"
+            web_server.PROXY_STATE_FILE = temp_path / "proxy_state.json"
+            web_server.PROXY_STATE_FILE.write_text(
+                json.dumps(
+                    {
+                        "active_session_id": SESSION_ID,
+                        "sessions": [
+                            {
+                                "id": SESSION_ID,
+                                "updated_at": "2026-05-09T00:00:00Z",
+                                "request_log": [
+                                    {
+                                        "body": {
+                                            "input": [
+                                                proxy_server.provider_message("developer", DEVELOPER_INSTRUCTIONS),
+                                                proxy_server.provider_message("user", ENVIRONMENT_CONTEXT),
+                                            ]
+                                        }
+                                    }
+                                ],
+                            }
+                        ],
+                    }
+                ),
+                encoding="utf-8",
+            )
+
+            assert web_server.latest_proxy_instruction_prefix_records()
+            assert web_server.codex_local_session_transcript(SESSION_ID) == []
+        finally:
+            web_server.CODEX_LOCAL_SESSIONS_DIR = previous_sessions_dir
+            web_server.PROXY_STATE_FILE = previous_proxy_state_file
+
+
+def test_prefix_only_codex_local_session_is_not_conversation() -> None:
+    transcript = web_server.normalize_transcript(
+        [
+            record("developer", DEVELOPER_INSTRUCTIONS),
+            record("user", ENVIRONMENT_CONTEXT),
+        ]
+    )
+
+    assert not web_server.transcript_has_conversation_records(transcript)
+
+
 def test_local_compact_without_override_preserves_codex_body() -> None:
     with tempfile.TemporaryDirectory() as temp_dir:
         store = proxy_server.ProxyStore(Path(temp_dir) / "proxy_state.json")

web_server.py

@@ -19,6 +19,7 @@
 from urllib import request as urllib_request
 from urllib.parse import quote, urlparse, urlunparse
 
+from codex_context import is_conversation_record
 from dotenv import load_dotenv
 
 try:
@@ -3441,6 +3442,10 @@ def transcript_has_instruction_prefix(records: list[dict[str, Any]]) -> bool:
     )
 
 
+def transcript_has_conversation_records(records: list[dict[str, Any]]) -> bool:
+    return any(is_conversation_record(record) for record in records)
+
+
 def provider_message_record(item: dict[str, Any]) -> dict[str, Any] | None:
     role = sanitize_text(item.get("role") or "").strip()
     if role not in CONTEXT_INPUT_MESSAGE_ROLES:
@@ -3514,7 +3519,7 @@ def latest_proxy_instruction_prefix_records() -> list[dict[str, Any]]:
 def codex_local_session_transcript(session_id: str) -> list[dict[str, Any]]:
     session_file = find_codex_local_session_file(session_id)
     if session_file is None:
-        return latest_proxy_instruction_prefix_records()
+        return []
 
     records: list[dict[str, Any]] = []
     try:
@@ -3551,6 +3556,9 @@ def codex_local_session_transcript(session_id: str) -> list[dict[str, Any]]:
             }
         )
 
+    if not records:
+        return []
+
     if not transcript_has_instruction_prefix(records):
         records = [*latest_proxy_instruction_prefix_records(), *records]
 
@@ -7302,7 +7310,7 @@ def do_POST(self) -> None:
                 if not session_id:
                     raise ValueError("session_id is required")
                 transcript = codex_local_session_transcript(session_id)
-                if not transcript:
+                if not transcript or not transcript_has_conversation_records(transcript):
                     self._send_json(
                         {
                             "error": "Codex local session was not found or has no transcript",