From 86a1441010b061623cbec725b32028324f481941 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 23:40:16 +0800
Subject: [PATCH 1/5] refactor(replan): centralize typed history windows and
 count logical turns

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../control_plane/effect_runtime_handlers.ts  |   2 +
 .../status/autonomous_replan_projection.py    |  10 +-
 .../work_items/autonomous_replan_ack.py       |  21 +-
 .../autonomous_replan_obligation.py           | 398 ++----------------
 .../work_items/progress_observation.py        |  53 +--
 .../work_items/replan_history.py              | 145 +++++++
 .../work_items/replan_history.ts              | 304 +++++++++++++
 loopx/status.py                               |  10 +-
 .../test_replan_history_policy.py             | 146 +++++++
 .../test_replan_history_provider.py           |  44 ++
 tests/control_plane_ts/replan_history.test.ts | 167 ++++++++
 tsconfig.control-plane.json                   |   1 +
 12 files changed, 856 insertions(+), 445 deletions(-)
 create mode 100644 loopx/control_plane/work_items/replan_history.py
 create mode 100644 loopx/control_plane/work_items/replan_history.ts
 create mode 100644 tests/control_plane/test_replan_history_policy.py
 create mode 100644 tests/control_plane/test_replan_history_provider.py
 create mode 100644 tests/control_plane_ts/replan_history.test.ts

diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 0882c6941d..ca57f3aac2 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -118,6 +118,7 @@ import {
 } from "./turn_driver/delivery_continuity.ts";
 import { reduceTurnSettlementTransaction } from "./turn_driver/settlement.ts";
 import { evaluateHostTodoCompletion } from "./turn_driver/host_todo_completion.ts";
+import { projectReplanHistory } from "./work_items/replan_history.ts";
 import { projectReplanSemantics } from "./work_items/replan_semantics.ts";
 import {
   projectReplanSettlementContract,
@@ -784,6 +785,7 @@ export function createEffectRuntimeHandlers(
     ["turn.host_todo_completion.evaluate", evaluateHostTodoCompletion],
     ["work_item.replan_settlement.project", projectReplanSettlementContract],
     ["work_item.replan_semantics.project", projectReplanSemantics],
+  ["work_item.replan_history.project", projectReplanHistory],
     [
       "work_item.replan_settlement.reentry",
       projectTodoLifecycleSettlementReentry,
diff --git a/loopx/control_plane/status/autonomous_replan_projection.py b/loopx/control_plane/status/autonomous_replan_projection.py
index f479a25250..8f99882bf7 100644
--- a/loopx/control_plane/status/autonomous_replan_projection.py
+++ b/loopx/control_plane/status/autonomous_replan_projection.py
@@ -2,8 +2,13 @@
 
 from __future__ import annotations
 
+
 from typing import Any
 
+from ..work_items.replan_history import (
+    REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS as AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS,
+)
+
 from ..runtime.public_safety import public_safe_compact_text
 from ..work_items.autonomous_replan_ack import (
     AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW,
@@ -23,11 +28,6 @@
 DEAD_MONITOR_REPEAT_SCHEMA_VERSION = "dead_monitor_repeat_v0"
 AUTONOMOUS_REPLAN_SCHEMA_VERSION = "autonomous_replan_obligation_v0"
 AUTONOMOUS_REPLAN_PERIODIC_RUN_THRESHOLD = AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW
-AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS = {
-    "quota_slot_spent",
-    "quota_slot_voided",
-    "delivery_completion_spend_accounted_v0",
-}
 
 
 def build_autonomous_replan_obligation(
diff --git a/loopx/control_plane/work_items/autonomous_replan_ack.py b/loopx/control_plane/work_items/autonomous_replan_ack.py
index 8e00d7d9fc..13e400abe0 100644
--- a/loopx/control_plane/work_items/autonomous_replan_ack.py
+++ b/loopx/control_plane/work_items/autonomous_replan_ack.py
@@ -122,16 +122,10 @@ def compact_autonomous_replan_ack(run: dict[str, Any] | None) -> dict[str, Any]
         isinstance(outcomes, list)
         and "fresh_vision_path_outcome" in outcomes
     ):
-        agent_vision = (
-            run.get("agent_vision")
-            if isinstance(run.get("agent_vision"), dict)
-            else {}
-        )
-        path_delta = (
-            agent_vision.get("path_delta")
-            if isinstance(agent_vision.get("path_delta"), dict)
-            else {}
-        )
+        raw_vision = run.get("agent_vision")
+        agent_vision = raw_vision if isinstance(raw_vision, dict) else {}
+        raw_path = agent_vision.get("path_delta")
+        path_delta = raw_path if isinstance(raw_path, dict) else {}
         path_disposition = str(path_delta.get("outcome") or "").strip()
         if path_disposition in FRESH_VISION_PATH_DISPOSITIONS:
             result["path_disposition"] = path_disposition
@@ -211,11 +205,8 @@ def _latest_monitor_replan_frontier_identity(
     for run in latest_runs or []:
         if not isinstance(run, dict):
             continue
-        target = (
-            run.get("monitor_target")
-            if isinstance(run.get("monitor_target"), dict)
-            else {}
-        )
+        raw_target = run.get("monitor_target")
+        target = raw_target if isinstance(raw_target, dict) else {}
         if normalized_agent_id:
             run_agent_id = str(run.get("agent_id") or "").strip()
             target_agent_id = str(target.get("agent_id") or "").strip()
diff --git a/loopx/control_plane/work_items/autonomous_replan_obligation.py b/loopx/control_plane/work_items/autonomous_replan_obligation.py
index 86a4119df4..115394d381 100644
--- a/loopx/control_plane/work_items/autonomous_replan_obligation.py
+++ b/loopx/control_plane/work_items/autonomous_replan_obligation.py
@@ -6,15 +6,12 @@
 from collections.abc import Callable, Mapping
 from typing import Any
 
-from ..runtime.time import parse_timestamp
 from ..todos.contract import (
-    normalize_todo_claimed_by,
     normalize_todo_id,
-    normalize_todo_id_list,
     normalize_todo_replan_obligation_id,
 )
-from ..todos.resume_planning import project_todo_resume_planning
-from .progress_observation import replan_writeback_requirements, typed_progress_repeat_trigger
+from .progress_observation import replan_writeback_requirements
+from .replan_history import project_replan_history
 from .replan_settlement import (
     project_todo_lifecycle_settlement_reentry as project_todo_lifecycle_reentry_effect,
 )
@@ -258,7 +255,7 @@ def _single_public_agent_id(items: list[dict[str, Any]]) -> str | None:
     return next(iter(agent_ids)) if len(agent_ids) == 1 else None
 
 
-def run_history_agent_id(run: dict[str, Any]) -> str | None:
+def run_history_agent_id(run: Mapping[str, Any]) -> str | None:
     agent_id = str(run.get("agent_id") or "").strip()
     if agent_id:
         return agent_id
@@ -268,38 +265,7 @@ def run_history_agent_id(run: dict[str, Any]) -> str | None:
     return None
 
 
-def _latest_agent_run_history(
-    latest_runs: list[dict[str, Any]] | None,
-    *,
-    neutral_classifications: set[str],
-    agent_id: str | None = None,
-) -> list[dict[str, Any]]:
-    """Keep the newest attributable agent lane while preserving goal-level runs."""
-
-    accountable_agent_id = str(agent_id or "").strip() or None
-    if accountable_agent_id is None:
-        accountable_agent_id = next(
-            (
-                run_history_agent_id(run)
-                for run in latest_runs or []
-                if isinstance(run, dict)
-                and str(run.get("classification") or "").strip()
-                not in neutral_classifications
-                and run_history_agent_id(run)
-            ),
-            None,
-        )
-    if not accountable_agent_id:
-        return [run for run in latest_runs or [] if isinstance(run, dict)]
-    return [
-        run
-        for run in latest_runs or []
-        if isinstance(run, dict)
-        and run_history_agent_id(run) in {None, accountable_agent_id}
-    ]
-
-
-def run_history_monitor_target(run: dict[str, Any]) -> dict[str, Any] | None:
+def run_history_monitor_target(run: Mapping[str, Any]) -> dict[str, Any] | None:
     target = run.get("monitor_target")
     if isinstance(target, dict):
         return target
@@ -343,173 +309,14 @@ def autonomous_replan_periodic_review_from_runs(
     periodic_run_threshold: int,
     build_autonomous_replan_obligation: Callable[..., dict[str, Any] | None],
 ) -> dict[str, Any] | None:
-    durable_runs: list[dict[str, Any]] = []
-    for run in latest_runs or []:
-        if not isinstance(run, dict):
-            continue
-        if autonomous_replan_ack_recorded(run):
-            break
-        classification = str(run.get("classification") or "").strip()
-        if not classification:
-            continue
-        if classification in neutral_classifications:
-            continue
-        durable_runs.append(run)
-        if len(durable_runs) >= periodic_run_threshold:
-            break
-
-    if len(durable_runs) < periodic_run_threshold:
-        return None
-
-    evidence: list[dict[str, Any]] = [
-        {
-            "kind": "periodic_review_due",
-            "section": "run_history",
-            "text": (
-                f"latest {len(durable_runs)} durable public run records since last autonomous "
-                f"replan reached periodic review threshold {periodic_run_threshold}"
-            ),
-            "run_count": len(durable_runs),
-            "threshold": periodic_run_threshold,
-            "latest_generated_at": str(durable_runs[0].get("generated_at") or ""),
-            "oldest_counted_generated_at": str(durable_runs[-1].get("generated_at") or ""),
-            "agent_id": _single_public_agent_id(durable_runs),
-        }
-    ]
-    return build_autonomous_replan_obligation(evidence, agent_todos=agent_todos)
-
-
-def _monitor_no_change_evidence(
-    agent_todos: dict[str, Any] | None,
-    *,
-    threshold: int,
-    schema_version: str,
-) -> dict[str, Any] | None:
-    if not isinstance(agent_todos, dict):
-        return None
-    raw_monitors = agent_todos.get("monitor_open_items")
-    monitors = [
-        item
-        for item in (raw_monitors if isinstance(raw_monitors, list) else [])
-        if isinstance(item, dict)
-    ]
-    stalled: list[tuple[int, dict[str, Any]]] = []
-    for item in monitors:
-        try:
-            no_change_count = int(str(item.get("consecutive_no_change") or "0"))
-        except ValueError:
-            continue
-        if no_change_count >= threshold:
-            stalled.append((no_change_count, item))
-    if not stalled:
-        return None
-
-    stalled.sort(key=lambda pair: pair[0], reverse=True)
-    no_change_count, monitor = stalled[0]
-    agent_id = str(monitor.get("claimed_by") or "").strip() or None
-    raw_advancements = agent_todos.get("executable_backlog_items")
-    if not isinstance(raw_advancements, list):
-        raw_advancements = agent_todos.get("items")
-    for item in raw_advancements or []:
-        if not isinstance(item, dict):
-            continue
-        if str(item.get("status") or "").strip().lower() != "open":
-            continue
-        if str(item.get("task_class") or "").strip() != "advancement_task":
-            continue
-        claimed_by = str(item.get("claimed_by") or "").strip() or None
-        if not claimed_by or claimed_by == agent_id:
-            return None
-
-    monitor_target_id = str(
-        monitor.get("target_key") or monitor.get("todo_id") or "monitor"
-    ).strip()
-    return {
-        "kind": "monitor_no_change_streak",
-        "schema_version": schema_version,
-        "section": "agent_todos",
-        "text": (
-            f"monitor {monitor_target_id} recorded {no_change_count} "
-            "consecutive unchanged polls without runnable advancement"
-        ),
-        "run_count": no_change_count,
-        "threshold": threshold,
-        "monitor_target_id": monitor_target_id,
-        "agent_id": agent_id,
-    }
-
-
-def _future_due_blocking_monitor(
-    agent_todos: dict[str, Any] | None,
-    *,
-    latest_generated_at: str,
-    agent_id: str | None,
-) -> dict[str, str] | None:
-    if not isinstance(agent_todos, dict):
-        return None
-    observed_at = parse_timestamp(latest_generated_at)
-    if observed_at is None:
-        return None
-
-    accountable_agent_id = normalize_todo_claimed_by(agent_id)
-    if not accountable_agent_id:
-        return None
-    raw_monitors = agent_todos.get("monitor_open_items")
-    monitor_items = raw_monitors if isinstance(raw_monitors, list) else []
-    monitors_by_id: dict[str, dict[str, Any]] = {}
-    for monitor in monitor_items:
-        if not isinstance(monitor, dict):
-            continue
-        monitor_todo_id = normalize_todo_id(monitor.get("todo_id"))
-        claimed_by = normalize_todo_claimed_by(monitor.get("claimed_by"))
-        if not monitor_todo_id:
-            continue
-        if accountable_agent_id and claimed_by and claimed_by != accountable_agent_id:
-            continue
-        monitors_by_id[monitor_todo_id] = monitor
-
-    blocking_monitor_ids: set[str] = set()
-    resume_planning = project_todo_resume_planning(agent_todos)
-    blocked_items = [
-        *resume_planning["monitor_blocked_items"],
-        *resume_planning["deferred_items"],
-    ]
-    for item in blocked_items:
-        claimed_by = normalize_todo_claimed_by(item.get("claimed_by"))
-        if accountable_agent_id and claimed_by and claimed_by != accountable_agent_id:
-            continue
-        raw_condition = item.get("resume_condition")
-        condition = raw_condition if isinstance(raw_condition, dict) else {}
-        monitor_todo_id = normalize_todo_id(
-            item.get("blocking_monitor_todo_id")
-            or condition.get("target_todo_id")
-            or condition.get("target")
-        )
-        if monitor_todo_id in monitors_by_id:
-            blocking_monitor_ids.add(monitor_todo_id)
-        for successor_todo_id in normalize_todo_id_list(
-            item.get("successor_todo_ids")
-        ):
-            if successor_todo_id in monitors_by_id:
-                blocking_monitor_ids.add(successor_todo_id)
-    if not blocking_monitor_ids:
-        return None
-
-    for monitor_todo_id in blocking_monitor_ids:
-        monitor = monitors_by_id[monitor_todo_id]
-        next_due_at = parse_timestamp(monitor.get("next_due_at"))
-        if next_due_at is None or next_due_at <= observed_at:
-            continue
-        expires_at = parse_timestamp(monitor.get("expires_at"))
-        if expires_at is not None and (
-            expires_at <= observed_at or expires_at <= next_due_at
-        ):
-            continue
-        return {
-            "todo_id": monitor_todo_id,
-            "next_due_at": str(monitor.get("next_due_at") or ""),
-        }
-    return None
+    trigger = project_replan_history(
+        latest_runs or [], operation="periodic",
+        ack_recorded=autonomous_replan_ack_recorded,
+        neutral_classifications=neutral_classifications,
+        periodic_threshold=periodic_run_threshold,
+    )
+    return (build_autonomous_replan_obligation([trigger], agent_todos=agent_todos)
+            if trigger else None)
 
 
 def build_autonomous_replan_obligation(
@@ -523,10 +330,10 @@ def build_autonomous_replan_obligation(
     dead_monitor_repeat_schema_version: str,
 ) -> dict[str, Any] | None:
     if not evidence:
-        monitor_evidence = _monitor_no_change_evidence(
-            agent_todos,
-            threshold=MONITOR_NO_CHANGE_STREAK_THRESHOLD,
-            schema_version=dead_monitor_repeat_schema_version,
+        monitor_evidence = project_replan_history(
+            operation="monitor_streak", agent_todos=agent_todos,
+            streak_threshold=MONITOR_NO_CHANGE_STREAK_THRESHOLD,
+            monitor_schema=dead_monitor_repeat_schema_version,
         )
         if monitor_evidence:
             evidence = [monitor_evidence]
@@ -775,169 +582,14 @@ def autonomous_replan_obligation_from_runs(
     dead_monitor_repeat_schema_version: str,
     periodic_run_threshold: int,
 ) -> dict[str, Any] | None:
-    scoped_latest_runs = _latest_agent_run_history(
-        latest_runs,
+    trigger = project_replan_history(
+        latest_runs or [], agent_todos=agent_todos, agent_id=agent_id,
+        ack_recorded=autonomous_replan_ack_recorded,
         neutral_classifications=neutral_classifications,
-        agent_id=agent_id,
+        stall_threshold=autonomous_replan_stall_threshold,
+        monitor_threshold=dead_monitor_repeat_threshold,
+        monitor_schema=dead_monitor_repeat_schema_version,
+        periodic_threshold=periodic_run_threshold,
     )
-
-    def periodic_review() -> dict[str, Any] | None:
-        return autonomous_replan_periodic_review_from_runs(
-            scoped_latest_runs,
-            agent_todos=agent_todos,
-            autonomous_replan_ack_recorded=autonomous_replan_ack_recorded,
-            neutral_classifications=neutral_classifications,
-            periodic_run_threshold=periodic_run_threshold,
-            build_autonomous_replan_obligation=build_autonomous_replan_obligation,
-        )
-
-    typed_repeat = typed_progress_repeat_trigger(
-        scoped_latest_runs,
-        agent_id=agent_id,
-        threshold=autonomous_replan_stall_threshold,
-    )
-    if typed_repeat:
-        return build_autonomous_replan_obligation(
-            [typed_repeat],
-            agent_todos=agent_todos,
-        )
-
-    # Monitor rows already carry a typed monitor target. Keep this explicit
-    # state-machine input; do not infer monitor/stall state from prose fields.
-    monitor_signals: list[dict[str, Any]] = []
-    signal_scan_limit = max(
-        autonomous_replan_stall_threshold,
-        dead_monitor_repeat_threshold,
-    )
-    for run in scoped_latest_runs:
-        if not isinstance(run, dict):
-            continue
-        if autonomous_replan_ack_recorded(run):
-            break
-        classification = str(run.get("classification") or "").strip()
-        if classification in neutral_classifications:
-            continue
-        if classification != "quota_monitor_poll":
-            break
-        monitor_target = run_history_monitor_target(run)
-        if not isinstance(monitor_target, dict):
-            break
-        monitor_target_id = str(monitor_target.get("target_id") or "").strip()
-        monitor_mode = str(monitor_target.get("monitor_mode") or "").strip()
-        if not monitor_target_id or not monitor_mode:
-            break
-        signal: dict[str, Any] = {
-            "classification": classification,
-            "generated_at": str(run.get("generated_at") or ""),
-            "agent_id": run_history_agent_id(run),
-            "monitor_target_id": monitor_target_id,
-            "monitor_target": {
-                key: monitor_target.get(key)
-                for key in (
-                    "schema_version",
-                    "target_id",
-                    "monitor_mode",
-                    "effective_action",
-                    "agent_id",
-                    "frontier_identity",
-                )
-                if monitor_target.get(key)
-            },
-        }
-        monitor_event = run.get("monitor_event")
-        if not isinstance(monitor_event, dict):
-            monitor_event = {}
-        todo_id = str(run.get("todo_id") or monitor_event.get("todo_id") or "").strip()
-        target_key = str(
-            run.get("target_key") or monitor_event.get("target_key") or ""
-        ).strip()
-        if todo_id:
-            signal["todo_id"] = todo_id
-        if target_key:
-            signal["target_key"] = target_key
-        turn_instance_id = str(run.get("turn_instance_id") or "").strip()
-        if turn_instance_id:
-            signal["turn_instance_id"] = turn_instance_id
-        monitor_signals.append(signal)
-        if len(monitor_signals) >= signal_scan_limit:
-            break
-
-    if len(monitor_signals) < autonomous_replan_stall_threshold:
-        return periodic_review()
-
-    blocked_successor_signals = monitor_signals[:autonomous_replan_stall_threshold]
-    blocked_successor_modes = {
-        str((signal.get("monitor_target") or {}).get("monitor_mode") or "")
-        for signal in blocked_successor_signals
-    }
-    if blocked_successor_modes == {
-        "blocked_successor_wait_without_material_transition"
-    }:
-        signal_agent_id = _single_public_agent_id(blocked_successor_signals)
-        if _future_due_blocking_monitor(
-            agent_todos,
-            latest_generated_at=str(monitor_signals[0].get("generated_at") or ""),
-            agent_id=signal_agent_id or agent_id,
-        ):
-            return periodic_review()
-        monitor_target_ids = {
-            str(signal.get("monitor_target_id") or "")
-            for signal in blocked_successor_signals
-            if signal.get("monitor_target_id")
-        }
-        frontier_identities = {
-            str((signal.get("monitor_target") or {}).get("frontier_identity") or "")
-            for signal in blocked_successor_signals
-            if (signal.get("monitor_target") or {}).get("frontier_identity")
-        }
-        if len(monitor_target_ids) != 1 or len(frontier_identities) != 1:
-            return periodic_review()
-        evidence = [
-            {
-                "kind": "blocked_successor_no_progress_repeat",
-                "section": "run_history",
-                "run_count": len(blocked_successor_signals),
-                "threshold": autonomous_replan_stall_threshold,
-                "monitor_target_id": next(iter(monitor_target_ids)),
-                "frontier_identity": next(iter(frontier_identities)),
-                "latest_generated_at": monitor_signals[0].get("generated_at"),
-                "agent_id": _single_public_agent_id(blocked_successor_signals),
-            }
-        ]
-        return build_autonomous_replan_obligation(evidence, agent_todos=agent_todos)
-
-    executed_monitor_signals = [
-        signal
-        for signal in monitor_signals
-        if (
-            (signal.get("todo_id") or signal.get("target_key"))
-            and str((signal.get("monitor_target") or {}).get("monitor_mode") or "")
-            in {
-                "due_monitor_observed_without_material_transition",
-                "external_monitor_observed_without_material_transition",
-            }
-        )
-    ]
-    repeated_monitors = executed_monitor_signals[:dead_monitor_repeat_threshold]
-    if len(repeated_monitors) < dead_monitor_repeat_threshold:
-        return periodic_review()
-    monitor_target_ids = {
-        str(signal.get("monitor_target_id") or "")
-        for signal in repeated_monitors
-        if signal.get("monitor_target_id")
-    }
-    if len(monitor_target_ids) != 1:
-        return periodic_review()
-    evidence = [
-        {
-            "kind": "dead_monitor_repeat",
-            "schema_version": dead_monitor_repeat_schema_version,
-            "section": "run_history",
-            "run_count": len(repeated_monitors),
-            "threshold": dead_monitor_repeat_threshold,
-            "monitor_target_id": next(iter(monitor_target_ids)),
-            "latest_generated_at": repeated_monitors[0].get("generated_at"),
-            "agent_id": _single_public_agent_id(repeated_monitors),
-        }
-    ]
-    return build_autonomous_replan_obligation(evidence, agent_todos=agent_todos)
+    return (build_autonomous_replan_obligation([trigger], agent_todos=agent_todos)
+            if trigger else None)
diff --git a/loopx/control_plane/work_items/progress_observation.py b/loopx/control_plane/work_items/progress_observation.py
index 3afd4d115d..4e25c6e123 100644
--- a/loopx/control_plane/work_items/progress_observation.py
+++ b/loopx/control_plane/work_items/progress_observation.py
@@ -193,54 +193,13 @@ def typed_progress_repeat_trigger(
     agent_id: str | None,
     threshold: int = PROGRESS_REPEAT_THRESHOLD,
 ) -> dict[str, Any] | None:
-    """Return a repeat trigger only for consecutive equivalent typed rows."""
+    """Compatibility entrypoint for the shared TypeScript history window."""
+    from .replan_history import project_replan_history
 
-    required_count = max(2, int(threshold))
-    normalized_agent_id = str(agent_id or "").strip()
-    observations: list[tuple[Mapping[str, Any], dict[str, Any]]] = []
-    observed_turn_instance_ids: set[str] = set()
-    for run in newest_first_runs:
-        run_agent_id = str(run.get("agent_id") or "").strip()
-        if normalized_agent_id and run_agent_id not in {"", normalized_agent_id}:
-            continue
-        turn_instance_id = _progress_turn_instance_id(run)
-        if turn_instance_id and turn_instance_id in observed_turn_instance_ids:
-            continue
-        observation = progress_observation_from_run(run)
-        if observation is None:
-            if observations:
-                break
-            continue
-        observations.append((run, observation))
-        if turn_instance_id:
-            observed_turn_instance_ids.add(turn_instance_id)
-        if len(observations) >= required_count:
-            break
-    if len(observations) < required_count:
-        return None
-    fingerprints = {item[1]["fingerprint"] for item in observations}
-    if len(fingerprints) != 1:
-        return None
-    result_class = observations[0][1]["result_class"]
-    if result_class not in {
-        ProgressResultClass.UNCHANGED.value,
-        ProgressResultClass.BLOCKED.value,
-    }:
-        return None
-    baseline = observations[0][1]
-    return {
-        "kind": PROGRESS_REPEAT_TRIGGER_KIND,
-        "schema_version": PROGRESS_OBSERVATION_SCHEMA_VERSION,
-        "agent_id": normalized_agent_id or None,
-        "run_count": required_count,
-        "threshold": required_count,
-        "progress_fingerprint": baseline["fingerprint"],
-        "progress_baseline": baseline,
-        "latest_generated_at": str(observations[0][0].get("generated_at") or ""),
-        "oldest_counted_generated_at": str(
-            observations[-1][0].get("generated_at") or ""
-        ),
-    }
+    return project_replan_history(
+        newest_first_runs, operation="progress", agent_id=agent_id,
+        stall_threshold=threshold,
+    )
 
 
 def _has_new_terminal_coverage(
diff --git a/loopx/control_plane/work_items/replan_history.py b/loopx/control_plane/work_items/replan_history.py
new file mode 100644
index 0000000000..93faa65b1c
--- /dev/null
+++ b/loopx/control_plane/work_items/replan_history.py
@@ -0,0 +1,145 @@
+"""Historical codecs for the TypeScript-owned replan history policy.
+
+The wire input contains consumed facts, never full run records or Todo prose.
+Persisted observation fingerprints and legacy timestamp parsing keep their
+existing Python codecs; trigger selection and history windows have one owner.
+"""
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable, Mapping
+from typing import Any, Literal
+
+from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
+from ..runtime.time import parse_timestamp
+from ..todos.contract import normalize_todo_claimed_by, normalize_todo_id, normalize_todo_id_list
+from ..todos.resume_planning import build_todo_resume_planning_request
+
+REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS = {
+    "quota_slot_spent", "quota_slot_voided", "delivery_completion_spend_accounted_v0",
+}
+
+
+def _timestamp(value: Any) -> float | None:
+    parsed = parse_timestamp(value)
+    return parsed.timestamp() if parsed is not None else None
+
+
+def _run_fact(run: Mapping[str, Any], ack_recorded: Callable[..., bool]) -> dict[str, Any]:
+    # Lazy imports keep the existing public codecs callable without an import cycle.
+    from .autonomous_replan_obligation import run_history_agent_id, run_history_monitor_target
+    from .progress_observation import _progress_turn_instance_id, progress_observation_from_run
+
+    target = run_history_monitor_target(run) or {}
+    event = run.get("monitor_event")
+    event = event if isinstance(event, dict) else {}
+    return {
+        "agent_id": run_history_agent_id(run),
+        "monitor_agent_id": normalize_todo_claimed_by(run_history_agent_id(run)),
+        "public_agent_id": str(run.get("agent_id") or "").strip() or None,
+        "classification": str(run.get("classification") or "").strip(),
+        "generated_at": str(run.get("generated_at") or ""),
+        "observed_at": _timestamp(run.get("generated_at")),
+        "turn_id": _progress_turn_instance_id(run),
+        "accepted_ack": ack_recorded(run),
+        "progress": progress_observation_from_run(run),
+        "monitor": {
+            "target_id": str(target.get("target_id") or "").strip() or None,
+            "mode": str(target.get("monitor_mode") or "").strip() or None,
+            "frontier": str(target.get("frontier_identity") or "") or None,
+            "todo_id": str(run.get("todo_id") or event.get("todo_id") or "").strip() or None,
+            "target_key": str(run.get("target_key") or event.get("target_key") or "").strip() or None,
+        },
+    }
+
+
+def _todo_facts(value: Any, *, include_resume: bool) -> dict[str, Any]:
+    value = value if isinstance(value, dict) else {}
+    raw_monitors = value.get("monitor_open_items")
+    monitors = []
+    for item in raw_monitors if isinstance(raw_monitors, list) else []:
+        if not isinstance(item, dict):
+            continue
+        try:
+            count = int(str(item.get("consecutive_no_change") or "0"))
+        except ValueError:
+            count = 0
+        monitors.append({
+            "id": normalize_todo_id(item.get("todo_id")),
+            "claim": normalize_todo_claimed_by(item.get("claimed_by")),
+            "public_claim": str(item.get("claimed_by") or "").strip() or None,
+            "target": str(item.get("target_key") or item.get("todo_id") or "monitor").strip(),
+            "no_change_count": count,
+            "due_at": _timestamp(item.get("next_due_at")),
+            "expires_at": _timestamp(item.get("expires_at")),
+        })
+    raw_advancements = value.get("executable_backlog_items")
+    if not isinstance(raw_advancements, list):
+        raw_advancements = value.get("items")
+    advancements = [{
+        "status": str(item.get("status") or "").strip().lower(),
+        "task_class": str(item.get("task_class") or "").strip(),
+        "claim": str(item.get("claimed_by") or "").strip() or None,
+    } for item in (raw_advancements or []) if isinstance(item, dict)]
+    # The existing resume codec is shared. TS composes the planner in-process;
+    # Python must not make a second RPC or decide which monitor blocks replan.
+    resume = build_todo_resume_planning_request(value) if include_resume else None
+    if resume:
+        for source in resume["sources"].values():
+            for entry in source:
+                payload = entry["payload"]
+                # Resume planning tests truthiness of text, not its content.
+                payload["text"] = "item" if payload.get("text") else ""
+                payload["claimed_by"] = normalize_todo_claimed_by(payload.get("claimed_by"))
+                condition = payload.get("resume_condition")
+                condition = condition if isinstance(condition, dict) else {}
+                payload["blocking_monitor_todo_id"] = normalize_todo_id(
+                    payload.get("blocking_monitor_todo_id")
+                    or condition.get("target_todo_id") or condition.get("target"))
+                payload["successor_todo_ids"] = normalize_todo_id_list(payload.get("successor_todo_ids"))
+                entry["payload"] = {key: payload[key] for key in (
+                    "index", "text", "task_class", "todo_id", "claimed_by",
+                    "resume_when", "resume_ready", "resume_condition",
+                    "blocking_monitor_todo_id", "successor_todo_ids",
+                ) if key in payload}
+    return {"monitors": monitors, "advancements": advancements, "resume": resume}
+
+
+def project_replan_history(
+    runs: Iterable[Mapping[str, Any]] = (), *,
+    operation: Literal["all", "progress", "periodic", "monitor_streak"] = "all",
+    agent_id: str | None = None, agent_todos: Any = None,
+    ack_recorded: Callable[..., bool] | None = None,
+    neutral_classifications: set[str] | None = None,
+    stall_threshold: int = 2, periodic_threshold: int = 20,
+    monitor_threshold: int = 6, streak_threshold: int = 5,
+    monitor_schema: str = "dead_monitor_repeat_v0",
+) -> dict[str, Any] | None:
+    if ack_recorded is None:
+        from .autonomous_replan_ack import autonomous_replan_ack_recorded
+        ack_recorded = autonomous_replan_ack_recorded
+    facts = [_run_fact(row, ack_recorded) for row in runs if isinstance(row, Mapping)]
+    needs_resume = operation == "all" and any(
+        row["monitor"]["mode"] == "blocked_successor_wait_without_material_transition"
+        for row in facts)
+    params = {
+        "schema_version": "replan_history_request_v0", "operation": operation,
+        "runs": facts,
+        "agent_id": str(agent_id or "").strip() or None,
+        "monitor_agent_id": normalize_todo_claimed_by(agent_id),
+        "neutral_classifications": sorted(REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS
+            if neutral_classifications is None else neutral_classifications),
+        "stall_threshold": max(2, int(stall_threshold)),
+        "periodic_threshold": periodic_threshold, "monitor_threshold": monitor_threshold,
+        "streak_threshold": streak_threshold, "monitor_schema": monitor_schema,
+        "todos": _todo_facts(agent_todos, include_resume=needs_resume),
+    }
+    try:
+        result = effect_runtime_result("work_item.replan_history.project", params)
+    except EffectRuntimeRejected as exc:
+        raise ValueError(str(exc)) from None
+    if not isinstance(result, dict) or result.get("schema_version") != "replan_history_result_v0":
+        raise RuntimeError("TypeScript replan history shape mismatch")
+    trigger = result.get("trigger")
+    if trigger is not None and not isinstance(trigger, dict):
+        raise RuntimeError("TypeScript replan history trigger mismatch")
+    return trigger
diff --git a/loopx/control_plane/work_items/replan_history.ts b/loopx/control_plane/work_items/replan_history.ts
new file mode 100644
index 0000000000..87481ce3d7
--- /dev/null
+++ b/loopx/control_plane/work_items/replan_history.ts
@@ -0,0 +1,304 @@
+/** Deterministic history-to-replan policy. No IO, writes, or model authority. */
+import type { JsonObject } from "../effect_program.ts";
+import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
+import {
+  jsonObject, requireJsonObject, requireBoolean, requireNonEmptyString,
+  requireStringArray, requireStringLiteral, optionalNonEmptyString,
+} from "../runtime_decode.ts";
+import { projectTodoResumePlanning } from "../todos/resume_planning.ts";
+
+interface Progress {
+  fingerprint: string;
+  result: "advanced" | "unchanged" | "blocked" | "exploration_exhausted" | "no_followup";
+  payload: JsonObject;
+}
+interface Monitor {
+  target: string | null;
+  mode: string | null;
+  frontier: string | null;
+  todo: string | null;
+  key: string | null;
+}
+interface Run {
+  agent: string | null;
+  publicAgent: string | null;
+  monitorAgent: string | null;
+  classification: string;
+  generatedAt: string;
+  observedAt: number | null;
+  turn: string | null;
+  ack: boolean;
+  progress: Progress | null;
+  monitor: Monitor;
+}
+interface MonitorTodo {
+  id: string | null;
+  claim: string | null;
+  publicClaim: string | null;
+  target: string;
+  count: number;
+  due: number | null;
+  expiry: number | null;
+}
+interface Todos {
+  monitors: readonly MonitorTodo[];
+  advancements: readonly { status: string; taskClass: string; claim: string | null }[];
+  resume: JsonObject | null;
+}
+interface Request {
+  operation: "all" | "progress" | "periodic" | "monitor_streak";
+  agent: string | null;
+  monitorAgent: string | null;
+  runs: readonly Run[];
+  neutral: ReadonlySet<string>;
+  stall: number;
+  periodic: number;
+  monitor: number;
+  streak: number;
+  monitorSchema: string;
+  todos: Todos;
+}
+
+type Trigger = { run_count: number; threshold: number; agent_id: string | null } & (
+  | { kind: "typed_progress_repeat"; schema_version: "typed_progress_observation_v0";
+      progress_baseline: JsonObject; progress_fingerprint: string;
+      latest_generated_at: string; oldest_counted_generated_at: string }
+  | { kind: "periodic_review_due"; section: "run_history"; text: string;
+      latest_generated_at: string; oldest_counted_generated_at: string }
+  | { kind: "blocked_successor_no_progress_repeat"; section: "run_history";
+      frontier_identity: string; monitor_target_id: string; latest_generated_at: string }
+  | { kind: "dead_monitor_repeat"; schema_version: string; section: "run_history";
+      monitor_target_id: string; latest_generated_at: string }
+  | { kind: "monitor_no_change_streak"; schema_version: string; section: "agent_todos";
+      monitor_target_id: string; text: string }
+);
+
+function text(value: unknown, label: string): string {
+  if (typeof value !== "string") throw new EffectRuntimeRequestError(`${label} must be a string`);
+  return value;
+}
+function integer(value: unknown, label: string, minimum: number): number {
+  if (typeof value !== "number" || !Number.isSafeInteger(value) || value < minimum) {
+    throw new EffectRuntimeRequestError(`${label} must be a safe integer >= ${minimum}`);
+  }
+  return value;
+}
+function time(value: unknown, label: string): number | null {
+  if (value === null) return null;
+  if (typeof value !== "number" || !Number.isFinite(value)) {
+    throw new EffectRuntimeRequestError(`${label} must be a finite timestamp or null`);
+  }
+  return value;
+}
+function rows<T>(value: unknown, label: string, decode: (row: JsonObject) => T): T[] {
+  if (!Array.isArray(value)) throw new EffectRuntimeRequestError(`${label} must be an array`);
+  return value.map(row => decode(requireJsonObject(row, label)));
+}
+function decodeProgress(value: unknown): Progress | null {
+  if (value === null) return null;
+  const payload = requireJsonObject(value, "progress");
+  requireStringLiteral(payload.schema_version, ["typed_progress_observation_v0"], "progress schema");
+  return {
+    payload, fingerprint: requireNonEmptyString(payload.fingerprint, "fingerprint"),
+    result: requireStringLiteral(payload.result_class,
+      ["advanced", "unchanged", "blocked", "exploration_exhausted", "no_followup"], "result_class"),
+  };
+}
+function decode(value: unknown): Request {
+  const raw = requireJsonObject(value, "replan history");
+  requireStringLiteral(raw.schema_version, ["replan_history_request_v0"], "replan history schema");
+  const todo = requireJsonObject(raw.todos, "todos");
+  return {
+    operation: requireStringLiteral(raw.operation, ["all", "progress", "periodic", "monitor_streak"], "operation"),
+    agent: optionalNonEmptyString(raw.agent_id, "agent_id"),
+    monitorAgent: optionalNonEmptyString(raw.monitor_agent_id, "monitor_agent_id"),
+    neutral: new Set(requireStringArray(raw.neutral_classifications, "neutral_classifications")),
+    stall: integer(raw.stall_threshold, "stall_threshold", 2),
+    periodic: integer(raw.periodic_threshold, "periodic_threshold", 1),
+    monitor: integer(raw.monitor_threshold, "monitor_threshold", 1),
+    streak: integer(raw.streak_threshold, "streak_threshold", 1),
+    monitorSchema: requireNonEmptyString(raw.monitor_schema, "monitor_schema"),
+    runs: rows(raw.runs, "runs", row => {
+      const monitor = requireJsonObject(row.monitor, "monitor");
+      return {
+        agent: optionalNonEmptyString(row.agent_id, "run.agent_id"),
+        publicAgent: optionalNonEmptyString(row.public_agent_id, "public_agent_id"),
+        monitorAgent: optionalNonEmptyString(row.monitor_agent_id, "run.monitor_agent_id"),
+        classification: text(row.classification, "classification"),
+        generatedAt: text(row.generated_at, "generated_at"), observedAt: time(row.observed_at, "observed_at"),
+        turn: optionalNonEmptyString(row.turn_id, "turn_id"), ack: requireBoolean(row.accepted_ack, "accepted_ack"),
+        progress: decodeProgress(row.progress),
+        monitor: {
+          target: optionalNonEmptyString(monitor.target_id, "target_id"),
+          mode: optionalNonEmptyString(monitor.mode, "mode"),
+          frontier: optionalNonEmptyString(monitor.frontier, "frontier"),
+          todo: optionalNonEmptyString(monitor.todo_id, "todo_id"),
+          key: optionalNonEmptyString(monitor.target_key, "target_key"),
+        },
+      };
+    }),
+    todos: {
+      resume: todo.resume === null ? null : requireJsonObject(todo.resume, "resume"),
+      monitors: rows(todo.monitors, "monitors", row => ({
+        id: optionalNonEmptyString(row.id, "monitor.id"), claim: optionalNonEmptyString(row.claim, "claim"),
+        publicClaim: optionalNonEmptyString(row.public_claim, "public_claim"),
+        target: text(row.target, "monitor.target"),
+        count: integer(row.no_change_count, "no_change_count", Number.MIN_SAFE_INTEGER),
+        due: time(row.due_at, "due_at"), expiry: time(row.expires_at, "expires_at"),
+      })),
+      advancements: rows(todo.advancements, "advancements", row => ({
+        status: text(row.status, "status"), taskClass: text(row.task_class, "task_class"),
+        claim: optionalNonEmptyString(row.claim, "claim"),
+      })),
+    },
+  };
+}
+
+/** Scope before ACK: a peer's acknowledgement cannot discharge this lane. */
+function historyWindow(request: Request): Run[] {
+  const agent = request.agent ?? (request.operation === "all"
+    ? request.runs.find(run => !request.neutral.has(run.classification) && run.agent)?.agent : null);
+  const result: Run[] = [];
+  for (const run of request.runs) {
+    if (agent && run.agent && run.agent !== agent) continue;
+    if (run.ack) break;
+    if (request.neutral.has(run.classification)) continue;
+    result.push(run);
+  }
+  return result;
+}
+
+/** De-duplicate accepted evidence, not arbitrary records sharing its turn.
+ * A leading untyped row cannot erase an older typed observation from that turn.
+ * Null legacy identities remain independent; neutral rows were already removed.
+ */
+function* distinctTurns(runs: readonly Run[], counts: (run: Run) => boolean = () => true): Generator<Run, void, unknown> {
+  const seen = new Set<string>();
+  for (const run of runs) {
+    const key = run.turn ? JSON.stringify([run.agent, run.turn]) : null;
+    if (key && seen.has(key)) continue;
+    if (key && counts(run)) seen.add(key);
+    yield run;
+  }
+}
+function sole(values: readonly (string | null)[]): string | null {
+  const unique = new Set(values.filter((value): value is string => value !== null));
+  return unique.size === 1 ? [...unique][0]! : null;
+}
+function progressTrigger(runs: readonly Run[], request: Request): Trigger | null {
+  const observed: (Run & { progress: Progress })[] = [];
+  for (const run of distinctTurns(runs, row => row.progress !== null)) {
+    if (run.progress === null) {
+      if (observed.length) break;
+      continue;
+    }
+    observed.push({ ...run, progress: run.progress });
+    if (observed.length >= request.stall) break;
+  }
+  if (observed.length < request.stall) return null;
+  const first = observed[0]!;
+  if (!sole(observed.map(row => row.progress.fingerprint)) ||
+      !["unchanged", "blocked"].includes(first.progress.result)) return null;
+  return {
+    kind: "typed_progress_repeat", schema_version: "typed_progress_observation_v0",
+    agent_id: request.agent, run_count: request.stall, threshold: request.stall,
+    progress_fingerprint: first.progress.fingerprint, progress_baseline: first.progress.payload,
+    latest_generated_at: first.generatedAt, oldest_counted_generated_at: observed.at(-1)!.generatedAt,
+  };
+}
+function periodicTrigger(runs: readonly Run[], request: Request): Trigger | null {
+  const durable: Run[] = [];
+  for (const run of distinctTurns(runs.filter(row => row.classification))) {
+    durable.push(run);
+    if (durable.length >= request.periodic) break;
+  }
+  if (durable.length < request.periodic) return null;
+  return {
+    kind: "periodic_review_due", section: "run_history",
+    text: `latest ${durable.length} durable public run records since last autonomous replan reached periodic review threshold ${request.periodic}`,
+    run_count: durable.length, threshold: request.periodic,
+    latest_generated_at: durable[0]!.generatedAt, oldest_counted_generated_at: durable.at(-1)!.generatedAt,
+    agent_id: sole(durable.map(run => run.publicAgent)),
+  };
+}
+function futureBlockingMonitor(request: Request, signals: readonly Run[]): boolean {
+  const observedAt = signals[0]!.observedAt;
+  const agent = sole(signals.map(run => run.monitorAgent)) ?? request.monitorAgent;
+  if (observedAt === null || !agent || !request.todos.resume) return false;
+  const monitors = new Map(request.todos.monitors.filter(row => row.id && (!row.claim || row.claim === agent))
+    .map(row => [row.id!, row]));
+  if (!monitors.size) return false;
+  // Compose the established planner here: no nested process/bridge call and no
+  // second Python copy of Todo resume or ownership rules.
+  const plan = projectTodoResumePlanning(request.todos.resume);
+  const items = [...plan.monitor_blocked_items as JsonObject[], ...plan.deferred_items as JsonObject[]];
+  for (const row of items) {
+    if (row.claimed_by && row.claimed_by !== agent) continue;
+    const condition = jsonObject(row.resume_condition);
+    const ids = [row.blocking_monitor_todo_id ?? condition?.target_todo_id ?? condition?.target,
+      ...(Array.isArray(row.successor_todo_ids) ? row.successor_todo_ids : [])];
+    for (const id of ids) {
+      const monitor = typeof id === "string" ? monitors.get(id) : undefined;
+      if (monitor?.due !== null && monitor?.due !== undefined && monitor.due > observedAt &&
+          (monitor.expiry === null || (monitor.expiry > observedAt && monitor.expiry > monitor.due))) return true;
+    }
+  }
+  return false;
+}
+function monitorTrigger(runs: readonly Run[], request: Request): Trigger | null {
+  const signals: Run[] = [];
+  for (const run of distinctTurns(runs)) {
+    if (run.classification !== "quota_monitor_poll" || !run.monitor.target || !run.monitor.mode) break;
+    signals.push(run);
+    if (signals.length >= Math.max(request.stall, request.monitor)) break;
+  }
+  if (signals.length < request.stall) return null;
+  const blocked = signals.slice(0, request.stall);
+  if (blocked.every(run => run.monitor.mode === "blocked_successor_wait_without_material_transition")) {
+    if (futureBlockingMonitor(request, blocked)) return null;
+    const target = sole(blocked.map(run => run.monitor.target));
+    const frontier = sole(blocked.map(run => run.monitor.frontier));
+    if (!target || !frontier) return null;
+    return {
+      kind: "blocked_successor_no_progress_repeat", section: "run_history",
+      run_count: blocked.length, threshold: request.stall, monitor_target_id: target,
+      frontier_identity: frontier, latest_generated_at: blocked[0]!.generatedAt,
+      agent_id: sole(blocked.map(run => run.agent)),
+    };
+  }
+  const repeated = signals.filter(run => (run.monitor.todo || run.monitor.key) &&
+    ["due_monitor_observed_without_material_transition", "external_monitor_observed_without_material_transition"]
+      .includes(run.monitor.mode!)).slice(0, request.monitor);
+  const target = sole(repeated.map(run => run.monitor.target));
+  if (repeated.length < request.monitor || !target) return null;
+  return {
+    kind: "dead_monitor_repeat", schema_version: request.monitorSchema, section: "run_history",
+    run_count: repeated.length, threshold: request.monitor, monitor_target_id: target,
+    latest_generated_at: repeated[0]!.generatedAt, agent_id: sole(repeated.map(run => run.agent)),
+  };
+}
+function monitorStreak(request: Request): Trigger | null {
+  const stalled = request.todos.monitors.filter(row => row.count >= request.streak)
+    .sort((a, b) => b.count - a.count)[0];
+  if (!stalled || request.todos.advancements.some(row => row.status === "open" &&
+      row.taskClass === "advancement_task" && (!row.claim || row.claim === stalled.publicClaim))) return null;
+  return {
+    kind: "monitor_no_change_streak", schema_version: request.monitorSchema, section: "agent_todos",
+    text: `monitor ${stalled.target} recorded ${stalled.count} consecutive unchanged polls without runnable advancement`,
+    run_count: stalled.count, threshold: request.streak,
+    monitor_target_id: stalled.target, agent_id: stalled.publicClaim,
+  };
+}
+
+export function projectReplanHistory(value: unknown): JsonObject {
+  const request = decode(value);
+  const runs = historyWindow(request);
+  let trigger: Trigger | null;
+  switch (request.operation) {
+    case "all": trigger = progressTrigger(runs, request) ?? monitorTrigger(runs, request) ?? periodicTrigger(runs, request); break;
+    case "progress": trigger = progressTrigger(runs, request); break;
+    case "periodic": trigger = periodicTrigger(runs, request); break;
+    case "monitor_streak": trigger = monitorStreak(request); break;
+  }
+  return { schema_version: "replan_history_result_v0", trigger };
+}
diff --git a/loopx/status.py b/loopx/status.py
index 9dec20bf9f..3bf05471ce 100644
--- a/loopx/status.py
+++ b/loopx/status.py
@@ -1,10 +1,15 @@
 from __future__ import annotations
 
+
 from collections.abc import Mapping, Sequence
 from pathlib import Path
 import re
 from typing import Any
 
+from .control_plane.work_items.replan_history import (
+    REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS as AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS,  # noqa: F401 - compatibility export
+)
+
 from .control_plane import compact_control_plane_policy
 from .control_plane.effect_runtime import effect_runtime_request_scope
 from .control_plane.status.collection import (
@@ -346,11 +351,6 @@
 )
 AUTONOMOUS_REPLAN_SCHEMA_VERSION = "autonomous_replan_obligation_v0"
 DEAD_MONITOR_REPEAT_SCHEMA_VERSION = "dead_monitor_repeat_v0"
-AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS = {
-    "quota_slot_spent",
-    "quota_slot_voided",
-    "delivery_completion_spend_accounted_v0",
-}
 
 
 
diff --git a/tests/control_plane/test_replan_history_policy.py b/tests/control_plane/test_replan_history_policy.py
new file mode 100644
index 0000000000..9a3b1c4113
--- /dev/null
+++ b/tests/control_plane/test_replan_history_policy.py
@@ -0,0 +1,146 @@
+"""Independent invariants for the public history-to-replan boundary."""
+
+from __future__ import annotations
+
+from loopx.status import autonomous_replan_obligation_from_runs
+from loopx.control_plane.work_items.progress_observation import typed_progress_repeat_trigger
+
+AGENT = "history-worker"
+
+
+def run(n: int, *, turn: str | None = None, agent: str = AGENT) -> dict:
+    value = {
+        "agent_id": agent,
+        "generated_at": f"2026-09-22T00:00:{n:02d}Z",
+        "classification": "bounded_delivery",
+    }
+    if turn is not None:
+        value["turn_instance_id"] = turn
+    return value
+
+
+def observation() -> dict:
+    return {"schema_version": "typed_progress_observation_v0",
+            "result_class": "unchanged", "surface_id": "surface-fixture",
+            "evidence_ids": ["evidence-fixture"]}
+
+
+def monitor(n: int, *, turn: str | None = None, agent: str = AGENT) -> dict:
+    return {**run(n, turn=turn, agent=agent), "classification": "quota_monitor_poll",
+            "todo_id": "todo_monitor_fixture",
+            "monitor_target": {"target_id": "target-fixture", "agent_id": agent,
+                               "monitor_mode": "due_monitor_observed_without_material_transition"}}
+
+
+def obligation(rows: list[dict]) -> dict | None:
+    return autonomous_replan_obligation_from_runs(rows, agent_todos=None, agent_id=AGENT)
+
+
+def test_periodic_review_counts_work_turns_not_retry_records() -> None:
+    assert obligation([run(n, turn="one-turn") for n in range(30, 0, -1)]) is None
+
+
+def test_monitor_repeat_counts_work_turns_not_retry_records() -> None:
+    assert obligation([monitor(n, turn="one-poll") for n in range(6, 0, -1)]) is None
+
+
+def test_progress_ack_ends_repeat_window() -> None:
+    ack = {**run(3), "autonomous_replan_ack": {
+        "recorded": True, "semantic_delta": {"accepted": True}}}
+    rows = [ack, {**run(2), "progress_observation": observation()},
+            {**run(1), "progress_observation": observation()}]
+    assert typed_progress_repeat_trigger(rows, agent_id=AGENT) is None
+    assert obligation(rows) is None
+
+
+def test_neutral_accounting_does_not_interrupt_equivalent_progress() -> None:
+    rows = [{**run(3), "progress_observation": observation()},
+            {**run(2), "classification": "quota_slot_voided"},
+            {**run(1), "progress_observation": observation()}]
+    result = obligation(rows)
+    assert result is not None
+    assert result["triggers"][0]["kind"] == "typed_progress_repeat"
+
+
+def test_peer_ack_does_not_discharge_current_lane() -> None:
+    rows = [{**run(3, agent="peer"), "autonomous_replan_ack": {
+        "recorded": True, "semantic_delta": {"accepted": True}}},
+        {**run(2), "progress_observation": observation()},
+        {**run(1), "progress_observation": observation()}]
+    assert obligation(rows) is not None
+
+
+def test_twenty_distinct_turns_retain_periodic_identity() -> None:
+    rows = [run(n, turn=f"turn-{n}") for n in range(20, 0, -1)]
+    first = obligation(rows)
+    assert first is not None
+    trigger = first["triggers"][0]
+    assert trigger["kind"] == "periodic_review_due"
+    assert trigger["run_count"] == 20
+    assert trigger["latest_generated_at"] == rows[0]["generated_at"]
+    assert trigger["oldest_counted_generated_at"] == rows[-1]["generated_at"]
+    assert obligation(rows) == first
+
+
+def test_six_distinct_polls_retain_dead_monitor_contract() -> None:
+    result = obligation([monitor(n, turn=f"poll-{n}") for n in range(6, 0, -1)])
+    assert result is not None
+    assert result["triggers"][0]["kind"] == "dead_monitor_repeat"
+    assert result["dead_monitor_detector"]["run_count"] == 6
+
+
+def test_genuinely_unknown_work_still_breaks_progress_streak() -> None:
+    rows = [{**run(3), "progress_observation": observation()}, run(2),
+            {**run(1), "progress_observation": observation()}]
+    assert obligation(rows) is None
+
+
+def test_ack_needs_accepted_semantic_evidence() -> None:
+    ack = {**run(3), "autonomous_replan_ack": {"recorded": True}}
+    rows = [ack, {**run(2), "progress_observation": observation()},
+            {**run(1), "progress_observation": observation()}]
+    assert obligation(rows) is not None
+
+
+def test_retry_identity_is_scoped_to_agent() -> None:
+    rows = [{**run(3, turn="shared", agent="peer"), "progress_observation": observation()},
+            {**run(2, turn="shared"), "progress_observation": observation()},
+            {**run(1, turn="prior"), "progress_observation": observation()}]
+    result = obligation(rows)
+    assert result is not None and result["agent_id"] == AGENT
+
+
+def test_future_monitor_expiry_must_cover_the_due_instant() -> None:
+    from test_goal_vision_blocked_successor import (
+        _blocked_wait_polls, _monitor_blocked_advancement_items, _quota_with_replan_runs,
+    )
+    for expiry, suppressed in [
+        ("2026-07-16T00:01:00Z", False),
+        ("2099-01-01T00:00:00Z", False),
+        ("2099-01-01T00:00:00.000001Z", True),
+    ]:
+        items = _monitor_blocked_advancement_items(next_due_at="2099-01-01T00:00:00Z")
+        items[0]["expires_at"] = expiry
+        guard = _quota_with_replan_runs(_blocked_wait_polls(), extra_agent_items=items)
+        kinds = [row["kind"] for row in (guard.get("autonomous_replan_obligation") or {}).get("triggers", [])]
+        assert ("blocked_successor_no_progress_repeat" not in kinds) is suppressed
+
+
+def test_codec_keeps_prose_out_of_the_history_decision_request(monkeypatch) -> None:
+    import json
+    from loopx.control_plane.work_items import replan_history
+
+    requests = []
+    def capture(method, params):
+        requests.append((method, params))
+        return {"schema_version": "replan_history_result_v0", "trigger": None}
+    monkeypatch.setattr(replan_history, "effect_runtime_result", capture)
+    prose = "private-unconsumed-prose" * 10000
+    rows = [{**monitor(2), "prompt": prose, "summary": prose}, monitor(1)]
+    replan_history.project_replan_history(rows, agent_id=AGENT)
+    assert len(requests) == 1
+    assert requests[0][0] == "work_item.replan_history.project"
+    encoded = json.dumps(requests[0][1])
+    assert prose not in encoded
+    assert len(encoded) < 5000
+    assert requests[0][1]["todos"]["resume"] is None
diff --git a/tests/control_plane/test_replan_history_provider.py b/tests/control_plane/test_replan_history_provider.py
new file mode 100644
index 0000000000..b1e28beebf
--- /dev/null
+++ b/tests/control_plane/test_replan_history_provider.py
@@ -0,0 +1,44 @@
+"""Real canonical stores feed the public quota consumer, including retry history."""
+from copy import deepcopy
+import json
+
+import pytest
+
+from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+from test_canonical_frontier_revision import _fixture
+from test_goal_amendment_proposal import _write_fixture, _stall_runs, _ack_run, GOAL_ID
+from loopx.control_plane.testing.canary_harness import run_json_cli_result
+from loopx.status import active_state_todo_fields, autonomous_replan_obligation_from_runs
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_complex_provider_snapshot_replan_and_quota_readback(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    paths = _write_fixture(tmp_path, runs=[])
+    projection = _fixture()
+    initialize_canonical_authority(paths["runtime"], GOAL_ID, projection,
+                                   state_path=paths["state_file"], provider=provider)
+    # The display is absent: the production reader must use persisted authority.
+    paths["state_file"].unlink()
+    goal = json.loads(paths["registry"].read_text())["goals"][0]
+    summary = active_state_todo_fields(goal, runtime_root=paths["runtime"])["agent_todos"]
+    history = list(reversed(_stall_runs()))
+    current = autonomous_replan_obligation_from_runs(history, agent_todos=summary, agent_id="agent-a")
+    assert current["triggers"][0]["kind"] == "typed_progress_repeat"
+    assert autonomous_replan_obligation_from_runs([history[0]] * 8,
+        agent_todos=summary, agent_id="agent-a") is None
+    ack = _ack_run(current["obligation_id"])
+    assert autonomous_replan_obligation_from_runs([ack, *history],
+        agent_todos=summary, agent_id="agent-a") is None
+    peer_ack = {**deepcopy(ack), "agent_id": "agent-b"}
+    assert autonomous_replan_obligation_from_runs([peer_ack, *history],
+        agent_todos=summary, agent_id="agent-a") == current
+    index = paths["runtime"] / "goals" / GOAL_ID / "runs" / "index.jsonl"
+    index.parent.mkdir(parents=True, exist_ok=True)
+    index.write_text("".join(json.dumps(row) + "\n" for row in reversed(history)))
+    code, result = run_json_cli_result("quota", "should-run", "--goal-id", GOAL_ID,
+        "--agent-id", "agent-a", registry_path=paths["registry"])
+    assert code == 0, result
+    assert "typed_progress_repeat" in json.dumps(result)
+    assert active_state_todo_fields(goal, runtime_root=paths["runtime"])["agent_todos"] == summary
+    assert not paths["state_file"].exists()
diff --git a/tests/control_plane_ts/replan_history.test.ts b/tests/control_plane_ts/replan_history.test.ts
new file mode 100644
index 0000000000..09193d7763
--- /dev/null
+++ b/tests/control_plane_ts/replan_history.test.ts
@@ -0,0 +1,167 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import type { JsonObject } from "../../loopx/control_plane/effect_program.ts";
+import { projectReplanHistory } from "../../loopx/control_plane/work_items/replan_history.ts";
+
+const neutral = ["quota_slot_spent", "quota_slot_voided", "delivery_completion_spend_accounted_v0"];
+function run(id: number, patch: JsonObject = {}): JsonObject {
+  return {
+    agent_id: "worker-a", public_agent_id: "worker-a", monitor_agent_id: "worker-a",
+    classification: "bounded_delivery", generated_at: `row-${id}`, observed_at: id,
+    turn_id: `turn-${id}`, accepted_ack: false, progress: null,
+    monitor: { target_id: null, mode: null, frontier: null, todo_id: null, target_key: null }, ...patch,
+  };
+}
+const observation = {
+  schema_version: "typed_progress_observation_v0", result_class: "unchanged",
+  fingerprint: "same-surface", surface_id: "fixture", evidence_ids: ["evidence-fixture"],
+};
+function poll(id: number, patch: JsonObject = {}): JsonObject {
+  return run(id, { classification: "quota_monitor_poll", monitor: {
+    target_id: "watch", mode: "due_monitor_observed_without_material_transition",
+    frontier: null, todo_id: "todo_watch", target_key: null, ...patch,
+  } });
+}
+function request(runs: JsonObject[], patch: JsonObject = {}): JsonObject {
+  return {
+    schema_version: "replan_history_request_v0", operation: "all", runs,
+    agent_id: "worker-a", monitor_agent_id: "worker-a", neutral_classifications: neutral,
+    stall_threshold: 2, periodic_threshold: 20, monitor_threshold: 6, streak_threshold: 5,
+    monitor_schema: "dead_monitor_repeat_v0", todos: { monitors: [], advancements: [], resume: null }, ...patch,
+  };
+}
+function trigger(runs: JsonObject[], patch: JsonObject = {}): JsonObject | null {
+  return projectReplanHistory(request(runs, patch)).trigger as JsonObject | null;
+}
+const many = (count: number, build = run): JsonObject[] => Array.from({ length: count }, (_, i) => build(count - i));
+
+test("periodic work counts distinct logical turns and keeps the existing public trigger", () => {
+  assert.equal(trigger(many(30, () => run(1))), null);
+  assert.deepEqual(trigger(many(20)), {
+    kind: "periodic_review_due", section: "run_history",
+    text: "latest 20 durable public run records since last autonomous replan reached periodic review threshold 20",
+    run_count: 20, threshold: 20, latest_generated_at: "row-20", oldest_counted_generated_at: "row-1",
+    agent_id: "worker-a",
+  });
+  assert.equal(trigger(many(19)), null);
+});
+
+test("ACK is a lane-scoped cutoff, including an ACK sharing the newest turn id", () => {
+  for (const operation of ["all", "progress", "periodic"]) {
+    const material = many(25, n => run(n, { progress: observation }));
+    assert.equal(trigger([{ ...material[0], accepted_ack: true }, ...material], { operation }), null);
+    assert.notEqual(trigger([{ ...run(30), accepted_ack: true, agent_id: "peer" }, ...material], { operation }), null);
+  }
+  assert.equal(trigger([run(25, { progress: observation }), run(24, { accepted_ack: true }),
+    run(23, { progress: observation })]), null);
+});
+
+test("neutral rows are transparent even when they carry the material turn id", () => {
+  for (const classification of neutral) {
+    const rows = [run(2, { classification }), run(2, { progress: observation }),
+      run(1, { classification }), run(1, { progress: observation })];
+    assert.equal(trigger(rows)?.kind, "typed_progress_repeat");
+  }
+});
+
+test("unknown work breaks a started progress streak; leading untyped history remains compatible", () => {
+  const p = (n: number) => run(n, { progress: observation });
+  assert.equal(trigger([p(3), run(2), p(1)]), null);
+  assert.equal(trigger([run(3), p(2), p(1)])?.kind, "typed_progress_repeat");
+  assert.equal(trigger([run(2), p(2), p(1)])?.kind, "typed_progress_repeat",
+    "an untyped record does not consume the progress evidence identity");
+  assert.equal(trigger([p(3), run(2, { classification: "quota_slot_voided_suffix" }), p(1)]), null);
+});
+
+test("progress fingerprints and outcomes govern repetition, never prose", () => {
+  for (const result_class of ["advanced", "exploration_exhausted", "no_followup"]) {
+    assert.equal(trigger(many(2, n => run(n, { progress: { ...observation, result_class } }))), null);
+  }
+  assert.equal(trigger([run(2, { progress: observation }), run(1, {
+    progress: { ...observation, fingerprint: "different-surface" },
+  })]), null);
+  assert.equal(trigger(many(2, n => run(n, { progress: { ...observation, result_class: "blocked" },
+    ignored_text: `advanced ${n}` })))?.kind, "typed_progress_repeat");
+});
+
+test("one retried monitor poll never becomes six observations", () => {
+  assert.equal(trigger(many(30, () => poll(1))), null);
+  assert.equal(trigger(many(6, poll))?.kind, "dead_monitor_repeat");
+  assert.equal(trigger(many(5, poll)), null);
+  assert.equal(trigger([poll(6, { target_id: "other" }), ...many(5, poll)]), null);
+  assert.equal(trigger(many(6, n => poll(n, { todo_id: null }))), null);
+});
+
+test("blocked successor compares typed target and frontier; ordinary work ends the monitor prefix", () => {
+  const blocked = (n: number, frontier = "frontier-a") => poll(n, {
+    mode: "blocked_successor_wait_without_material_transition", frontier,
+  });
+  assert.equal(trigger([blocked(2), blocked(1)])?.kind, "blocked_successor_no_progress_repeat");
+  assert.equal(trigger([blocked(2), blocked(1, "frontier-b")]), null);
+  assert.equal(trigger([blocked(3), run(2), blocked(1)]), null);
+});
+
+test("each historical trigger has deterministic precedence", () => {
+  const all = many(20, n => ({ ...poll(n), progress: observation }));
+  assert.equal(trigger(all)?.kind, "typed_progress_repeat");
+  assert.equal(trigger(many(20, poll))?.kind, "dead_monitor_repeat");
+  assert.equal(trigger(many(20))?.kind, "periodic_review_due");
+});
+
+test("multi-agent fixture retains a complete lane through peer ACKs, retries, and accounting", () => {
+  const rows = many(20).flatMap(row => [
+    { ...row, classification: neutral[0] }, row, { ...row },
+    { ...row, agent_id: "peer", accepted_ack: true },
+    ...many(10, n => run(n, { agent_id: `peer-${n}` })),
+  ]);
+  assert.equal(rows.length, 280);
+  const before = structuredClone(rows);
+  assert.equal(trigger(rows)?.run_count, 20);
+  assert.equal(trigger(rows.slice(0, -14)), null);
+  assert.deepEqual(rows, before, "the reducer may not reorder or mutate source history");
+  assert.equal(trigger(many(2, n => run(n, { turn_id: null, progress: observation })))?.kind,
+    "typed_progress_repeat", "untrustworthy legacy ids must not be deduplicated");
+});
+
+test("inferred lane follows the newest non-neutral attributed work", () => {
+  const rows = [run(30, { agent_id: "peer", classification: neutral[0] }), ...many(20)];
+  assert.equal(trigger(rows, { agent_id: null })?.agent_id, "worker-a");
+  assert.equal(trigger([...many(20, n => run(n, { agent_id: null, public_agent_id: null }))],
+    { agent_id: null })?.agent_id, null);
+});
+
+function monitorTodo(patch: JsonObject = {}): JsonObject {
+  return { id: "todo_watch", claim: "worker-a", public_claim: "worker-a", target: "watch",
+    no_change_count: 5, due_at: null, expires_at: null, ...patch };
+}
+test("persisted monitor streak retains stable tie order and advancement ownership", () => {
+  const todos = { monitors: [monitorTodo(), monitorTodo({ target: "second" })], advancements: [], resume: null };
+  assert.equal(trigger([], { operation: "monitor_streak", todos })?.monitor_target_id, "watch");
+  for (const claim of [null, "worker-a", "peer"]) {
+    const result = trigger([], { operation: "monitor_streak", todos: { ...todos,
+      advancements: [{ status: "open", task_class: "advancement_task", claim }],
+    } });
+    assert.equal(result === null, claim !== "peer");
+  }
+  assert.equal(trigger([], { operation: "monitor_streak", todos: { ...todos,
+    monitors: [monitorTodo({ no_change_count: 4 })],
+  } }), null);
+});
+
+for (const [key, bad, message] of [
+  ["schema_version", "future_schema", /schema/], ["operation", "dispatch", /operation/],
+  ["runs", {}, /runs/], ["stall_threshold", 1, /stall_threshold/],
+  ["periodic_threshold", 0, /periodic_threshold/], ["monitor_threshold", 1.5, /monitor_threshold/],
+  ["streak_threshold", Number.MAX_SAFE_INTEGER + 1, /streak_threshold/],
+  ["neutral_classifications", [true], /neutral_classifications/],
+] as const) {
+  test(`reject malformed boundary field ${key}`, () => {
+    assert.throws(() => projectReplanHistory(request([], { [key]: bad })), message);
+  });
+}
+test("malformed typed facts fail visibly rather than falling back to Python policy", () => {
+  for (const patch of [
+    { accepted_ack: "true" }, { observed_at: NaN }, { progress: { ...observation, result_class: "success" } },
+    { progress: { ...observation, fingerprint: "" } }, { monitor: [] },
+  ]) assert.throws(() => projectReplanHistory(request([run(1, patch)])));
+});
diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json
index a464d28745..0203cdd6d7 100644
--- a/tsconfig.control-plane.json
+++ b/tsconfig.control-plane.json
@@ -146,6 +146,7 @@
     "tests/control_plane_ts/vision_checkpoint.test.ts",
     "tests/control_plane_ts/vision_wait_coverage.test.ts",
     "tests/control_plane_ts/shared_goal_alignment.test.ts",
+    "tests/control_plane_ts/replan_history.test.ts",
     "tests/control_plane_ts/goal_amendment_proposal.test.ts"
   ]
 }

From 47d30bc0f35318419adc39fc99f67f27f05309b9 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 23:40:35 +0800
Subject: [PATCH 2/5] docs(replan): record history semantics and qualified
 migration boundary

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../2026-09-22-replan-consumer-checkpoint.md  | 22 +++++++++
 ...-09-22-replan-consumer-checkpoint.zh-CN.md | 16 ++++++
 .../2026-09-22-replan-history-policy.md       | 30 ++++++++++++
 .../2026-09-22-replan-history-policy.zh-CN.md | 21 ++++++++
 .../goal-vision-replan-contract-v0.md         | 49 +++++++++++++++++++
 5 files changed, 138 insertions(+)
 create mode 100644 docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-22-replan-consumer-checkpoint.md
 create mode 100644 docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-22-replan-consumer-checkpoint.zh-CN.md
 create mode 100644 docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.md
 create mode 100644 docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.zh-CN.md

diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-22-replan-consumer-checkpoint.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-22-replan-consumer-checkpoint.md
new file mode 100644
index 0000000000..2e72375436
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-22-replan-consumer-checkpoint.md
@@ -0,0 +1,22 @@
+# Replan consumer checkpoint and default-path estimate
+
+The replan history decision family now shares a TS owner across legacy and
+canonical consumers. See the [TS checkpoint](../typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.md)
+for the four semantic fixes and isolated real-source readback. This advances T3;
+it does not establish a new provider durability, capture, or promotion claim.
+
+The planning range remains **5–8 cohesive delivery packages**, conditional on
+acceptance rather than line counts: remaining caller/executor fences (1–2),
+consumer/projection recovery (1), SQLite durability and qualification (1–2),
+whole-Goal capture/migration (1–2), then default onboarding and bounded retirement
+(1). These categories can overlap in a complete package; their maxima are not
+independent additive promises. The current history slice does not retire an
+entire package. The existing SQLite long-running qualification owner (#4224)
+and the required elapsed soak remain dependencies.
+
+A complete capture attempt exposed a still-open archive dependency role/class
+hold. Active read-model parity is useful but cannot discharge this hold. Existing
+Goal migration must keep its exact-source qualification, old-writer fence and
+rollback acceptance; no active Goal was promoted for this validation. PostgreSQL
+remains separately qualified and opt-in; this PR changes no store transaction,
+connection, selection, schema or migration contract.
diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-22-replan-consumer-checkpoint.zh-CN.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-22-replan-consumer-checkpoint.zh-CN.md
new file mode 100644
index 0000000000..5224a15807
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-22-replan-consumer-checkpoint.zh-CN.md
@@ -0,0 +1,16 @@
+# Replan 消费侧检查点与默认路径估算
+
+历史 replan 决策族已由旧路径与 canonical 消费侧共享 TS 规则；四项语义修复与
+隔离真实读后核对见 [TS 检查点](../typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.zh-CN.md)。
+这推进 T3，不新增 provider 持久性、完整捕获或晋升资格声明。
+
+仍估计需要 **5–8 个完整交付包**：剩余 caller/executor fence 1–2、消费侧和投影
+恢复 1、SQLite 持久性资格 1–2、整 Goal 捕获及迁移 1–2、默认 onboarding 和受控
+删除 1。相关类别可以在一个完整包中重叠，不能把各自最大值当作独立承诺相加。
+本次历史触发 slice 没有独自清空其中一个完整包；SQLite 长程资格原有负责人
+（#4224）及实际 soak 时间仍是依赖，不能用 PR 数量抵扣。
+
+完整捕获还遇到归档依赖 role/class 不完整的门禁。活动读模型一致不能消除这个
+阻塞。既有 Goal 迁移仍需精确源资格、旧 writer fence 及回滚验收；本次没有晋升
+活动 Goal。PostgreSQL 继续独立资格化并显式选择，本次不修改其事务、连接、
+选择、数据库 schema 或迁移合同。
diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.md
new file mode 100644
index 0000000000..ba4495eeaf
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.md
@@ -0,0 +1,30 @@
+# Replan history decision owner
+
+Goal/source: overall roadmap #4574, typed control-plane T3 consumer ownership.
+The observed gap was independent Python history scans with different retry,
+ACK, neutral-accounting and lane semantics. The delivered boundary is one
+`work_item.replan_history.project` request, with in-process reuse of the existing
+Todo resume planner. Python keeps legacy codecs and public obligation rendering;
+its replaced historical trigger scans and duplicated neutral vocabulary are removed.
+
+Four independently specified regression cases failed on baseline `709734cd6`:
+periodic retry overcount, Monitor retry overcount, accepted ACK not resetting
+progress repetition, and neutral accounting breaking equivalent progress. The
+new owner corrects them while retaining thresholds, precedence and obligation
+identity. Typed negative cases and the 280-row multi-agent interleaving fixture
+cover scope-before-ACK, retries, missing identities, invalid input and source
+immutability. Real File/SQLite consumer tests delete the display before reading
+status and invoking quota, rather than substituting an in-memory store.
+
+A read-only local-source rehearsal covered 345 active Todos, 600 history rows
+and five agent lanes: all five historical projections matched baseline; isolated
+File/SQLite readback and public quota CLI passed without changing the source.
+This was an active read-model snapshot, not whole-Goal promotion. Complete source
+capture independently rejected an archived dependency with missing/incompatible
+role/task-class facts. That migration hold remains and was not bypassed or repaired
+in active state.
+
+This closes the history-trigger decision family, not all T3 or default adoption.
+Progress fingerprint codecs, obligation assembly, frontier settlement and broader
+capture/provider qualification retain their owners. No new model observer,
+provider selection, frontend configuration or optional capability is introduced.
diff --git a/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.zh-CN.md b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.zh-CN.md
new file mode 100644
index 0000000000..4b7ff07a7f
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/typescript-control-plane-migration-v0/2026-09-22-replan-history-policy.zh-CN.md
@@ -0,0 +1,21 @@
+# Replan 历史决策归属
+
+目标来源为总纲 #4574 与 T3 消费侧 TS 迁移。原有 Python 历史扫描分别处理重试、
+ACK、中性记账及 agent 归属，规则已经分叉。本次统一到一次
+`work_item.replan_history.project` 请求，并在 TS 内直接复用 Todo resume planner。
+Python 保留历史解码、指纹及 obligation 呈现；删除被替代的触发扫描和重复词表。
+
+基线 `709734cd6` 上独立编写的四个反例失败：周期复盘重复累计重试、监控重复
+累计重试、ACK 未截断进展停滞、记账打断同质进展。新实现修复这些问题，保持
+阈值、优先级和 obligation 标识。280 行多 agent 交错 fixture 与边界负例验证
+先归属后 ACK、重试、无效身份、输入拒绝及不修改源数据。File/SQLite 消费侧测试
+使用真实持久化 provider，删除展示文件后读取状态并执行 quota CLI。
+
+本机只读演练覆盖 345 条活动 Todo、600 条历史记录和五个 agent：五个历史投影
+与基线一致，隔离 File/SQLite 读后核对及公开 quota CLI 通过，活动源未修改。
+该证据仅限活动读模型，不证明完整 Goal 晋升。独立的完整捕获因归档依赖缺少或
+不兼容的 role/task-class 事实而拒绝；这个迁移阻塞保留，没有绕过门禁或改写活动状态。
+
+本次闭合历史触发决策族，不代表全部 T3 或默认切换。进展指纹解码、obligation
+组装、frontier 结算及完整捕获资格仍有各自归属。没有加入模型观察器、provider
+切换、前端配置项或可选 capability。
diff --git a/docs/reference/protocols/goal-vision-replan-contract-v0.md b/docs/reference/protocols/goal-vision-replan-contract-v0.md
index 5baeb1d85a..01387b9468 100644
--- a/docs/reference/protocols/goal-vision-replan-contract-v0.md
+++ b/docs/reference/protocols/goal-vision-replan-contract-v0.md
@@ -730,3 +730,52 @@ A change satisfies this contract only when:
 - auto-research remains a thin preset over the reusable kernel; and
 - public docs and smokes cover the budget, state machine, and `quota.py`
   boundary without private material.
+
+## History-trigger ownership and retry semantics
+
+The built-in `work_items/replan_history.ts` decision owns historical progress
+repetition, blocked-successor repetition, repeated executed Monitor polls,
+periodic review, and the persisted unchanged-Monitor streak. The Python codec
+preserves historical observation fingerprints and timestamp parsing, then sends
+one bounded fact request. Obligation rendering and identity serialization retain
+the existing public contract. This is deterministic policy; no observer model,
+new capability, provider selection, or additional permission is introduced.
+
+History is newest first. Agent scoping precedes an accepted replan ACK cutoff;
+a peer ACK cannot clear another lane. The three existing neutral accounting
+classifications are transparent. A valid logical turn id is counted once per
+agent, including an id carried by settlement identity. Missing, malformed, or
+conflicting historical ids remain separate rows; the reader does not invent an
+identity. Unknown material work still breaks an established progress streak.
+
+The default thresholds remain two equivalent typed observations, two blocked
+successor waits, six executed unchanged Monitor turns, twenty material turns
+for periodic review, and five persisted unchanged polls for a Monitor-only
+lane. Trigger precedence remains progress, Monitor, then periodic review.
+Accepted ACKs reset the historical window; clearing another frontier obligation
+still requires its existing typed semantic outcome and revision rules. A future
+blocking Monitor suppresses premature wait replanning only while its schedule
+and expiry are valid; the decision reuses the Todo resume planner.
+
+These are enforced replan conditions, not advisory hints. Relative to the older
+reader, retry records no longer accelerate periodic/Monitor thresholds, accepted
+ACKs now stop typed-progress repetition, and neutral accounting no longer hides
+repetition. These changes apply to legacy, File, and SQLite status/quota callers
+without an opt-in. Frontend and Lark consume the existing obligation shape and
+need no new setting or editor. Read back with `loopx status --goal-id <id>` and
+`loopx quota should-run --goal-id <id> --agent-id <agent>`.
+
+### 中文：历史触发与重试语义
+
+历史触发规则由现有 work_items 的 TypeScript 边界统一维护，Python 负责旧数据
+解码及原有 obligation 呈现。没有新增模型、capability、provider 选择或权限。
+先按 agent 筛选，再遇到已接受的 replan ACK 截断窗口；其他 agent 的 ACK 不能
+清空当前窗口。中性额度记账不计数、不打断停滞；同一 agent 的有效 Turn ID
+只计一次。无效、缺失或相互矛盾的历史 ID 不被猜测性合并。
+
+阈值仍为：2 次相同 typed progress、2 次 successor 等待、6 次已执行监控、
+20 次实质工作轮次、5 次持久化监控无变化。优先级及 obligation 标识保持原样。
+修复的是计数单位、ACK 截断和记账透明性，适用于旧路径及 File/SQLite；这些是
+机器执行的 replan 条件。尚未到期且在到期时仍有效的关联监控继续抑制提前重规划。
+这不替代其他 frontier 的语义验收、版本检查或权限。前端与 Lark 继续使用原有
+返回结构；可用上面的 status/quota 命令核对。

From c7e59215d93f4737f9c70ab6f277210a79f56750 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 23:43:15 +0800
Subject: [PATCH 3/5] fix(replan): retain turn identity when legacy retry
 attribution is absent

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/control_plane/work_items/replan_history.ts | 5 ++++-
 tests/control_plane_ts/replan_history.test.ts    | 8 ++++++++
 2 files changed, 12 insertions(+), 1 deletion(-)

diff --git a/loopx/control_plane/work_items/replan_history.ts b/loopx/control_plane/work_items/replan_history.ts
index 87481ce3d7..9eb7c7a9f9 100644
--- a/loopx/control_plane/work_items/replan_history.ts
+++ b/loopx/control_plane/work_items/replan_history.ts
@@ -174,8 +174,11 @@ function historyWindow(request: Request): Run[] {
  */
 function* distinctTurns(runs: readonly Run[], counts: (run: Run) => boolean = () => true): Generator<Run, void, unknown> {
   const seen = new Set<string>();
+  // Historical goal-level rows can omit attribution. Within a single scoped
+  // lane, keep the original turn identity across an attributed/unattributed retry.
+  const lane = sole(runs.map(run => run.agent));
   for (const run of runs) {
-    const key = run.turn ? JSON.stringify([run.agent, run.turn]) : null;
+    const key = run.turn ? JSON.stringify([run.agent ?? lane, run.turn]) : null;
     if (key && seen.has(key)) continue;
     if (key && counts(run)) seen.add(key);
     yield run;
diff --git a/tests/control_plane_ts/replan_history.test.ts b/tests/control_plane_ts/replan_history.test.ts
index 09193d7763..23a5ec17be 100644
--- a/tests/control_plane_ts/replan_history.test.ts
+++ b/tests/control_plane_ts/replan_history.test.ts
@@ -92,6 +92,14 @@ test("one retried monitor poll never becomes six observations", () => {
   assert.equal(trigger(many(6, n => poll(n, { todo_id: null }))), null);
 });
 
+test("legacy missing attribution retains turn identity within a scoped lane", () => {
+  const attributed = run(1, { progress: observation });
+  const goalLevel = { ...attributed, agent_id: null, public_agent_id: null, monitor_agent_id: null };
+  assert.equal(trigger([attributed, goalLevel]), null);
+  assert.equal(trigger(many(20, n => ({ ...attributed, ...(n % 2 ? goalLevel : {}) })),
+    { operation: "periodic" }), null);
+});
+
 test("blocked successor compares typed target and frontier; ordinary work ends the monitor prefix", () => {
   const blocked = (n: number, frontier = "frontier-a") => poll(n, {
     mode: "blocked_successor_wait_without_material_transition", frontier,

From 1182f7c6ef448e621d84d359dc56b40cde256975 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Tue, 22 Sep 2026 23:45:01 +0800
Subject: [PATCH 4/5] refactor(replan): distinguish historical codec from the
 typed policy owner

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/control_plane/status/autonomous_replan_projection.py      | 2 +-
 loopx/control_plane/work_items/autonomous_replan_obligation.py  | 2 +-
 loopx/control_plane/work_items/progress_observation.py          | 2 +-
 .../work_items/{replan_history.py => replan_history_codec.py}   | 0
 loopx/status.py                                                 | 2 +-
 tests/control_plane/test_replan_history_policy.py               | 2 +-
 6 files changed, 5 insertions(+), 5 deletions(-)
 rename loopx/control_plane/work_items/{replan_history.py => replan_history_codec.py} (100%)

diff --git a/loopx/control_plane/status/autonomous_replan_projection.py b/loopx/control_plane/status/autonomous_replan_projection.py
index 8f99882bf7..25c63535bb 100644
--- a/loopx/control_plane/status/autonomous_replan_projection.py
+++ b/loopx/control_plane/status/autonomous_replan_projection.py
@@ -5,7 +5,7 @@
 
 from typing import Any
 
-from ..work_items.replan_history import (
+from ..work_items.replan_history_codec import (
     REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS as AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS,
 )
 
diff --git a/loopx/control_plane/work_items/autonomous_replan_obligation.py b/loopx/control_plane/work_items/autonomous_replan_obligation.py
index 115394d381..fcad861689 100644
--- a/loopx/control_plane/work_items/autonomous_replan_obligation.py
+++ b/loopx/control_plane/work_items/autonomous_replan_obligation.py
@@ -11,7 +11,7 @@
     normalize_todo_replan_obligation_id,
 )
 from .progress_observation import replan_writeback_requirements
-from .replan_history import project_replan_history
+from .replan_history_codec import project_replan_history
 from .replan_settlement import (
     project_todo_lifecycle_settlement_reentry as project_todo_lifecycle_reentry_effect,
 )
diff --git a/loopx/control_plane/work_items/progress_observation.py b/loopx/control_plane/work_items/progress_observation.py
index 4e25c6e123..3198026b3d 100644
--- a/loopx/control_plane/work_items/progress_observation.py
+++ b/loopx/control_plane/work_items/progress_observation.py
@@ -194,7 +194,7 @@ def typed_progress_repeat_trigger(
     threshold: int = PROGRESS_REPEAT_THRESHOLD,
 ) -> dict[str, Any] | None:
     """Compatibility entrypoint for the shared TypeScript history window."""
-    from .replan_history import project_replan_history
+    from .replan_history_codec import project_replan_history
 
     return project_replan_history(
         newest_first_runs, operation="progress", agent_id=agent_id,
diff --git a/loopx/control_plane/work_items/replan_history.py b/loopx/control_plane/work_items/replan_history_codec.py
similarity index 100%
rename from loopx/control_plane/work_items/replan_history.py
rename to loopx/control_plane/work_items/replan_history_codec.py
diff --git a/loopx/status.py b/loopx/status.py
index 3bf05471ce..e441d73b85 100644
--- a/loopx/status.py
+++ b/loopx/status.py
@@ -6,7 +6,7 @@
 import re
 from typing import Any
 
-from .control_plane.work_items.replan_history import (
+from .control_plane.work_items.replan_history_codec import (
     REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS as AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS,  # noqa: F401 - compatibility export
 )
 
diff --git a/tests/control_plane/test_replan_history_policy.py b/tests/control_plane/test_replan_history_policy.py
index 9a3b1c4113..4238acd0dc 100644
--- a/tests/control_plane/test_replan_history_policy.py
+++ b/tests/control_plane/test_replan_history_policy.py
@@ -128,7 +128,7 @@ def test_future_monitor_expiry_must_cover_the_due_instant() -> None:
 
 def test_codec_keeps_prose_out_of_the_history_decision_request(monkeypatch) -> None:
     import json
-    from loopx.control_plane.work_items import replan_history
+    from loopx.control_plane.work_items import replan_history_codec as replan_history
 
     requests = []
     def capture(method, params):

From 344f18af31732f593349249a8d92713288e22856 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Wed, 23 Sep 2026 09:56:35 +0800
Subject: [PATCH 5/5] fix(status): keep the facade export inside the audited
 allowlist

The replan refactor replaced the local AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS
literal with an aliased import, which the facade audit reads as a new unlisted
public import-only binding, so
tests/architecture/test_control_plane_import_boundaries.py::test_public_facade_import_only_reexports_match_the_audited_allowlist
regressed from base to head. Import the codec name privately and keep the
established public name as an identity alias, the same shape the facade already
uses for STATUS_NEUTRAL_CLASSIFICATIONS. The vocabulary stays single-sourced in
the codec, the public name and its value are unchanged for existing callers, and
the audited allowlist needs no new entry.

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/status.py | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/loopx/status.py b/loopx/status.py
index e441d73b85..69fed8cd07 100644
--- a/loopx/status.py
+++ b/loopx/status.py
@@ -1,13 +1,12 @@
 from __future__ import annotations
 
-
 from collections.abc import Mapping, Sequence
 from pathlib import Path
 import re
 from typing import Any
 
 from .control_plane.work_items.replan_history_codec import (
-    REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS as AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS,  # noqa: F401 - compatibility export
+    REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS as _REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS,
 )
 
 from .control_plane import compact_control_plane_policy
@@ -351,6 +350,14 @@
 )
 AUTONOMOUS_REPLAN_SCHEMA_VERSION = "autonomous_replan_obligation_v0"
 DEAD_MONITOR_REPEAT_SCHEMA_VERSION = "dead_monitor_repeat_v0"
+# Refs #4447: one definition for this vocabulary, owned by the control-plane
+# codec that now feeds the replan history policy across status and quota. The
+# facade keeps exporting the established public name for existing callers as an
+# identity alias instead of restating the values, which also keeps the audited
+# import-only export allowlist unchanged.
+AUTONOMOUS_RUN_HISTORY_NEUTRAL_CLASSIFICATIONS = (
+    _REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS
+)