diff --git a/pyproject.toml b/pyproject.toml index 8827e33c17..e26d5fabf3 100644 --- a/pyproject.toml +++ b/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"] diff --git a/tests/control_plane/test_status_rollout_event_snapshot.py b/tests/control_plane/test_status_rollout_event_snapshot.py index 91a4befe84..ae2cf46451 100644 --- a/tests/control_plane/test_status_rollout_event_snapshot.py +++ b/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: diff --git a/tests/extensions/test_lark_reply_handoff.py b/tests/extensions/test_lark_reply_handoff.py index f345df1d3f..46d627159d 100644 --- a/tests/extensions/test_lark_reply_handoff.py +++ b/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"] diff --git a/tests/test_chat_delegation_journey.py b/tests/test_chat_delegation_journey.py index 96c93b6f44..7d78dc9ccf 100644 --- a/tests/test_chat_delegation_journey.py +++ b/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( diff --git a/tests/test_chat_manager_model_adoption.py b/tests/test_chat_manager_model_adoption.py index dc8a553276..38a93825c6 100644 --- a/tests/test_chat_manager_model_adoption.py +++ b/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" diff --git a/tests/test_chat_queue_preparation.py b/tests/test_chat_queue_preparation.py index 07ea8573de..d8621dc468 100644 --- a/tests/test_chat_queue_preparation.py +++ b/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" diff --git a/tests/test_native_steward_private.py b/tests/test_native_steward_private.py index 80594ada73..c4697fa787 100644 --- a/tests/test_native_steward_private.py +++ b/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":''')