diff --git a/apps/presentation/dashboard/smoke/delegation-read-cancel.ts b/apps/presentation/dashboard/smoke/delegation-read-cancel.ts index 53fe4739cc..6ab8c6998b 100644 --- a/apps/presentation/dashboard/smoke/delegation-read-cancel.ts +++ b/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(); diff --git a/examples/personal-workspace-browser/steward_requests_fixture.py b/examples/personal-workspace-browser/steward_requests_fixture.py index e9d2107ff4..035a6daaab 100644 --- a/examples/personal-workspace-browser/steward_requests_fixture.py +++ b/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, diff --git a/loopx/control_plane/collaboration/source_chat_observation.py b/loopx/control_plane/collaboration/source_chat_observation.py index 466b2eb984..2552904432 100644 --- a/loopx/control_plane/collaboration/source_chat_observation.py +++ b/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 diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py index d714b7cd9d..b845b4a559 100644 --- a/tests/control_plane/test_chat_request_inventory.py +++ b/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 diff --git a/tests/control_plane/test_goal_create_completion.py b/tests/control_plane/test_goal_create_completion.py index c03bb20600..18a0af6e7a 100644 --- a/tests/control_plane/test_goal_create_completion.py +++ b/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)