test: align release regressions with accepted readback contracts

pyproject.toml

@@ -172,8 +172,8 @@ include = ["loopx*"]
 
 [tool.pytest.ini_options]
 markers = ["stage2c_e2e: real process Stage 2C correctness and recovery acceptance"]
-# Private snapshots are not source tests; optional provider tests run in ark-turn.yml.
-norecursedirs = [".local", "deprecate", "packages/loopx-ark-turn"]
+# Private snapshots and packaging copies are not source tests; optional provider tests run in ark-turn.yml.
+norecursedirs = [".local", "build", "dist", "deprecate", "packages/loopx-ark-turn"]
 
 [tool.coverage.run]
 source = ["loopx"]

tests/control_plane/test_status_rollout_event_snapshot.py

@@ -10,6 +10,7 @@
 import loopx.status as status_module
 from loopx.control_plane import effect_runtime
 from loopx.control_plane.todos.todo_index import build_todo_index
+from loopx.control_plane.todos.resume_condition import evaluate_todo_resume_conditions
 from loopx.rollout_event_log import (
     ROLLOUT_EVENT_SCHEMA_VERSION,
     RolloutEventSnapshot,
@@ -135,7 +136,7 @@ def _indexed_todo_ids(payload: dict[str, Any]) -> set[str]:
     }
 
 
-def test_collect_status_reuses_one_goal_snapshot_and_next_request_is_fresh(
+def test_collect_status_reuses_display_snapshot_but_pr_wait_reads_complete_facts(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -181,7 +182,13 @@ def load_and_append(path: Path, *, limit: int | None = None):
 
     assert loads == [(GOAL_ID, 500)]
     assert _deferred_todo(first, "todo_wait_pr_41")["resume_ready"] is True
-    assert _deferred_todo(first, "todo_wait_pr_42")["resume_ready"] is False
+    # The retained PR proof is read by the typed Todo owner, independently of
+    # the bounded display snapshot. Evidence receipts and Todo index still
+    # share the original snapshot within this status request.
+    resumed = _deferred_todo(first, "todo_wait_pr_42")
+    assert resumed["resume_ready"] is True
+    assert resumed["resume_condition"]["matched_event_id"] == appended_events[0]["event_id"]
+    assert resumed["resume_condition"]["matched_pr_ref"] == f"{REPOSITORY}#42"
     assert [
         receipt["event_id"]
         for receipt in _goal_queue_item(first)["evidence_log_read_receipts"]
@@ -209,7 +216,7 @@ def load_and_append(path: Path, *, limit: int | None = None):
     assert "todo_event_appended" in _indexed_todo_ids(second)
 
 
-def test_explicit_empty_events_do_not_fall_back_to_disk(
+def test_empty_display_events_keep_complete_pr_proof_and_pure_evaluation_stays_empty(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -245,8 +252,20 @@ def fail_load(*_args: Any, **_kwargs: Any) -> list[dict[str, Any]]:
         for item in fresh["agent_todos"]["deferred_items"]
         if item["todo_id"] == "todo_wait_pr_41"
     )
-    assert empty_todo["resume_ready"] is False
+    # Empty display history must not truncate the retained authority for a PR
+    # dependency. The Python display loader remains unused as asserted above.
+    assert empty_todo["resume_ready"] is True
     assert fresh_todo["resume_ready"] is True
+    assert empty_todo["resume_condition"]["matched_pr_ref"] == f"{REPOSITORY}#41"
+    assert empty_todo["resume_condition"]["matched_event_id"] == fresh_todo["resume_condition"]["matched_event_id"]
+    # The pure evaluator has no local source pointer and must still respect
+    # explicitly empty evidence; it cannot invent a merge or read the log.
+    waiting = {"todo_id": "todo_wait_pr_41", "role": "agent", "status": "deferred",
+               "task_class": "advancement_task", "resume_when": "pr_merged:#41",
+               "task_repository": f"git:github.com/{REPOSITORY}"}
+    pure = evaluate_todo_resume_conditions([waiting], source_items=[waiting], rollout_events=[])
+    assert pure[waiting["todo_id"]]["satisfied"] is False
+    assert "matched_event_id" not in pure[waiting["todo_id"]]
 
 
 def test_direct_todo_index_calls_keep_fresh_disk_reads(tmp_path: Path) -> None:

tests/extensions/test_lark_reply_handoff.py

@@ -123,12 +123,11 @@ def test_short_reply_handoff_preserves_source_and_returns_once(
         answer = answer_lark_goal_topic(**options)
         assert "**已转交给 `worker`。**" in answer
         assert "是否已开始处理尚未核实" in answer
-        if context_state == "near_record_limit":
-            assert "完整简报已投递" in answer
-        else:
-            assert handoff_target["brief"]["purpose"] in answer
-            assert handoff_target["brief"]["context"] not in answer
-            assert "Do not publish." in answer
+        # The audience-processed model answer survives the handoff. The full
+        # brief is read back from the recipient inbox, not duplicated in Chat.
+        assert answer.startswith("Preparing handoff.\n\n")
+        assert answer.count("Preparing handoff.") == 1
+        assert handoff_target["brief"]["context"] not in answer
         entry, = pending(tmp_path, **target)["items"]
         assert entry["brief"] == handoff_target["brief"]
         forwarded = entry["message"]

tests/test_chat_delegation_journey.py

@@ -134,10 +134,9 @@ def start_turn(self, message, sink):
             assert completed["status"] == "completed", completed
             reply = completed["response"]["message"]
             assert "**已转交给 `writer`。**" in reply
-            assert drafts[index][1] in reply
+            assert reply.startswith("Passing the brief to the existing owner.\n\n")
+            assert reply.count("Passing the brief to the existing owner.") == 1
             assert "是否已开始处理尚未核实" in reply
-            if index:
-                assert "Chinese first" in reply and "Do not publish" in reply
             receipt = completed["response"]["context_handoff_receipt"]
             assert receipt["agent_id"] == "writer"
             assert not completed["response"]["proposals"]
@@ -158,6 +157,7 @@ def start_turn(self, message, sink):
                 if row["request_id"] == receipt["request_id"]
             )
             assert received["message"] == message
+            assert received["brief"]["purpose"] == drafts[index][1]
             assert received["brief"]["constraints"] == drafts[index][2]
             # This is a receiver decision, separate from supplied/delivered.
             acknowledge(

tests/test_chat_manager_model_adoption.py

@@ -107,7 +107,7 @@ def test_idle_conversation_adopts_model_preserving_context_and_authority(convers
     assert len(starts) == 2
 
 
-@pytest.mark.parametrize("turn_status", ["queued", "starting", "running", "interrupting"])
+@pytest.mark.parametrize("turn_status", ["starting", "running", "interrupting"])
 def test_model_edit_never_replaces_an_in_flight_turn(conversation, turn_status):
     root, store, runtime, session, starts = conversation
     # Store transitions are covered independently; this fixture models the
@@ -122,16 +122,19 @@ def test_model_edit_never_replaces_an_in_flight_turn(conversation, turn_status):
     assert store.load_turn(session["session_id"], turn["turn_id"])["status"] == turn_status
 
 
-def test_accepted_queued_turn_can_adopt_before_starting_upstream(conversation):
+@pytest.mark.parametrize("explicit_acceptance", [False, True])
+def test_accepted_queued_turn_can_adopt_before_starting_upstream(conversation, explicit_acceptance):
     root, store, runtime, session, starts = conversation
     turn, _ = store.create_turn(session["session_id"], client_turn_id="next", message="Continue.")
-    queued = store.load_session(session["session_id"])
+    queued = store.update_session(session["session_id"], active_turn_id=turn["turn_id"], status="busy")
     configure(root, model="gpt-6.1-sol", effort="high")
     with runtime._session_adapter_lock(session["session_id"]):
         runtime._ensure_adapter_locked(
-            queued, work_dir=root, objective="manager", accepted_turn_id=turn["turn_id"],
+            queued, work_dir=root, objective="manager",
+            **({"accepted_turn_id": turn["turn_id"]} if explicit_acceptance else {}),
         )
-    assert len(starts) == 2
+    assert len(starts) == 2 and starts[0][1].closed
+    assert store.load_session(session["session_id"])["active_turn_id"] == turn["turn_id"]
     assert starts[1][0]["executor_model"]["model"] == "gpt-6.1-sol"
     assert store.load_turn(session["session_id"], turn["turn_id"])["status"] == "queued"
 

tests/test_chat_queue_preparation.py

@@ -6,6 +6,7 @@
 import sys
 import threading
 import time
+import urllib.error
 import urllib.request
 
 import pytest
@@ -252,8 +253,8 @@ def preparation(*args, **kwargs):
 
 
 @pytest.mark.parametrize("delay_stage", ["context", "provider"])
-def test_http_stop_remains_effective_after_worker_wait_expires(tmp_path, monkeypatch, delay_stage):
-    """A late worker cannot dispatch or hand off after the stop receipt commits."""
+def test_http_stop_retains_fence_until_late_worker_acknowledges(tmp_path, monkeypatch, delay_stage):
+    """A failed stop retains authority; retry settles only after the worker exits."""
     import loopx.chat_coordination as coordination
     from loopx.capabilities import manager_context
 
@@ -304,17 +305,27 @@ def context(controller, adapter, session, turn_id, sink, **kwargs):
             f"http://127.0.0.1:{server.server_port}/api/chat/sessions/{sid}/turns/{tid}/interrupt",
             data=b"{}", headers={"Content-Type": "application/json"}, method="POST",
         )
+        with pytest.raises(urllib.error.HTTPError) as rejected:
+            urllib.request.urlopen(request, timeout=10)
+        assert rejected.value.code == 424
+        rejected.value.close()
+        assert not old_worker_done.is_set(), "the late worker must still be blocked"
+        assert (sid, tid) in runtime.cancelled_turns
+        assert store.load_turn(sid, tid)["status"] == "interrupting"
+        assert store.load_session(sid)["active_turn_id"] == tid
+        assert not any(e["kind"] == "turn.interrupted" for e in store.events_after(sid, tid, None))
+        replay, created = runtime.submit_turn(session_id=sid, client_turn_id="old-request", message="Old request", work_dir=tmp_path, objective="Research")
+        assert not created and replay["status"] == "interrupting"
+        release.set()
+        assert old_worker_done.wait(3)
         with urllib.request.urlopen(request, timeout=10) as response:
             assert response.status == 200
             assert json.load(response)["status"] == "interrupted"
-        assert not old_worker_done.is_set(), "the late worker must still be blocked"
         assert (sid, tid) not in runtime.cancelled_turns
-        fresh, _ = runtime.submit_turn(session_id=sid, client_turn_id="fresh-request", message="Fresh request", work_dir=tmp_path, objective="Research")
-        assert runtime.wait_for_turn(session_id=sid, turn_id=fresh["turn_id"], timeout_sec=3)["response"]["message"] == "Fresh result"
         replay, created = runtime.submit_turn(session_id=sid, client_turn_id="old-request", message="Old request", work_dir=tmp_path, objective="Research")
         assert not created and replay["status"] == "interrupted"
-        release.set()
-        assert old_worker_done.wait(3)
+        fresh, _ = runtime.submit_turn(session_id=sid, client_turn_id="fresh-request", message="Fresh request", work_dir=tmp_path, objective="Research")
+        assert runtime.wait_for_turn(session_id=sid, turn_id=fresh["turn_id"], timeout_sec=3)["response"]["message"] == "Fresh result"
         assert len(dispatches) == (1 if delay_stage == "context" else 2)
         assert handoffs == []
         assert store.load_turn(sid, tid)["status"] == "interrupted"

tests/test_native_steward_private.py

@@ -48,11 +48,13 @@ def steward(tmp_path):
         if native_goal["status"] == "active":
             turn = "native-first"
             active_turn = turn
+            turn_status[turn] = "inProgress"
             print(json.dumps({"method": "turn/started", "params": {"threadId": "durable-thread", "turn": {"id": turn}}}), flush=True)
             if "wait for interrupt" in str(native_goal.get("objective")):
                 continue
             print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": turn, "delta": "Verified synthetic result: the workspace has one README and no inherited portfolio."}}), flush=True)
             native_goal["status"] = "complete"
+            turn_status[turn] = "completed"
             print(json.dumps({"method": "turn/completed", "params": {"threadId": "durable-thread", "turn": {"id": turn, "status": "completed"}}}), flush=True)
         continue
     elif method == "turn/start":''')