fix(daemon): persist judgment scheduler receipts (#3896)

docs/plans/layering.yaml

@@ -100,7 +100,8 @@ writer_modules:
         upsert_blackboard_note, upsert_comparative_judgment_assertion, upsert_correction, upsert_mark,
         upsert_pathology_findings_as_assertions,
         upsert_recall_pack, upsert_saved_view, upsert_session_metadata_assertion, upsert_session_tag_assertion,
-        upsert_suppression, upsert_transform_candidate_assertions, upsert_findings_as_assertions, upsert_workspace]
+        upsert_suppression, upsert_transform_candidate_assertions, upsert_findings_as_assertions, upsert_workspace,
+        ack_judgment_automation_receipt_outbox, upsert_judgment_automation_receipt_outbox]
     - path: polylogue/storage/sqlite/archive_tiers/user_annotations.py
       surfaces:
         - tier: user

polylogue/api/archive.py

@@ -1508,6 +1508,10 @@ def _archive_assertion_candidate_queue_health(
 ) -> AssertionCandidateQueueHealthPayload:
     """Project queue depth, retention, producer telemetry, and scheduler health."""
 
+    from polylogue.daemon.judgment_automation import (
+        is_valid_judgment_automation_receipt_payload,
+        judgment_automation_receipt_freshness_window_ms,
+    )
     from polylogue.daemon.lifecycle import DAEMON_HEARTBEAT_STALE_AFTER_SECONDS
     from polylogue.storage.sqlite.archive_tiers.user_write import (
         ASSERTION_CANDIDATE_JUDGMENT_KINDS,
@@ -1516,6 +1520,14 @@ def _archive_assertion_candidate_queue_health(
     from polylogue.surfaces.payloads import AssertionCandidateQueueHealthPayload
 
     observed_at_ms = int(datetime.now(UTC).timestamp() * 1000) if now_ms is None else now_ms
+    interval_s = getattr(config, "judgment_automation_interval_s", None)
+    if isinstance(interval_s, bool) or not isinstance(interval_s, int):
+        return AssertionCandidateQueueHealthPayload(
+            state="unavailable",
+            observed_at_ms=observed_at_ms,
+            pending_count=0,
+            caveats=("judgment scheduler interval authority is unavailable; freshness is unverified",),
+        )
     archive_root = _active_archive_root(config)
     user_db = archive_root / "user.db"
     if not user_db.exists():
@@ -1607,6 +1619,9 @@ def _archive_assertion_candidate_queue_health(
     producer_debt_count = 0
     scheduler_state: Literal["fresh", "stale", "stopped", "unknown"] = "unknown"
     scheduler_heartbeat_at_ms: int | None = None
+    judgment_scheduler_receipt_status: Literal["completed", "parked", "failed", "unknown"] = "unknown"
+    judgment_scheduler_receipt_at_ms: int | None = None
+    judgment_scheduler_receipt_reason: str | None = None
     caveats: list[str] = []
     ops_db = archive_root / "ops.db"
     if not ops_db.exists():
@@ -1661,6 +1676,38 @@ def _archive_assertion_candidate_queue_health(
                             scheduler_state = "fresh"
                         else:
                             scheduler_state = "stale"
+                if ops_conn.execute(
+                    "SELECT 1 FROM sqlite_master WHERE type='table' AND name='daemon_events'"
+                ).fetchone():
+                    receipt_row = ops_conn.execute(
+                        """
+                        SELECT ts_ms, payload_json
+                        FROM daemon_events
+                        WHERE kind = 'judgment-automation'
+                        ORDER BY id DESC
+                        LIMIT 1
+                        """
+                    ).fetchone()
+                    if receipt_row is not None:
+                        try:
+                            receipt_payload = json.loads(str(receipt_row[1]))
+                        except (TypeError, ValueError):
+                            receipt_payload = None
+                        receipt_is_valid = is_valid_judgment_automation_receipt_payload(receipt_payload)
+                        raw_status = (
+                            str(receipt_payload.get("status", "unknown"))
+                            if receipt_is_valid and isinstance(receipt_payload, dict)
+                            else "unknown"
+                        )
+                        if not receipt_is_valid:
+                            caveats.append("latest judgment scheduler receipt is malformed")
+                        if raw_status in {"completed", "parked", "failed"}:
+                            judgment_scheduler_receipt_status = cast(
+                                Literal["completed", "parked", "failed"], raw_status
+                            )
+                        judgment_scheduler_receipt_at_ms = int(receipt_row[0])
+                        if receipt_is_valid and isinstance(receipt_payload, dict):
+                            judgment_scheduler_receipt_reason = str(receipt_payload["reason"])
             finally:
                 ops_conn.close()
         except sqlite3.Error as exc:
@@ -1675,12 +1722,43 @@ def _archive_assertion_candidate_queue_health(
         and producer_age_ms is not None
         and producer_age_ms <= 24 * 60 * 60 * 1000
     )
+    judgment_receipt_age_ms = (
+        None if judgment_scheduler_receipt_at_ms is None else max(0, observed_at_ms - judgment_scheduler_receipt_at_ms)
+    )
+    receipt_freshness_window_ms = judgment_automation_receipt_freshness_window_ms(interval_s)
+    parked_receipt_freshness_window_ms = judgment_automation_receipt_freshness_window_ms(interval_s, parked=True)
+    judgment_receipt_fresh = (
+        judgment_scheduler_receipt_status == "completed"
+        and judgment_receipt_age_ms is not None
+        and judgment_receipt_age_ms <= receipt_freshness_window_ms
+    )
+    judgment_parked_receipt_fresh = (
+        judgment_scheduler_receipt_status == "parked"
+        and judgment_receipt_age_ms is not None
+        and judgment_receipt_age_ms <= parked_receipt_freshness_window_ms
+    )
 
     state: AssertionCandidateQueueState
     if producer_debt_count or producer_status in failed_producer_statuses or scheduler_state in {"stale", "stopped"}:
         state = "producer-stalled"
     elif stale_pending_count:
         state = "stale-pending"
+    elif pending_count and judgment_scheduler_receipt_status == "parked" and not judgment_parked_receipt_fresh:
+        state = "scheduler-stalled"
+        caveats.append(
+            "judgment scheduler has no fresh parked receipt; the bounded retry route is the next daemon tick"
+        )
+    elif pending_count and judgment_scheduler_receipt_status in {"parked", "unknown"}:
+        state = "parked-pending"
+        if judgment_scheduler_receipt_status == "parked":
+            caveats.append("judgment scheduler is parked; the bounded retry route is the next enabled daemon tick")
+        else:
+            caveats.append("no judgment scheduler receipt is observable; pending candidates are not converged")
+    elif pending_count and (judgment_scheduler_receipt_status == "failed" or not judgment_receipt_fresh):
+        state = "scheduler-stalled"
+        caveats.append(
+            "judgment scheduler has no fresh successful receipt; the bounded retry route is the next daemon tick"
+        )
     elif pending_count:
         state = "pending"
     elif producer_fresh and scheduler_state == "fresh":
@@ -1710,6 +1788,10 @@ def _archive_assertion_candidate_queue_health(
         scheduler_state=scheduler_state,
         scheduler_heartbeat_at_ms=scheduler_heartbeat_at_ms,
         scheduler_heartbeat_age_ms=heartbeat_age_ms,
+        judgment_scheduler_receipt_status=judgment_scheduler_receipt_status,
+        judgment_scheduler_receipt_at_ms=judgment_scheduler_receipt_at_ms,
+        judgment_scheduler_receipt_age_ms=judgment_receipt_age_ms,
+        judgment_scheduler_receipt_reason=judgment_scheduler_receipt_reason,
         producer_debt_count=producer_debt_count,
         caveats=tuple(dict.fromkeys(caveats)),
     )

polylogue/cli/commands/status.py

@@ -6,6 +6,7 @@
 import sqlite3
 import time
 from collections.abc import Sequence
+from contextlib import suppress
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -1339,6 +1340,10 @@ def _show_daemon_status(env: AppEnv, status: dict[str, Any], *, compact: bool =
     if isinstance(raw_frontier, dict):
         _render_raw_frontier_integrity(env, raw_frontier)
 
+    assertion_candidate_queue = status.get("assertion_candidate_queue")
+    if isinstance(assertion_candidate_queue, dict):
+        _render_assertion_candidate_queue(env, assertion_candidate_queue)
+
     # Sizes
     db_bytes = status.get("db_size_bytes", 0)
     disk_free = status.get("disk_free_bytes", 0)
@@ -1446,6 +1451,10 @@ def _compact_status_payload(status: dict[str, Any], *, source: str) -> dict[str,
     if archive_debt:
         payload["archive_debt"] = archive_debt
 
+    assertion_candidate_queue = status.get("assertion_candidate_queue")
+    if isinstance(assertion_candidate_queue, dict):
+        payload["assertion_candidate_queue"] = assertion_candidate_queue
+
     raw_materialization = _compact_mapping_without(
         status.get("raw_materialization_readiness"),
         {"sampled_rows"},
@@ -1702,6 +1711,28 @@ def _show_direct_json(
     raw_materialization_readiness = _direct_raw_materialization_readiness(active_root)
     raw_frontier_integrity = _direct_raw_frontier_integrity(active_root, raw_materialization_readiness)
     raw_failure_status = _direct_raw_failure_status(root)
+    from polylogue.config import Config, resolve_runtime_config
+    from polylogue.daemon.status import assertion_candidate_queue_status_summary
+    from polylogue.paths import render_root
+
+    try:
+        resolved_runtime_config = resolve_runtime_config().as_config()
+    except Exception as exc:
+        assertion_candidate_queue = {
+            "mode": "assertion-candidate-queue-health",
+            "state": "unavailable",
+            "pending_count": 0,
+            "caveats": [f"queue health configuration unavailable: {exc}"],
+        }
+    else:
+        queue_config = Config(
+            archive_root=active_root,
+            render_root=render_root(),
+            sources=[],
+            db_path=active_db if active_db is not None else active_root / "index.db",
+            judgment_automation_interval_s=resolved_runtime_config.judgment_automation_interval_s,
+        )
+        assertion_candidate_queue = assertion_candidate_queue_status_summary(config=queue_config)
     component_readiness = _direct_component_readiness(
         env,
         active_root=active_root,
@@ -1736,6 +1767,7 @@ def _show_direct_json(
         "archive_facade_routes": _archive_facade_route_status(),
         "archive_cli_routes": _archive_cli_route_status(),
         "archive_runtime_paths": _archive_runtime_path_status(),
+        "assertion_candidate_queue": assertion_candidate_queue,
         "raw_materialization_readiness": raw_materialization_readiness,
         "raw_frontier_integrity": raw_frontier_integrity,
         "component_readiness": component_readiness,
@@ -2471,13 +2503,32 @@ def _render_assertion_candidate_queue(env: AppEnv, queue: dict[str, Any]) -> Non
     state = str(queue.get("state") or "unavailable")
     pending = _safe_int(queue.get("pending_count"))
     color = "green" if state == "healthy-empty" else "yellow"
-    if state in {"producer-stalled", "stale-pending", "unavailable"}:
+    if state in {"producer-stalled", "scheduler-stalled", "parked-pending", "stale-pending", "unavailable"}:
         color = "red"
     line = f"  Assertion candidate queue: [{color}]{state}, {pending} pending[/{color}]"
     oldest_age = queue.get("oldest_pending_age_ms")
     if isinstance(oldest_age, int | float):
         line += f", oldest={float(oldest_age) / (24 * 60 * 60 * 1000):.1f}d"
     env.ui.console.print(line)
+    receipt_status = queue.get("judgment_scheduler_receipt_status")
+    if receipt_status is not None:
+        receipt_details = [str(receipt_status)]
+        receipt_at_ms = queue.get("judgment_scheduler_receipt_at_ms")
+        if isinstance(receipt_at_ms, int | float) and not isinstance(receipt_at_ms, bool):
+            with suppress(OverflowError, OSError, ValueError):
+                receipt_timestamp = datetime.fromtimestamp(float(receipt_at_ms) / 1000, tz=UTC).isoformat()
+                receipt_details.append(f"at={receipt_timestamp}")
+        receipt_age_ms = queue.get("judgment_scheduler_receipt_age_ms")
+        if isinstance(receipt_age_ms, int | float) and not isinstance(receipt_age_ms, bool):
+            receipt_age_s = float(receipt_age_ms) / 1000
+            if receipt_age_s >= 24 * 60 * 60:
+                receipt_details.append(f"age={receipt_age_s / (24 * 60 * 60):.1f}d")
+            else:
+                receipt_details.append(f"age={receipt_age_s:.1f}s")
+        receipt_reason = queue.get("judgment_scheduler_receipt_reason")
+        if receipt_reason:
+            receipt_details.append(f"reason={receipt_reason}")
+        env.ui.console.print(f"    judgment scheduler receipt: {', '.join(receipt_details)}")
 
 
 def _render_sqlite_maintenance(env: AppEnv, status: dict[str, Any]) -> None:

polylogue/config.py

@@ -28,6 +28,9 @@ class ConfigError(PolylogueError):
     """Configuration error."""
 
 
+JUDGMENT_AUTOMATION_BATCH_LIMIT_DEFAULT = 200
+
+
 @dataclass(frozen=True, slots=True)
 class Source:
     """A session source (local path, Drive folder, or both)."""
@@ -102,6 +105,7 @@ class Config:
     index_config: IndexConfig | None = None
     embedding_model: str = "voyage-4-lite"
     embedding_dimension: int = 1024
+    judgment_automation_interval_s: int = 3600
 
     def __init__(
         self,
@@ -113,6 +117,7 @@ def __init__(
         index_config: IndexConfig | None = None,
         embedding_model: str = "voyage-4-lite",
         embedding_dimension: int = 1024,
+        judgment_automation_interval_s: int = 3600,
     ) -> None:
         self.archive_root = archive_root
         self.render_root = render_root
@@ -122,12 +127,15 @@ def __init__(
         self.index_config = index_config
         self.embedding_model = embedding_model
         self.embedding_dimension = embedding_dimension
+        self.judgment_automation_interval_s = judgment_automation_interval_s
         for attr in ("archive_root", "render_root", "db_path"):
             value = getattr(self, attr)
             if not isinstance(value, Path):
                 raise ConfigError(f"Config.{attr} must be a Path, got {type(value).__name__}")
             if not value.is_absolute():
                 raise ConfigError(f"Config.{attr} must be an absolute path, got {value!r}")
+        if isinstance(judgment_automation_interval_s, bool) or not isinstance(judgment_automation_interval_s, int):
+            raise ConfigError("Config.judgment_automation_interval_s must be an integer")
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, Config):
@@ -141,14 +149,16 @@ def __eq__(self, other: object) -> bool:
             and self.index_config == other.index_config
             and self.embedding_model == other.embedding_model
             and self.embedding_dimension == other.embedding_dimension
+            and self.judgment_automation_interval_s == other.judgment_automation_interval_s
         )
 
     def __repr__(self) -> str:
         return (
             f"Config(archive_root={self.archive_root!r}, render_root={self.render_root!r}, "
             f"sources={self.sources!r}, db_path={self.db_path!r}, "
             f"drive_config={self.drive_config!r}, index_config={self.index_config!r}, "
-            f"embedding_model={self.embedding_model!r}, embedding_dimension={self.embedding_dimension!r})"
+            f"embedding_model={self.embedding_model!r}, embedding_dimension={self.embedding_dimension!r}, "
+            f"judgment_automation_interval_s={self.judgment_automation_interval_s!r})"
         )
 
     def with_sources(self, sources: list[Source]) -> Config:
@@ -161,6 +171,7 @@ def with_sources(self, sources: list[Source]) -> Config:
             index_config=self.index_config,
             embedding_model=self.embedding_model,
             embedding_dimension=self.embedding_dimension,
+            judgment_automation_interval_s=self.judgment_automation_interval_s,
         )
 
 
@@ -832,7 +843,16 @@ def judgment_automation_interval_s(self) -> int:
     @property
     def judgment_automation_batch_limit(self) -> int:
         """Maximum candidates judged per judgment-automation sweep (polylogue-6qjc)."""
-        return int(str(self._data.get("judgment_automation_batch_limit", 200)))
+        raw_value = self._data.get("judgment_automation_batch_limit", JUDGMENT_AUTOMATION_BATCH_LIMIT_DEFAULT)
+        if isinstance(raw_value, bool):
+            raise ConfigError("judgment_automation_batch_limit must be a positive integer")
+        try:
+            value = int(str(raw_value).strip(), 10)
+        except (TypeError, ValueError) as exc:
+            raise ConfigError("judgment_automation_batch_limit must be a positive integer") from exc
+        if value <= 0:
+            raise ConfigError("judgment_automation_batch_limit must be a positive integer")
+        return value
 
     @property
     def judgment_automation_policy(self) -> dict[str, object]:
@@ -2173,6 +2193,7 @@ def as_config(self) -> Config:
             index_config=self.index_config,
             embedding_model=self.settings.embedding_model,
             embedding_dimension=self.settings.embedding_dimension,
+            judgment_automation_interval_s=self.settings.judgment_automation_interval_s,
         )
 
 
@@ -2898,6 +2919,7 @@ def format_config_toml(cfg: dict[str, object]) -> str:
     "DEFAULT_SITE_CONFIG_PATH",
     "DriveConfig",
     "IndexConfig",
+    "JUDGMENT_AUTOMATION_BATCH_LIMIT_DEFAULT",
     "PolylogueConfig",
     "SECRET_CONFIG_KEYS",
     "SECRET_SET_PLACEHOLDER",

polylogue/daemon/cli.py

@@ -2513,7 +2513,10 @@ async def run_daemon_services(
                 periodic_embedding_orphan_reconcile_check(catch_up_complete=catch_up_complete_gate),
                 _periodic_db_optimize(),
                 _periodic_status_snapshot_refresh(),
-                periodic_judgment_automation_sweep(catch_up_complete=catch_up_complete_gate),
+                periodic_judgment_automation_sweep(
+                    catch_up_complete=catch_up_complete_gate,
+                    archive_root_path=archive_root_path,
+                ),
                 periodic_fts_identity_drift_recompute(catch_up_complete=catch_up_complete_gate),
                 periodic_fts_orphan_audit(catch_up_complete=catch_up_complete_gate),
                 periodic_blob_gc_check(catch_up_complete=catch_up_complete_gate),