v7next F1 pilot: domain D15 verified against upstream; its test split transplanted

All 15 D15 runtime modules are unsplit by design - upstream bytes stand as-is:
7 byte-identical to the reference, 6 pure upstream drift (nothing to do), and
2 (consciousness, reflection) carry v7 deltas of the D02 family that must NOT
be replayed verbatim - reflection's status-set delta would invert over
upstream's own handling (the re-prove trap is documented in
docs/v7next/LEDGER_CORRECTIONS.md together with a MIGRATION row already
superseded by upstream).
The test side transplants the oracle's D15 split: the 2386-line evolution
integrity giant becomes six themed suites (lossless: 65==65 test functions,
zero upstream drift of the giant since merge-base) plus three siblings carried
verbatim; adaptations are exclusively reverse-mappings of OTHER domains' v7
spellings back to upstream signatures, each keyed to the original monolith.
98 passed in isolation (re-verified independently); every file <=617 lines;
HEAD held through six pytest runs.

docs/v7next/LEDGER_CORRECTIONS.md

@@ -0,0 +1,22 @@
+# Ledger corrections discovered during v7next transplants (append-only)
+
+Rows of the reference MIGRATION_v7.md / DOMAIN_MAP.md falsified by upstream drift,
+with evidence, found lane by lane. Applied to the campaign's carried ledger at F5.
+
+## From the D15 pilot (base b9f7597f, 2026-08-30)
+1. MIGRATION row 351 (`tools/core.py::_filter_out_project_store` ->
+   `project_facts.py::filter_out_project_store`, status "pending upstream
+   transfer") — SUPERSEDED-BY-UPSTREAM: the tip already carries the extraction
+   (project_facts.py byte-identical to the reference; core.py keeps only the
+   import alias at :17 with two call sites).
+2. DOMAIN_MAP §D15 "v7 delta" prose — remeasure from the new base: consolidator
+   delta absorbed upstream (now 0); the true residue is +23/-12 in two files
+   (consciousness.py, reflection.py), not +25/-14 in three.
+3. RE-PROVE TRAP (D02 family, reflection.py): the reference's
+   `_trace_call_errored` reads `_OK_TOOL_STATUSES` (with "untyped") from the v7
+   leaf `_outcome_tool_errors`, which upstream does not have; a verbatim replay
+   of the delta over upstream's own status handling would invert the fix. The
+   D02 adoption must re-derive the delta against upstream bytes.
+4. MIGRATION row 166 (retirement of 4 CLAUDE_CODE markers, id "none") — needs an
+   explicit ADOPTION disposition (umbrella under D02 or its own row): zero
+   production emitters of those markers exist at this tip (claim re-proven).

tests/_evolution_state_shared.py

@@ -0,0 +1,59 @@
+"""Builders and stubs shared by the evolution state-integrity suites.
+
+Split out of ``tests/test_evolution_state_integrity_v3.py`` when that module was divided by
+theme; every definition is verbatim, so each sibling suite keeps the exact campaign and
+transaction shape it was written against.
+"""
+
+from __future__ import annotations
+
+import pathlib
+
+
+def _patch_commit_seam(monkeypatch, name, value):
+    """Stub one reviewed-commit seam on every module that resolves it.
+
+    ``_repo_commit_push`` spans ``tools/git.py`` and, on trees where the v7
+    split landed, its extracted owners (``git_review_cycle``, ``git_evolution``),
+    so a seam stub has to reach whichever module resolves the name at call time.
+    On the upstream layout only ``tools/git.py`` exists; the extracted owners
+    are patched when importable.
+    """
+    from ouroboros.tools import git as git_tools
+
+    modules = [git_tools]
+    try:
+        from ouroboros.tools import git_evolution, git_review_cycle
+        modules.extend((git_review_cycle, git_evolution))
+    except ImportError:
+        pass
+
+    for module in modules:
+        if hasattr(module, name):
+            monkeypatch.setattr(module, name, value)
+
+
+def _active_transaction(tmp_path: pathlib.Path, task_id: str = "evo-task"):
+    from supervisor import evolution_lifecycle, queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    queue.init_queue_refs([], {}, {"value": 0})
+    campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+    live = state.load_state()
+    live.update({
+        "owner_chat_id": 1,
+        "evolution_mode_enabled": True,
+        "evolution_owner_stopped": False,
+    })
+    state.save_state(live)
+    tx = evolution_lifecycle.begin_evolution_transaction(task_id, cycle=1, campaign=campaign)
+    return campaign, tx
+
+
+class _CaptureQueue:
+    def __init__(self):
+        self.items = []
+
+    def put(self, item):
+        self.items.append(item)

tests/test_context_memory.py

@@ -0,0 +1,461 @@
+"""Recent chat, consolidation offsets and the memory sections around them.
+
+Split verbatim out of ``tests/test_context.py`` by theme (merged there from the former
+``test_context_memory_overhaul.py``). This module owns the offset a consolidation
+leaves, the full-awareness main thread against a project thread's own view, the
+workpad/journal that may not be silently sliced, the low-mode dialogue tail, the stale
+offset a rotation invalidates, the world profile, the retired dialogue summaries, the
+process logs filtered by task id, and the installed-skills verdict.
+"""
+
+from __future__ import annotations
+
+import json
+
+
+
+
+
+# ===========================================================================
+# Memory / consolidation offset behavior (merged from former
+# test_context_memory_overhaul.py).  Inspect-only `limit=50` / `limit=1000`
+# source-string pins were dropped — behavioral coverage below already
+# exercises the offset path.  test_no_identity_truncation_in_consolidator_
+# prompts was also dropped (inspect-only); identity-truncation is covered
+# behaviorally by consolidator tests.
+# ===========================================================================
+
+
+def test_recent_chat_starts_after_consolidated_offset(tmp_path):
+    from ouroboros.context import build_recent_sections
+    from ouroboros.memory import Memory
+
+    logs_dir = tmp_path / "logs"
+    memory_dir = tmp_path / "memory"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    memory_dir.mkdir(parents=True, exist_ok=True)
+    entries = [
+        {"ts": f"2026-03-19T16:{i:02d}:00Z", "direction": "in", "username": "User", "text": f"msg-{i}"}
+        for i in range(5)
+    ]
+    (logs_dir / "chat.jsonl").write_text(
+        "\n".join(json.dumps(entry) for entry in entries) + "\n",
+        encoding="utf-8",
+    )
+    memory = Memory(drive_root=tmp_path)
+    (memory_dir / "dialogue_meta.json").write_text(
+        json.dumps({
+            "last_consolidated_offset": 3,
+            "chat_log_signature": memory.jsonl_generation_signature("chat.jsonl"),
+        }),
+        encoding="utf-8",
+    )
+
+    sections = build_recent_sections(memory, env=None)
+    combined = "\n\n".join(sections)
+
+    assert "msg-0" not in combined
+    assert "msg-1" not in combined
+    assert "msg-2" not in combined
+    assert "msg-3" in combined
+    assert "msg-4" in combined
+
+
+def test_recent_chat_main_includes_all_threads_full_awareness(tmp_path):
+    """Full project awareness (v6.32.0): the one identity's main/global context
+    sees its WHOLE conversation — main + project threads alike (BIBLE P1, one
+    awareness across direct chat, project rooms, and consciousness). Project chat
+    is part of the one mind's memory, NOT partitioned out; only A2A virtual
+    transport is excluded (covered elsewhere)."""
+    from ouroboros.context import build_recent_sections
+    from ouroboros.memory import Memory
+    from ouroboros.projects_registry import create_project
+
+    logs_dir = tmp_path / "logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+
+    project = create_project(tmp_path, "racer")
+    project_chat = int(project["chat_id"])
+    transport_chat = 555000111  # large NON-project id (e.g. a Telegram mirror)
+
+    entries = [
+        {"chat_id": 1, "direction": "in", "username": "User", "text": "main-keep"},
+        {"chat_id": project_chat, "direction": "in", "username": "User", "text": "project-visible"},
+        {"chat_id": transport_chat, "direction": "in", "username": "User", "text": "transport-keep"},
+        {"direction": "in", "username": "User", "text": "legacy-keep"},  # no chat_id -> main
+    ]
+    (logs_dir / "chat.jsonl").write_text(
+        "\n".join(json.dumps(entry) for entry in entries) + "\n",
+        encoding="utf-8",
+    )
+
+    combined = "\n\n".join(build_recent_sections(Memory(drive_root=tmp_path), env=None))
+
+    assert "main-keep" in combined
+    assert "legacy-keep" in combined
+    assert "transport-keep" in combined
+    assert "project-visible" in combined  # full awareness: the one mind sees project chat
+
+
+def test_recent_chat_for_project_thread_shows_only_its_own_thread(tmp_path):
+    """A project TASK gets a FOCUSED working view of its own thread (full
+    awareness, v6.32.0): its "## Recent chat" is its own project thread, not the
+    штаб's main chat nor a sibling project's chat, so cross-project noise does not
+    bloat its working context. This is focus, not memory isolation — the one mind
+    still sees everything via the main/background path. Pins that thread_chat_id
+    selects the project's own raw tail rather than the main consolidation stream."""
+    from ouroboros.context import build_recent_sections
+    from ouroboros.memory import Memory
+    from ouroboros.projects_registry import create_project
+
+    logs_dir = tmp_path / "logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+
+    proj_a = create_project(tmp_path, "racer")
+    proj_b = create_project(tmp_path, "research")
+    chat_a = int(proj_a["chat_id"])
+    chat_b = int(proj_b["chat_id"])
+
+    entries = [
+        {"chat_id": 1, "direction": "in", "username": "User", "text": "main-stab-chat"},
+        {"chat_id": chat_a, "direction": "in", "username": "User", "text": "project-a-own-thread"},
+        {"chat_id": chat_b, "direction": "in", "username": "User", "text": "project-b-sibling"},
+    ]
+    (logs_dir / "chat.jsonl").write_text(
+        "\n".join(json.dumps(entry) for entry in entries) + "\n",
+        encoding="utf-8",
+    )
+
+    combined = "\n\n".join(build_recent_sections(
+        Memory(drive_root=tmp_path), env=None, thread_chat_id=chat_a))
+
+    assert "project-a-own-thread" in combined   # its own thread is visible
+    assert "project-b-sibling" not in combined  # sibling project not in focused view
+    assert "main-stab-chat" not in combined     # main chat not in focused project view
+
+
+def test_project_workpad_and_journal_not_silently_sliced(tmp_path, monkeypatch):
+    """BIBLE P1 (no silent truncation): project cognitive artifacts are not
+    prefix-sliced into context. The workpad rides in FULL; journal milestones show
+    full text (no per-row [:N]) with a visible journal_read pointer for older."""
+    import types
+
+    monkeypatch.setattr("ouroboros.config.DATA_DIR", tmp_path)
+    from ouroboros.context import build_knowledge_sections
+    from ouroboros.project_facts import project_journal_path, project_workpad_path
+    from ouroboros.utils import append_jsonl
+
+    pid = "builder"
+    wp = project_workpad_path(pid)
+    wp.parent.mkdir(parents=True, exist_ok=True)
+    tail = "WORKPAD_TAIL_MARKER"
+    wp.write_text("A" * 20_000 + tail, encoding="utf-8")  # > old 12_000 slice
+    append_jsonl(project_journal_path(pid), {
+        "ts": "2026-06-14T00:00:00Z", "kind": "checkpoint", "text": "M" * 600,  # > old 200 slice
+    })
+
+    env = types.SimpleNamespace(drive_path=lambda rel: tmp_path / rel)
+    combined = "\n\n".join(build_knowledge_sections(env, project_id=pid))
+
+    assert tail in combined          # full workpad, not prefix-sliced to 12_000
+    assert ("M" * 600) in combined   # full journal milestone, not sliced to 200
+
+
+def test_append_journal_milestone_bounds_over_limit_with_pointer(tmp_path, monkeypatch):
+    """An AUTOMATIC completion milestone honors the journal's durable per-row cap:
+    over-limit text is bounded with a VISIBLE pointer (recorded, never silently
+    sliced nor dropped) — same _MAX_TEXT_CHARS contract as the journal_write tool,
+    so emit_task_results cannot append a raw unbounded row."""
+    monkeypatch.setattr("ouroboros.config.DATA_DIR", tmp_path)
+    from ouroboros.project_facts import project_journal_path
+    from ouroboros.tools.project_journal import _MAX_TEXT_CHARS, append_journal_milestone
+    from ouroboros.utils import iter_jsonl_objects
+
+    pid = "lh"
+    append_journal_milestone(pid, "done", "Z" * (_MAX_TEXT_CHARS + 500), task_id="t1")
+    rows = [r for r in iter_jsonl_objects(project_journal_path(pid)) if isinstance(r, dict)]
+    assert len(rows) == 1                      # recorded (not dropped/rejected)
+    txt = rows[0]["text"]
+    assert len(txt) <= _MAX_TEXT_CHARS         # honors the durable per-row contract
+    assert "task_results" in txt               # VISIBLE pointer to the full text
+
+
+def test_low_mode_preserves_full_unconsolidated_dialogue_suffix(tmp_path, monkeypatch):
+    from ouroboros.context import build_recent_sections
+    from ouroboros.memory import Memory
+
+    logs_dir = tmp_path / "logs"
+    memory_dir = tmp_path / "memory"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    memory_dir.mkdir(parents=True, exist_ok=True)
+    fresh_count = 305
+    entries = [
+        {"chat_id": 1, "direction": "in", "username": "User", "text": f"consolidated-{i}"}
+        for i in range(3)
+    ] + [
+        {"chat_id": 1, "direction": "in", "username": "User", "text": f"fresh-{i}"}
+        for i in range(fresh_count)
+    ]
+    (logs_dir / "chat.jsonl").write_text(
+        "\n".join(json.dumps(entry) for entry in entries) + "\n",
+        encoding="utf-8",
+    )
+    memory = Memory(drive_root=tmp_path)
+    (memory_dir / "dialogue_meta.json").write_text(
+        json.dumps({
+            "last_consolidated_offset": 3,
+            "chat_log_signature": memory.jsonl_generation_signature("chat.jsonl"),
+        }),
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("OUROBOROS_CONTEXT_MODE", "low")
+
+    combined = "\n\n".join(build_recent_sections(memory, env=None))
+
+    assert "consolidated-0" not in combined
+    assert "fresh-0" in combined
+    assert f"fresh-{fresh_count - 1}" in combined
+
+
+def test_low_mode_without_consolidation_keeps_max_raw_dialogue_tail(tmp_path, monkeypatch):
+    from ouroboros.context import build_recent_sections
+    from ouroboros.context_budget import MAX_RECENT_CHAT_TAIL
+    from ouroboros.memory import Memory
+
+    logs_dir = tmp_path / "logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    fresh_count = 305
+    entries = [
+        {"chat_id": 1, "direction": "in", "username": "User", "text": f"fresh-{i}"}
+        for i in range(fresh_count)
+    ]
+    (logs_dir / "chat.jsonl").write_text(
+        "\n".join(json.dumps(entry) for entry in entries) + "\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("OUROBOROS_CONTEXT_MODE", "low")
+
+    combined = "\n\n".join(build_recent_sections(Memory(drive_root=tmp_path), env=None))
+
+    assert fresh_count < MAX_RECENT_CHAT_TAIL
+    assert "fresh-0" in combined
+    assert f"fresh-{fresh_count - 1}" in combined
+
+
+def test_recent_chat_offset_uses_filtered_dialogue_entries(tmp_path):
+    from ouroboros.context import build_recent_sections
+    from ouroboros.memory import Memory
+
+    logs_dir = tmp_path / "logs"
+    memory_dir = tmp_path / "memory"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    memory_dir.mkdir(parents=True, exist_ok=True)
+    entries = [
+        {"chat_id": 1, "direction": "in", "username": "User", "text": "consolidated-0"},
+        {"chat_id": -1, "direction": "in", "username": "Agent", "text": "a2a-noise"},
+        {"chat_id": 1, "direction": "in", "username": "User", "text": "consolidated-1"},
+        {"chat_id": 1, "direction": "in", "username": "User", "text": "fresh"},
+    ]
+    (logs_dir / "chat.jsonl").write_text(
+        "\n".join(json.dumps(entry) for entry in entries) + "\n",
+        encoding="utf-8",
+    )
+    memory = Memory(drive_root=tmp_path)
+    (memory_dir / "dialogue_meta.json").write_text(
+        json.dumps({
+            "last_consolidated_offset": 2,
+            "chat_log_signature": memory.jsonl_generation_signature("chat.jsonl"),
+        }),
+        encoding="utf-8",
+    )
+
+    combined = "\n\n".join(build_recent_sections(memory, env=None))
+
+    assert "consolidated-0" not in combined
+    assert "consolidated-1" not in combined
+    assert "a2a-noise" not in combined
+    assert "fresh" in combined
+
+
+def test_recent_chat_ignores_stale_consolidation_offset_after_rotation(tmp_path):
+    from ouroboros.context import build_recent_sections
+    from ouroboros.memory import Memory
+
+    logs_dir = tmp_path / "logs"
+    memory_dir = tmp_path / "memory"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    memory_dir.mkdir(parents=True, exist_ok=True)
+    initial = [
+        {"chat_id": 1, "direction": "in", "username": "User", "text": f"early-{i}"}
+        for i in range(3)
+    ]
+    (logs_dir / "chat.jsonl").write_text(
+        "\n".join(json.dumps(entry) for entry in initial) + "\n",
+        encoding="utf-8",
+    )
+    memory = Memory(drive_root=tmp_path)
+    stale_signature = memory.jsonl_generation_signature("chat.jsonl")
+    (memory_dir / "dialogue_meta.json").write_text(
+        json.dumps({
+            "last_consolidated_offset": 3,
+            "chat_log_signature": stale_signature,
+        }),
+        encoding="utf-8",
+    )
+
+    rotated = [
+        {"chat_id": 1, "direction": "in", "username": "User", "text": f"post-rotate-{i}"}
+        for i in range(2)
+    ]
+    (logs_dir / "chat.jsonl").write_text(
+        "\n".join(json.dumps(entry) for entry in rotated) + "\n",
+        encoding="utf-8",
+    )
+
+    combined = "\n\n".join(build_recent_sections(memory, env=None))
+
+    # Rotation invalidates the stale offset; rotated entries appear.
+    assert "post-rotate-0" in combined
+    assert "post-rotate-1" in combined
+
+
+def test_recent_chat_keeps_offset_when_same_log_gets_appended(tmp_path):
+    from ouroboros.context import build_recent_sections
+    from ouroboros.memory import Memory
+
+    logs_dir = tmp_path / "logs"
+    memory_dir = tmp_path / "memory"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    memory_dir.mkdir(parents=True, exist_ok=True)
+    initial = [
+        {"chat_id": 1, "direction": "in", "username": "User", "text": f"old-{i}"}
+        for i in range(3)
+    ]
+    (logs_dir / "chat.jsonl").write_text(
+        "\n".join(json.dumps(entry) for entry in initial) + "\n",
+        encoding="utf-8",
+    )
+    memory = Memory(drive_root=tmp_path)
+    (memory_dir / "dialogue_meta.json").write_text(
+        json.dumps({
+            "last_consolidated_offset": 3,
+            "chat_log_signature": memory.jsonl_generation_signature("chat.jsonl"),
+        }),
+        encoding="utf-8",
+    )
+
+    with open(logs_dir / "chat.jsonl", "a", encoding="utf-8") as handle:
+        handle.write(json.dumps({"chat_id": 1, "direction": "in", "username": "User", "text": "new"}) + "\n")
+
+    combined = "\n\n".join(build_recent_sections(memory, env=None))
+
+    assert "old-0" not in combined
+    assert "new" in combined
+
+
+def test_world_profile_is_loaded_with_stable_memory(tmp_path):
+    from ouroboros.context import build_memory_sections
+    from ouroboros.memory import Memory
+
+    (tmp_path / "memory").mkdir(parents=True, exist_ok=True)
+    (tmp_path / "memory" / "WORLD.md").write_text("world-profile-data", encoding="utf-8")
+    memory = Memory(drive_root=tmp_path)
+
+    sections = build_memory_sections(memory)
+    combined = "\n\n".join(sections)
+
+    assert "world-profile-data" in combined
+
+
+def test_retired_dialogue_summary_remains_visible_when_blocks_exist(tmp_path):
+    from ouroboros.context import build_memory_sections
+    from ouroboros.memory import Memory
+
+    memory_dir = tmp_path / "memory"
+    memory_dir.mkdir(parents=True, exist_ok=True)
+    (memory_dir / "dialogue_summary.md").write_text("legacy dialogue", encoding="utf-8")
+    (memory_dir / "dialogue_blocks.json").write_text(
+        json.dumps([{"content": "new dialogue block"}]),
+        encoding="utf-8",
+    )
+    memory = Memory(drive_root=tmp_path)
+
+    combined = "\n\n".join(build_memory_sections(memory, partition="volatile"))
+
+    assert "## Dialogue History" in combined
+    assert "new dialogue block" in combined
+    assert "## Legacy Dialogue Summary (retired flat format, read-only fallback)" in combined
+    assert "legacy dialogue" in combined
+
+
+def test_retired_dialogue_summary_fallback_preserves_continuity_without_blocks(tmp_path):
+    from ouroboros.context import build_memory_sections
+    from ouroboros.memory import Memory
+
+    memory_dir = tmp_path / "memory"
+    memory_dir.mkdir(parents=True, exist_ok=True)
+    (memory_dir / "dialogue_summary.md").write_text("legacy dialogue only", encoding="utf-8")
+    memory = Memory(drive_root=tmp_path)
+
+    combined = "\n\n".join(build_memory_sections(memory, partition="volatile"))
+
+    assert "## Legacy Dialogue Summary (retired flat format, read-only fallback)" in combined
+    assert "legacy dialogue only" in combined
+
+
+def test_recent_sections_filter_process_logs_by_task_id(tmp_path):
+    from ouroboros.context import build_recent_sections
+    from ouroboros.memory import Memory
+
+    logs_dir = tmp_path / "logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    (logs_dir / "progress.jsonl").write_text(
+        "\n".join([
+            json.dumps({"task_id": "task-a", "text": "in-scope"}),
+            json.dumps({"task_id": "task-b", "text": "out-of-scope"}),
+        ]) + "\n",
+        encoding="utf-8",
+    )
+    (logs_dir / "tools.jsonl").write_text(
+        "\n".join([
+            json.dumps({"task_id": "task-a", "tool": "shell"}),
+            json.dumps({"task_id": "task-b", "tool": "shell"}),
+        ]) + "\n",
+        encoding="utf-8",
+    )
+
+    memory = Memory(drive_root=tmp_path)
+    sections = build_recent_sections(memory, env=None, task_id="task-a")
+    combined = "\n\n".join(sections)
+    assert "in-scope" in combined
+    assert "out-of-scope" not in combined
+
+
+def test_installed_skills_section_includes_warnings_verdict(tmp_path, monkeypatch):
+    from ouroboros.context import _build_installed_skills_section
+
+    class FakeEnv:
+        drive_root = tmp_path
+
+    monkeypatch.setattr(
+        "ouroboros.skill_loader.summarize_skills",
+        lambda _root: {
+            "skills": [
+                {
+                    "name": "weather",
+                    "type": "script",
+                    "enabled": True,
+                    "review_status": "warnings",
+                    "executable_review": True,
+                    "review_stale": False,
+                    "description": "Weather helper",
+                }
+            ]
+        },
+    )
+
+    section = _build_installed_skills_section(FakeEnv())
+
+    assert "## Installed Skills" in section
+    assert "weather" in section
+    assert "warnings" in section

tests/test_evolution_commit_receipt.py

@@ -0,0 +1,336 @@
+"""The exact commit receipt: what binds it, what may read it, and what may never erase it.
+
+Split out of ``tests/test_evolution_state_integrity_v3.py`` by theme: the receipt bound to
+campaign, transaction and task; the second commit blocked before review; the receipt race
+ahead of the git commit; revoked authority; the rescue link and campaign sidecar that share
+the CAS; and the stale, terminal and panicking writers that must not overwrite it.
+"""
+
+from __future__ import annotations
+
+import pathlib
+import threading
+from types import SimpleNamespace
+
+import pytest
+
+from tests._evolution_state_shared import (
+    _active_transaction,
+    _patch_commit_seam,
+)
+
+
+def test_exact_commit_receipt_is_bound_to_campaign_transaction_and_task(tmp_path):
+    from supervisor import evolution_lifecycle
+
+    campaign, tx = _active_transaction(tmp_path)
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+    }
+    assert evolution_lifecycle.check_evolution_authority(**claim)["ok"] is True
+
+    receipt = evolution_lifecycle.record_evolution_commit(**claim, commit_sha="a" * 40)
+
+    assert receipt["ok"] is True
+    assert receipt["commit_sha"] == "a" * 40
+    stored = evolution_lifecycle._read_evolution_campaign()["active_transaction"]
+    assert stored["commit_receipt"] == receipt
+    assert evolution_lifecycle.check_evolution_authority(
+        **claim, commit_sha="b" * 40,
+    )["reason"] == "commit_receipt_mismatch"
+
+    campaign_state = evolution_lifecycle._read_evolution_campaign()
+    campaign_state["active_transaction"].pop("commit_receipt")
+    assert evolution_lifecycle._write_evolution_campaign(campaign_state) is True
+    assert evolution_lifecycle.check_evolution_authority(
+        **claim, commit_sha="a" * 40,
+    )["reason"] == "commit_receipt_missing"
+
+
+def test_second_evolution_commit_is_blocked_before_review(tmp_path, monkeypatch):
+    from ouroboros.tools import git as git_tools
+    from supervisor import evolution_lifecycle
+
+    campaign, tx = _active_transaction(tmp_path)
+    assert evolution_lifecycle.record_evolution_commit(
+        campaign["id"], tx["transaction_id"], tx["task_id"], "a" * 40,
+    )["ok"] is True
+    review_calls = []
+    monkeypatch.setattr(git_tools, "_task_attributed_commit_paths", lambda *a, **k: (None, None, "", None))
+    monkeypatch.setattr(git_tools, "_check_overlapping_review_attempt", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_record_commit_attempt", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_acquire_git_lock", lambda *a, **k: pathlib.Path("lock"))
+    monkeypatch.setattr(git_tools, "_release_git_lock", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_prepare_review_commit_worktree", lambda *a, **k: (False, ""))
+    _patch_commit_seam(monkeypatch, "_run_reviewed_stage_cycle",
+        lambda *a, **k: review_calls.append(True) or {"status": "passed"},
+    )
+    ctx = SimpleNamespace(
+        repo_dir=tmp_path,
+        drive_root=tmp_path,
+        current_task_type="evolution",
+        task_id=tx["task_id"],
+        task_metadata={"evolution_transaction": tx},
+    )
+
+    result = git_tools._repo_commit_push(ctx, "second commit")
+
+    assert "transaction_already_committed" in result
+    assert "No reviewer was called" in result
+    assert review_calls == []
+
+
+def test_receipt_race_blocks_evolution_before_git_commit(tmp_path, monkeypatch):
+    from ouroboros.tools import git as git_tools
+    from supervisor import evolution_lifecycle
+
+    campaign, tx = _active_transaction(tmp_path)
+    ctx = SimpleNamespace(
+        repo_dir=tmp_path,
+        current_task_type="evolution",
+        task_id=tx["task_id"],
+        task_metadata={"evolution_transaction": tx},
+    )
+    _patch_commit_seam(monkeypatch, "_record_commit_attempt", lambda *a, **k: None)
+    claim, error = git_tools._check_evolution_commit_stage(
+        ctx, "commit", 0.0, phase="pre_review_authority",
+    )
+    assert error == ""
+    assert evolution_lifecycle.record_evolution_commit(
+        **claim, commit_sha="b" * 40,
+    )["ok"] is True
+
+    _claim, error = git_tools._check_evolution_commit_stage(
+        ctx, "commit", 0.0, phase="pre_commit_authority",
+    )
+
+    assert "transaction_already_committed" in error
+    assert "Nothing was committed" in error
+
+
+def test_revoked_authority_leaves_commit_unrecorded(tmp_path):
+    from supervisor import evolution_lifecycle, state
+
+    campaign, tx = _active_transaction(tmp_path)
+    live = state.load_state()
+    live["evolution_mode_enabled"] = False
+    live["evolution_owner_stopped"] = True
+    state.save_state(live)
+
+    receipt = evolution_lifecycle.record_evolution_commit(
+        campaign["id"], tx["transaction_id"], tx["task_id"], "c" * 40,
+    )
+
+    assert receipt == {"ok": False, "reason": "owner_stopped", "commit_sha": "c" * 40}
+    assert evolution_lifecycle._read_evolution_campaign()["active_transaction"]["commit_sha"] == ""
+
+
+def test_exact_receipt_remains_authority_after_post_task_autostop(tmp_path):
+    from supervisor import evolution_lifecycle, state
+
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "9" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim, commit_sha=sha)["ok"] is True
+    state.update_state(lambda live: live.update(
+        evolution_mode_enabled=False,
+        post_task_autostop=False,
+    ))
+
+    assert evolution_lifecycle.check_evolution_authority(
+        **claim, commit_sha=sha,
+    )["ok"] is True
+    assert evolution_lifecycle.check_evolution_authority(**claim)["reason"] == "evolution_disabled"
+
+
+@pytest.mark.parametrize("held_lock", ["state", "campaign"])
+def test_rescue_link_uses_shared_campaign_cas_and_preserves_commit_receipt(
+    tmp_path, monkeypatch, held_lock,
+):
+    from ouroboros.platform_layer import (
+        acquire_exclusive_file_lock,
+        release_exclusive_file_lock,
+    )
+    from ouroboros.utils import atomic_write_json
+    from supervisor import evolution_lifecycle, git_ops, state
+
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "3" * 40
+    assert evolution_lifecycle.record_evolution_commit(
+        campaign["id"], tx["transaction_id"], tx["task_id"], sha,
+    )["ok"] is True
+    monkeypatch.setattr(evolution_lifecycle, "EVOLUTION_CAMPAIGN_CAS_TIMEOUT_SEC", 1.0)
+    monkeypatch.setattr(git_ops, "DRIVE_ROOT", tmp_path)
+    campaign_path = tmp_path / "state" / "evolution_campaign.json"
+    if held_lock == "state":
+        lock_path = tmp_path / "locks" / "state.lock"
+        lock_fd = state.acquire_file_lock(lock_path, timeout_sec=1.0)
+        release = state.release_file_lock
+    else:
+        lock_path = campaign_path.with_name(campaign_path.name + ".lock")
+        lock_fd = acquire_exclusive_file_lock(lock_path, timeout_sec=1.0)
+        release = release_exclusive_file_lock
+    assert lock_fd is not None
+    done = threading.Event()
+
+    def _link() -> None:
+        git_ops._link_rescue_to_evolution_transaction(
+            {"rescue_ref": "rescue/test", "path": "/tmp/rescue-test"},
+            "test",
+        )
+        done.set()
+
+    thread = threading.Thread(target=_link, daemon=True)
+    thread.start()
+    try:
+        assert done.wait(0.1) is False
+        current = evolution_lifecycle._read_evolution_campaign()
+        current["active_transaction"]["interleaved"] = held_lock
+        atomic_write_json(campaign_path, current, trailing_newline=True)
+    finally:
+        release(lock_path, lock_fd)
+    assert done.wait(2.0) is True
+    thread.join(timeout=1.0)
+
+    stored = evolution_lifecycle._read_evolution_campaign()["active_transaction"]
+    assert stored["commit_sha"] == sha
+    assert stored["commit_receipt"]["commit_sha"] == sha
+    assert stored["rescue_ref"] == "rescue/test"
+    assert stored["interleaved"] == held_lock
+
+
+def test_commit_receipt_uses_campaign_sidecar_before_rescue(tmp_path, monkeypatch):
+    from ouroboros.platform_layer import (
+        acquire_exclusive_file_lock,
+        release_exclusive_file_lock,
+    )
+    from supervisor import evolution_lifecycle
+
+    campaign, tx = _active_transaction(tmp_path)
+    monkeypatch.setattr(evolution_lifecycle, "EVOLUTION_CAMPAIGN_CAS_TIMEOUT_SEC", 1.0)
+    campaign_path = tmp_path / "state" / "evolution_campaign.json"
+    lock_path = campaign_path.with_name(campaign_path.name + ".lock")
+    lock_fd = acquire_exclusive_file_lock(lock_path, timeout_sec=1.0)
+    assert lock_fd is not None
+    done = threading.Event()
+    result = {}
+
+    def _record() -> None:
+        result.update(evolution_lifecycle.record_evolution_commit(
+            campaign["id"], tx["transaction_id"], tx["task_id"], "4" * 40,
+        ))
+        done.set()
+
+    thread = threading.Thread(target=_record, daemon=True)
+    thread.start()
+    try:
+        assert done.wait(0.1) is False
+    finally:
+        release_exclusive_file_lock(lock_path, lock_fd)
+    assert done.wait(2.0) is True
+    thread.join(timeout=1.0)
+    assert result["ok"] is True
+    assert evolution_lifecycle._read_evolution_campaign()["active_transaction"][
+        "commit_receipt"
+    ]["commit_sha"] == "4" * 40
+
+
+def test_campaign_sidecar_contention_releases_state_lock_quickly(tmp_path):
+    from ouroboros.platform_layer import (
+        acquire_exclusive_file_lock,
+        release_exclusive_file_lock,
+    )
+    from supervisor import evolution_lifecycle, state
+
+    campaign, tx = _active_transaction(tmp_path)
+    campaign_path = tmp_path / "state" / "evolution_campaign.json"
+    sidecar = campaign_path.with_name(campaign_path.name + ".lock")
+    sidecar_fd = acquire_exclusive_file_lock(sidecar, timeout_sec=1.0)
+    assert sidecar_fd is not None
+    try:
+        result = evolution_lifecycle.record_evolution_commit(
+            campaign["id"], tx["transaction_id"], tx["task_id"], "6" * 40,
+        )
+        assert result["ok"] is False
+        state_fd = state.acquire_file_lock(state.STATE_LOCK_PATH, timeout_sec=0.2)
+        assert state_fd is not None
+        state.release_file_lock(state.STATE_LOCK_PATH, state_fd)
+    finally:
+        release_exclusive_file_lock(sidecar, sidecar_fd)
+
+
+def test_sent_owner_report_clear_cannot_erase_concurrent_commit_receipt(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle, queue
+
+    campaign, tx = _active_transaction(tmp_path)
+    report = {"cycle_outcome": "absorbed", "task_id": "previous"}
+    current = evolution_lifecycle._read_evolution_campaign()
+    current["pending_owner_report"] = report
+    assert evolution_lifecycle._write_evolution_campaign(current) is True
+
+    def _send_then_record(*args, **kwargs):
+        receipt = evolution_lifecycle.record_evolution_commit(
+            campaign["id"], tx["transaction_id"], tx["task_id"], "5" * 40,
+        )
+        assert receipt["ok"] is True
+
+    monkeypatch.setattr(queue, "notify_owner_cycle_outcome", _send_then_record)
+
+    queue._deliver_pending_owner_report()
+
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert "pending_owner_report" not in stored
+    assert stored["active_transaction"]["commit_receipt"]["commit_sha"] == "5" * 40
+
+
+def test_terminal_campaign_cannot_be_resurrected_by_a_stale_writer(tmp_path):
+    from supervisor import evolution_lifecycle
+
+    campaign, _ = _active_transaction(tmp_path)
+    stale = dict(campaign)
+    evolution_lifecycle.complete_evolution_campaign("owner stop", cleanup_worktree=False)
+    stale["status"] = "active"
+
+    assert evolution_lifecycle._write_evolution_campaign(stale) is False
+    assert evolution_lifecycle._read_evolution_campaign()["status"] == "stopped"
+
+
+def test_stale_campaign_cannot_overwrite_a_new_campaign(tmp_path):
+    from supervisor import evolution_lifecycle, queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    first = evolution_lifecycle.start_evolution_campaign("First", source="test")
+    stale = dict(first)
+    evolution_lifecycle.complete_evolution_campaign("done", cleanup_worktree=False)
+    second = evolution_lifecycle.start_evolution_campaign("Second", source="test")
+
+    stale["status"] = "active"
+    assert evolution_lifecycle._write_evolution_campaign(stale) is False
+    assert evolution_lifecycle._read_evolution_campaign()["id"] == second["id"]
+
+
+def test_panic_campaign_close_uses_nonblocking_state_lock(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle, queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+    timeouts = []
+    monkeypatch.setattr(
+        state,
+        "acquire_file_lock",
+        lambda path, timeout_sec=4.0, **kw: timeouts.append(timeout_sec) or None,
+    )
+
+    evolution_lifecycle.complete_evolution_campaign(
+        "panic stop", status="stopped", cleanup_worktree=False,
+    )
+
+    assert timeouts == [0.001]

tests/test_evolution_publication.py

@@ -0,0 +1,617 @@
+"""Publishing an evolution commit: the orphan ref, the git lock, and the authority to promote.
+
+Split out of ``tests/test_evolution_state_integrity_v3.py`` by theme: the orphan ref a later
+normal push cannot publish and its safe CAS fallback, the review refused when the claim is
+gone, the post-commit CAS and binding failures, the push that alone stays under the git
+lock, the revoked publication that anchors nothing, and the exact claim a promote carries.
+"""
+
+from __future__ import annotations
+
+import pathlib
+import subprocess
+from types import SimpleNamespace
+
+import pytest
+
+from tests._evolution_state_shared import _patch_commit_seam
+
+
+def test_evolution_orphan_ref_cannot_be_published_by_later_normal_push(
+    tmp_path, monkeypatch,
+):
+    from ouroboros.tools import git as git_tools
+    from supervisor import git_ops
+
+    repo, remote = tmp_path / "repo", tmp_path / "remote.git"
+
+    def _git(*args, cwd=repo, check=True):
+        return subprocess.run(
+            ["git", *args], cwd=cwd, check=check, capture_output=True, text=True,
+        )
+
+    subprocess.run(
+        ["git", "init", "--bare", str(remote)], check=True, capture_output=True, text=True,
+    )
+    repo.mkdir()
+    _git("init", "-b", "ouroboros")
+    _git("config", "user.name", "Test")
+    _git("config", "user.email", "test@example.com")
+    _git("remote", "add", "origin", str(remote))
+    (repo / "file.txt").write_text("base\n", encoding="utf-8")
+    (repo / "peer.txt").write_text("peer-base\n", encoding="utf-8")
+    _git("add", ".")
+    _git("commit", "-m", "base")
+    base_sha = _git("rev-parse", "HEAD").stdout.strip()
+    _git("tag", "-a", "v-base", "-m", "base")
+    _git("push", "-u", "origin", "ouroboros")
+    _git("push", "origin", "--tags")
+    (repo / "file.txt").write_text("orphan\n", encoding="utf-8")
+    _git("add", ".")
+    _git("commit", "-m", "orphan")
+    orphan_sha = _git("rev-parse", "HEAD").stdout.strip()
+    _git("tag", "-a", "v-orphan", "-m", "orphan")
+    (repo / "peer.txt").write_text("peer-concurrent-edit\n", encoding="utf-8")
+
+    note = git_tools._preserve_evolution_orphan(
+        SimpleNamespace(repo_dir=repo), orphan_sha, created_tag="v-orphan",
+    )
+
+    assert "CONTAINMENT_FAILED" not in note
+    assert _git("rev-parse", "HEAD").stdout.strip() == base_sha
+    private_ref = f"refs/ouroboros/evolution-orphans/{orphan_sha}"
+    assert _git("rev-parse", private_ref).stdout.strip() == orphan_sha
+    assert _git("show-ref", "--verify", "refs/tags/v-orphan", check=False).returncode != 0
+    assert _git("rev-parse", "refs/tags/v-base^{commit}").stdout.strip() == base_sha
+    assert (repo / "peer.txt").read_text(encoding="utf-8") == "peer-concurrent-edit\n"
+    assert _git("status", "--porcelain").stdout.strip()
+
+    monkeypatch.setattr(git_ops, "REPO_DIR", repo)
+    pushed, _message = git_ops.push_to_remote("ouroboros", push_tags=True)
+
+    assert pushed is True
+    assert _git("rev-parse", "refs/heads/ouroboros", cwd=remote).stdout.strip() == base_sha
+    assert _git("show-ref", "--verify", "refs/tags/v-orphan", cwd=remote, check=False).returncode != 0
+    assert _git("show-ref", "--verify", private_ref, cwd=remote, check=False).returncode != 0
+    assert _git("cat-file", "-e", orphan_sha, cwd=remote, check=False).returncode != 0
+
+    # A separate Git writer may advance the branch after the atomic containment
+    # transaction. Worktree alignment must not move that ref back to the parent.
+    (repo / "file.txt").write_text("second orphan\n", encoding="utf-8")
+    _git("add", ".")
+    _git("commit", "-m", "second orphan")
+    second_orphan = _git("rev-parse", "HEAD").stdout.strip()
+    base_tree = _git("rev-parse", f"{base_sha}^{{tree}}").stdout.strip()
+    concurrent = subprocess.run(
+        ["git", "commit-tree", base_tree, "-p", base_sha],
+        cwd=repo,
+        input="concurrent branch update\n",
+        text=True,
+        check=True,
+        capture_output=True,
+    ).stdout.strip()
+    real_subprocess_run = subprocess.run
+    interleaved = {"done": False}
+
+    def _interleave_after_ref_transaction(cmd, *args, **kwargs):
+        proc = real_subprocess_run(cmd, *args, **kwargs)
+        if cmd[:3] == ["git", "update-ref", "--stdin"] and proc.returncode == 0 and not interleaved["done"]:
+            real_subprocess_run(
+                ["git", "update-ref", "refs/heads/ouroboros", concurrent, base_sha],
+                cwd=repo, check=True, capture_output=True, text=True,
+            )
+            interleaved["done"] = True
+        return proc
+
+    monkeypatch.setattr(git_tools.subprocess, "run", _interleave_after_ref_transaction)
+    note = git_tools._preserve_evolution_orphan(
+        SimpleNamespace(repo_dir=repo), second_orphan,
+    )
+
+    assert "CONTAINMENT_FAILED" not in note
+    assert "concurrent branch update" in note
+    assert _git("rev-parse", "HEAD").stdout.strip() == concurrent
+    assert _git(
+        "rev-parse", f"refs/ouroboros/evolution-orphans/{second_orphan}",
+    ).stdout.strip() == second_orphan
+
+
+def test_orphan_ref_transaction_failure_falls_back_to_safe_ref_cas(tmp_path, monkeypatch):
+    from ouroboros.tools import git as git_tools
+    from supervisor import git_ops
+
+    repo, remote = tmp_path / "repo", tmp_path / "remote.git"
+    real_run = subprocess.run
+
+    def _git(*args, cwd=repo, check=True):
+        return real_run(
+            ["git", *args], cwd=cwd, check=check, capture_output=True, text=True,
+        )
+
+    real_run(["git", "init", "--bare", str(remote)], check=True, capture_output=True, text=True)
+    repo.mkdir()
+    _git("init", "-b", "ouroboros")
+    _git("config", "user.name", "Test")
+    _git("config", "user.email", "test@example.com")
+    _git("remote", "add", "origin", str(remote))
+    (repo / "file.txt").write_text("base\n", encoding="utf-8")
+    _git("add", ".")
+    _git("commit", "-m", "base")
+    base_sha = _git("rev-parse", "HEAD").stdout.strip()
+    _git("push", "-u", "origin", "ouroboros")
+    (repo / "file.txt").write_text("orphan\n", encoding="utf-8")
+    _git("add", ".")
+    _git("commit", "-m", "orphan")
+    orphan_sha = _git("rev-parse", "HEAD").stdout.strip()
+    _git("tag", "-a", "v-orphan", "-m", "orphan")
+
+    def _fail_transactions(cmd, *args, **kwargs):
+        if cmd[:3] == ["git", "update-ref", "--stdin"]:
+            # BYTES streams: the transaction call deliberately runs in binary mode
+            # (text-mode pipes CRLF-mangle --stdin commands on Windows).
+            return subprocess.CompletedProcess(cmd, 1, b"", b"injected transaction failure")
+        return real_run(cmd, *args, **kwargs)
+
+    monkeypatch.setattr(git_tools.subprocess, "run", _fail_transactions)
+
+    note = git_tools._preserve_evolution_orphan(
+        SimpleNamespace(repo_dir=repo), orphan_sha, created_tag="v-orphan",
+    )
+
+    assert "CONTAINMENT_FAILED" not in note
+    assert _git("rev-parse", "HEAD").stdout.strip() == base_sha
+    assert _git(
+        "rev-parse", f"refs/ouroboros/evolution-orphans/{orphan_sha}",
+    ).stdout.strip() == orphan_sha
+    assert _git("show-ref", "--verify", "refs/tags/v-orphan", check=False).returncode != 0
+
+    monkeypatch.setattr(git_ops, "REPO_DIR", repo)
+    pushed, _message = git_ops.push_to_remote("ouroboros", push_tags=True)
+
+    assert pushed is True
+    assert _git("rev-parse", "refs/heads/ouroboros", cwd=remote).stdout.strip() == base_sha
+    assert _git("cat-file", "-e", orphan_sha, cwd=remote, check=False).returncode != 0
+
+
+def test_evolution_commit_refuses_review_when_claim_is_gone(tmp_path, monkeypatch):
+    from ouroboros.tools import git as git_tools
+
+    reviewed = []
+    monkeypatch.setattr(git_tools, "_task_attributed_commit_paths", lambda *a, **k: (None, None, "", None))
+    monkeypatch.setattr(git_tools, "_check_overlapping_review_attempt", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_record_commit_attempt", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_acquire_git_lock", lambda *a, **k: pathlib.Path("lock"))
+    monkeypatch.setattr(git_tools, "_release_git_lock", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_prepare_review_commit_worktree", lambda *a, **k: (False, ""))
+    _patch_commit_seam(monkeypatch, "_evolution_commit_authority",
+        lambda *a, **k: ({}, {"ok": False, "reason": "owner_stopped"}),
+    )
+    _patch_commit_seam(monkeypatch, "_run_reviewed_stage_cycle",
+        lambda *a, **k: reviewed.append(True) or {"status": "passed"},
+    )
+    ctx = SimpleNamespace(
+        repo_dir=tmp_path,
+        drive_root=tmp_path,
+        branch_dev="ouroboros",
+        current_task_type="evolution",
+        task_id="evo",
+        task_metadata={},
+    )
+
+    result = git_tools._repo_commit_push(ctx, "test commit")
+
+    assert "EVOLUTION_AUTHORITY_REVOKED" in result
+    assert reviewed == []
+
+
+def test_postcommit_cas_failure_returns_local_orphan_after_binding(tmp_path, monkeypatch):
+    from ouroboros.tools import git as git_tools
+    from supervisor import evolution_lifecycle
+
+    claim = {"campaign_id": "camp", "transaction_id": "tx", "task_id": "evo"}
+    tagged, contained = [], []
+    monkeypatch.setattr(git_tools, "_task_attributed_commit_paths", lambda *a, **k: (None, None, "", None))
+    monkeypatch.setattr(git_tools, "_check_overlapping_review_attempt", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_record_commit_attempt", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_acquire_git_lock", lambda *a, **k: pathlib.Path("lock"))
+    monkeypatch.setattr(git_tools, "_release_git_lock", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_prepare_review_commit_worktree", lambda *a, **k: (False, ""))
+    _patch_commit_seam(monkeypatch, "_evolution_commit_authority", lambda *a, **k: (claim, {"ok": True}))
+    _patch_commit_seam(monkeypatch, "_verify_reviewed_commit_binding", lambda *a, **k: (True, ""))
+    _patch_commit_seam(monkeypatch, "_run_reviewed_stage_cycle",
+        lambda *a, **k: {
+            "status": "passed",
+            "pre_fingerprint": {"fingerprint": "pre"},
+            "post_fingerprint": {"fingerprint": "post", "binding": {}},
+        },
+    )
+    _patch_commit_seam(monkeypatch, "run_cmd",
+        lambda cmd, cwd=None: "d" * 40 if cmd[:3] == ["git", "rev-parse", "HEAD"] else "",
+    )
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "record_evolution_commit",
+        lambda **kwargs: {"ok": False, "reason": "owner_stopped", "commit_sha": kwargs["commit_sha"]},
+    )
+    monkeypatch.setattr(
+        git_tools,
+        "_auto_tag_on_version_bump",
+        lambda *a, **k: tagged.append(True) or "",
+    )
+    _patch_commit_seam(monkeypatch, "_preserve_evolution_orphan",
+        lambda *a, **k: contained.append((a, k)) or "contained",
+    )
+    ctx = SimpleNamespace(
+        repo_dir=tmp_path,
+        drive_root=tmp_path,
+        branch_dev="ouroboros",
+        current_task_type="evolution",
+        task_id="evo",
+        task_metadata={"evolution_transaction": claim},
+    )
+
+    result = git_tools._repo_commit_push(ctx, "test commit")
+
+    assert "EVOLUTION_COMMIT_ORPHANED" in result
+    assert "d" * 40 in result
+    assert tagged == [True]
+    assert len(contained) == 1
+
+
+@pytest.mark.parametrize(
+    ("task_type", "expected_order"),
+    [
+        ("evolution", ["authority", "push", "release", "publish"]),
+        ("task", ["release", "push", "publish"]),
+    ],
+)
+def test_only_evolution_push_stays_under_git_lock(
+    tmp_path, monkeypatch, task_type, expected_order,
+):
+    from ouroboros.tools import git as git_tools
+
+    claim = {"campaign_id": "camp", "transaction_id": "tx", "task_id": "evo"}
+    order = []
+    monkeypatch.setattr(git_tools, "_task_attributed_commit_paths", lambda *a, **k: (None, None, "", None))
+    monkeypatch.setattr(git_tools, "_check_overlapping_review_attempt", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_record_commit_attempt", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_acquire_git_lock", lambda *a, **k: pathlib.Path("lock"))
+    monkeypatch.setattr(git_tools, "_release_git_lock", lambda *a, **k: order.append("release"))
+    monkeypatch.setattr(git_tools, "_prepare_review_commit_worktree", lambda *a, **k: (False, ""))
+    _patch_commit_seam(monkeypatch, "_evolution_commit_authority", lambda *a, **k: (claim, {"ok": True}))
+    _patch_commit_seam(monkeypatch, "_verify_reviewed_commit_binding", lambda *a, **k: (True, ""))
+    _patch_commit_seam(monkeypatch, "_run_reviewed_stage_cycle",
+        lambda *a, **k: {
+            "status": "passed",
+            "pre_fingerprint": {"fingerprint": "pre"},
+            "post_fingerprint": {"fingerprint": "post", "binding": {}},
+        },
+    )
+    _patch_commit_seam(monkeypatch, "run_cmd",
+        lambda cmd, cwd=None: "d" * 40 if cmd[:3] == ["git", "rev-parse", "HEAD"] else "",
+    )
+    monkeypatch.setattr(git_tools, "_auto_tag_on_version_bump", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_record_evolution_commit_receipt", lambda *a, **k: "")
+    monkeypatch.setattr(
+        git_tools,
+        "_evolution_publication_stopped_result",
+        lambda *a, **k: order.append("authority") or "",
+    )
+    monkeypatch.setattr(git_tools, "_auto_push", lambda *a, **k: order.append("push") or "")
+    monkeypatch.setattr(
+        git_tools,
+        "_publish_reviewed_commit",
+        lambda *a, **k: order.append("publish") or "ok",
+    )
+    ctx = SimpleNamespace(
+        repo_dir=tmp_path,
+        drive_root=tmp_path,
+        branch_dev="ouroboros",
+        current_task_type=task_type,
+        task_id="evo",
+        task_metadata={"evolution_transaction": claim},
+    )
+
+    assert git_tools._repo_commit_push(ctx, "test commit", skip_tests=True) == "ok"
+    assert order == expected_order
+
+
+def test_revoked_publication_does_not_record_or_anchor_success(tmp_path, monkeypatch):
+    from ouroboros import mutation_attribution
+    from ouroboros.tools import git as git_tools
+
+    claim = {"campaign_id": "camp", "transaction_id": "tx", "task_id": "evo"}
+    sha = "d" * 40
+    attempts, baselines, contained, pushed = [], [], [], []
+    authority = iter([
+        (claim, {"ok": True}),
+        (claim, {"ok": True}),
+        (claim, {"ok": True}),
+        (claim, {"ok": False, "reason": "owner_stopped"}),
+    ])
+    monkeypatch.setattr(git_tools, "_task_attributed_commit_paths", lambda *a, **k: (None, None, "", ("root", "task")))
+    monkeypatch.setattr(git_tools, "_check_overlapping_review_attempt", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_record_commit_attempt", lambda *a, **k: attempts.append((k.get("status") or a[2], k)))
+    monkeypatch.setattr(git_tools, "_acquire_git_lock", lambda *a, **k: pathlib.Path("lock"))
+    monkeypatch.setattr(git_tools, "_release_git_lock", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_prepare_review_commit_worktree", lambda *a, **k: (False, ""))
+    _patch_commit_seam(monkeypatch, "_evolution_commit_authority", lambda *a, **k: next(authority))
+    _patch_commit_seam(monkeypatch, "_verify_reviewed_commit_binding", lambda *a, **k: (True, ""))
+    def reviewed(review_ctx, *args, **kwargs):
+        review_ctx._last_triad_raw_results = [{"raw": "triad"}]
+        review_ctx._last_scope_raw_result = {"raw": "scope"}
+        review_ctx._review_degraded_reasons = ["recorded"]
+        return {
+            "status": "passed",
+            "pre_fingerprint": {"fingerprint": "pre"},
+            "post_fingerprint": {"fingerprint": "post", "binding": {}},
+        }
+    _patch_commit_seam(monkeypatch, "_run_reviewed_stage_cycle", reviewed)
+    _patch_commit_seam(monkeypatch, "run_cmd", lambda cmd, cwd=None: sha if cmd[:3] == ["git", "rev-parse", "HEAD"] else "")
+    monkeypatch.setattr(git_tools, "_auto_tag_on_version_bump", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_record_evolution_commit_receipt", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_preserve_evolution_orphan", lambda *a, **k: contained.append(True) or "contained")
+    monkeypatch.setattr(git_tools, "_auto_push", lambda *a, **k: pushed.append(True) or "")
+    monkeypatch.setattr(mutation_attribution, "advance_mutation_baseline", lambda *a, **k: baselines.append(a))
+    ctx = SimpleNamespace(
+        repo_dir=tmp_path, drive_root=tmp_path, branch_dev="ouroboros",
+        current_task_type="evolution", task_id="evo",
+        task_metadata={"evolution_transaction": claim},
+        _scope_review_history={"keep": True},
+    )
+
+    result = git_tools._repo_commit_push(ctx, "test commit", skip_tests=True)
+
+    assert "EVOLUTION_PUBLICATION_STOPPED" in result
+    assert contained == [True]
+    assert pushed == []
+    assert baselines == []
+    statuses = [status for status, _details in attempts]
+    assert "succeeded" not in statuses and statuses[-1] == "failed"
+    failed = attempts[-1][1]
+    assert failed["fingerprint_status"] == "matched"
+    assert failed["pre_review_fingerprint"] == "pre"
+    assert failed["post_review_fingerprint"] == "post"
+    assert failed["triad_raw_results"] == [{"raw": "triad"}]
+    assert failed["scope_raw_result"] == {"raw": "scope"}
+    assert failed["degraded_reasons"] == ["recorded"]
+    assert not getattr(ctx, "last_reviewed_commit_sha", "")
+    assert ctx._scope_review_history == {"keep": True}
+
+
+def test_postcommit_binding_failure_contains_evolution_commit(tmp_path, monkeypatch):
+    from ouroboros.tools import git as git_tools
+
+    claim = {"campaign_id": "camp", "transaction_id": "tx", "task_id": "evo"}
+    contained = []
+    monkeypatch.setattr(git_tools, "_task_attributed_commit_paths", lambda *a, **k: (None, None, "", None))
+    monkeypatch.setattr(git_tools, "_check_overlapping_review_attempt", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_record_commit_attempt", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_acquire_git_lock", lambda *a, **k: pathlib.Path("lock"))
+    monkeypatch.setattr(git_tools, "_release_git_lock", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_prepare_review_commit_worktree", lambda *a, **k: (False, ""))
+    _patch_commit_seam(monkeypatch, "_evolution_commit_authority", lambda *a, **k: (claim, {"ok": True}))
+    _patch_commit_seam(monkeypatch, "_run_reviewed_stage_cycle",
+        lambda *a, **k: {
+            "status": "passed",
+            "pre_fingerprint": {"fingerprint": "pre"},
+            "post_fingerprint": {"fingerprint": "post", "binding": {}},
+        },
+    )
+    _patch_commit_seam(monkeypatch, "run_cmd",
+        lambda cmd, cwd=None: "2" * 40 if cmd[:3] == ["git", "rev-parse", "HEAD"] else "",
+    )
+    _patch_commit_seam(monkeypatch, "_verify_reviewed_commit_binding",
+        lambda *a, **k: (False, "tree mismatch"),
+    )
+    _patch_commit_seam(monkeypatch, "_preserve_evolution_orphan",
+        lambda *a, **k: contained.append((a, k)) or "contained",
+    )
+    ctx = SimpleNamespace(
+        repo_dir=tmp_path,
+        drive_root=tmp_path,
+        current_task_type="evolution",
+        task_id="evo",
+        task_metadata={"evolution_transaction": claim},
+    )
+
+    result = git_tools._repo_commit_push(ctx, "test commit")
+
+    assert "REVIEW_BINDING_FAILED" in result
+    assert "contained" in result
+    assert len(contained) == 1
+    assert contained[0][1] == {}
+
+
+def test_final_tag_binding_failure_cannot_record_restart_receipt(tmp_path, monkeypatch):
+    from ouroboros.tools import git as git_tools
+
+    claim = {"campaign_id": "camp", "transaction_id": "tx", "task_id": "evo"}
+    recorded = []
+    contained = []
+    binding_results = iter([(True, ""), (False, "tag target mismatch")])
+    monkeypatch.setattr(git_tools, "_task_attributed_commit_paths", lambda *a, **k: (None, None, "", None))
+    monkeypatch.setattr(git_tools, "_check_overlapping_review_attempt", lambda *a, **k: "")
+    _patch_commit_seam(monkeypatch, "_record_commit_attempt", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_acquire_git_lock", lambda *a, **k: pathlib.Path("lock"))
+    monkeypatch.setattr(git_tools, "_release_git_lock", lambda *a, **k: None)
+    monkeypatch.setattr(git_tools, "_prepare_review_commit_worktree", lambda *a, **k: (False, ""))
+    _patch_commit_seam(monkeypatch, "_evolution_commit_authority", lambda *a, **k: (claim, {"ok": True}))
+    _patch_commit_seam(monkeypatch, "_verify_reviewed_commit_binding", lambda *a, **k: next(binding_results),
+    )
+    _patch_commit_seam(monkeypatch, "_run_reviewed_stage_cycle",
+        lambda *a, **k: {
+            "status": "passed",
+            "pre_fingerprint": {"fingerprint": "pre"},
+            "post_fingerprint": {
+                "fingerprint": "post",
+                "binding": {"expected_tag": "v-test"},
+            },
+        },
+    )
+    _patch_commit_seam(monkeypatch, "run_cmd",
+        lambda cmd, cwd=None: "1" * 40 if cmd[:3] == ["git", "rev-parse", "HEAD"] else "",
+    )
+    monkeypatch.setattr(
+        git_tools,
+        "_auto_tag_on_version_bump",
+        lambda *a, **k: " [tagged: v-test]",
+    )
+    _patch_commit_seam(monkeypatch, "_preserve_evolution_orphan",
+        lambda *a, **k: contained.append((a, k)) or "contained",
+    )
+    _patch_commit_seam(monkeypatch, "_record_evolution_commit_receipt",
+        lambda *a, **k: recorded.append(True) or "",
+    )
+    ctx = SimpleNamespace(
+        repo_dir=tmp_path,
+        drive_root=tmp_path,
+        branch_dev="ouroboros",
+        current_task_type="evolution",
+        task_id="evo",
+        task_metadata={"evolution_transaction": claim},
+    )
+
+    result = git_tools._repo_commit_push(ctx, "test commit")
+
+    assert "REVIEW_BINDING_FAILED" in result
+    assert recorded == []
+    assert len(contained) == 1
+    assert contained[0][1]["created_tag"] == "v-test"
+
+
+def test_evolution_publication_authority_requires_exact_head(tmp_path, monkeypatch):
+    from ouroboros.tools import git as git_tools
+
+    monkeypatch.setattr(
+        "supervisor.evolution_lifecycle.check_evolution_authority",
+        lambda **kwargs: {"ok": True, "reason": ""},
+    )
+    _patch_commit_seam(monkeypatch, "run_cmd",
+        lambda cmd, cwd=None: "b" * 40 if cmd[:3] == ["git", "rev-parse", "HEAD"] else "",
+    )
+    ctx = SimpleNamespace(
+        repo_dir=tmp_path,
+        task_id="evo",
+        task_metadata={"evolution_transaction": {
+            "campaign_id": "camp",
+            "transaction_id": "tx",
+            "task_id": "evo",
+        }},
+    )
+
+    _, authority = git_tools._evolution_commit_authority(ctx, commit_sha="a" * 40)
+
+    assert authority["ok"] is False
+    assert authority["reason"] == "head_mismatch"
+
+
+def test_evolution_promote_event_carries_exact_claim():
+    from ouroboros.tools import control
+
+    ctx = SimpleNamespace(
+        current_task_type="evolution",
+        task_id="evo-task",
+        task_metadata={"evolution_transaction": {
+            "campaign_id": "campaign",
+            "transaction_id": "transaction",
+            "task_id": "evo-task",
+            "commit_sha": "",
+        }},
+        last_reviewed_commit_sha="a" * 40,
+        pending_events=[],
+    )
+
+    control._promote_to_stable(ctx, "reviewed")
+
+    event = ctx.pending_events[0]
+    assert event["type"] == "promote_to_stable"
+    assert event["reason"] == "reviewed"
+    assert event["evolution_claim"] == {
+        "campaign_id": "campaign",
+        "transaction_id": "transaction",
+        "task_id": "evo-task",
+        "commit_sha": "a" * 40,
+    }
+
+
+def test_promote_to_stable_rechecks_evolution_claim_without_changing_normal_flow(
+    tmp_path, monkeypatch,
+):
+    from supervisor import events, evolution_lifecycle
+
+    repo = tmp_path / "repo"
+    repo.mkdir()
+
+    def _git(*args):
+        return subprocess.run(
+            ["git", *args], cwd=repo, check=True, capture_output=True, text=True,
+        ).stdout.strip()
+
+    _git("init", "-b", "ouroboros")
+    _git("config", "user.name", "Test")
+    _git("config", "user.email", "test@example.com")
+    (repo / "file.txt").write_text("base\n", encoding="utf-8")
+    _git("add", ".")
+    _git("commit", "-m", "base")
+    base_sha = _git("rev-parse", "HEAD")
+    _git("branch", "ouroboros-stable", base_sha)
+    (repo / "file.txt").write_text("reviewed\n", encoding="utf-8")
+    _git("add", ".")
+    _git("commit", "-m", "reviewed")
+    reviewed_sha = _git("rev-parse", "HEAD")
+    sent = []
+    ctx = SimpleNamespace(
+        REPO_DIR=repo,
+        BRANCH_DEV="ouroboros",
+        BRANCH_STABLE="ouroboros-stable",
+        load_state=lambda: {"owner_chat_id": 1},
+        send_with_budget=lambda chat_id, message: sent.append(message),
+    )
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "check_evolution_authority",
+        lambda **claim: {
+            "ok": claim.get("campaign_id") == "valid",
+            "reason": "owner_stopped" if claim.get("campaign_id") != "valid" else "",
+        },
+    )
+
+    events._handle_promote_to_stable({
+        "type": "promote_to_stable",
+        "evolution_claim": {
+            "campaign_id": "revoked",
+            "transaction_id": "tx",
+            "task_id": "evo",
+            "commit_sha": reviewed_sha,
+        },
+    }, ctx)
+    assert _git("rev-parse", "ouroboros-stable") == base_sha
+    assert "owner_stopped" in sent[-1]
+
+    events._handle_promote_to_stable({
+        "type": "promote_to_stable",
+        "evolution_claim": {
+            "campaign_id": "",
+            "transaction_id": "",
+            "task_id": "",
+            "commit_sha": "",
+        },
+    }, ctx)
+    assert _git("rev-parse", "ouroboros-stable") == base_sha
+    assert "commit_receipt_missing" in sent[-1]
+
+    events._handle_promote_to_stable({
+        "type": "promote_to_stable",
+        "evolution_claim": {
+            "campaign_id": "valid",
+            "transaction_id": "tx",
+            "task_id": "evo",
+            "commit_sha": reviewed_sha,
+        },
+    }, ctx)
+    assert _git("rev-parse", "ouroboros-stable") == reviewed_sha
+
+    _git("branch", "-f", "ouroboros-stable", base_sha)
+    events._handle_promote_to_stable({"type": "promote_to_stable"}, ctx)
+    assert _git("rev-parse", "ouroboros-stable") == reviewed_sha

tests/test_evolution_restart_claims.py

@@ -0,0 +1,614 @@
+"""The restart claim: who may take it, who must wait, and what boot reconciliation may revive.
+
+Split out of ``tests/test_evolution_state_integrity_v3.py`` by theme: the exact active
+receipt a restart requires, the v2 claim verified only after a new generation, the losers
+that wait instead of bypassing, the dead claim reclaimed, the write failures that restore
+the claim for retry, the owner-stopped campaign boot cannot resurrect, and the supervisor
+rechecks that gate an evolution restart against a stale marker or a moved head.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import threading
+from types import SimpleNamespace
+
+import pytest
+
+from tests._evolution_state_shared import _active_transaction
+
+
+def test_restart_requires_the_exact_active_commit_receipt(tmp_path, monkeypatch):
+    from ouroboros.tools import control
+    from supervisor import evolution_lifecycle, state
+
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "e" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim, commit_sha=sha)["ok"] is True
+    monkeypatch.setattr(
+        control,
+        "run_cmd",
+        lambda cmd, cwd=None: "" if cmd[:2] == ["git", "status"] else sha,
+    )
+    ctx = SimpleNamespace(
+        current_task_type="evolution",
+        repo_dir=tmp_path,
+        task_id=tx["task_id"],
+        task_metadata={"evolution_transaction": tx},
+        last_reviewed_commit_sha=sha,
+    )
+    assert control._evolution_restart_block_reason(ctx) == ""
+
+    live = state.load_state()
+    live["evolution_owner_stopped"] = True
+    live["evolution_mode_enabled"] = False
+    state.save_state(live)
+    assert "owner_stopped" in control._evolution_restart_block_reason(ctx)
+
+
+def test_boot_restart_verifies_exact_v2_claim_only_after_new_generation(
+    tmp_path, monkeypatch,
+):
+    from ouroboros import agent_startup_checks, process_custody
+    from supervisor import evolution_lifecycle
+
+    generation = {"value": "server-a"}
+    monkeypatch.setattr(
+        process_custody, "current_custody_session_id", lambda: generation["value"],
+    )
+    campaign, tx = _active_transaction(tmp_path)
+    assert tx["schema_version"] == 2
+    sha = "8" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim)["ok"] is True
+    marker = tmp_path / "state" / "pending_restart_verify.json"
+    marker.write_text(json.dumps({"expected_sha": sha, "evolution_claim": claim}))
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+
+    agent_startup_checks.verify_restart(env, sha)
+
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert stored["active_transaction"]["transaction_id"] == tx["transaction_id"]
+    assert int(stored.get("absorbed_cycles_done") or 0) == 0
+    assert marker.is_file()
+
+    generation["value"] = "server-b"
+    agent_startup_checks.verify_restart(env, sha)
+
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert "active_transaction" not in stored
+    assert stored["transaction_history"][-1]["cycle_outcome"] == "absorbed"
+    assert stored["last_boot_reconcile_gen"] == "server-b"
+    assert not marker.exists()
+
+
+def test_boot_restart_rejects_mismatched_claim_without_loser_bypass(
+    tmp_path, monkeypatch,
+):
+    from ouroboros import agent_startup_checks, process_custody
+    from supervisor import evolution_lifecycle
+
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "9" * 40
+    exact = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(**exact)["ok"] is True
+    stale = {**exact, "transaction_id": "stale-transaction"}
+    marker = tmp_path / "state" / "pending_restart_verify.json"
+    marker.write_text(json.dumps({"expected_sha": sha, "evolution_claim": stale}))
+    monkeypatch.setattr(process_custody, "current_custody_session_id", lambda: "boot-gen")
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+
+    agent_startup_checks.verify_restart(env, sha)
+    agent_startup_checks.verify_restart(env, sha)  # a rename loser must not reconcile it
+
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert stored["active_transaction"]["transaction_id"] == tx["transaction_id"]
+    assert stored["active_transaction"]["restart_authority_error"] == "restart_claim_mismatch"
+    assert int(stored.get("absorbed_cycles_done") or 0) == 0
+    event = json.loads((tmp_path / "logs" / "events.jsonl").read_text().splitlines()[-1])
+    assert event["type"] == "restart_verify"
+    assert event["error"] == "restart_claim_mismatch"
+
+
+def test_boot_markerless_v2_missing_receipt_stays_unresolved(tmp_path, monkeypatch):
+    from ouroboros import agent_startup_checks, process_custody
+    from supervisor import evolution_lifecycle
+
+    campaign, tx = _active_transaction(tmp_path)
+    stored = evolution_lifecycle._read_evolution_campaign()
+    stored["active_transaction"]["commit_sha"] = "a" * 40
+    assert evolution_lifecycle._write_evolution_campaign(stored) is True
+    monkeypatch.setattr(process_custody, "current_custody_session_id", lambda: "boot-gen")
+    monkeypatch.setattr(
+        agent_startup_checks.subprocess,
+        "run",
+        lambda *a, **k: SimpleNamespace(returncode=0, stdout="", stderr=""),
+    )
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+
+    agent_startup_checks.verify_restart(env, "head")
+
+    current = evolution_lifecycle._read_evolution_campaign()
+    assert current["active_transaction"]["transaction_id"] == tx["transaction_id"]
+    assert current["active_transaction"]["restart_authority_error"] == "commit_receipt_missing"
+    assert int(current.get("absorbed_cycles_done") or 0) == 0
+    event = json.loads((tmp_path / "logs" / "events.jsonl").read_text().splitlines()[-1])
+    assert event["type"] == "evolution_tx_reconcile_blocked"
+    assert event["reason"] == "commit_receipt_missing"
+
+
+def test_boot_rename_loser_waits_for_claim_winner(tmp_path):
+    from ouroboros import agent_startup_checks
+    from supervisor import evolution_lifecycle
+
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "b" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim)["ok"] is True
+    claimed = tmp_path / "state" / f"pending_restart_verify.claimed.{os.getpid()}.json"
+    claimed.write_text(json.dumps({"expected_sha": sha, "evolution_claim": claim}))
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+
+    agent_startup_checks.verify_restart(env, sha)
+
+    current = evolution_lifecycle._read_evolution_campaign()
+    assert current["active_transaction"]["transaction_id"] == tx["transaction_id"]
+    assert int(current.get("absorbed_cycles_done") or 0) == 0
+
+
+def test_boot_reclaims_dead_restart_claim(tmp_path, monkeypatch):
+    from ouroboros import agent_startup_checks, platform_layer, process_custody
+    from supervisor import evolution_lifecycle
+
+    generation = {"value": "server-a"}
+    monkeypatch.setattr(
+        process_custody, "current_custody_session_id", lambda: generation["value"],
+    )
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "7" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim)["ok"] is True
+    claimed = tmp_path / "state" / "pending_restart_verify.claimed.999999999.json"
+    claimed.write_text(json.dumps({"expected_sha": sha, "evolution_claim": claim}))
+    monkeypatch.setattr(platform_layer, "pid_is_alive", lambda pid: False)
+    generation["value"] = "server-b"
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+
+    agent_startup_checks.verify_restart(env, sha)
+
+    current = evolution_lifecycle._read_evolution_campaign()
+    assert "active_transaction" not in current
+    assert current["transaction_history"][-1]["cycle_outcome"] == "absorbed"
+    assert list((tmp_path / "state").glob("pending_restart_verify.claimed.*.json")) == []
+
+
+def test_new_campaign_is_stamped_for_same_generation_worker_respawns(tmp_path, monkeypatch):
+    from ouroboros import agent_startup_checks, process_custody
+    from supervisor import evolution_lifecycle, queue, state
+
+    monkeypatch.setattr(process_custody, "current_custody_session_id", lambda: "same-server")
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    queue.init_queue_refs([], {}, {"value": 0})
+    campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+    assert campaign["last_boot_reconcile_gen"] == "same-server"
+    state.update_state(lambda live: live.update(evolution_mode_enabled=True))
+    tx = evolution_lifecycle.begin_evolution_transaction("respawn", cycle=1, campaign=campaign)
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": "6" * 40,
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim)["ok"] is True
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+
+    agent_startup_checks.verify_restart(env, claim["commit_sha"])
+
+    current = evolution_lifecycle._read_evolution_campaign()
+    assert current["active_transaction"]["transaction_id"] == tx["transaction_id"]
+    assert int(current.get("absorbed_cycles_done") or 0) == 0
+
+
+def test_boot_reconcile_cannot_resurrect_owner_stopped_campaign(tmp_path, monkeypatch):
+    from ouroboros import agent_startup_checks, process_custody
+    from supervisor import evolution_lifecycle
+
+    generation = {"value": "before-restart"}
+    monkeypatch.setattr(
+        process_custody, "current_custody_session_id", lambda: generation["value"],
+    )
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "4" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim)["ok"] is True
+    generation["value"] = "after-restart"
+    reached = threading.Event()
+    release = threading.Event()
+
+    def delayed_merge_base(*args, **kwargs):
+        reached.set()
+        assert release.wait(timeout=2)
+        return SimpleNamespace(returncode=0)
+
+    monkeypatch.setattr(agent_startup_checks.subprocess, "run", delayed_merge_base)
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+    worker = threading.Thread(target=agent_startup_checks.verify_restart, args=(env, "5" * 40))
+    worker.start()
+    assert reached.wait(timeout=2)
+    evolution_lifecycle.complete_evolution_campaign("owner stop", cleanup_worktree=False)
+    release.set()
+    worker.join(timeout=2)
+
+    current = evolution_lifecycle._read_evolution_campaign()
+    assert current["status"] == "stopped"
+    assert current["completion_reason"] == "owner stop"
+    assert "active_transaction" not in current
+
+
+def test_owner_stop_preserves_prior_boot_reconciliation_evidence(tmp_path, monkeypatch):
+    from ouroboros import agent_startup_checks, process_custody
+    from supervisor import evolution_lifecycle
+
+    generation = {"value": "before-restart"}
+    monkeypatch.setattr(
+        process_custody, "current_custody_session_id", lambda: generation["value"],
+    )
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "3" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim)["ok"] is True
+    generation["value"] = "after-restart"
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+    agent_startup_checks.verify_restart(env, sha)
+    evolution_lifecycle.complete_evolution_campaign("owner stop", cleanup_worktree=False)
+
+    current = evolution_lifecycle._read_evolution_campaign()
+    assert current["status"] == "stopped"
+    assert current["absorbed_cycles_done"] == 1
+    assert current["last_boot_reconcile_gen"] == "after-restart"
+    assert current["transaction_history"][-1]["cycle_outcome"] == "absorbed"
+
+
+def test_boot_restart_writers_obey_live_root_fuse(tmp_path, monkeypatch):
+    from ouroboros import agent_startup_checks
+    from supervisor import evolution_lifecycle
+
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "2" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim)["ok"] is True
+    campaign_path = tmp_path / "state" / "evolution_campaign.json"
+    before = campaign_path.read_bytes()
+    monkeypatch.setenv("OUROBOROS_PYTEST_ACTIVE", "1")
+    monkeypatch.setenv("OUROBOROS_TEST_LIVE_DATA_ROOT", str(tmp_path))
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+
+    with pytest.raises(RuntimeError, match="PYTEST_LIVE_DATA_WRITE_BLOCKED"):
+        agent_startup_checks.verify_restart(env, sha)
+
+    assert campaign_path.read_bytes() == before
+    assert list((tmp_path / "state").glob("pending_restart_verify.claimed.*.json")) == []
+
+
+def test_boot_restart_write_failure_restores_claim_for_retry(tmp_path, monkeypatch):
+    from ouroboros import agent_startup_checks, process_custody
+    from supervisor import evolution_lifecycle
+
+    generation = {"value": "server-a"}
+    monkeypatch.setattr(
+        process_custody, "current_custody_session_id", lambda: generation["value"],
+    )
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "c" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(**claim)["ok"] is True
+    pending = tmp_path / "state" / "pending_restart_verify.json"
+    pending.write_text(json.dumps({"expected_sha": sha, "evolution_claim": claim}))
+    real_write = agent_startup_checks.atomic_write_json
+    calls = {"count": 0}
+
+    def fail_once(*args, **kwargs):
+        calls["count"] += 1
+        if calls["count"] == 1:
+            raise OSError("temporary write failure")
+        return real_write(*args, **kwargs)
+
+    monkeypatch.setattr(agent_startup_checks, "atomic_write_json", fail_once)
+    generation["value"] = "server-b"
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+
+    agent_startup_checks.verify_restart(env, sha)
+    assert pending.is_file()
+    assert list((tmp_path / "state").glob("pending_restart_verify.claimed.*.json")) == []
+
+    agent_startup_checks.verify_restart(env, sha)
+    current = evolution_lifecycle._read_evolution_campaign()
+    assert "active_transaction" not in current
+    assert current["transaction_history"][-1]["cycle_outcome"] == "absorbed"
+
+
+def test_boot_exact_claim_never_passes_without_active_transaction(tmp_path):
+    from ouroboros import agent_startup_checks
+
+    (tmp_path / "state").mkdir(parents=True)
+    (tmp_path / "logs").mkdir(parents=True)
+    sha = "d" * 40
+    claim = {
+        "campaign_id": "campaign",
+        "transaction_id": "transaction",
+        "task_id": "task",
+        "commit_sha": sha,
+    }
+    marker = tmp_path / "state" / "pending_restart_verify.json"
+    marker.write_text(json.dumps({"expected_sha": sha, "evolution_claim": claim}))
+    campaign_path = tmp_path / "state" / "evolution_campaign.json"
+    campaign_path.write_text("{bad")
+    env = SimpleNamespace(
+        drive_path=lambda name: tmp_path / name,
+        drive_root=tmp_path,
+        repo_dir=tmp_path,
+    )
+
+    agent_startup_checks.verify_restart(env, sha)
+
+    event = json.loads((tmp_path / "logs" / "events.jsonl").read_text().splitlines()[-1])
+    assert event["type"] == "restart_verify"
+    assert event["ok"] is False
+    assert event["error"] == "transaction_missing"
+    assert marker.exists() is False
+    assert campaign_path.read_text() == "{bad"
+
+
+def test_evolution_restart_write_failure_does_not_become_generic_restart(tmp_path, monkeypatch):
+    from ouroboros.tools import control
+
+    monkeypatch.setattr(control, "_evolution_restart_block_reason", lambda ctx: "")
+    monkeypatch.setattr(
+        control, "run_cmd",
+        lambda cmd, cwd=None: "f" * 40 if cmd[-1] == "HEAD" else "ouroboros",
+    )
+    monkeypatch.setattr(
+        control, "atomic_write_json",
+        lambda *a, **k: (_ for _ in ()).throw(OSError("disk full")),
+    )
+    ctx = SimpleNamespace(
+        current_task_type="evolution",
+        repo_dir=tmp_path,
+        drive_path=lambda name: tmp_path / name,
+        task_id="evo",
+        task_metadata={"evolution_transaction": {}},
+        pending_restart_reason=None,
+        last_push_succeeded=True,
+        last_reviewed_commit_sha="f" * 40,
+    )
+
+    result = control._request_restart(ctx, "apply reviewed evolution")
+
+    assert "RESTART_BLOCKED" in result
+    assert ctx.pending_restart_reason is None
+
+
+def test_supervisor_rechecks_evolution_claim_immediately_before_restart(tmp_path):
+    import server
+    from supervisor import evolution_lifecycle, state
+
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "1" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(
+        campaign["id"], tx["transaction_id"], tx["task_id"], sha,
+    )["ok"] is True
+    marker = tmp_path / "state" / "pending_restart_verify.json"
+    marker.write_text(json.dumps({
+        "reason": "evolution restart",
+        "expected_sha": sha,
+        "evolution_claim": claim,
+    }))
+    live = state.load_state()
+    live.update({"evolution_mode_enabled": False, "evolution_owner_stopped": True})
+    state.save_state(live)
+    restarted = []
+    messages = []
+    ctx = SimpleNamespace(
+        DRIVE_ROOT=tmp_path,
+        load_state=state.load_state,
+        safe_restart=lambda **k: restarted.append(k) or (True, "ok"),
+        send_with_budget=lambda *a: messages.append(a),
+    )
+
+    server._perform_supervisor_restart(
+        ctx, restart_reason="evolution restart", evolution_restart=True,
+    )
+
+    assert restarted == []
+    assert "owner_stopped" in messages[0][1]
+
+
+def test_supervisor_blocks_evolution_restart_if_marker_disappears_during_drain(tmp_path):
+    import server
+
+    restarted = []
+    messages = []
+    ctx = SimpleNamespace(
+        DRIVE_ROOT=tmp_path,
+        load_state=lambda: {"owner_chat_id": 1},
+        safe_restart=lambda **k: restarted.append(k) or (True, "ok"),
+        send_with_budget=lambda *a: messages.append(a),
+    )
+
+    server._perform_supervisor_restart(
+        ctx,
+        restart_reason="agent_requested_restart",
+        evolution_restart=True,
+    )
+
+    assert restarted == []
+    assert "receipt is missing" in messages[0][1]
+
+
+def test_generic_restart_ignores_stale_evolution_marker(tmp_path, monkeypatch):
+    import server
+
+    marker = tmp_path / "state" / "pending_restart_verify.json"
+    marker.parent.mkdir(parents=True)
+    marker.write_text(json.dumps({
+        "reason": "agent_requested_restart",
+        "evolution_claim": {"campaign_id": "stale"},
+    }))
+    restarted = []
+    exited = []
+    monkeypatch.setattr(server, "_request_restart_exit", lambda: exited.append(True))
+    ctx = SimpleNamespace(
+        DRIVE_ROOT=tmp_path,
+        load_state=lambda: {},
+        safe_restart=lambda **k: restarted.append(k) or (True, "ok"),
+        kill_workers=lambda **k: None,
+        save_state=lambda state: None,
+        persist_queue_snapshot=lambda **k: None,
+    )
+
+    server._perform_supervisor_restart(
+        ctx, restart_reason="agent_requested_restart", evolution_restart=False,
+    )
+
+    assert restarted
+    assert exited == [True]
+
+
+def test_supervisor_blocks_restart_when_head_moved_after_receipt(tmp_path):
+    import server
+    from supervisor import evolution_lifecycle
+
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
+    subprocess.run(["git", "config", "user.name", "Test"], cwd=repo, check=True)
+    subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=repo, check=True)
+    (repo / "file.txt").write_text("current\n")
+    subprocess.run(["git", "add", "."], cwd=repo, check=True)
+    subprocess.run(["git", "commit", "-m", "current"], cwd=repo, check=True, capture_output=True)
+    campaign, tx = _active_transaction(tmp_path)
+    reviewed_sha = "2" * 40
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": reviewed_sha,
+    }
+    assert evolution_lifecycle.record_evolution_commit(
+        campaign["id"], tx["transaction_id"], tx["task_id"], reviewed_sha,
+    )["ok"] is True
+    (tmp_path / "state" / "pending_restart_verify.json").write_text(json.dumps({
+        "reason": "evolution restart",
+        "expected_sha": reviewed_sha,
+        "evolution_claim": claim,
+    }))
+    restarted = []
+    messages = []
+    ctx = SimpleNamespace(
+        DRIVE_ROOT=tmp_path,
+        REPO_DIR=repo,
+        load_state=lambda: {"owner_chat_id": 1},
+        safe_restart=lambda **k: restarted.append(k) or (True, "ok"),
+        send_with_budget=lambda *a: messages.append(a),
+    )
+
+    server._perform_supervisor_restart(
+        ctx, restart_reason="evolution restart", evolution_restart=True,
+    )
+
+    assert restarted == []
+    assert "no longer matches" in messages[0][1]

tests/test_evolution_scheduler.py

@@ -0,0 +1,320 @@
+"""What the evolution scheduler enqueues, replaces and refuses, and what assignment dispatches.
+
+Split out of ``tests/test_evolution_state_integrity_v3.py`` by theme: the bare flag with no
+campaign, the active campaign with no source, the owner resume that repairs a legacy
+source, the transaction attach and its owner-stop recheck, the uncommitted transaction
+replaced only when no worker is reaping, and the exact claim assignment must see.
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+from tests._evolution_state_shared import (
+    _CaptureQueue,
+    _active_transaction,
+)
+
+
+def _assignment_case(tmp_path, monkeypatch, task_id="assign-evo"):
+    from supervisor import evolution_lifecycle, queue, state, workers
+
+    state.init(tmp_path)
+    monkeypatch.setattr(state, "TOTAL_BUDGET_LIMIT", 0.0)
+    pending, running = [], {}
+    monkeypatch.setattr(workers, "PENDING", pending)
+    monkeypatch.setattr(workers, "RUNNING", running)
+    workers.init(tmp_path, tmp_path, 1, 600, 1800, 0.0)
+    campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+    state.update_state(lambda live: live.update(
+        evolution_mode_enabled=True,
+        evolution_owner_stopped=False,
+    ))
+    tx = evolution_lifecycle.begin_evolution_transaction(task_id, cycle=1, campaign=campaign)
+    task = {
+        "id": task_id,
+        "type": "evolution",
+        "text": "Improve",
+        "metadata": {"evolution_transaction": dict(tx)},
+    }
+    pending.append(task)
+    inbox, events = _CaptureQueue(), _CaptureQueue()
+    worker = SimpleNamespace(wid=1, busy_task_id=None, reaping=False, in_q=inbox)
+    monkeypatch.setattr(workers, "WORKERS", {1: worker})
+    monkeypatch.setattr(workers, "get_event_q", lambda: events)
+    monkeypatch.setattr(queue, "persist_queue_snapshot", lambda reason="": None)
+    monkeypatch.setattr(evolution_lifecycle, "evolution_block_reason", lambda: "")
+    return workers, task, tx, worker, inbox, events
+
+
+def test_scheduler_disables_a_bare_flag_without_campaign(tmp_path, monkeypatch):
+    from supervisor import queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    pending = []
+    queue.init_queue_refs(pending, {}, {"value": 0})
+    live = state.load_state()
+    live.update({
+        "owner_chat_id": 1,
+        "evolution_mode_enabled": True,
+        "post_task_autostop": True,
+    })
+    state.save_state(live)
+    sent = []
+    monkeypatch.setattr(queue, "send_with_budget", lambda *args, **kwargs: sent.append(args[1]))
+
+    queue.enqueue_evolution_task_if_needed()
+
+    assert pending == []
+    assert state.load_state()["evolution_mode_enabled"] is False
+    assert state.load_state()["post_task_autostop"] is False
+    assert "active campaign authority" in sent[0]
+    event = json.loads((tmp_path / "logs" / "events.jsonl").read_text().splitlines()[-1])
+    assert event["type"] == "evolution_authority_missing"
+
+
+def test_scheduler_refuses_active_campaign_without_source(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle, queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    queue.init_queue_refs([], {}, {"value": 0})
+    campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+    campaign.pop("source")
+    assert evolution_lifecycle._write_evolution_campaign(campaign) is True
+    live = state.load_state()
+    live.update({"owner_chat_id": 1, "evolution_mode_enabled": True})
+    state.save_state(live)
+    monkeypatch.setattr(queue, "send_with_budget", lambda *a, **k: None)
+
+    queue.enqueue_evolution_task_if_needed()
+
+    assert state.load_state()["evolution_mode_enabled"] is False
+
+
+def test_owner_resume_repairs_missing_legacy_campaign_source(tmp_path):
+    from supervisor import evolution_lifecycle, queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+    campaign["status"] = "paused"
+    campaign.pop("source")
+    assert evolution_lifecycle._write_evolution_campaign(campaign) is True
+
+    resumed = evolution_lifecycle.start_evolution_campaign("", source="owner_chat")
+
+    assert resumed["status"] == "active"
+    assert resumed["source"] == "owner_chat"
+
+
+def test_scheduler_does_not_enqueue_when_transaction_attach_fails(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle, queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    pending = []
+    queue.init_queue_refs(pending, {}, {"value": 0})
+    evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+    live = state.load_state()
+    live.update({"owner_chat_id": 1, "evolution_mode_enabled": True})
+    state.save_state(live)
+    monkeypatch.setattr(queue, "begin_evolution_transaction", lambda *a, **k: {})
+    monkeypatch.setattr(queue, "send_with_budget", lambda *a, **k: None)
+
+    queue.enqueue_evolution_task_if_needed()
+
+    assert pending == []
+    assert state.load_state()["evolution_mode_enabled"] is False
+
+
+def test_transaction_attach_rechecks_owner_stop_under_state_lock(tmp_path):
+    from supervisor import evolution_lifecycle, queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+    live = state.load_state()
+    live.update({"evolution_mode_enabled": False, "evolution_owner_stopped": True})
+    state.save_state(live)
+
+    tx = evolution_lifecycle.begin_evolution_transaction(
+        "too-late", cycle=1, campaign=campaign,
+    )
+
+    assert tx == {}
+    assert "active_transaction" not in evolution_lifecycle._read_evolution_campaign()
+
+
+def test_scheduler_replaces_uncommitted_transaction_lost_before_enqueue(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle, queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    pending = []
+    queue.init_queue_refs(pending, {}, {"value": 0})
+    campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+    live = state.load_state()
+    live.update({"owner_chat_id": 1, "evolution_mode_enabled": True})
+    state.save_state(live)
+    lost = evolution_lifecycle.begin_evolution_transaction(
+        "lost-before-enqueue", cycle=1, campaign=campaign,
+    )
+    monkeypatch.setattr(queue, "send_with_budget", lambda *a, **k: None)
+
+    queue.enqueue_evolution_task_if_needed()
+
+    assert len(pending) == 1
+    replacement = pending[0]["metadata"]["evolution_transaction"]
+    assert replacement["transaction_id"] != lost["transaction_id"]
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert stored["active_transaction"]["transaction_id"] == replacement["transaction_id"]
+    assert stored["transaction_history"][-1]["abandoned_reason"] == "dispatch_not_persisted"
+
+
+def test_scheduler_does_not_replace_transaction_while_worker_is_reaping(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle, queue
+
+    _campaign, tx = _active_transaction(tmp_path, task_id="reaping-evolution")
+    pending = []
+    queue.init_queue_refs(pending, {}, {"value": 0})
+    assert evolution_lifecycle.update_evolution_transaction(
+        tx["task_id"], dispatch_status="reaping",
+    )
+    monkeypatch.setattr(queue, "send_with_budget", lambda *args, **kwargs: None)
+
+    queue.enqueue_evolution_task_if_needed()
+
+    assert pending == []
+    stored = evolution_lifecycle._read_evolution_campaign()["active_transaction"]
+    assert stored["transaction_id"] == tx["transaction_id"]
+    assert stored["dispatch_status"] == "reaping"
+
+
+def test_timeout_marks_evolution_reaping_before_scheduler_can_replace_it(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle, queue
+
+    _campaign, tx = _active_transaction(tmp_path, task_id="timeout-evolution")
+    pending = []
+    running = {
+        tx["task_id"]: {
+            "task": {
+                "id": tx["task_id"],
+                "type": "evolution",
+                "chat_id": 1,
+                "metadata": {"evolution_transaction": dict(tx)},
+            },
+            "started_at": 1.0,
+            "last_heartbeat_at": 1.0,
+            "worker_id": 7,
+            "attempt": 1,
+        }
+    }
+    queue.init_queue_refs(pending, running, {"value": 0})
+    worker = SimpleNamespace(busy_task_id=tx["task_id"], proc=None, reaping=False)
+    workers_view = SimpleNamespace(WORKERS={7: worker})
+    reaper_jobs = _CaptureQueue()
+    monkeypatch.setattr(queue, "FINALIZATION_GRACE_SEC", 0)
+    monkeypatch.setattr(queue, "get_task_idle_timeout_sec", lambda: 1)
+    monkeypatch.setattr(queue, "get_per_call_timeout_ceiling_sec", lambda: 1)
+    monkeypatch.setattr(queue, "get_task_abs_ceiling_sec", lambda: 10)
+    monkeypatch.setattr(queue, "_ensure_reaper_started", lambda: None)
+    monkeypatch.setattr(queue, "_reap_queue", reaper_jobs)
+    monkeypatch.setattr(queue, "persist_queue_snapshot", lambda reason="": True)
+    monkeypatch.setattr(queue, "send_with_budget", lambda *args, **kwargs: None)
+
+    queue._enforce_task_timeouts_locked(
+        workers_view, now=1000.0, owner_chat_id=1,
+        st={"evolution_mode_enabled": True},
+    )
+
+    assert running == {}
+    assert worker.reaping is True
+    assert len(reaper_jobs.items) == 1
+    stored = evolution_lifecycle._read_evolution_campaign()["active_transaction"]
+    assert stored["dispatch_status"] == "reaping"
+
+    queue.enqueue_evolution_task_if_needed()
+    assert pending == []
+    assert evolution_lifecycle._read_evolution_campaign()["active_transaction"][
+        "transaction_id"
+    ] == tx["transaction_id"]
+
+
+def test_assignment_dispatches_exact_uncommitted_evolution_claim(tmp_path, monkeypatch):
+    workers, task, _tx, worker, inbox, events = _assignment_case(tmp_path, monkeypatch)
+
+    workers.assign_tasks()
+
+    assert inbox.items == [task]
+    assert worker.busy_task_id == task["id"]
+    assert workers.RUNNING[task["id"]]["task"] == task
+    assert events.items == []
+
+
+def test_assignment_rejects_stale_or_committed_evolution_claim(tmp_path, monkeypatch):
+    from ouroboros.task_results import load_task_result
+    from supervisor import evolution_lifecycle
+
+    workers, task, tx, worker, inbox, events = _assignment_case(tmp_path, monkeypatch)
+    task["metadata"]["evolution_transaction"]["task_id"] = "other-task"
+
+    workers.assign_tasks()
+
+    assert inbox.items == []
+    assert workers.RUNNING == {}
+    assert worker.busy_task_id is None
+    stored = load_task_result(tmp_path, task["id"])
+    assert stored["status"] == "cancelled"
+    assert stored["reason_code"] == "evolution_authority_missing"
+    assert stored["authority_reason"] == "task_mismatch"
+    assert events.items[-1]["metadata"]["evolution_transaction"]["task_id"] == "other-task"
+
+    workers, task, tx, _worker, inbox, _events = _assignment_case(
+        tmp_path / "committed", monkeypatch, task_id="committed-evo",
+    )
+    campaign = evolution_lifecycle._read_evolution_campaign()
+    assert evolution_lifecycle.record_evolution_commit(
+        campaign["id"], tx["transaction_id"], tx["task_id"], "a" * 40,
+    )["ok"] is True
+
+    workers.assign_tasks()
+
+    assert inbox.items == []
+    assert load_task_result(tmp_path / "committed", task["id"])["authority_reason"] == (
+        "transaction_already_committed"
+    )
+
+
+def test_assignment_keeps_invalid_evolution_pending_when_cancel_write_fails(
+    tmp_path, monkeypatch,
+):
+    from supervisor import workers as workers_module
+
+    workers, task, _tx, worker, inbox, events = _assignment_case(tmp_path, monkeypatch)
+    task["metadata"]["evolution_transaction"]["task_id"] = "other-task"
+    monkeypatch.setattr(
+        "ouroboros.task_results.write_task_result",
+        lambda *_a, **_k: (_ for _ in ()).throw(OSError("disk full")),
+    )
+
+    workers.assign_tasks()
+
+    assert workers_module.PENDING == [task]
+    assert worker.busy_task_id is None
+    assert inbox.items == []
+    assert events.items == []
+
+
+def test_benchmark_seed_creates_campaign_before_enabling(tmp_path):
+    from devtools.benchmarks.common.server_runner import seed_owner_state
+
+    seed_owner_state(tmp_path, evolution_enabled=True)
+
+    state = json.loads((tmp_path / "state" / "state.json").read_text())
+    campaign = json.loads((tmp_path / "state" / "evolution_campaign.json").read_text())
+    assert campaign["status"] == "active"
+    assert campaign["id"]
+    assert state["evolution_mode_enabled"] is True

tests/test_evolution_terminal_events.py

@@ -0,0 +1,351 @@
+"""The terminal event write: which campaign it may touch, and what it must leave alone.
+
+Split out of ``tests/test_evolution_state_integrity_v3.py`` by theme: the terminal that
+cannot write into a different campaign, the metadata-less one that cannot mutate the
+active campaign, the duplicates that resume pending cleanup and a missing restart request,
+the exact model reason a restart preserves, the serialized concurrent pause, and the
+exception and rejection paths that leave no lifecycle side effects.
+"""
+
+from __future__ import annotations
+
+import json
+import pathlib
+import threading
+from types import SimpleNamespace
+
+from tests._evolution_state_shared import (
+    _CaptureQueue,
+    _active_transaction,
+)
+
+
+def test_terminal_event_cannot_write_into_a_different_campaign(tmp_path):
+    from supervisor import evolution_lifecycle
+
+    campaign, tx = _active_transaction(tmp_path, task_id="same-task")
+    stale = {
+        **tx,
+        "campaign_id": "old-campaign",
+        "transaction_id": "old-transaction",
+    }
+
+    result = evolution_lifecycle.update_evolution_campaign_after_task(
+        "same-task",
+        cost_usd=1.0,
+        outcome_axes={"execution": {"status": "ok"}},
+        rounds=1,
+        transaction=stale,
+    )
+
+    assert result == {
+        "accepted": False,
+        "persisted": False,
+        "replay": False,
+        "reason": "transaction_mismatch",
+        "transaction": {},
+    }
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert stored["id"] == campaign["id"]
+    assert stored["active_transaction"]["transaction_id"] == tx["transaction_id"]
+    assert stored.get("history", []) == []
+
+
+def test_metadata_less_terminal_cannot_mutate_active_campaign(tmp_path):
+    from supervisor import evolution_lifecycle, queue, state
+
+    state.init(tmp_path)
+    queue.init(tmp_path, 600, 1800)
+    campaign = evolution_lifecycle.start_evolution_campaign("Improve", source="test")
+
+    result = evolution_lifecycle.update_evolution_campaign_after_task(
+        "stale-task",
+        cost_usd=1.25,
+        outcome_axes={"execution": {"status": "failed"}},
+        rounds=1,
+    )
+
+    assert result == {
+        "accepted": False,
+        "persisted": False,
+        "replay": False,
+        "reason": "transaction_missing",
+        "transaction": {},
+    }
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert stored["id"] == campaign["id"]
+    assert stored["cycles_done"] == 0
+    assert stored["budget_spent_usd"] == 0.0
+    assert stored.get("history", []) == []
+
+
+def test_duplicate_terminal_resumes_pending_cleanup_and_owner_report(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle
+
+    _campaign, tx = _active_transaction(tmp_path)
+    assert evolution_lifecycle.update_evolution_transaction(
+        tx["task_id"], rescue_ref="refs/ouroboros/rescue/test",
+    )
+    real_resume = evolution_lifecycle._resume_evolution_terminal_effects
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "_resume_evolution_terminal_effects",
+        lambda _campaign_id, _task_id, value: dict(value),
+    )
+
+    first = evolution_lifecycle.update_evolution_campaign_after_task(
+        tx["task_id"],
+        cost_usd=1.0,
+        outcome_axes={"execution": {"status": "failed"}},
+        rounds=1,
+        transaction=tx,
+    )
+
+    assert first["persisted"] is True
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert stored["history"][0]["transaction"]["cleanup_status"] == "pending"
+    assert stored["pending_owner_report"]["cycle_outcome"] == "abandoned"
+
+    cleanup_calls = []
+    reports = []
+
+    def _cleanup(value, *_args, **_kwargs):
+        cleanup_calls.append(value["transaction_id"])
+        value["cleanup_status"] = "already_clean"
+
+    monkeypatch.setattr(evolution_lifecycle, "_resume_evolution_terminal_effects", real_resume)
+    monkeypatch.setattr(evolution_lifecycle, "_cleanup_worktree_after_cycle", _cleanup)
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "notify_owner_cycle_outcome",
+        lambda campaign, value: reports.append((campaign["id"], value["cycle_outcome"])),
+    )
+
+    replay = evolution_lifecycle.update_evolution_campaign_after_task(
+        tx["task_id"],
+        cost_usd=1.0,
+        outcome_axes={"execution": {"status": "failed"}},
+        rounds=1,
+        transaction=tx,
+    )
+
+    assert replay["replay"] is True
+    assert cleanup_calls == [tx["transaction_id"]]
+    assert reports == [(_campaign["id"], "abandoned")]
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert stored["history"][0]["transaction"]["cleanup_status"] == "already_clean"
+    assert "pending_owner_report" not in stored
+
+
+def test_duplicate_terminal_resumes_missing_restart_request(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle
+
+    _campaign, tx = _active_transaction(tmp_path)
+    receipt = evolution_lifecycle.record_evolution_commit(
+        campaign_id=tx["campaign_id"],
+        transaction_id=tx["transaction_id"],
+        task_id=tx["task_id"],
+        commit_sha="a" * 40,
+    )
+    assert receipt["ok"] is True
+    real_resume = evolution_lifecycle._resume_evolution_terminal_effects
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "_resume_evolution_terminal_effects",
+        lambda _campaign_id, _task_id, value: dict(value),
+    )
+
+    first = evolution_lifecycle.update_evolution_campaign_after_task(
+        tx["task_id"],
+        cost_usd=1.0,
+        outcome_axes={"execution": {"status": "ok"}},
+        rounds=1,
+        transaction=tx,
+    )
+
+    assert first["transaction"]["cycle_outcome"] == "waiting_for_restart"
+    restart_calls = []
+    monkeypatch.setattr(evolution_lifecycle, "_resume_evolution_terminal_effects", real_resume)
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "request_evolution_restart",
+        lambda drive_root, value, log=None: restart_calls.append(
+            (pathlib.Path(drive_root), value["commit_sha"])
+        ),
+    )
+
+    replay = evolution_lifecycle.update_evolution_campaign_after_task(
+        tx["task_id"],
+        cost_usd=1.0,
+        outcome_axes={"execution": {"status": "ok"}},
+        rounds=1,
+        transaction=tx,
+    )
+
+    assert replay["replay"] is True
+    assert restart_calls == [(tmp_path, "a" * 40)]
+
+
+def test_terminal_restart_preserves_exact_model_reason(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle, workers
+
+    campaign, tx = _active_transaction(tmp_path)
+    sha = "c" * 40
+    assert evolution_lifecycle.record_evolution_commit(
+        campaign["id"], tx["transaction_id"], tx["task_id"], sha,
+    )["ok"] is True
+    current_tx = evolution_lifecycle._read_evolution_campaign()["active_transaction"]
+    claim = {
+        "campaign_id": campaign["id"],
+        "transaction_id": tx["transaction_id"],
+        "task_id": tx["task_id"],
+        "commit_sha": sha,
+    }
+    marker = tmp_path / "state" / "pending_restart_verify.json"
+    marker.write_text(json.dumps({
+        "expected_sha": sha,
+        "reason": "apply reviewed evolution",
+        "evolution_claim": claim,
+    }))
+    events = _CaptureQueue()
+    monkeypatch.setenv("OUROBOROS_EVOLUTION_AUTO_RESTART", "true")
+    monkeypatch.setattr(workers, "get_event_q", lambda: events)
+
+    evolution_lifecycle.request_evolution_restart(tmp_path, current_tx)
+
+    assert json.loads(marker.read_text())["reason"] == "apply reviewed evolution"
+    assert len(events.items) == 1
+    assert events.items[0]["reason"] == "apply reviewed evolution"
+    assert events.items[0]["evolution_restart"] is True
+
+
+def test_terminal_write_serializes_concurrent_campaign_pause(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle
+
+    _campaign, tx = _active_transaction(tmp_path)
+    real_write = evolution_lifecycle._write_evolution_campaign
+    entered = threading.Event()
+    release = threading.Event()
+    terminal_result = {}
+    pause_result = {}
+
+    def _hold_terminal_write(data, **kwargs):
+        entered.set()
+        assert release.wait(timeout=2)
+        return real_write(data, **kwargs)
+
+    monkeypatch.setattr(evolution_lifecycle, "_write_evolution_campaign", _hold_terminal_write)
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "_cleanup_worktree_after_cycle",
+        lambda tx, *_a, **_k: tx.update(cleanup_status="already_clean"),
+    )
+
+    def _terminal():
+        terminal_result.update(evolution_lifecycle.update_evolution_campaign_after_task(
+            tx["task_id"],
+            cost_usd=1.0,
+            outcome_axes={"execution": {"status": "ok"}},
+            rounds=1,
+            transaction=tx,
+        ))
+
+    terminal_thread = threading.Thread(target=_terminal)
+    terminal_thread.start()
+    assert entered.wait(timeout=2)
+
+    def _pause():
+        pause_result.update(evolution_lifecycle.pause_evolution_campaign("concurrent pause"))
+
+    pause_thread = threading.Thread(target=_pause)
+    pause_thread.start()
+    pause_thread.join(timeout=0.05)
+    assert pause_thread.is_alive()
+    release.set()
+    terminal_thread.join(timeout=2)
+    pause_thread.join(timeout=2)
+
+    assert terminal_result["persisted"] is True
+    assert pause_result["status"] == "paused"
+    stored = evolution_lifecycle._read_evolution_campaign()
+    assert stored["status"] == "paused"
+    assert stored["pause_reason"] == "concurrent pause"
+    assert stored["history"][0]["task_id"] == tx["task_id"]
+
+
+def test_terminal_write_exception_has_no_lifecycle_side_effects(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle
+
+    _campaign, tx = _active_transaction(tmp_path)
+    side_effects = []
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "_write_evolution_campaign",
+        lambda *_a, **_k: (_ for _ in ()).throw(OSError("disk full")),
+    )
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "_cleanup_worktree_after_cycle",
+        lambda *_a, **_k: side_effects.append("cleanup"),
+    )
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "notify_owner_cycle_outcome",
+        lambda *_a, **_k: side_effects.append("notify"),
+    )
+
+    result = evolution_lifecycle.update_evolution_campaign_after_task(
+        tx["task_id"],
+        cost_usd=1.0,
+        outcome_axes={"execution": {"status": "ok"}},
+        rounds=1,
+        transaction=tx,
+    )
+
+    assert result["persisted"] is False
+    assert result["reason"] == "campaign_write_failed"
+    assert side_effects == []
+
+
+def test_rejected_terminal_does_not_consume_global_evolution_state(tmp_path, monkeypatch):
+    from supervisor import evolution_lifecycle, state
+    from supervisor.events import _handle_evolution_task_done
+
+    state.init(tmp_path)
+    state.update_state(lambda live: live.update(
+        evolution_mode_enabled=True,
+        post_task_autostop=True,
+        evolution_consecutive_failures=4,
+    ))
+    monkeypatch.setattr(
+        evolution_lifecycle,
+        "update_evolution_campaign_after_task",
+        lambda *_a, **_k: {
+            "accepted": True, "persisted": False, "replay": False,
+            "reason": "campaign_write_refused", "transaction": {},
+        },
+    )
+    checkpoints = []
+    monkeypatch.setattr(
+        "ouroboros.evolution_checkpoints.append_evolution_checkpoint",
+        lambda *_a, **_k: checkpoints.append(True),
+    )
+    ctx = SimpleNamespace(DRIVE_ROOT=tmp_path, REPO_DIR=tmp_path)
+    task = {"metadata": {"evolution_transaction": {"transaction_id": "stale"}}}
+
+    _handle_evolution_task_done(
+        ctx,
+        evt={},
+        task_id="stale",
+        task=task,
+        task_done_event={"status": "failed"},
+        outcome_axes={"execution": {"status": "failed"}},
+        cost=1.0,
+        rounds=1,
+    )
+
+    live = state.load_state()
+    assert live["evolution_mode_enabled"] is True
+    assert live["post_task_autostop"] is True
+    assert live["evolution_consecutive_failures"] == 4
+    assert checkpoints == []

tests/test_post_task_reflection.py

@@ -0,0 +1,147 @@
+"""Post-task reflection and backlog promotion in ``ouroboros.agent_task_pipeline``.
+
+Split out of ``tests/test_agent_task_pipeline.py`` when that module was divided
+by theme; every moved block is verbatim. Covers `_run_reflection` entry
+generation, `_update_improvement_backlog`, and the project-scoped channel
+split: project memory stays project-local while backlog promotion goes to the
+global drive through `_run_global_backlog_promotion_only`.
+"""
+
+import json
+from types import SimpleNamespace
+
+import ouroboros.agent_task_pipeline as pipeline
+
+
+def test_project_scoped_post_task_processing_feeds_global_backlog_but_project_memory(tmp_path, monkeypatch):
+    import ouroboros.post_task_evolution as post_task_evolution
+
+    calls = []
+    reflection = {"backlog_candidates": [{"summary": "tool friction"}], "memory_actions": [{"kind": "note"}]}
+    monkeypatch.setattr(pipeline, "_run_task_summary", lambda *args, **kwargs: calls.append(("summary",)))
+    monkeypatch.setattr(pipeline, "_run_reflection", lambda *args, **kwargs: reflection)
+    monkeypatch.setattr(pipeline, "_update_improvement_backlog", lambda _env, entry: calls.append(("backlog", entry)) or 1)
+    monkeypatch.setattr(
+        pipeline,
+        "_apply_reflection_memory_actions",
+        lambda _env, entry, project_id="": calls.append(("memory", project_id, entry)) or 1,
+    )
+    monkeypatch.setattr(post_task_evolution, "maybe_promote", lambda _env, task, entry, _llm: calls.append(("promote", task.get("project_id"), entry)))
+    env = SimpleNamespace(repo_dir=tmp_path, drive_root=tmp_path, drive_path=lambda rel: tmp_path / rel)
+
+    pipeline._run_post_task_processing_async(
+        env,
+        {"id": "task-1", "type": "task", "project_id": "proj-1", "text": "fix workspace"},
+        {"rounds": 3, "cost": 0.1},
+        {"tool_calls": [], "reasoning_notes": []},
+        {},
+        tmp_path / "logs",
+        blocking=True,
+    )
+
+    assert ("backlog", reflection) in calls
+    assert ("memory", "proj-1", reflection) in calls
+    assert ("promote", "proj-1", reflection) in calls
+
+
+def test_project_global_promotion_uses_real_maybe_promote_without_project_scope(tmp_path, monkeypatch):
+    import ouroboros.post_task_evolution as post_task_evolution
+
+    monkeypatch.setattr("ouroboros.config.get_post_task_evolution_enabled", lambda: True)
+    monkeypatch.setattr("ouroboros.config.get_runtime_mode", lambda: "pro")
+    monkeypatch.setattr("ouroboros.config.get_post_task_evolution_cadence", lambda: "every_n:1")
+    monkeypatch.setattr(
+        post_task_evolution,
+        "_decide_promotion",
+        lambda *_args, **_kwargs: {
+            "promote": True,
+            "objective": "Improve Ouroboros workspace tool feedback",
+            "requires_plan_review": True,
+            "backlog_id": "",
+        },
+    )
+    env = SimpleNamespace(drive_root=tmp_path, drive_path=lambda rel: tmp_path / rel)
+    reflection = {
+        "reflection": "Project-specific detail should not be forwarded.",
+        "memory_actions": [{"kind": "note"}],
+        "backlog_candidates": [{"summary": "Improve Ouroboros workspace tool feedback"}],
+    }
+
+    pipeline._run_global_backlog_promotion_only(
+        env,
+        {
+            "id": "task-project",
+            "project_id": "proj-1",
+            "workspace_root": "/tmp/project",
+            "workspace_mode": "external",
+            "metadata": {"workspace_preflight": {"git": {"head": "abc"}}},
+        },
+        reflection,
+        object(),
+    )
+
+    req = json.loads((tmp_path / "state" / "post_task_evolution_request.json").read_text(encoding="utf-8"))
+    assert req["objective"] == "Improve Ouroboros workspace tool feedback"
+    backlog = (tmp_path / "memory" / "knowledge" / "improvement-backlog.md").read_text(encoding="utf-8")
+    assert "Project-specific detail" not in backlog
+
+
+def test_update_improvement_backlog_appends_candidates(tmp_path):
+    env = SimpleNamespace(drive_root=tmp_path)
+
+    added = pipeline._update_improvement_backlog(
+        env,
+        {
+            "backlog_candidates": [{
+                "summary": "Reduce recurring task friction around REVIEW_BLOCKED",
+                "category": "process",
+                "source": "execution_reflection",
+                "task_id": "task-backlog",
+                "evidence": "REVIEW_BLOCKED",
+                "context": "The task retried blocked review loops without narrowing scope.",
+                "proposed_next_step": "Run plan_task before touching review prompts again.",
+            }],
+        },
+    )
+
+    assert added == 1
+    backlog_path = tmp_path / "memory" / "knowledge" / "improvement-backlog.md"
+    assert backlog_path.exists()
+    text = backlog_path.read_text(encoding="utf-8")
+    assert "Reduce recurring task friction around REVIEW_BLOCKED" in text
+
+
+def test_run_reflection_returns_entry_when_generated(tmp_path):
+    captured = {}
+
+    class FakeLlm:
+        def chat(self, *, messages, model, reasoning_effort, max_tokens):
+            captured["prompt"] = messages[0]["content"]
+            return {
+                "content": (
+                    "Reflection text.\n"
+                    "BACKLOG_CANDIDATES_JSON: "
+                    "[{\"summary\":\"Reduce recurring task friction around REVIEW_BLOCKED\"," 
+                    "\"category\":\"process\"," 
+                    "\"source\":\"execution_reflection\"," 
+                    "\"evidence\":\"REVIEW_BLOCKED\"}]"
+                )
+            }, {"cost": 0}
+
+    env = SimpleNamespace(drive_root=tmp_path)
+    (tmp_path / "logs").mkdir(parents=True)
+
+    entry = pipeline._run_reflection(
+        env,
+        FakeLlm(),
+        {"id": "task-reflect", "type": "task", "text": "Fix it"},
+        {"rounds": 2, "cost": 0.01},
+        {"tool_calls": [{"tool": "commit_reviewed", "is_error": False, "result": "⚠️ REVIEW_BLOCKED"}]},
+        {"recent_attempts": [], "open_obligations": [{"item": "tests_affected", "reason": "Fix the failing test before commit"}]},
+    )
+
+    assert entry is not None
+    assert entry["task_id"] == "task-reflect"
+    assert entry["reflection"] == "Reflection text."
+    assert len(entry["backlog_candidates"]) == 1
+    assert entry["backlog_candidates"][0]["summary"] == "Reduce recurring task friction around REVIEW_BLOCKED"

tests/test_root_post_task_synthesis.py

@@ -0,0 +1,557 @@
+"""The root post-task synthesis phase of ``ouroboros.agent_task_pipeline``.
+
+Split out of ``tests/test_agent_task_pipeline.py`` when that module was divided
+by theme; every moved block is verbatim. Covers the durable
+`root_phase_checkpoint` state machine and its exact-subtree cost
+reconciliation, startup recovery of pending/indeterminate synthesis, the
+shared pre-synthesis usage snapshot taken once before worker dispatch, and
+that snapshot reaching (or staying out of) the summary and reflection prompts.
+"""
+
+from types import SimpleNamespace
+
+import ouroboros.agent_task_pipeline as pipeline
+
+
+def test_root_phase_checkpoint_is_durable_and_completion_is_idempotent(tmp_path):
+    env = SimpleNamespace(drive_root=tmp_path, repo_dir=tmp_path)
+    task = {"id": "root-checkpoint", "root_task_id": "root-checkpoint", "type": "task"}
+    trace = {
+        "tool_calls": [],
+        "reasoning_notes": [],
+        "root_phase_checkpoint": {
+            "phase": "task_acceptance",
+            "status": "pass",
+            "pass_index": 1,
+            "post_task_synthesis": "pending_once",
+        },
+    }
+    pipeline._store_task_result(
+        env, task, "done", {"rounds": 1, "cost": 0.0}, trace,
+    )
+    stored = pipeline.load_task_result(tmp_path, "root-checkpoint")
+    assert stored["root_phase_checkpoint"]["post_task_synthesis"] == "pending_once"
+    pipeline._set_root_post_task_checkpoint(env, task, "completed")
+    assert pipeline._root_post_task_already_completed(env, task) is True
+
+    # A repeated result materialization must preserve the terminal phase marker.
+    pipeline._store_task_result(
+        env, task, "done again", {"rounds": 1, "cost": 0.0}, trace,
+    )
+    stored = pipeline.load_task_result(tmp_path, "root-checkpoint")
+    assert stored["root_phase_checkpoint"]["post_task_synthesis"] == "completed"
+
+    degraded_task = {"id": "root-degraded", "root_task_id": "root-degraded"}
+    pipeline.write_task_result(
+        tmp_path, "root-degraded", pipeline.STATUS_COMPLETED,
+        root_phase_checkpoint={"post_task_synthesis": "degraded"},
+    )
+    assert pipeline._root_post_task_already_completed(env, degraded_task) is True
+
+
+def test_root_checkpoint_reconciles_exact_subtree_and_late_namer_cost(tmp_path):
+    from ouroboros import usage_accounting as accounting
+
+    env = SimpleNamespace(drive_root=tmp_path, repo_dir=tmp_path)
+    task = {
+        "id": "root-cost", "root_task_id": "root-cost", "type": "task",
+        "budget_drive_root": str(tmp_path),
+    }
+    pipeline.write_task_result(
+        tmp_path, "root-cost", pipeline.STATUS_COMPLETED,
+        root_task_id="root-cost", cost_usd=99.0, cost_final=True,
+        root_phase_checkpoint={"post_task_synthesis": "running"},
+    )
+
+    def settle(task_id, cost):
+        reservation = accounting.reserve_attempt(accounting.AttemptRequest(
+            model="openai/gpt-5.2", provider="openai", reservation_usd=cost,
+            drive_root=tmp_path, task_id=task_id, root_task_id="root-cost",
+            global_limit_usd=10.0, root_limit_usd=10.0,
+        ))
+        accounting.mark_dispatched(reservation)
+        accounting.settle_attempt(reservation, {}, cost_usd=cost, cost_final=True)
+
+    settle("root-cost", 1.0)
+    settle("abnormal-child", 2.0)
+    pipeline._set_root_post_task_checkpoint(env, task, "completed")
+    stored = pipeline.load_task_result(tmp_path, "root-cost")
+    assert stored["cost_usd"] == 1.0
+    assert stored["cost_usd_with_children"] == 3.0
+    assert stored["cost_final"] is True
+    assert stored["cost_with_children_partial"] is False
+
+    settle("root-cost", 0.25)
+    pipeline._set_root_post_task_checkpoint(env, task, "refresh")
+    stored = pipeline.load_task_result(tmp_path, "root-cost")
+    assert stored["root_phase_checkpoint"]["post_task_synthesis"] == "completed"
+    assert stored["cost_usd"] == 1.25
+    assert stored["cost_usd_with_children"] == 3.25
+
+
+def test_retry_root_checkpoint_preserves_logical_subtree_cost(tmp_path):
+    from ouroboros import usage_accounting as accounting
+
+    env = SimpleNamespace(drive_root=tmp_path, repo_dir=tmp_path)
+    task = {
+        "id": "retry-2",
+        "root_task_id": "logical-root",
+        "parent_task_id": "",
+        "delegation_role": "root",
+        "original_task_id": "retry-1",
+        "timeout_retry_from": "retry-1",
+        "budget_drive_root": str(tmp_path),
+    }
+    assert pipeline._is_root_post_task(task) is True
+    assert pipeline._is_root_post_task({
+        **task,
+        "timeout_retry_from": "different-attempt",
+    }) is False
+    pipeline.write_task_result(
+        tmp_path,
+        "retry-2",
+        pipeline.STATUS_COMPLETED,
+        **{key: value for key, value in task.items() if key != "id"},
+        root_phase_checkpoint={"post_task_synthesis": "running"},
+    )
+
+    def settle(task_id, cost):
+        reservation = accounting.reserve_attempt(accounting.AttemptRequest(
+            model="openai/gpt-5.2",
+            provider="openai",
+            reservation_usd=cost,
+            drive_root=tmp_path,
+            task_id=task_id,
+            root_task_id="logical-root",
+            global_limit_usd=10.0,
+            root_limit_usd=10.0,
+        ))
+        accounting.mark_dispatched(reservation)
+        accounting.settle_attempt(
+            reservation, {}, cost_usd=cost, cost_final=True,
+        )
+
+    settle("logical-root", 1.25)
+    settle("retry-2", 0.75)
+    pipeline._set_root_post_task_checkpoint(env, task, "completed")
+
+    stored = pipeline.load_task_result(tmp_path, "retry-2")
+    assert stored["root_task_id"] == "logical-root"
+    assert stored["cost_usd"] == 0.75
+    assert stored["cost_usd_with_children"] == 2.0
+    assert stored["cost_final"] is True
+
+
+def test_startup_recovery_reuses_pending_root_result_checkpoint(tmp_path, monkeypatch):
+    pipeline.write_task_result(
+        tmp_path,
+        "recover-root",
+        pipeline.STATUS_COMPLETED,
+        root_task_id="recover-root",
+        objective="finish recovery",
+        total_rounds=3,
+        cost_usd=0.25,
+        root_phase_checkpoint={
+            "phase": "task_acceptance",
+            "status": "pass",
+            "post_task_synthesis": "pending_once",
+        },
+    )
+    calls = []
+
+    def fake_run(env, task, usage, trace, evidence, drive_logs, *, blocking=False,
+                 sealed_final=None):
+        calls.append((env.drive_root, task, usage, trace, evidence, drive_logs, blocking))
+        pipeline._set_root_post_task_checkpoint(env, task, "completed")
+
+    monkeypatch.setattr(pipeline, "_run_post_task_processing_async", fake_run)
+    assert pipeline.recover_pending_root_post_task_synthesis(tmp_path, tmp_path / "repo") == 1
+    assert calls[0][1]["id"] == "recover-root"
+    assert calls[0][2]["rounds"] == 3
+    assert calls[0][3]["recovered_post_task_synthesis"] is True
+    assert calls[0][-1] is False
+    assert pipeline.recover_pending_root_post_task_synthesis(tmp_path, tmp_path / "repo") == 0
+
+
+def test_startup_recovery_never_replays_indeterminate_paid_post_task_phase(tmp_path, monkeypatch):
+    pipeline.write_task_result(
+        tmp_path,
+        "crashed-root",
+        pipeline.STATUS_COMPLETED,
+        root_task_id="crashed-root",
+        root_phase_checkpoint={
+            "phase": "task_acceptance",
+            "status": "pass",
+            "post_task_synthesis": "running",
+        },
+    )
+    paid_replays = []
+    monkeypatch.setattr(
+        pipeline,
+        "_run_post_task_processing_async",
+        lambda *args, **kwargs: paid_replays.append((args, kwargs)),
+    )
+
+    assert pipeline.recover_pending_root_post_task_synthesis(tmp_path, tmp_path / "repo") == 1
+    assert paid_replays == []
+    stored = pipeline.load_task_result(tmp_path, "crashed-root")
+    checkpoint = stored["root_phase_checkpoint"]
+    assert checkpoint["post_task_synthesis"] == "degraded"
+    assert checkpoint["post_task_stop_reason"] == "restart_indeterminate_running"
+    assert pipeline.recover_pending_root_post_task_synthesis(tmp_path, tmp_path / "repo") == 0
+
+
+def test_periodic_orphan_reconcile_does_not_degrade_live_post_task_synthesis(tmp_path):
+    from ouroboros.task_status import reconcile_orphaned_running_tasks
+
+    pipeline.write_task_result(
+        tmp_path,
+        "live-synthesis",
+        pipeline.STATUS_COMPLETED,
+        root_task_id="live-synthesis",
+        root_phase_checkpoint={
+            "phase": "task_acceptance",
+            "status": "pass",
+            "post_task_synthesis": "running",
+        },
+    )
+
+    assert reconcile_orphaned_running_tasks(tmp_path) == 0
+    stored = pipeline.load_task_result(tmp_path, "live-synthesis")
+    assert stored["root_phase_checkpoint"]["post_task_synthesis"] == "running"
+
+
+def test_root_synthesis_uses_one_shared_nonfinal_subtree_cost_snapshot(tmp_path, monkeypatch):
+    import ouroboros.memory as memory_mod
+    import ouroboros.post_task_evolution as post_task_evolution
+    import ouroboros.usage_accounting as accounting
+    import ouroboros.llm as llm_mod
+
+    reads = []
+    order = []
+    snapshots = []
+
+    def fake_breakdown(root, *, root_task_id="", task_id=""):
+        order.append("snapshot")
+        reads.append((root, root_task_id, task_id))
+        return {
+            "accounted_usd": 4.75,
+            "reserved_usd": 1.5,
+            "unresolved_upper_bound_usd": 0.75,
+            "unknown_unmetered": 2,
+            "integrity_degraded": False,
+        }
+
+    monkeypatch.setattr(accounting, "usage_breakdown", fake_breakdown)
+    monkeypatch.setattr(llm_mod, "LLMClient", lambda: object())
+    monkeypatch.setattr(memory_mod, "Memory", lambda **_kwargs: object())
+    monkeypatch.setattr(
+        pipeline, "_run_chat_consolidation",
+        lambda *args, **kwargs: order.append("chat_consolidation"),
+    )
+    monkeypatch.setattr(
+        pipeline, "_run_scratchpad_consolidation",
+        lambda *args, **kwargs: order.append("scratchpad_consolidation"),
+    )
+    monkeypatch.setattr(
+        pipeline,
+        "_run_task_summary",
+        lambda _env, _llm, _task, usage, *_args, **_kwargs: (
+            order.append("summary"), snapshots.append(usage)
+        ),
+    )
+    monkeypatch.setattr(
+        pipeline,
+        "_run_reflection",
+        lambda _env, _llm, _task, usage, *_args, **_kwargs: (
+            order.append("reflection"), snapshots.append(usage)
+        ),
+    )
+    monkeypatch.setattr(pipeline, "_update_improvement_backlog", lambda *args, **kwargs: 0)
+    monkeypatch.setattr(pipeline, "_apply_reflection_memory_actions", lambda *args, **kwargs: 0)
+    monkeypatch.setattr(post_task_evolution, "maybe_promote", lambda *args, **kwargs: None)
+    monkeypatch.setattr(pipeline, "_set_root_post_task_checkpoint", lambda *args, **kwargs: None)
+
+    env = SimpleNamespace(
+        repo_dir=tmp_path,
+        drive_root=tmp_path,
+        drive_path=lambda rel: tmp_path / rel,
+    )
+    pipeline._run_post_task_processing_async(
+        env,
+        {
+            "id": "root-synthesis",
+            "root_task_id": "root-synthesis",
+            "budget_drive_root": str(tmp_path),
+        },
+        {"rounds": 8, "cost": 1.25},
+        {"tool_calls": [], "reasoning_notes": []},
+        {},
+        tmp_path / "logs",
+        blocking=True,
+    )
+
+    assert reads == [(tmp_path, "root-synthesis", "")]
+    assert order[:5] == [
+        "snapshot", "chat_consolidation", "scratchpad_consolidation",
+        "summary", "reflection",
+    ]
+    assert len(snapshots) == 2 and snapshots[0] is snapshots[1]
+    snapshot = snapshots[0]
+    assert snapshot["cost_usd_with_children"] == 4.75
+    assert snapshot["reserved_usd"] == 1.5
+    assert snapshot["unresolved_upper_bound_usd"] == 0.75
+    assert snapshot["unknown_unmetered"] == 2
+    assert snapshot["ledger_integrity"] == "ok"
+    assert snapshot["cost_final"] is False
+    assert snapshot["cost_with_children_partial"] is True
+
+
+def test_nonblocking_post_task_snapshot_precedes_worker_dispatch(tmp_path, monkeypatch):
+    import ouroboros.usage_accounting as accounting
+
+    order = []
+    worker_targets = []
+
+    monkeypatch.setattr(
+        accounting,
+        "usage_breakdown",
+        lambda *_args, **_kwargs: order.append("snapshot") or {
+            "accounted_usd": 1.0,
+            "reserved_usd": 0.0,
+            "unresolved_upper_bound_usd": 0.0,
+            "unknown_unmetered": 0,
+            "integrity_degraded": False,
+        },
+    )
+    monkeypatch.setattr(pipeline, "_set_root_post_task_checkpoint", lambda *args, **kwargs: None)
+
+    class DeferredThread:
+        def __init__(self, *, target, daemon):
+            assert order == ["snapshot"]
+            assert daemon is True
+            worker_targets.append(target)
+
+        def start(self):
+            order.append("thread_start")
+
+    monkeypatch.setattr(pipeline.threading, "Thread", DeferredThread)
+    env = SimpleNamespace(
+        repo_dir=tmp_path,
+        drive_root=tmp_path,
+        drive_path=lambda rel: tmp_path / rel,
+    )
+
+    pipeline._run_post_task_processing_async(
+        env,
+        {
+            "id": "async-root",
+            "root_task_id": "async-root",
+            "budget_drive_root": str(tmp_path),
+        },
+        {"cost": 0.5},
+        {},
+        {},
+        tmp_path / "logs",
+    )
+
+    assert order == ["snapshot", "thread_start"]
+    assert len(worker_targets) == 1
+    with pipeline._POST_TASK_SYNTHESIS_LOCK:
+        pipeline._POST_TASK_SYNTHESIS_INFLIGHT.discard(
+            (str(tmp_path.resolve(strict=False)), "async-root")
+        )
+
+
+def test_pre_synthesis_cost_failure_is_unavailable_not_zero(tmp_path, monkeypatch):
+    import ouroboros.usage_accounting as accounting
+
+    monkeypatch.setattr(
+        accounting,
+        "usage_breakdown",
+        lambda *args, **kwargs: (_ for _ in ()).throw(OSError("ledger unavailable")),
+    )
+    env = SimpleNamespace(drive_root=tmp_path)
+    snapshot = pipeline._pre_synthesis_usage_snapshot(
+        env,
+        {"id": "root", "root_task_id": "root", "budget_drive_root": str(tmp_path)},
+        {"rounds": 2, "cost": 1.0},
+    )
+
+    assert snapshot["cost_usd_with_children"] is None
+    assert snapshot["reserved_usd"] is None
+    assert snapshot["unresolved_upper_bound_usd"] is None
+    assert snapshot["unknown_unmetered"] is None
+    assert snapshot["ledger_integrity"] == "unavailable"
+    assert pipeline._synthesis_cost_text(snapshot) == "cost unavailable (non-final)"
+
+
+def _capture_summary_and_reflection_prompts(
+    tmp_path, monkeypatch, usage, *, task_overrides=None,
+):
+    import ouroboros.consolidator as consolidator
+
+    monkeypatch.setattr(
+        consolidator,
+        "_consolidation_route",
+        lambda: ("test/synthesis-model", False),
+    )
+
+    class CapturingLlm:
+        def __init__(self):
+            self.prompts = []
+
+        def chat(self, *, messages, **_kwargs):
+            self.prompts.append(messages[0]["content"])
+            return {"content": "captured synthesis"}, {}
+
+    drive_logs = tmp_path / "logs"
+    drive_logs.mkdir(parents=True, exist_ok=True)
+    task = {
+        "id": "root-synthesis-prompt",
+        "root_task_id": "root-synthesis-prompt",
+        "type": "task",
+        "text": "Inspect the shared cost snapshot",
+        "drive_root": str(tmp_path),
+    }
+    task.update(task_overrides or {})
+    trace = {
+        "tool_calls": [{
+            "tool": "run_command",
+            "status": "error",
+            "is_error": True,
+            "result": "TOOL_ERROR: synthetic prompt-capture trigger",
+        }],
+        "reasoning_notes": [],
+    }
+
+    summary_llm = CapturingLlm()
+    pipeline._run_task_summary(
+        env=None,
+        llm=summary_llm,
+        task=task,
+        usage=usage,
+        llm_trace=trace,
+        drive_logs=drive_logs,
+    )
+
+    reflection_llm = CapturingLlm()
+    entry = pipeline._run_reflection(
+        SimpleNamespace(drive_root=tmp_path),
+        reflection_llm,
+        task,
+        usage,
+        trace,
+        {},
+    )
+
+    assert entry is not None
+    assert len(summary_llm.prompts) == 1
+    assert len(reflection_llm.prompts) == 1
+    return summary_llm.prompts[0], reflection_llm.prompts[0]
+
+
+def test_shared_cost_snapshot_reaches_summary_and_reflection_prompts(tmp_path, monkeypatch):
+    snapshot = {
+        "rounds": 8,
+        "cost": 1.25,
+        "cost_usd_with_children": 4.75,
+        "reserved_usd": 1.5,
+        "unresolved_upper_bound_usd": 0.75,
+        "unknown_unmetered": 2,
+        "ledger_integrity": "ok",
+        "cost_snapshot_at": "2026-07-15T12:34:56+00:00",
+        "cost_final": False,
+        "cost_with_children_partial": True,
+        "cost_accounting_status": "available",
+        "reason_code": "child_results_deferred",
+        "outcome_axes": {
+            "execution": {"status": "degraded"},
+            "objective": {"status": "best_effort"},
+            "review": {"status": "degraded"},
+        },
+    }
+
+    prompts = _capture_summary_and_reflection_prompts(
+        tmp_path, monkeypatch, snapshot,
+    )
+    snapshot_text = pipeline._synthesis_usage_snapshot_text(snapshot)
+    expected_fragments = (
+        '"cost_usd_with_children": 4.75',
+        '"reserved_usd": 1.5',
+        '"unresolved_upper_bound_usd": 0.75',
+        '"unknown_unmetered": 2',
+        '"ledger_integrity": "ok"',
+        '"cost_snapshot_at": "2026-07-15T12:34:56+00:00"',
+        '"cost_final": false',
+        '"cost_with_children_partial": true',
+        '"cost_accounting_status": "available"',
+        '"reason_code": "child_results_deferred"',
+        '"status": "best_effort"',
+    )
+    for prompt in prompts:
+        assert snapshot_text in prompt
+        assert "accounted subtree cost only" in prompt
+        assert "separate non-final exposure fields" in prompt
+        assert "including the reserved" not in prompt
+        assert "outcome_axes` is canonical task truth" in prompt
+        assert '"review": {' in prompt
+        for fragment in expected_fragments:
+            assert fragment in prompt
+
+
+def test_unavailable_cost_snapshot_is_null_not_zero_in_both_prompts(tmp_path, monkeypatch):
+    snapshot = {
+        "rounds": 8,
+        "cost": 1.25,
+        "cost_usd_with_children": None,
+        "reserved_usd": None,
+        "unresolved_upper_bound_usd": None,
+        "unknown_unmetered": None,
+        "ledger_integrity": "unavailable",
+        "cost_snapshot_at": "2026-07-15T12:35:00+00:00",
+        "cost_final": False,
+        "cost_with_children_partial": True,
+        "cost_accounting_status": "unavailable",
+    }
+
+    prompts = _capture_summary_and_reflection_prompts(
+        tmp_path, monkeypatch, snapshot,
+    )
+    snapshot_text = pipeline._synthesis_usage_snapshot_text(snapshot)
+    null_fields = (
+        "cost_usd_with_children",
+        "reserved_usd",
+        "unresolved_upper_bound_usd",
+        "unknown_unmetered",
+    )
+    for prompt in prompts:
+        assert snapshot_text in prompt
+        for field in null_fields:
+            assert f'"{field}": null' in prompt
+        assert '"ledger_integrity": "unavailable"' in prompt
+        assert '"cost_snapshot_at": "2026-07-15T12:35:00+00:00"' in prompt
+        assert '"cost_final": false' in prompt
+        assert '"cost_with_children_partial": true' in prompt
+        assert '"cost_accounting_status": "unavailable"' in prompt
+        assert "$0" not in prompt
+
+
+def test_child_legacy_usage_does_not_claim_a_subtree_snapshot(tmp_path, monkeypatch):
+    prompts = _capture_summary_and_reflection_prompts(
+        tmp_path,
+        monkeypatch,
+        {"rounds": 8, "cost": 1.25},
+        task_overrides={
+            "id": "child-synthesis-prompt",
+            "root_task_id": "root-synthesis-prompt",
+            "parent_task_id": "root-synthesis-prompt",
+            "delegation_role": "subagent",
+        },
+    )
+
+    for prompt in prompts:
+        assert "Shared pre-synthesis cost snapshot" not in prompt
+        assert "cost_usd_with_children" not in prompt
+        assert "cost_snapshot_at" not in prompt
+    assert "Cost: $1.25" in prompts[0]