feat(control-plane): sqlite_v0 bounded active view projection, in-transaction auto-archive, and goal-scoped stale_source_registry

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: ",
     )
 

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,

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.")

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(),

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;
@@ -133,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;
 }
@@ -1929,6 +1933,45 @@ 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 = 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,
+        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 +2002,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 +2023,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}]),
   ] : [];

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",

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

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: