feat(resolvers): binky_rhythm_status query in state-hub resolver

Due-items for the three Binky operating-rhythm definitions
(BINKY-WP-0004-T02). Dueness derives from hub progress events recorded
by the executing session (binky_daily_brief / binky_mail_intake /
binky_weekly_review, detail.repo scoped); weekly_review carries
milestone_moved from event_type=milestone events in the last 7 days
(RISK-005 signal). Definitions' resolver comments updated; definitions
stay enabled:false until cutover (BINKY-WP-0004-T06).

7 new tests; resolver test file 31/31 green. Pre-existing failures in
test_railiance_ops_inventory_wiring/test_schedule_health are untouched.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/activity_core/context_resolvers/state_hub.py

@@ -17,6 +17,8 @@ Supported queries:
   - consistency_sweep_remote_all: POST {STATE_HUB_URL}/consistency/sweep/remote-all
   - phase5_stabilization_check: hub-visible Phase 5 stabilization gates
   - legacy_meter_weekly_review: GET {STATE_HUB_URL}/legacy-meter/weekly-review
+  - binky_rhythm_status: due-items for the Binky operating-rhythm definitions,
+                      derived from /progress/ events (see _binky_rhythm_status)
 
 When STATE_HUB_URL points at the state-hub edge relay, allowlisted GET reads may
 be served from a stale local cache during upstream outages (`X-StateHub-Edge-Cache:
@@ -154,6 +156,8 @@ class StateHubContextResolver(ContextResolver):
             return _phase5_stabilization_check(params)
         if query == "legacy_meter_weekly_review":
             return _legacy_meter_weekly_review(params)
+        if query == "binky_rhythm_status":
+            return _binky_rhythm_status(params)
         return {}
 
 
@@ -358,6 +362,114 @@ def _phase5_stabilization_check(params: dict[str, Any]) -> dict[str, Any]:
 CONTEXT_RESOLVER_REGISTRY["state-hub"] = StateHubContextResolver
 
 
+# Rhythm kinds understood by binky_rhythm_status, mapped to the progress
+# event_type an executing session records on completion. A run is only
+# "seen" by the idempotence guard if the session logs that event with
+# detail.repo set to the target repo.
+_BINKY_RHYTHM_EVENT_TYPES = {
+    "daily_brief": "binky_daily_brief",
+    "mail_intake": "binky_mail_intake",
+    "weekly_review": "binky_weekly_review",
+}
+_BINKY_MAIL_INTAKE_WINDOW_DAYS = 7
+_BINKY_MILESTONE_WINDOW_DAYS = 7
+
+
+def _binky_rhythm_status(params: dict[str, Any]) -> dict[str, Any]:
+    """Due-items for the Binky operating-rhythm definitions.
+
+    Params: repo (default "binky-control"), kind (optional: one of
+    _BINKY_RHYTHM_EVENT_TYPES; absent = daily_brief), timezone (IANA name,
+    default Europe/Berlin — "today" is evaluated in this zone).
+
+    Dueness is derived from hub progress events, which the executing session
+    records at completion (event_type per _BINKY_RHYTHM_EVENT_TYPES, with
+    detail.repo naming the target repo):
+      - daily_brief:   due when no completion event exists for today
+      - mail_intake:   due when no completion event exists in the last 7 days
+      - weekly_review: due when no completion event exists for today; the item
+        additionally carries milestone_moved — whether an event_type=milestone
+        event for the repo was recorded in the last 7 days (RISK-005 signal)
+
+    Events whose detail.repo names a different repo are ignored; events with
+    no detail.repo are accepted (the binky-prefixed event types are already
+    repo-specific in practice).
+    """
+    repo = str(params.get("repo") or "binky-control")
+    kind = str(params.get("kind") or "daily_brief")
+    if kind not in _BINKY_RHYTHM_EVENT_TYPES:
+        return {"items": [], "repo": repo, "error": f"unknown kind: {kind}"}
+
+    tz = _binky_timezone(params.get("timezone"))
+    now_local = _utc_now().astimezone(tz)
+    today = now_local.date()
+
+    event_type = _BINKY_RHYTHM_EVENT_TYPES[kind]
+    events = _binky_progress_events(event_type, repo)
+    # _parse_progress_timestamp returns a datetime.min sentinel for missing or
+    # unparsable created_at; those events cannot prove a run happened.
+    timestamps = [
+        ts
+        for ts in (
+            _parse_progress_timestamp(item.get("created_at")) for item in events
+        )
+        if ts > datetime.min.replace(tzinfo=timezone.utc)
+    ]
+    last_run = max(timestamps, default=None)
+
+    if kind == "mail_intake":
+        due = last_run is None or (
+            (_utc_now() - last_run).days >= _BINKY_MAIL_INTAKE_WINDOW_DAYS
+        )
+    else:
+        due = last_run is None or last_run.astimezone(tz).date() != today
+
+    item: dict[str, Any] = {
+        "kind": kind,
+        "due": due,
+        "date": today.isoformat(),
+        "last_run_at": last_run.isoformat() if last_run else None,
+    }
+    if kind == "weekly_review":
+        milestone_events = _binky_progress_events("milestone", repo)
+        cutoff = _utc_now().timestamp() - _BINKY_MILESTONE_WINDOW_DAYS * 86400
+        item["milestone_moved"] = any(
+            _parse_progress_timestamp(event.get("created_at")).timestamp() > cutoff
+            for event in milestone_events
+        )
+
+    return {
+        "items": [item],
+        "repo": repo,
+        "generated_at": _utc_now().isoformat(),
+    }
+
+
+def _binky_timezone(raw: Any) -> Any:
+    from zoneinfo import ZoneInfo
+
+    name = str(raw or "Europe/Berlin")
+    try:
+        return ZoneInfo(name)
+    except Exception:
+        return timezone.utc
+
+
+def _binky_progress_events(event_type: str, repo: str) -> list[dict[str, Any]]:
+    items = _fetch_json("/progress/", {"event_type": event_type, "limit": 50})
+    if not isinstance(items, list):
+        return []
+    matched: list[dict[str, Any]] = []
+    for item in items:
+        if not isinstance(item, dict) or item.get("event_type") != event_type:
+            continue
+        detail_repo = _progress_detail(item).get("repo")
+        if detail_repo and str(detail_repo) != repo:
+            continue
+        matched.append(item)
+    return matched
+
+
 def _repo_sbom_status(params: dict[str, Any]) -> dict[str, Any]:
     """Resolve SBOM staleness against the State Hub.