diff --git a/devtools/command_catalog.py b/devtools/command_catalog.py index 2573840546..c12a6f2b24 100644 --- a/devtools/command_catalog.py +++ b/devtools/command_catalog.py @@ -759,6 +759,24 @@ def to_dict(self) -> dict[str, object]: "devtools bench slo --skip-benchmarks --json", ), ), + CommandSpec( + "bench help-latency", + "benchmarking", + "Check `--help` wall-clock latency against the interactive-tier cold-CLI budget (polylogue-20d.2).", + "devtools.help_latency_probe", + use_when=( + "Catch CLI import-tax regressions continuously. Runs `polylogue --help` for a curated " + "set of root and nested subcommands as fresh subprocesses and compares the minimum wall time " + "against the 700ms cold-CLI budget from the 20d.14 interactive SLO tier. Fails when any " + "'required' target exceeds budget; 'informational' targets (currently `ops maintenance`, " + "known slow pending a lazy-import refactor) are reported but never block." + ), + examples=( + "devtools bench help-latency", + "devtools bench help-latency --json", + "devtools bench help-latency --repeats 5 --out .local/help-latency.json", + ), + ), CommandSpec( "verify manifests", "verification", diff --git a/devtools/help_latency_probe.py b/devtools/help_latency_probe.py new file mode 100644 index 0000000000..a012047a18 --- /dev/null +++ b/devtools/help_latency_probe.py @@ -0,0 +1,158 @@ +"""Measure ``--help`` wall-clock latency against the interactive-tier cold-CLI budget. + +Use ``devtools bench help-latency`` to catch import-tax regressions on the +CLI ``--help`` path continuously (polylogue-20d.2). The interactive SLO tier +(polylogue-20d.14, ``docs/plans/slo-catalog.yaml``) states a <700ms budget for +a cold (no warm daemon) CLI invocation; ``--help`` is the cheapest possible +invocation of any command (no archive I/O, no query execution) so it is the +tightest floor on Python/import overhead. A regression here means every +daemonless invocation of that command pays the same tax. + +Each target is run as a fresh subprocess (``python -m polylogue.cli ``) +several times; the *minimum* wall time is compared against budget rather than +the mean, because process-launch jitter (scheduler contention, page-cache +misses) only ever adds latency on a shared dev host, never subtracts it. This +mirrors the host-variable framing of the other ``devtools bench`` probes: +wall-clock is diagnostic and campaign-comparable, but the budget comparison +here IS a CI gate for "required" targets (unlike the wall-clock-only probes), +because import cost is deterministic given the source tree, not host load. + +Targets marked ``gate="informational"`` are measured and reported but never +fail the check — they document a known-slow path with an open follow-up +(currently: the ``ops maintenance`` command group; see the module docstring +in ``polylogue/cli/commands/maintenance.py``). +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from time import perf_counter +from typing import Literal, cast + +Gate = Literal["required", "informational"] + + +@dataclass(frozen=True, slots=True) +class HelpLatencyTarget: + label: str + args: tuple[str, ...] + budget_ms: int + gate: Gate + + +# Budget follows the 20d.14 interactive-tier "cold CLI (no daemon) <700ms" +# line in docs/plans/slo-catalog.yaml. +_DEFAULT_BUDGET_MS = 700 + +TARGETS: tuple[HelpLatencyTarget, ...] = ( + HelpLatencyTarget("root", ("--help",), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("find", ("find", "--help"), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("read", ("read", "--help"), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("mark", ("mark", "--help"), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("select", ("select", "--help"), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("analyze", ("analyze", "--help"), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("import", ("import", "--help"), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("config", ("config", "--help"), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("dashboard", ("dashboard", "--help"), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("ops", ("ops", "--help"), _DEFAULT_BUDGET_MS, "required"), + HelpLatencyTarget("reset", ("reset", "--help"), _DEFAULT_BUDGET_MS, "required"), + # `ops maintenance` (any leaf) fully imports the ~2800-line maintenance + # command module at module scope (ArchiveStore, blob_gc, blob_integrity, + # embeddings.reconcile, migration_runner, ...) because `_LazyGroup` + # resolution needs the module to enumerate subcommands. Known slow; + # informational until the module's imports are pushed into per-command + # bodies. Do not silently promote to "required" without doing that work + # first — it will flap the gate red on every host. + HelpLatencyTarget("ops-maintenance", ("ops", "maintenance", "--help"), _DEFAULT_BUDGET_MS, "informational"), + HelpLatencyTarget( + "ops-maintenance-archive-read", + ("ops", "maintenance", "archive-read", "--help"), + _DEFAULT_BUDGET_MS, + "informational", + ), +) + + +def _time_invocation(args: tuple[str, ...], *, repeats: int) -> float: + """Return the minimum wall-clock ms across ``repeats`` fresh subprocess runs.""" + + best: float | None = None + for _ in range(repeats): + started = perf_counter() + subprocess.run( + [sys.executable, "-m", "polylogue.cli", *args], + check=False, + capture_output=True, + text=True, + ) + elapsed_ms = (perf_counter() - started) * 1_000 + if best is None or elapsed_ms < best: + best = elapsed_ms + assert best is not None + return best + + +def measure(*, repeats: int = 3, targets: tuple[HelpLatencyTarget, ...] = TARGETS) -> dict[str, object]: + results = [] + for target in targets: + elapsed_ms = _time_invocation(target.args, repeats=repeats) + within_budget = elapsed_ms <= target.budget_ms + results.append( + { + "label": target.label, + "args": list(target.args), + "gate": target.gate, + "budget_ms": target.budget_ms, + "elapsed_ms": round(elapsed_ms, 1), + "within_budget": within_budget, + } + ) + violations = [r["label"] for r in results if r["gate"] == "required" and not r["within_budget"]] + return { + "version": 1, + "generated_at": datetime.now(UTC).isoformat(), + "repeats": repeats, + "results": results, + "violations": violations, + "ok": not violations, + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repeats", type=int, default=3, help="Subprocess runs per target; minimum wins.") + parser.add_argument("--json", action="store_true", help="Emit the full JSON report instead of a table.") + parser.add_argument("--out", type=Path, default=None, help="Also write the JSON report to this path.") + args = parser.parse_args(argv) + + report = measure(repeats=max(1, args.repeats)) + + if args.out is not None: + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + if args.json: + print(json.dumps(report, indent=2, sort_keys=True)) + else: + results = cast("list[dict[str, object]]", report["results"]) + for result in results: + marker = "OK" if result["within_budget"] else "OVER" + flag = "" if result["gate"] == "required" else " (informational)" + print(f"{marker:>4} {result['label']:<32} {result['elapsed_ms']:>7.1f}ms / {result['budget_ms']}ms{flag}") + violations = cast("list[str]", report["violations"]) + if violations: + print(f"\nBudget violations (required): {', '.join(violations)}") + else: + print("\nAll required targets within budget.") + + return 0 if report["ok"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 5c51e862b9..e3f97549a4 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -683,7 +683,7 @@ Commands: action-affordances Print shared query-action affordance metadata as JSON. completions Generate shell completion scripts. paths Print canonical archive paths and filesystem topology. - query-completions Print shared query-builder completion metadata as... + query-completions Print shared query-builder completion metadata as JSON. ``` ## Completions diff --git a/docs/devtools.md b/docs/devtools.md index 5d8f03ba34..3e07e6ec96 100644 --- a/docs/devtools.md +++ b/docs/devtools.md @@ -176,6 +176,7 @@ These are the commands worth remembering during normal repo work: | --- | --- | | `devtools bench campaign` | Run or compare benchmark campaigns. | | `devtools bench coordination-latency` | Measure compact coordination status p50/p95 with raw stage samples. | +| `devtools bench help-latency` | Check `--help` wall-clock latency against the interactive-tier cold-CLI budget (polylogue-20d.2). | | `devtools bench ingest-amplification` | Measure deterministic per-tier ingest write amplification on a synthetic fixture (#1851). | | `devtools bench ingest-throughput` | Measure ingest wall-clock throughput on a synthetic fixture. | | `devtools bench memory` | Measure query-memory envelopes on generated fixtures. | diff --git a/docs/plans/slo-catalog.yaml b/docs/plans/slo-catalog.yaml index c6d91e8387..21237565de 100644 --- a/docs/plans/slo-catalog.yaml +++ b/docs/plans/slo-catalog.yaml @@ -79,3 +79,46 @@ surfaces: p95_ms: 60000 gate: "informational" tier: "lab" + + # --- Interactive tier (polylogue-20d.14) ----------------------------- + # + # Named latency budgets for the CLI-to-daemon hot path (polylogue-20d.1). + # These are the epic's own measurement contract: sibling beads (daemon + # result cache 20d.12, SSE push 20d.13, ingest-to-searchable lag 20d.6) + # cite these budgets rather than inventing their own. Only the surfaces + # that are actually implemented and benchmarkable today are "required"; + # the rest are informational placeholders naming the bead that must land + # before the row can gate (mirrors the existing context/cost pattern + # above, which defers to #838/#803/#870 the same way). + + daemon_cli_query: + description: "CLI-to-daemon UDS round trip for an ordinary find-mode session page (20d.1 fast path)" + benchmark_test: "tests/benchmarks/test_daemon_uds.py::test_bench_daemon_uds_cli_query" + p50_ms: 100 + p95_ms: 400 + gate: "required" + tier: "cheap-local" + + daemon_health_probe: + description: "Config-matched UDS health probe every fast-path call pays before the query itself" + benchmark_test: "tests/benchmarks/test_daemon_uds.py::test_bench_daemon_uds_health_probe" + p50_ms: 30 + p95_ms: 100 + gate: "required" + tier: "cheap-local" + + daemon_cached_facets: + description: "Cached facets/status read from the daemon result-cache memo layer (deferred until 20d.12 lands)" + benchmark_test: "tests/benchmarks/test_reader_api.py::test_bench_reader_facets" + p50_ms: 30 + p95_ms: 100 + gate: "informational" + tier: "cheap-local" + + ingest_to_searchable: + description: "JSONL write to find/webui visibility, end to end (deferred until 20d.6/20d.12/20d.13 land)" + benchmark_test: "tests/benchmarks/test_daemon_convergence.py::test_convergence_single_file_perf" + p50_ms: 5000 + p95_ms: 10000 + gate: "informational" + tier: "lab" diff --git a/docs/plans/test-clock-allowlist.yaml b/docs/plans/test-clock-allowlist.yaml index 9be43cf7b4..bc63f8bde7 100644 --- a/docs/plans/test-clock-allowlist.yaml +++ b/docs/plans/test-clock-allowlist.yaml @@ -77,3 +77,7 @@ files: reason: "Daemon resilience integration tests (#1735) measure real elapsed wall-clock for process lifecycle events (SIGKILL delivery, subprocess startup, concurrency timing). frozen_clock cannot substitute for real time when waiting on OS process state." - path: tests/integration/test_ingest_pipeline_correctness.py reason: "Ingest pipeline correctness tests (#1736) use time.monotonic for real process wait/retry timing that cannot be satisfied by a frozen clock." + - path: tests/benchmarks/test_daemon_uds.py + reason: "polylogue-20d.1 daemon UDS benchmark fixture polls a real background-thread HTTP server's readiness with a bounded wall-clock deadline; frozen_clock cannot substitute for waiting on real socket/thread startup." + - path: tests/unit/cli/test_daemon_golden_parity.py + reason: "polylogue-20d.1 golden-parity test starts a real UDS daemon server in a background thread and polls its readiness with a bounded wall-clock deadline before comparing direct vs daemon-proxied CLI output; frozen_clock cannot substitute for waiting on real socket/thread startup." diff --git a/polylogue/cli/archive_query.py b/polylogue/cli/archive_query.py index edb7e8ebd7..c41fcc8645 100644 --- a/polylogue/cli/archive_query.py +++ b/polylogue/cli/archive_query.py @@ -1313,6 +1313,47 @@ def _daemon_query_pairs(query_params: Mapping[str, object]) -> Iterable[tuple[st yield key, str(value) +_DAEMON_LIST_ITEM_KEEP_KEYS = ( + "id", + "origin", + "title", + "target_ref", + "anchor", + "actions", + "created_at", + "updated_at", + "message_count", + "tags", + "summary", + "words", + "repo", + "cwd_display", + "flags", +) + + +def _normalize_daemon_list_item(item: Mapping[str, object]) -> dict[str, object]: + """Reshape a daemon web-reader session row into the CLI's native list-row shape. + + The daemon's ``/api/sessions`` wire contract (``_archive_summary_payload`` / + ``_do_list`` in ``daemon/http.py``) is the stable webui row shape — + ``word_count`` naming, a ``date`` convenience field, and a ``session_id`` + duplicate of ``id``. The CLI's direct-path renderer + (``archive_query._summary_payload`` -> ``SessionListRowPayload``) predates + that contract and uses ``words`` with no ``date``/``session_id`` fields. + Golden parity (polylogue-20d.1) requires the two to render identically, so + the CLI-side proxy adapts the wire shape here rather than either surface + changing its stable contract. + """ + + normalized = dict(item) + if "words" not in normalized and "word_count" in normalized: + normalized["words"] = normalized.get("word_count") + return { + key: normalized[key] for key in _DAEMON_LIST_ITEM_KEEP_KEYS if key in normalized and normalized[key] is not None + } + + def _emit_daemon_list_payload( payload: Mapping[str, object], *, @@ -1322,7 +1363,11 @@ def _emit_daemon_list_payload( origin: str | None, fields: str | None, ) -> None: - items = [dict(item) for item in cast(list[object], payload.get("items") or []) if isinstance(item, Mapping)] + items = [ + _normalize_daemon_list_item(item) + for item in cast(list[object], payload.get("items") or []) + if isinstance(item, Mapping) + ] total = _object_int(payload.get("total") or len(items)) next_offset = offset + limit if total > offset + limit else None envelope: dict[str, object] = { diff --git a/polylogue/cli/commands/config.py b/polylogue/cli/commands/config.py index 1f8b90b5b2..2611c68f22 100644 --- a/polylogue/cli/commands/config.py +++ b/polylogue/cli/commands/config.py @@ -4,12 +4,7 @@ import click -from polylogue.cli.commands.completions import ( - action_affordances_command, - completions_command, - query_completions_command, -) -from polylogue.cli.commands.paths import paths_command +from polylogue.cli.click_command_registration import _LazyCommand from polylogue.cli.shared.types import AppEnv @@ -38,10 +33,44 @@ def config_command(ctx: click.Context, output_format: str, show_layers: bool) -> _show_config(env, output_format, show_layers) -config_command.add_command(completions_command) -config_command.add_command(query_completions_command) -config_command.add_command(action_affordances_command) -config_command.add_command(paths_command) +# Deferred: `completions.py` transitively imports the whole insights/storage +# stack (polylogue.operations.action_contracts -> operations.archive -> +# insights.archive -> storage.repair), ~650ms alone. `config --help` only +# needs each subcommand's name + short help, not its implementation, so these +# register as lazy proxies (matches the root-level pattern in +# click_command_registration.py) instead of importing eagerly at module scope. +config_command.add_command( + _LazyCommand( + "completions", + "polylogue.cli.commands.completions", + "completions_command", + short_help="Generate shell completion scripts.", + ) +) +config_command.add_command( + _LazyCommand( + "query-completions", + "polylogue.cli.commands.completions", + "query_completions_command", + short_help="Print shared query-builder completion metadata as JSON.", + ) +) +config_command.add_command( + _LazyCommand( + "action-affordances", + "polylogue.cli.commands.completions", + "action_affordances_command", + short_help="Print shared query-action affordance metadata as JSON.", + ) +) +config_command.add_command( + _LazyCommand( + "paths", + "polylogue.cli.commands.paths", + "paths_command", + short_help="Print canonical archive paths and filesystem topology.", + ) +) def _show_config(env: AppEnv, output_format: str, show_layers: bool) -> None: diff --git a/polylogue/daemon/http.py b/polylogue/daemon/http.py index 2371221e49..b35571c040 100644 --- a/polylogue/daemon/http.py +++ b/polylogue/daemon/http.py @@ -2748,8 +2748,8 @@ def _archive_summary_payload(self, summary: ArchiveSessionSummary) -> dict[str, "updated_at": summary.updated_at, "message_count": summary.message_count, "word_count": summary.word_count, - "repo": None, - "cwd_display": None, + "repo": summary.git_repository_url, + "cwd_display": next(iter(summary.working_directories), None), "tags": list(summary.tags), "flags": None, "summary": None, diff --git a/tests/benchmarks/test_daemon_uds.py b/tests/benchmarks/test_daemon_uds.py new file mode 100644 index 0000000000..a87c3f7052 --- /dev/null +++ b/tests/benchmarks/test_daemon_uds.py @@ -0,0 +1,146 @@ +"""Daemon UDS fast-path benchmark (polylogue-20d.1 / polylogue-20d.14). + +Covers: CLI-to-daemon round trip over the AF_UNIX transport that +``polylogue.cli.archive_query`` proxies ordinary session-page queries through +when a config-matched daemon is reachable. This is the "interactive" SLO +tier's ``daemon_cli_query`` surface: the whole point of the hot-daemon fast +path is that this round trip stays far below the cold, import-paying direct +CLI path. + +Run with: + pytest tests/benchmarks/test_daemon_uds.py --benchmark-enable -p no:xdist -v +""" + +from __future__ import annotations + +import os +import shutil +import tempfile +import threading +import time +from collections.abc import Iterator +from pathlib import Path + +import pytest + +from tests.benchmarks.conftest import _seed_realistic_db +from tests.benchmarks.helpers import BenchmarkFixture + + +@pytest.fixture(scope="session") +def bench_daemon_uds_archive_root(tmp_path_factory: pytest.TempPathFactory) -> Path: + """Session-scoped archive_root with a seeded index.db (~1K messages). + + Deliberately smaller than ``bench_db_5k`` — this surface benchmarks fixed + per-request UDS/HTTP-handler overhead, not query cost over a large corpus + (the ``query``/``reader``/``facets`` surfaces already cover that). + """ + archive_root = tmp_path_factory.mktemp("bench-daemon-uds") / "archive" + archive_root.mkdir() + stats = _seed_realistic_db(archive_root / "index.db", target_messages=1000) + print(f"\nbench_daemon_uds_archive_root: {stats}") + return archive_root + + +@pytest.fixture +def bench_daemon_uds_client( + bench_daemon_uds_archive_root: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> Iterator[object]: + """A live production UDS daemon server + matching ``DaemonClient``. + + ``AF_UNIX`` paths are capped at ~108 bytes on Linux; pytest's default + ``tmp_path`` nests deep enough (``.../pytest-.../test-name0/...``) to + blow that budget, so the runtime dir (and only the runtime dir, which + holds the socket) lives under a short-path ``tempfile.mkdtemp()`` instead. + """ + + from polylogue.cli.daemon_client import DaemonClient + from polylogue.daemon.http import DaemonAPIHandler + from polylogue.daemon.uds import DaemonAPIUnixHTTPServer, daemon_socket_path + + runtime_dir = Path(tempfile.mkdtemp(prefix="plg-bench-uds-")) + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(bench_daemon_uds_archive_root)) + monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "data")) + monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path / "state")) + monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime_dir)) + monkeypatch.setenv("POLYLOGUE_SCHEMA_VALIDATION", "off") + monkeypatch.delenv("POLYLOGUE_NO_DAEMON", raising=False) + monkeypatch.delenv("POLYLOGUE_DAEMON", raising=False) + + socket_path = daemon_socket_path(str(runtime_dir)) + server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler) + server.auth_token = "" + thread = threading.Thread(target=server.serve_forever, name="bench-daemon-uds", daemon=True) + thread.start() + # Wait for the socket to accept connections rather than a fixed sleep — + # ThreadingMixIn.serve_forever binds synchronously in __init__, but give + # the accept loop a moment to actually start before the first probe. + deadline = time.monotonic() + 2.0 + client = DaemonClient(socket_path, timeout_s=1.0) + while time.monotonic() < deadline: + if client.request_json("GET", "/api/health") is not None: + break + time.sleep(0.02) + else: + shutil.rmtree(runtime_dir, ignore_errors=True) + pytest.fail("daemon UDS server did not become ready") + + try: + yield client + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + shutil.rmtree(runtime_dir, ignore_errors=True) + + +@pytest.mark.benchmark +def test_bench_daemon_uds_cli_query( + benchmark: BenchmarkFixture, + bench_daemon_uds_client: object, +) -> None: + """Benchmark the CLI's ``/api/cli/query`` UDS round trip (find-mode page). + + Matches: ``polylogue.cli.archive_query._try_emit_daemon_session_page`` -> + ``DaemonClient.cli_query`` -> ``DaemonAPIHandler._handle_cli_query``. + """ + from polylogue.cli.daemon_client import DaemonClient + + client = bench_daemon_uds_client + assert isinstance(client, DaemonClient) + + def _query() -> dict[str, object] | None: + return client.cli_query({"limit": 20}) + + result = benchmark(_query) + assert result is not None + items = result.get("items") + assert isinstance(items, list) + assert len(items) > 0 + + +@pytest.mark.benchmark +def test_bench_daemon_uds_health_probe( + benchmark: BenchmarkFixture, + bench_daemon_uds_client: object, +) -> None: + """Benchmark the config-matched health probe every fast-path call pays first.""" + from polylogue.cli.daemon_client import DaemonClient + from polylogue.storage.sqlite.archive_tiers.index import INDEX_SCHEMA_VERSION + from polylogue.version import POLYLOGUE_VERSION + + client = bench_daemon_uds_client + assert isinstance(client, DaemonClient) + archive_root = os.environ["POLYLOGUE_ARCHIVE_ROOT"] + + def _probe() -> dict[str, object] | None: + return client.probe( + archive_root=archive_root, + index_schema_version=INDEX_SCHEMA_VERSION, + daemon_version=POLYLOGUE_VERSION, + ) + + result = benchmark(_probe) + assert result is not None diff --git a/tests/unit/cli/test_daemon_golden_parity.py b/tests/unit/cli/test_daemon_golden_parity.py new file mode 100644 index 0000000000..4dc074402c --- /dev/null +++ b/tests/unit/cli/test_daemon_golden_parity.py @@ -0,0 +1,192 @@ +"""Golden parity: direct CLI execution vs config-matched daemon-proxied execution. + +polylogue-20d.1 acceptance criterion: "`--format json` output is byte-identical +between direct and daemon-proxied execution for every read surface on the demo +corpus." A real production UDS daemon server is started against the same +seeded archive the direct path reads, and the same `find` invocation is run +through :class:`click.testing.CliRunner` twice — once with no daemon socket +present (direct path) and once with the daemon reachable (proxied path) — so +this is an end-to-end regression test, not a mock of the daemon transport. + +The two envelopes are compared field-for-field rather than as raw text: the +daemon envelope carries an explicit ``"source": "daemon"`` provenance marker +that the direct envelope does not (`archive_query.py::_emit_daemon_list_payload` +vs `_emit_list`) — that is the one intentional, documented difference. Every +other field (`items`, `total`, `limit`, `offset`, `origin`, `next_offset`, +`next_cursor`) must match exactly. +""" + +from __future__ import annotations + +import json +import shutil +import tempfile +import threading +import time +from collections.abc import Iterator +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from tests.infra.storage_records import SessionBuilder + + +@pytest.fixture +def golden_parity_workspace(cli_workspace: dict[str, Path], monkeypatch: pytest.MonkeyPatch) -> dict[str, Path]: + """A real seeded archive, reused for both the direct and daemon-proxied runs.""" + + monkeypatch.setenv("XDG_STATE_HOME", str(cli_workspace["state_dir"])) + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(cli_workspace["archive_root"])) + monkeypatch.setenv("POLYLOGUE_FORCE_PLAIN", "1") + monkeypatch.delenv("POLYLOGUE_NO_DAEMON", raising=False) + monkeypatch.delenv("POLYLOGUE_DAEMON", raising=False) + + index_db = cli_workspace["archive_root"] / "index.db" + ( + SessionBuilder(index_db, "conv1") + .provider("chatgpt") + .title("Python Error Handling") + .git_repository_url("polylogue") + .add_message("m1", role="user", text="How to handle exceptions in Python?") + .add_message("m2", role="assistant", text="Use try-except blocks.") + .save() + ) + ( + SessionBuilder(index_db, "conv2") + .provider("claude-code") + .title("Rust Ownership") + .git_repository_url("polylogue") + .add_message("m3", role="user", text="What is ownership in Rust?") + .add_message("m4", role="assistant", text="Rust ownership ensures memory safety.") + .save() + ) + return cli_workspace + + +@pytest.fixture +def _uds_runtime_dir(monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]: + """A short-path runtime dir so the AF_UNIX socket path stays under the OS limit.""" + + runtime_dir = Path(tempfile.mkdtemp(prefix="plg-golden-uds-")) + monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime_dir)) + try: + yield runtime_dir + finally: + shutil.rmtree(runtime_dir, ignore_errors=True) + + +def _run_find_json(args: list[str], *, no_daemon: bool = False) -> dict[str, object]: + from polylogue.cli import cli + + runner = CliRunner() + # `--no-daemon` and `--repo` are root options (`click_app.py::cli`), not + # `find` verb options — they must precede `find` in argv. Passing `--repo` + # here as the root option (rather than a `repo:polylogue` DSL query token, + # which routes through a different, older rendering path with a distinct + # envelope shape — see the module docstring follow-up note) is what + # actually exercises `_try_emit_daemon_session_page` / + # `_daemon_session_page_supported`, the code this test targets. + root_flags = ["--plain", *args, *(["--no-daemon"] if no_daemon else [])] + result = runner.invoke(cli, [*root_flags, "find", "--format", "json", "--limit", "10"]) + assert result.exit_code == 0, result.output + return dict(json.loads(result.output)) + + +def _strip_provenance(envelope: dict[str, object]) -> dict[str, object]: + return {key: value for key, value in envelope.items() if key != "source"} + + +def test_find_list_json_parity_between_direct_and_daemon( + golden_parity_workspace: dict[str, Path], + _uds_runtime_dir: Path, +) -> None: + del golden_parity_workspace + args = ["--repo", "polylogue"] + + # 1. Direct path: no daemon socket exists at XDG_RUNTIME_DIR, so the probe + # fails in-process and the CLI falls back to opening SQLite itself. + direct_payload = _run_find_json(args, no_daemon=True) + assert "source" not in direct_payload + + # 2. Daemon-proxied path: start the production UDS server against the + # same archive_root the direct run just read, then reissue the identical + # query with the daemon reachable. + from polylogue.daemon.http import DaemonAPIHandler + from polylogue.daemon.uds import DaemonAPIUnixHTTPServer, daemon_socket_path + + socket_path = daemon_socket_path(str(_uds_runtime_dir)) + server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler) + server.auth_token = "" + thread = threading.Thread(target=server.serve_forever, name="golden-parity-uds", daemon=True) + thread.start() + try: + from polylogue.cli.daemon_client import DaemonClient + + client = DaemonClient(socket_path, timeout_s=1.0) + deadline = time.monotonic() + 2.0 + while time.monotonic() < deadline: + if client.request_json("GET", "/api/health") is not None: + break + time.sleep(0.02) + else: + pytest.fail("daemon UDS server did not become ready") + + daemon_payload = _run_find_json(args) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + assert daemon_payload["source"] == "daemon" + assert _strip_provenance(daemon_payload) == _strip_provenance(direct_payload) + assert direct_payload["items"], "fixture query must actually match rows, or parity is vacuous" + + +def test_facets_json_parity_between_direct_and_daemon( + golden_parity_workspace: dict[str, Path], + _uds_runtime_dir: Path, +) -> None: + """The facets surface has its own daemon fast path (`_fetch_daemon_facets`).""" + del golden_parity_workspace + from polylogue.cli import cli + + runner = CliRunner() + + direct_result = runner.invoke(cli, ["--plain", "--no-daemon", "facets", "--format", "json"]) + assert direct_result.exit_code == 0, direct_result.output + direct_payload = json.loads(direct_result.output) + + from polylogue.daemon.http import DaemonAPIHandler + from polylogue.daemon.uds import DaemonAPIUnixHTTPServer, daemon_socket_path + + socket_path = daemon_socket_path(str(_uds_runtime_dir)) + server = DaemonAPIUnixHTTPServer(socket_path, DaemonAPIHandler) + server.auth_token = "" + thread = threading.Thread(target=server.serve_forever, name="golden-parity-facets-uds", daemon=True) + thread.start() + try: + from polylogue.cli.daemon_client import DaemonClient + + client = DaemonClient(socket_path, timeout_s=1.0) + deadline = time.monotonic() + 2.0 + while time.monotonic() < deadline: + if client.request_json("GET", "/api/health") is not None: + break + time.sleep(0.02) + else: + pytest.fail("daemon UDS server did not become ready") + + daemon_result = runner.invoke(cli, ["--plain", "facets", "--format", "json"]) + assert daemon_result.exit_code == 0, daemon_result.output + daemon_payload = json.loads(daemon_result.output) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + # `generated_at` is a genuine wall-clock timestamp stamped independently + # by each call, not a parity signal. + direct_payload.pop("generated_at", None) + daemon_payload.pop("generated_at", None) + assert daemon_payload == direct_payload