test(coordination): restore Stage2C priority and recovery oracles

tests/control_plane/test_cold_source_disposition_e2e.py

@@ -238,7 +238,8 @@ def test_original_lease_disposition_requires_release_before_cold_import(
     fixture.cli("todo", "add", "--role", "agent", "--text", "Later lease-import write",
         "--operation-id", "later-lease-import", "--claimed-by", "agent-a")
     current = provider_todos(receiver, fixture)
-    assert {row["text"] for row in current["todos"]} == {"Original leased task", "Later lease-import write"}
+    assert {row["text"] for row in current["todos"]} == {"[P1] Original leased task", "[P1] Later lease-import write"}
+    assert {row["priority"] for row in current["todos"]} == {"P1"}
     fixture.state.unlink()
     replay = cold_cli(fixture, receiver_package, "recover-import", "--operation-id", "lease-import",
         "--plan-sha256", plan["plan_sha256"], "--execute")
@@ -305,15 +306,17 @@ def test_unproved_original_archival_then_cold_import_keeps_history_inert(
     assert applied["status"] == "applied" and applied["execution_authority_granted"] is False
     imported = provider_todos(receiver, fixture)
     assert imported["source_authority"] == f"{provider}_v0"
-    assert [(row["todo_id"], row["text"]) for row in imported["todos"]] == [(source_todo, "Later source requirement")]
+    assert [(row["todo_id"], row["text"], row["priority"]) for row in imported["todos"]] == [
+        (source_todo, "[P1] Later source requirement", "P1")]
     # The later current source, not the candidate or ambiguous queue, is the
     # import basis. Unproved entries remain archived and never become receipts.
     assert Path(rolled["candidate_archive_path"]).read_bytes() == candidate
     assert originals(Path(rolled["outbox_archive_path"])) == pending
     fixture.cli("todo", "add", "--role", "agent", "--text", "Later canonical write",
         "--operation-id", "after-ambiguous-import", "--claimed-by", "agent-a")
     current = provider_todos(receiver, fixture)
-    assert {row["text"] for row in current["todos"]} == {"Later source requirement", "Later canonical write"}
+    assert {row["text"] for row in current["todos"]} == {"[P1] Later source requirement", "[P1] Later canonical write"}
+    assert {row["priority"] for row in current["todos"]} == {"P1"}
     fixture.state.unlink()
     recovered = cold_cli(fixture, receiver_package, "recover-import", "--operation-id", "ambiguous-import",
         "--plan-sha256", plan["plan_sha256"], "--execute")
@@ -415,7 +418,8 @@ def test_original_outbox_disposition_then_cold_import_preserves_receipts_and_lat
     current = provider_todos(receiver, fixture)
     assert current["source_authority"] == f"{provider}_v0"
     assert {row["text"] for row in current["todos"]} == (
-        {"Interrupted original", "Later canonical write"} if committed else {"Later canonical write"})
+        {"[P1] Interrupted original", "[P1] Later canonical write"} if committed else {"[P1] Later canonical write"})
+    assert {row["priority"] for row in current["todos"]} == {"P1"}
     assert len(current["todos"]) == (2 if committed else 1)
     assert later["todo_id"] in {row["todo_id"] for row in current["todos"]}
     fixture.state.unlink()

tests/control_plane/test_runtime_shadow_bounded_e2e.py

@@ -450,7 +450,8 @@ def test_public_handoff_todo_and_atomic_monitor_batch_capture_each_primary_mutat
     assert len(transactions) == 5  # Baseline, handoff, todo add, monitor add, atomic Monitor batch.
     batch_todos = {todo["todo_id"]: todo for todo in transactions[-1]["projection"]["todos"]}
     assert batch_todos[monitor["todo_id"]]["result_hash"] == "release-v1"
-    assert batch_todos[result["successor_todo_ids"][0]]["text"] == "Validate the released head"
+    successor = batch_todos[result["successor_todo_ids"][0]]
+    assert (successor["text"], successor["priority"]) == ("[P1] Validate the released head", "P1")
     receipts = [transaction["receipts"][0] for transaction in transactions[1:]]
     assert len({receipt["entry_id"] for receipt in receipts}) == 4
     assert [receipt["seq"] for receipt in receipts] == [1, 2, 3, 4]
@@ -499,6 +500,9 @@ def test_native_lease_writer_cannot_reuse_a_sequence_when_its_cursor_is_missing(
     assert json.loads(pending[0].read_text())["seq"] == 2
     assert not cursor.exists()  # The writer recovers a sequence, never a cursor.
     cli(registry, runtime, "todo", "add", "--goal-id", "goal-a", "--role", "agent", "--text", "Trigger bounded drain")
+    # Inline capture may leave receipt-proven cleanup for explicit recovery.
+    drained = cli(registry, runtime, "authority-shadow", "drain", "--goal-id", "goal-a")
+    assert drained["pending_after"] == drained["prepared_only_after"] == 0, drained
     transactions = history(tmp_path, runtime)
     lease_receipts = [row["receipts"][0] for row in transactions[1:] if row["receipts"][0]["partition"] == "leases"]
     assert [receipt["seq"] for receipt in lease_receipts] == [1, 2]

tests/control_plane/test_shadow_fence_caller_parity_e2e.py

@@ -83,11 +83,12 @@ def observe(self, args: tuple[str, ...]) -> dict:
         return {
             "envelope": self.normalize(envelope),
             "exit": self.exit(),
-            "effect": {
+            # Keep the exact effect shape and target; Todo IDs are opaque.
+            "effect": self.normalize({
                 "added": sorted(k for k in after if k not in before),
                 "removed": sorted(k for k in before if k not in after),
                 "changed": sorted(k for k in after if k in before and after[k] != before[k]),
-            },
+            }),
             "outbox_added": sorted(outbox_after - outbox_before),
         }
 

tests/control_plane/test_shadow_observable_e2e.py

@@ -93,14 +93,16 @@ def test_todo_argument_intent_and_rejections(caller: Caller) -> None:
     preview = w.call('todo', 'add', '--role', 'agent', '--text', 'Retain operator intent', '--dry-run')
     assert preview['ok'] is True and w.primary() == before
     todo = w.add('Retain operator intent', '--claimed-by', 'agent-a', '--note', 'Initial context')
-    assert w.read(todo)['claimed_by'] == 'agent-a'
+    created = w.read(todo)
+    assert created['claimed_by'] == 'agent-a'
+    assert (created['text'], created['priority']) == ('[P1] Retain operator intent', 'P1')
     for args, expected_note in [(('--note', 'Updated context'), 'Updated context'),
                                 (('--text', 'Corrected operator intent'), 'Updated context')]:
         result = w.call('todo', 'update', '--todo-id', todo, '--agent-id', 'agent-a', *args)
         assert result['ok'] is True, result
         current = w.read(todo)
         assert current['note'] == expected_note and current['claimed_by'] == 'agent-a'
-    assert w.read(todo)['text'] == 'Corrected operator intent'
+    assert (current['text'], current['priority']) == ('[P1] Corrected operator intent', 'P1')
     # CLI mutable-field validation treats an empty note as absent, not a clear.
     before = w.primary()
     empty = w.call('todo', 'update', '--todo-id', todo, '--agent-id', 'agent-a', '--note', '')
@@ -308,7 +310,8 @@ def test_monitor_successor_retains_caller_routing(caller: Caller) -> None:
         '--next-claimed-by', 'agent-a', '--execute')
     assert result['ok'] is True and len(result['successor_todo_ids']) == 1, result
     successor = w.read(result['successor_todo_ids'][0])
-    assert successor['text'] == 'Validate released head' and successor['claimed_by'] == 'agent-a'
+    assert successor['text'] == '[P1] Validate released head' and successor['claimed_by'] == 'agent-a'
+    assert successor['priority'] == 'P1'
     assert successor['action_kind'] == 'validate_release_head' and successor['required_capabilities'] == ['network']
 
 
@@ -321,7 +324,8 @@ def test_retired_observation_request_preserves_runtime_capture(caller: Caller) -
         assert w.primary() == before
     assert not (w.root / 'authority-shadow' / 'file' / 'observable').exists()
     todo = w.add('Independent observation contract')
-    assert w.read(todo)['text'] == 'Independent observation contract'
+    created = w.read(todo)
+    assert (created['text'], created['priority']) == ('[P1] Independent observation contract', 'P1')
 
 
 def test_turn_input_rejection_has_no_host_or_primary_effect(caller: Caller) -> None:
@@ -344,7 +348,8 @@ def test_registry_relative_root_does_not_depend_on_callers_cwd(caller: Caller) -
         'todo', 'add', '--goal-id', 'observable', '--role', 'agent', '--text', 'Registry relative root']
     added = w.invoke(command, command[3:], cwd=cwd)
     assert added['ok'] is True, added
-    assert w.read(added['todo_id'])['text'] == 'Registry relative root'
+    created = w.read(added['todo_id'])
+    assert (created['text'], created['priority']) == ('[P1] Registry relative root', 'P1')
     assert (w.root / 'goals' / 'observable' / 'rollout-event-log.jsonl').exists()
     assert not (cwd / 'runtime').exists()
     assert w.call('handoff-mode', 'set', '--mode', 'hard_lease')['ok'] is True

tests/fixtures/control_plane/legacy_writer_fence_caller_parity_v0.json

@@ -1068,7 +1068,7 @@
       },
       "effect": {
         "added": [
-          "runtime/goals/observable/task-leases/todo_f49c7f1b755a.json"
+          "runtime/goals/observable/task-leases/{todo_b}.json"
         ],
         "removed": [],
         "changed": []