test(todos): compare canonical reads without the time-varying envelope

loopx-project#6874 added projection_envelope to the canonical todo read with wall-clock observed_at/served_at/last_read_at values. Three tests compared consecutive whole payloads, so they now fail on the timestamps alone: the required windows-powershell lane (test_sqlite_authority_cli.py) and two tests collected by the required test-shard job (test_canonical_todo_writer_isolation.py, test_research_context_adoption.py).

Use the existing assert_todo_read_unchanged helper from canonical_authority_fixture, which validates and ignores only those clock fields and compares every other envelope and body fact.

Fixes loopx-project#7152

Signed-off-by: JasonBuildAI <jasonbuildai@gmail.com>

Author: JasonBuildAI

tests/control_plane/test_canonical_todo_writer_isolation.py

@@ -12,7 +12,7 @@
 
 import pytest
 
-from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture
+from canonical_authority_fixture import assert_todo_read_unchanged, isolate_sqlite_runtime, promoted_create_fixture
 from test_quota_authority_settlement_journey import _source
 import test_quota_settlement_cli as settlement
 
@@ -185,19 +185,19 @@ def cli(*args, succeeds=True, override=False):
     before = cli("todo", "list")
     change = ("handoff-mode", "set", "--mode", "hard_lease", "--operation-id", "original-mode")
     assert cli(*change, "--dry-run")["status"] == "planned"
-    assert cli("todo", "list") == before
+    assert_todo_read_unchanged(cli("todo", "list"), before)
     assert cli(*change)["changed"] is True
     assert cli("handoff-mode", "show", override=True)["handoff_mode"] == "hard_lease"
     assert cli("handoff-mode", "set", "--mode", "soft_claim", "--operation-id", "later-mode")["changed"] is True
     later = cli("todo", "list")
     replay = cli(*change)
     assert replay["status"] == "replayed" and replay["changed"] is False
     assert replay["handoff_mode"] == "hard_lease"
-    assert cli("todo", "list") == later
+    assert_todo_read_unchanged(cli("todo", "list"), later)
     assert cli("handoff-mode", "show")["handoff_mode"] == "soft_claim"
     mismatch = cli("handoff-mode", "set", "--mode", "soft_claim", "--operation-id", "original-mode", succeeds=False)
     assert mismatch["error_code"] == "coordination_operation_identity_mismatch"
-    assert cli("todo", "list") == later
+    assert_todo_read_unchanged(cli("todo", "list"), later)
     assert display.read_bytes() == original_display
 
     backend = runtime / "authority" / f"{provider}-v0"
@@ -211,7 +211,7 @@ def cli(*args, succeeds=True, override=False):
             assert not backend.exists() and display.read_bytes() == original_display
     finally:
         offline.rename(backend)
-    assert cli("todo", "list") == later
+    assert_todo_read_unchanged(cli("todo", "list"), later)
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])

tests/control_plane/test_research_context_adoption.py

@@ -13,7 +13,7 @@
 from loopx.cli import main
 from loopx.control_plane.effect_runtime import effect_runtime_result
 from loopx.hosts.research_checkpoint import canonical_context, research_context, verify_context_sources, adopt_todo_context, _encoded
-from canonical_authority_fixture import isolate_sqlite_runtime
+from canonical_authority_fixture import assert_todo_read_unchanged, isolate_sqlite_runtime
 from loopx.control_plane.turn_driver.codex_sessions import _store_codex_cli_session
 
 
@@ -227,7 +227,7 @@ def lost_handoff(env, payload):
     assert 'source-revision:'+source_context['snapshot']['provider_revision'] in inspected['evidence_refs']
     assert 'Candidate 3 failed' in Path(result['artifact']).read_text()
     assert 'Candidate 999' not in Path(result['artifact']).read_text()
-    assert source_run('todo','list','--goal-id','source','--todo-id',source_todo)==source_before
+    assert_todo_read_unchanged(source_run('todo','list','--goal-id','source','--todo-id',source_todo),source_before)
     assert run('task-lease','inspect','--goal-id','branch','--todo-id',destination_todo)['lease']==lease['lease']
     if with_knowledge and lost_response is None:
         _ordinary_review_and_work(run, workspace, tmp_path, destination_todo, digest, monkeypatch)
@@ -238,7 +238,7 @@ def lost_handoff(env, payload):
     if with_knowledge:
         assert replay['explore_knowledge']['appended_event_count'] == 0
         assert log.read_bytes() == imported_bytes
-    assert run('todo','list','--goal-id','branch','--todo-id',destination_todo)==after
+    assert_todo_read_unchanged(run('todo','list','--goal-id','branch','--todo-id',destination_todo),after)
     with pytest.raises(ValueError,match='execution authority changed'):
         adopt_todo_context(env,unit,digest,**{**args,'lease_version':999})
     with pytest.raises(ValueError,match='already has context'):
@@ -258,7 +258,7 @@ def lost_handoff(env, payload):
     moved=run('todo','list','--goal-id','branch','--todo-id',destination_todo)
     with pytest.raises(ValueError,match='continuation rejected'):
         adopt_todo_context(env,unit,digest,**args)
-    assert run('todo','list','--goal-id','branch','--todo-id',destination_todo)==moved
+    assert_todo_read_unchanged(run('todo','list','--goal-id','branch','--todo-id',destination_todo),moved)
 
 
 def _ordinary_review_and_work(run, workspace, tmp_path, todo, digest, monkeypatch, work=None):

tests/control_plane/test_sqlite_authority_cli.py

@@ -12,7 +12,9 @@
 
 from loopx.control_plane.effect_runtime import effect_runtime_result
 
-from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+from tests.control_plane.canonical_authority_fixture import (
+    assert_todo_read_unchanged, initialize_canonical_authority, isolate_sqlite_runtime,
+)
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
 from loopx.control_plane.coordination.coordination_state_contract import TODO_DOMAIN_READ_RECORD_SCHEMA_VERSION, TODO_DOMAIN_RECORD_FIELDS
 from loopx.control_plane.coordination.local_authority_shadow_projection import canonical_bytes
@@ -238,7 +240,10 @@ def todos():
         refusal = command("authority-archive", "plan-migration", "--goal-id", goal,
                           "--provider", "file", "--plan", str(refused_plan), expected_code=1)
         assert "settled task leases" in refusal["reason"]
-        assert not refused_plan.exists() and todos() == before_refusal
+        assert not refused_plan.exists()
+        # The projection envelope records when each read happened; the refused
+        # exit must leave every other read fact unchanged.
+        assert_todo_read_unchanged(todos(), before_refusal)
         command("task-lease", "release", "--goal-id", goal, "--todo-id", todo,
                 "--owner", agent, "--idempotency-key", "trial-work",
                 "--expected-version", str(lease["lease"]["version"]))