.json`。
-
-## 回放 example data 并人工检查
-
-下面的命令读取 `examples/data/` 中的 ATIF、ACTF 和 OpenAI Messages 样例,把每个可
-回放的 agent/inference step 转换成 Chat Completions 请求,经过真实 Gateway 和本地 Echo,
-最后将输入、HTTP 响应和 canonical capture 汇总到一个长期保留的 review bundle:
-
-```bash
-just benchmark-gateway-replay
-```
-
-默认输出根目录是 `benchmark/gateway/results/replay-review/`。每次运行创建独立的 UTC
-时间戳子目录,并把该目录写入 `latest.txt`,不会覆盖以前的 review。也可以指定输入和输出:
-
-```bash
-just benchmark-gateway-replay examples/data /tmp/gateway-replay-review
-```
-
-建议先打开 bundle 中的 `REVIEW.md` 看每个 case 的来源、参考回答、Echo 回答和自动检查;
-需要逐字段核对时再查看 `review.jsonl`。后者把 source record、实际 Gateway request/response、
-解析后的 canonical request/response 以及检查结果 join 在同一行。`captured-events.jsonl` 是
-未经 join 的完整事件导出,`dataset/` 和 `gateway-state/` 分别保留 Lance Dataset 与 WAL。
-
-这里的 source response 只作为人工 review 的参考,不参与相等性断言。本地 Echo 固定返回
-最后一条 user message,因此这个 replay 验证的是 Gateway wire handling、事件配对和 capture
-保真度,不声称能够复现原始模型回答。
-
-## Links
-
-- [Gateway architecture](../../docs/src/en/pchronicle/guides/serve-gateway.md)
-- [`persisting-gateway`](../../crates/persisting-gateway/README.md)
-- [Regression tests](../../tests/regression/README.md)
diff --git a/benchmark/gateway/benchmark.py b/benchmark/gateway/benchmark.py
deleted file mode 100755
index 7fa8d1715..000000000
--- a/benchmark/gateway/benchmark.py
+++ /dev/null
@@ -1,628 +0,0 @@
-#!/usr/bin/env python3
-"""Black-box throughput benchmark for Gateway forwarding plus durable capture."""
-
-from __future__ import annotations
-
-import argparse
-import concurrent.futures
-import http.client
-import json
-import os
-import platform
-import shutil
-import signal
-import statistics
-import subprocess
-import sys
-import tempfile
-import threading
-import time
-import urllib.parse
-import uuid
-from dataclasses import asdict, dataclass
-from datetime import UTC, datetime
-from pathlib import Path
-from typing import TextIO
-
-SCENARIO_DIR = Path(__file__).resolve().parent
-REPO_ROOT = SCENARIO_DIR.parents[1]
-sys.path.insert(0, str(REPO_ROOT / "tests" / "regression"))
-
-from gateway_harness import ( # noqa: E402
- require_subcommand,
- resolve_binary,
- stop_process,
- wait_http,
- wait_logged_url,
- without_proxy_environment,
-)
-
-
-@dataclass
-class LoadResult:
- elapsed_seconds: float
- requests: int
- errors: int
- requests_per_second: float
- latency_ms_mean: float
- latency_ms_p50: float
- latency_ms_p95: float
- latency_ms_p99: float
- latency_ms_max: float
- estimated_mean_in_flight: float
- error_samples: list[str]
-
-
-def percentile(values: list[float], quantile: float) -> float:
- if not values:
- return 0.0
- ordered = sorted(values)
- index = max(0, min(len(ordered) - 1, int(len(ordered) * quantile + 0.999999) - 1))
- return ordered[index]
-
-
-def one_worker(
- base_url: str,
- start: threading.Event,
- deadline: list[float],
- request_budget: list[int] | None,
- request_budget_lock: threading.Lock,
- worker: int,
- concurrency: int,
- sessions: int,
- session_prefix: str,
- payload: str,
-) -> tuple[list[float], int, list[str]]:
- parsed = urllib.parse.urlparse(base_url)
- if parsed.scheme != "http" or parsed.hostname is None or parsed.port is None:
- raise ValueError(f"benchmark target must be an explicit HTTP URL: {base_url}")
- path = f"{parsed.path.rstrip('/')}/v1/chat/completions"
- body = json.dumps(
- {
- "model": "gateway-benchmark",
- "messages": [{"role": "user", "content": payload}],
- "stream": False,
- },
- separators=(",", ":"),
- ).encode()
- headers = {
- "authorization": "Bearer benchmark-local",
- "content-type": "application/json",
- }
- latencies: list[float] = []
- error_count = 0
- errors: list[str] = []
- connection: http.client.HTTPConnection | None = None
- request_index = 0
- start.wait()
- while time.monotonic() < deadline[0]:
- if request_budget is not None:
- with request_budget_lock:
- if request_budget[0] <= 0:
- break
- request_budget[0] -= 1
- request_started = time.perf_counter()
- try:
- if connection is None:
- connection = http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=15)
- # Keep storage topology constant across a concurrency sweep. Advancing
- # by `concurrency` makes all workers collectively round-robin over the
- # configured session set without changing its cardinality.
- session = (worker + request_index * concurrency) % sessions
- headers["x-persisting-benchmark-session"] = f"{session_prefix}-{session}"
- request_index += 1
- connection.request("POST", path, body=body, headers=headers)
- response = connection.getresponse()
- response_body = response.read()
- if response.status != 200:
- raise RuntimeError(f"HTTP {response.status}: {response_body[:160]!r}")
- document = json.loads(response_body)
- content = document["choices"][0]["message"]["content"]
- if content != payload:
- raise RuntimeError("echo response content mismatch")
- latencies.append((time.perf_counter() - request_started) * 1000.0)
- except Exception as error: # benchmark must report transport and protocol failures
- error_count += 1
- if len(errors) < 10:
- errors.append(f"worker {worker}: {error}")
- if connection is not None:
- connection.close()
- connection = None
- if connection is not None:
- connection.close()
- return latencies, error_count, errors
-
-
-def run_load(
- base_url: str,
- *,
- duration: float,
- concurrency: int,
- sessions: int,
- max_requests: int | None,
- session_prefix: str,
- payload: str,
-) -> LoadResult:
- start = threading.Event()
- deadline = [0.0]
- request_budget = [max_requests] if max_requests is not None else None
- request_budget_lock = threading.Lock()
- started = time.perf_counter()
- with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor:
- futures = [
- executor.submit(
- one_worker,
- base_url,
- start,
- deadline,
- request_budget,
- request_budget_lock,
- worker,
- concurrency,
- sessions,
- session_prefix,
- payload,
- )
- for worker in range(concurrency)
- ]
- deadline[0] = time.monotonic() + duration
- start.set()
- worker_results = [future.result() for future in futures]
- elapsed = time.perf_counter() - started
- latencies = [latency for worker, _, _ in worker_results for latency in worker]
- error_count = sum(count for _, count, _ in worker_results)
- errors = [error for _, _, worker_errors in worker_results for error in worker_errors][:10]
- return LoadResult(
- elapsed_seconds=elapsed,
- requests=len(latencies),
- errors=error_count,
- requests_per_second=len(latencies) / elapsed if elapsed else 0.0,
- latency_ms_mean=statistics.fmean(latencies) if latencies else 0.0,
- latency_ms_p50=percentile(latencies, 0.50),
- latency_ms_p95=percentile(latencies, 0.95),
- latency_ms_p99=percentile(latencies, 0.99),
- latency_ms_max=max(latencies, default=0.0),
- # Little's law: throughput * residence time estimates how many requests
- # were resident in the measured path on average. In this closed-loop
- # driver, a value near `concurrency` means the client kept the path full.
- estimated_mean_in_flight=(
- (len(latencies) / elapsed) * statistics.fmean(latencies) / 1000.0
- if elapsed and latencies
- else 0.0
- ),
- error_samples=errors,
- )
-
-
-def write_configs(work_dir: Path, echo_url: str) -> tuple[Path, Path]:
- warehouse = work_dir / "warehouse.toml"
- gateway = work_dir / "gateway.toml"
- warehouse.write_text(
- f'''default_dataset = "captures"
-
-[[datasets]]
-name = "captures"
-uri = "{work_dir / "dataset"}"
-''',
- encoding="utf-8",
- )
- gateway.write_text(
- f'''listen = "127.0.0.1:0"
-admin_listen = "127.0.0.1:0"
-agent_id = "gateway-benchmark"
-session_header = "x-persisting-benchmark-session"
-capture_level = "full"
-
-[[models]]
-name = "gateway-benchmark"
-provider = "openai"
-upstream = "{echo_url}/v1"
-''',
- encoding="utf-8",
- )
- return warehouse, gateway
-
-
-def captured_manifest_stats(dataset: Path, agent_id: str, prefix: str) -> dict[str, int]:
- """Count published capture rows without opening every Lance version.
-
- The epoch-fenced manifest is the canonical visibility pointer for an event
- log. Reading one manifest at a time keeps file-descriptor use constant even
- when a benchmark produces many sessions and Lance versions.
- """
- agent_dir = dataset / agent_id
- if not agent_dir.is_dir():
- raise RuntimeError(f"capture agent directory is missing: {agent_dir}")
-
- session_prefix = f"{prefix}-"
- sessions = sorted(
- path
- for path in agent_dir.iterdir()
- if path.is_dir() and path.name.startswith(session_prefix)
- )
- if not sessions:
- raise RuntimeError(f"no capture sessions found for prefix {prefix!r}")
-
- published_events = 0
- visible_segments = 0
- max_segment_level = 0
- min_revision: int | None = None
- max_revision: int | None = None
- for session in sessions:
- manifest_path = session / "events.lance" / "_manifest.json"
- try:
- manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
- except FileNotFoundError as error:
- raise RuntimeError(f"capture manifest is missing: {manifest_path}") from error
- except (OSError, json.JSONDecodeError) as error:
- raise RuntimeError(f"cannot read capture manifest {manifest_path}: {error}") from error
-
- if manifest.get("schema_version") != 1:
- raise RuntimeError(
- f"unsupported capture manifest schema in {manifest_path}: "
- f"{manifest.get('schema_version')!r}"
- )
- revision = manifest.get("revision")
- segments = manifest.get("segments")
- if (
- not isinstance(revision, int)
- or isinstance(revision, bool)
- or revision < 0
- or not isinstance(segments, list)
- ):
- raise RuntimeError(f"invalid capture manifest metadata: {manifest_path}")
-
- for segment in segments:
- rows = segment.get("rows") if isinstance(segment, dict) else None
- level = segment.get("level", 0) if isinstance(segment, dict) else None
- if not isinstance(rows, int) or isinstance(rows, bool) or rows < 0:
- raise RuntimeError(f"invalid segment row count in {manifest_path}: {rows!r}")
- if not isinstance(level, int) or isinstance(level, bool) or level < 0:
- raise RuntimeError(f"invalid segment level in {manifest_path}: {level!r}")
- published_events += rows
- max_segment_level = max(max_segment_level, level)
- visible_segments += len(segments)
- min_revision = revision if min_revision is None else min(min_revision, revision)
- max_revision = revision if max_revision is None else max(max_revision, revision)
-
- return {
- "sessions": len(sessions),
- "visible_segments": visible_segments,
- "max_segment_level": max_segment_level,
- "published_events": published_events,
- "min_manifest_revision": min_revision or 0,
- "max_manifest_revision": max_revision or 0,
- }
-
-
-def captured_event_counts(
- pchronicle: Path, dataset: Path, prefix: str, output: Path
-) -> dict[str, int]:
- """Exercise the real query path after storage-fragment maintenance."""
- query = (
- "SELECT kind, COUNT(*) AS count FROM dataset.events "
- f"WHERE session_id LIKE '{prefix}%' GROUP BY kind ORDER BY kind"
- )
- subprocess.run(
- [
- str(pchronicle),
- "query",
- str(dataset),
- query,
- "--format",
- "jsonl",
- "--output",
- str(output),
- ],
- cwd=REPO_ROOT,
- check=True,
- )
- counts: dict[str, int] = {}
- for line in output.read_text(encoding="utf-8").splitlines():
- row = json.loads(line)
- counts[str(row["kind"])] = int(row["count"])
- return counts
-
-
-def benchmark(args: argparse.Namespace, work_dir: Path, pchronicle: Path) -> dict[str, object]:
- logs = work_dir / "logs"
- dataset = work_dir / "dataset"
- state = work_dir / "gateway-state"
- logs.mkdir()
- dataset.mkdir()
- state.mkdir()
- payload = "x" * args.payload_bytes
- run_token = uuid.uuid4().hex
- measured_prefix = f"gateway-bench-{run_token}"
- processes: list[tuple[str, subprocess.Popen[bytes]]] = []
- handles: list[TextIO] = []
- serve_process: subprocess.Popen[bytes] | None = None
- try:
- echo_log = (logs / "echo.log").open("w", encoding="utf-8")
- handles.append(echo_log)
- echo_process = subprocess.Popen(
- [str(pchronicle), "echo", "--listen", "127.0.0.1:0", "--encoding", "plain"],
- cwd=REPO_ROOT,
- stdout=echo_log,
- stderr=subprocess.STDOUT,
- )
- processes.append(("pChronicle Echo", echo_process))
- echo_url = wait_logged_url(
- logs / "echo.log", "pChronicle Echo: ", echo_process, "pChronicle Echo"
- )
- wait_http(f"{echo_url}/health", echo_process, "pChronicle Echo")
-
- baseline: LoadResult | None = None
- with without_proxy_environment():
- if not args.skip_baseline:
- if args.warmup > 0:
- warmup = run_load(
- echo_url,
- duration=args.warmup,
- concurrency=args.concurrency,
- sessions=args.sessions,
- max_requests=min(args.requests, 64),
- session_prefix=f"baseline-warmup-{run_token}",
- payload=payload,
- )
- if warmup.errors:
- raise RuntimeError(f"Echo baseline warmup failed: {warmup.error_samples}")
- baseline = run_load(
- echo_url,
- duration=args.duration,
- concurrency=args.concurrency,
- sessions=args.sessions,
- max_requests=args.requests,
- session_prefix=f"baseline-{run_token}",
- payload=payload,
- )
- if baseline.errors:
- raise RuntimeError(f"Echo baseline failed: {baseline.error_samples}")
-
- warehouse, gateway = write_configs(work_dir, echo_url)
- serve_log = (logs / "serve.log").open("w", encoding="utf-8")
- handles.append(serve_log)
- serve_process = subprocess.Popen(
- [
- str(pchronicle),
- "serve",
- "--config",
- str(warehouse),
- "--listen",
- "127.0.0.1:0",
- "--gateway",
- str(gateway),
- "--gateway-dataset",
- "captures",
- "--gateway-state",
- str(state),
- ],
- cwd=REPO_ROOT,
- stdout=serve_log,
- stderr=subprocess.STDOUT,
- )
- processes.append(("pChronicle serve", serve_process))
- gateway_url = wait_logged_url(
- logs / "serve.log", "pChronicle Gateway: ", serve_process, "pChronicle Gateway"
- )
- admin_url = wait_logged_url(
- logs / "serve.log",
- "pChronicle Gateway admin: ",
- serve_process,
- "pChronicle Gateway admin",
- )
- wait_http(f"{admin_url}/admin/status", serve_process, "pChronicle Gateway admin")
-
- with without_proxy_environment():
- if args.warmup > 0:
- warmup = run_load(
- gateway_url,
- duration=args.warmup,
- concurrency=args.concurrency,
- sessions=args.sessions,
- max_requests=min(args.requests, 64),
- session_prefix=f"gateway-warmup-{run_token}",
- payload=payload,
- )
- if warmup.errors:
- raise RuntimeError(f"Gateway warmup failed: {warmup.error_samples}")
- gateway_result = run_load(
- gateway_url,
- duration=args.duration,
- concurrency=args.concurrency,
- sessions=args.sessions,
- max_requests=args.requests,
- session_prefix=measured_prefix,
- payload=payload,
- )
- if gateway_result.errors:
- raise RuntimeError(f"Gateway load failed: {gateway_result.error_samples}")
- if gateway_result.requests_per_second < args.min_rps:
- raise RuntimeError(
- f"Gateway throughput {gateway_result.requests_per_second:.2f} req/s is below "
- f"--min-rps {args.min_rps:.2f}"
- )
-
- drain_started = time.perf_counter()
- stop_process(serve_process, label="pChronicle serve", require_success=True)
- shutdown_drain_seconds = time.perf_counter() - drain_started
- serve_process = None
- capture = captured_manifest_stats(dataset, "gateway-benchmark", measured_prefix)
- expected_requests = gateway_result.requests
- expected_events = expected_requests * 2
- if capture["published_events"] != expected_events:
- raise RuntimeError(
- "durable capture count mismatch: "
- f"requests={expected_requests}, expected_events={expected_events}, "
- f"capture={json.dumps(capture, sort_keys=True)}"
- )
- event_counts = captured_event_counts(
- pchronicle, dataset, measured_prefix, logs / "capture-counts.jsonl"
- )
- if (
- event_counts.get("llm.request", 0) != expected_requests
- or event_counts.get("llm.response", 0) != expected_requests
- ):
- raise RuntimeError(
- "durable capture kind mismatch: "
- f"requests={expected_requests}, counts={json.dumps(event_counts, sort_keys=True)}"
- )
-
- comparison = None
- if baseline is not None:
- comparison = {
- "throughput_ratio": (
- gateway_result.requests_per_second / baseline.requests_per_second
- if baseline.requests_per_second
- else None
- ),
- "p95_latency_ratio": (
- gateway_result.latency_ms_p95 / baseline.latency_ms_p95
- if baseline.latency_ms_p95
- else None
- ),
- }
- return {
- "schema_version": 2,
- "generated_at": datetime.now(UTC).isoformat(),
- "system": {
- "platform": platform.platform(),
- "python": platform.python_version(),
- "cpu_count": os.cpu_count(),
- },
- "config": {
- "duration_seconds": args.duration,
- "request_limit": args.requests,
- "warmup_seconds": args.warmup,
- "concurrency": args.concurrency,
- "load_model": "closed_loop",
- "max_in_flight": args.concurrency,
- "sessions": args.sessions,
- "payload_bytes": args.payload_bytes,
- "capture_level": "full",
- },
- "echo_baseline": asdict(baseline) if baseline is not None else None,
- "gateway": asdict(gateway_result),
- "comparison": comparison,
- "durable_capture": {
- "expected_requests": expected_requests,
- "expected_events": expected_events,
- "shutdown_drain_seconds": shutdown_drain_seconds,
- "events_per_drain_second": (
- expected_events / shutdown_drain_seconds
- if shutdown_drain_seconds
- else None
- ),
- **capture,
- "event_counts": event_counts,
- "validated": True,
- },
- }
- finally:
- stop_process(serve_process, label="pChronicle serve")
- for label, process in reversed(processes):
- stop_process(process, label=label)
- for handle in handles:
- handle.close()
-
-
-def parse_args() -> argparse.Namespace:
- parser = argparse.ArgumentParser(description=__doc__)
- parser.add_argument("--duration", type=float, default=10.0)
- parser.add_argument("--warmup", type=float, default=1.0)
- parser.add_argument("--concurrency", type=int, default=16)
- parser.add_argument(
- "--requests",
- type=int,
- default=1024,
- help="maximum requests per measured phase; duration remains a time ceiling",
- )
- parser.add_argument(
- "--sessions",
- type=int,
- default=16,
- help="fixed capture-session cardinality (independent of concurrency)",
- )
- parser.add_argument("--payload-bytes", type=int, default=256)
- parser.add_argument("--min-rps", type=float, default=0.0)
- parser.add_argument("--skip-baseline", action="store_true")
- parser.add_argument("--keep-artifacts", action="store_true")
- parser.add_argument("--output", type=Path)
- args = parser.parse_args()
- if args.duration <= 0 or args.warmup < 0:
- parser.error("duration must be positive and warmup must be non-negative")
- if args.concurrency <= 0 or args.sessions <= 0 or args.requests <= 0:
- parser.error("concurrency, sessions, and requests must be positive")
- if args.payload_bytes < 0 or args.min_rps < 0:
- parser.error("payload-bytes and min-rps cannot be negative")
- return args
-
-
-def interrupt_on_sigterm(_signum: int, _frame: object) -> None:
- raise KeyboardInterrupt
-
-
-def main() -> None:
- args = parse_args()
- signal.signal(signal.SIGTERM, interrupt_on_sigterm)
- pchronicle = resolve_binary(
- "PERSISTING_PCHRONICLE_BIN",
- REPO_ROOT / "target" / "release" / "pchronicle",
- "cargo build --release --locked -p persisting-pchronicle-cli --bin pchronicle",
- )
- for subcommand in ["echo", "serve", "query"]:
- require_subcommand(
- pchronicle,
- subcommand,
- "cargo build --release --locked -p persisting-pchronicle-cli --bin pchronicle",
- )
- work_dir = Path(tempfile.mkdtemp(prefix="persisting-gateway-benchmark."))
- success = False
- try:
- result = benchmark(args, work_dir, pchronicle)
- output = args.output or SCENARIO_DIR / "results" / "latest.json"
- output = output.expanduser().resolve()
- output.parent.mkdir(parents=True, exist_ok=True)
- temporary = output.with_suffix(f"{output.suffix}.tmp")
- temporary.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
- temporary.replace(output)
- gateway = result["gateway"]
- baseline = result["echo_baseline"]
- print(
- "Gateway: "
- f"{gateway['requests_per_second']:.2f} req/s, "
- f"mean={gateway['latency_ms_mean']:.2f} ms, "
- f"p50={gateway['latency_ms_p50']:.2f} ms, "
- f"p95={gateway['latency_ms_p95']:.2f} ms, "
- f"p99={gateway['latency_ms_p99']:.2f} ms, "
- f"requests={gateway['requests']}"
- )
- print(
- "Load: closed-loop, "
- f"max_in_flight={args.concurrency}, "
- f"estimated_mean_in_flight={gateway['estimated_mean_in_flight']:.2f}"
- )
- if baseline is not None:
- print(
- "Echo baseline: "
- f"{baseline['requests_per_second']:.2f} req/s, "
- f"p95={baseline['latency_ms_p95']:.2f} ms"
- )
- print(f"Durable capture: {json.dumps(result['durable_capture'], sort_keys=True)}")
- capture = result["durable_capture"]
- print(
- "Capture drain: "
- f"{capture['shutdown_drain_seconds']:.2f} s, "
- f"{capture['events_per_drain_second']:.2f} events/s"
- )
- print(f"Result: {output}")
- success = True
- finally:
- if args.keep_artifacts or not success:
- print(f"Gateway benchmark artifacts: {work_dir}", file=sys.stderr)
- else:
- shutil.rmtree(work_dir)
-
-
-if __name__ == "__main__":
- main()
diff --git a/benchmark/gateway/replay.py b/benchmark/gateway/replay.py
deleted file mode 100644
index 90acc6ffb..000000000
--- a/benchmark/gateway/replay.py
+++ /dev/null
@@ -1,668 +0,0 @@
-#!/usr/bin/env python3
-"""Replay the example trajectory data through Gateway into a reviewable bundle."""
-
-from __future__ import annotations
-
-import argparse
-import copy
-import http.client
-import json
-import os
-import re
-import signal
-import subprocess
-import sys
-import time
-import urllib.parse
-from collections import defaultdict
-from dataclasses import asdict, dataclass
-from datetime import UTC, datetime
-from pathlib import Path
-from typing import Any, TextIO
-
-SCENARIO_DIR = Path(__file__).resolve().parent
-REPO_ROOT = SCENARIO_DIR.parents[1]
-sys.path.insert(0, str(REPO_ROOT / "tests" / "regression"))
-
-from gateway_harness import ( # noqa: E402
- append_jsonl,
- require_subcommand,
- resolve_binary,
- stop_process,
- wait_http,
- wait_logged_url,
- without_proxy_environment,
-)
-
-
-@dataclass(frozen=True)
-class ReplayCase:
- case_id: str
- source_file: str
- source_format: str
- source_record_id: str
- session_id: str
- model: str
- messages: list[dict[str, Any]]
- reference_response: Any
- source_record: dict[str, Any]
-
-
-def source_label(path: Path) -> str:
- try:
- return str(path.relative_to(REPO_ROOT))
- except ValueError:
- return str(path)
-
-
-def safe_id(value: object) -> str:
- normalized = re.sub(r"[^A-Za-z0-9_.-]+", "-", str(value)).strip("-.")
- return normalized or "unknown"
-
-
-def message_text(content: Any) -> str:
- if isinstance(content, str):
- return content
- if isinstance(content, list):
- parts: list[str] = []
- for part in content:
- if isinstance(part, str):
- parts.append(part)
- elif isinstance(part, dict):
- text = part.get("text", part.get("input_text"))
- if isinstance(text, str):
- parts.append(text)
- return "".join(parts)
- if isinstance(content, dict):
- text = content.get("text")
- return text if isinstance(text, str) else ""
- return ""
-
-
-def last_user_text(messages: list[dict[str, Any]]) -> str:
- for message in reversed(messages):
- if message.get("role") == "user":
- return message_text(message.get("content"))
- return ""
-
-
-def openai_cases(path: Path, document: list[Any]) -> list[ReplayCase]:
- cases: list[ReplayCase] = []
- for index, row in enumerate(document):
- if not isinstance(row, dict) or not isinstance(row.get("messages"), list):
- raise ValueError(f"invalid OpenAI Messages row {index} in {path}")
- record_id = str(row.get("id", index))
- session = safe_id(row.get("session_id", record_id))
- model = str(row.get("agent_model", "example-openai-model"))
- cases.append(
- ReplayCase(
- case_id=f"openai-{safe_id(record_id)}",
- source_file=source_label(path),
- source_format="openai-messages",
- source_record_id=record_id,
- session_id=f"gateway-replay-openai-{session}",
- model=model,
- messages=copy.deepcopy(row["messages"]),
- reference_response=copy.deepcopy(row.get("response")),
- source_record=copy.deepcopy(row),
- )
- )
- return cases
-
-
-def tool_call_message(tool_calls: Any) -> list[dict[str, Any]]:
- if not isinstance(tool_calls, list):
- return []
- converted: list[dict[str, Any]] = []
- for index, call in enumerate(tool_calls):
- if not isinstance(call, dict):
- continue
- call_id = str(call.get("tool_call_id", call.get("id", f"call-{index}")))
- name = str(call.get("function_name", call.get("name", "unknown_tool")))
- arguments = call.get("arguments", call.get("input", {}))
- converted.append(
- {
- "id": call_id,
- "type": "function",
- "function": {
- "name": name,
- "arguments": json.dumps(arguments, ensure_ascii=False, sort_keys=True),
- },
- }
- )
- return converted
-
-
-def append_tool_results(history: list[dict[str, Any]], tool_calls: Any, observations: Any) -> None:
- if isinstance(tool_calls, list):
- for index, call in enumerate(tool_calls):
- if not isinstance(call, dict) or "result" not in call:
- continue
- call_id = str(call.get("tool_call_id", call.get("id", f"call-{index}")))
- history.append(
- {
- "role": "tool",
- "tool_call_id": call_id,
- "content": json.dumps(call["result"], ensure_ascii=False, sort_keys=True),
- }
- )
- if isinstance(observations, list):
- for index, observation in enumerate(observations):
- if not isinstance(observation, dict):
- continue
- call_id = str(observation.get("tool_use_id", f"observation-{index}"))
- history.append(
- {
- "role": "tool",
- "tool_call_id": call_id,
- "content": message_text(observation.get("content")),
- }
- )
-
-
-def atif_cases(path: Path, document: dict[str, Any]) -> list[ReplayCase]:
- session = safe_id(document.get("session_id", document.get("trajectory_id", path.stem)))
- agent = document.get("agent") if isinstance(document.get("agent"), dict) else {}
- model = str(agent.get("model_name", "example-atif-model"))
- history: list[dict[str, Any]] = []
- cases: list[ReplayCase] = []
- for index, step in enumerate(document.get("steps", [])):
- if not isinstance(step, dict):
- raise ValueError(f"invalid ATIF step {index} in {path}")
- source = step.get("source")
- message = message_text(step.get("message"))
- if source == "user":
- history.append({"role": "user", "content": message})
- continue
- if source != "agent":
- continue
- step_id = str(step.get("step_id", index))
- cases.append(
- ReplayCase(
- case_id=f"atif-{session}-step-{safe_id(step_id)}",
- source_file=source_label(path),
- source_format="atif",
- source_record_id=step_id,
- session_id=f"gateway-replay-atif-{session}",
- model=str(step.get("model_name", model)),
- messages=copy.deepcopy(history),
- reference_response={
- "message": step.get("message"),
- "tool_calls": copy.deepcopy(step.get("tool_calls", [])),
- },
- source_record=copy.deepcopy(step),
- )
- )
- assistant: dict[str, Any] = {"role": "assistant", "content": message}
- converted_calls = tool_call_message(step.get("tool_calls"))
- if converted_calls:
- assistant["tool_calls"] = converted_calls
- history.append(assistant)
- append_tool_results(history, step.get("tool_calls"), None)
- return cases
-
-
-def actf_cases(path: Path, document: dict[str, Any]) -> list[ReplayCase]:
- task_id = safe_id(document.get("task_id", path.stem))
- attempts = document.get("attempts")
- if not isinstance(attempts, dict):
- raise ValueError(f"ACTF attempts must be an object in {path}")
- cases: list[ReplayCase] = []
- for attempt_id, attempt in sorted(attempts.items(), key=lambda item: str(item[0])):
- if not isinstance(attempt, dict):
- raise ValueError(f"invalid ACTF attempt {attempt_id} in {path}")
- trajectory = attempt.get("trajectory")
- if not isinstance(trajectory, dict) or not isinstance(trajectory.get("steps"), list):
- raise ValueError(f"ACTF attempt {attempt_id} has no trajectory steps in {path}")
- session = f"gateway-replay-actf-{task_id}-attempt-{safe_id(attempt_id)}"
- history: list[dict[str, Any]] = []
- current_system: str | None = None
- for index, step in enumerate(trajectory["steps"]):
- if not isinstance(step, dict):
- raise ValueError(f"invalid ACTF step {index} in {path}")
- system_prompt = message_text(step.get("system_prompt"))
- if system_prompt and system_prompt != current_system:
- history.append({"role": "system", "content": system_prompt})
- current_system = system_prompt
- user_content = message_text(step.get("user_content"))
- if user_content:
- history.append({"role": "user", "content": user_content})
- assistant_content = step.get("assistant_content")
- if not isinstance(assistant_content, dict):
- assistant_content = {"content": message_text(assistant_content)}
- step_id = str(step.get("step_id", index))
- cases.append(
- ReplayCase(
- case_id=(
- f"actf-{task_id}-attempt-{safe_id(attempt_id)}-step-{safe_id(step_id)}"
- ),
- source_file=source_label(path),
- source_format="actf",
- source_record_id=f"attempt-{attempt_id}/step-{step_id}",
- session_id=session,
- model="example-actf-model",
- messages=copy.deepcopy(history),
- reference_response=copy.deepcopy(assistant_content),
- source_record=copy.deepcopy(step),
- )
- )
- assistant: dict[str, Any] = {
- "role": "assistant",
- "content": message_text(assistant_content.get("content")),
- }
- converted_calls = tool_call_message(assistant_content.get("tool_calls"))
- if converted_calls:
- assistant["tool_calls"] = converted_calls
- history.append(assistant)
- append_tool_results(history, None, step.get("observation"))
- return cases
-
-
-def load_cases(data_dir: Path) -> list[ReplayCase]:
- cases: list[ReplayCase] = []
- json_paths = sorted(data_dir.rglob("*.json"))
- if not json_paths:
- raise ValueError(f"no JSON example data found under {data_dir}")
- for path in json_paths:
- document = json.loads(path.read_text(encoding="utf-8"))
- if isinstance(document, list):
- cases.extend(openai_cases(path, document))
- elif isinstance(document, dict) and str(document.get("schema_version", "")).startswith(
- "ATIF-"
- ):
- cases.extend(atif_cases(path, document))
- elif isinstance(document, dict) and "attempts" in document:
- cases.extend(actf_cases(path, document))
- else:
- raise ValueError(f"unsupported example JSON format: {path}")
- if not cases:
- raise ValueError(f"example data produced no replayable Gateway calls: {data_dir}")
- case_ids = [case.case_id for case in cases]
- if len(case_ids) != len(set(case_ids)):
- raise ValueError("example data produced duplicate replay case IDs")
- return cases
-
-
-def write_configs(run_dir: Path, echo_url: str, cases: list[ReplayCase]) -> tuple[Path, Path]:
- warehouse = run_dir / "warehouse.toml"
- gateway = run_dir / "gateway.toml"
- warehouse.write_text(
- 'default_dataset = "captures"\n\n'
- '[[datasets]]\nname = "captures"\n'
- f"uri = {json.dumps(str(run_dir / 'dataset'))}\n",
- encoding="utf-8",
- )
- routes = []
- for model in sorted({case.model for case in cases}):
- routes.append(
- "[[models]]\n"
- f"name = {json.dumps(model)}\n"
- 'provider = "openai"\n'
- f"upstream = {json.dumps(echo_url + '/v1')}\n"
- )
- gateway.write_text(
- 'listen = "127.0.0.1:0"\n'
- 'admin_listen = "127.0.0.1:0"\n'
- 'agent_id = "gateway-example-replay"\n'
- 'capture_level = "full"\n\n'
- + "\n".join(routes),
- encoding="utf-8",
- )
- return warehouse, gateway
-
-
-def send_case(gateway_url: str, case: ReplayCase) -> dict[str, Any]:
- parsed = urllib.parse.urlparse(gateway_url)
- if parsed.scheme != "http" or parsed.hostname is None or parsed.port is None:
- raise ValueError(f"Gateway must publish an explicit HTTP URL: {gateway_url}")
- request = {"model": case.model, "messages": case.messages, "stream": False}
- body = json.dumps(request, ensure_ascii=False, separators=(",", ":")).encode()
- headers = {
- "authorization": "Bearer gateway-example-replay",
- "content-type": "application/json",
- "x-persisting-session-id": case.session_id,
- }
- started = time.perf_counter()
- connection = http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=15)
- try:
- connection.request("POST", "/v1/chat/completions", body=body, headers=headers)
- response = connection.getresponse()
- response_body = response.read()
- finally:
- connection.close()
- latency_ms = (time.perf_counter() - started) * 1000
- try:
- response_json = json.loads(response_body)
- except json.JSONDecodeError:
- response_json = None
- result = {
- "case_id": case.case_id,
- "session_id": case.session_id,
- "http_status": response.status,
- "latency_ms": latency_ms,
- "request": request,
- "response": response_json,
- "response_body": response_body.decode("utf-8", errors="replace"),
- }
- if response.status != 200 or not isinstance(response_json, dict):
- raise RuntimeError(
- f"Gateway replay failed for {case.case_id}: HTTP {response.status} "
- f"body={result['response_body'][:500]!r}"
- )
- return result
-
-
-def query_events(pchronicle: Path, dataset: Path, output: Path) -> list[dict[str, Any]]:
- subprocess.run(
- [
- str(pchronicle),
- "query",
- str(dataset),
- (
- "SELECT seq, kind, session_id, model, call_id, payload_json "
- "FROM dataset.events ORDER BY session_id, seq"
- ),
- "--format",
- "jsonl",
- "--output",
- str(output),
- ],
- cwd=REPO_ROOT,
- check=True,
- )
- return [json.loads(line) for line in output.read_text(encoding="utf-8").splitlines()]
-
-
-def response_text(result: dict[str, Any]) -> str:
- try:
- return str(result["response"]["choices"][0]["message"]["content"] or "")
- except (KeyError, IndexError, TypeError):
- return ""
-
-
-def reference_text(reference: Any) -> str:
- if not isinstance(reference, dict):
- return message_text(reference)
- for field in ["content", "message", "reasoning_content"]:
- text = message_text(reference.get(field))
- if text:
- return text
- return ""
-
-
-def captured_payload(event: dict[str, Any]) -> dict[str, Any]:
- payload = event.get("payload_json")
- if not isinstance(payload, str):
- raise ValueError(f"captured event has no payload_json: {event}")
- parsed = json.loads(payload)
- if not isinstance(parsed, dict):
- raise ValueError(f"captured payload is not an object: {event}")
- return parsed
-
-
-def build_review(
- run_dir: Path,
- data_dir: Path,
- cases: list[ReplayCase],
- results: list[dict[str, Any]],
- events: list[dict[str, Any]],
-) -> None:
- results_by_case = {result["case_id"]: result for result in results}
- requests_by_session: dict[str, list[dict[str, Any]]] = defaultdict(list)
- responses_by_session: dict[str, list[dict[str, Any]]] = defaultdict(list)
- for event in events:
- kind = event.get("kind")
- if kind == "llm.request":
- requests_by_session[str(event.get("session_id"))].append(event)
- elif kind in {"llm.response", "llm.response.stream"}:
- responses_by_session[str(event.get("session_id"))].append(event)
-
- session_offsets: dict[str, int] = defaultdict(int)
- review_path = run_dir / "review.jsonl"
- checks_passed = 0
- markdown_rows: list[str] = []
- for case in cases:
- offset = session_offsets[case.session_id]
- session_offsets[case.session_id] += 1
- captured_requests = requests_by_session.get(case.session_id, [])
- captured_responses = responses_by_session.get(case.session_id, [])
- request_event = captured_requests[offset] if offset < len(captured_requests) else None
- response_event = captured_responses[offset] if offset < len(captured_responses) else None
- result = results_by_case[case.case_id]
- actual = response_text(result)
- expected_echo = last_user_text(case.messages)
- request_capture = None if request_event is None else captured_payload(request_event)
- response_capture = None if response_event is None else captured_payload(response_event)
- checks = {
- "http_200": result["http_status"] == 200,
- "echo_matches_last_user": actual == expected_echo,
- "captured_request": request_event is not None,
- "captured_response": response_event is not None,
- "capture_call_id_matches": (
- request_event is not None
- and response_event is not None
- and request_event.get("call_id") == response_event.get("call_id")
- ),
- "captured_request_wire_matches": (
- request_capture is not None
- and request_capture.get("payload", {}).get("http", {}).get("request_body")
- == result["request"]
- ),
- "captured_response_wire_matches": (
- response_capture is not None
- and response_capture.get("payload", {}).get("http", {}).get("response_body")
- == result["response"]
- ),
- }
- passed = all(checks.values())
- checks_passed += int(passed)
- append_jsonl(
- review_path,
- {
- "case": asdict(case),
- "gateway_result": result,
- "expected_echo_response": expected_echo,
- "captured_request": request_capture,
- "captured_response": response_capture,
- "checks": checks,
- "passed": passed,
- },
- )
- short_actual = actual.replace("\n", "\\n")[:64].replace("|", "\\|")
- short_reference = (
- reference_text(case.reference_response)
- .replace("\n", "\\n")[:64]
- .replace("|", "\\|")
- )
- markdown_rows.append(
- f"| `{case.case_id}` | `{case.source_format}` | {len(case.messages)} | "
- f"{'PASS' if passed else 'FAIL'} | {short_reference} | {short_actual} |"
- )
-
- expected_events = len(cases) * 2
- global_checks = {
- "exact_event_count": len(events) == expected_events,
- "only_request_response_events": all(
- event.get("kind") in {"llm.request", "llm.response"} for event in events
- ),
- }
- summary = {
- "schema_version": 1,
- "generated_at": datetime.now(UTC).isoformat(),
- "source_directory": str(data_dir),
- "output_directory": str(run_dir),
- "cases": len(cases),
- "expected_events": expected_events,
- "captured_events": len(events),
- "passed": checks_passed,
- "failed": len(cases) - checks_passed,
- "global_checks": global_checks,
- }
- (run_dir / "summary.json").write_text(
- json.dumps(summary, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
- encoding="utf-8",
- )
- (run_dir / "REVIEW.md").write_text(
- "# Gateway example-data replay review\n\n"
- f"Source: `{data_dir}` \n"
- f"Cases: {len(cases)} \n"
- f"Canonical events: {len(events)}/{expected_events} \n"
- f"Automated checks: {checks_passed}/{len(cases)} passed\n\n"
- "The recorded source response is reference material only. The local Echo upstream "
- "deterministically returns the last user message, so this run reviews Gateway wire "
- "handling and capture fidelity rather than model-output equivalence.\n\n"
- "## Files\n\n"
- "- `inputs.jsonl`: source record plus the replayed Chat Completions messages.\n"
- "- `client-results.jsonl`: exact HTTP request and response seen by the driver.\n"
- "- `captured-events.jsonl`: canonical events exported from the Lance Dataset.\n"
- "- `review.jsonl`: one joined record per call, including parsed canonical payloads.\n"
- "- `dataset/`: durable Lance capture; `gateway-state/`: Gateway WAL/state.\n"
- "- `logs/`: Echo and Gateway process logs.\n\n"
- "## Calls\n\n"
- "| Case | Source format | Request messages | Checks | Reference response | Echo output |\n"
- "|---|---:|---:|---:|---|---|\n"
- + "\n".join(markdown_rows)
- + "\n",
- encoding="utf-8",
- )
- if checks_passed != len(cases) or not all(global_checks.values()):
- raise RuntimeError(
- f"replay review checks failed: cases={checks_passed}/{len(cases)} "
- f"global={global_checks}; see {review_path}"
- )
-
-
-def run_replay(run_dir: Path, data_dir: Path, pchronicle: Path) -> None:
- cases = load_cases(data_dir)
- logs = run_dir / "logs"
- dataset = run_dir / "dataset"
- state = run_dir / "gateway-state"
- logs.mkdir()
- dataset.mkdir()
- state.mkdir()
- for case in cases:
- append_jsonl(run_dir / "inputs.jsonl", asdict(case))
-
- processes: list[tuple[str, subprocess.Popen[bytes]]] = []
- handles: list[TextIO] = []
- serve_process: subprocess.Popen[bytes] | None = None
- try:
- echo_log = (logs / "echo.log").open("w", encoding="utf-8")
- handles.append(echo_log)
- echo_process = subprocess.Popen(
- [str(pchronicle), "echo", "--listen", "127.0.0.1:0", "--encoding", "plain"],
- cwd=REPO_ROOT,
- stdout=echo_log,
- stderr=subprocess.STDOUT,
- )
- processes.append(("pChronicle Echo", echo_process))
- echo_url = wait_logged_url(
- logs / "echo.log", "pChronicle Echo: ", echo_process, "pChronicle Echo"
- )
- wait_http(f"{echo_url}/health", echo_process, "pChronicle Echo")
-
- warehouse, gateway = write_configs(run_dir, echo_url, cases)
- serve_log = (logs / "serve.log").open("w", encoding="utf-8")
- handles.append(serve_log)
- serve_process = subprocess.Popen(
- [
- str(pchronicle),
- "serve",
- "--config",
- str(warehouse),
- "--listen",
- "127.0.0.1:0",
- "--gateway",
- str(gateway),
- "--gateway-dataset",
- "captures",
- "--gateway-state",
- str(state),
- ],
- cwd=REPO_ROOT,
- stdout=serve_log,
- stderr=subprocess.STDOUT,
- )
- processes.append(("pChronicle serve", serve_process))
- gateway_url = wait_logged_url(
- logs / "serve.log", "pChronicle Gateway: ", serve_process, "pChronicle Gateway"
- )
- admin_url = wait_logged_url(
- logs / "serve.log",
- "pChronicle Gateway admin: ",
- serve_process,
- "pChronicle Gateway admin",
- )
- wait_http(f"{admin_url}/admin/status", serve_process, "pChronicle Gateway admin")
-
- results: list[dict[str, Any]] = []
- with without_proxy_environment():
- for case in cases:
- result = send_case(gateway_url, case)
- results.append(result)
- append_jsonl(run_dir / "client-results.jsonl", result)
-
- stop_process(serve_process, label="pChronicle serve", require_success=True)
- serve_process = None
- events = query_events(pchronicle, dataset, run_dir / "captured-events.jsonl")
- build_review(run_dir, data_dir, cases, results, events)
- finally:
- stop_process(serve_process, label="pChronicle serve")
- for label, process in reversed(processes):
- stop_process(process, label=label)
- for handle in handles:
- handle.close()
-
-
-def parse_args() -> argparse.Namespace:
- parser = argparse.ArgumentParser(description=__doc__)
- parser.add_argument("--data", type=Path, default=REPO_ROOT / "examples" / "data")
- parser.add_argument(
- "--output",
- type=Path,
- default=SCENARIO_DIR / "results" / "replay-review",
- help="parent directory for timestamped review bundles",
- )
- return parser.parse_args()
-
-
-def interrupt_on_sigterm(_signum: int, _frame: object) -> None:
- raise KeyboardInterrupt
-
-
-def main() -> None:
- args = parse_args()
- signal.signal(signal.SIGTERM, interrupt_on_sigterm)
- data_dir = args.data.expanduser().resolve()
- if not data_dir.is_dir():
- raise ValueError(f"example data directory does not exist: {data_dir}")
- output_root = args.output.expanduser().resolve()
- output_root.mkdir(parents=True, exist_ok=True)
- run_id = datetime.now(UTC).strftime("%Y%m%dT%H%M%S.%fZ")
- run_dir = output_root / run_id
- run_dir.mkdir()
- (output_root / "latest.txt").write_text(str(run_dir) + "\n", encoding="utf-8")
- print(f"Gateway example replay review: {run_dir}", flush=True)
-
- pchronicle = resolve_binary(
- "PERSISTING_PCHRONICLE_BIN",
- REPO_ROOT / "target" / "release" / "pchronicle",
- "cargo build --release --locked -p persisting-pchronicle-cli --bin pchronicle",
- )
- for subcommand in ["echo", "serve", "query"]:
- require_subcommand(
- pchronicle,
- subcommand,
- "cargo build --release --locked -p persisting-pchronicle-cli --bin pchronicle",
- )
- run_replay(run_dir, data_dir, pchronicle)
- summary = json.loads((run_dir / "summary.json").read_text(encoding="utf-8"))
- print(
- f"PASS Gateway example replay: cases={summary['cases']} "
- f"events={summary['captured_events']} review={run_dir / 'REVIEW.md'}",
- flush=True,
- )
-
-
-if __name__ == "__main__":
- main()
diff --git a/benchmark/gateway/replay.sh b/benchmark/gateway/replay.sh
deleted file mode 100755
index a7f1e3194..000000000
--- a/benchmark/gateway/replay.sh
+++ /dev/null
@@ -1,5 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-scenario_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
-exec python3 "$scenario_dir/replay.py" "$@"
diff --git a/benchmark/gateway/run.sh b/benchmark/gateway/run.sh
deleted file mode 100755
index 37beb8849..000000000
--- a/benchmark/gateway/run.sh
+++ /dev/null
@@ -1,5 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-scenario_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
-exec python3 "$scenario_dir/benchmark.py" "$@"
diff --git a/benchmark/langfuse-pchronicle-review/README.md b/benchmark/langfuse-pchronicle-review/README.md
deleted file mode 100644
index 063594192..000000000
--- a/benchmark/langfuse-pchronicle-review/README.md
+++ /dev/null
@@ -1,132 +0,0 @@
-# Langfuse → pChronicle backend feasibility probe
-
-**Review-only probe: replace Langfuse's ClickHouse analytics backend with
-pChronicle, without changing Langfuse's public API or pChronicle storage
-formats.**
-
-Owns the adapter experiment, deterministic fixture, and recorded verdicts.
-Postgres, Redis, and object storage remain unchanged. TTAS, Queue/Sampler,
-Search, and `persisting-dlcapt` are outside this probe.
-
-The original recorded verdict and evidence are in [REPORT.zh.md](REPORT.zh.md).
-The lazy-catalog retest is in
-[RETEST-2026-08-12.zh.md](RETEST-2026-08-12.zh.md), with machine-readable
-results in
-[recorded-results-2026-08-12.json](recorded-results-2026-08-12.json). The
-Server source-routing implementation and release measurements are in
-[SERVER-ACCELERATION-2026-08-12.zh.md](SERVER-ACCELERATION-2026-08-12.zh.md),
-with machine-readable results in
-[server-acceleration-results-2026-08-12.json](server-acceleration-results-2026-08-12.json). The
-original run remains in
-[recorded-results-2026-08-11.json](recorded-results-2026-08-11.json).
-
-Pinned review baselines:
-
-- Langfuse: `d18a59ad663ffc7c04afc61354186c141b3ec0f3` (v4.7.1)
-- Persisting: `94531cf903e5abc336de347588fb1858e9d52b6a` plus the local,
- uncommitted catalog/query-engine work present during the review
-- ClickHouse: 25.12.x; the recorded run used 25.12.11.4
-
-## Workload
-
-The default deterministic fixture contains two projects and:
-
-- 100,000 trace/span events, including 100 duplicate logical versions
-- 10,000 scores
-- 2,000 dataset-run items
-- 1,000 blob-storage log rows
-- metadata, tags, tool names, large I/O values, and 12-digit decimal costs
-
-Trace IDs map to pChronicle Run IDs. Rows without a trace use synthetic Run
-IDs. `project_id` maps to `agent_id`; the complete logical row is retained in
-the event payload. The default 200 traces plus synthetic Runs create 210 Lance
-event datasets.
-
-The executable defines the review-only `LangfuseAnalyticsBackend` contract and
-implements append, point/list/aggregate reads, JSON streaming, flush, and
-health. Update and delete methods deliberately return unsupported errors. An
-implementation that hides those errors behind a new mutable projection engine
-would exceed the bounded-integration decision rule.
-
-## Run
-
-Use an empty temporary directory; the fixture is about 238 MB.
-
-```bash
-PROBE_DIR="$(mktemp -d /tmp/langfuse-pchronicle-review.XXXXXX)"
-PCHRONICLE_LANGFUSE_WORKDIR="$PROBE_DIR" \
- cargo run --release -p persisting-pchronicle \
- --example langfuse_backend_feasibility
-```
-
-The command writes:
-
-- `$PROBE_DIR/logical_rows.jsonl` — shared deterministic input
-- `$PROBE_DIR/pchronicle-store/` — pChronicle data
-- `$PROBE_DIR/pchronicle-report.json` — measurements and capability failures
-
-The load phase samples visibility after every acknowledged single-row append,
-so `visibility_p95_ms` is an observed percentile rather than a final-row spot
-check.
-
-Useful scale controls are `PCHRONICLE_LANGFUSE_EVENTS`,
-`PCHRONICLE_LANGFUSE_SCORES`, `PCHRONICLE_LANGFUSE_DATASET_RUN_ITEMS`,
-`PCHRONICLE_LANGFUSE_BLOB_LOG_ROWS`, `PCHRONICLE_LANGFUSE_TRACES`,
-`PCHRONICLE_LANGFUSE_PRELOAD_BATCH_ROWS`, and
-`PCHRONICLE_LANGFUSE_LOAD_SECONDS`.
-
-Start an isolated ClickHouse 25.12 instance. One Docker example is:
-
-```bash
-docker run --rm --name langfuse-pchronicle-review \
- -e CLICKHOUSE_PASSWORD=review-only \
- -p 127.0.0.1:18123:8123 \
- clickhouse/clickhouse-server:25.12
-```
-
-In another shell, run the same fixture:
-
-```bash
-python3 benchmark/langfuse-pchronicle-review/clickhouse_baseline.py \
- --url http://127.0.0.1:18123 \
- --user default \
- --password review-only \
- --fixture "$PROBE_DIR/logical_rows.jsonl" \
- --output "$PROBE_DIR/clickhouse-report.json"
-```
-
-The script sends credentials only in ClickHouse authentication headers. It
-drops and recreates only the disposable database
-`langfuse_pchronicle_review`. Its schema mirrors the relevant Langfuse v4
-patterns: `ReplacingMergeTree`, project/time/trace ordering, full/core event
-tables with an incremental materialized view, text and skipping indexes,
-async inserts with acknowledgement, and separate score/dataset/blob tables.
-It samples visibility after every acknowledged 10-row load batch.
-
-```bash
-cargo test -p persisting-pchronicle --test langfuse_backend_faults -- --nocapture
-cargo test -p persisting-pchronicle --test production_scale --test query_engine
-cargo test -p persisting-pchronicle store::catalog
-```
-
-These cover SIGKILL after an acknowledged append, writer fencing, pinned
-snapshots during append, independent Run isolation, maintenance/restart, the
-read-only query boundary, and catalog refresh behavior.
-
-## Acceptance gates
-
-- No acknowledged loss and no duplicate logical rows
-- Visibility P95 ≤ 2 s; update/delete visibility ≤ 5 s
-- Point P95 ≤ 500 ms; list/facet P95 ≤ 1 s
-- Dashboard, full-text, and export first-byte P95 ≤ 2 s
-- No query class more than 5× slower than the ClickHouse baseline
-- Catalog cold start ≤ 30 s and RSS ≤ 2 GiB
-- Exact project isolation and full Langfuse v4 update/delete/export semantics
-
-Isolation, durability, mutation, or public-semantic failure is an immediate
-NO-GO. A localized performance-only miss may be conditional.
-
-## Links
-
-- [pChronicle design](../../docs/src/pchronicle/design/index.md)
-- [`persisting-pchronicle`](../../crates/persisting-pchronicle/README.md)
diff --git a/benchmark/langfuse-pchronicle-review/REPORT.zh.md b/benchmark/langfuse-pchronicle-review/REPORT.zh.md
deleted file mode 100644
index cda11d2f6..000000000
--- a/benchmark/langfuse-pchronicle-review/REPORT.zh.md
+++ /dev/null
@@ -1,205 +0,0 @@
-# Langfuse ClickHouse → pChronicle 存储后端可行性评审
-
-评审日期:2026-08-11
-
-> [KNOWN, HIGH] 2026-08-12 的 lazy-catalog 优化已显著改变性能结论;性能章节请以
-> [RETEST-2026-08-12.zh.md](RETEST-2026-08-12.zh.md) 为准。本报告的 full-v4
-> 功能兼容结论保留为历史基线。
-
-## 先看最强支持理由
-
-[COMPUTED, HIGH] 如果目标被收窄为“低吞吐、追加式、按轨迹回放的原始事件库”,pChronicle 在本次 100k event PoC 中是可用的:113,000 行预载及 30 行 10 eps 在线写入均无 ack 后丢失;写入可见延迟 P95 为 4.13 ms;项目交叉点查为 0 行;SIGKILL-after-ack、writer fencing、固定快照、maintenance 后重启均通过;cold catalog 2.10 s,进程 RSS 1.04 GiB。
-
-[INFERRED, HIGH] 这说明 pChronicle 适合承接 Langfuse 的 canonical/cold trajectory、回放和证据归档角色,也说明“完全不适合 Langfuse 数据”这个更强的否定命题不成立。
-
-## 结论
-
-**[INFERRED, HIGH] NO-GO:在钉住的代码状态和已批准的验收标准下,不应以 pChronicle 直接替换 Langfuse 的 ClickHouse 后端。**
-
-[COMPUTED, HIGH] 直接替换触发了四个立即否决项:重复逻辑版本仍作为物理行暴露;`bookmarked/public` 更新不支持;trace/project/retention 删除不支持;Langfuse dashboard、FTS、去重和 Parquet export 所依赖的关键 ClickHouse 语义不支持。
-
-[COMPUTED, HIGH] 即使只看已可表达的查询,最终记录跑中 pChronicle point/list P95 分别是 ClickHouse 的 82.02×/190.13×,超过“不慢于 5×”门槛。第二次完整 pChronicle 跑的 point P95 为 527.94 ms,说明 500 ms 的绝对 point 门槛也不稳定。
-
-[INFERRED, HIGH] 补齐这些缺口至少需要新的 latest-version/tombstone 投影、typed OLAP projections、ClickHouse 查询语义兼容层、多租户网络服务与 catalog pruning;这已是一个新的多租户 OLAP 产品,而不是有界适配器,因此违反本次评审的停止条件。
-
-[INFERRED, HIGH] 推荐保留 ClickHouse 作为 hot OLAP 和 Langfuse API/UI 的查询后端,把 pChronicle 放在旁路,作为追加式 canonical/cold trajectory store。Postgres、Redis 和现有对象存储保持不变。
-
-## 评审范围与基线
-
-[KNOWN, HIGH] 本评审只讨论替换 ClickHouse。Langfuse 的 Postgres 事务数据、Redis 队列/缓存、S3/对象存储不在替换范围内。Langfuse 官方架构也将这些职责分开:[Architecture](https://langfuse.com/handbook/product-engineering/architecture)、[ClickHouse infrastructure](https://langfuse.com/self-hosting/deployment/infrastructure/clickhouse)。
-
-[KNOWN, HIGH] 钉住的基线为:
-
-- Persisting `94531cf903e5abc336de347588fb1858e9d52b6a`,加评审时已存在、未提交的 catalog/query-engine 实验代码;
-- Langfuse `d18a59ad663ffc7c04afc61354186c141b3ec0f3`,工作区干净;
-- ClickHouse `25.12.11.4`;
-- Langfuse v4 语义。v4 对 ClickHouse 的版本和表结构要求可参见官方 [v3 → v4 upgrade guide](https://langfuse.com/self-hosting/upgrade/upgrade-guides/upgrade-v3-to-v4)。
-
-[KNOWN, HIGH] Persisting 的 TTAS、Queue/Sampler、Search 和独立 `persisting-dlcapt` 不在本评审范围内。FTS 只检查现有非 Search 的 DataFusion/Lance 路径,没有借助另一个搜索子系统掩盖后端缺口。
-
-[KNOWN, HIGH] 新增的实现与证据文件是:
-
-- `crates/persisting-pchronicle/examples/langfuse_backend_feasibility.rs`:有界 backend contract、fixture、语义/性能/资源/maintenance 探针;
-- `crates/persisting-pchronicle/tests/langfuse_backend_faults.rs`:ack 后强制终止进程再恢复读取;
-- `benchmark/langfuse-pchronicle-review/clickhouse_baseline.py`:相同 fixture 的 ClickHouse 25.12 对照;
-- `benchmark/langfuse-pchronicle-review/recorded-results-2026-08-11.json`:最终机器可读结果;
-- `benchmark/langfuse-pchronicle-review/README.md`:复现命令。
-
-## 数据映射与 workload
-
-[KNOWN, HIGH] PoC 使用以下映射;完整 Langfuse 逻辑行同时保留在 `payload_json` 中:
-
-| Langfuse | pChronicle |
-|---|---|
-| `project_id` | `agent_id` |
-| `trace_id` | Run / `root_session_id` |
-| `span_id` | `call_id` |
-| `parent_span_id` | `parent_call_id` |
-| 无 trace 的 score/dataset/blob 行 | synthetic Run |
-
-[COMPUTED, HIGH] 两个后端读取的是同一份 238,470,862-byte JSONL,SHA-256 为 `1b2ef8ed2f51e6d85a9907bbbebbdafbc4c961d8c4368b2a20b737f94cfc4c90`。它包含两个 project、100,000 events(100 个重复逻辑版本、99,900 个 distinct logical IDs)、10,000 scores、2,000 dataset-run items 和 1,000 blob-log rows;另含 metadata、tags、tool names、65 KiB I/O 和 12 位小数 cost。200 条 trace 加 synthetic Runs 在 pChronicle 中形成 210 个 source datasets。
-
-[KNOWN, HIGH] 在线阶段持续 3 秒、目标 10 eps,并运行一个并发查询;查询基准每类重复 7 次。该规模对应已批准的 team self-hosted/dev 验收,不代表生产容量。
-
-## 验收结果
-
-[COMPUTED, HIGH] 下表采用最终记录跑;原始摘要在 `recorded-results-2026-08-11.json`。
-
-| Gate / workload | pChronicle | ClickHouse | 判定 |
-|---|---:|---:|---|
-| 113k 预载 ack loss | 0 | 0 | `[COMPUTED, HIGH]` 通过 |
-| 10 eps 在线阶段 | 10.26 eps | 9.86 eps | `[COMPUTED, HIGH]` 通过 |
-| 在线 ack loss | 0 / 30 | 0 / 30 | `[COMPUTED, HIGH]` 通过 |
-| 可见延迟 P95 | 4.13 ms | 6.51 ms | `[COMPUTED, HIGH]` 通过 ≤2 s |
-| 重复版本后的物理/逻辑 event 行 | 100,000 / 99,900 | 99,900 / 99,900 | `[COMPUTED, HIGH]` pChronicle 失败 |
-| point P95 | 395.87 ms | 4.83 ms | `[COMPUTED, HIGH]` 绝对值通过,82.02× 相对门槛失败 |
-| list P95 | 564.82 ms | 2.97 ms | `[COMPUTED, HIGH]` 绝对值通过,190.13× 相对门槛失败 |
-| simple group/facet P95 | 806.70 ms(model group) | 3.47 ms(tag facet) | `[COMPUTED, MED]` 都过 1 s;查询并非完全同构,不用其比值作单独否决 |
-| 并发查询 P95 | 350.41 ms | 13.96 ms | `[COMPUTED, HIGH]` pChronicle 为 25.10× |
-| dashboard | ClickHouse curried quantile/`WITH FILL` 不支持 | 4.53 ms | `[COMPUTED, HIGH]` 失败 |
-| FTS | `hasAllTokens` 不支持 | 3.47 ms | `[COMPUTED, HIGH]` 失败 |
-| JSON export TTFB | 429.09 ms / 50k 行 | 68.77 ms / 10k 行 | `[COMPUTED, MED]` 都过 2 s;行数不同,不比较倍数 |
-| Parquet export TTFB | 不支持 | 28.27 ms | `[COMPUTED, HIGH]` 失败 |
-| update flags | 不支持 | 25.32 ms | `[COMPUTED, HIGH]` 失败 |
-| trace/project/retention delete | 全部不支持 | 8.98 / 5.96 / 7.64 ms | `[COMPUTED, HIGH]` 失败 |
-| catalog cold start | 2.10 s | N/A(常驻服务) | `[COMPUTED, HIGH]` 通过 ≤30 s |
-| RSS / resident memory | 1,117,782,016 B | 1,422,721,024 B | `[COMPUTED, MED]` 两者采集边界不同;pChronicle 通过 ≤2 GiB |
-| maintenance + restart visibility | 38.36 ms + 4.16 s,可见 | N/A | `[COMPUTED, HIGH]` 通过本次故障探针 |
-| cross-project negative point | 0 行 | 0 行 | `[COMPUTED, HIGH]` 数据映射级通过 |
-
-[COMPUTED, HIGH] bulk append 吞吐为 pChronicle 9,374.59 rows/s、ClickHouse 73,449.48 rows/s,ClickHouse 高 7.83×;两者都足以覆盖本次 10 eps 目标,因此吞吐本身不是立即否决项。
-
-[COMPUTED, HIGH] 另一轮相同 fixture 的完整跑中,pChronicle point/list P95 为 527.94/666.71 ms,ClickHouse 为 4.18/4.00 ms。跨跑波动会改变 point 是否略过 500 ms,但不会改变远超 5× 相对门槛的结论。
-
-## 语义兼容矩阵
-
-| Langfuse v4 能力 | 当前 pChronicle | 结论 |
-|---|---|---|
-| acknowledged append / batch | 原生 append 可用,PoC 零 ack loss | `[COMPUTED, HIGH]` 可适配 |
-| latest-version dedup | 明确 at-least-once;重复 `event_id` 是合法物理事实 | `[KNOWN, HIGH]` 不兼容 |
-| `events_full` wide typed schema | 只有少量索引列和完整 JSON payload | `[KNOWN, HIGH]` 不兼容 |
-| `events_core` truncated projection/MV | 无对应 typed incremental projection | `[KNOWN, HIGH]` 不兼容 |
-| scores / dataset-run items / blob log | PoC 只能以 generic event + synthetic Run 保存 | `[COMPUTED, HIGH]` 存得下,不等于 API/查询兼容 |
-| `bookmarked` / `public` 更新 | adapter 明确返回 unsupported | `[COMPUTED, HIGH]` 不兼容 |
-| trace/project/retention 删除 | 无逻辑 tombstone/delete contract | `[KNOWN, HIGH]` 不兼容 |
-| point/list | 可由少量索引列表达 | `[COMPUTED, HIGH]` 功能可做,性能门槛失败 |
-| tags/metadata/tools/cost facets | 主要埋在 JSON;无 Array/Map/Decimal 等同构列 | `[KNOWN, HIGH]` 不兼容 |
-| FTS | 非 Search 路径不支持 Langfuse token SQL | `[COMPUTED, HIGH]` 不兼容 |
-| dashboards/session aggregates | `argMaxIf`、`sumMap`、curried quantile、`WITH FILL` 等不兼容 | `[COMPUTED, HIGH]` 不兼容 |
-| JSONEachRow export | 可流式输出 generic event JSONL | `[COMPUTED, HIGH]` 格式/列语义仍不等同 |
-| Parquet export | public query adapter 不支持 | `[COMPUTED, HIGH]` 不兼容 |
-| Langfuse migrations/direct schema assumptions | 无 ClickHouse-compatible schema/migration surface | `[KNOWN, HIGH]` 不兼容 |
-
-## 静态代码证据
-
-[KNOWN, HIGH] Langfuse `events_full` 是 wide typed OLAP 表:`Array`/`Map`/`Decimal(18,12)`、materialized cost、bloom/minmax/text indexes、`ReplacingMergeTree(event_ts, is_deleted)`,且主键/排序键以 project、minute、trace、span、time 为中心。见相邻 Langfuse 仓库的 `packages/shared/clickhouse/migrations/unclustered/0039_create_events_full.up.sql:1`。
-
-[KNOWN, HIGH] `events_core` 由增量 materialized view 从 full 表生成,并将 I/O 和 metadata 值截断到 200 字符,承担轻量列表查询。见相邻 Langfuse 仓库的 `packages/shared/clickhouse/migrations/unclustered/0041_create_events_core_mv.up.sql:1`。
-
-[KNOWN, HIGH] Langfuse writer 为 events、scores、dataset-run items 和 blob log 提供 batch/retry/async acknowledged inserts;client 强制 `async_insert=1`、`wait_for_async_insert=1`。见相邻 Langfuse 仓库的 `worker/src/services/ClickhouseWriter/index.ts:35` 和 `packages/shared/src/server/clickhouse/client.ts:190`。
-
-[KNOWN, HIGH] Langfuse repository 会同步更新 full/core 的 `bookmarked/public`,并按 project+trace、project、project+time 执行删除;同时提供 JSON/raw text/Parquet 流式 export。见相邻 Langfuse 仓库的 `packages/shared/src/server/repositories/events.ts:1948`、`:2246`、`:2334`、`:2400`、`:3139`。
-
-[KNOWN, HIGH] query builder 使用 `argMaxIf`、`sumMap`、`groupUniqArray*` 和 `LIMIT 1 BY`,FTS 使用 `tokens` + `hasAllTokens` 再叠加精确 predicate。见相邻 Langfuse 仓库的 `packages/shared/src/server/queries/clickhouse-sql/event-query-builder.ts:522`、`:710`、`:1437` 和 `packages/shared/src/server/queries/clickhouse-sql/fts.ts:74`。
-
-[COMPUTED, HIGH] 对 `web/`、`worker/`、`packages/shared/` 下 `.ts/.tsx` 的词法扫描得到 423 个文件含 ClickHouse 大小写变体;`LIMIT 1 BY` 53 次、`argMaxIf` 35 次、`sumMap` 38 次、`hasAllTokens` 16 次、`WITH FILL` 21 次。这些是包含测试/注释的 occurrence counts,不是迁移工期估算,但足以反驳“只替换一个数据库 client 即可”的假设。
-
-[KNOWN, HIGH] pChronicle 的 canonical Lance writer 自述为 at-least-once、append-only,重复 `event_id` 每次都成为物理行;重维护不在 ingestion path。见 `crates/persisting-pchronicle/src/store/events/mod.rs`。
-
-[KNOWN, HIGH] pChronicle `EventRow` 只有 seq、标识/时间、kind/source、agent/session/call/trace/model 等索引字段,其他内容进入 `payload_json`。见 `crates/persisting-pchronicle/src/store/event_row.rs:7`。
-
-[KNOWN, HIGH] query engine 是 read-only,只允许单条 `SELECT/VALUES/DESCRIBE/EXPLAIN`。见 `crates/persisting-pchronicle/src/store/query_engine.rs:93`、`:569`。
-
-[KNOWN, HIGH] 当前实验 catalog 会在 snapshot discovery 时打开每个 source,并对 event source 执行 `SELECT * FROM events ORDER BY seq`、collect 后物化 normalized MemTables。见 `crates/persisting-pchronicle/src/store/catalog/`。
-
-[INFERRED, HIGH] Langfuse 的主访问模式是 project/time 范围 OLAP;当前 pChronicle 以 Run 为物理单元并在 catalog cold path 打开、归一化每个 source。210 sources 尚可在 2.10 s 打开,但该机制没有给出 source 数量增长时的稳定上界,因此不能从本次小规模结果外推到长期多租户项目。
-
-## SQL 兼容探针
-
-[COMPUTED, HIGH] 当前 DataFusion 路径无法 plan:`approx_top_k`、`argMax`、`arrayJoin`、`hasAllTokens`、`JSONExtractString`、`LIMIT 1 BY`、curried `quantile`、`sumMap`、`WITH FILL`。
-
-[COMPUTED, HIGH] `FINAL` 和 `PREWHERE` 文本在探针中没有报 parser/planner error。
-
-[INFERRED, HIGH] 这不证明语义兼容:`FINAL` 可能被当作 alias/no-op 接受,而 `PREWHERE` 能被解析也不意味着拥有 ClickHouse 的存储级 prewhere 和 latest-version 行为。因此二者没有记为通过。
-
-## ClickHouse best-practices 检查
-
-[KNOWN, HIGH] 已逐项检查该仓库 skill 中的 28 条规则。对本评审有直接影响的结论如下:
-
-- [KNOWN, HIGH] 按 `schema-pk-plan-before-creation`、`schema-pk-prioritize-filters`、`schema-pk-filter-on-orderby`、`schema-pk-cardinality-order`,对照 schema 保留了 Langfuse 的 project/time/trace 优先顺序;这正是 point/list/facet 快的关键结构之一。
-- [KNOWN, HIGH] 按 `schema-types-native-types`、`schema-types-lowcardinality`、`schema-types-minimize-bitwidth`、`schema-json-when-to-use`,常用过滤/聚合字段使用 native Array/Map/Decimal/LowCardinality;JSON 只作为完整 payload,不替代 typed projection。pChronicle 当前恰好缺这层 projection。
-- [KNOWN, HIGH] 按 `schema-partition-lifecycle`、`schema-partition-low-cardinality`、`schema-partition-query-tradeoffs`、`schema-partition-start-without`,复核了 Langfuse 的月分区与 retention 模型;本 PoC 没有为 pChronicle 创造第二套分区系统。
-- [KNOWN, HIGH] 按 `insert-batch-size`、`insert-async-small-batches`,ClickHouse 对照使用 batch 和 `async_insert=1, wait_for_async_insert=1`。`insert-format-native` 建议更高效的 Native 格式,但对照保留 Langfuse 实际 `JSONEachRow`,避免用更优但非现状的 producer 美化基线。
-- [KNOWN, HIGH] 按 `insert-optimize-avoid-final`、`insert-mutation-avoid-update`、`insert-mutation-avoid-delete`,频繁 mutation/`FINAL` 应慎用;但 Langfuse 已公开要求 bookmarks/public/delete/retention 语义,不能以“最佳实践”名义删掉。可行的替代仍需版本行+tombstone+latest projection,而当前 pChronicle 没有。
-- [KNOWN, HIGH] 按 `query-index-skipping-indices`、`query-mv-incremental`,对照保留 text/bloom/minmax indexes 与 full→core 增量 MV。`query-mv-refreshable` 不适用于这条 ingestion path。
-- [KNOWN, HIGH] `query-join-filter-before`、`query-join-consider-alternatives`、`query-join-use-any`、`query-join-choose-algorithm`、`query-join-null-handling` 已检查;本次 point/list/facet 对照没有 join,故它们不是判定因素。
-- [KNOWN, HIGH] `schema-types-avoid-nullable` 和 `schema-types-enum` 已检查;Langfuse schema 的 nullable 时间/版本字段和 LowCardinality strings 属于现有兼容约束,本 PoC 不借机重构它们。
-
-## 故障、隔离与安全
-
-[COMPUTED, HIGH] `langfuse_backend_faults` 通过 2/2:子进程 append 并输出 ack 后被强制终止,新的 process/runtime 仍读到唯一目标行。
-
-[COMPUTED, HIGH] targeted regression tests 通过:`production_scale` 5 passed、1 ignored;`query_engine` 14 passed;`store::catalog` filter 6 passed。覆盖 writer fencing、固定 snapshot 在 append 期间不漂移、独立 Runs、maintenance/restart 和 read-only SQL boundary。
-
-[KNOWN, HIGH] 当前 pChronicle server 是 loopback-only single-user UI,并明确因为没有 authentication 而拒绝 non-loopback bind;它提供 query/export/maintenance routes,但没有供 Langfuse web/worker 使用的 canonical authenticated network append service。见 `crates/persisting-pchronicle-server/src/lib.rs:1`、`:155`、`:168`。
-
-[COMPUTED, HIGH] PoC 中 `project_id → agent_id` 且 cross-project negative point 为 0,证明当前测试 adapter 没有在该查询中串项目。
-
-[INFERRED, HIGH] 这不等于服务级 tenant isolation:共享 catalog 能看见多个项目,本身没有 Langfuse project RBAC/row policy。若暴露任意 SQL endpoint,调用者可能绕过 adapter 的 project predicate。
-
-[INFERRED, HIGH] 因此未来旁路 dual-write 只能使用服务端固定 endpoint、强认证(mTLS 或短期 service token)、每请求显式 project scope、最小化 SQL surface、传输/静态加密和无 payload/credential 日志;不得允许 tenant 自定义出站 host。此次实现没有新增生产网络端点或扩大权限。
-
-## 为什么不是“再补几个函数”
-
-[INFERRED, HIGH] 满足 full Langfuse v4 至少要新增以下五个耦合层:
-
-1. event version、logical tombstone、latest-row projection,以及 update/delete/retention 的一致性契约;
-2. `events_full/events_core/scores/dataset_run_items/blob_log` 的 typed projections、索引、迁移和生命周期;
-3. dashboard、sessions、facets、FTS、export 对应的 SQL/执行语义和差分测试;
-4. web/worker 可用的 authenticated network write/query service 与 tenant authorization;
-5. 跨大量 Runs 的 durable metadata、source pruning、增量 catalog,而不是 cold path 全打开和全 normalize。
-
-[INFERRED, HIGH] 这些层彼此依赖:没有 version projection 就无法正确聚合;没有 tenant-aware service 就不能部署;没有 typed/pruned projection 就无法达到现有查询延迟。它们构成产品级扩展,不属于批准的 bounded integration。
-
-## 推荐落点
-
-[INFERRED, HIGH] 建议采用并行角色,而非替换角色:
-
-- ClickHouse 继续承载 Langfuse hot OLAP、lists/facets/dashboards/FTS、更新删除以及 JSON/Parquet export;
-- pChronicle 接收原始、追加式 trajectory 副本,服务 replay、审计证据、冷归档、离线研究;
-- dual-write 失败必须有独立重试/回补水位,不能让 pChronicle ack 替代现有 ClickHouse ack;
-- project deletion/retention 必须在两端有可审计的 lifecycle 协议后,才可处理真实租户数据。
-
-[INFERRED, MED] 一个后续有界 pilot 可以只做“单 project、原始 event dual-write + trace replay”,不接管 dashboards、FTS、facets、updates/deletes,也不宣称它是 Langfuse analytics backend。
-
-## 限制与未知边界
-
-[KNOWN, HIGH] 记录环境为 Apple M4、24 GiB RAM、macOS 26.5;pChronicle 是本地 release binary,ClickHouse 是本地 Docker container。没有隔离所有后台负载,因此数值不是可发布的容量基准。
-
-[KNOWN, HIGH] 本次只有单进程、一个并发查询、200 traces/210 sources、两次主要完整跑;没有 HA、远端对象存储故障、长期 compaction、百万级 Runs、多日 retention 或真实租户 payload。
-
-[KNOWN, HIGH] ClickHouse PoC schema 是 Langfuse v4 相关模式的代表性子集,不是完整部署;full API/UI 没有做 E2E,因为静态兼容矩阵已触发 hard NO-GO。该限制不会反转 mutation/dedup/SQL/export 的确定性失败,但会限制对所有边缘查询的覆盖度。
-
-[KNOWN, HIGH] pChronicle catalog/query-engine 基于评审时未提交的实验代码。结论只对上文钉住的 snapshot 有效,后续实现变化必须重跑同一套 fixture、fault tests 和语义矩阵。
-
-[RULES I BROKE]: none
diff --git a/benchmark/langfuse-pchronicle-review/RETEST-2026-08-12.zh.md b/benchmark/langfuse-pchronicle-review/RETEST-2026-08-12.zh.md
deleted file mode 100644
index 5db05dfef..000000000
--- a/benchmark/langfuse-pchronicle-review/RETEST-2026-08-12.zh.md
+++ /dev/null
@@ -1,86 +0,0 @@
-# pChronicle 优化后复测(2026-08-12)
-
-## 结论
-
-**[INFERRED, HIGH] 本次 lazy catalog / source-pruning 优化有效;按“serverless Lite tracing backend”范围,性能与可靠性可以进入 Conditional GO 原型阶段。**
-
-[COMPUTED, HIGH] 相同 113,000-row fixture 下,catalog cold start 从 2,104.04 ms 降到 35.04 ms,warm point/list/simple aggregate P95 分别改善 34.8%/49.9%/63.2%。
-
-[COMPUTED, HIGH] 当 adapter 能把 Run 映射成 exact `_file_` predicate 时,catalog point P95 为 3.01 ms;此前记录的 ClickHouse point P95 为 4.83 ms。这个本地结果说明 per-Run Lance format 不是 point path 的主要性能障碍。
-
-[COMPUTED, HIGH] project-path pruning 后 list P95 为 167.17 ms,满足 Lite 场景 1 s 绝对门槛,但仍是 ClickHouse 记录值的 56.3×。跨 Run list/aggregate 仍是主要差距。
-
-[KNOWN, HIGH] full Langfuse v4 harness 仍会报告 unsupported mutation/delete/FTS/dashboard/Parquet 和 duplicate-version 语义。这些能力本次没有实现;若 Lite profile 明确关闭它们,就不应把它们解读成性能回归。
-
-## 可比性
-
-[COMPUTED, HIGH] 新旧 pChronicle 与 ClickHouse 使用相同 fixture:238,470,862 bytes、113,000 rows,SHA-256 `1b2ef8ed2f51e6d85a9907bbbebbdafbc4c961d8c4368b2a20b737f94cfc4c90`。
-
-[KNOWN, HIGH] Persisting HEAD 仍为 `94531cf903e5abc336de347588fb1858e9d52b6a`;本次测量包含 2026-08-12 工作区内未提交的 lazy catalog/query changes。关键文件 SHA 已写入 `recorded-results-2026-08-12.json`。
-
-[KNOWN, HIGH] ClickHouse 未重新启动;因 fixture SHA 未变,复用 2026-08-11 在同一机器记录的 ClickHouse 25.12.11.4 baseline,避免额外容器噪声。
-
-## 结果
-
-| Metric | 旧 pChronicle | 新 pChronicle | 变化 |
-|---|---:|---:|---:|
-| catalog cold start | 2,104.04 ms | 35.04 ms | `[COMPUTED, HIGH]` -98.3% |
-| catalog 后 RSS | 1,117,782,016 B | 316,669,952 B | `[COMPUTED, HIGH]` -71.7% |
-| 新版执行 unpruned queries 后 RSS | N/A | 624,885,760 B | `[COMPUTED, HIGH]` 仍低于旧 eager catalog 44.1% |
-| warm point P95 | 395.87 ms | 258.13 ms | `[COMPUTED, HIGH]` -34.8% |
-| warm list P95 | 564.82 ms | 282.80 ms | `[COMPUTED, HIGH]` -49.9% |
-| simple model aggregate P95 | 806.70 ms | 296.71 ms | `[COMPUTED, HIGH]` -63.2% |
-| concurrent query P95 @10 eps | 350.41 ms | 243.45 ms | `[COMPUTED, HIGH]` -30.5% |
-| JSON export TTFB / 50k rows | 429.09 ms | 336.69 ms | `[COMPUTED, HIGH]` -21.5% |
-| append throughput | 9,374.59 rows/s | 9,185.15 rows/s | `[COMPUTED, MED]` -2.0%,视为无改善/运行噪声 |
-| visibility P95 | 4.13 ms | 5.46 ms | `[COMPUTED, HIGH]` 仍远低于 2 s |
-
-[COMPUTED, HIGH] 当前代码三次完整跑的 catalog cold start 为 35.04–35.36 ms,warm point 为 258.13–262.03 ms,aggregate 为 292.39–299.31 ms;这些核心改善具有重复性。warm list 为 282.80–352.99 ms,波动较大但仍明显好于旧版 564.82–666.71 ms。
-
-## Lazy/pruning 路径
-
-| Path | Catalog/open | Query P95 | End-to-end approximation |
-|---|---:|---:|---:|
-| fresh global point,210 sources | 10.84 ms | 302.94 ms | `[COMPUTED, HIGH]` 313.77 ms |
-| fresh global list,210 sources | 12.47 ms | 326.68 ms | `[COMPUTED, HIGH]` 339.16 ms |
-| direct Run point | 0.51 ms | 2.14 ms | `[COMPUTED, HIGH]` 2.64 ms |
-| exact `_file_` catalog point | 12.34 ms | 3.01 ms | `[COMPUTED, HIGH]` 15.35 ms |
-| project `_file_ LIKE` list | 12.02 ms | 167.17 ms | `[COMPUTED, HIGH]` 179.20 ms |
-
-[INFERRED, HIGH] 结果支持此前的因果判断:主要瓶颈是跨 Run fan-out,而不是 Lance 文件格式。exact source 被选中后,catalog provider 能在 resolve 前排除其他 209 个 source,point latency 即降到毫秒级。
-
-[INFERRED, HIGH] Langfuse Lite adapter 必须把已知 project/trace/Run 映射成 `_file_` predicate 或直接 Run URI;只传 `agent_id/event_id` 会触发全 catalog source resolution,丢失本次优化的大部分收益。
-
-[COMPUTED, HIGH] restart 的 append+catalog 为 51.76 ms,随后首次全局 visibility query 为 344.02 ms,端到端 395.78 ms且目标行可见。该拆分避免把 lazy catalog 的快速构建误报成完整 restart latency。
-
-## Durability 与回归测试
-
-[COMPUTED, HIGH] 所有 targeted tests 通过:
-
-- `langfuse_backend_faults`: 2 passed;
-- `production_scale`: 5 passed,1 ignored stress test;
-- `query_engine`: 14 passed;
-- catalog unit tests: 9 passed,包括 lazy resolution 与 source pruning;
-- raw event Lance/row unit tests: 21 passed;
-- `persisting-pchronicle-server --lib`: 16 passed。
-
-[KNOWN, HIGH] 唯一重复出现的 warning 是 macOS linker 的 `__eh_frame section too large`;没有测试因此失败。
-
-## Rules checked
-
-- [KNOWN, HIGH] Per `schema-pk-prioritize-filters` / `schema-pk-filter-on-orderby`:exact `_file_` 相当于先使用物理 source key;这是新版 point path 的关键合规点。
-- [KNOWN, HIGH] Per `schema-partition-low-cardinality`:per-Run source 仍是高基数布局;lazy resolution 缓解 metadata/open cost,但 project-wide query 仍随选中 Run 数增长。
-- [KNOWN, HIGH] Per `query-index-skipping-indices`:source-level pruning 已有效;source 内 `event_id/time` scalar index 仍可进一步降低 scan。
-- [KNOWN, HIGH] Per `schema-types-native-types` / `schema-json-when-to-use`:当前 timestamp/known Langfuse fields 仍偏 UTF-8/opaque JSON;这更影响跨 Run sort/aggregate,而非已路由的 point。
-- [KNOWN, HIGH] Per `query-mv-incremental`:Lite profile 若不做 dashboards,可暂不引入完整 OLAP MV;若未来需要 project-wide analytics,应增加小型 typed projection,而不是重新 eager-normalize catalog。
-- [KNOWN, HIGH] Per `insert-batch-size` / `insert-async-small-batches`:本次优化未改善 append throughput;per-Run commit/fragment batching 仍是独立优化方向。
-
-## 判定
-
-[INFERRED, HIGH] 对明确缩减能力的 `PCHRONICLE_LITE`:性能、cold start、内存、visibility 和 durability 达到本次 dev-scale absolute gates,可以继续实现 adapter prototype。
-
-[INFERRED, HIGH] 对完整 ClickHouse-compatible backend:项目级查询相对 ClickHouse 的差距和未实现语义仍存在,因此旧的 full-v4 NO-GO 不被本次性能优化推翻。
-
-[INFERRED, HIGH] 下一优先级不是替换存储格式,而是让 adapter 强制走 project/Run source pruning,并增加 project-level run index;之后再测 1/10/100/1000 Runs 的规模曲线。
-
-[RULES I BROKE]: none
diff --git a/benchmark/langfuse-pchronicle-review/SERVER-ACCELERATION-2026-08-12.zh.md b/benchmark/langfuse-pchronicle-review/SERVER-ACCELERATION-2026-08-12.zh.md
deleted file mode 100644
index 003f1e006..000000000
--- a/benchmark/langfuse-pchronicle-review/SERVER-ACCELERATION-2026-08-12.zh.md
+++ /dev/null
@@ -1,70 +0,0 @@
-# pChronicle Server source-routing 加速复测
-
-日期:2026-08-12
-
-## 结论
-
-本轮 Server 加速把可唯一定位的 warm point query 从 pChronicle 原来的 `258.13ms` 降到
-HTTP P95 `4.72ms`,与此前 ClickHouse `4.83ms` 基线处于同一数量级。项目 list 和 aggregate
-分别降到 `151.05ms` 和 `165.41ms`,绝对延迟可用,但仍约为 ClickHouse 的 `50.8x` 和
-`47.7x`。
-
-因此 source-routing index 已经消除了 point query 的主要 fan-out 成本;项目级扫描仍需要
-访问 106 个 source,剩余差距主要来自多 source scan/merge/sort,而不是 Catalog 初始化。
-
-## 实现边界
-
-- `DatasetCatalogSnapshot`、`CatalogDataset`、`DiscoveredSource` 和持久化格式未改变。
-- `CatalogRuntime` 同代持有可重建的 `ServerAcceleration`。
-- Run、event identity、event partition 分别 lazy single-flight 构建。
-- value 使用每代随机 keyed 64-bit fingerprint,collision 只扩大 source 候选;原 SQL 谓词
- 负责最终过滤。
-- 只改写简单单表查询的顶层必要等值/`IN` 条件;不确定时回退原 SQL。
-- 索引构建使用 Arrow stream,并限制为 100 万行、100 万 distinct value。
-- refresh 原子发布新 snapshot/engine/空索引,旧索引不会跨 generation 复用。
-
-## 113k 行 release 结果
-
-| 场景 | 结果 |
-|---|---:|
-| Catalog cold(3 次中位数) | `37.15ms` |
-| cold identity point,含首次 identity index | `134.26–166.17ms`,中位数 `159.45ms` |
-| Catalog + cold identity point(中位数相加) | `196.60ms` |
-| warm point HTTP P50 / P95 | `4.15 / 4.72ms` |
-| 首次 project list,含 partition index | `312.94ms` |
-| warm project list HTTP P50 / P95 | `147.24 / 151.05ms` |
-| warm project aggregate HTTP P50 / P95 | `158.63 / 165.41ms` |
-
-Warm 数字各采样 20 次。point 命中 1 个 source;project-a list/aggregate 命中 106 个 source。
-
-## 与 ClickHouse 基线比较
-
-| 场景 | pChronicle Server P95 | ClickHouse P95 | 比率 |
-|---|---:|---:|---:|
-| point | `4.72ms` | `4.83ms` | `0.98x` |
-| project list | `151.05ms` | `2.97ms` | `50.84x` |
-| project aggregate | `165.41ms` | `3.47ms` | `47.72x` |
-
-这里 pChronicle 包含 Axum 和 loopback HTTP 开销,ClickHouse 数字复用 2026-08-11 的相同
-fixture 基线,并非同轮并发 A/B。因此 point 的约 2% 差异只能解释为接近持平,不能解释为
-严格引擎胜负。
-
-## 内存说明
-
-partition index 记录 113,031 行、211 个 source、213 个 distinct fingerprint;identity index
-记录相同行数和 113,132 个 distinct fingerprint。fresh process 中 Catalog 后 RSS 为
-`26.2MiB`,完成 identity index 与 point query 后为 `101.9MiB`。
-
-这个 RSS 增量同时包含 211 个 LazySource 的首次解析、Lance/DataFusion scan buffer、索引和
-allocator retention,不能当作索引对象自身大小。两级 lazy index 的意义是:只运行项目列表
-时不会建立 113k-value identity map;超出硬限制时索引不会发布,查询回退原 fan-out 路径。
-
-## 验证
-
-- `cargo test -p persisting-pchronicle-server --lib`:21 passed。
-- `cargo clippy -p persisting-pchronicle-server --all-targets -- -D warnings`:passed。
-- 真实双 `events.lance` 测试验证 point/project 路由、结果等价和两级 lazy build。
-- Server 集成测试验证 `_file_` 注入、quoted alias、显式 `_file_` 保留以及 refresh 清空索引。
-
-机器可读数据见
-[`server-acceleration-results-2026-08-12.json`](server-acceleration-results-2026-08-12.json)。
diff --git a/benchmark/langfuse-pchronicle-review/clickhouse_baseline.py b/benchmark/langfuse-pchronicle-review/clickhouse_baseline.py
deleted file mode 100644
index a0eb15f20..000000000
--- a/benchmark/langfuse-pchronicle-review/clickhouse_baseline.py
+++ /dev/null
@@ -1,647 +0,0 @@
-#!/usr/bin/env python3
-"""Run the Langfuse-shaped fixture against an isolated ClickHouse 25.12 server.
-
-The script uses only the ClickHouse HTTP API, never prints credentials, and
-writes one JSON report. It is intended for the disposable database named
-``langfuse_pchronicle_review``.
-"""
-
-from __future__ import annotations
-
-import argparse
-import json
-import math
-import pathlib
-import threading
-import time
-import urllib.error
-import urllib.parse
-import urllib.request
-from typing import Any, Iterable
-
-DATABASE = "langfuse_pchronicle_review"
-LOAD_ROWS_PER_SECOND = 10
-
-COLUMNS = """
- row_kind LowCardinality(String),
- project_id String,
- trace_id String,
- span_id String,
- parent_span_id String,
- logical_id String,
- event_ts DateTime64(3),
- start_time DateTime64(3),
- name LowCardinality(String),
- event_type LowCardinality(String),
- session_id String,
- user_id String,
- environment LowCardinality(String),
- tags Array(String),
- metadata_names Array(String),
- metadata_values Array(String),
- tool_names Array(String),
- model LowCardinality(String),
- input String CODEC(ZSTD(3)),
- output String CODEC(ZSTD(3)),
- usage_input UInt64,
- usage_output UInt64,
- total_cost Decimal(38, 12),
- version UInt32,
- is_deleted Bool,
- bookmarked Bool,
- public Bool,
- dataset_id String,
- dataset_run_id String,
- storage_run_id String,
- payload_json String CODEC(ZSTD(3))
-"""
-
-COLUMN_NAMES = [
- "row_kind",
- "project_id",
- "trace_id",
- "span_id",
- "parent_span_id",
- "logical_id",
- "event_ts",
- "start_time",
- "name",
- "event_type",
- "session_id",
- "user_id",
- "environment",
- "tags",
- "metadata_names",
- "metadata_values",
- "tool_names",
- "model",
- "input",
- "output",
- "usage_input",
- "usage_output",
- "total_cost",
- "version",
- "is_deleted",
- "bookmarked",
- "public",
- "dataset_id",
- "dataset_run_id",
- "storage_run_id",
- "payload_json",
-]
-
-
-class ClickHouseHttp:
- def __init__(self, url: str, user: str, password: str, timeout: float = 120.0):
- self.url = url.rstrip("/")
- self.timeout = timeout
- self.headers = {
- "X-ClickHouse-User": user,
- "X-ClickHouse-Key": password,
- }
-
- def request(
- self,
- query: str,
- body: bytes | None = None,
- settings: dict[str, str | int] | None = None,
- ) -> bytes:
- params = dict(settings or {})
- if body is not None:
- params["query"] = query
- data = body
- else:
- data = query.encode("utf-8")
- url = self.url
- if params:
- url += "?" + urllib.parse.urlencode(params)
- request = urllib.request.Request(url, data=data, headers=self.headers, method="POST")
- try:
- with urllib.request.urlopen(request, timeout=self.timeout) as response:
- return response.read()
- except urllib.error.HTTPError as error:
- detail = error.read().decode("utf-8", errors="replace")
- raise RuntimeError(f"ClickHouse HTTP {error.code}: {detail[:1000]}") from error
-
- def json_query(self, query: str) -> dict[str, Any]:
- document = self.request(query.rstrip().rstrip(";") + " FORMAT JSON")
- return json.loads(document)
-
- def scalar(self, query: str) -> Any:
- result = self.json_query(query)
- if len(result.get("data", [])) != 1 or len(result["data"][0]) != 1:
- raise RuntimeError(f"expected one scalar result for query: {query}")
- return next(iter(result["data"][0].values()))
-
- def stream(self, query: str, output_format: str) -> dict[str, Any]:
- sql = query.rstrip().rstrip(";") + f" FORMAT {output_format}"
- request = urllib.request.Request(
- self.url,
- data=sql.encode("utf-8"),
- headers=self.headers,
- method="POST",
- )
- started = time.perf_counter()
- with urllib.request.urlopen(request, timeout=self.timeout) as response:
- first = response.read(1)
- first_byte = time.perf_counter()
- remainder = response.read()
- finished = time.perf_counter()
- return {
- "bytes": len(first) + len(remainder),
- "first_byte_ms": (first_byte - started) * 1000.0,
- "elapsed_ms": (finished - started) * 1000.0,
- }
-
-
-def parse_args() -> argparse.Namespace:
- parser = argparse.ArgumentParser()
- parser.add_argument("--url", required=True, help="ClickHouse HTTP URL")
- parser.add_argument("--user", default="default")
- parser.add_argument("--password", default="")
- parser.add_argument("--fixture", required=True, type=pathlib.Path)
- parser.add_argument("--output", required=True, type=pathlib.Path)
- parser.add_argument("--load-seconds", type=int, default=3)
- parser.add_argument("--insert-batch-rows", type=int, default=10_000)
- return parser.parse_args()
-
-
-def execute_ddl(client: ClickHouseHttp) -> None:
- client.request(f"DROP DATABASE IF EXISTS {DATABASE}")
- client.request(f"CREATE DATABASE {DATABASE}")
- client.request(
- f"""
- CREATE TABLE {DATABASE}.staging ({COLUMNS})
- ENGINE = MergeTree
- ORDER BY (row_kind, project_id, event_ts, logical_id)
- """
- )
- client.request(
- f"""
- CREATE TABLE {DATABASE}.events_full (
- {COLUMNS},
- INDEX idx_span_id span_id TYPE bloom_filter(0.01) GRANULARITY 1,
- INDEX idx_trace_id trace_id TYPE bloom_filter(0.01) GRANULARITY 1,
- INDEX idx_user_id user_id TYPE bloom_filter(0.01) GRANULARITY 1,
- INDEX idx_session_id session_id TYPE bloom_filter(0.01) GRANULARITY 1,
- INDEX idx_fts_input lower(input) TYPE text(tokenizer = splitByNonAlpha),
- INDEX idx_fts_output lower(output) TYPE text(tokenizer = splitByNonAlpha),
- INDEX idx_fts_metadata_values metadata_values TYPE text(tokenizer = splitByNonAlpha)
- )
- ENGINE = ReplacingMergeTree(event_ts, is_deleted)
- PARTITION BY toYYYYMM(start_time)
- PRIMARY KEY (project_id, toStartOfMinute(start_time), xxHash32(trace_id))
- ORDER BY (project_id, toStartOfMinute(start_time), xxHash32(trace_id), span_id, start_time)
- SAMPLE BY xxHash32(trace_id)
- SETTINGS index_granularity_bytes = '64Mi', enable_full_text_index = 1
- """
- )
- client.request(
- f"""
- CREATE TABLE {DATABASE}.events_core (
- {COLUMNS},
- INDEX idx_trace_id trace_id TYPE bloom_filter(0.01) GRANULARITY 1,
- INDEX idx_fts_metadata_values metadata_values TYPE text(tokenizer = splitByNonAlpha)
- )
- ENGINE = ReplacingMergeTree(event_ts, is_deleted)
- PARTITION BY toYYYYMM(start_time)
- PRIMARY KEY (project_id, toStartOfMinute(start_time), xxHash32(trace_id))
- ORDER BY (project_id, toStartOfMinute(start_time), xxHash32(trace_id), span_id, start_time)
- SAMPLE BY xxHash32(trace_id)
- SETTINGS enable_full_text_index = 1
- """
- )
- projected = []
- for column in COLUMN_NAMES:
- if column == "input":
- projected.append("leftUTF8(input, 200) AS input")
- elif column == "output":
- projected.append("leftUTF8(output, 200) AS output")
- elif column == "metadata_values":
- projected.append(
- "arrayMap(value -> leftUTF8(value, 200), metadata_values) AS metadata_values"
- )
- else:
- projected.append(column)
- client.request(
- f"""
- CREATE MATERIALIZED VIEW {DATABASE}.events_core_mv TO {DATABASE}.events_core AS
- SELECT {", ".join(projected)} FROM {DATABASE}.events_full
- """
- )
- for table, order_by in [
- ("scores", "(project_id, name, event_ts, logical_id)"),
- ("dataset_run_items_rmt", "(project_id, dataset_id, dataset_run_id, logical_id)"),
- ("blob_storage_file_log", "(project_id, event_ts, logical_id)"),
- (
- "mutation_events",
- "(project_id, toStartOfMinute(start_time), trace_id, span_id, start_time)",
- ),
- ]:
- client.request(
- f"""
- CREATE TABLE {DATABASE}.{table} ({COLUMNS})
- ENGINE = ReplacingMergeTree(event_ts, is_deleted)
- PARTITION BY toYYYYMM(start_time)
- ORDER BY {order_by}
- """
- )
-
-
-def batches(lines: Iterable[bytes], size: int) -> Iterable[bytes]:
- batch: list[bytes] = []
- for line in lines:
- if line.strip():
- batch.append(line)
- if len(batch) == size:
- yield b"".join(batch)
- batch = []
- if batch:
- yield b"".join(batch)
-
-
-def load_fixture(client: ClickHouseHttp, fixture: pathlib.Path, batch_rows: int) -> dict[str, Any]:
- started = time.perf_counter()
- inserted = 0
- insert_latencies: list[float] = []
- with fixture.open("rb") as source:
- for body in batches(source, batch_rows):
- row_count = body.count(b"\n")
- insert_started = time.perf_counter()
- client.request(
- f"INSERT INTO {DATABASE}.staging FORMAT JSONEachRow",
- body=body,
- settings={
- "async_insert": 1,
- "wait_for_async_insert": 1,
- "input_format_skip_unknown_fields": 1,
- },
- )
- insert_latencies.append((time.perf_counter() - insert_started) * 1000.0)
- inserted += row_count
- elapsed = time.perf_counter() - started
- return {
- "acknowledged_rows": inserted,
- "elapsed_ms": elapsed * 1000.0,
- "rows_per_second": inserted / elapsed,
- "batch_ack_p95_ms": percentile(insert_latencies, 0.95),
- }
-
-
-def populate_tables(client: ClickHouseHttp) -> None:
- columns = ", ".join(COLUMN_NAMES)
- for table, row_kind in [
- ("events_full", "event"),
- ("scores", "score"),
- ("dataset_run_items_rmt", "dataset_run_item"),
- ("blob_storage_file_log", "blob_storage_file_log"),
- ]:
- client.request(
- f"INSERT INTO {DATABASE}.{table} ({columns}) "
- f"SELECT {columns} FROM {DATABASE}.staging WHERE row_kind = '{row_kind}'"
- )
- client.request(
- f"INSERT INTO {DATABASE}.mutation_events ({columns}) "
- f"SELECT {columns} FROM {DATABASE}.staging WHERE row_kind = 'event'"
- )
-
-
-def measure_query(client: ClickHouseHttp, sql: str, repetitions: int = 7) -> dict[str, Any]:
- samples: list[float] = []
- result_rows = 0
- for _ in range(repetitions):
- started = time.perf_counter()
- result = client.json_query(sql)
- samples.append((time.perf_counter() - started) * 1000.0)
- result_rows = int(result.get("rows", len(result.get("data", []))))
- return {
- "repetitions": repetitions,
- "result_rows": result_rows,
- "p50_ms": percentile(samples, 0.50),
- "p95_ms": percentile(samples, 0.95),
- "max_ms": max(samples),
- }
-
-
-def run_load_phase(
- client: ClickHouseHttp, fixture: pathlib.Path, load_seconds: int
-) -> dict[str, Any]:
- with fixture.open("r", encoding="utf-8") as source:
- template = json.loads(next(source))
- query_samples: list[float] = []
- query_errors: list[str] = []
- stop = threading.Event()
-
- def query_loop() -> None:
- while not stop.is_set():
- started = time.perf_counter()
- try:
- client.scalar(
- f"SELECT count() FROM {DATABASE}.events_core WHERE project_id = 'project-a'"
- )
- query_samples.append((time.perf_counter() - started) * 1000.0)
- except Exception as error: # recorded and surfaced in the report
- query_errors.append(str(error)[:500])
- return
- stop.wait(0.25)
-
- thread = threading.Thread(target=query_loop, name="clickhouse-review-query", daemon=True)
- thread.start()
- ack_samples: list[float] = []
- visibility_samples: list[float] = []
- load_rows = 0
- phase_started = time.perf_counter()
- load_elapsed = 0.0
- try:
- for second in range(load_seconds):
- rows = []
- for offset in range(LOAD_ROWS_PER_SECOND):
- index = second * LOAD_ROWS_PER_SECOND + offset
- row = dict(template)
- row["logical_id"] = f"project-a-clickhouse-load-{index:04d}"
- row["trace_id"] = "project-a-clickhouse-load-trace"
- row["span_id"] = f"project-a-clickhouse-load-span-{index:04d}"
- row["storage_run_id"] = row["trace_id"]
- row["event_ts"] = f"2026-01-01 00:05:{index // 10:02d}.{offset * 100:03d}"
- row["start_time"] = row["event_ts"]
- row["payload_json"] = json.dumps(row, ensure_ascii=False, separators=(",", ":"))
- rows.append(json.dumps(row, ensure_ascii=False, separators=(",", ":")))
- target = phase_started + second + 1
- remaining = target - time.perf_counter()
- if remaining > 0:
- time.sleep(remaining)
- body = ("\n".join(rows) + "\n").encode("utf-8")
- started = time.perf_counter()
- client.request(
- f"INSERT INTO {DATABASE}.events_full FORMAT JSONEachRow",
- body=body,
- settings={
- "async_insert": 1,
- "wait_for_async_insert": 1,
- "input_format_skip_unknown_fields": 1,
- },
- )
- ack_samples.append((time.perf_counter() - started) * 1000.0)
- load_rows += len(rows)
- visibility_started = time.perf_counter()
- visible = int(
- client.scalar(
- f"SELECT count() FROM {DATABASE}.events_full "
- "WHERE logical_id LIKE 'project-a-clickhouse-load-%'"
- )
- )
- if visible != load_rows:
- raise RuntimeError(
- f"acknowledged ClickHouse rows not visible: expected {load_rows}, got {visible}"
- )
- visibility_samples.append((time.perf_counter() - visibility_started) * 1000.0)
- load_elapsed = time.perf_counter() - phase_started
- finally:
- stop.set()
- thread.join(timeout=10)
- visible = int(
- client.scalar(
- f"SELECT count() FROM {DATABASE}.events_full "
- "WHERE logical_id LIKE 'project-a-clickhouse-load-%'"
- )
- )
- return {
- "rows": load_rows,
- "elapsed_ms": load_elapsed * 1000.0,
- "effective_rows_per_second": load_rows / load_elapsed,
- "visible_rows": visible,
- "zero_acknowledged_loss": visible == load_rows,
- "ack_p50_ms": percentile(ack_samples, 0.50),
- "ack_p95_ms": percentile(ack_samples, 0.95),
- "visibility_p50_ms": percentile(visibility_samples, 0.50),
- "visibility_p95_ms": percentile(visibility_samples, 0.95),
- "concurrent_query_p95_ms": percentile(query_samples, 0.95),
- "concurrent_query_errors": query_errors,
- }
-
-
-def run_mutations(client: ClickHouseHttp) -> dict[str, Any]:
- update_started = time.perf_counter()
- for table in ("events_full", "events_core"):
- client.request(
- f"ALTER TABLE {DATABASE}.{table} UPDATE bookmarked = true, public = true "
- "WHERE project_id = 'project-a' AND logical_id = 'project-a-event-00000000' "
- "SETTINGS mutations_sync = 2"
- )
- update_ms = (time.perf_counter() - update_started) * 1000.0
- updated_full = int(
- client.scalar(
- f"SELECT count() FROM {DATABASE}.events_full WHERE project_id = 'project-a' "
- "AND logical_id = 'project-a-event-00000000' AND bookmarked AND public"
- )
- )
- updated_core = int(
- client.scalar(
- f"SELECT count() FROM {DATABASE}.events_core WHERE project_id = 'project-a' "
- "AND logical_id = 'project-a-event-00000000' AND bookmarked AND public"
- )
- )
-
- trace_started = time.perf_counter()
- client.request(
- f"DELETE FROM {DATABASE}.mutation_events "
- "WHERE project_id = 'project-a' AND trace_id = 'project-a-trace-0000' "
- "SETTINGS mutations_sync = 2"
- )
- trace_delete_ms = (time.perf_counter() - trace_started) * 1000.0
- trace_remaining = int(
- client.scalar(
- f"SELECT count() FROM {DATABASE}.mutation_events "
- "WHERE project_id = 'project-a' AND trace_id = 'project-a-trace-0000'"
- )
- )
-
- retention_started = time.perf_counter()
- client.request(
- f"DELETE FROM {DATABASE}.mutation_events "
- "WHERE project_id = 'project-a' AND start_time < '2026-01-01 00:00:30.000' "
- "SETTINGS mutations_sync = 2"
- )
- retention_delete_ms = (time.perf_counter() - retention_started) * 1000.0
- retention_remaining = int(
- client.scalar(
- f"SELECT count() FROM {DATABASE}.mutation_events "
- "WHERE project_id = 'project-a' AND start_time < '2026-01-01 00:00:30.000'"
- )
- )
-
- project_started = time.perf_counter()
- client.request(
- f"DELETE FROM {DATABASE}.mutation_events WHERE project_id = 'project-b' "
- "SETTINGS mutations_sync = 2"
- )
- project_delete_ms = (time.perf_counter() - project_started) * 1000.0
- project_remaining = int(
- client.scalar(
- f"SELECT count() FROM {DATABASE}.mutation_events WHERE project_id = 'project-b'"
- )
- )
- return {
- "update_flags": {
- "elapsed_ms": update_ms,
- "events_full_visible": updated_full > 0,
- "events_core_visible": updated_core > 0,
- },
- "delete_trace": {
- "elapsed_ms": trace_delete_ms,
- "remaining_rows": trace_remaining,
- },
- "delete_retention": {
- "elapsed_ms": retention_delete_ms,
- "remaining_rows": retention_remaining,
- },
- "delete_project": {
- "elapsed_ms": project_delete_ms,
- "remaining_rows": project_remaining,
- },
- }
-
-
-def percentile(samples: list[float], quantile: float) -> float:
- if not samples:
- return 0.0
- ordered = sorted(samples)
- rank = math.ceil((len(ordered) - 1) * quantile)
- return ordered[min(rank, len(ordered) - 1)]
-
-
-def main() -> None:
- args = parse_args()
- if args.load_seconds <= 0 or args.insert_batch_rows <= 0:
- raise SystemExit("load seconds and insert batch rows must be positive")
- if not args.fixture.is_file():
- raise SystemExit(f"fixture does not exist: {args.fixture}")
- if args.output.exists():
- raise SystemExit(f"refusing to overwrite report: {args.output}")
-
- client = ClickHouseHttp(args.url, args.user, args.password)
- version = str(client.scalar("SELECT version()"))
- execute_ddl(client)
- insert_metrics = load_fixture(client, args.fixture, args.insert_batch_rows)
- populate_started = time.perf_counter()
- populate_tables(client)
- populate_ms = (time.perf_counter() - populate_started) * 1000.0
-
- counts = {
- table: int(client.scalar(f"SELECT count() FROM {DATABASE}.{table}"))
- for table in (
- "staging",
- "events_full",
- "events_core",
- "scores",
- "dataset_run_items_rmt",
- "blob_storage_file_log",
- )
- }
- event_distinct_ids = int(
- client.scalar(f"SELECT uniqExact(logical_id) FROM {DATABASE}.events_full")
- )
- point_sql = f"""
- SELECT logical_id, event_ts, event_type, trace_id, span_id
- FROM {DATABASE}.events_full
- WHERE project_id = 'project-a'
- AND start_time >= '2026-01-01 00:00:00.000'
- AND trace_id = 'project-a-trace-0000'
- AND span_id = 'project-a-span-00000000'
- ORDER BY event_ts DESC LIMIT 1
- """
- list_sql = f"""
- SELECT logical_id, event_ts, event_type, trace_id, span_id
- FROM {DATABASE}.events_core
- WHERE project_id = 'project-a'
- AND start_time >= '2026-01-01 00:00:00.000'
- ORDER BY start_time DESC LIMIT 100
- """
- facet_sql = f"""
- SELECT arrayJoin(tags) AS tag, count() AS rows
- FROM {DATABASE}.events_core
- WHERE project_id = 'project-a'
- GROUP BY tag ORDER BY rows DESC LIMIT 20
- """
- dashboard_sql = f"""
- SELECT toStartOfInterval(start_time, INTERVAL 10 SECOND) AS bucket,
- quantile(0.95)(usage_input) AS p95_input,
- sum(total_cost) AS total_cost
- FROM {DATABASE}.events_core
- WHERE project_id = 'project-a'
- AND start_time >= '2026-01-01 00:00:00.000'
- AND start_time < '2026-01-01 00:02:00.000'
- GROUP BY bucket ORDER BY bucket
- WITH FILL FROM toDateTime('2026-01-01 00:00:00')
- TO toDateTime('2026-01-01 00:02:00')
- STEP INTERVAL 10 SECOND
- """
- fts_sql = f"""
- SELECT logical_id
- FROM {DATABASE}.events_full
- WHERE project_id = 'project-a'
- AND hasAllTokens(lower(input),
- arraySlice(arrayDistinct(tokens(lower('needle-token'))), 1, 64))
- AND positionCaseInsensitiveUTF8(input, 'needle-token') > 0
- LIMIT 100
- SETTINGS enable_full_text_index = 1
- """
- score_sql = f"""
- SELECT name, count() AS rows
- FROM {DATABASE}.scores
- WHERE project_id = 'project-a'
- GROUP BY name ORDER BY rows DESC
- """
- query_metrics = {
- "point": measure_query(client, point_sql),
- "list": measure_query(client, list_sql),
- "facet": measure_query(client, facet_sql),
- "dashboard": measure_query(client, dashboard_sql),
- "full_text": measure_query(client, fts_sql),
- "scores": measure_query(client, score_sql),
- }
- cross_project_point_rows = int(
- client.scalar(
- f"SELECT count() FROM {DATABASE}.events_full "
- "WHERE project_id = 'project-a' AND logical_id = 'project-b-event-00000500'"
- )
- )
- export_query = (
- f"SELECT * FROM {DATABASE}.events_full "
- "WHERE project_id = 'project-a' ORDER BY start_time LIMIT 10000"
- )
- exports = {
- "json_each_row": client.stream(export_query, "JSONEachRow"),
- "parquet": client.stream(export_query, "Parquet"),
- }
- load_phase = run_load_phase(client, args.fixture, args.load_seconds)
- mutations = run_mutations(client)
- memory_resident = client.scalar(
- "SELECT value FROM system.asynchronous_metrics WHERE metric = 'MemoryResident'"
- )
-
- report = {
- "probe": "langfuse-clickhouse-baseline",
- "clickhouse_version": version,
- "database": DATABASE,
- "fixture": str(args.fixture),
- "insert": insert_metrics,
- "populate_ms": populate_ms,
- "counts": counts,
- "event_physical_rows": counts["events_full"],
- "event_distinct_logical_ids": event_distinct_ids,
- "duplicate_versions_present": counts["events_full"] > event_distinct_ids,
- "cross_project_point_rows": cross_project_point_rows,
- "query_latency": query_metrics,
- "exports": exports,
- "load_phase": load_phase,
- "mutations": mutations,
- "memory_resident_bytes": int(memory_resident),
- }
- args.output.parent.mkdir(parents=True, exist_ok=True)
- args.output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
- print(json.dumps(report, indent=2, sort_keys=True))
-
-
-if __name__ == "__main__":
- main()
diff --git a/benchmark/langfuse-pchronicle-review/recorded-results-2026-08-11.json b/benchmark/langfuse-pchronicle-review/recorded-results-2026-08-11.json
deleted file mode 100644
index 82fe3602c..000000000
--- a/benchmark/langfuse-pchronicle-review/recorded-results-2026-08-11.json
+++ /dev/null
@@ -1,146 +0,0 @@
-{
- "status": "experimental-local-review",
- "recorded_at": "2026-08-11",
- "verdict": "NO-GO_DIRECT_REPLACEMENT",
- "baselines": {
- "persisting": "94531cf903e5abc336de347588fb1858e9d52b6a plus local uncommitted catalog/query-engine work",
- "langfuse": "d18a59ad663ffc7c04afc61354186c141b3ec0f3",
- "clickhouse": "25.12.11.4"
- },
- "environment": {
- "kernel": "Darwin 25.5.0 arm64",
- "macos": "26.5",
- "cpu": "Apple M4",
- "memory_bytes": 25769803776,
- "rustc": "1.97.1 (8bab26f4f 2026-07-14)",
- "python": "3.14.5",
- "docker_server": "29.6.2"
- },
- "fixture": {
- "sha256": "1b2ef8ed2f51e6d85a9907bbbebbdafbc4c961d8c4368b2a20b737f94cfc4c90",
- "bytes": 238470862,
- "rows": 113000,
- "events": 100000,
- "event_distinct_logical_ids": 99900,
- "scores": 10000,
- "dataset_run_items": 2000,
- "blob_log_rows": 1000,
- "projects": 2,
- "traces": 200,
- "pchronicle_sources": 210
- },
- "pchronicle": {
- "append": {
- "acknowledged_rows": 113000,
- "elapsed_ms": 12053.854958,
- "rows_per_second": 9374.594301468946,
- "zero_acknowledged_loss": true
- },
- "load_phase": {
- "rows": 30,
- "effective_rows_per_second": 10.263871301317753,
- "ack_p95_ms": 80.864458,
- "visibility_p95_ms": 4.1275,
- "concurrent_query_p95_ms": 350.409292,
- "visible_rows": 30,
- "zero_acknowledged_loss": true,
- "pinned_count_after_append": 113000,
- "refreshed_count": 113030
- },
- "query_p95_ms": {
- "point": 395.870541,
- "list": 564.816625,
- "simple_model_aggregate": 806.700875
- },
- "catalog": {
- "cold_start_ms": 2104.038375,
- "rss_after_bytes": 1117782016,
- "rss_before_bytes": 308281344
- },
- "maintenance_restart": {
- "fragments_removed": 3,
- "maintenance_ms": 38.360458,
- "restart_append_and_catalog_ms": 4160.961459,
- "restart_row_visible": true
- },
- "json_export": {
- "rows": 50000,
- "bytes": 96902959,
- "first_byte_ms": 429.086209,
- "elapsed_ms": 585.346084
- },
- "semantic_results": {
- "event_physical_rows": 100000,
- "event_distinct_logical_ids": 99900,
- "duplicate_versions_remain_physical": true,
- "cross_project_point_rows": 0,
- "update_flags": "unsupported",
- "delete_trace": "unsupported",
- "delete_project": "unsupported",
- "delete_retention": "unsupported",
- "parquet_export": "unsupported"
- }
- },
- "clickhouse": {
- "append": {
- "acknowledged_rows": 113000,
- "elapsed_ms": 1538.472415995784,
- "rows_per_second": 73449.48068299306
- },
- "load_phase": {
- "rows": 30,
- "effective_rows_per_second": 9.857884896590003,
- "ack_p95_ms": 107.69920799066313,
- "visibility_p95_ms": 6.510250008432195,
- "concurrent_query_p95_ms": 13.96033400669694,
- "visible_rows": 30,
- "zero_acknowledged_loss": true
- },
- "query_p95_ms": {
- "point": 4.826582997338846,
- "list": 2.9707499779760838,
- "facet": 3.466208989266306,
- "dashboard": 4.525625001406297,
- "full_text": 3.4713750064838678,
- "scores": 2.282792003825307
- },
- "exports": {
- "json_first_byte_ms": 68.76837500021793,
- "parquet_first_byte_ms": 28.269458998693153
- },
- "mutations_ms": {
- "update_flags": 25.31504098442383,
- "delete_trace": 8.981124992715195,
- "delete_project": 5.9625419962685555,
- "delete_retention": 7.64183298451826
- },
- "semantic_results": {
- "event_physical_rows": 99900,
- "event_distinct_logical_ids": 99900,
- "duplicate_versions_present": false,
- "cross_project_point_rows": 0
- },
- "memory_resident_bytes": 1422721024
- },
- "relative_results": {
- "clickhouse_append_throughput_over_pchronicle": 7.83,
- "pchronicle_point_p95_over_clickhouse": 82.02,
- "pchronicle_list_p95_over_clickhouse": 190.13,
- "pchronicle_simple_aggregate_over_clickhouse_tag_facet_not_semantically_identical": 232.73,
- "pchronicle_concurrent_query_p95_over_clickhouse": 25.10
- },
- "same_fixture_repeat_observations": {
- "pchronicle_append_rows_per_second": [9374.594301468946, 9594.449551771359],
- "pchronicle_point_p95_ms": [395.870541, 527.936125],
- "pchronicle_list_p95_ms": [564.816625, 666.709167],
- "clickhouse_append_rows_per_second": [73449.48068299306, 73661.05953160369],
- "clickhouse_point_p95_ms": [4.183750017546117, 4.826582997338846],
- "clickhouse_list_p95_ms": [2.9707499779760838, 3.9958749839570373]
- },
- "tests": {
- "langfuse_backend_faults": "2 passed; 0 failed",
- "production_scale": "5 passed; 0 failed; 1 ignored",
- "query_engine": "14 passed; 0 failed",
- "store_catalog_filter": "6 passed; 0 failed; 182 filtered"
- }
-}
diff --git a/benchmark/langfuse-pchronicle-review/recorded-results-2026-08-12.json b/benchmark/langfuse-pchronicle-review/recorded-results-2026-08-12.json
deleted file mode 100644
index abd9fa7ab..000000000
--- a/benchmark/langfuse-pchronicle-review/recorded-results-2026-08-12.json
+++ /dev/null
@@ -1,133 +0,0 @@
-{
- "status": "experimental-local-retest",
- "recorded_at": "2026-08-12",
- "scope": "pChronicle lazy catalog and source-pruning retest",
- "baseline": {
- "persisting_head": "94531cf903e5abc336de347588fb1858e9d52b6a",
- "state": "head plus local uncommitted pChronicle catalog/query changes",
- "previous_results": "recorded-results-2026-08-11.json",
- "clickhouse_version": "25.12.11.4"
- },
- "source_sha256": {
- "catalog.rs": "f5066a96a0c8a8eb062b50b205b4f81a2272cf98385f17708fe496cea787872a",
- "raw_event_datafusion.rs": "e4f2a19b1e62ffaa9e7538430d7b033d72bfca4d5ef0ccb816a8fe59f9dbba78",
- "raw_event_lance.rs": "42089b5d10c966360f9a9beb72db74222a7d3fec499f579ccdd732916f544ce2",
- "query_engine.rs": "40c95f0bc10de90cd04f97fc81cd5a0410321b56ad001df4c487b332042c26ea",
- "storyline_datafusion.rs": "3f56eacf94735364b2b190313973510f6b0bcf8b04726c65054097de8f40e71f",
- "langfuse_backend_feasibility.rs": "8c822ab13cdf7c949e3b93c0eab6b1c7335a560b3e24720165e7eaa4ad12f1b8"
- },
- "fixture": {
- "sha256": "1b2ef8ed2f51e6d85a9907bbbebbdafbc4c961d8c4368b2a20b737f94cfc4c90",
- "bytes": 238470862,
- "rows": 113000,
- "events": 100000,
- "event_distinct_logical_ids": 99900,
- "scores": 10000,
- "dataset_run_items": 2000,
- "blob_log_rows": 1000,
- "projects": 2,
- "traces": 200,
- "pchronicle_sources": 210
- },
- "pchronicle": {
- "append": {
- "acknowledged_rows": 113000,
- "elapsed_ms": 12302.462166,
- "rows_per_second": 9185.15322179126,
- "zero_acknowledged_loss": true
- },
- "catalog": {
- "cold_start_ms": 35.041416999999996,
- "rss_before_bytes": 308871168,
- "rss_after_catalog_bytes": 316669952,
- "rss_after_unpruned_queries_bytes": 624885760,
- "sources": 210
- },
- "warm_query_p95_ms": {
- "point": 258.13295899999997,
- "list": 282.798583,
- "simple_model_aggregate": 296.706625
- },
- "fresh_global": {
- "point_catalog_setup_ms": 10.836791999999999,
- "point_p95_ms": 302.93704199999996,
- "list_catalog_setup_ms": 12.47275,
- "list_p95_ms": 326.682458
- },
- "catalog_pruning": {
- "direct_run_open_ms": 0.5066660000000001,
- "direct_run_point_p95_ms": 2.1355,
- "exact_file_catalog_setup_ms": 12.337958,
- "exact_file_point_p95_ms": 3.0130410000000003,
- "project_file_catalog_setup_ms": 12.02425,
- "project_file_list_p95_ms": 167.17120799999998
- },
- "load_phase": {
- "rows": 30,
- "effective_rows_per_second": 9.582487175227183,
- "ack_p95_ms": 27.738083,
- "visibility_p95_ms": 5.46075,
- "concurrent_query_p95_ms": 243.45462500000002,
- "visible_rows": 30,
- "zero_acknowledged_loss": true,
- "pinned_count_after_append": 113000,
- "refreshed_count": 113030
- },
- "json_export": {
- "rows": 50000,
- "bytes": 96902959,
- "first_byte_ms": 336.693958,
- "elapsed_ms": 503.48304200000007
- },
- "maintenance_restart": {
- "maintenance_ms": 37.512542,
- "fragments_removed": 3,
- "restart_append_and_catalog_ms": 51.761459,
- "restart_visibility_ms": 344.021583,
- "restart_end_to_end_ms": 395.78304199999997,
- "restart_row_visible": true
- },
- "semantics": {
- "event_physical_rows": 100000,
- "event_distinct_logical_ids": 99900,
- "duplicate_versions_remain_physical": true,
- "cross_project_point_rows": 0
- }
- },
- "same_code_repeat_ranges": {
- "catalog_cold_start_ms": [35.041416999999996, 35.360584],
- "append_rows_per_second": [9105.865426522427, 9239.77754263836],
- "warm_point_p95_ms": [258.13295899999997, 262.034875],
- "warm_list_p95_ms": [282.798583, 352.99425],
- "simple_model_aggregate_p95_ms": [292.394042, 299.305875],
- "exact_file_point_p95_ms": [2.978792, 3.0130410000000003],
- "project_file_list_p95_ms": [166.093667, 167.17120799999998]
- },
- "comparison_to_2026_08_11_pct": {
- "catalog_cold_start_reduction": 98.33456378855257,
- "cold_catalog_rss_reduction": 71.66979362101313,
- "old_eager_catalog_vs_new_post_query_rss_reduction": 44.09591932457786,
- "warm_point_latency_reduction": 34.79359228197787,
- "warm_list_latency_reduction": 49.930903149318596,
- "simple_model_aggregate_latency_reduction": 63.219746724583636,
- "concurrent_query_latency_reduction": 30.52278276912816,
- "append_throughput_reduction": 2.020792298691799
- },
- "comparison_to_clickhouse_2026_08_11": {
- "clickhouse_point_p95_ms": 4.826582997338846,
- "exact_file_point_ratio": 0.6242596473864122,
- "clickhouse_list_p95_ms": 2.9707499779760838,
- "project_file_list_ratio": 56.27239223742773,
- "warm_global_point_ratio": 53.481512519793505,
- "warm_global_list_ratio": 95.19433984567944
- },
- "tests": {
- "langfuse_backend_faults": "2 passed; 0 failed",
- "production_scale": "5 passed; 0 failed; 1 ignored",
- "query_engine": "14 passed; 0 failed",
- "catalog": "9 passed; 0 failed",
- "raw_event_lance_and_rows": "21 passed; 0 failed",
- "pchronicle_server_lib": "16 passed; 0 failed"
- },
- "artifact_directory": "/tmp/langfuse-pchronicle-retest-final.n9fgAW"
-}
diff --git a/benchmark/langfuse-pchronicle-review/server-acceleration-results-2026-08-12.json b/benchmark/langfuse-pchronicle-review/server-acceleration-results-2026-08-12.json
deleted file mode 100644
index b311f49d8..000000000
--- a/benchmark/langfuse-pchronicle-review/server-acceleration-results-2026-08-12.json
+++ /dev/null
@@ -1,80 +0,0 @@
-{
- "recorded_at": "2026-08-12",
- "scope": "pChronicle release Server source-routing acceleration",
- "fixture": {
- "storage": "/tmp/langfuse-pchronicle-retest-final.n9fgAW/pchronicle-store",
- "catalog_sources": 211,
- "indexed_event_rows": 113031
- },
- "routing_indexes": {
- "event_partition": {
- "rows": 113031,
- "sources": 211,
- "distinct_fingerprints": 213,
- "project_a_candidate_sources": 106
- },
- "event_identity": {
- "rows": 113031,
- "sources": 211,
- "distinct_fingerprints": 113132
- }
- },
- "cold": {
- "catalog_ms": 37.149,
- "identity_point_ms_repeats": [166.173, 159.452, 134.261],
- "identity_point_median_ms": 159.452,
- "catalog_plus_identity_point_ms": 196.601,
- "first_project_list_ms": 312.941
- },
- "warm_http_p95_ms": {
- "point": 4.724,
- "project_list": 151.047,
- "project_aggregate": 165.406
- },
- "warm_http_p50_ms": {
- "point": 4.149,
- "project_list": 147.244,
- "project_aggregate": 158.632
- },
- "warm_samples": {
- "point": 20,
- "project_list": 20,
- "project_aggregate": 20
- },
- "process_rss_kib": {
- "startup": 13456,
- "after_catalog": 26880,
- "after_partition_index_and_query": 129344,
- "after_partition_then_identity_index_and_queries": 216928,
- "fresh_after_catalog_identity_index_and_query": 104384,
- "note": "RSS includes lazy source resolution, Lance/DataFusion buffers and allocator retention; it is not the routing-index heap size"
- },
- "comparison": {
- "previous_pchronicle_p95_ms": {
- "global_point": 258.132959,
- "global_list": 282.798583,
- "global_aggregate": 296.706625
- },
- "improvement_vs_previous_pchronicle_pct": {
- "warm_point": 98.170,
- "warm_project_list": 46.588,
- "warm_project_aggregate": 44.253
- },
- "clickhouse_2026_08_11_p95_ms": {
- "point": 4.826583,
- "list": 2.970750,
- "aggregate": 3.466209
- },
- "pchronicle_to_clickhouse_ratio": {
- "warm_point": 0.978746,
- "warm_project_list": 50.844737,
- "warm_project_aggregate": 47.719569
- }
- },
- "methodology": {
- "build": "cargo build --release -p persisting-cli",
- "server": "target/release/persisting chronicle serve ... --listen 127.0.0.1:19877",
- "client": "curl over loopback HTTP/1.1",
- "caveat": "pChronicle measurements include Axum and loopback HTTP; reused ClickHouse baseline was not rerun concurrently"
- }
-}
diff --git a/benchmark/pchronicle/README.md b/benchmark/pchronicle/README.md
index 52d585574..590c29f3a 100644
--- a/benchmark/pchronicle/README.md
+++ b/benchmark/pchronicle/README.md
@@ -8,10 +8,7 @@
The guarded paths are:
- ATIF parse and round-trip conversion;
-- canonical events → Storyline projection;
- Storyline split into and reconstruction from the three relational tables;
-- canonical Lance event append;
-- full and incremental Storyline projection, including verification;
- Storyline point reads, replacement, indexed SQL, and group-by queries;
- projected JSON streaming latency, throughput, allocations, and peak RSS.
@@ -58,9 +55,9 @@ array. Every leaf has one stable JSONPath address:
{
"measurements": {
"system": {
- "projection_pipeline": {
- "projection_incremental": {
- "sync_ms": {
+ "lance_vs_json": {
+ "lifecycle": {
+ "cold_query_ms": {
"value": 42.1,
"unit": "ms",
"direction": "lower",
@@ -79,7 +76,7 @@ escaping. Read a value with:
```bash
python3 benchmark/pchronicle/bench.py jsonpath-get \
--document target/pchronicle-benchmark/current/raw-report.json \
- --path '$["measurements"]["system"]["projection_pipeline"]["projection_incremental"]["sync_ms"]'
+ --path '$["measurements"]["system"]["lance_vs_json"]["lifecycle"]["cold_query_ms"]'
```
Future benchmark producers can insert a new leaf with `jsonpath-set`. Duplicate
@@ -115,5 +112,5 @@ unbounded repository history.
## Links
-- [pChronicle design](../../docs/src/pchronicle/design/index.md)
+- [pChronicle design](../../docs/src/en/pchronicle/design/index.md)
- [`persisting-pchronicle`](../../crates/persisting-pchronicle/README.md)
diff --git a/benchmark/pchronicle/bench.py b/benchmark/pchronicle/bench.py
index 4b636f765..fcd65068f 100644
--- a/benchmark/pchronicle/bench.py
+++ b/benchmark/pchronicle/bench.py
@@ -22,7 +22,6 @@
import sys
from typing import Any
-
SCHEMA = "pchronicle-benchmark"
README_START = ""
README_END = ""
@@ -82,9 +81,10 @@ def command_output(command: list[str], cwd: pathlib.Path) -> str:
def environment(repo: pathlib.Path, suite: str) -> dict[str, Any]:
- dirty = subprocess.run(
- ["git", "diff", "--quiet", "--ignore-submodules", "--"], cwd=repo
- ).returncode != 0
+ dirty = (
+ subprocess.run(["git", "diff", "--quiet", "--ignore-submodules", "--"], cwd=repo).returncode
+ != 0
+ )
return {
"suite": suite,
"git_commit": command_output(["git", "rev-parse", "HEAD"], repo),
@@ -149,9 +149,7 @@ def jsonpath_segments(path: str) -> list[str]:
while position < len(path):
matched = JSONPATH_TOKEN_RE.match(path, position)
if matched is None:
- raise ValueError(
- "benchmark JSONPath supports only bracketed name selectors: " + path
- )
+ raise ValueError("benchmark JSONPath supports only bracketed name selectors: " + path)
key = json.loads(matched.group("key"))
if not isinstance(key, str) or not key:
raise ValueError(f"JSONPath name selector must be a non-empty string: {path}")
@@ -169,9 +167,7 @@ def jsonpath_get(document: dict[str, Any], path: str) -> Any:
return current
-def jsonpath_set(
- document: dict[str, Any], path: str, value: Any, *, replace: bool = False
-) -> None:
+def jsonpath_set(document: dict[str, Any], path: str, value: Any, *, replace: bool = False) -> None:
segments = jsonpath_segments(path)
if not segments:
raise ValueError("the benchmark JSONPath root cannot be replaced")
@@ -190,9 +186,7 @@ def jsonpath_set(
def store_metrics(document: dict[str, Any], metrics: list[dict[str, Any]]) -> None:
for item in metrics:
path = metric_jsonpath(item["case"], item["metric"])
- measurement = {
- key: value for key, value in item.items() if key not in {"case", "metric"}
- }
+ measurement = {key: value for key, value in item.items() if key not in {"case", "metric"}}
jsonpath_set(document, path, measurement)
@@ -249,9 +243,7 @@ def run_criterion(
{
"CARGO_TARGET_DIR": str(target_dir),
"PCHRONICLE_CRITERION_SAMPLES": str(config["criterion_samples"]),
- "PCHRONICLE_CRITERION_MEASUREMENT_MS": str(
- config["criterion_measurement_ms"]
- ),
+ "PCHRONICLE_CRITERION_MEASUREMENT_MS": str(config["criterion_measurement_ms"]),
"PCHRONICLE_CRITERION_WARMUP_MS": str(config["criterion_warmup_ms"]),
}
)
@@ -290,12 +282,8 @@ def run_criterion(
"ns",
"lower",
"criterion",
- float(confidence["lower_bound"])
- if "lower_bound" in confidence
- else None,
- float(confidence["upper_bound"])
- if "upper_bound" in confidence
- else None,
+ float(confidence["lower_bound"]) if "lower_bound" in confidence else None,
+ float(confidence["upper_bound"]) if "upper_bound" in confidence else None,
)
)
if not metrics:
@@ -488,18 +476,6 @@ def run_suite(args: argparse.Namespace) -> None:
config["hyperfine_runs"],
)
)
- if bench_target_exists(repo, "projection_pipeline"):
- projection = build_bench_executable(repo, target_dir, "projection_pipeline", output)
- metrics.extend(
- run_system_scenario(
- repo,
- output,
- projection,
- "projection_pipeline",
- common,
- config["hyperfine_runs"],
- )
- )
for shape in ("ndjson", "array"):
metrics.extend(
run_system_scenario(
@@ -762,14 +738,12 @@ def html_report(
th:first-child,td:first-child{{text-align:left}} th{{background:#f5f7fa}}
pChronicle benchmark
{verdict}{len(regressions)} regressions · {len(improvements)} improvements
-{''.join(sections)}Complete Markdown report
{html.escape(markdown)}