Skip to content

Commit 8da34be

Browse files
authored
fix(chat): recover prepared requests under fresh binding fences (#6695)
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
1 parent 04b929a commit 8da34be

5 files changed

Lines changed: 136 additions & 5 deletions

File tree

‎docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,11 @@ without private bindings do not add provider authentication; configured aliases
7272
reuse one request-scoped verified identity observation. Lark HTTP composition
7373
resides in the extension, while the typed binding owner remains provider-neutral.
7474

75+
Binding resolution rereads the exact grant after provider identity observation.
76+
Revoking or replacing that binding during the observation rejects the request;
77+
an unrelated binding edit does not invalidate it. The typed owner still checks
78+
the current context, with no identity cache shared across verification calls.
79+
7580
The same Settings read model includes Core bindings for locally installed
7681
conversation transports. Listener status comes from the exact host-composed
7782
provider's optional `health_snapshot()` returning a content-free `status`, or
@@ -128,7 +133,10 @@ transport resume can fail. Native observation may only narrow that frozen set;
128133
a retry cannot add later completed/active Turns or switch to a newer Session.
129134
The existing Lark listener returns after this durable freeze; its existing reply
130135
workers recover the prepared request to observe and cancel native background
131-
commands. Slow or unavailable native control therefore does not hold inbound
136+
commands. Prepared recovery uses admission's fresh authority check inside both
137+
the request and source-route fences, without a duplicate provider probe before
138+
waiting for those fences. Revocation while waiting prevents control and delivery.
139+
Slow or unavailable native control therefore does not hold inbound
132140
admission. This is scheduling only: it introduces no queue or completion state,
133141
does not defer an active Turn's exact interrupt, and sends no successful stop
134142
reply until native control and the original delivery checks have succeeded.

‎docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,9 @@ provider 读回确认回复;发生没有 receipt 的不确定写入时不盲
5151
时,同一请求内的别名检查与连接共用一次已验证身份观测。飞书 HTTP 组合位于
5252
extension,typed binding owner 继续保持 provider-neutral。
5353

54+
绑定解析在 provider 身份观测后重读确切授权。观测期间撤销或替换该绑定会拒绝请求;
55+
其它绑定的修改不使当前绑定失效。typed owner 仍检查当前上下文,不跨次缓存身份核验。
56+
5457
同一设置读模型也展示本机已安装 transport 的 Core binding。监听状态取确切宿主组合
5558
provider 的可选 `health_snapshot()`,返回仅含状态的观测;默认 profile 使用已有
5659
Lark listener 观测。该本地 hook 只读取缓存,不发起网络请求。仅投影已有监听标签
@@ -89,7 +92,9 @@ App 身份时,并行读取两条原 provider 消息。两项只读观测全部
8992
普通 Codex 话题空闲时,`/stop` 也覆盖同一原生 Session 中已完成 Turn 遗留的后台命令。
9093
Core 先在原请求日志固定 Session 与候选 Turn,原 listener 随即继续受理消息;现有
9194
回复工作池再恢复这条 prepared 请求,读取原生命令清单并取消原目标。恢复只能缩小
92-
固定候选,不能追停后来的 Turn 或新 Session。慢速或失败的原生控制不阻塞入站,
95+
固定候选,不能追停后来的 Turn 或新 Session。prepared 恢复复用受理路径在原请求与
96+
来源路由锁内的新鲜授权核验,不在等待锁前重复探测 provider;等待期间撤权即阻止
97+
控制和投递。慢速或失败的原生控制不阻塞入站,
9398
也不提前回复“已停止”;没有固定目标的旧请求保持未知,活跃 Turn 仍沿原确切中断路径。
9499
这只调整现有 IO 调度,不新增队列、消费者或完成状态;实际安装、停止时延与完整社区
95100
准出仍需分别验收。

‎loopx/capabilities/native_chat/conversation_bindings.py‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -183,10 +183,17 @@ def resolve(self, *, binding_id: str, source_ref: str, sender_ref: str,
183183
row = next((item for item in current.get("bindings", []) if item.get("binding_id") == binding_id), None)
184184
if row is None:
185185
raise ValueError("conversation binding is no longer authorized")
186+
observation = dict(self.observe(row["transport_ref"]))
187+
# Provider verification can wait on IO. It cannot authorize a binding
188+
# that was revoked or replaced while that observation was in flight.
189+
# Unrelated binding changes do not invalidate this exact source.
190+
current = self.read()
191+
if next((item for item in current.get("bindings", [])
192+
if item.get("binding_id") == binding_id), None) != row:
193+
raise ValueError("conversation binding changed during identity verification")
186194
registry_scope = {}
187195
if row.get("goal_scope") == "all_registered":
188196
registry_scope["available_goal_ids"] = self.goal_scope_ids(row)
189-
observation = dict(self.observe(row["transport_ref"]))
190197
selected = self._core("collaboration.conversation.bound_context", {
191198
"current": current, "binding_id": binding_id, "source_ref": source_ref, "sender_ref": sender_ref,
192199
"private_human_message": private_human_message, "observation": observation,

‎loopx/capabilities/native_chat/external_conversations.py‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -610,12 +610,16 @@ def recover(self, *, request_ref: str | None = None) -> None:
610610
# delivering their pending response; do not add an idle probe.
611611
if row["status"] == "command_completed" and not row.get("commission_adoption_pending"):
612612
continue
613-
self.bindings.resolve(binding_id=row["binding_id"], **row["source"])
614613
if row["status"] == "prepared":
614+
# Admission resolves current authority after both request
615+
# and route fences. A separate pre-fence network probe
616+
# neither authorizes recovery nor needs to precede it.
615617
self.admit(binding_id=row["binding_id"], source=row["source"], request_ref=row["request_ref"],
616618
message=row["message"], command=row["command"],
617619
attachments=row.get("attachments"), origin=row.get("origin", "lark"))
618-
elif row["status"] == "command_queued":
620+
continue
621+
self.bindings.resolve(binding_id=row["binding_id"], **row["source"])
622+
if row["status"] == "command_queued":
619623
self._run_steward_action(row)
620624
elif row["status"] == "accepted":
621625
if row.get("agent_target"):

‎tests/test_chat_conversation_bindings.py‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -259,6 +259,113 @@ def observe(profile):
259259
runtime.close()
260260

261261

262+
@pytest.mark.parametrize("group", [False, True])
263+
@pytest.mark.parametrize("change", ["revoke", "replace", "unrelated"])
264+
def test_binding_change_while_identity_is_observed_uses_current_grant(ordinary, monkeypatch, group, change): # noqa: F811
265+
from test_lark_private_conversations import connect, connect_group
266+
from loopx.extensions.lark.conversation_identity import lark_private_source
267+
268+
store, runtime, provider, transport = (connect_group if group else connect)(ordinary)
269+
bindings = transport.bindings
270+
binding = transport._binding("notes-app")
271+
event = provider.topic("grant-observation", "Read project") if group else provider.event("notes-app", "grant-observation", "Read project")
272+
source = lark_private_source(provider_ref=binding["provider_ref"], event=event)
273+
observe, armed = bindings.observe, True
274+
275+
def changed(profile, **kwargs):
276+
nonlocal armed
277+
result = observe(profile, **kwargs)
278+
if armed:
279+
armed = False
280+
if change == "revoke":
281+
bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"])
282+
else:
283+
target = "steward-app" if change == "unrelated" else "notes-app"
284+
if change == "replace":
285+
provider.profile_apps[target] = "cli_replaced_app"
286+
bindings.configure(transport_ref=target, project_ref=binding["project_ref"],
287+
executor_endpoint_id="codex", audience="group" if group else None,
288+
group_refs=binding.get("group_refs"), available_group_refs=binding.get("group_refs"))
289+
return result
290+
291+
monkeypatch.setattr(bindings, "observe", changed)
292+
try:
293+
if change == "unrelated":
294+
selected = bindings.resolve(binding_id=binding["binding_id"], **source)
295+
assert selected["binding"] == binding
296+
else:
297+
with pytest.raises(ValueError, match="binding|identity|authorized"):
298+
bindings.resolve(binding_id=binding["binding_id"], **source)
299+
assert not any(row["enabled"] and row["binding_id"] == binding["binding_id"]
300+
for row in bindings.read()["bindings"])
301+
assert not store.list_sessions() and not provider.writes
302+
finally:
303+
runtime.close()
304+
305+
306+
@pytest.mark.parametrize("revoke", [False, True])
307+
def test_prepared_stop_recovery_waits_for_route_fence_before_its_identity_probe(ordinary, monkeypatch, revoke): # noqa: F811
308+
import threading
309+
from concurrent.futures import ThreadPoolExecutor
310+
from test_chat_completed_background_stop import _completed_with_later_answer
311+
from loopx.file_lock import exclusive_file_lock
312+
313+
store, runtime, provider, transport, root, first, wire = _completed_with_later_answer(ordinary, monkeypatch)
314+
stop = provider.topic("fenced-recovery-stop", "/stop", root=root["message_id"])
315+
transport.admit("notes-app", stop, defer_background_stop=True)
316+
frozen = next(row for row in transport.core.pending() if row["command"] == "stop")
317+
assert frozen["status"] == "prepared" and frozen["target_recorded"]
318+
fence = transport.core.root / "source-fences" / f"{frozen['binding_id']}.{frozen['source']['source_ref']}.json"
319+
started, observed = threading.Event(), threading.Event()
320+
observe, finish = transport.bindings.observe, transport.core._finish_background_stop
321+
before_native = []
322+
323+
def observing(*args, **kwargs):
324+
observed.set()
325+
return observe(*args, **kwargs)
326+
327+
def finishing(*args, **kwargs):
328+
before_native.append(sum("auth" in call for call in provider.calls))
329+
return finish(*args, **kwargs)
330+
331+
def recover():
332+
started.set()
333+
transport.core.recover(request_ref=frozen["request_ref"])
334+
335+
monkeypatch.setattr(transport.bindings, "observe", observing)
336+
monkeypatch.setattr(transport.core, "_finish_background_stop", finishing)
337+
provider.calls.clear()
338+
writes_before = list(provider.writes)
339+
receipts = {turn_id: store.load_turn(first["session_id"], turn_id)
340+
for turn_id in frozen["background_turn_ids"]}
341+
try:
342+
with ThreadPoolExecutor(max_workers=1) as pool:
343+
with exclusive_file_lock(fence, operation="hold_original_source_route"):
344+
future = pool.submit(recover)
345+
assert started.wait(5)
346+
assert not observed.wait(.1), "A pre-fence probe cannot authorize prepared recovery"
347+
if revoke:
348+
transport.bindings.disconnect(frozen["binding_id"], expected_revision=transport.bindings.read()["revision"])
349+
future.result(timeout=10)
350+
current = transport.core.read_request(frozen["request_ref"])
351+
if revoke:
352+
assert current["status"] == "prepared" and current["background_turn_ids"] == frozen["background_turn_ids"]
353+
assert not before_native and wire.children["owned"].poll() is None
354+
assert provider.writes == writes_before
355+
else:
356+
assert before_native == [1], "The fenced admission already owns the fresh identity proof"
357+
assert current["status"] == "command_completed" and current["background_turn_ids"] == [first["turn_id"]]
358+
assert wire.children["owned"].poll() is not None
359+
transport.reconcile()
360+
assert sum("停止" in text for _, text in provider.writes) == 1
361+
assert wire.children["unowned"].poll() is None
362+
assert all(store.load_turn(first["session_id"], turn_id) == receipt
363+
for turn_id, receipt in receipts.items())
364+
finally:
365+
wire.close()
366+
runtime.close()
367+
368+
262369
@pytest.mark.parametrize("resumed", [False, True])
263370
def test_group_disconnect_stops_running_and_queued_turns_without_stopping_other_session(ordinary, resumed): # noqa: F811
264371
from test_lark_private_conversations import connect_group

0 commit comments

Comments
 (0)