fix(codex-bridge): preserve plaintext peer tasks and fail invalid transfers without retries

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

@@ -1,9 +1,10 @@
 """Regression fixture matching actual Codex child delivery, with no user data."""
 import copy
+import json
 import unittest
 
 import gecis_context
-import kopru
+from test_model_switch import env as wire_bridge
 
 
 class AgentPayloadTests(unittest.TestCase):
@@ -20,7 +21,7 @@ class AgentPayloadTests(unittest.TestCase):
         task = "Run exec_command: Write-Output (157 * 29); report the exit code."
         request = self.request(task)
         original = copy.deepcopy(request)
-        kopru.flatten_request(request)
+        wire_bridge["flatten_request"](request)
         # An OpenAI-compatible translator consumes input_text, not Codex's
         # encrypted_content part. Checking the JSON string alone misses loss.
         text = "\n".join(part["text"] for item in request["input"]
@@ -48,5 +49,52 @@ class AgentPayloadTests(unittest.TestCase):
         self.assertEqual(gecis_context.portable_reasoning(request)["input"], [])
 
 
+class PlaintextAgentWireTests(unittest.TestCase):
+    def item(self, message="Read the complete task and report its result.", **extra):
+        return dict(type="function_call", namespace="collaboration", name="spawn_agent",
+                    id="fc_peer", call_id="call_peer", arguments=json.dumps({
+                        "task_name": "peer", "message": message}), **extra)
+
+    def test_final_plaintext_peer_calls_are_explicit_without_argument_changes(self):
+        for name in ("spawn_agent", "send_message", "followup_task"):
+            original = self.item(); original["name"] = name
+            for stage in ("response.output_item.done", "response.completed"):
+                with self.subTest(name=name, stage=stage):
+                    adapter = wire_bridge["SseRewriter"](set())
+                    item = copy.deepcopy(original)
+                    event = ({"type": stage, "item": item} if stage.endswith(".done") else
+                             {"type": stage, "response": {"output": [item]}})
+                    adapter._fix(event)
+                    self.assertEqual(item.pop("encrypted_function_args"), [])
+                    self.assertEqual(item, original)
+
+    def test_partial_empty_or_opaque_messages_never_gain_plaintext_metadata(self):
+        for arguments in ('', '{"message":', '{}', '[]', 'null',
+                          json.dumps({"message": " "}), json.dumps({"message": None}),
+                          json.dumps({"message": "gAAAA_ciphertext"}),
+                          json.dumps({"message": "A" * 200})):
+            for stage in ("response.output_item.added", "response.output_item.done", "response.completed"):
+                with self.subTest(arguments=arguments, stage=stage):
+                    item = self.item(); item["arguments"] = arguments
+                    wire_bridge["SseRewriter"](set())._restore_tool_name(item, plaintext=stage != "response.output_item.added")
+                    self.assertNotIn("encrypted_function_args", item)
+        item = self.item()
+        wire_bridge["SseRewriter"](set())._fix({"type": "response.output_item.added", "item": item})
+        self.assertNotIn("encrypted_function_args", item)
+
+    def test_existing_encryption_metadata_and_other_tools_are_untouched(self):
+        for marker in (None, [], ["message"]):
+            item = self.item(encrypted_function_args=marker)
+            original = copy.deepcopy(item)
+            wire_bridge["SseRewriter"](set())._fix({"type": "response.output_item.done", "item": item})
+            self.assertEqual(item, original)
+        for namespace, name in (("other", "spawn_agent"), ("collaboration", "wait_agent"),
+                                (None, "spawn_agent")):
+            item = self.item(); item.update(namespace=namespace, name=name)
+            original = copy.deepcopy(item)
+            wire_bridge["SseRewriter"](set())._fix({"type": "response.output_item.done", "item": item})
+            self.assertEqual(item, original)
+
+
 if __name__ == "__main__":
     unittest.main()