Preserve native Codex history and add configurable cross-model compression

Separate encrypted reasoning from actual compaction, stop migration after native return, and retain full history and recent exchanges. Add typed settings, Gemini fallback, verified provider windows, bounded summaries and durable concurrent checkpoints.

Validation: 245 bridge tests plus focused concurrency tests; 71 settings/connection tests; 66 supplementary tests; TypeScript checks; renderer and Electron bundles; real primary/native/fallback summary probes and 700K input acceptance.

apps/desktop/assets/codex-bridge/test_gecis_context.py

@@ -8,6 +8,10 @@ from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 import unittest
 import urllib.request
+import tempfile
+from types import SimpleNamespace
+import compression_engine
+import compression_settings
 
 import gecis_context as gc
 import ozet_kopru
@@ -18,6 +22,34 @@ from test_model_switch import env
 
 
 class History(unittest.TestCase):
+    def test_native_function_results_are_not_foreign_or_renamed(self):
+        req = {"input": [{"type": "function_call_output", "id": "fco_native",
+                          "call_id": "call_native", "output": "kept"}]}
+        self.assertFalse(gc.foreign(req))
+        self.assertEqual(gc.normalize(req), req)
+
+    def test_encrypted_reasoning_does_not_block_visible_history_summary(self):
+        req = {"input": [{"role": "user", "content": "KARAR-4827 " + "a" * 60000},
+                         {"type": "reasoning", "encrypted_content": "OPAQUE_OLD",
+                          "summary": [{"type": "summary_text", "text": "VISIBLE_NOTE"}]},
+                         {"role": "assistant", "content": "Work done"},
+                         {"role": "user", "content": "Latest"},
+                         {"type": "reasoning", "encrypted_content": "OPAQUE_RECENT"}]}
+        before = copy.deepcopy(req)
+        calls = []
+        def summarize(text, budget):
+            calls.append(text)
+            self.assertNotIn("OPAQUE_", text)
+            self.assertIn("VISIBLE_NOTE", text)
+            self.assertIn("KARAR-4827", text)
+            return "KARAR-4827: Visible decisions and work preserved. " * 4
+        result = gc.Preparer().prepare(req, 32768, summarize)
+        self.assertEqual(len(calls), 1)
+        self.assertNotIn("OPAQUE_", json.dumps(result))
+        self.assertEqual(req, before)
+        self.assertLessEqual(gc.tokens(result), gc.input_budget(req, 32768))
+        self.assertEqual(result["input"][-1], {"role": "user", "content": "Latest"})
+
     def test_legacy_summary_and_tool_ids_repaired_without_mutating_source(self):
         req = {"input": [ozet_kopru.ozet_ogesi("KARAR-4827 korunacak"),
                {"type": "custom_tool_call", "id": "fc_call_173683", "call_id": "call_1", "input": "abc"},
@@ -112,6 +144,12 @@ class PreparationService(unittest.TestCase):
             duz(200, raw, "text/event-stream")(h, body)
         self.server = sunucu(upstream)
         self.ns["ROUTES"] = {"aiturk": {"url": "http://127.0.0.1:%d/responses" % self.server.server_address[1], "key": "gateway-test"}}
+        self.temp = tempfile.TemporaryDirectory()
+        settings = dict(compression_settings.DEFAULTS, keepTurns=1, primaryModel='ide/test', timeoutSeconds=5)
+        Path(self.temp.name, 'compression-settings.json').write_text(json.dumps(settings), encoding='utf-8')
+        self.ns.update(HOME=self.temp.name, compression_engine=compression_engine,
+                       CHECKPOINTS=compression_settings.Checkpoints(self.temp.name), __name__='offline')
+        self.ns['sys'] = SimpleNamespace(modules={'offline': SimpleNamespace(**self.ns)})
         cls = ast.ClassDef(name="Service", bases=[], keywords=[], body=methods, decorator_list=[])
         exec(compile(ast.fix_missing_locations(ast.Module(body=[cls], type_ignores=[])), "preparation", "exec"), self.ns)
         self.service = self.ns["Service"]()
@@ -121,6 +159,7 @@ class PreparationService(unittest.TestCase):
     def tearDown(self):
         self.server.shutdown()
         self.server.server_close()
+        self.temp.cleanup()
 
     def test_large_external_preflight_uses_aiturk_only_and_fits(self):
         req = {"model": "ide/test", "input": [{"role": "user", "content": "x" * 60000 + "KARAR-4827"},
@@ -148,14 +187,110 @@ class PreparationService(unittest.TestCase):
             self.service._prepare_transfer({"request": req})
         self.assertEqual(self.server.gelen, [])
 
-    def test_external_compaction_uses_external_summary_not_subscription(self):
-        req = {"model": "ide/test", "input": [{"role": "user", "content": "KARAR-4827"}, {"type": "compaction_trigger"}]}
+    def test_external_small_history_strips_only_opaque_reasoning(self):
+        req = {"model": "ide/test", "input": [
+            {"role": "user", "content": "KARAR-4827"},
+            {"type": "reasoning", "id": "rs_native", "encrypted_content": "OPAQUE_NATIVE",
+             "summary": [{"type": "summary_text", "text": "VISIBLE_NOTE"}]},
+            {"type": "function_call", "id": "fc_native", "call_id": "call_native", "name": "read", "arguments": "{}"},
+            {"type": "function_call_output", "id": "fco_native", "call_id": "call_native", "output": "TOOL_RESULT"},
+            {"role": "assistant", "content": "ANSWER"}]}
+        before = copy.deepcopy(req)
+        result = self.service._prepare_transfer({"request": req})["request"]
+        self.assertNotIn("OPAQUE_NATIVE", json.dumps(result))
+        self.assertIn("VISIBLE_NOTE", json.dumps(result))
+        for item in (req["input"][0], *req["input"][2:]):
+            self.assertIn(item, result["input"])
+        self.assertEqual(req, before)
+        self.assertEqual(self.server.gelen, [])
+
+    def test_external_compaction_never_sends_reasoning_ciphertext_to_summarizer(self):
+        req = {"model": "ide/test", "input": [
+            {"role": "assistant", "content": "KARAR-4827: previous decision"},
+            {"role": "user", "content": "KARAR-4827"},
+            {"type": "reasoning", "encrypted_content": "OPAQUE_NATIVE"},
+            {"type": "compaction_trigger"}]}
         result = self.service._prepare_transfer({"request": req})
-        self.assertIn("KARAR-4827", ozet_kopru.ozet_metni(result["compaction"]))
+        self.assertIn("KARAR-4827", ozet_kopru.ozet_metni(result["compaction_items"][0]))
+        self.assertEqual(len(self.server.gelen), 1)
+        self.assertNotIn(b"OPAQUE_NATIVE", self.server.gelen[0]["govde"])
+
+    def test_native_small_history_keeps_reasoning_unchanged(self):
+        req = {"model": "gpt-6-astra", "input": [
+            {"type": "reasoning", "encrypted_content": "OPAQUE_NATIVE"},
+            {"role": "user", "content": "Latest"}]}
+        self.assertEqual(self.service._prepare_transfer({"request": req})["request"], req)
+        self.assertEqual(self.server.gelen, [])
+
+    def test_real_compaction_still_uses_owner_while_reasoning_is_portable(self):
+        compact = {"type": "compaction", "encrypted_content": "REAL_COMPACTION"}
+        calls = []
+        def decode(_n, item, _cfg):
+            calls.append(item)
+            return "COMPACTED_DECISION_4827"
+        self.service._openai_ozet_coz = decode
+        req = {"model": "ide/test", "input": [compact,
+            {"type": "reasoning", "encrypted_content": "OPAQUE_REASONING"},
+            {"role": "user", "content": "Latest"}]}
+        before = copy.deepcopy(req)
+        result = self.service._prepare_transfer({"request": req})["request"]
+        self.assertEqual(calls, [compact])
+        self.assertIn("COMPACTED_DECISION_4827", json.dumps(result))
+        self.assertNotIn("OPAQUE_REASONING", json.dumps(result))
+        self.assertEqual(req, before)
+
+    def test_large_native_return_summarizes_visible_history_without_ciphertext(self):
+        req = {"model": "gpt-6-astra", "input": [
+            {"role": "user", "content": "KARAR-4827 " + "x" * 190000},
+            {"type": "reasoning", "encrypted_content": "OPAQUE_NATIVE"},
+            {"role": "user", "content": "Latest"}]}
+        before = copy.deepcopy(req)
+        result = self.service._prepare_transfer({"request": req, "source_model": "ide/test"})["request"]
+        self.assertEqual(result["model"], "gpt-6-astra")
+        self.assertIn("KARAR-4827", json.dumps(result))
+        self.assertLessEqual(gc.tokens(result), gc.input_budget(req, 128000))
+        self.assertEqual(req, before)
+        self.assertTrue(self.server.gelen)
+        self.assertTrue(all(b"OPAQUE_NATIVE" not in call["govde"] for call in self.server.gelen))
+
+    def test_external_compaction_uses_external_summary_not_subscription(self):
+        req = {"model": "ide/test", "input": [{"role": "assistant", "content": "KARAR-4827"}, {"role": "user", "content": "KARAR-4827"}, {"type": "compaction_trigger"}]}
+        result = self.service._prepare_transfer({"request": req})
+        self.assertIn("KARAR-4827", ozet_kopru.ozet_metni(result["compaction_items"][0]))
         self.assertEqual(len(self.server.gelen), 1)
 
 
 class SocketMigration(fixtures.Ayirici):
+    def test_completed_return_ends_preparation_for_following_native_turns(self):
+        self.ozgun.cevap = duz(200, b'data: {"type":"response.completed"}\n\n', 'text/event-stream')
+        req = {"model": "ide/test", "prompt_cache_key": "return-once", "input": []}
+        self.iste("POST", "/backend-api/codex/responses", json.dumps(req).encode())
+        req.update(model="gpt-6-astra", input=[{"role": "user", "content": "x" * 100000}])
+        self.iste("POST", "/backend-api/codex/responses", json.dumps(req).encode())
+        count = len(self.kopru.gelen)
+        req.update(model="gpt-6-luna")
+        req["input"].append({"type": "reasoning", "encrypted_content": "NATIVE_OPAQUE"})
+        raw = zstd(json.dumps(req).encode())
+        self.iste("POST", "/backend-api/codex/responses", raw, {"Content-Encoding": "zstd"})
+        self.assertEqual(len(self.kopru.gelen), count)
+        self.assertEqual(self.ozgun.gelen[-1]["govde"], raw)
+
+    def test_native_large_tool_history_bypasses_transfer_byte_for_byte(self):
+        for model in ("gpt-6-astra", "gpt-6-luna"):
+            req = {"model": model, "prompt_cache_key": "native-only", "input": [
+                {"role": "user", "content": "x" * 100000},
+                {"type": "reasoning", "id": "rs_native", "encrypted_content": "OPAQUE_NATIVE"},
+                {"type": "function_call", "id": "fc_native", "call_id": "c", "name": "read", "arguments": "{}"},
+                {"type": "function_call_output", "id": "fco_native", "call_id": "c", "output": "result"},
+                {"role": "user", "content": "Latest"}]}
+            raw = json.dumps(req).encode()
+            for encoding, body in (("", raw), ("zstd", zstd(raw))):
+                headers = {"Content-Encoding": encoding} if encoding else {}
+                status, _, _ = self.iste("POST", "/backend-api/codex/responses", body, headers)
+                self.assertEqual(status, 200)
+                self.assertEqual(self.ozgun.gelen[-1]["govde"], body)
+                self.assertEqual(self.kopru.gelen, [])
+
     def test_fifty_simultaneous_chats_keep_their_provider_routes(self):
         start = threading.Barrier(50)
         def call(index):