fix: status readers and readiness reds from the 2026-09-05 corpus

devtools/command_catalog.py

@@ -137,15 +137,6 @@ def to_dict(self) -> dict[str, object]:
         examples=("devtools verify", "devtools verify --quick", "devtools verify --all"),
         featured=True,
     ),
-    CommandSpec(
-        "schema-manifest",
-        "verification",
-        "Compare canonical SQLite schema manifests with archive tier files.",
-        "devtools.verify_schema_manifest",
-        json_flag=True,
-        use_when="Verify every tier's canonical create route and optionally compare it with an archive root.",
-        examples=("devtools schema-manifest", "devtools schema-manifest --archive-root /path/to/archive --json"),
-    ),
     CommandSpec(
         "gate",
         "verification",

docs/devtools.md

@@ -70,7 +70,6 @@ These are the commands worth remembering during normal repo work:
 | --- | --- |
 | `devtools gate` | Run one named invariant check. |
 | `devtools scenario` | Run a named archive verification scenario. |
-| `devtools schema-manifest` | Compare canonical SQLite schema manifests with archive tier files. |
 | `devtools smoke` | Probe deployed Polylogue binaries, daemon/web routes, and browser-capture archive flow. |
 | `devtools verify` | Run the local verification baseline: every quick gate, then the selected or complete test corpus. |
 

polylogue/cli/commands/status.py

@@ -244,7 +244,14 @@ def _status_operation_result(
     from polylogue.daemon.socket_path import daemon_socket_path
     from polylogue.version import POLYLOGUE_VERSION
 
-    config = load_effective_config(env) if daemon_url in (None, _BUILTIN_DAEMON_URL) else None
+    config: Any = None
+    if daemon_url in (None, _BUILTIN_DAEMON_URL):
+        try:
+            config = load_effective_config(env)
+        except Exception:
+            # Without a resolved config there is no socket to address, but a
+            # daemon on a discovered API port is still reachable over HTTP.
+            config = None
     client = (
         DaemonClient(
             daemon_socket_path(config.archive_root),
@@ -260,15 +267,22 @@ def _status_operation_result(
     request = OperationRequest("status", {"include_archive_readiness": include_archive_readiness})
 
     def daemon_call(lowered: Any) -> Any:
-        if daemon_url is None or daemon_url == _BUILTIN_DAEMON_URL:
-            assert config is not None and client is not None
-            return client.operation(
-                lowered.operation,
-                dict(lowered.payload),
-                archive_root=str(config.archive_root),
-                daemon_version=POLYLOGUE_VERSION,
-            )
-        for candidate in _candidate_daemon_urls(daemon_url):
+        if client is not None and config is not None:
+            try:
+                uds_value = client.operation(
+                    lowered.operation,
+                    dict(lowered.payload),
+                    archive_root=str(config.archive_root),
+                    daemon_version=POLYLOGUE_VERSION,
+                )
+            except Exception:
+                uds_value = None
+            if uds_value is not None:
+                return uds_value
+            # No socket for this archive root does not mean no daemon: a
+            # dev-loop daemon on a discovered API port still answers over
+            # HTTP, which is what _candidate_daemon_urls is for.
+        for candidate in _candidate_daemon_urls(daemon_url or _BUILTIN_DAEMON_URL):
             try:
                 request = Request(f"{candidate}/api/status", headers={"Accept": "application/json"}, method="GET")
                 with urlopen(request, timeout=_FULL_TIMEOUT_S) as response:
@@ -278,15 +292,10 @@ def daemon_call(lowered: Any) -> Any:
                 continue
         return None
 
-    return OperationKernel(
-        daemon_call,
-        lambda _lowered: _show_direct_json(
-            env,
-            full=True,
-            include_archive_readiness=include_archive_readiness,
-            emit=False,
-        ),
-    ).execute(request)
+    # No direct fallback here: both callers render the direct surface
+    # themselves, so a fallback would assemble the whole direct payload once
+    # to be discarded. An unreachable daemon raises OperationKernelError.
+    return OperationKernel(daemon_call).execute(request)
 
 
 def _candidate_daemon_urls(primary_url: str) -> tuple[str, ...]:
@@ -492,7 +501,7 @@ def _archive_one_tier_status(tier: str, path: Path) -> dict[str, Any]:
         return status
 
     try:
-        conn = open_readonly_connection(path, timeout_class="interactive-read")
+        conn = _open_status_connection(path)
         try:
             counts, precision = _archive_table_counts(conn, _ARCHIVE_TIER_TABLES[tier], db_size_bytes=probe.size_bytes)
             status["table_counts"] = counts
@@ -583,6 +592,17 @@ def _sqlite_stat1_rows(conn: sqlite3.Connection) -> int:
     return _fast_count(conn, "SELECT COUNT(*) FROM sqlite_stat1")
 
 
+def _open_status_connection(path: Path) -> sqlite3.Connection:
+    """Open a tier read-only for reporting.
+
+    Status reports a skewed or unstamped tier as data. Validating the schema
+    on the way in raises the whole status surface out of service on exactly
+    the archive whose state the operator is asking about, so these reads
+    never validate.
+    """
+    return open_readonly_connection(path, timeout_class="interactive-read", validate_schema=False)
+
+
 def _sqlite_maintenance_status(root: Path) -> dict[str, Any]:
     tiers: dict[str, dict[str, Any]] = {}
     total_wal_bytes = 0
@@ -598,7 +618,7 @@ def _sqlite_maintenance_status(root: Path) -> dict[str, Any]:
         total_wal_bytes += int(tier_status["wal_bytes"])
         if path.exists():
             try:
-                conn = open_readonly_connection(path, timeout_class="interactive-read")
+                conn = _open_status_connection(path)
                 try:
                     stat_rows = _sqlite_stat1_rows(conn)
                 finally:
@@ -678,7 +698,7 @@ def _archive_source_table_count(conn: Any, *, table: str, sql: str, configured_r
         if not source_db.exists():
             return -1
         try:
-            source_conn = open_readonly_connection(source_db, timeout_class="interactive-read")
+            source_conn = _open_status_connection(source_db)
             try:
                 if not _table_exists(source_conn, table):
                     return -1
@@ -698,7 +718,7 @@ def _archive_source_table_count(conn: Any, *, table: str, sql: str, configured_r
     if not source_db.exists():
         return -1
     try:
-        source_conn = open_readonly_connection(source_db, timeout_class="interactive-read")
+        source_conn = _open_status_connection(source_db)
         try:
             if not _table_exists(source_conn, table):
                 return -1
@@ -730,7 +750,7 @@ def _ops_workload_status(active_root: Path, *, now_ms: int) -> dict[str, Any]:
     if not ops_db.exists():
         return {"available": False, "reason": "missing_ops_tier"}
     try:
-        conn = open_readonly_connection(ops_db, timeout_class="interactive-read")
+        conn = _open_status_connection(ops_db)
     except sqlite3.Error as exc:
         return {"available": False, "reason": f"ops workload status unavailable: {exc}"}
     try:
@@ -999,6 +1019,22 @@ def status_command(
             )
         except Exception:
             operation_result = None
+        # An unparseable or timed-out /api/status is not a stopped daemon. When
+        # the named daemon still answers its liveness probe, say so instead of
+        # publishing local archive facts as daemon truth.
+        daemon_answered = operation_result is not None and operation_result.authority.get("mode") != "direct"
+        if (
+            not daemon_answered
+            and daemon_url not in (None, _BUILTIN_DAEMON_URL)
+            and _daemon_live(daemon_url, timeout=_FAST_TIMEOUT_S)
+        ):
+            obs.attributes["daemon_reachable"] = True
+            obs.daemon_path = "daemon"
+            if output_format == "json":
+                _show_daemon_status_unavailable_json(env)
+            else:
+                _show_daemon_status_unavailable(env, compact=not full_payload)
+            raise click.exceptions.Exit(1)
         if operation_result is None:
             obs.attributes["daemon_reachable"] = False
             obs.daemon_path = "direct"
@@ -1886,7 +1922,7 @@ def _direct_assertion_component(active_root: Path) -> dict[str, Any]:
         return component_from_assertion_substrate(table_exists=False).to_dict()
 
     try:
-        conn = open_readonly_connection(user_db, timeout_class="interactive-read")
+        conn = _open_status_connection(user_db)
         try:
             table_exists = _table_exists(conn, "assertions")
             if not table_exists:

polylogue/daemon/fts_status.py

@@ -259,8 +259,11 @@ def _archive_blocks_surface(conn: sqlite3.Connection) -> dict[str, int | bool |
         if source_exists and recorded_state == "ready" and source_rows == 0 and indexed_rows == 0
         else False
     )
+    # No ledger table at all is a legacy tier: readiness is structural. A
+    # present ledger that holds no row for this surface is an unmeasured
+    # surface, and stays not-ready.
     freshness_ready = (
-        False
+        True
         if freshness_records is None
         else freshness_ready_record_trusted(
             state=recorded_state,
@@ -277,11 +280,13 @@ def _archive_blocks_surface(conn: sqlite3.Connection) -> dict[str, int | bool |
             source_has_rows=source_has_rows,
         )
     )
-    # Ledger state is a cache of a measurement, not authority to refuse a
-    # read. Re-measure a non-trusted recorded scope against the rows the
-    # executor will actually search; a deferred single-session observation
-    # must not masquerade as archive-wide incompleteness.
-    if freshness is not None and not freshness_ready:
+    # A recorded state below `ready` is the ledger admitting it holds no
+    # current measurement: re-measure once against the rows the executor will
+    # actually search, so a deferred single-session observation cannot
+    # masquerade as archive-wide incompleteness. A recorded `ready` that fails
+    # the trust check already carries counts and a verdict — report it stale or
+    # unknown, because the request-safe path must not scan the source.
+    if freshness is not None and not freshness_ready and recorded_state != "ready":
         measured = _archive_exact_blocks_surface(conn)
         measured.update(
             {

tests/infra/cli_interaction.py

@@ -111,6 +111,7 @@ def _owners(values: Iterable[str], owner: str) -> dict[str, str]:
     "hooks": "tests/unit/cli/test_hooks.py",
     "import": "tests/unit/cli/test_import.py",
     "init": "tests/unit/cli/test_init.py",
+    "insights": "tests/unit/cli/test_insights.py",
     "judge": "tests/unit/cli/test_judge_command.py",
     "manual": "tests/unit/cli/test_manual_command.py",
     "mark": "tests/unit/cli/test_mark_note_identity.py",

tests/infra/local_timezone.py

@@ -0,0 +1,59 @@
+"""Pin a named local timezone for tests that render host-local times.
+
+`datetime.astimezone()` resolves the local zone through libc, which finds a
+named zone only when a zoneinfo database is on disk at a location it knows.
+This host ships no `/usr/share/zoneinfo`, so libc resolves names only via
+`TZDIR`, and an environment that does not forward `TZDIR` silently degrades
+`TZ=America/Los_Angeles` to a POSIX zero-offset zone named `America`. Deriving
+`TZDIR` from `zoneinfo.TZPATH` keeps the pin independent of ambient state.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+import zoneinfo
+from collections.abc import Iterator
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+
+def _zoneinfo_directory(zone: str) -> str | None:
+    for candidate in zoneinfo.TZPATH:
+        if (Path(candidate) / zone).exists():
+            return candidate
+    return None
+
+
+@contextmanager
+def pinned_local_timezone(monkeypatch: pytest.MonkeyPatch, zone: str) -> Iterator[None]:
+    """Make libc-local time render `zone`, restoring the prior zone after."""
+    previous_tz = os.environ.get("TZ")
+    previous_tzdir = os.environ.get("TZDIR")
+    directory = _zoneinfo_directory(zone)
+    if directory is not None:
+        monkeypatch.setenv("TZDIR", directory)
+    monkeypatch.setenv("TZ", zone)
+    time.tzset()
+    probe = datetime(2026, 7, 1, 12, 0, tzinfo=timezone.utc)
+    expected = probe.astimezone(zoneinfo.ZoneInfo(zone)).utcoffset()
+    if probe.astimezone().utcoffset() != expected:
+        raise AssertionError(
+            f"could not pin local timezone to {zone!r}: libc reported {time.tzname!r}; "
+            "no zoneinfo database was reachable"
+        )
+    try:
+        yield
+    finally:
+        for key, value in (("TZ", previous_tz), ("TZDIR", previous_tzdir)):
+            if value is None:
+                monkeypatch.delenv(key, raising=False)
+            else:
+                monkeypatch.setenv(key, value)
+        time.tzset()
+
+
+__all__ = ["pinned_local_timezone"]

tests/infra/test_local_timezone.py

@@ -0,0 +1,34 @@
+"""Anti-vacuity: without this helper the pin silently degrades.
+
+Deleting the `TZDIR` derivation in `pinned_local_timezone` turns
+`TZ=America/Los_Angeles` into a zero-offset POSIX zone on a host with no
+`/usr/share/zoneinfo`, and `test_pin_holds_without_an_ambient_zoneinfo_directory`
+goes red.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from tests.infra.local_timezone import pinned_local_timezone
+
+
+def _rendered(monkeypatch: pytest.MonkeyPatch) -> str:
+    with pinned_local_timezone(monkeypatch, "America/Los_Angeles"):
+        return datetime(2026, 7, 1, 19, 48, tzinfo=timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M %Z")
+
+
+def test_pin_holds_without_an_ambient_zoneinfo_directory(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("TZDIR", raising=False)
+
+    assert _rendered(monkeypatch) == "2026-07-01 12:48 PDT"
+
+
+def test_pin_restores_the_previous_zone(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("TZ", "UTC")
+
+    _rendered(monkeypatch)
+
+    assert datetime(2026, 7, 1, 19, 48, tzinfo=timezone.utc).astimezone().utcoffset() == timedelta(0)

tests/unit/cli/test_insights.py

@@ -589,7 +589,10 @@ def test_insights_status_plain(cli_workspace: CliWorkspace) -> None:
 
     assert result.exit_code == 0
     assert "Convergence: caught up" in result.output
-    assert "session_profiles: degraded" in result.output
+    # Readiness is convergence plus counts: one row per seeded session, and the
+    # expected denominator gated on the product table. A reader that stopped
+    # reporting either would print a bare name.
+    assert "session_profiles: rows=2 expected=2" in result.output
 
 
 def test_insights_hermes_health_json_reports_disabled_without_a_hermes_root(

tests/unit/cli/test_insights_command_runtime.py

@@ -194,7 +194,7 @@ def test_render_status_plain_and_export_plain_cover_optional_sections(
     output = capsys.readouterr().out
     assert "Convergence: debt in derived" in output
     assert "Scope: origin=codex-session since=2026-04-01 until=2026-04-30" in output
-    assert "session_profiles: partial rows=7 expected=10" in output
+    assert "session_profiles: rows=7 expected=10" in output
     assert "missing=1 stale=2 orphan=3 incompatible=4" in output
     assert "origins: codex-session=7" in output
     assert "versions: materializer_version={'4': 7}" in output