feat(periodic-report): add bounded segment milestone trigger (#3527)

* feat(periodic-report): add bounded segment milestones

Signed-off-by: cmyk-labs <niangxizhuang01@gmail.com>

* test(periodic-report): cover bounded segment triggers

Signed-off-by: cmyk-labs <niangxizhuang01@gmail.com>

* docs(periodic-report): document segment milestone contract

Signed-off-by: cmyk-labs <niangxizhuang01@gmail.com>

---------

Signed-off-by: cmyk-labs <niangxizhuang01@gmail.com>

Author: cmyk-labs

docs/reference/protocols/periodic-report-v0.md

@@ -159,6 +159,8 @@ candidate facts. The built-in kinds are:
   and a successor is established or the goal is terminal;
 - `primary_goal_outcome`: the primary delivery outcome is validated and has a
   durable writeback;
+- `bounded_segment_milestone`: a named, bounded work segment completed or
+  entered replan and was durably written back, even when open todos remain;
 - `material_decision`: an approved, rejected, or cancelled decision changed
   the execution route and was durably recorded;
 - `material_blocker`: a new or escalated P0 blocker stops the primary path;
@@ -179,6 +181,16 @@ the selected and coalesced ids, every suppression reason, cooldown state, and
 the report kind (`cadence_digest`, `milestone_update`, `exception_update`, or
 `manual_update`).
 
+`bounded_segment_milestone` accepts only `segment_ref`, `transition`,
+`delivered_count`, `remaining_todo_count`, and `durable_writeback` facts.
+`segment_ref` and `transition` are required. Counts are non-negative and
+default to zero when omitted. A `segment_completed` or `replan_entered`
+transition is material only after `durable_writeback=true`;
+`vision_checkpoint` remains non-material. Remaining todos do not suppress an
+otherwise material segment milestone. The trigger has priority 40, between
+`material_decision` and `cadence_due`, maps to `milestone_update`, and obeys
+the normal profile cooldown.
+
 An eligible decision may be embedded as `trigger_receipt` in a
 `periodic_report_run_request_v0`. Its `report_key` and `report_kind` then
 participate in run identity, so a milestone update and a scheduled digest over

examples/fixtures/periodic-report-product-profiles.public.json

@@ -14,6 +14,7 @@
       "trigger_policy": {
         "enabled_kinds": [
           "manual",
+          "bounded_segment_milestone",
           "cadence_due",
           "primary_goal_outcome",
           "vision_closed",

loopx/capabilities/periodic_report/core.py

@@ -24,6 +24,7 @@
     "milestone_update",
 }
 _REPORTABLE_TRIGGER_KINDS = {
+    "bounded_segment_milestone",
     "cadence_due",
     "manual",
     "material_blocker",

loopx/capabilities/periodic_report/presets.py

@@ -30,6 +30,7 @@
     "trigger_policy": {
         "enabled_kinds": [
             "manual",
+            "bounded_segment_milestone",
             "cadence_due",
             "primary_goal_outcome",
             "vision_closed",

loopx/capabilities/periodic_report/triggers.py

@@ -22,6 +22,7 @@
 TRIGGER_DECISION_SCHEMA = "periodic_report_trigger_decision_v0"
 
 _REPORTABLE_KINDS = {
+    "bounded_segment_milestone",
     "cadence_due",
     "manual",
     "material_blocker",
@@ -44,9 +45,11 @@
     "material_blocker": 70,
     "material_recovery": 60,
     "material_decision": 50,
+    "bounded_segment_milestone": 40,
     "cadence_due": 10,
 }
 _REPORT_KIND = {
+    "bounded_segment_milestone": "milestone_update",
     "cadence_due": "cadence_digest",
     "manual": "manual_update",
     "material_blocker": "exception_update",
@@ -147,6 +150,39 @@ def _materiality(kind: str, facts: Mapping[str, Any], label: str) -> tuple[bool,
             _boolean(facts.get("due"), f"{label}.due"),
             "cadence_due" if facts.get("due") is True else "cadence_not_due",
         )
+    if kind == "bounded_segment_milestone":
+        _reject_unknown_facts(
+            facts,
+            allowed={
+                "delivered_count",
+                "durable_writeback",
+                "remaining_todo_count",
+                "segment_ref",
+                "transition",
+            },
+            label=label,
+        )
+        _token(facts.get("segment_ref"), f"{label}.segment_ref")
+        transition = _token(facts.get("transition"), f"{label}.transition")
+        if transition not in {
+            "replan_entered",
+            "segment_completed",
+            "vision_checkpoint",
+        }:
+            raise ValueError(f"{label}.transition is invalid")
+        _integer(facts.get("delivered_count", 0), f"{label}.delivered_count")
+        _integer(
+            facts.get("remaining_todo_count", 0),
+            f"{label}.remaining_todo_count",
+        )
+        durable_writeback = _boolean(
+            facts.get("durable_writeback"), f"{label}.durable_writeback"
+        )
+        if transition not in {"replan_entered", "segment_completed"}:
+            return False, "segment_checkpoint_only"
+        if not durable_writeback:
+            return False, "segment_writeback_pending"
+        return True, "bounded_segment_advanced"
     if kind == "vision_closed":
         _reject_unknown_facts(
             facts,

tests/capabilities/test_periodic_report_profile.py

@@ -116,6 +116,9 @@ def test_weekly_preset_is_portable_domain_neutral_and_aliasable() -> None:
     assert {
         item["adapter_id"] for item in activation["profile"]["renderer_bindings"]
     } == {"markdown_v0", "html_artifact_v0"}
+    assert "bounded_segment_milestone" in activation["profile"]["trigger_policy"][
+        "enabled_kinds"
+    ]
     serialized = json.dumps(activation, ensure_ascii=False).lower()
     assert "issue_fix" not in serialized and "pull_request" not in serialized
 

tests/capabilities/test_periodic_report_triggers.py

@@ -40,6 +40,7 @@ def _trigger_request(*candidates: dict[str, object]) -> dict[str, object]:
         },
         "trigger_policy": {
             "enabled_kinds": [
+                "bounded_segment_milestone",
                 "cadence_due",
                 "manual",
                 "material_blocker",
@@ -242,6 +243,119 @@ def test_vision_requires_validated_closure_and_continuation() -> None:
     assert payload["report_kind"] == "milestone_update"
 
 
+@pytest.mark.parametrize("transition", ["segment_completed", "replan_entered"])
+def test_bounded_segment_with_remaining_todos_triggers_milestone_update(
+    transition: str,
+) -> None:
+    request = _trigger_request(
+        _candidate(
+            "cadence_due",
+            source_ref="scheduler:weekly-window/2026-w29",
+            evidence_digest="sha256:weekly-due",
+            facts={"due": True},
+        ),
+        _candidate(
+            "bounded_segment_milestone",
+            source_ref="segment:2026-w29",
+            evidence_digest=f"sha256:{transition}",
+            facts={
+                "segment_ref": "week-2026-w29",
+                "transition": transition,
+                "delivered_count": 4,
+                "remaining_todo_count": 12,
+                "durable_writeback": True,
+            },
+        ),
+    )
+
+    payload = build_periodic_report_trigger_decision(request)
+    run = build_periodic_report_run(_run_request(payload))
+
+    assert payload["eligible"] is True
+    assert payload["selected_trigger_kind"] == "bounded_segment_milestone"
+    assert payload["report_kind"] == "milestone_update"
+    assert len(payload["coalesced_trigger_ids"]) == 2
+    assert run["trigger_receipt"]["report_kind"] == "milestone_update"
+
+
+@pytest.mark.parametrize(
+    ("transition", "durable_writeback", "reason"),
+    [
+        ("vision_checkpoint", True, "segment_checkpoint_only"),
+        ("replan_entered", False, "segment_writeback_pending"),
+    ],
+)
+def test_bounded_segment_requires_material_transition_and_writeback(
+    transition: str,
+    durable_writeback: bool,
+    reason: str,
+) -> None:
+    payload = build_periodic_report_trigger_decision(
+        _trigger_request(
+            _candidate(
+                "bounded_segment_milestone",
+                source_ref="segment:2026-w29",
+                evidence_digest=f"sha256:{transition}",
+                facts={
+                    "segment_ref": "week-2026-w29",
+                    "transition": transition,
+                    "delivered_count": 4,
+                    "remaining_todo_count": 12,
+                    "durable_writeback": durable_writeback,
+                },
+            )
+        )
+    )
+
+    assert payload["eligible"] is False
+    assert payload["suppressed_triggers"][0]["reason"] == reason
+
+
+@pytest.mark.parametrize(
+    "facts, error",
+    [
+        (
+            {
+                "transition": "segment_completed",
+                "durable_writeback": True,
+            },
+            "segment_ref is required",
+        ),
+        (
+            {
+                "segment_ref": "week-2026-w29",
+                "transition": "unknown_transition",
+                "durable_writeback": True,
+            },
+            "transition is invalid",
+        ),
+        (
+            {
+                "segment_ref": "week-2026-w29",
+                "transition": "segment_completed",
+                "remaining_todo_count": -1,
+                "durable_writeback": True,
+            },
+            "remaining_todo_count must be between",
+        ),
+    ],
+)
+def test_bounded_segment_facts_are_strictly_validated(
+    facts: dict[str, object], error: str
+) -> None:
+    with pytest.raises(ValueError, match=error):
+        build_periodic_report_trigger_decision(
+            _trigger_request(
+                _candidate(
+                    "bounded_segment_milestone",
+                    source_ref="segment:2026-w29",
+                    evidence_digest="sha256:invalid-segment",
+                    facts=facts,
+                )
+            )
+        )
+
+
 def test_urgent_blocker_bypasses_cooldown_and_coalesces_material_decision() -> None:
     request = _trigger_request(
         _candidate(