fix(collaboration): reject ambiguous original Turn provenance

apps/presentation/dashboard/smoke/delegation-read-cancel.ts

@@ -135,6 +135,20 @@ for (const storage of ["file", "sqlite"]) {
     await fixture("continue"); await refresh.press("Enter"); await rows(61); await release();
     assert.equal(await status.count(), 0);
 
+    // Ambiguous original provenance must become partial coverage, never a
+    // directory-order choice. Other requests and exact storage recovery work.
+    await fixture("turn-identity-conflict"); await refresh.click(); await rows(60);
+    assert.equal(await original.count(), 0);
+    assert.equal(await dialog.getByRole("button", {name: "部分委派无法核验", exact: true}).count(), 1);
+    await page.setViewportSize({width: 1440, height: 1000});
+    await page.screenshot({path: resolve(output, `source-conflict-${storage}-desktop.png`)});
+    await page.setViewportSize({width: 390, height: 844});
+    assert.ok(await dialog.evaluate(node => node.scrollWidth <= node.clientWidth + 1));
+    await page.screenshot({path: resolve(output, `source-conflict-${storage}-narrow.png`)});
+    await fixture("turn-identity-recover"); await refresh.click(); await rows(61);
+    assert.deepEqual(await snapshot(), allRows);
+    assert.equal(await dialog.getByRole("button", {name: "部分委派无法核验", exact: true}).count(), 0);
+
     // Source loss still withholds content. Cancellation must not turn that
     // refusal into empty success or keep another source's previous records.
     await fixture("local-source-unavailable"); await refresh.click(); await dialog.getByRole("alert").waitFor();

examples/personal-workspace-browser/steward_requests_fixture.py

@@ -8,6 +8,7 @@
 local-source-recover, history-corrupt/history-recover (--long-history only),
 report-partial, complete-partial, report-readable, hold, hold-window, continue, release, stop. Window faults/returns apply
 after the first page to challenge multi-window refresh readback.
+turn-identity-conflict/turn-identity-recover inject and remove one owned historical Turn conflict.
 """
 from __future__ import annotations
 
@@ -321,6 +322,14 @@ def source_hashes():
                     local_session_path.write_text("{invalid synthetic session", encoding="utf-8")
                 elif command == "local-source-recover":
                     local_session_path.write_bytes(local_session_bytes)
+                elif command == "turn-identity-conflict":
+                    original = store.turn_for_client(session["session_id"], "synthetic-0")
+                    duplicate = store._turn_path(session["session_id"], "ambiguous-source")
+                    if original is None or duplicate.exists():
+                        raise ValueError("Synthetic source conflict is already present or unavailable")
+                    duplicate.write_text(json.dumps({**original, "turn_id": "ambiguous-source"}))
+                elif command == "turn-identity-recover":
+                    store._turn_path(session["session_id"], "ambiguous-source").unlink()
                 elif command != "inspect":
                     raise ValueError("Unknown fixture command")
                 print(json.dumps({"event": command, "posts": server.posts, "inventory_reads": server.inventory_reads,

loopx/control_plane/collaboration/source_chat_observation.py

@@ -32,20 +32,27 @@ def original_turn_reader(store: ChatSessionStore, session_id: str) -> Callable[[
     caller discards this locator after its read, including between pages.
     """
     turns_dir = store._session_dir(session_id) / "turns"
-    paths: dict[str, Path] = {}
+    paths: dict[str, Path | None] = {}
     for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []:
         if path.name.endswith(".events.json"):
             continue
         client_id = _read_json(path).get("client_turn_id")
         if isinstance(client_id, str):
-            paths.setdefault(client_id, path)
+            # Directory order cannot identify the original of two retained
+            # Turns claiming the same client identity. Leave it unavailable.
+            paths[client_id] = path if client_id not in paths else None
 
     def read(client_turn_id: str) -> dict[str, Any] | None:
         path = paths.get(client_turn_id)
         if path is None:
             return None
         payload = _read_json(path)
-        return payload if payload.get("client_turn_id") == client_turn_id else None
+        return payload if (
+            payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION
+            and payload.get("session_id") == session_id
+            and payload.get("turn_id") == path.stem
+            and payload.get("client_turn_id") == client_turn_id
+        ) else None
 
     return read
 

tests/control_plane/test_chat_request_inventory.py

@@ -253,6 +253,57 @@ def test_turn_locator_rereads_content_and_rejects_lost_identity(api):
     assert _inventory(api, route["session_id"])[1]["pending_count"] == 1
 
 
+@pytest.mark.parametrize("conflict", ["duplicate_client", "foreign_session", "foreign_turn", "unknown_schema"])
+def test_unproved_original_turn_is_withheld_without_hiding_other_requests(api, conflict):
+    import json
+
+    _, _, route = _source(api)
+    _, fixture, server = api
+    store = server.chat_store
+    session = store.load_session(route["session_id"])
+    other, _ = store.create_turn(session["session_id"], client_turn_id="unambiguous-source",
+        message="Check the other artifact", origin="web")
+    delivered = deliver(fixture["runtime"], fixture["registry"], session=session, turn=other,
+        request={"goal_id": fixture["ref"]["goal_id"], "agent_id": "agent-a"}, source_store=store)
+    store.update_turn(session["session_id"], other["turn_id"], status="completed",
+        response={"context_handoff_receipt": delivered})
+    store.finalize_managed_turn_completion(session["session_id"], other["turn_id"])
+    assert _inventory(api, session["session_id"])[1]["pending_count"] == 2
+
+    # Model retained historical corruption in this disposable Chat store only.
+    # A matching client identity cannot prove its schema, Session or Turn.
+    original = store.load_turn(session["session_id"], route["turn_id"])
+    original_path = store._turn_path(session["session_id"], route["turn_id"])
+    original_bytes = original_path.read_bytes()
+    damaged = store._turn_path(session["session_id"], "duplicate-source") if conflict == "duplicate_client" else original_path
+    patch = {"turn_id": "duplicate-source"} if conflict == "duplicate_client" else {
+        "foreign_session": {"session_id": "another-session"},
+        "foreign_turn": {"turn_id": "another-turn"},
+        "unknown_schema": {"schema_version": "unsupported-turn-schema"},
+    }[conflict]
+    damaged.write_text(json.dumps({**original, **patch}))
+    root = _root(fixture["runtime"])
+    before = {path: path.read_bytes() for path in root.rglob("*.json")}
+    turns_before = {path: path.read_bytes() for path in original_path.parent.glob("*.json")}
+    for query in ({}, {"view": "history"}, {"view": "history", "search": "Check the current contract"}):
+        code, value = _inventory(api, session["session_id"], **query)
+        assert code == 200, value
+        assert value["coverage"]["unreadable"] == 1, value
+        assert value["pending_count"] == 1, value
+        assert route["request_id"] not in {row["request_id"] for row in value["rows"]}
+        if "search" not in query:
+            assert [row["request_id"] for row in value["rows"]] == [delivered["request_id"]]
+    assert {path: path.read_bytes() for path in root.rglob("*.json")} == before
+    assert {path: path.read_bytes() for path in original_path.parent.glob("*.json")} == turns_before
+    if conflict == "duplicate_client":
+        damaged.unlink()
+    else:
+        original_path.write_bytes(original_bytes)
+    restored = _inventory(api, session["session_id"])[1]
+    assert restored["pending_count"] == 2 and restored["coverage"]["unreadable"] == 0
+    assert {row["request_id"] for row in restored["rows"]} == {route["request_id"], delivered["request_id"]}
+
+
 def test_refresh_window_preserves_full_pages_and_source_checks(api, monkeypatch):
     from loopx.presentation.request_control_api import RequestControlMixin
 

tests/control_plane/test_goal_create_completion.py

@@ -181,7 +181,7 @@ def test_managed_creation_retains_requirements_before_one_host_dispatch(environm
     original = getattr(target, method)
 
     def lose_response(*args, **kwargs):
-        result = original(*args, **kwargs)
+        original(*args, **kwargs)
         raise TimeoutError("Committed creation response was lost")
 
     monkeypatch.setattr(target, method, lose_response)