diff --git a/loopx/control_plane/scheduler/monitor_poll_writeback.py b/loopx/control_plane/scheduler/monitor_poll_writeback.py index 2629e94b4f..285fae8242 100644 --- a/loopx/control_plane/scheduler/monitor_poll_writeback.py +++ b/loopx/control_plane/scheduler/monitor_poll_writeback.py @@ -1,5 +1,6 @@ from __future__ import annotations +from copy import deepcopy from pathlib import Path from typing import Any from uuid import uuid4 @@ -42,16 +43,28 @@ def resolve_monitor_todo_item( for item in authority["todos"] if item.get("role") == "agent" and item.get("archive_state") != "archive" ] + # DTO only: selectMonitorTodo remains the rule owner. Preserve the complete + # candidate set, but keep prose/evidence and schedule fields in the source + # snapshot. The selected fact is mapped back to its complete original row. + fields = ("todo_id", "target_key", "status", "archive_state", "role", "task_class") + facts = [{key: item[key] for key in fields if key in item} for item in items] try: result = effect_runtime_result("scheduler.monitor_target.select", { - "schema_version": "loopx_monitor_target_request_v0", "items": items, + "schema_version": "loopx_monitor_target_request_v0", "items": facts, "todo_id": normalized_todo_id, "target_key": safe_target_key or None, }) except EffectRuntimeRejected as exc: + # A source/transport outage is not a definitive target rejection. + # Due lookup catches ValueError for semantic misses only. + if exc.diagnostic_code != "invalid_request": + raise raise ValueError(str(exc)) from exc if not isinstance(result, dict) or result.get("schema_version") != "loopx_monitor_target_result_v0": raise TypeError("TypeScript Monitor target selection shape mismatch") - return result["todo"] + for item, fact in zip(items, facts, strict=True): + if fact == result.get("todo"): + return deepcopy(item) + raise TypeError("TypeScript Monitor target selection does not match its source snapshot") def write_monitor_poll_todo_state( diff --git a/tests/control_plane/test_monitor_target_transport.py b/tests/control_plane/test_monitor_target_transport.py new file mode 100644 index 0000000000..0fd50b283e --- /dev/null +++ b/tests/control_plane/test_monitor_target_transport.py @@ -0,0 +1,181 @@ +"""Monitor target lookup retains full records without transporting their prose.""" +from __future__ import annotations + +from copy import deepcopy +from datetime import datetime, timezone +import json + +import pytest + +from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from test_monitor_followthrough_contract import AGENT_ID, GOAL_ID, _write_fixture +from loopx.control_plane.coordination.local_authority import ( + LocalCoordinationAuthorityUnavailable, read_canonical_todos_if_promoted, +) +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.control_plane.effect_runtime import EffectRuntimeRejected, MAX_REQUEST_BYTES +from loopx.control_plane.quota.monitor_poll import resolve_due_monitor_candidate +from loopx.control_plane.scheduler import monitor_poll_writeback, monitor_todo +from loopx.control_plane.testing.canary_harness import run_json_cli + + +def _monitor(**fields): + return { + "schema_version": "todo_item_v0", "source_section": "Agent Todo", "index": 321, + "todo_id": "todo_transport_watch", "role": "agent", "status": "open", + "task_class": "continuous_monitor", "archive_state": "active", "done": False, + "text": "Observe an external revision", "claimed_by": AGENT_ID, + "target_key": "transport-watch", "cadence": "1h", + "next_due_at": "2000-01-01T00:00:00Z", **fields, + } + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("target_repetitions", [1, 40_000, 120_000]) +def test_real_large_authority_returns_complete_target(tmp_path, monkeypatch, provider, target_repetitions): + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, runtime, state = _write_fixture(tmp_path, canonical=False) + note = "Measured evidence 演示 " * target_repetitions + if target_repetitions == 120_000: + assert len(note.encode()) > MAX_REQUEST_BYTES + target = _monitor(note=note, evidence="source:retained-result", + required_capabilities=["network"], resume_when="todo_done:todo_input", + resume_ready=True, expires_at="2099-01-01T00:00:00Z") + items = [ + {"schema_version": "todo_item_v0", "source_section": "Agent Todo", "index": i + 1, + "todo_id": f"todo_history_{i}", "role": "agent", "status": "done", + "done": True, "task_class": "advancement_task", "archive_state": "active", + "text": "Retained task", "note": "evidence 字 " * 650} + for i in range(320) + ] + [target] + items[0]["todo_id"] = "todo_input" + assert len(json.dumps(items, ensure_ascii=False).encode()) > MAX_REQUEST_BYTES + projection = build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=items, + handoff_mode="soft_claim") + initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider) + state.unlink() # No legacy Markdown fallback can satisfy the lookup. + + # The canonical source has its own per-record page ceiling. Keep that + # refusal; this transport repair must not loosen it or call it "not due". + if target_repetitions == 120_000: + with pytest.raises(LocalCoordinationAuthorityUnavailable) as error: + resolve_due_monitor_candidate(registry_path=registry, runtime_root=runtime, + goal_id=GOAL_ID, todo_id=target["todo_id"], target_key=None) + assert error.value.code == "canonical_snapshot_record_too_large" + assert not state.exists() + return + + selected = monitor_poll_writeback.resolve_monitor_todo_item( + registry_path=registry, runtime_root=runtime, goal_id=GOAL_ID, + todo_id=target["todo_id"], target_key=target["target_key"], + ) + for field in ("todo_id", "note", "evidence", "claimed_by", "required_capabilities", + "resume_when", "resume_ready", "next_due_at", "expires_at"): + assert selected[field] == target[field] + assert resolve_due_monitor_candidate(registry_path=registry, runtime_root=runtime, + goal_id=GOAL_ID, todo_id=target["todo_id"], target_key=None) == { + "due": True, "todo_id": target["todo_id"], "target_key": target["target_key"], + "task_class": "continuous_monitor", "claimed_by": AGENT_ID, + } + assert not state.exists() + if target_repetitions == 1: + scope = ["--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--runtime-profile", "codex_app_ssh_goal", + "--turn-instance-id", "monitor-large-history"] + guard = ["quota", "should-run", *scope, + "--available-capability", "network"] + admitted = run_json_cli(*guard, registry_path=registry, runtime_root=runtime) + assert admitted["selected_todo"]["todo_id"] == target["todo_id"] + poll = ["quota", "monitor-poll", *scope, "--todo-id", target["todo_id"], + "--target-key", target["target_key"], "--result-hash", "observed-v2", + "--material-change", "--next-due-at", "2099-01-01T00:00:00Z", "--execute"] + first = run_json_cli(*poll, registry_path=registry, runtime_root=runtime) + assert first["turn_continuation"]["current_turn_settled"] is True + replay = run_json_cli(*poll, registry_path=registry, runtime_root=runtime) + assert replay["replayed"] is True and replay["appended"] is False + closed = run_json_cli(*guard, registry_path=registry, runtime_root=runtime) + assert closed["effective_action"] == "heartbeat_settled_skip" + canonical = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID) + assert canonical["source_authority"] == ("sqlite_v0" if provider == "sqlite" else "file_v0") + assert len(canonical["todos"]) == 321 + persisted = next(row for row in canonical["todos"] if row["todo_id"] == target["todo_id"]) + assert persisted["material_change_generation"] == 1 + assert persisted["note"] == note and persisted["result_hash"] == "observed-v2" + events = [json.loads(line) for line in + (runtime / "goals" / GOAL_ID / "runs" / "index.jsonl").read_text().splitlines()] + assert sum(row["classification"] == "quota_monitor_poll" for row in events) == 1 + assert not any(row["classification"] == "quota_slot_spent" for row in events) + + +@pytest.mark.parametrize(("overrides", "message"), [ + ({"status": "done", "done": True}, "unfinished"), + ({"role": "user"}, "not found"), + ({"archive_state": "archive"}, "not found"), + ({"task_class": "advancement_task"}, "continuous_monitor"), +]) +def test_target_rejections_retain_native_rules(tmp_path, monkeypatch, overrides, message): + source = [_monitor(**overrides)] + monkeypatch.setattr(monitor_poll_writeback, "read_canonical_todos_if_promoted", + lambda **_: {"todos": source}) + with pytest.raises(ValueError, match=message): + monitor_poll_writeback.resolve_monitor_todo_item(registry_path=tmp_path / "registry.json", + runtime_root=tmp_path, goal_id=GOAL_ID, todo_id="todo_transport_watch") + + +def test_duplicate_key_exact_identity_and_full_record_isolation(tmp_path, monkeypatch): + source = [_monitor(), _monitor(todo_id="todo_other", note="second source", + required_capabilities=["network"])] + original = deepcopy(source) + monkeypatch.setattr(monitor_poll_writeback, "read_canonical_todos_if_promoted", + lambda **_: {"todos": source}) + args = dict(registry_path=tmp_path / "registry.json", runtime_root=tmp_path, goal_id=GOAL_ID) + with pytest.raises(ValueError, match="multiple"): + monitor_poll_writeback.resolve_monitor_todo_item(**args, target_key="transport-watch") + with pytest.raises(ValueError, match="resolves to"): + monitor_poll_writeback.resolve_monitor_todo_item(**args, todo_id="todo_other", target_key="wrong") + selected = monitor_poll_writeback.resolve_monitor_todo_item(**args, todo_id="todo_other", + target_key="transport-watch") + assert selected == source[1] + selected["required_capabilities"].append("synthetic-change") + assert source == original + + +@pytest.mark.parametrize("overrides", [ + {"next_due_at": "2099-01-01T00:00:00Z"}, + {"expires_at": "2000-01-01T00:00:00Z"}, + {"resume_when": "todo_done:todo_input", "resume_ready": False}, + {"status": "blocked"}, {"done": True}, +]) +def test_original_schedule_and_wait_fields_still_control_due(tmp_path, monkeypatch, overrides): + source = [_monitor(**overrides)] + monkeypatch.setattr(monitor_poll_writeback, "read_canonical_todos_if_promoted", + lambda **_: {"todos": source}) + monkeypatch.setattr(monitor_todo, "now_utc", lambda: datetime(2030, 1, 1, tzinfo=timezone.utc)) + assert resolve_due_monitor_candidate(registry_path=tmp_path / "registry.json", + runtime_root=tmp_path, goal_id=GOAL_ID, todo_id="todo_transport_watch", target_key=None) == {} + + +@pytest.mark.parametrize("code", ["request_too_large", "authentication_failed"]) +def test_transport_rejection_is_not_a_not_due_result(tmp_path, monkeypatch, code): + monkeypatch.setattr(monitor_poll_writeback, "read_canonical_todos_if_promoted", + lambda **_: {"todos": [_monitor()]}) + + def unavailable(*_args, **_kwargs): + raise EffectRuntimeRejected("Synthetic transport rejection", diagnostic_code=code) + + monkeypatch.setattr(monitor_poll_writeback, "effect_runtime_result", unavailable) + with pytest.raises(EffectRuntimeRejected) as failure: + resolve_due_monitor_candidate(registry_path=tmp_path / "registry.json", + runtime_root=tmp_path, goal_id=GOAL_ID, todo_id="todo_transport_watch", target_key=None) + assert failure.value.diagnostic_code == code + + +def test_foreign_result_cannot_replace_the_source_record(tmp_path, monkeypatch): + monkeypatch.setattr(monitor_poll_writeback, "read_canonical_todos_if_promoted", + lambda **_: {"todos": [_monitor()]}) + monkeypatch.setattr(monitor_poll_writeback, "effect_runtime_result", lambda *_: { + "schema_version": "loopx_monitor_target_result_v0", "todo": _monitor(todo_id="todo_foreign"), + }) + with pytest.raises(TypeError, match="source"): + monitor_poll_writeback.resolve_monitor_todo_item(registry_path=tmp_path / "registry.json", + runtime_root=tmp_path, goal_id=GOAL_ID, todo_id="todo_transport_watch")