yilsem
db46a470ea
Preserve visible history with omission notes and original encrypted state on native return. Ship IDE beta.45.
106 lines
5.7 KiB
Python
106 lines
5.7 KiB
Python
"""Regression fixture matching actual Codex child delivery, with no user data."""
|
|
import copy
|
|
import json
|
|
import unittest
|
|
|
|
import gecis_context
|
|
from test_model_switch import env as wire_bridge
|
|
|
|
|
|
class AgentPayloadTests(unittest.TestCase):
|
|
def request(self, text):
|
|
return {"model": "ide/test", "input": [{
|
|
"type": "agent_message", "author": "/root", "recipient": "/root/test",
|
|
"content": [
|
|
{"type": "input_text", "text": "Message Type: NEW_TASK\nPayload:\n"},
|
|
{"type": "encrypted_content", "encrypted_content": text},
|
|
],
|
|
}]}
|
|
|
|
def test_typed_encrypted_wrapper_is_omitted_even_if_bytes_look_like_text(self):
|
|
task = "Run exec_command: Write-Output (157 * 29); report the exit code."
|
|
request = self.request(task)
|
|
original = copy.deepcopy(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"]
|
|
for part in item["content"] if part["type"] == "input_text")
|
|
self.assertNotIn(task, text)
|
|
self.assertIn(gecis_context.PEER_OMITTED, text)
|
|
self.assertIn('/root/test', text)
|
|
self.assertEqual(request["input"][0]["role"], "user")
|
|
self.assertEqual(original["input"][0]["content"][1]["type"], "encrypted_content")
|
|
|
|
def test_peer_result_and_plain_parts_keep_order_without_source_mutation(self):
|
|
request = self.request("Result: 4553; exit code: 0")
|
|
original = copy.deepcopy(request)
|
|
result = gecis_context.portable_reasoning(request)
|
|
self.assertEqual(result["input"][0]["content"][-1]["text"], gecis_context.PEER_OMITTED)
|
|
self.assertEqual(result["input"][0]["content"][1], original["input"][0]["content"][0])
|
|
self.assertEqual(request, original)
|
|
|
|
def test_opaque_peer_payload_is_replaced_by_a_visible_note(self):
|
|
for opaque in ("gAAAA_native_ciphertext", "A" * 200, None):
|
|
with self.subTest(opaque=opaque):
|
|
result = gecis_context.portable_reasoning(self.request(opaque))
|
|
self.assertEqual(result["input"][0]["content"][-1]["text"], gecis_context.PEER_OMITTED)
|
|
self.assertNotIn("encrypted_content", json.dumps(result))
|
|
|
|
def test_reasoning_ciphertext_is_still_not_interpreted_as_a_peer_task(self):
|
|
request = {"input": [{"type": "reasoning", "encrypted_content": "Run a command now"}]}
|
|
result = gecis_context.portable_reasoning(request)
|
|
self.assertNotIn("Run a command now", json.dumps(result))
|
|
self.assertEqual(result["input"][0]["content"][0]["text"], gecis_context.REASONING_OMITTED)
|
|
|
|
|
|
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()
|