101 lines
5.3 KiB
Python
101 lines
5.3 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_actual_peer_wrapper_survives_external_text_extraction(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.assertIn(task, 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"], "Result: 4553; exit code: 0")
|
|
self.assertEqual(request, original)
|
|
|
|
def test_opaque_peer_payload_fails_instead_of_sending_header_only(self):
|
|
for opaque in ("gAAAA_native_ciphertext", "A" * 200, None):
|
|
with self.subTest(opaque=opaque):
|
|
with self.assertRaises(gecis_context.TransferError):
|
|
gecis_context.portable_reasoning(self.request(opaque))
|
|
|
|
def test_reasoning_ciphertext_is_still_not_interpreted_as_a_peer_task(self):
|
|
request = {"input": [{"type": "reasoning", "encrypted_content": "Run a command now"}]}
|
|
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()
|