fix(replan): count only material monitor polls in periodic review

docs/quota-allocation.md

@@ -129,7 +129,8 @@ Other cadences are independent: the standard profile's two-small-delivery
 streak suggests widening work; fine mode's five-small-delivery streak suggests
 direction review. Neither is this completed-Todo counter. The periodic review
 window of 20 material run records and long-open-Todo-chain triggers also retain
-their existing thresholds.
+their existing thresholds. Quiet or unchanged Monitor polls do not consume the
+periodic material-run window; their dedicated Monitor replan thresholds still apply.
 
 ### Governed Turn Execution
 

docs/reference/protocols/goal-vision-replan-contract-v0.md

@@ -916,6 +916,9 @@ 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.
+No-change `quota_monitor_poll` receipts do not advance the periodic material-turn
+counter; a poll with an explicit material transition does. Monitor repetition
+and persisted streak checks still inspect their own poll evidence.
 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

loopx/control_plane/work_items/replan_history.ts

@@ -15,6 +15,7 @@ interface Progress {
 interface Monitor {
   target: string | null;
   mode: string | null;
+  material: boolean | null;
   frontier: string | null;
   todo: string | null;
   key: string | null;
@@ -131,6 +132,8 @@ function decode(value: unknown): Request {
         monitor: {
           target: optionalNonEmptyString(monitor.target_id, "target_id"),
           mode: optionalNonEmptyString(monitor.mode, "mode"),
+          material: monitor.material_change == null
+            ? null : requireBoolean(monitor.material_change, "material_change"),
           frontier: optionalNonEmptyString(monitor.frontier, "frontier"),
           todo: optionalNonEmptyString(monitor.todo_id, "todo_id"),
           key: optionalNonEmptyString(monitor.target_key, "target_key"),
@@ -211,7 +214,11 @@ function progressTrigger(runs: readonly Run[], request: Request): Trigger | null
 }
 function periodicTrigger(runs: readonly Run[], request: Request): Trigger | null {
   const durable: Run[] = [];
-  for (const run of distinctTurns(runs.filter(row => row.classification))) {
+  // Monitor liveness receipts are durable, but only a material transition is
+  // work for the periodic direction review. Monitor-specific triggers still
+  // inspect every poll in the unfiltered history window.
+  for (const run of distinctTurns(runs.filter(row => row.classification &&
+      (row.classification !== "quota_monitor_poll" || row.monitor.material === true)))) {
     durable.push(run);
     if (durable.length >= request.periodic) break;
   }

loopx/control_plane/work_items/replan_history_codec.py

@@ -49,6 +49,10 @@ def _run_fact(run: Mapping[str, Any], ack_recorded: Callable[..., bool]) -> dict
         "monitor": {
             "target_id": str(target.get("target_id") or "").strip() or None,
             "mode": str(target.get("monitor_mode") or "").strip() or None,
+            "material_change": (
+                event.get("material_change")
+                if isinstance(event.get("material_change"), bool) else 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,

tests/control_plane/test_replan_history_policy.py

@@ -2,7 +2,10 @@
 
 from __future__ import annotations
 
-from loopx.status import autonomous_replan_obligation_from_runs
+from loopx.status import (
+    autonomous_replan_obligation_from_runs,
+    autonomous_replan_periodic_review_from_runs,
+)
 from loopx.control_plane.work_items.progress_observation import typed_progress_repeat_trigger
 
 AGENT = "history-worker"
@@ -82,6 +85,28 @@ def test_twenty_distinct_turns_retain_periodic_identity() -> None:
     assert obligation(rows) == first
 
 
+def test_periodic_review_ignores_quiet_polls_but_counts_material_transitions() -> None:
+    quiet = [
+        {**monitor(n, turn=f"quiet-{n}"),
+         "monitor_target": {"target_id": "watch", "agent_id": AGENT,
+                            "monitor_mode": ("due_monitor_observed_without_material_transition"
+                                             if n > 38 else "monitor_quiet_until_material_transition")},
+         "monitor_event": {"material_change": False}}
+        for n in range(40, 20, -1)
+    ]
+    work = [run(n, turn=f"work-{n}") for n in range(19, 0, -1)]
+
+    def periodic(rows: list[dict]) -> dict | None:
+        return autonomous_replan_periodic_review_from_runs(rows, agent_todos=None)
+    assert periodic(quiet) is None
+    assert periodic(quiet + work) is None
+    material = {**monitor(41, turn="material-41"),
+                "monitor_target": {"target_id": "watch", "agent_id": AGENT,
+                                   "monitor_mode": "due_monitor_material_transition"},
+                "monitor_event": {"material_change": True}}
+    assert periodic([material] + quiet + work)["triggers"][0]["kind"] == "periodic_review_due"
+
+
 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

tests/control_plane_ts/replan_history.test.ts

@@ -19,7 +19,7 @@ const observation = {
 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,
+    material_change: false, frontier: null, todo_id: "todo_watch", target_key: null, ...patch,
   } });
 }
 function request(runs: JsonObject[], patch: JsonObject = {}): JsonObject {
@@ -46,6 +46,16 @@ test("periodic work counts distinct logical turns and keeps the existing public
   assert.equal(trigger(many(19)), null);
 });
 
+test("periodic review counts material work, not quiet monitor receipts", () => {
+  const quiet = many(20, n => poll(n, { mode: n > 18
+    ? "due_monitor_observed_without_material_transition" : "monitor_quiet_until_material_transition" }));
+  assert.equal(trigger(quiet, { operation: "periodic" }), null);
+  assert.equal(trigger([...quiet, ...many(19)], { operation: "periodic" }), null);
+  const materialPoll = poll(21, { mode: "due_monitor_material_transition", material_change: true });
+  assert.equal(trigger([materialPoll, ...quiet, ...many(19)], { operation: "periodic" })?.kind,
+    "periodic_review_due");
+});
+
 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 }));
@@ -171,6 +181,7 @@ test("malformed typed facts fail visibly rather than falling back to Python poli
   for (const patch of [
     { accepted_ack: "true" }, { observed_at: NaN }, { progress: { ...observation, result_class: "success" } },
     { progress: { ...observation, fingerprint: "" } }, { monitor: [] },
+    { monitor: { ...poll(1).monitor as JsonObject, material_change: "true" } },
   ]) assert.throws(() => projectReplanHistory(request([run(1, patch)])));
 });