Compare commits

..
Author SHA1 Message Date
yilsem 1aa033a751 fix(codex-bridge): preserve plaintext peer tasks and fail invalid transfers without retries
Server CI and Windows package / verify (push) Successful in 6m11s
Server CI and Windows package / windows-package (push) Successful in 17m49s
2026-10-20 07:35:41 +03:00
8 changed files with 143 additions and 10 deletions

No files matched your search

+11 -1
View File
@@ -35,7 +35,7 @@ import gecis_context
import terminal_stream
HERE = os.path.dirname(os.path.abspath(__file__))
SURUM = "21" # AITURK IDE'nin bekledigi kopru surumu (codex-link-config.ts CODEX_BRIDGE_VERSION)
SURUM = "22" # AITURK IDE'nin bekledigi kopru surumu (codex-link-config.ts CODEX_BRIDGE_VERSION)
try:
with open(os.path.abspath(__file__), "rb") as _f:
KOD = hashlib.sha256(_f.read()).hexdigest()[:8]
@@ -521,6 +521,16 @@ class H(http.server.BaseHTTPRequestHandler):
except Exception:
req = {}
if external:
# Opaque peer tasks cannot be sent to another provider. Detect
# this permanent transfer error before committing SSE headers:
# stock Codex retries unknown response.failed stream errors.
# Projection validates a copy; the saved/request history stays
# intact and no task is dropped or sent to the worker.
try:
gecis_context.portable_reasoning(req)
except gecis_context.TransferError as e:
return self._json(400, {"error": {"type": "invalid_request_error",
"code": "aiturk_transfer_failed", "message": str(e), "retryable": False}})
self._failure_key = terminal_stream.request_key(req or {"raw": govde.hex()}, self.headers)
previous_failure = self.server.failure_replays.take(self._failure_key)
if previous_failure:
+20 -4
View File
@@ -42,7 +42,7 @@ from bekci_watch import MetadataFilter
# Installed only by owned native launcher, never an HTTP control endpoint.
WATCH_OWNER = None
SURUM = "20"
SURUM = "21"
HERE = os.path.dirname(os.path.abspath(__file__))
try:
with open(os.path.abspath(__file__), "rb") as _f:
@@ -883,11 +883,27 @@ class SseRewriter:
self.bekleyen = d
return once
return once + [d]
def _restore_tool_name(self, item):
def _restore_tool_name(self, item, plaintext=False):
if isinstance(item, dict) and item.get("type") in ("function_call", "custom_tool_call"):
identity = self.tool_names.get(item.get("name"))
if identity:
item["namespace"], item["name"] = identity
if (plaintext and item.get("type") == "function_call"
and item.get("namespace") == "collaboration"
and item.get("name") in ("spawn_agent", "send_message", "followup_task")
and "encrypted_function_args" not in item):
# Codex distinguishes missing encryption metadata (opaque) from
# an explicit empty list (plaintext). External providers return
# ordinary JSON strings; stamp only a complete readable message,
# after its final arguments arrive. Never override owner state.
try:
message = json.loads(item.get("arguments") or "{}").get("message")
except (ValueError, AttributeError, TypeError):
message = None
if (isinstance(message, str) and message.strip()
and not message.startswith("gAAAA")
and not re.fullmatch(r"[A-Za-z0-9_+/=-]{128,}", message)):
item["encrypted_function_args"] = []
return item
def _fix(self, d):
t = d.get("type", "")
@@ -923,10 +939,10 @@ class SseRewriter:
item = {"type": "custom_tool_call", "id": item.get("id"), "call_id": item.get("call_id"),
"name": item.get("name"), "input": inp, "status": "completed"}
fix_tool_ids({"input": [item]})
out.append(self._restore_tool_name(item))
out.append(self._restore_tool_name(item, plaintext=True))
d["response"]["output"] = out
fix_tool_ids({"input": out})
self._restore_tool_name(d.get("item"))
self._restore_tool_name(d.get("item"), plaintext=t == "response.output_item.done")
return d
def feed(self, chunk):
self.buf += chunk
@@ -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()
@@ -281,6 +281,42 @@ class PreparationService(unittest.TestCase):
class SocketMigration(fixtures.Ayirici):
def test_opaque_peer_transfer_fails_as_http_400_before_sse_or_upstream(self):
req = {"model": "ide/agy-gemini-3.8-flash-high", "prompt_cache_key": "opaque-peer",
"input": [{"type": "agent_message", "author": "/root", "recipient": "/root/peer",
"content": [{"type": "input_text", "text": "Message Type: NEW_TASK\nPayload:\n"},
{"type": "encrypted_content", "encrypted_content": "gAAAA_synthetic_opaque"}]}]}
original = copy.deepcopy(req)
raw = json.dumps(req).encode()
# Identical retries must receive the same permanent error without a
# preparation request, a stream heartbeat, or provider inference.
for _ in range(2):
status, body, headers = self.iste("POST", "/backend-api/codex/responses", raw)
self.assertEqual(status, 400)
self.assertEqual(headers["Content-Type"].split(";")[0], "application/json")
self.assertNotIn("Transfer-Encoding", headers)
error = json.loads(body)["error"]
self.assertEqual(error["code"], "aiturk_transfer_failed")
self.assertEqual(error["type"], "invalid_request_error")
self.assertFalse(error["retryable"])
self.assertIn("görev eksik gönderilmedi", error["message"])
self.assertNotIn(b"response.failed", body)
self.assertEqual(req, original)
self.assertEqual(self.ozgun.gelen, [])
self.assertEqual(self.kopru.gelen, [])
self.assertEqual(bridge.TRANSFERS, {})
def test_native_opaque_peer_history_remains_byte_identical_on_owner_route(self):
req = {"model": "gpt-6-astra", "input": [{"type": "agent_message", "author": "/root",
"recipient": "/root/peer", "content": [{"type": "encrypted_content",
"encrypted_content": "gAAAA_synthetic_native_owner"}]}]}
raw = zstd(json.dumps(req).encode())
status, _, _ = self.iste("POST", "/backend-api/codex/responses", raw,
{"Content-Encoding": "zstd"})
self.assertEqual(status, 200)
self.assertEqual(self.ozgun.gelen[-1]["govde"], raw)
self.assertEqual(self.kopru.gelen, [])
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": []}
@@ -45,6 +45,29 @@ class ToolNamespaces(unittest.TestCase):
self.assertEqual(restored["id"], "ctc_patch")
self.assertEqual(restored["input"], "patch text")
def test_plaintext_peer_marker_survives_namespace_restore_and_split_frames(self):
req = self.request(); names = {}; custom = env["flatten_request"](req, names)
alias = req["tools"][0]["name"]
arguments = json.dumps({"task_name": "child", "message": "Execute the complete delegated task."})
item = {"type": "function_call", "name": alias, "id": "fc_peer", "call_id": "call_peer", "arguments": arguments}
events = [
{"type": "response.output_item.added", "item": dict(item, arguments="")},
{"type": "response.function_call_arguments.delta", "item_id": "fc_peer", "delta": arguments[:17]},
{"type": "response.function_call_arguments.delta", "item_id": "fc_peer", "delta": arguments[17:]},
{"type": "response.output_item.done", "item": dict(item)},
{"type": "response.completed", "response": {"output": [dict(item)]}},
]
raw = b"".join(("data: " + json.dumps(event) + "\n\n").encode() for event in events)
adapter = env["SseRewriter"](custom, tool_names=names)
output = b"".join(adapter.feed(raw[i:i+13]) for i in range(0, len(raw), 13)) + adapter.flush()
restored = [json.loads(line[6:]) for line in output.decode().splitlines() if line.startswith("data: ")]
self.assertNotIn("encrypted_function_args", restored[0]["item"])
self.assertEqual(restored[1:3], events[1:3])
for final in (restored[3]["item"], restored[4]["response"]["output"][0]):
self.assertEqual(final["encrypted_function_args"], [])
self.assertEqual((final["namespace"], final["name"]), ("collaboration", "spawn_agent"))
self.assertEqual((final["id"], final["call_id"], final["arguments"]), ("fc_peer", "call_peer", arguments))
def test_plain_tools_stay_plain_and_history_not_mutated(self):
req = self.request(); original = copy.deepcopy(req); old_item = req["input"][0]
env["flatten_request"](req, {})
+1 -1
View File
@@ -27,7 +27,7 @@ export const CODEX_LINK_REFRESH_URL = `${CODEX_LINK_ORIGIN}/aiturk/codex-baglama
export const CODEX_LINK_MARKER = '# AITURK IDE Codex Baglama'
// Must equal SURUM in assets/codex-bridge/bridge.py; a running bridge that reports
// another version is restarted so an updated IDE never keeps an old bridge alive.
export const CODEX_BRIDGE_VERSION = '21'
export const CODEX_BRIDGE_VERSION = '22'
export const CODEX_LINK_LINE = `openai_base_url = "${CODEX_LINK_ORIGIN}/backend-api/codex" ${CODEX_LINK_MARKER}`
export const CODEX_LINK_RUN_KEY = 'HKCU\\Software\\Microsoft\\Windows\\CurrentVersion\\Run'
+1 -1
View File
@@ -2,7 +2,7 @@
"name": "aiturk-ide",
"productName": "AITURK IDE",
"private": true,
"version": "1.0.0-beta.43",
"version": "1.0.0-beta.44",
"description": "TurkServis resmi yapay zekâ geliştirme ortamı. Hermes Agent tabanlıdır.",
"author": "AITURK / TurkServis",
"repository": {
+1 -1
View File
@@ -65,7 +65,7 @@
},
"apps/desktop": {
"name": "aiturk-ide",
"version": "1.0.0-beta.43",
"version": "1.0.0-beta.44",
"license": "MIT",
"dependencies": {
"@assistant-ui/core": "0.2.23",