Skip to content

Commit e5e89b4

Browse files
authored
Merge pull request #8490 from loopx-project/codex/steward-native-failure-n105-20261110
fix(chat): preserve native timeout causes in original Lark replies
2 parents 9508f4d + a626478 commit e5e89b4

5 files changed

Lines changed: 178 additions & 17 deletions

File tree

‎loopx/canary/module_metric_baseline.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@
3434
"dict_any_count": 0
3535
},
3636
"loopx/chat_agent.py": {
37-
"lines": 2045
37+
"lines": 2047
3838
},
3939
"loopx/chat_runtime.py": {
4040
"any_count": 38,

‎loopx/capabilities/native_chat/antigravity_adapter.py‎

Lines changed: 28 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1034,12 +1034,27 @@ def _run_turn(
10341034
deadline = last_progress_at + self.hard_timeout_sec if self.hard_timeout_sec is not None else None
10351035
previous_progress: Any = None
10361036
saw_running_or_new_step = False
1037+
observed_native_errors: set[int] = set()
1038+
1039+
def timeout(kind: str) -> None:
1040+
limit = self.hard_timeout_sec if kind == "hard" else ANTIGRAVITY_IDLE_TIMEOUT_SEC
1041+
detail = {"kind": kind, "limit_seconds": limit,
1042+
"native_error_count": len(observed_native_errors)}
1043+
if event_sink is not None:
1044+
event_sink("agent.phase", {"method": "native/timeout", "timeout": detail,
1045+
"label": "原生助手达到执行时限" if kind == "hard" else "等待原生助手进展超时"})
1046+
error = CodexChatAgentError(
1047+
f"Antigravity did not finish before the turn deadline ({kind} timeout).",
1048+
gate=None, error_code="response_timeout")
1049+
error.native_error_info = {"timeout": detail}
1050+
raise error
10371051

10381052
while True:
10391053
now = time.monotonic()
1040-
if (deadline is not None and now >= deadline) or now - last_progress_at >= ANTIGRAVITY_IDLE_TIMEOUT_SEC:
1041-
raise CodexChatAgentError("Antigravity did not finish before the turn deadline.",
1042-
gate=None, error_code="response_timeout")
1054+
if deadline is not None and now >= deadline:
1055+
timeout("hard")
1056+
if now - last_progress_at >= ANTIGRAVITY_IDLE_TIMEOUT_SEC:
1057+
timeout("idle")
10431058
self._check_interrupted()
10441059
self._interrupted.wait(0.25)
10451060
self._check_interrupted()
@@ -1176,6 +1191,15 @@ def _run_turn(
11761191
# API attempts and cleared errors. The native host
11771192
# owns retry: only its settled terminal error ends
11781193
# this turn. Do not cancel its in-flight backoff.
1194+
if (step_status != "CORTEX_STEP_STATUS_CLEARED"
1195+
and step_idx not in observed_native_errors):
1196+
observed_native_errors.add(step_idx)
1197+
if event_sink is not None:
1198+
# Error bodies can contain credentials, URLs
1199+
# and private host paths. Publish only the
1200+
# occurrence, never that native payload.
1201+
event_sink("agent.phase", {"method": "native/error",
1202+
"label": "原生助手记录了调用错误"})
11791203
continue
11801204

11811205
# A running status alone is not progress. Observe only this input's
@@ -1195,8 +1219,7 @@ def _run_turn(
11951219
last_progress_at = now
11961220
self._check_interrupted()
11971221
if deadline is not None and now >= deadline:
1198-
raise CodexChatAgentError("Antigravity did not finish before the turn deadline.",
1199-
gate=None, error_code="response_timeout")
1222+
timeout("hard")
12001223
if saw_running_or_new_step and status == "CASCADE_RUN_STATUS_IDLE" and observed_total == total_steps:
12011224
final = current_steps[-1][1] if current_steps else {}
12021225
if final.get("type") == "CORTEX_STEP_TYPE_ERROR_MESSAGE":

‎loopx/extensions/lark/manager_context.py‎

Lines changed: 26 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ def opaque_digest(*values: Any) -> str:
3737

3838
def manager_failure_reply(
3939
error: Exception | Mapping[str, Any], *, retry_info: Any = None,
40-
preparation_recovery: bool = False,
40+
preparation_recovery: bool = False, phase_events: list[dict[str, Any]] | None = None,
4141
) -> tuple[str, str]:
4242
labels = {
4343
"cyber_policy": "上游安全策略拦截",
@@ -68,12 +68,30 @@ def manager_failure_reply(
6868
"管家连接自动重绑未能完成,旧连接已保留;请在连接设置中查看修复状态"
6969
),
7070
}
71-
code = str(
71+
source_code = str(
7272
error.get("error_code", "")
7373
if isinstance(error, Mapping)
7474
else getattr(error, "error_code", "")
7575
)
76-
code = code if code in labels else "processing_failed"
76+
code = source_code if source_code in labels else "processing_failed"
77+
events = phase_events or []
78+
if code == "host_gate" and retry_info is None:
79+
retry_info = next((event.get("payload", {}).get("retry", {}).get("codex_error_info")
80+
for event in reversed(events) if event.get("kind") == "agent.phase"
81+
and event.get("payload", {}).get("method") == "error"), None)
82+
timeout_info = next((event.get("payload", {}).get("timeout")
83+
for event in reversed(events) if event.get("kind") == "agent.phase"
84+
and event.get("payload", {}).get("method") == "native/timeout"), None) if source_code == "response_timeout" else None
85+
if source_code == "response_timeout" and isinstance(timeout_info, Mapping):
86+
# Use only the adapter's typed observation. Do not infer the deadline
87+
# from error prose or publish arbitrary host diagnostics.
88+
kind = timeout_info.get("kind")
89+
if isinstance(kind, str) and kind in {"idle", "hard"}:
90+
label = "等待原生助手有效进展超时" if kind == "idle" else "原生助手达到本次执行时限"
91+
count = timeout_info.get("native_error_count")
92+
observed = "期间原生助手记录了调用错误。" if type(count) is int and count > 0 else ""
93+
return code, (f"本次未能完成:{label}。{observed}"
94+
"没有生成完整答复,本次请求不会自动重放。")
7795
if code == "manager_authorization_unavailable" and preparation_recovery:
7896
return code, ("已收到你的消息,但本轮在准备授权上下文时失败。请求已保留;只有核验为模型调用前失败、"
7997
"连接恢复且没有后续重交时,原队列才会限次接续。可用 /status 查看状态。")
@@ -92,7 +110,11 @@ def manager_failure_reply(
92110
return code, "上游响应流曾中断,自动重试后本轮仍未完成。没有生成完整答复,本次请求不会自动重放。"
93111
if code == "server_overloaded":
94112
return code, "当前模型繁忙,本轮未完成。请先查看 /status 与已有结果,再决定是否稍后重试;本次请求不会自动重放。"
95-
return code, f"已收到你的消息,但本次未能完成:{labels.get(code, '管家处理失败')}。没有生成完整答复,本次请求不会自动重放。"
113+
# This owner also supplies failure codes to group admission. Improve the
114+
# display without promoting previously generic errors into routing input.
115+
label = labels.get(code) or {"response_timeout": "等待助手响应超时",
116+
"antigravity_step_error": "原生助手调用失败"}.get(source_code, "管家处理失败")
117+
return code, f"已收到你的消息,但本次未能完成:{label}。没有生成完整答复,本次请求不会自动重放。"
96118

97119

98120
def session_turn_effect(route: Mapping[str, Any]) -> dict[str, Any]:

‎loopx/extensions/lark/private_conversations.py‎

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2081,15 +2081,11 @@ def inbox() -> Path:
20812081
if not turn or turn["status"] not in TERMINAL_TURN_STATES:
20822082
return 0
20832083
if turn["status"] == "failed":
2084-
retry_info = None
2085-
if turn.get("error_code") == "host_gate":
2086-
events = self.core.controller.store.events_after(record["session_id"], record["turn_id"], None)
2087-
retry_info = next((event.get("payload", {}).get("retry", {}).get("codex_error_info")
2088-
for event in reversed(events) if event.get("kind") == "agent.phase"
2089-
and event.get("payload", {}).get("method") == "error"), None)
2084+
events = self.core.controller.store.events_after(record["session_id"], record["turn_id"], None
2085+
) if turn.get("error_code") in {"host_gate", "response_timeout"} else []
20902086
preparation_recovery = (record["source"].get("private_human_message") is True
20912087
and selected["binding"].get("context_kind") in {"project", "steward"} and not native.get("agent_target"))
2092-
response = manager_failure_reply(turn, retry_info=retry_info,
2088+
response = manager_failure_reply(turn, phase_events=events,
20932089
preparation_recovery=preparation_recovery)[1] + " 原会话已保留。"
20942090
else:
20952091
response = str((turn.get("response") or {}).get("message") or "") if turn["status"] == "completed" else (

‎tests/test_steward_interrupted_answer.py‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,126 @@
1616
from loopx.extensions.lark.conversation_identity import identity_ref
1717

1818

19+
@pytest.mark.parametrize("hard_limit, native_error", [(None, True), (150.0, True), (None, False)])
20+
def test_native_timeout_reports_its_clock_without_extending_native_retry(
21+
tmp_path, language_server, monkeypatch, hard_limit, native_error,
22+
):
23+
from test_antigravity_steward_adapter import _native_clock, adapter_for
24+
from loopx.chat_agent import CodexChatAgentError
25+
26+
user = {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE",
27+
"metadata": {"sourceTrajectoryStepInfo": {"stepIndex": 0}}}
28+
steps, frames = [user], []
29+
for idx, at in enumerate((1.0, 120.0, 240.0, 305.0), 1):
30+
if native_error:
31+
steps = [*steps, {"type": "CORTEX_STEP_TYPE_ERROR_MESSAGE", "status": "CORTEX_STEP_STATUS_DONE",
32+
"metadata": {"sourceTrajectoryStepInfo": {"stepIndex": idx}},
33+
"errorMessage": {"error": {"shortError": f"EOF {idx} NATIVE_PRIVATE_SENTINEL /private/host"}}}]
34+
frames.append({"at": at, "steps": steps, "status": "CASCADE_RUN_STATUS_RUNNING"})
35+
_native_clock(monkeypatch, language_server, frames)
36+
events = []
37+
with pytest.raises(CodexChatAgentError) as raised:
38+
adapter_for(tmp_path, hard_timeout_sec=hard_limit).start_turn(
39+
"Original task", lambda kind, data: events.append((kind, data)))
40+
assert raised.value.error_code == "response_timeout"
41+
diagnostic = next(data["timeout"] for kind, data in events
42+
if kind == "agent.phase" and data.get("method") == "native/timeout")
43+
assert diagnostic["kind"] == ("hard" if hard_limit is not None else "idle")
44+
assert diagnostic["limit_seconds"] == (hard_limit or 300.0)
45+
assert diagnostic["native_error_count"] == ((3 if hard_limit else 4) if native_error else 0)
46+
assert raised.value.native_error_info == {"timeout": diagnostic}
47+
assert language_server["now"] == (240.0 if hard_limit else 305.0)
48+
assert language_server["cancelled"].is_set()
49+
assert [method for method, _ in language_server["calls"]].count("SendUserCascadeMessage") == 1
50+
assert not any(kind in {"answer.final", "turn.native_result"} for kind, _ in events)
51+
serialized = json.dumps(events)
52+
assert "NATIVE_PRIVATE_SENTINEL" not in serialized and "/private/host" not in serialized
53+
errors = [data for kind, data in events if data.get("method") == "native/error"]
54+
assert len(errors) == diagnostic["native_error_count"]
55+
56+
57+
@pytest.mark.parametrize("kind, expected", [
58+
("idle", "等待原生助手有效进展超时"), ("hard", "原生助手达到本次执行时限"),
59+
("unknown-host-value", "等待助手响应超时"), (None, "等待助手响应超时"),
60+
])
61+
def test_original_lark_timeout_keeps_typed_reason_and_never_replays(
62+
steward, language_server, monkeypatch, kind, expected,
63+
):
64+
from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter
65+
from loopx.chat_agent import CodexChatAgentError
66+
from loopx.extensions.lark.private_conversations import LarkPrivateConversations
67+
68+
starts = []
69+
70+
def fail_after_observed_progress(_adapter, message, event_sink):
71+
starts.append(message)
72+
event_sink("turn.started", {"upstream_turn_id": "synthetic-timeout"})
73+
event_sink("agent.phase", {"method": "native/timeout", "label": "private label",
74+
"timeout": {"kind": kind, "native_error_count": 4, "raw": "NATIVE_PRIVATE_SENTINEL"}})
75+
raise CodexChatAgentError("Private host diagnostic NATIVE_PRIVATE_SENTINEL", gate=None,
76+
error_code="response_timeout")
77+
78+
monkeypatch.setattr(CodexAppServerAdapter, "start_turn", fail_after_observed_progress)
79+
store, runtime, provider, transport, binding, _, _ = steward
80+
native_route = kind in {"idle", "hard"}
81+
if native_route:
82+
from test_antigravity_steward_adapter import _native_clock, _native_frames
83+
monkeypatch.setenv("LOOPX_MANAGER_MODEL", "gemini-4-argon-high")
84+
monkeypatch.setenv("LOOPX_MANAGER_REASONING_EFFORT", "high")
85+
binding = transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"],
86+
executor_endpoint_id="antigravity", context_kind="steward")
87+
runtime.hard_timeout_sec = 150.0 if kind == "hard" else None
88+
frames = _native_frames(stagnant=True)
89+
for index, frame in enumerate(frames):
90+
frame["steps"] += [{"type": "CORTEX_STEP_TYPE_ERROR_MESSAGE", "status": "CORTEX_STEP_STATUS_DONE",
91+
"metadata": {"sourceTrajectoryStepInfo": {"stepIndex": 2 + attempt}},
92+
"errorMessage": {"error": {"shortError": "EOF NATIVE_PRIVATE_SENTINEL"}}}
93+
for attempt in range(index)]
94+
_native_clock(monkeypatch, language_server, frames)
95+
source = provider.event("steward-app", "native-timeout-source", "Complete this original request")
96+
transport.admit("steward-app", source)
97+
ref = identity_ref(binding["provider_ref"], source["message_id"])
98+
row = transport.core.read_request(ref)
99+
failed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=8)
100+
assert failed["status"] == "failed" and failed["error_code"] == "response_timeout"
101+
original = store._turn_path(row["session_id"], row["turn_id"]).read_bytes()
102+
assert transport.reconcile() == 1
103+
saved = json.loads((transport.root / f"{ref}.json").read_text())
104+
terminal = saved["deliveries"]["terminal"]
105+
assert terminal["verified"] and expected in terminal["text"]
106+
assert ("期间原生助手记录了调用错误" in terminal["text"]) is (kind in {"idle", "hard"})
107+
assert "不会自动重放" in terminal["text"]
108+
assert "NATIVE_PRIVATE_SENTINEL" not in terminal["text"] and "private label" not in terminal["text"]
109+
returned = provider.messages[terminal["attempt"]["message_ref"]]
110+
assert returned["chat_id"] == source["chat_id"]
111+
assert expected in json.dumps(returned["body"], ensure_ascii=False)
112+
restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
113+
runner=provider, cli_bin="lark-cli")
114+
before = list(provider.writes)
115+
assert restarted.reconcile() == 0
116+
assert restarted.health()[row["binding_id"]] == {"pending_count": 1, "recovery_count": 0}
117+
assert (not starts and [method for method, _ in language_server["calls"]].count("SendUserCascadeMessage") == 1
118+
if native_route else len(starts) == 1)
119+
assert provider.writes == before
120+
assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == original
121+
122+
123+
@pytest.mark.parametrize("info", [None, [], {"kind": []}, {"kind": {}}, {"kind": "idle", "native_error_count": True}])
124+
def test_timeout_presentation_does_not_trust_unknown_diagnostic_text(info):
125+
from loopx.extensions.lark.manager_context import manager_failure_reply
126+
code, text = manager_failure_reply({"error_code": "response_timeout", "error": "NATIVE_PRIVATE_SENTINEL"},
127+
phase_events=[{"kind": "agent.phase", "payload": {"method": "native/timeout", "timeout": info}}])
128+
assert code == "processing_failed" and "超时" in text and "不会自动重放" in text
129+
assert "NATIVE_PRIVATE_SENTINEL" not in text and "期间原生助手记录了调用错误" not in text
130+
131+
132+
@pytest.mark.parametrize("source_code", ["response_timeout", "antigravity_step_error", "unknown"])
133+
def test_native_error_labels_preserve_shared_admission_failure_category(source_code):
134+
from loopx.extensions.lark.manager_context import manager_failure_reply
135+
code, _ = manager_failure_reply({"error_code": source_code})
136+
assert code == "processing_failed"
137+
138+
19139
@pytest.mark.parametrize("fault, expected", [
20140
("done", True), ("error", True), ("canceled", True), ("cleared", True),
21141
("running", False), ("generating", False), ("waiting", False),

0 commit comments

Comments
 (0)