From e63f79d0cd647bea568be7d042312c2638eeb24b Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 22:55:17 +0530
Subject: [PATCH 1/7] feat(control-plane): auto-archive completed todos, cap
 sqlite markdown archive projection, and scope stale_source_registry checks

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../goals/global_registry_health.py           |  65 ++-
 .../todos/machine_section_projection.py       |  63 ++-
 .../todos/provider_projection.py              |   6 +
 .../todos/provider_terminal_lifecycle.py      |  62 ++-
 ...test_sqlite_archive_and_registry_health.py | 404 ++++++++++++++++++
 5 files changed, 595 insertions(+), 5 deletions(-)
 create mode 100644 tests/control_plane/test_sqlite_archive_and_registry_health.py

diff --git a/loopx/control_plane/goals/global_registry_health.py b/loopx/control_plane/goals/global_registry_health.py
index eaa2990121..ab80dfd693 100644
--- a/loopx/control_plane/goals/global_registry_health.py
+++ b/loopx/control_plane/goals/global_registry_health.py
@@ -31,6 +31,58 @@ def global_registry_finding(
     return finding
 
 
+def _multi_goal_source_entry_unchanged(
+    goal: dict[str, Any],
+    *,
+    goal_id: str,
+    source_path: Path,
+    source_goals_cache: dict[str, list[dict[str, Any]] | None],
+    load_registry: Callable[[Path], dict[str, Any]],
+    registry_goals: Callable[[dict[str, Any]], list[dict[str, Any]]],
+) -> bool:
+    cache_key = str(source_path)
+    if cache_key not in source_goals_cache:
+        try:
+            source_goals_cache[cache_key] = registry_goals(load_registry(source_path))
+        except Exception:  # noqa: BLE001
+            source_goals_cache[cache_key] = None
+    source_goals = source_goals_cache[cache_key]
+    if not source_goals or len(source_goals) <= 1:
+        return False
+    candidates = [
+        row for row in source_goals if str(row.get("id") or "") == goal_id
+    ]
+    if len(candidates) != 1:
+        return False
+    source_goal = candidates[0]
+    if source_goal.get("source_registry"):
+        return False
+    try:
+        from ...global_registry import (
+            preserve_attention_override,
+            sanitize_goal_for_global,
+        )
+
+        resolved_source = source_path.expanduser().resolve()
+        expected = preserve_attention_override(
+            {},
+            sanitize_goal_for_global(
+                source_goal, source_registry=resolved_source, synced_at=""
+            ),
+        )
+        observed = preserve_attention_override(
+            {},
+            {
+                **goal,
+                "source_registry": str(resolved_source),
+                "synced_at": "",
+            },
+        )
+        return observed == expected
+    except Exception:  # noqa: BLE001
+        return False
+
+
 def collect_global_registry_health(
     *,
     registry_path: Path,
@@ -104,6 +156,7 @@ def collect_global_registry_health(
     global_ids = [str(goal.get("id")) for goal in global_goals if goal.get("id")]
     global_id_set = set(global_ids)
     source_registries: set[str] = set()
+    source_goals_cache: dict[str, list[dict[str, Any]] | None] = {}
     findings: list[dict[str, Any]] = []
     checks: list[str] = list(current_checks)
 
@@ -161,7 +214,17 @@ def collect_global_registry_health(
                 synced_at = parse_timestamp(goal.get("synced_at"))
                 if synced_at:
                     source_mtime = datetime.fromtimestamp(source_path.stat().st_mtime, tz=timezone.utc)
-                    if source_mtime > synced_at.astimezone(timezone.utc) + timedelta(seconds=5):
+                    if (
+                        source_mtime > synced_at.astimezone(timezone.utc) + timedelta(seconds=5)
+                        and not _multi_goal_source_entry_unchanged(
+                            goal,
+                            goal_id=goal_id,
+                            source_path=source_path,
+                            source_goals_cache=source_goals_cache,
+                            load_registry=load_registry,
+                            registry_goals=registry_goals,
+                        )
+                    ):
                         findings.append(
                             global_registry_finding(
                                 kind="stale_source_registry",
diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py
index a5cce107f4..fbb57c703e 100644
--- a/loopx/control_plane/todos/machine_section_projection.py
+++ b/loopx/control_plane/todos/machine_section_projection.py
@@ -56,6 +56,7 @@
 
 
 TODO_SECTION_PROJECTION_SCHEMA_VERSION = "loopx_todo_section_projection_v0"
+DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT = 20
 _MARKER_PATTERN = re.compile(
     r"(?m)^<!-- loopx:todo-section-projection-v0 "
     r"role=(?P<role>user|agent|archive) "
@@ -407,12 +408,62 @@ def _parity_records(records: Sequence[Mapping[str, object]]) -> list[dict[str, o
     ]
 
 
+def _active_referenced_todo_ids(
+    active_records: Sequence[Mapping[str, object]],
+) -> set[str]:
+    referenced: set[str] = set()
+    for record in active_records:
+        raw_resume = record.get("resume_when")
+        if not isinstance(raw_resume, str):
+            continue
+        for clause in re.split(r"[,;]", raw_resume):
+            token = clause.strip()
+            if token.startswith("todo_done:"):
+                target_id = token.partition(":")[2].strip()
+                if target_id:
+                    referenced.add(target_id)
+    return referenced
+
+
+def _select_projected_archive_records(
+    archived: list[dict[str, object]],
+    *,
+    active_records: Sequence[Mapping[str, object]],
+    max_projected_archive_records: int | None,
+) -> list[dict[str, object]]:
+    if max_projected_archive_records is None:
+        return archived
+    if isinstance(max_projected_archive_records, bool) or not isinstance(
+        max_projected_archive_records, int
+    ) or max_projected_archive_records < 0:
+        raise TodoSectionProjectionError(
+            "max_projected_archive_records must be a non-negative integer or None"
+        )
+    if len(archived) <= max_projected_archive_records:
+        return archived
+    referenced_ids = _active_referenced_todo_ids(active_records)
+    tail_records = (
+        archived[-max_projected_archive_records:]
+        if max_projected_archive_records > 0
+        else []
+    )
+    retained_ids = referenced_ids | {
+        str(record.get("todo_id") or "") for record in tail_records
+    }
+    return [
+        record
+        for record in archived
+        if str(record.get("todo_id") or "") in retained_ids
+    ]
+
+
 def render_canonical_todo_sections(
     markdown: str,
     records: Sequence[Mapping[str, object]],
     *,
     provider_revision: str,
     private_validation_declarations: Mapping[str, Mapping[str, Any]] | None = None,
+    max_projected_archive_records: int | None = None,
 ) -> TodoSectionProjectionResult:
     """Replace only Todo sections and verify deterministic parse/render parity."""
 
@@ -483,9 +534,14 @@ def render_canonical_todo_sections(
         )
         for role in TODO_SECTION_HEADINGS
     }
-    archived = sorted(
-        [record for record in canonical if record.get("archive_state") == "archive"],
-        key=_record_sort_key,
+    all_active = [*by_role["user"], *by_role["agent"]]
+    archived = _select_projected_archive_records(
+        sorted(
+            [record for record in canonical if record.get("archive_state") == "archive"],
+            key=_record_sort_key,
+        ),
+        active_records=all_active,
+        max_projected_archive_records=max_projected_archive_records,
     )
     newline = "\r\n" if "\r\n" in markdown else "\n"
     rendered_sections: dict[str, str] = {}
@@ -567,6 +623,7 @@ def inspect_todo_section_projection(markdown: str) -> dict[str, object]:
 
 
 __all__ = [
+    "DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT",
     "TODO_SECTION_PROJECTION_SCHEMA_VERSION",
     "TodoSectionProjectionError",
     "TodoSectionProjectionResult",
diff --git a/loopx/control_plane/todos/provider_projection.py b/loopx/control_plane/todos/provider_projection.py
index 48b1ff7875..624ed4bc92 100644
--- a/loopx/control_plane/todos/provider_projection.py
+++ b/loopx/control_plane/todos/provider_projection.py
@@ -24,6 +24,7 @@
     local_authority_is_promoted,
 )
 from .machine_section_projection import (
+    DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT,
     TodoSectionProjectionError,
     render_canonical_todo_sections,
 )
@@ -141,6 +142,11 @@ def project_current_canonical_todos(
                     goal_id=goal_id,
                     todos=authority_read["todos"],
                 ),
+                max_projected_archive_records=(
+                    DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT
+                    if authority_read.get("source_authority") == "sqlite_v0"
+                    else None
+                ),
             )
             if execute and projection.changed:
                 # The projection is a primary-state write: it must respect the
diff --git a/loopx/control_plane/todos/provider_terminal_lifecycle.py b/loopx/control_plane/todos/provider_terminal_lifecycle.py
index 1b8c7ff7a9..096793bda8 100644
--- a/loopx/control_plane/todos/provider_terminal_lifecycle.py
+++ b/loopx/control_plane/todos/provider_terminal_lifecycle.py
@@ -27,6 +27,10 @@
     effect_runtime_result,
 )
 from .completion_policy import build_completion_policy_request
+from .completed_archive import (
+    ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE,
+    DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE,
+)
 from .completion_transaction import require_completion_successor_todo_ids
 from .completion_validation import (
     resolve_private_completion_validation_declaration,
@@ -517,7 +521,7 @@ def terminal_canonical_todo_if_promoted(
         "mutation_authority": terminal_decision,
         "task_lease_fence": terminal_decision,
     }
-    return _projection_payload(
+    settled = _projection_payload(
         settle_canonical_todo_projection(
             response,
             registry_path=registry_path,
@@ -527,6 +531,62 @@ def terminal_canonical_todo_if_promoted(
             state_file=state_file,
         )
     )
+    target_role = str(target.get("role") or role or "agent")
+    if (
+        not dry_run
+        and command == "complete"
+        and payload.get("changed") is True
+        and target_role in {"user", "agent"}
+    ):
+        active_done_before = sum(
+            1
+            for item in canonical.get("todos", [])
+            if isinstance(item, Mapping)
+            and str(item.get("role") or "agent") == target_role
+            and item.get("archive_state") != "archive"
+            and (
+                item.get("status") == "done"
+                or (item.get("done") is True and item.get("status") != "deferred")
+            )
+        )
+        active_done_after = active_done_before + (
+            0
+            if target.get("status") == "done"
+            or (target.get("done") is True and target.get("status") != "deferred")
+            else 1
+        )
+        if active_done_after > DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE:
+            try:
+                archived = archive_canonical_todos_if_promoted(
+                    registry_path=registry_path,
+                    runtime_root=runtime_root,
+                    goal_id=goal_id,
+                    role=target_role,
+                    max_active_done=ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE,
+                    dry_run=False,
+                    project=project,
+                    state_file=state_file,
+                )
+                if isinstance(archived, Mapping):
+                    settled["auto_archive"] = {
+                        "status": archived.get("status"),
+                        "role": target_role,
+                        "moved_count": archived.get("moved_count", 0),
+                        "moved_todo_ids": archived.get("moved_todo_ids", []),
+                        "active_done_before": archived.get("active_done_before"),
+                        "active_done_after": archived.get("active_done_after"),
+                        "max_active_done": ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE,
+                        "provider_revision": archived.get("provider_revision"),
+                    }
+            except Exception as error:  # noqa: BLE001 - completion already committed
+                settled["auto_archive"] = {
+                    "status": "pending",
+                    "role": target_role,
+                    "reason_code": "auto_archive_unavailable",
+                    "error_class": error.__class__.__name__,
+                    "retryable": True,
+                }
+    return settled
 
 
 def archive_canonical_todos_if_promoted(
diff --git a/tests/control_plane/test_sqlite_archive_and_registry_health.py b/tests/control_plane/test_sqlite_archive_and_registry_health.py
new file mode 100644
index 0000000000..6afae7c426
--- /dev/null
+++ b/tests/control_plane/test_sqlite_archive_and_registry_health.py
@@ -0,0 +1,404 @@
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+
+import pytest
+
+from canonical_authority_fixture import (
+    initialize_canonical_authority,
+    isolate_sqlite_runtime,
+)
+from loopx import status as status_module
+from loopx.control_plane.coordination.local_authority import (
+    read_canonical_todos_if_promoted,
+)
+from loopx.control_plane.coordination.runtime_shadow import (
+    build_todo_runtime_shadow_projection,
+)
+from loopx.control_plane.todos import provider_projection
+from loopx.control_plane.todos.active_state_todo_parser import parse_todo_source
+from loopx.control_plane.todos.completed_archive import (
+    ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE,
+    DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE,
+)
+from loopx.control_plane.todos.machine_section_projection import (
+    DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT,
+    render_canonical_todo_sections,
+)
+from loopx.control_plane.todos.provider_terminal_lifecycle import (
+    terminal_canonical_todo_if_promoted,
+)
+from loopx.global_registry import sanitize_goal_for_global
+from loopx.paths import global_registry_path
+
+
+def test_sqlite_projection_caps_markdown_archive_and_preserves_active_dependencies() -> None:
+    source = (
+        "# Goal\n\n"
+        "Human narrative stays intact.\n\n"
+        "## User Todo / Owner Review Reading Queue\n\n"
+        "## Agent Todo\n\n"
+        "## Completed Work Archive\n\n"
+        "## Next Action\n\n"
+        "- Continue.\n"
+    )
+    records: list[dict[str, object]] = [
+        {
+            "schema_version": "todo_domain_record_v0",
+            "todo_id": "todo_waiting_on_oldest",
+            "role": "agent",
+            "status": "open",
+            "done": False,
+            "text": "Follow up after oldest archived prerequisite.",
+            "archive_state": "active",
+            "task_class": "advancement_task",
+            "resume_when": "todo_done:todo_arch_000",
+        }
+    ]
+    for index in range(30):
+        records.append(
+            {
+                "schema_version": "todo_domain_record_v0",
+                "todo_id": f"todo_arch_{index:03d}",
+                "role": "agent",
+                "status": "done",
+                "done": True,
+                "text": f"Archived work item {index}.",
+                "archive_state": "archive",
+                "task_class": "advancement_task",
+            }
+        )
+
+    projected = render_canonical_todo_sections(
+        source,
+        records,
+        provider_revision="sqlite:cap:1",
+        max_projected_archive_records=DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT,
+    )
+    assert projected.changed is True
+    assert projected.todo_count == 31
+    _, archived_rows, _ = parse_todo_source(projected.markdown)
+    archived_ids = [row["todo_id"] for row in archived_rows]
+    # 20 tail archived items + 1 dependency-referenced oldest item (todo_arch_000)
+    assert len(archived_ids) == DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT + 1
+    assert "todo_arch_000" in archived_ids
+    assert "todo_arch_001" not in archived_ids
+    assert "todo_arch_029" in archived_ids
+
+    replay = render_canonical_todo_sections(
+        projected.markdown,
+        records,
+        provider_revision="sqlite:cap:1",
+        max_projected_archive_records=DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT,
+    )
+    assert replay.changed is False
+    assert replay.markdown == projected.markdown
+
+
+def test_provider_projection_uses_archive_cap_only_for_sqlite_v0(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    runtime_root = tmp_path / "runtime"
+    project = tmp_path / "project"
+    state_file = project / ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md"
+    state_file.parent.mkdir(parents=True)
+    state_file.write_text(
+        "# Goal\n\n## User Todo / Owner Review Reading Queue\n\n"
+        "## Agent Todo\n\n## Completed Work Archive\n",
+        encoding="utf-8",
+    )
+    registry = tmp_path / "registry.json"
+    registry.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "common_runtime_root": str(runtime_root),
+                "goals": [
+                    {
+                        "id": "goal-a",
+                        "status": "active",
+                        "repo": str(project),
+                        "state_file": ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md",
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    todos = [
+        {
+            "schema_version": "todo_domain_record_v0",
+            "todo_id": f"todo_arch_{index:03d}",
+            "role": "agent",
+            "status": "done",
+            "done": True,
+            "text": f"Archived item {index}.",
+            "archive_state": "archive",
+            "task_class": "advancement_task",
+        }
+        for index in range(25)
+    ]
+
+    def fake_read(**kwargs):
+        result = {
+            "status": "loaded",
+            "source_authority": "sqlite_v0",
+            "provider_revision": "sqlite:25:abc",
+            "todos": todos,
+        }
+        if "projection_readback" in kwargs:
+            witness = kwargs["projection_readback"]
+            result["projection_readback"] = {
+                "provider_revision": witness["provider_revision"],
+                "observed_provider_revision": result["provider_revision"],
+                "status": "delivered" if witness["changed"] else "current",
+                "next_action": "finish",
+            }
+        return result
+
+    monkeypatch.setattr(
+        provider_projection, "read_canonical_todos_if_promoted", fake_read
+    )
+    delivered = provider_projection.project_current_canonical_todos(
+        registry_path=registry,
+        runtime_root=runtime_root,
+        goal_id="goal-a",
+    )
+    assert delivered["status"] == "delivered"
+    assert delivered["todo_count"] == 25
+    _, archived_rows, _ = parse_todo_source(state_file.read_text(encoding="utf-8"))
+    assert len(archived_rows) == DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT
+
+
+def test_completing_13th_todo_in_sqlite_auto_archives_to_10_active_done(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    runtime_root = tmp_path / "runtime"
+    project = tmp_path / "project"
+    state_file = project / ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md"
+    state_file.parent.mkdir(parents=True)
+    state_file.write_text(
+        "# Goal\n\n## User Todo / Owner Review Reading Queue\n\n"
+        "## Agent Todo\n\n## Completed Work Archive\n",
+        encoding="utf-8",
+    )
+    registry_path = tmp_path / "registry.json"
+    registry_path.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "common_runtime_root": str(runtime_root),
+                "goals": [
+                    {
+                        "id": "goal-a",
+                        "status": "active",
+                        "repo": str(project),
+                        "state_file": ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md",
+                        "coordination": {
+                            "registered_agents": ["agent-a"],
+                        },
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    initial_todos: list[dict[str, object]] = []
+    for index in range(DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE):
+        initial_todos.append(
+            {
+                "schema_version": "todo_item_v0",
+                "todo_id": f"todo_done_{index:02d}",
+                "role": "agent",
+                "status": "done",
+                "done": True,
+                "text": f"Completed step {index}.",
+                "archive_state": "active",
+                "source_section": "Agent Todo",
+                "index": index + 1,
+                "task_class": "advancement_task",
+            }
+        )
+    initial_todos.extend(
+        [
+            {
+                "schema_version": "todo_item_v0",
+                "todo_id": "todo_13th",
+                "role": "agent",
+                "status": "open",
+                "done": False,
+                "text": "Complete the 13th step.",
+                "archive_state": "active",
+                "source_section": "Agent Todo",
+                "index": 13,
+                "task_class": "advancement_task",
+                "claimed_by": "agent-a",
+            },
+            {
+                "schema_version": "todo_item_v0",
+                "todo_id": "todo_successor",
+                "role": "agent",
+                "status": "open",
+                "done": False,
+                "text": "Continue next bounded slice.",
+                "archive_state": "active",
+                "source_section": "Agent Todo",
+                "index": 14,
+                "task_class": "advancement_task",
+                "claimed_by": "agent-a",
+            },
+        ]
+    )
+    projection = build_todo_runtime_shadow_projection(
+        goal_id="goal-a", todos=initial_todos, handoff_mode="soft_claim"
+    )
+    initialize_canonical_authority(
+        runtime_root, "goal-a", projection, state_path=state_file, provider="sqlite"
+    )
+
+    result = terminal_canonical_todo_if_promoted(
+        registry_path=registry_path,
+        runtime_root=runtime_root,
+        goal_id="goal-a",
+        command="complete",
+        todo_id="todo_13th",
+        role="agent",
+        actor_agent_id="agent-a",
+        authority_reason=None,
+        decision_outcome=None,
+        evidence="verified 13th step",
+        completion_result_file=None,
+        note="completed",
+        reason=None,
+        completion_turn_key="turn-13",
+        completion_identity_source=None,
+        completion_delivery_workspace=None,
+        completion_validation_workspace_path=None,
+        claim_operation_id=None,
+        task_lease_idempotency_key=None,
+        task_lease_expected_version=None,
+        no_followup=False,
+        successor_todo_ids=["todo_successor"],
+        claimed_by="agent-a",
+        clear_claim=False,
+        next_agent_todo=None,
+        next_user_todo=None,
+        next_user_task_class=None,
+        next_claimed_by=None,
+        next_task_class=None,
+        next_action_kind=None,
+        next_task_repository=None,
+        next_required_capabilities=None,
+        next_continuation_policy=None,
+        next_excluded_agents=None,
+        self_merged=False,
+        dry_run=False,
+        project=project,
+        state_file=state_file,
+    )
+    assert result is not None
+    assert result["ok"] is True
+    assert "auto_archive" in result
+    assert result["auto_archive"]["moved_count"] == 3
+    assert (
+        result["auto_archive"]["active_done_after"]
+        == ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE
+    )
+
+    after_read = read_canonical_todos_if_promoted(
+        runtime_root=runtime_root, goal_id="goal-a"
+    )
+    assert after_read is not None
+    all_done = [
+        item for item in after_read["todos"] if item.get("status") == "done"
+    ]
+    active_done = [
+        item
+        for item in all_done
+        if item.get("archive_state") != "archive"
+    ]
+    archived_done = [
+        item
+        for item in all_done
+        if item.get("archive_state") == "archive"
+    ]
+    assert len(all_done) == 13
+    assert len(active_done) == ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE
+    assert len(archived_done) == 3
+
+
+def test_multi_goal_source_registry_only_flags_goals_whose_entry_drifted(
+    tmp_path: Path,
+) -> None:
+    runtime = tmp_path / "runtime"
+    project = tmp_path / "project"
+    source_registry_path = project / ".loopx" / "registry.json"
+    source_registry_path.parent.mkdir(parents=True)
+
+    goals_in_source = []
+    for goal_id in ("goal-synced", "goal-untouched", "goal-drifted"):
+        state_rel = f".codex/goals/{goal_id}/ACTIVE_GOAL_STATE.md"
+        state_path = project / state_rel
+        state_path.parent.mkdir(parents=True)
+        state_path.write_text("# Goal\n", encoding="utf-8")
+        goals_in_source.append(
+            {
+                "id": goal_id,
+                "status": "active",
+                "repo": str(project),
+                "state_file": state_rel,
+            }
+        )
+
+    old_synced_at = "2026-01-01T00:00:00+00:00"
+    new_synced_at = "2026-11-01T12:00:00+00:00"
+    global_goals = [
+        sanitize_goal_for_global(
+            goals_in_source[0],
+            source_registry=source_registry_path,
+            synced_at=new_synced_at,
+        ),
+        sanitize_goal_for_global(
+            goals_in_source[1],
+            source_registry=source_registry_path,
+            synced_at=old_synced_at,
+        ),
+        sanitize_goal_for_global(
+            goals_in_source[2],
+            source_registry=source_registry_path,
+            synced_at=old_synced_at,
+        ),
+    ]
+    # Mutate goal-drifted in source_registry without syncing it to global
+    goals_in_source[2] = {**goals_in_source[2], "status": "paused"}
+    source_registry = {"schema_version": 1, "goals": goals_in_source}
+    source_registry_path.write_text(
+        json.dumps(source_registry, indent=2) + "\n", encoding="utf-8"
+    )
+    now = time.time()
+    os.utime(source_registry_path, (now, now))
+
+    global_path = global_registry_path(runtime)
+    global_path.parent.mkdir(parents=True, exist_ok=True)
+    global_path.write_text(
+        json.dumps({"schema_version": 1, "goals": global_goals}, indent=2) + "\n",
+        encoding="utf-8",
+    )
+
+    health = status_module.collect_global_registry_health(
+        registry_path=source_registry_path,
+        runtime_root=runtime,
+        current_registry=source_registry,
+    )
+    stale_goal_ids = [
+        finding.get("goal_id")
+        for finding in health["findings"]
+        if finding.get("kind") == "stale_source_registry"
+    ]
+    assert stale_goal_ids == ["goal-drifted"]

From c9c6510362c7574920fd8088febf34f3988a487a Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 23:06:01 +0530
Subject: [PATCH 2/7] feat(control-plane): bounded active view projection,
 in-transaction auto-archive, and goal-scoped stale_source_registry

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../coordination/todo_terminal_lifecycle.ts   |  57 +++++-
 loopx/control_plane/todos/machine_region.py   |   1 +
 .../todos/machine_section_projection.py       |  96 ++++++++--
 .../todos/provider_projection.py              |   6 +-
 .../todos/provider_terminal_lifecycle.py      |  62 +------
 loopx/global_registry.py                      |  36 +++-
 ...test_sqlite_archive_and_registry_health.py | 175 ++++++++++++------
 7 files changed, 287 insertions(+), 146 deletions(-)

diff --git a/loopx/control_plane/coordination/todo_terminal_lifecycle.ts b/loopx/control_plane/coordination/todo_terminal_lifecycle.ts
index 8f57471c64..c8f3efd9d2 100644
--- a/loopx/control_plane/coordination/todo_terminal_lifecycle.ts
+++ b/loopx/control_plane/coordination/todo_terminal_lifecycle.ts
@@ -79,11 +79,14 @@ import {
 import { BARE_SHA256_PATTERN } from "../content_digest.ts";
 import {todoExecutionDependencyRejection} from "./todo_execution_dependency.ts";
 import {isLifecycleCloseout, prepareTerminalReentry, requiresReentrySettlement} from "./todo_terminal_reentry.ts";
+import {selectCoordinationTodoArchive} from "./todo_archive_selection.ts";
 
 export const COORDINATION_TODO_TERMINAL_LIFECYCLE_RESULT_SCHEMA =
   "loopx_coordination_todo_terminal_lifecycle_result_v0";
 export const COORDINATION_TODO_TERMINAL_LIFECYCLE_RECEIPT_SCHEMA =
   "loopx_coordination_todo_terminal_lifecycle_receipt_v0";
+export const DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE = 12;
+export const ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE = 10;
 const TERMINAL_COMMANDS = ["complete", "supersede"] as const;
 const TODO_ROLES = ["agent", "user"] as const;
 const DECISION_OUTCOMES = ["approve", "reject", "cancel"] as const;
@@ -1929,6 +1932,44 @@ export async function executeCoordinationTodoTerminalLifecycle(
     ? projection.todos.get(String(todo.unblocks_todo_id)) : undefined;
   const changed = authority.outcome === "apply" || edit?.changed === true ||
     terminalUpgradeReceipt !== null;
+  const targetRole = (todo.role === "user" ? "user" : "agent") as TodoRole;
+  const postTerminalTodos = changed && target.todo.status === "done"
+    ? projection.todo_ids.map((todoId) => {
+        if (todoId === input.todo_id) return target.todo;
+        if (dependent && followthrough && todoId === String(todo.unblocks_todo_id)) {
+          return {...dependent, ...followthrough.updates, updated_at: target.todo.updated_at};
+        }
+        return projection.todos.get(todoId)!;
+      })
+    : [];
+  const activeDoneCount = postTerminalTodos.filter(
+    (item) => item.role === targetRole && item.archive_state === "active" && item.status === "done",
+  ).length;
+  const autoArchiveSelection = activeDoneCount > DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE
+    ? selectCoordinationTodoArchive({
+        role: targetRole,
+        max_active_done: ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE,
+        todos: postTerminalTodos,
+      })
+    : null;
+  const autoArchivedIds = new Set(autoArchiveSelection?.moved_todo_ids ?? []);
+  const finalTargetTodo = autoArchivedIds.has(input.todo_id)
+    ? {...target.todo, archive_state: "archive"}
+    : target.todo;
+  const autoArchiveMutations: CoordinationProjectionMutation[] = autoArchiveSelection === null
+    ? []
+    : autoArchiveSelection.moved_todo_ids
+        .filter((todoId) => todoId !== input.todo_id)
+        .map((todoId) => ({
+          kind: "todo_upsert" as const,
+          todo: {
+            ...(dependent && followthrough && todoId === String(todo.unblocks_todo_id)
+              ? {...dependent, ...followthrough.updates}
+              : projection.todos.get(todoId)!),
+            archive_state: "archive",
+            updated_at: target.todo.updated_at,
+          },
+        }));
   const result: JsonObject = {
     todo_id: input.todo_id,
     command: input.command,
@@ -1959,6 +2000,17 @@ export async function executeCoordinationTodoTerminalLifecycle(
     ...(completionResult === null ? {} : {completion_result: completionResult}),
     ...(acceptanceEvidence === null ? {} : {goal_acceptance_completion: acceptanceEvidence}),
     ...(capabilityCompletionEvidence === null ? {} : {capability_completion_evidence: capabilityCompletionEvidence}),
+    ...(autoArchiveSelection !== null && autoArchiveSelection.moved_count > 0 ? {
+      auto_archive: {
+        role: autoArchiveSelection.role,
+        active_done_before: autoArchiveSelection.active_done_before,
+        active_done_after: autoArchiveSelection.active_done_after,
+        max_active_done: autoArchiveSelection.max_active_done,
+        moved_count: autoArchiveSelection.moved_count,
+        moved_todo_ids: autoArchiveSelection.moved_todo_ids,
+        retained_standing_decision_count: autoArchiveSelection.retained_standing_decision_count,
+      },
+    } : {}),
     // A preview that omits this would show an unconditional close for work the
     // real call still gates. Name the criteria the real call must run; never
     // their argv, which stays out of every projection.
@@ -1969,9 +2021,10 @@ export async function executeCoordinationTodoTerminalLifecycle(
       : {}),
   };
   const mutations: CoordinationProjectionMutation[] = changed ? [
-    {kind: "todo_upsert", todo: target.todo, clear_fields: target.clear_fields},
-    ...(dependent && followthrough ? [{kind: "todo_upsert" as const,
+    {kind: "todo_upsert", todo: finalTargetTodo, clear_fields: target.clear_fields},
+    ...(dependent && followthrough && !autoArchivedIds.has(String(todo.unblocks_todo_id)) ? [{kind: "todo_upsert" as const,
       todo: {...dependent, ...followthrough.updates, updated_at: target.todo.updated_at}}] : []),
+    ...autoArchiveMutations,
     ...successorCandidates.map((successor) => ({kind: "todo_upsert" as const, todo: successor})),
     ...(released === null ? [] : [{kind: "lease_upsert" as const, lease: released}]),
   ] : [];
diff --git a/loopx/control_plane/todos/machine_region.py b/loopx/control_plane/todos/machine_region.py
index 75b50c99dc..f9647b8ba1 100644
--- a/loopx/control_plane/todos/machine_region.py
+++ b/loopx/control_plane/todos/machine_region.py
@@ -110,6 +110,7 @@ def find_todo_regions(lines: list[str], *, visible: frozenset[int] | None = None
                 or task
                 or parse_todo_metadata_line(content) is not None
                 or content.startswith("<!-- loopx:todo-section-projection-v0 ")
+                or content.startswith("<!-- loopx:todo-archive-pointer ")
                 or legacy_continuation
             )
             if not generated:
diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py
index fbb57c703e..03da98ad28 100644
--- a/loopx/control_plane/todos/machine_section_projection.py
+++ b/loopx/control_plane/todos/machine_section_projection.py
@@ -57,6 +57,15 @@
 
 TODO_SECTION_PROJECTION_SCHEMA_VERSION = "loopx_todo_section_projection_v0"
 DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT = 20
+SQLITE_ARCHIVE_POINTER_COMMENT = (
+    '<!-- loopx:todo-archive-pointer authority=sqlite_v0 '
+    'query="loopx todo list --include-archived" -->'
+)
+BOUNDED_ACTIVE_VIEW_TEXT_LIMIT = 160
+BOUNDED_ACTIVE_VIEW_MAX_SCOPE_CHARS = 120
+BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS = frozenset(
+    {"note", "evidence", "reason", "completion_result"}
+)
 _MARKER_PATTERN = re.compile(
     r"(?m)^<!-- loopx:todo-section-projection-v0 "
     r"role=(?P<role>user|agent|archive) "
@@ -238,6 +247,7 @@ def _render_section(
     provider_revision: str,
     newline: str,
     private_validation: Mapping[str, Mapping[str, object]],
+    archive_pointer_comment: str | None = None,
 ) -> tuple[str, str]:
     digest = _sha256_text(_canonical_json(records))
     heading = (
@@ -250,6 +260,7 @@ def _render_section(
         todo_region_marker(role, "begin"),
         f"<!-- loopx:todo-section-projection-v0 role={role} "
         f"provider_revision={provider_revision} records_sha256={digest} -->",
+        *([archive_pointer_comment] if archive_pointer_comment else []),
         "",
     ]
     for record in records:
@@ -457,6 +468,36 @@ def _select_projected_archive_records(
     ]
 
 
+def _compact_record_for_bounded_active_view(
+    record: Mapping[str, object],
+) -> dict[str, object]:
+    compacted: dict[str, object] = {
+        key: value
+        for key, value in record.items()
+        if key not in BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS
+    }
+    scopes = compacted.get("required_write_scopes")
+    if (
+        isinstance(scopes, list)
+        and sum(len(str(item)) for item in scopes) > BOUNDED_ACTIVE_VIEW_MAX_SCOPE_CHARS
+    ):
+        compacted.pop("required_write_scopes", None)
+    raw_text = " ".join(str(compacted.get("text") or "").strip().split())
+    if len(raw_text) > BOUNDED_ACTIVE_VIEW_TEXT_LIMIT:
+        truncated_text = (
+            raw_text[: BOUNDED_ACTIVE_VIEW_TEXT_LIMIT - 3].rstrip() + "..."
+        )
+        compacted["text"] = truncated_text
+        priority, title = todo_priority_parts(truncated_text)
+        if priority:
+            compacted["priority"] = priority
+            compacted["title"] = normalize_todo_text(title, limit=None)
+        else:
+            compacted.pop("priority", None)
+            compacted.pop("title", None)
+    return compacted
+
+
 def render_canonical_todo_sections(
     markdown: str,
     records: Sequence[Mapping[str, object]],
@@ -464,6 +505,7 @@ def render_canonical_todo_sections(
     provider_revision: str,
     private_validation_declarations: Mapping[str, Mapping[str, Any]] | None = None,
     max_projected_archive_records: int | None = None,
+    bounded_active_view: bool = False,
 ) -> TodoSectionProjectionResult:
     """Replace only Todo sections and verify deterministic parse/render parity."""
 
@@ -523,25 +565,40 @@ def render_canonical_todo_sections(
             )
         private_validation[todo_id] = source[1]
     by_role = {
-        role: sorted(
-            [
-                record
-                for record in canonical
-                if record.get("role") == role
-                and record.get("archive_state") == "active"
-            ],
-            key=_record_sort_key,
-        )
+        role: [
+            (
+                _compact_record_for_bounded_active_view(record)
+                if bounded_active_view
+                else record
+            )
+            for record in sorted(
+                [
+                    record
+                    for record in canonical
+                    if record.get("role") == role
+                    and record.get("archive_state") == "active"
+                ],
+                key=_record_sort_key,
+            )
+        ]
         for role in TODO_SECTION_HEADINGS
     }
     all_active = [*by_role["user"], *by_role["agent"]]
-    archived = _select_projected_archive_records(
-        sorted(
-            [record for record in canonical if record.get("archive_state") == "archive"],
-            key=_record_sort_key,
-        ),
-        active_records=all_active,
-        max_projected_archive_records=max_projected_archive_records,
+    archived = (
+        []
+        if bounded_active_view
+        else _select_projected_archive_records(
+            sorted(
+                [
+                    record
+                    for record in canonical
+                    if record.get("archive_state") == "archive"
+                ],
+                key=_record_sort_key,
+            ),
+            active_records=all_active,
+            max_projected_archive_records=max_projected_archive_records,
+        )
     )
     newline = "\r\n" if "\r\n" in markdown else "\n"
     rendered_sections: dict[str, str] = {}
@@ -561,6 +618,9 @@ def render_canonical_todo_sections(
             provider_revision=provider_revision,
             newline=newline,
             private_validation=private_validation,
+            archive_pointer_comment=(
+                SQLITE_ARCHIVE_POINTER_COMMENT if bounded_active_view else None
+            ),
         )
 
     try:
@@ -623,7 +683,11 @@ def inspect_todo_section_projection(markdown: str) -> dict[str, object]:
 
 
 __all__ = [
+    "BOUNDED_ACTIVE_VIEW_MAX_SCOPE_CHARS",
+    "BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS",
+    "BOUNDED_ACTIVE_VIEW_TEXT_LIMIT",
     "DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT",
+    "SQLITE_ARCHIVE_POINTER_COMMENT",
     "TODO_SECTION_PROJECTION_SCHEMA_VERSION",
     "TodoSectionProjectionError",
     "TodoSectionProjectionResult",
diff --git a/loopx/control_plane/todos/provider_projection.py b/loopx/control_plane/todos/provider_projection.py
index 624ed4bc92..1a47be6c1b 100644
--- a/loopx/control_plane/todos/provider_projection.py
+++ b/loopx/control_plane/todos/provider_projection.py
@@ -142,10 +142,8 @@ def project_current_canonical_todos(
                     goal_id=goal_id,
                     todos=authority_read["todos"],
                 ),
-                max_projected_archive_records=(
-                    DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT
-                    if authority_read.get("source_authority") == "sqlite_v0"
-                    else None
+                bounded_active_view=(
+                    authority_read.get("source_authority") == "sqlite_v0"
                 ),
             )
             if execute and projection.changed:
diff --git a/loopx/control_plane/todos/provider_terminal_lifecycle.py b/loopx/control_plane/todos/provider_terminal_lifecycle.py
index 096793bda8..1b8c7ff7a9 100644
--- a/loopx/control_plane/todos/provider_terminal_lifecycle.py
+++ b/loopx/control_plane/todos/provider_terminal_lifecycle.py
@@ -27,10 +27,6 @@
     effect_runtime_result,
 )
 from .completion_policy import build_completion_policy_request
-from .completed_archive import (
-    ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE,
-    DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE,
-)
 from .completion_transaction import require_completion_successor_todo_ids
 from .completion_validation import (
     resolve_private_completion_validation_declaration,
@@ -521,7 +517,7 @@ def terminal_canonical_todo_if_promoted(
         "mutation_authority": terminal_decision,
         "task_lease_fence": terminal_decision,
     }
-    settled = _projection_payload(
+    return _projection_payload(
         settle_canonical_todo_projection(
             response,
             registry_path=registry_path,
@@ -531,62 +527,6 @@ def terminal_canonical_todo_if_promoted(
             state_file=state_file,
         )
     )
-    target_role = str(target.get("role") or role or "agent")
-    if (
-        not dry_run
-        and command == "complete"
-        and payload.get("changed") is True
-        and target_role in {"user", "agent"}
-    ):
-        active_done_before = sum(
-            1
-            for item in canonical.get("todos", [])
-            if isinstance(item, Mapping)
-            and str(item.get("role") or "agent") == target_role
-            and item.get("archive_state") != "archive"
-            and (
-                item.get("status") == "done"
-                or (item.get("done") is True and item.get("status") != "deferred")
-            )
-        )
-        active_done_after = active_done_before + (
-            0
-            if target.get("status") == "done"
-            or (target.get("done") is True and target.get("status") != "deferred")
-            else 1
-        )
-        if active_done_after > DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE:
-            try:
-                archived = archive_canonical_todos_if_promoted(
-                    registry_path=registry_path,
-                    runtime_root=runtime_root,
-                    goal_id=goal_id,
-                    role=target_role,
-                    max_active_done=ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE,
-                    dry_run=False,
-                    project=project,
-                    state_file=state_file,
-                )
-                if isinstance(archived, Mapping):
-                    settled["auto_archive"] = {
-                        "status": archived.get("status"),
-                        "role": target_role,
-                        "moved_count": archived.get("moved_count", 0),
-                        "moved_todo_ids": archived.get("moved_todo_ids", []),
-                        "active_done_before": archived.get("active_done_before"),
-                        "active_done_after": archived.get("active_done_after"),
-                        "max_active_done": ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE,
-                        "provider_revision": archived.get("provider_revision"),
-                    }
-            except Exception as error:  # noqa: BLE001 - completion already committed
-                settled["auto_archive"] = {
-                    "status": "pending",
-                    "role": target_role,
-                    "reason_code": "auto_archive_unavailable",
-                    "error_class": error.__class__.__name__,
-                    "retryable": True,
-                }
-    return settled
 
 
 def archive_canonical_todos_if_promoted(
diff --git a/loopx/global_registry.py b/loopx/global_registry.py
index 72a5260f8c..26ef21643c 100644
--- a/loopx/global_registry.py
+++ b/loopx/global_registry.py
@@ -278,9 +278,15 @@ def merge_goal_entries(
     incoming: list[dict[str, Any]],
     *,
     allow_route_replacement: bool = False,
+    unchanged_siblings: list[dict[str, Any]] | None = None,
 ) -> tuple[list[Any], list[str], list[str], list[dict[str, Any]]]:
     merged: list[Any] = []
     seen_incoming = {str(goal.get("id")) for goal in incoming if goal.get("id")}
+    sibling_by_id = {
+        str(goal.get("id")): goal
+        for goal in (unchanged_siblings or [])
+        if isinstance(goal, dict) and goal.get("id")
+    }
     actions: list[str] = []
     synced_ids: list[str] = []
     collisions: list[dict[str, Any]] = []
@@ -294,6 +300,15 @@ def merge_goal_entries(
     for item in existing:
         if isinstance(item, dict) and str(item.get("id")) in seen_incoming:
             continue
+        if isinstance(item, dict):
+            sibling = sibling_by_id.get(str(item.get("id") or ""))
+            if (
+                sibling is not None
+                and same_source_registry(item, sibling)
+                and preserve_attention_override({}, {**item, "synced_at": ""})
+                == preserve_attention_override(item, {**sibling, "synced_at": ""})
+            ):
+                item = {**item, "synced_at": sibling["synced_at"]}
         merged.append(item)
 
     for goal in incoming:
@@ -366,6 +381,7 @@ def _merge_global_registry_payload(
     schema_version_fallback: str,
     runtime_root: Path,
     synced_at: str,
+    unchanged_siblings: list[dict[str, Any]] | None = None,
 ) -> tuple[dict[str, Any], list[Any], list[str], list[str], list[dict[str, Any]]]:
     """Reduce one global-registry snapshot with sanitized incoming goals."""
 
@@ -376,6 +392,7 @@ def _merge_global_registry_payload(
         existing_goals,
         incoming,
         allow_route_replacement=allow_route_replacement,
+        unchanged_siblings=unchanged_siblings,
     )
     existing_projects = existing.get("projects")
     if existing_projects is None:
@@ -405,6 +422,7 @@ def _sync_global_registry_reduction(
     schema_version_fallback: str,
     runtime_root: Path,
     synced_at: str,
+    unchanged_siblings: list[dict[str, Any]] | None = None,
 ) -> GlobalRegistryReduction:
     payload, merged_goals, actions, synced_ids, collisions = (
         _merge_global_registry_payload(
@@ -415,6 +433,7 @@ def _sync_global_registry_reduction(
             schema_version_fallback=schema_version_fallback,
             runtime_root=runtime_root,
             synced_at=synced_at,
+            unchanged_siblings=unchanged_siblings,
         )
     )
     return GlobalRegistryReduction(
@@ -686,9 +705,10 @@ def _sync_project_registry_to_global_once(
             "actions": [],
         }
 
-    goals = registry_goals(project_registry)
+    all_source_goals = registry_goals(project_registry)
+    goals = all_source_goals
     if goal_id:
-        goals = [goal for goal in goals if str(goal.get("id")) == goal_id]
+        goals = [goal for goal in all_source_goals if str(goal.get("id")) == goal_id]
     if goal_id and not goals:
         raise ValueError(f"goal id not found in source registry: {goal_id}")
 
@@ -699,6 +719,17 @@ def _sync_project_registry_to_global_once(
         )
         for goal in goals
     ]
+    unchanged_siblings = (
+        [
+            sanitize_goal_for_global(
+                sibling, source_registry=registry_path, synced_at=synced_at
+            )
+            for sibling in all_source_goals
+            if str(sibling.get("id")) != goal_id
+        ]
+        if goal_id
+        else []
+    )
     source_projects = project_registry.get("projects")
     if source_projects is None:
         source_projects = []
@@ -724,6 +755,7 @@ def _sync_project_registry_to_global_once(
         "schema_version_fallback": str(project_registry.get("schema_version") or "0.1"),
         "runtime_root": runtime_root,
         "synced_at": synced_at,
+        "unchanged_siblings": unchanged_siblings,
     }
     preview = _sync_global_registry_reduction(
         _load_global_registry(global_path),
diff --git a/tests/control_plane/test_sqlite_archive_and_registry_health.py b/tests/control_plane/test_sqlite_archive_and_registry_health.py
index 6afae7c426..963d6a7f24 100644
--- a/tests/control_plane/test_sqlite_archive_and_registry_health.py
+++ b/tests/control_plane/test_sqlite_archive_and_registry_health.py
@@ -25,37 +25,46 @@
     DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE,
 )
 from loopx.control_plane.todos.machine_section_projection import (
-    DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT,
+    BOUNDED_ACTIVE_VIEW_TEXT_LIMIT,
+    SQLITE_ARCHIVE_POINTER_COMMENT,
     render_canonical_todo_sections,
 )
 from loopx.control_plane.todos.provider_terminal_lifecycle import (
     terminal_canonical_todo_if_promoted,
 )
-from loopx.global_registry import sanitize_goal_for_global
+from loopx.global_registry import (
+    sanitize_goal_for_global,
+    sync_project_registry_to_global,
+)
 from loopx.paths import global_registry_path
 
 
-def test_sqlite_projection_caps_markdown_archive_and_preserves_active_dependencies() -> None:
+def test_sqlite_bounded_active_view_projection_omits_archive_and_large_inline_fields() -> None:
     source = (
         "# Goal\n\n"
         "Human narrative stays intact.\n\n"
         "## User Todo / Owner Review Reading Queue\n\n"
         "## Agent Todo\n\n"
         "## Completed Work Archive\n\n"
+        "- [x] Legacy archived item that should be cleared.\n\n"
         "## Next Action\n\n"
         "- Continue.\n"
     )
+    long_title = "[P0] " + ("Very long active task description segment " * 8).strip()
     records: list[dict[str, object]] = [
         {
             "schema_version": "todo_domain_record_v0",
-            "todo_id": "todo_waiting_on_oldest",
+            "todo_id": "todo_active_compact",
             "role": "agent",
             "status": "open",
             "done": False,
-            "text": "Follow up after oldest archived prerequisite.",
+            "text": long_title,
             "archive_state": "active",
             "task_class": "advancement_task",
-            "resume_when": "todo_done:todo_arch_000",
+            "note": "x" * 12000,
+            "evidence": "y" * 8000,
+            "reason": "z" * 4000,
+            "required_write_scopes": [f"loopx/module_{idx}/**/*.py" for idx in range(20)],
         }
     ]
     for index in range(30):
@@ -69,36 +78,40 @@ def test_sqlite_projection_caps_markdown_archive_and_preserves_active_dependenci
                 "text": f"Archived work item {index}.",
                 "archive_state": "archive",
                 "task_class": "advancement_task",
+                "note": "archived-note-" * 500,
             }
         )
 
     projected = render_canonical_todo_sections(
         source,
         records,
-        provider_revision="sqlite:cap:1",
-        max_projected_archive_records=DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT,
+        provider_revision="sqlite:bounded:1",
+        bounded_active_view=True,
     )
     assert projected.changed is True
     assert projected.todo_count == 31
-    _, archived_rows, _ = parse_todo_source(projected.markdown)
-    archived_ids = [row["todo_id"] for row in archived_rows]
-    # 20 tail archived items + 1 dependency-referenced oldest item (todo_arch_000)
-    assert len(archived_ids) == DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT + 1
-    assert "todo_arch_000" in archived_ids
-    assert "todo_arch_001" not in archived_ids
-    assert "todo_arch_029" in archived_ids
+    assert SQLITE_ARCHIVE_POINTER_COMMENT in projected.markdown
+    assert "note=" not in projected.markdown
+    assert "evidence=" not in projected.markdown
+    assert "reason=" not in projected.markdown
+    assert "required_write_scopes=" not in projected.markdown
+    assert "Archived work item" not in projected.markdown
+    active_by_role, archived_rows, _ = parse_todo_source(projected.markdown)
+    assert archived_rows == []
+    assert len(active_by_role["agent"]) == 1
+    assert len(active_by_role["agent"][0]["text"]) <= BOUNDED_ACTIVE_VIEW_TEXT_LIMIT
 
     replay = render_canonical_todo_sections(
         projected.markdown,
         records,
-        provider_revision="sqlite:cap:1",
-        max_projected_archive_records=DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT,
+        provider_revision="sqlite:bounded:1",
+        bounded_active_view=True,
     )
     assert replay.changed is False
     assert replay.markdown == projected.markdown
 
 
-def test_provider_projection_uses_archive_cap_only_for_sqlite_v0(
+def test_provider_projection_uses_bounded_active_view_for_sqlite_v0(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
 ) -> None:
@@ -132,22 +145,36 @@ def test_provider_projection_uses_archive_cap_only_for_sqlite_v0(
     todos = [
         {
             "schema_version": "todo_domain_record_v0",
-            "todo_id": f"todo_arch_{index:03d}",
+            "todo_id": "todo_open_1",
             "role": "agent",
-            "status": "done",
-            "done": True,
-            "text": f"Archived item {index}.",
-            "archive_state": "archive",
+            "status": "open",
+            "done": False,
+            "text": "Active task in SQLite.",
+            "archive_state": "active",
             "task_class": "advancement_task",
-        }
-        for index in range(25)
+            "note": "large-note-" * 400,
+            "evidence": "large-evidence-" * 400,
+        },
+        *[
+            {
+                "schema_version": "todo_domain_record_v0",
+                "todo_id": f"todo_arch_{index:03d}",
+                "role": "agent",
+                "status": "done",
+                "done": True,
+                "text": f"Archived item {index}.",
+                "archive_state": "archive",
+                "task_class": "advancement_task",
+            }
+            for index in range(25)
+        ],
     ]
 
     def fake_read(**kwargs):
         result = {
             "status": "loaded",
             "source_authority": "sqlite_v0",
-            "provider_revision": "sqlite:25:abc",
+            "provider_revision": "sqlite:26:abc",
             "todos": todos,
         }
         if "projection_readback" in kwargs:
@@ -169,14 +196,20 @@ def fake_read(**kwargs):
         goal_id="goal-a",
     )
     assert delivered["status"] == "delivered"
-    assert delivered["todo_count"] == 25
-    _, archived_rows, _ = parse_todo_source(state_file.read_text(encoding="utf-8"))
-    assert len(archived_rows) == DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT
+    assert delivered["todo_count"] == 26
+    rendered = state_file.read_text(encoding="utf-8")
+    assert SQLITE_ARCHIVE_POINTER_COMMENT in rendered
+    assert "note=" not in rendered
+    assert "evidence=" not in rendered
+    _, archived_rows, _ = parse_todo_source(rendered)
+    assert archived_rows == []
 
 
-def test_completing_13th_todo_in_sqlite_auto_archives_to_10_active_done(
+@pytest.mark.parametrize("command", ["complete", "supersede"])
+def test_terminal_13th_todo_in_sqlite_auto_archives_in_same_transaction(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
+    command: str,
 ) -> None:
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     runtime_root = tmp_path / "runtime"
@@ -233,7 +266,7 @@ def test_completing_13th_todo_in_sqlite_auto_archives_to_10_active_done(
                 "role": "agent",
                 "status": "open",
                 "done": False,
-                "text": "Complete the 13th step.",
+                "text": "Complete or supersede the 13th step.",
                 "archive_state": "active",
                 "source_section": "Agent Todo",
                 "index": 13,
@@ -258,25 +291,26 @@ def test_completing_13th_todo_in_sqlite_auto_archives_to_10_active_done(
     projection = build_todo_runtime_shadow_projection(
         goal_id="goal-a", todos=initial_todos, handoff_mode="soft_claim"
     )
-    initialize_canonical_authority(
+    init_result = initialize_canonical_authority(
         runtime_root, "goal-a", projection, state_path=state_file, provider="sqlite"
     )
+    assert init_result["cursor"] == "1"
 
     result = terminal_canonical_todo_if_promoted(
         registry_path=registry_path,
         runtime_root=runtime_root,
         goal_id="goal-a",
-        command="complete",
+        command=command,
         todo_id="todo_13th",
         role="agent",
         actor_agent_id="agent-a",
         authority_reason=None,
         decision_outcome=None,
-        evidence="verified 13th step",
+        evidence="verified 13th step" if command == "complete" else None,
         completion_result_file=None,
-        note="completed",
-        reason=None,
-        completion_turn_key="turn-13",
+        note="completed" if command == "complete" else "superseded",
+        reason=None if command == "complete" else "replaced by todo_successor",
+        completion_turn_key="turn-13" if command == "complete" else None,
         completion_identity_source=None,
         completion_delivery_workspace=None,
         completion_validation_workspace_path=None,
@@ -285,7 +319,7 @@ def test_completing_13th_todo_in_sqlite_auto_archives_to_10_active_done(
         task_lease_expected_version=None,
         no_followup=False,
         successor_todo_ids=["todo_successor"],
-        claimed_by="agent-a",
+        claimed_by="agent-a" if command == "complete" else None,
         clear_claim=False,
         next_agent_todo=None,
         next_user_todo=None,
@@ -304,6 +338,8 @@ def test_completing_13th_todo_in_sqlite_auto_archives_to_10_active_done(
     )
     assert result is not None
     assert result["ok"] is True
+    # Single atomic transaction: cursor advances from 1 -> 2 (not 3!)
+    assert result["cursor"] == "2"
     assert "auto_archive" in result
     assert result["auto_archive"]["moved_count"] == 3
     assert (
@@ -315,6 +351,7 @@ def test_completing_13th_todo_in_sqlite_auto_archives_to_10_active_done(
         runtime_root=runtime_root, goal_id="goal-a"
     )
     assert after_read is not None
+    assert after_read["cursor"] == "2"
     all_done = [
         item for item in after_read["todos"] if item.get("status") == "done"
     ]
@@ -333,7 +370,7 @@ def test_completing_13th_todo_in_sqlite_auto_archives_to_10_active_done(
     assert len(archived_done) == 3
 
 
-def test_multi_goal_source_registry_only_flags_goals_whose_entry_drifted(
+def test_multi_goal_source_registry_sync_and_health_prevent_sibling_false_positives(
     tmp_path: Path,
 ) -> None:
     runtime = tmp_path / "runtime"
@@ -357,39 +394,55 @@ def test_multi_goal_source_registry_only_flags_goals_whose_entry_drifted(
         )
 
     old_synced_at = "2026-01-01T00:00:00+00:00"
-    new_synced_at = "2026-11-01T12:00:00+00:00"
     global_goals = [
         sanitize_goal_for_global(
-            goals_in_source[0],
-            source_registry=source_registry_path,
-            synced_at=new_synced_at,
-        ),
-        sanitize_goal_for_global(
-            goals_in_source[1],
-            source_registry=source_registry_path,
-            synced_at=old_synced_at,
-        ),
-        sanitize_goal_for_global(
-            goals_in_source[2],
+            goal,
             source_registry=source_registry_path,
             synced_at=old_synced_at,
-        ),
+        )
+        for goal in goals_in_source
     ]
-    # Mutate goal-drifted in source_registry without syncing it to global
+    global_path = global_registry_path(runtime)
+    global_path.parent.mkdir(parents=True, exist_ok=True)
+    global_path.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "common_runtime_root": str(runtime),
+                "goals": global_goals,
+            },
+            indent=2,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+
+    # Modify goal-synced and goal-drifted in source_registry, then sync ONLY goal-synced
+    goals_in_source[0] = {**goals_in_source[0], "domain": "updated-domain"}
     goals_in_source[2] = {**goals_in_source[2], "status": "paused"}
-    source_registry = {"schema_version": 1, "goals": goals_in_source}
+    source_registry = {
+        "schema_version": 1,
+        "common_runtime_root": str(runtime),
+        "goals": goals_in_source,
+    }
     source_registry_path.write_text(
         json.dumps(source_registry, indent=2) + "\n", encoding="utf-8"
     )
-    now = time.time()
+    sync_receipt = sync_project_registry_to_global(
+        registry_path=source_registry_path,
+        runtime_root_override=str(runtime),
+        goal_id="goal-synced",
+    )
+    assert sync_receipt["synced_goal_ids"] == ["goal-synced"]
+    now = time.time() + 10
     os.utime(source_registry_path, (now, now))
 
-    global_path = global_registry_path(runtime)
-    global_path.parent.mkdir(parents=True, exist_ok=True)
-    global_path.write_text(
-        json.dumps({"schema_version": 1, "goals": global_goals}, indent=2) + "\n",
-        encoding="utf-8",
-    )
+    # Verify that goal-untouched had its synced_at refreshed during the single-goal sync
+    # AND that even with future mtime on source_registry_path, only goal-drifted is flagged!
+    global_after = json.loads(global_path.read_text(encoding="utf-8"))
+    by_id = {g["id"]: g for g in global_after["goals"]}
+    assert by_id["goal-untouched"]["synced_at"] == sync_receipt["updated_at"]
+    assert by_id["goal-drifted"]["synced_at"] == old_synced_at
 
     health = status_module.collect_global_registry_health(
         registry_path=source_registry_path,

From 08218febd07de150024711a8ea1b8c6c2da8e4d9 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 23:23:11 +0530
Subject: [PATCH 3/7] control-plane: cap bounded active-view done items to 5
 and emit archive stats

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../todos/machine_section_projection.py       | 137 ++++++++++++++----
 ...test_sqlite_archive_and_registry_health.py |  37 ++++-
 2 files changed, 140 insertions(+), 34 deletions(-)

diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py
index 03da98ad28..aee5e6ae7d 100644
--- a/loopx/control_plane/todos/machine_section_projection.py
+++ b/loopx/control_plane/todos/machine_section_projection.py
@@ -57,10 +57,26 @@
 
 TODO_SECTION_PROJECTION_SCHEMA_VERSION = "loopx_todo_section_projection_v0"
 DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT = 20
+BOUNDED_ACTIVE_VIEW_MAX_DONE_ITEMS = 5
 SQLITE_ARCHIVE_POINTER_COMMENT = (
     '<!-- loopx:todo-archive-pointer authority=sqlite_v0 '
     'query="loopx todo list --include-archived" -->'
 )
+
+
+def format_sqlite_archive_pointer_comment(
+    *,
+    archived_count: int,
+    active_done_omitted: int = 0,
+) -> str:
+    return (
+        f"<!-- loopx:todo-archive-pointer authority=sqlite_v0 "
+        f"archived_count={archived_count} "
+        f"active_done_omitted={active_done_omitted} "
+        f'query="loopx todo list --include-archived" -->'
+    )
+
+
 BOUNDED_ACTIVE_VIEW_TEXT_LIMIT = 160
 BOUNDED_ACTIVE_VIEW_MAX_SCOPE_CHARS = 120
 BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS = frozenset(
@@ -468,6 +484,40 @@ def _select_projected_archive_records(
     ]
 
 
+def _select_bounded_active_role_records(
+    records: Sequence[dict[str, object]],
+    *,
+    referenced_ids: set[str],
+    max_done_items: int = BOUNDED_ACTIVE_VIEW_MAX_DONE_ITEMS,
+) -> tuple[list[dict[str, object]], int]:
+    done_records = [
+        record
+        for record in records
+        if normalize_todo_status(record.get("status")) == "done"
+        or record.get("done") is True
+    ]
+    if len(done_records) <= max_done_items:
+        return list(records), 0
+    tail_done_ids = {
+        str(record.get("todo_id") or "")
+        for record in (done_records[-max_done_items:] if max_done_items > 0 else [])
+    }
+    retained_done_ids = tail_done_ids | referenced_ids
+    selected: list[dict[str, object]] = []
+    omitted_done = 0
+    for record in records:
+        is_done = (
+            normalize_todo_status(record.get("status")) == "done"
+            or record.get("done") is True
+        )
+        todo_id = str(record.get("todo_id") or "")
+        if not is_done or todo_id in retained_done_ids:
+            selected.append(record)
+        else:
+            omitted_done += 1
+    return selected, omitted_done
+
+
 def _compact_record_for_bounded_active_view(
     record: Mapping[str, object],
 ) -> dict[str, object]:
@@ -564,39 +614,57 @@ def render_canonical_todo_sections(
                 f"Todo {todo_id!r} private validation declaration does not match authority"
             )
         private_validation[todo_id] = source[1]
-    by_role = {
-        role: [
-            (
-                _compact_record_for_bounded_active_view(record)
-                if bounded_active_view
-                else record
-            )
-            for record in sorted(
-                [
-                    record
-                    for record in canonical
-                    if record.get("role") == role
-                    and record.get("archive_state") == "active"
-                ],
-                key=_record_sort_key,
-            )
-        ]
+    raw_active_by_role = {
+        role: sorted(
+            [
+                record
+                for record in canonical
+                if record.get("role") == role
+                and record.get("archive_state") == "active"
+            ],
+            key=_record_sort_key,
+        )
         for role in TODO_SECTION_HEADINGS
     }
-    all_active = [*by_role["user"], *by_role["agent"]]
+    raw_all_active = [*raw_active_by_role["user"], *raw_active_by_role["agent"]]
+    referenced_active_ids = _active_referenced_todo_ids(
+        [
+            record
+            for record in raw_all_active
+            if normalize_todo_status(record.get("status")) != "done"
+            and record.get("done") is not True
+        ]
+    )
+    omitted_done_total = 0
+    by_role: dict[str, list[dict[str, object]]] = {}
+    for role in TODO_SECTION_HEADINGS:
+        role_records = raw_active_by_role[role]
+        if bounded_active_view:
+            selected_role_records, omitted_count = _select_bounded_active_role_records(
+                role_records,
+                referenced_ids=referenced_active_ids,
+            )
+            omitted_done_total += omitted_count
+            by_role[role] = [
+                _compact_record_for_bounded_active_view(record)
+                for record in selected_role_records
+            ]
+        else:
+            by_role[role] = role_records
+    canonical_archived = sorted(
+        [
+            record
+            for record in canonical
+            if record.get("archive_state") == "archive"
+        ],
+        key=_record_sort_key,
+    )
     archived = (
         []
         if bounded_active_view
         else _select_projected_archive_records(
-            sorted(
-                [
-                    record
-                    for record in canonical
-                    if record.get("archive_state") == "archive"
-                ],
-                key=_record_sort_key,
-            ),
-            active_records=all_active,
+            canonical_archived,
+            active_records=raw_all_active,
             max_projected_archive_records=max_projected_archive_records,
         )
     )
@@ -611,7 +679,11 @@ def render_canonical_todo_sections(
             newline=newline,
             private_validation=private_validation,
         )
-    if archived or "archive" in source_document.spans:
+    if (
+        archived
+        or "archive" in source_document.spans
+        or (bounded_active_view and (canonical_archived or omitted_done_total > 0))
+    ):
         rendered_sections["archive"], section_digests["archive"] = _render_section(
             role="archive",
             records=archived,
@@ -619,7 +691,12 @@ def render_canonical_todo_sections(
             newline=newline,
             private_validation=private_validation,
             archive_pointer_comment=(
-                SQLITE_ARCHIVE_POINTER_COMMENT if bounded_active_view else None
+                format_sqlite_archive_pointer_comment(
+                    archived_count=len(canonical_archived),
+                    active_done_omitted=omitted_done_total,
+                )
+                if bounded_active_view
+                else None
             ),
         )
 
@@ -683,6 +760,7 @@ def inspect_todo_section_projection(markdown: str) -> dict[str, object]:
 
 
 __all__ = [
+    "BOUNDED_ACTIVE_VIEW_MAX_DONE_ITEMS",
     "BOUNDED_ACTIVE_VIEW_MAX_SCOPE_CHARS",
     "BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS",
     "BOUNDED_ACTIVE_VIEW_TEXT_LIMIT",
@@ -691,6 +769,7 @@ def inspect_todo_section_projection(markdown: str) -> dict[str, object]:
     "TODO_SECTION_PROJECTION_SCHEMA_VERSION",
     "TodoSectionProjectionError",
     "TodoSectionProjectionResult",
+    "format_sqlite_archive_pointer_comment",
     "inspect_todo_section_projection",
     "render_canonical_todo_sections",
 ]
diff --git a/tests/control_plane/test_sqlite_archive_and_registry_health.py b/tests/control_plane/test_sqlite_archive_and_registry_health.py
index 963d6a7f24..e3561ab7c6 100644
--- a/tests/control_plane/test_sqlite_archive_and_registry_health.py
+++ b/tests/control_plane/test_sqlite_archive_and_registry_health.py
@@ -25,8 +25,9 @@
     DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE,
 )
 from loopx.control_plane.todos.machine_section_projection import (
+    BOUNDED_ACTIVE_VIEW_MAX_DONE_ITEMS,
     BOUNDED_ACTIVE_VIEW_TEXT_LIMIT,
-    SQLITE_ARCHIVE_POINTER_COMMENT,
+    format_sqlite_archive_pointer_comment,
     render_canonical_todo_sections,
 )
 from loopx.control_plane.todos.provider_terminal_lifecycle import (
@@ -67,6 +68,19 @@ def test_sqlite_bounded_active_view_projection_omits_archive_and_large_inline_fi
             "required_write_scopes": [f"loopx/module_{idx}/**/*.py" for idx in range(20)],
         }
     ]
+    for index in range(8):
+        records.append(
+            {
+                "schema_version": "todo_domain_record_v0",
+                "todo_id": f"todo_active_done_{index:02d}",
+                "role": "agent",
+                "status": "done",
+                "done": True,
+                "text": f"Active done step {index}.",
+                "archive_state": "active",
+                "task_class": "advancement_task",
+            }
+        )
     for index in range(30):
         records.append(
             {
@@ -89,16 +103,24 @@ def test_sqlite_bounded_active_view_projection_omits_archive_and_large_inline_fi
         bounded_active_view=True,
     )
     assert projected.changed is True
-    assert projected.todo_count == 31
-    assert SQLITE_ARCHIVE_POINTER_COMMENT in projected.markdown
+    assert projected.todo_count == 39
+    assert (
+        format_sqlite_archive_pointer_comment(
+            archived_count=30,
+            active_done_omitted=8 - BOUNDED_ACTIVE_VIEW_MAX_DONE_ITEMS,
+        )
+        in projected.markdown
+    )
     assert "note=" not in projected.markdown
     assert "evidence=" not in projected.markdown
     assert "reason=" not in projected.markdown
     assert "required_write_scopes=" not in projected.markdown
     assert "Archived work item" not in projected.markdown
+    assert "Active done step 0." not in projected.markdown
+    assert "Active done step 7." in projected.markdown
     active_by_role, archived_rows, _ = parse_todo_source(projected.markdown)
     assert archived_rows == []
-    assert len(active_by_role["agent"]) == 1
+    assert len(active_by_role["agent"]) == 1 + BOUNDED_ACTIVE_VIEW_MAX_DONE_ITEMS
     assert len(active_by_role["agent"][0]["text"]) <= BOUNDED_ACTIVE_VIEW_TEXT_LIMIT
 
     replay = render_canonical_todo_sections(
@@ -198,7 +220,12 @@ def fake_read(**kwargs):
     assert delivered["status"] == "delivered"
     assert delivered["todo_count"] == 26
     rendered = state_file.read_text(encoding="utf-8")
-    assert SQLITE_ARCHIVE_POINTER_COMMENT in rendered
+    assert (
+        format_sqlite_archive_pointer_comment(
+            archived_count=25, active_done_omitted=0
+        )
+        in rendered
+    )
     assert "note=" not in rendered
     assert "evidence=" not in rendered
     _, archived_rows, _ = parse_todo_source(rendered)

From adf82634317db8fa6d45c40a1469b1ec81cc0f4a Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 23:48:39 +0530
Subject: [PATCH 4/7] fix(todos): scope terminal auto-archive to runtime calls
 and clean ruff import

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/control_plane/coordination/local_authority_runtime.ts | 2 ++
 loopx/control_plane/coordination/todo_terminal_lifecycle.ts | 4 +++-
 loopx/control_plane/todos/provider_projection.py            | 1 -
 3 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/loopx/control_plane/coordination/local_authority_runtime.ts b/loopx/control_plane/coordination/local_authority_runtime.ts
index 87bfdff9bf..185cae934f 100644
--- a/loopx/control_plane/coordination/local_authority_runtime.ts
+++ b/loopx/control_plane/coordination/local_authority_runtime.ts
@@ -1414,6 +1414,8 @@ export async function terminalLifecycleLocalCoordinationTodo(
         completion_policy_request:
           input.completion_policy_request === null || input.completion_policy_request === undefined
             ? null : requireJsonObject(input.completion_policy_request, "completion_policy_request"),
+        auto_archive_completed: input.auto_archive_completed === undefined
+          ? true : Boolean(input.auto_archive_completed),
         dry_run: input.dry_run as boolean,
         now: claimObservedAt(input.observed_at),
       }, async () => await authoritySourcesCurrent() && researchEvidence.current(),
diff --git a/loopx/control_plane/coordination/todo_terminal_lifecycle.ts b/loopx/control_plane/coordination/todo_terminal_lifecycle.ts
index c8f3efd9d2..5d6e1e30a2 100644
--- a/loopx/control_plane/coordination/todo_terminal_lifecycle.ts
+++ b/loopx/control_plane/coordination/todo_terminal_lifecycle.ts
@@ -136,6 +136,7 @@ interface CoordinationTodoTerminalLifecycleBaseInput {
   readonly goal_acceptance_validation_receipts?: unknown;
   readonly completion_result?: JsonObject | null;
   readonly completion_policy_request: JsonObject | null;
+  readonly auto_archive_completed?: boolean;
   readonly dry_run: boolean;
   readonly now: Date;
 }
@@ -1945,7 +1946,8 @@ export async function executeCoordinationTodoTerminalLifecycle(
   const activeDoneCount = postTerminalTodos.filter(
     (item) => item.role === targetRole && item.archive_state === "active" && item.status === "done",
   ).length;
-  const autoArchiveSelection = activeDoneCount > DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE
+  const autoArchiveSelection = input.auto_archive_completed === true &&
+    activeDoneCount > DEFAULT_MAX_ACTIVE_DONE_TODOS_BEFORE_ARCHIVE
     ? selectCoordinationTodoArchive({
         role: targetRole,
         max_active_done: ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE,
diff --git a/loopx/control_plane/todos/provider_projection.py b/loopx/control_plane/todos/provider_projection.py
index 1a47be6c1b..e32ab27109 100644
--- a/loopx/control_plane/todos/provider_projection.py
+++ b/loopx/control_plane/todos/provider_projection.py
@@ -24,7 +24,6 @@
     local_authority_is_promoted,
 )
 from .machine_section_projection import (
-    DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT,
     TodoSectionProjectionError,
     render_canonical_todo_sections,
 )

From 9a54013ffda7e2fee6501843a1f30340b580a847 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 2 Nov 2026 08:51:46 +0530
Subject: [PATCH 5/7] feat(todos): add --include-archived to todo list and
 remove unused archive cap API

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../cli_commands/todo_argument_validation.py  |  7 +--
 loopx/cli_commands/todo_list.py               |  1 +
 loopx/cli_commands/todo_registration.py       |  9 ++++
 loopx/control_plane/todos/list_readback.py    |  2 +
 .../todos/machine_section_projection.py       | 49 ++-----------------
 loopx/control_plane/todos/read_transaction.ts | 10 ++--
 ...test_sqlite_archive_and_registry_health.py | 37 ++++++++++++++
 7 files changed, 63 insertions(+), 52 deletions(-)

diff --git a/loopx/cli_commands/todo_argument_validation.py b/loopx/cli_commands/todo_argument_validation.py
index 9dde461561..9829c1d473 100644
--- a/loopx/cli_commands/todo_argument_validation.py
+++ b/loopx/cli_commands/todo_argument_validation.py
@@ -81,6 +81,7 @@
     ("--limit", "todo_limit"),
     ("--thin", "todo_thin"),
     ("--summary", "todo_summary"),
+    ("--include-archived", "include_archived"),
     ("--state-file", "state_file"),
     ("--execute", "execute"),
     ("--provider-revision", "provider_revision"),
@@ -143,7 +144,7 @@
 # Read/claim subsets share validation and help. Update help reuses the mutable
 # field owner; other lifecycle commands retain full help until centralized.
 TODO_ACTION_FIELDS = {
-    "list": frozenset({"role", "todo_id", "status", "agent_id", "todo_limit", "todo_thin", "todo_summary", "state_file", "project", "dry_run"}),
+    "list": frozenset({"role", "todo_id", "status", "agent_id", "todo_limit", "todo_thin", "todo_summary", "include_archived", "state_file", "project", "dry_run"}),
     "receipt": frozenset({"operation_id"}),
     "result-read": frozenset({"todo_id"}),
     "plan": frozenset({"text", "agent_id", "project"}),
@@ -307,8 +308,8 @@ def validate_todo_list_options(args: argparse.Namespace) -> None:
         args,
         TODO_ACTION_FIELDS["list"],
         "todo list only accepts --goal-id, optional --role, --status, --todo-id, "
-        "--agent-id, --limit, --thin, --project, --state-file, --dry-run, and "
-        "--format; "
+        "--agent-id, --limit, --thin, --summary, --include-archived, --project, "
+        "--state-file, --dry-run, and --format; "
         "unsupported: ",
     )
 
diff --git a/loopx/cli_commands/todo_list.py b/loopx/cli_commands/todo_list.py
index e40082b2c5..60a3787bc9 100644
--- a/loopx/cli_commands/todo_list.py
+++ b/loopx/cli_commands/todo_list.py
@@ -39,6 +39,7 @@ def handle_todo_list_command(
             limit=args.todo_limit,
             thin=bool(args.todo_thin),
             **({"summary": True} if getattr(args, "todo_summary", False) else {}),
+            **({"include_archived": True} if getattr(args, "include_archived", False) else {}),
             project=Path(args.project).expanduser() if args.project else None,
             state_file=Path(args.state_file).expanduser() if args.state_file else None,
             runtime_root_arg=runtime_root_arg,
diff --git a/loopx/cli_commands/todo_registration.py b/loopx/cli_commands/todo_registration.py
index 44dd18f66e..5af11265db 100644
--- a/loopx/cli_commands/todo_registration.py
+++ b/loopx/cli_commands/todo_registration.py
@@ -468,6 +468,15 @@ def register_todo_command(
             "--thin or --todo-id."
         ),
     )
+    todo_parser.add_argument(
+        "--include-archived",
+        dest="include_archived",
+        action="store_true",
+        help=(
+            "For todo list, include archived Todos from canonical SQLite "
+            "authority alongside active Todos."
+        ),
+    )
     todo_parser.add_argument("--project", help="Project root. Defaults to the registry goal repo.")
     todo_parser.add_argument("--state-file", help="Active goal state path. Defaults to the registry goal state_file.")
     todo_parser.add_argument("--dry-run", action="store_true", help="Preview the active-state edit without writing.")
diff --git a/loopx/control_plane/todos/list_readback.py b/loopx/control_plane/todos/list_readback.py
index d9e5ed9487..233b94cec7 100644
--- a/loopx/control_plane/todos/list_readback.py
+++ b/loopx/control_plane/todos/list_readback.py
@@ -34,6 +34,7 @@ def list_goal_todos(
     thin: bool = False,
     full_text: bool = False,
     summary: bool = False,
+    include_archived: bool = False,
     read_scope: Literal["active", "completed_history"] = "active",
     json_references: bool = False,
 ) -> dict[str, Any]:
@@ -97,6 +98,7 @@ def list_goal_todos(
             "todo_id": normalized_todo_id, "agent_id": normalized_agent_id, "limit": limit,
             "read_scope": read_scope, "json_references": json_references, "thin": thin,
             **({"full_text": True} if full_text else {}),
+            **({"include_archived": True} if include_archived else {}),
             "state_file": str(resolved_state_file), "project": str(resolved_project) if resolved_project else None}}
     if summary:
         params["query"]["summary"] = True
diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py
index aee5e6ae7d..0af4740736 100644
--- a/loopx/control_plane/todos/machine_section_projection.py
+++ b/loopx/control_plane/todos/machine_section_projection.py
@@ -56,11 +56,10 @@
 
 
 TODO_SECTION_PROJECTION_SCHEMA_VERSION = "loopx_todo_section_projection_v0"
-DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT = 20
 BOUNDED_ACTIVE_VIEW_MAX_DONE_ITEMS = 5
 SQLITE_ARCHIVE_POINTER_COMMENT = (
     '<!-- loopx:todo-archive-pointer authority=sqlite_v0 '
-    'query="loopx todo list --include-archived" -->'
+    'query="loopx todo list --goal-id <goal-id> --include-archived" -->'
 )
 
 
@@ -73,7 +72,7 @@ def format_sqlite_archive_pointer_comment(
         f"<!-- loopx:todo-archive-pointer authority=sqlite_v0 "
         f"archived_count={archived_count} "
         f"active_done_omitted={active_done_omitted} "
-        f'query="loopx todo list --include-archived" -->'
+        f'query="loopx todo list --goal-id <goal-id> --include-archived" -->'
     )
 
 
@@ -452,38 +451,6 @@ def _active_referenced_todo_ids(
     return referenced
 
 
-def _select_projected_archive_records(
-    archived: list[dict[str, object]],
-    *,
-    active_records: Sequence[Mapping[str, object]],
-    max_projected_archive_records: int | None,
-) -> list[dict[str, object]]:
-    if max_projected_archive_records is None:
-        return archived
-    if isinstance(max_projected_archive_records, bool) or not isinstance(
-        max_projected_archive_records, int
-    ) or max_projected_archive_records < 0:
-        raise TodoSectionProjectionError(
-            "max_projected_archive_records must be a non-negative integer or None"
-        )
-    if len(archived) <= max_projected_archive_records:
-        return archived
-    referenced_ids = _active_referenced_todo_ids(active_records)
-    tail_records = (
-        archived[-max_projected_archive_records:]
-        if max_projected_archive_records > 0
-        else []
-    )
-    retained_ids = referenced_ids | {
-        str(record.get("todo_id") or "") for record in tail_records
-    }
-    return [
-        record
-        for record in archived
-        if str(record.get("todo_id") or "") in retained_ids
-    ]
-
-
 def _select_bounded_active_role_records(
     records: Sequence[dict[str, object]],
     *,
@@ -554,7 +521,6 @@ def render_canonical_todo_sections(
     *,
     provider_revision: str,
     private_validation_declarations: Mapping[str, Mapping[str, Any]] | None = None,
-    max_projected_archive_records: int | None = None,
     bounded_active_view: bool = False,
 ) -> TodoSectionProjectionResult:
     """Replace only Todo sections and verify deterministic parse/render parity."""
@@ -659,15 +625,7 @@ def render_canonical_todo_sections(
         ],
         key=_record_sort_key,
     )
-    archived = (
-        []
-        if bounded_active_view
-        else _select_projected_archive_records(
-            canonical_archived,
-            active_records=raw_all_active,
-            max_projected_archive_records=max_projected_archive_records,
-        )
-    )
+    archived = [] if bounded_active_view else canonical_archived
     newline = "\r\n" if "\r\n" in markdown else "\n"
     rendered_sections: dict[str, str] = {}
     section_digests: dict[str, str] = {}
@@ -764,7 +722,6 @@ def inspect_todo_section_projection(markdown: str) -> dict[str, object]:
     "BOUNDED_ACTIVE_VIEW_MAX_SCOPE_CHARS",
     "BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS",
     "BOUNDED_ACTIVE_VIEW_TEXT_LIMIT",
-    "DEFAULT_SQLITE_PROJECTED_ARCHIVE_LIMIT",
     "SQLITE_ARCHIVE_POINTER_COMMENT",
     "TODO_SECTION_PROJECTION_SCHEMA_VERSION",
     "TodoSectionProjectionError",
diff --git a/loopx/control_plane/todos/read_transaction.ts b/loopx/control_plane/todos/read_transaction.ts
index 1c636a9280..d0cdfb3ca4 100644
--- a/loopx/control_plane/todos/read_transaction.ts
+++ b/loopx/control_plane/todos/read_transaction.ts
@@ -37,7 +37,9 @@ export async function readCanonicalTodoProjection(input: JsonObject): Promise<Js
     source_authority: first!.source_authority, provider_revision: snapshot.provider_revision,
     cursor: snapshot.cursor, decision_read_from_provider: true, legacy_fallback_used: false,
     ...(Object.hasOwn(requireJsonObject(first!.metadata, "canonical metadata"), "goal_acceptance_contract") ? {goal_acceptance_work_guards: guards} : {})};
-  const projected = await projectCanonicalTodoFields({...input, todos, retained: requireJsonObject(input.query ?? {}, "Todo query").read_scope === "completed_history",
+  const query = requireJsonObject(input.query ?? {}, "Todo query");
+  const projected = await projectCanonicalTodoFields({...input, todos,
+    retained: query.read_scope === "completed_history" || query.include_archived === true,
     goal_acceptance_contract: canonical.goal_acceptance_contract ?? null,
     goal_acceptance_work_guards: guards});
   if (input.query == null || projected.rejection) return {...projected, canonical};
@@ -111,7 +113,8 @@ async function todoListPayload(input: JsonObject, original: JsonObject[], fields
     item.text = String(matching[0]!.text ?? "");
   }
   const matchedCount = todos.length;
-  const hot = !fullText && query.read_scope !== "completed_history" && query.agent_id && query.limit == null &&
+  const includeArchived = query.read_scope === "completed_history" || query.include_archived === true;
+  const hot = !fullText && !includeArchived && query.agent_id && query.limit == null &&
     query.role == null && query.status == null && query.todo_id == null;
   if (hot && !summary) {
     const summaries = requireJsonObject(projection.summaries, "Todo summaries");
@@ -121,7 +124,8 @@ async function todoListPayload(input: JsonObject, original: JsonObject[], fields
   }
   const payload: JsonObject = {ok: true, dry_run: true, read_only: true, command: "list", goal_id: query.goal_id,
     role: query.role ?? "all", status_filter: query.status ?? null, source: sourceName, todo_count: matchedCount,
-    todos, ...(fullText ? {full_text: true} : {}), state_file: query.state_file, project: query.project ?? null,
+    todos, ...(fullText ? {full_text: true} : {}), ...(query.include_archived === true ? {include_archived: true} : {}),
+    state_file: query.state_file, project: query.project ?? null,
     authority_read: {source_authority: canonical.source_authority, provider_revision: canonical.provider_revision,
       cursor: canonical.cursor, todo_read_model: canonical.todo_read_model,
       decision_read_from_provider: true, legacy_fallback_used: false}};
diff --git a/tests/control_plane/test_sqlite_archive_and_registry_health.py b/tests/control_plane/test_sqlite_archive_and_registry_health.py
index e3561ab7c6..c46d3a35fe 100644
--- a/tests/control_plane/test_sqlite_archive_and_registry_health.py
+++ b/tests/control_plane/test_sqlite_archive_and_registry_health.py
@@ -396,6 +396,43 @@ def test_terminal_13th_todo_in_sqlite_auto_archives_in_same_transaction(
     assert len(active_done) == ARCHIVE_COMPLETED_DEFAULT_MAX_ACTIVE_DONE
     assert len(archived_done) == 3
 
+    from loopx.cli import main
+    from loopx.control_plane.todos.list_readback import list_goal_todos
+
+    active_list = list_goal_todos(
+        registry_path=registry_path,
+        runtime_root_arg=str(runtime_root),
+        goal_id="goal-a",
+        include_archived=False,
+    )
+    archived_list = list_goal_todos(
+        registry_path=registry_path,
+        runtime_root_arg=str(runtime_root),
+        goal_id="goal-a",
+        include_archived=True,
+    )
+    active_ids = {item["todo_id"] for item in active_list["todos"]}
+    with_archived_ids = {item["todo_id"] for item in archived_list["todos"]}
+    assert "todo_done_00" not in active_ids
+    assert {"todo_done_00", "todo_done_01", "todo_done_02"} <= with_archived_ids
+    assert archived_list.get("include_archived") is True
+    rc = main(
+        [
+            "--registry",
+            str(registry_path),
+            "--runtime-root",
+            str(runtime_root),
+            "--format",
+            "json",
+            "todo",
+            "list",
+            "--goal-id",
+            "goal-a",
+            "--include-archived",
+        ]
+    )
+    assert rc == 0
+
 
 def test_multi_goal_source_registry_sync_and_health_prevent_sibling_false_positives(
     tmp_path: Path,

From 594fda45692be5f8122a583499f27b21f3e25736 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 2 Nov 2026 09:16:53 +0530
Subject: [PATCH 6/7] fix(todos): preserve bounded active note on open sqlite
 Todos for handoff projection

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../todos/machine_section_projection.py       | 23 +++++++++++++++----
 1 file changed, 18 insertions(+), 5 deletions(-)

diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py
index 0af4740736..6902e0cfbe 100644
--- a/loopx/control_plane/todos/machine_section_projection.py
+++ b/loopx/control_plane/todos/machine_section_projection.py
@@ -78,6 +78,7 @@ def format_sqlite_archive_pointer_comment(
 
 BOUNDED_ACTIVE_VIEW_TEXT_LIMIT = 160
 BOUNDED_ACTIVE_VIEW_MAX_SCOPE_CHARS = 120
+BOUNDED_ACTIVE_VIEW_MAX_ACTIVE_NOTE_CHARS = 512
 BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS = frozenset(
     {"note", "evidence", "reason", "completion_result"}
 )
@@ -488,11 +489,23 @@ def _select_bounded_active_role_records(
 def _compact_record_for_bounded_active_view(
     record: Mapping[str, object],
 ) -> dict[str, object]:
-    compacted: dict[str, object] = {
-        key: value
-        for key, value in record.items()
-        if key not in BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS
-    }
+    is_done_or_archived = (
+        normalize_todo_status(record.get("status")) == "done"
+        or record.get("done") is True
+        or record.get("archive_state") == "archive"
+    )
+    compacted: dict[str, object] = {}
+    for key, value in record.items():
+        if key in BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS:
+            if (
+                not is_done_or_archived
+                and key == "note"
+                and isinstance(value, str)
+                and len(value) <= BOUNDED_ACTIVE_VIEW_MAX_ACTIVE_NOTE_CHARS
+            ):
+                compacted[key] = value
+            continue
+        compacted[key] = value
     scopes = compacted.get("required_write_scopes")
     if (
         isinstance(scopes, list)

From fb5408038c8dfe033452d2381f73a580fe81ded6 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 2 Nov 2026 15:05:30 +0530
Subject: [PATCH 7/7] fix(control-plane): surface active field omissions and
 bound cold-read queries in sqlite_v0 projection

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../todos/machine_section_projection.py       | 240 ++++++++++++++++--
 .../todos/provider_projection.py              |   3 +
 ...test_sqlite_archive_and_registry_health.py | 200 +++++++++++++++
 3 files changed, 424 insertions(+), 19 deletions(-)

diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py
index 6902e0cfbe..037036f625 100644
--- a/loopx/control_plane/todos/machine_section_projection.py
+++ b/loopx/control_plane/todos/machine_section_projection.py
@@ -67,12 +67,58 @@ def format_sqlite_archive_pointer_comment(
     *,
     archived_count: int,
     active_done_omitted: int = 0,
+    active_fields_omitted: int = 0,
+    omitted_fields: Sequence[str] = (),
+    omitted_todo_ids: Sequence[str] = (),
+    goal_id: str | None = None,
 ) -> str:
+    resolved_goal = (
+        goal_id.strip()
+        if isinstance(goal_id, str) and goal_id.strip()
+        else "<goal-id>"
+    )
+    extra_parts: list[str] = []
+    if active_fields_omitted > 0:
+        extra_parts.append(f"active_fields_omitted={active_fields_omitted}")
+    if omitted_todo_ids:
+        extra_parts.append(f"omitted_todo_ids={','.join(omitted_todo_ids)}")
+    if omitted_fields:
+        extra_parts.append(f"omitted_fields={','.join(omitted_fields)}")
+    if goal_id and isinstance(goal_id, str) and goal_id.strip():
+        extra_parts.append(f"goal_id={resolved_goal}")
+    if omitted_todo_ids:
+        extra_parts.append(
+            f'todo_query="loopx todo list --goal-id {resolved_goal} '
+            f'--todo-id {omitted_todo_ids[0]}"'
+        )
+    extra_suffix = (" " + " ".join(extra_parts)) if extra_parts else ""
     return (
         f"<!-- loopx:todo-archive-pointer authority=sqlite_v0 "
         f"archived_count={archived_count} "
-        f"active_done_omitted={active_done_omitted} "
-        f'query="loopx todo list --goal-id <goal-id> --include-archived" -->'
+        f"active_done_omitted={active_done_omitted}"
+        f"{extra_suffix} "
+        f'query="loopx todo list --goal-id {resolved_goal} --include-archived" -->'
+    )
+
+
+def format_sqlite_active_omission_pointer_comment(
+    *,
+    goal_id: str | None,
+    role: str,
+    todo_id: str,
+    omitted_fields: Sequence[str],
+) -> str:
+    resolved_goal = (
+        goal_id.strip()
+        if isinstance(goal_id, str) and goal_id.strip()
+        else "<goal-id>"
+    )
+    fields_token = ",".join(omitted_fields)
+    return (
+        f"<!-- loopx:todo-archive-pointer authority=sqlite_v0 "
+        f"scope=active_todo goal_id={resolved_goal} role={role} "
+        f"todo_id={todo_id} omitted_fields={fields_token} "
+        f'query="loopx todo list --goal-id {resolved_goal} --todo-id {todo_id}" -->'
     )
 
 
@@ -104,6 +150,8 @@ class TodoSectionProjectionResult:
     provider_revision: str
     todo_count: int
     section_record_sha256: dict[str, str]
+    active_fields_omitted: int = 0
+    omitted_active_records: tuple[dict[str, object], ...] = ()
 
 
 def _sha256_text(value: str) -> str:
@@ -264,6 +312,7 @@ def _render_section(
     newline: str,
     private_validation: Mapping[str, Mapping[str, object]],
     archive_pointer_comment: str | None = None,
+    pointer_comments: Sequence[str] = (),
 ) -> tuple[str, str]:
     digest = _sha256_text(_canonical_json(records))
     heading = (
@@ -271,12 +320,16 @@ def _render_section(
         if role == "archive"
         else TODO_SECTION_HEADINGS[role]
     )
+    comments = [
+        *([archive_pointer_comment] if archive_pointer_comment else []),
+        *[comment for comment in pointer_comments if comment],
+    ]
     lines = [
         f"## {heading}",
         todo_region_marker(role, "begin"),
         f"<!-- loopx:todo-section-projection-v0 role={role} "
         f"provider_revision={provider_revision} records_sha256={digest} -->",
-        *([archive_pointer_comment] if archive_pointer_comment else []),
+        *comments,
         "",
     ]
     for record in records:
@@ -486,15 +539,16 @@ def _select_bounded_active_role_records(
     return selected, omitted_done
 
 
-def _compact_record_for_bounded_active_view(
+def _compact_record_for_bounded_active_view_with_omissions(
     record: Mapping[str, object],
-) -> dict[str, object]:
+) -> tuple[dict[str, object], tuple[str, ...]]:
     is_done_or_archived = (
         normalize_todo_status(record.get("status")) == "done"
         or record.get("done") is True
         or record.get("archive_state") == "archive"
     )
     compacted: dict[str, object] = {}
+    omitted_fields: list[str] = []
     for key, value in record.items():
         if key in BOUNDED_ACTIVE_VIEW_OMITTED_PROSE_FIELDS:
             if (
@@ -504,6 +558,8 @@ def _compact_record_for_bounded_active_view(
                 and len(value) <= BOUNDED_ACTIVE_VIEW_MAX_ACTIVE_NOTE_CHARS
             ):
                 compacted[key] = value
+            elif value not in (None, "", []):
+                omitted_fields.append(key)
             continue
         compacted[key] = value
     scopes = compacted.get("required_write_scopes")
@@ -512,12 +568,14 @@ def _compact_record_for_bounded_active_view(
         and sum(len(str(item)) for item in scopes) > BOUNDED_ACTIVE_VIEW_MAX_SCOPE_CHARS
     ):
         compacted.pop("required_write_scopes", None)
+        omitted_fields.append("required_write_scopes")
     raw_text = " ".join(str(compacted.get("text") or "").strip().split())
     if len(raw_text) > BOUNDED_ACTIVE_VIEW_TEXT_LIMIT:
         truncated_text = (
             raw_text[: BOUNDED_ACTIVE_VIEW_TEXT_LIMIT - 3].rstrip() + "..."
         )
         compacted["text"] = truncated_text
+        omitted_fields.append("text")
         priority, title = todo_priority_parts(truncated_text)
         if priority:
             compacted["priority"] = priority
@@ -525,9 +583,44 @@ def _compact_record_for_bounded_active_view(
         else:
             compacted.pop("priority", None)
             compacted.pop("title", None)
+    ordered_omitted = tuple(
+        field
+        for field in (
+            "text",
+            "note",
+            "evidence",
+            "reason",
+            "completion_result",
+            "required_write_scopes",
+        )
+        if field in omitted_fields
+    )
+    return compacted, ordered_omitted
+
+
+def _compact_record_for_bounded_active_view(
+    record: Mapping[str, object],
+) -> dict[str, object]:
+    compacted, _ = _compact_record_for_bounded_active_view_with_omissions(record)
     return compacted
 
 
+def _extract_frontmatter_goal_id(markdown: str) -> str | None:
+    lines = markdown.splitlines()
+    if not lines or lines[0].strip() != "---":
+        return None
+    for line in lines[1:40]:
+        stripped = line.strip()
+        if stripped in {"---", "..."}:
+            break
+        for prefix in ("goal_id:", "adapter_id:"):
+            if stripped.startswith(prefix):
+                candidate = stripped.partition(":")[2].strip().strip("\"'")
+                if candidate and re.fullmatch(r"[A-Za-z0-9_.:-]+", candidate):
+                    return candidate
+    return None
+
+
 def render_canonical_todo_sections(
     markdown: str,
     records: Sequence[Mapping[str, object]],
@@ -535,6 +628,7 @@ def render_canonical_todo_sections(
     provider_revision: str,
     private_validation_declarations: Mapping[str, Mapping[str, Any]] | None = None,
     bounded_active_view: bool = False,
+    goal_id: str | None = None,
 ) -> TodoSectionProjectionResult:
     """Replace only Todo sections and verify deterministic parse/render parity."""
 
@@ -614,8 +708,18 @@ def render_canonical_todo_sections(
             and record.get("done") is not True
         ]
     )
+    resolved_goal_id = (
+        goal_id.strip()
+        if isinstance(goal_id, str) and goal_id.strip()
+        else _extract_frontmatter_goal_id(markdown)
+    )
+    query_goal_id = resolved_goal_id or "<goal-id>"
     omitted_done_total = 0
     by_role: dict[str, list[dict[str, object]]] = {}
+    omitted_active_by_role: dict[str, list[dict[str, object]]] = {
+        role: [] for role in TODO_SECTION_HEADINGS
+    }
+    omitted_active_records: list[dict[str, object]] = []
     for role in TODO_SECTION_HEADINGS:
         role_records = raw_active_by_role[role]
         if bounded_active_view:
@@ -624,10 +728,26 @@ def render_canonical_todo_sections(
                 referenced_ids=referenced_active_ids,
             )
             omitted_done_total += omitted_count
-            by_role[role] = [
-                _compact_record_for_bounded_active_view(record)
-                for record in selected_role_records
-            ]
+            compacted_role_records: list[dict[str, object]] = []
+            for record in selected_role_records:
+                compacted, omitted_fields = (
+                    _compact_record_for_bounded_active_view_with_omissions(record)
+                )
+                compacted_role_records.append(compacted)
+                if omitted_fields:
+                    todo_id = str(record.get("todo_id") or "")
+                    entry: dict[str, object] = {
+                        "todo_id": todo_id,
+                        "role": role,
+                        "omitted_fields": list(omitted_fields),
+                        "query": (
+                            f"loopx todo list --goal-id {query_goal_id} "
+                            f"--todo-id {todo_id}"
+                        ),
+                    }
+                    omitted_active_by_role[role].append(entry)
+                    omitted_active_records.append(entry)
+            by_role[role] = compacted_role_records
         else:
             by_role[role] = role_records
     canonical_archived = sorted(
@@ -643,32 +763,83 @@ def render_canonical_todo_sections(
     rendered_sections: dict[str, str] = {}
     section_digests: dict[str, str] = {}
     for role in TODO_SECTION_HEADINGS:
+        role_pointer_comments = (
+            [
+                format_sqlite_active_omission_pointer_comment(
+                    goal_id=resolved_goal_id,
+                    role=role,
+                    todo_id=str(item["todo_id"]),
+                    omitted_fields=tuple(str(f) for f in item["omitted_fields"]),  # type: ignore[index]
+                )
+                for item in omitted_active_by_role[role]
+            ]
+            if bounded_active_view
+            else []
+        )
         rendered_sections[role], section_digests[role] = _render_section(
             role=role,
             records=by_role[role],
             provider_revision=provider_revision,
             newline=newline,
             private_validation=private_validation,
+            pointer_comments=role_pointer_comments,
         )
     if (
         archived
         or "archive" in source_document.spans
-        or (bounded_active_view and (canonical_archived or omitted_done_total > 0))
+        or (
+            bounded_active_view
+            and (
+                canonical_archived
+                or omitted_done_total > 0
+                or bool(omitted_active_records)
+            )
+        )
     ):
+        unique_omitted_fields = tuple(
+            field
+            for field in (
+                "text",
+                "note",
+                "evidence",
+                "reason",
+                "completion_result",
+                "required_write_scopes",
+            )
+            if any(
+                field in item["omitted_fields"]  # type: ignore[operator]
+                for item in omitted_active_records
+            )
+        )
+        omitted_todo_ids = tuple(
+            str(item["todo_id"]) for item in omitted_active_records
+        )
+        archive_pointer_comments: list[str] = []
+        if bounded_active_view:
+            archive_pointer_comments.append(
+                format_sqlite_archive_pointer_comment(
+                    archived_count=len(canonical_archived),
+                    active_done_omitted=omitted_done_total,
+                )
+            )
+            if omitted_active_records or resolved_goal_id:
+                archive_pointer_comments.append(
+                    format_sqlite_archive_pointer_comment(
+                        archived_count=len(canonical_archived),
+                        active_done_omitted=omitted_done_total,
+                        active_fields_omitted=len(omitted_active_records),
+                        omitted_fields=unique_omitted_fields,
+                        omitted_todo_ids=omitted_todo_ids,
+                        goal_id=resolved_goal_id,
+                    )
+                )
         rendered_sections["archive"], section_digests["archive"] = _render_section(
             role="archive",
             records=archived,
             provider_revision=provider_revision,
             newline=newline,
             private_validation=private_validation,
-            archive_pointer_comment=(
-                format_sqlite_archive_pointer_comment(
-                    archived_count=len(canonical_archived),
-                    active_done_omitted=omitted_done_total,
-                )
-                if bounded_active_view
-                else None
-            ),
+            pointer_comments=archive_pointer_comments,
         )
 
     try:
@@ -710,23 +881,53 @@ def render_canonical_todo_sections(
         provider_revision=provider_revision,
         todo_count=len(canonical),
         section_record_sha256=section_digests,
+        active_fields_omitted=len(omitted_active_records),
+        omitted_active_records=tuple(omitted_active_records),
     )
 
 
+_ACTIVE_OMISSION_POINTER_PATTERN = re.compile(
+    r"^<!-- loopx:todo-archive-pointer authority=sqlite_v0 "
+    r"scope=active_todo goal_id=(?P<goal_id>[^ ]+) role=(?P<role>user|agent) "
+    r"todo_id=(?P<todo_id>[A-Za-z0-9_.:-]+) "
+    r"omitted_fields=(?P<omitted_fields>[a-z_,]+) "
+    r'query="(?P<query>[^"]+)" -->$'
+)
+
+
 def inspect_todo_section_projection(markdown: str) -> dict[str, object]:
     """Return compact marker diagnostics without treating Markdown as truth."""
 
     document = TodoProjectionDocument.parse(markdown)
     markers = []
+    omitted_active_records: list[dict[str, object]] = []
     for region in document.regions:
         for line in document.lines[region.start:region.body_end]:
-            match = _MARKER_PATTERN.fullmatch(line.rstrip("\r\n"))
+            stripped = line.rstrip("\r\n")
+            match = _MARKER_PATTERN.fullmatch(stripped)
             if match is not None and match["role"] == region.role:
                 markers.append(match.groupdict())
+            omission = _ACTIVE_OMISSION_POINTER_PATTERN.fullmatch(stripped)
+            if omission is not None and omission["role"] == region.role:
+                omitted_active_records.append(
+                    {
+                        "todo_id": omission["todo_id"],
+                        "role": omission["role"],
+                        "goal_id": omission["goal_id"],
+                        "omitted_fields": [
+                            item
+                            for item in omission["omitted_fields"].split(",")
+                            if item
+                        ],
+                        "query": omission["query"],
+                    }
+                )
     return {
         "schema_version": TODO_SECTION_PROJECTION_SCHEMA_VERSION,
         "section_count": len(markers),
         "sections": markers,
+        "active_fields_omitted": len(omitted_active_records),
+        "omitted_active_records": omitted_active_records,
     }
 
 
@@ -739,6 +940,7 @@ def inspect_todo_section_projection(markdown: str) -> dict[str, object]:
     "TODO_SECTION_PROJECTION_SCHEMA_VERSION",
     "TodoSectionProjectionError",
     "TodoSectionProjectionResult",
+    "format_sqlite_active_omission_pointer_comment",
     "format_sqlite_archive_pointer_comment",
     "inspect_todo_section_projection",
     "render_canonical_todo_sections",
diff --git a/loopx/control_plane/todos/provider_projection.py b/loopx/control_plane/todos/provider_projection.py
index e32ab27109..062b7078e1 100644
--- a/loopx/control_plane/todos/provider_projection.py
+++ b/loopx/control_plane/todos/provider_projection.py
@@ -144,6 +144,7 @@ def project_current_canonical_todos(
                 bounded_active_view=(
                     authority_read.get("source_authority") == "sqlite_v0"
                 ),
+                goal_id=goal_id,
             )
             if execute and projection.changed:
                 # The projection is a primary-state write: it must respect the
@@ -202,6 +203,8 @@ def project_current_canonical_todos(
         "rendered_sha256": projection.rendered_sha256,
         "narrative_sha256": projection.narrative_sha256,
         "section_record_sha256": projection.section_record_sha256,
+        "active_fields_omitted": projection.active_fields_omitted,
+        "omitted_active_records": list(projection.omitted_active_records),
         "parse_render_parity": True,
         "narrative_preserved": not recovered_missing,
         **({"recovery_scope": "todo_sections_only"} if recovered_missing else {}),
diff --git a/tests/control_plane/test_sqlite_archive_and_registry_health.py b/tests/control_plane/test_sqlite_archive_and_registry_health.py
index c46d3a35fe..a2fef7ed49 100644
--- a/tests/control_plane/test_sqlite_archive_and_registry_health.py
+++ b/tests/control_plane/test_sqlite_archive_and_registry_health.py
@@ -519,3 +519,203 @@ def test_multi_goal_source_registry_sync_and_health_prevent_sibling_false_positi
         if finding.get("kind") == "stale_source_registry"
     ]
     assert stale_goal_ids == ["goal-drifted"]
+
+
+@pytest.mark.parametrize(
+    ("scenario", "todo_payload", "expected_omitted_fields", "expect_inline_note"),
+    [
+        (
+            "zero_archive_long_text_and_note",
+            {
+                "todo_id": "todo_open_both",
+                "text": "[P0] " + ("Acceptance requirement and stop condition " * 11)[:454],
+                "note": "stop-gate-detail-" * 50 + "end853",
+            },
+            ["text", "note"],
+            False,
+        ),
+        (
+            "zero_archive_long_text_only",
+            {
+                "todo_id": "todo_open_long_text",
+                "text": "[P1] " + ("Long body with acceptance tail " * 12).strip(),
+                "note": "Short stop condition retained inline.",
+            },
+            ["text"],
+            True,
+        ),
+        (
+            "zero_archive_long_note_and_scopes_only",
+            {
+                "todo_id": "todo_open_long_meta",
+                "text": "[P1] Compact title preserved in full.",
+                "note": "extended-note-stop-condition-" * 30,
+                "required_write_scopes": [
+                    f"loopx/control_plane/module_{idx}/**/*.py" for idx in range(10)
+                ],
+            },
+            ["note", "required_write_scopes"],
+            False,
+        ),
+    ],
+)
+def test_sqlite_project_markdown_zero_archive_active_field_omission_and_cold_read(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    scenario: str,
+    todo_payload: dict[str, object],
+    expected_omitted_fields: list[str],
+    expect_inline_note: bool,
+) -> None:
+    from loopx import cli
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    runtime_root = tmp_path / "runtime"
+    project = tmp_path / "project"
+    state_file = project / ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md"
+    state_file.parent.mkdir(parents=True)
+    # Intentionally omit "## Completed Work Archive" and keep zero done/archived items.
+    state_file.write_text(
+        "# Goal\n\n## User Todo / Owner Review Reading Queue\n\n## Agent Todo\n",
+        encoding="utf-8",
+    )
+    registry = tmp_path / "registry.json"
+    registry.write_text(
+        json.dumps(
+            {
+                "schema_version": 1,
+                "common_runtime_root": str(runtime_root),
+                "goals": [
+                    {
+                        "id": "goal-a",
+                        "status": "active",
+                        "repo": str(project),
+                        "state_file": ".codex/goals/goal-a/ACTIVE_GOAL_STATE.md",
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    todo_id = str(todo_payload["todo_id"])
+    full_text = str(todo_payload["text"])
+    full_note = str(todo_payload["note"])
+    projection = build_todo_runtime_shadow_projection(
+        goal_id="goal-a",
+        todos=[
+            {
+                "schema_version": "todo_item_v0",
+                "role": "agent",
+                "status": "open",
+                "done": False,
+                "archive_state": "active",
+                "source_section": "Agent Todo",
+                "index": 1,
+                "task_class": "advancement_task",
+                **todo_payload,
+            }
+        ],
+        handoff_mode="soft_claim",
+    )
+    initialize_canonical_authority(
+        runtime_root, "goal-a", projection, state_path=state_file, provider="sqlite"
+    )
+    loaded = read_canonical_todos_if_promoted(runtime_root=runtime_root, goal_id="goal-a")
+    assert loaded is not None
+    provider_revision = str(loaded["provider_revision"])
+
+    rc = cli.main(
+        [
+            "--registry",
+            str(registry),
+            "--runtime-root",
+            str(runtime_root),
+            "--format",
+            "json",
+            "todo",
+            "project-markdown",
+            "--goal-id",
+            "goal-a",
+            "--provider-revision",
+            provider_revision,
+            "--execute",
+        ]
+    )
+    assert rc == 0, scenario
+    projection_out = json.loads(capsys.readouterr().out)
+    assert projection_out["active_fields_omitted"] == 1
+    assert projection_out["omitted_active_records"] == [
+        {
+            "todo_id": todo_id,
+            "role": "agent",
+            "omitted_fields": expected_omitted_fields,
+            "query": f"loopx todo list --goal-id goal-a --todo-id {todo_id}",
+        }
+    ]
+
+    rendered = state_file.read_text(encoding="utf-8")
+    expected_comment = (
+        "<!-- loopx:todo-archive-pointer authority=sqlite_v0 "
+        f"scope=active_todo goal_id=goal-a role=agent todo_id={todo_id} "
+        f"omitted_fields={','.join(expected_omitted_fields)} "
+        f'query="loopx todo list --goal-id goal-a --todo-id {todo_id}" -->'
+    )
+    assert expected_comment in rendered
+    parsed_active, _, _ = parse_todo_source(rendered)
+    assert len(parsed_active["agent"]) == 1
+    if expect_inline_note:
+        assert parsed_active["agent"][0].get("note") == full_note
+    else:
+        assert "note=" not in rendered
+        assert parsed_active["agent"][0].get("note") is None
+
+    # Repeated rendering via project-markdown must be idempotent.
+    rc_repeat = cli.main(
+        [
+            "--registry",
+            str(registry),
+            "--runtime-root",
+            str(runtime_root),
+            "--format",
+            "json",
+            "todo",
+            "project-markdown",
+            "--goal-id",
+            "goal-a",
+            "--provider-revision",
+            provider_revision,
+            "--execute",
+        ]
+    )
+    assert rc_repeat == 0
+    repeat_out = json.loads(capsys.readouterr().out)
+    assert repeat_out["changed"] is False
+    assert state_file.read_text(encoding="utf-8") == rendered
+
+    # Execute the bound cold-read query from the pointer comment and verify full requirements.
+    rc_list = cli.main(
+        [
+            "--registry",
+            str(registry),
+            "--runtime-root",
+            str(runtime_root),
+            "--format",
+            "json",
+            "todo",
+            "list",
+            "--goal-id",
+            "goal-a",
+            "--todo-id",
+            todo_id,
+        ]
+    )
+    assert rc_list == 0
+    cold_read = json.loads(capsys.readouterr().out)
+    item = cold_read["todo"]
+    assert item is not None
+    assert item["todo_id"] == todo_id
+    assert item["text"] == full_text
+    assert item["note"] == full_note
+    if "required_write_scopes" in todo_payload:
+        assert item["required_write_scopes"] == todo_payload["required_write_scopes"]