README.md

@@ -65,7 +65,7 @@ machine (no parallel work). Every task writes a completion marker
 | 06 Convert to PostgreSQL | `convert_postgresql.py` | live PostgreSQL DB (+ dump when pg_dump is available) | implemented |
 | 07 Convert to RDF | `convert_rdf.py` | `out/rdf/dataset.nt.gz`, `dataset.ttl`, oxigraph store | implemented |
 | 08 Benchmark queries | `make_queries.py` | `out/bench/queries/`, `out/bench/expected/` | implemented |
-| 09 Benchmark execution | `bench_runner.py` | `out/bench/results_raw.csv`, `results_median.csv` | planned |
+| 09 Benchmark execution | `bench_runner.py` | `out/bench/results_raw.csv`, `results_median.csv` | implemented |
 | 10 Extrapolation | `extrapolate.py` | `out/bench/extrapolation.csv`, `hardware_sizing.csv` | planned |
 | 11 Reporting | `report.py` | `out/report/REPORT.md`, charts, tables | planned |
 
@@ -156,6 +156,7 @@ convert_sqlite.py     task 05 entry script
 convert_postgresql.py task 06 entry script (TRIBO_PG_DSN, TRIBO_PROM_URL)
 convert_rdf.py        task 07 entry script
 make_queries.py       task 08 entry script (canonical results need TRIBO_PG_DSN)
+bench_runner.py       task 09 entry script (TRIBO_PG_DSN, TRIBO_PROM_URL)
 requirements.txt      closed dependency list (docs/rules/code-python-style.md)
 ```
 

bench_runner.py

@@ -0,0 +1,527 @@
+"""Task 09 - Benchmark execution and resource measurement.
+
+Runs the 35-cell matrix (Q1-Q7 x 5 formats) in randomized order. Every
+cell: 1 warm-up run (recorded as run 0, excluded from statistics) + 3
+measured runs, each in its OWN subprocess sampled with psutil every 50 ms
+for peak RSS / CPU time / read bytes. Result rows of every run are
+validated against out/bench/expected/ (tolerance 1e-9) and recorded as
+checksum_ok. Writes results_raw.csv, results_median.csv, PG EXPLAIN plans,
+PG host/server metrics, the archival storage sizes, an environment record,
+and the 09.ok marker. Spec: docs/specs/09_benchmark_execution.md; protocol:
+docs/rules/bench-methodology.md.
+
+Environment adaptations (bench-methodology section 6):
+- the cold-cache matrix pass is SKIPPED (Windows host, no page-cache drop);
+- PostgreSQL runs on a remote host: no backend-PID psutil sampling;
+  instead best-effort Prometheus host CPU/RAM windows per PG cell and per
+  session, plus best-effort pg_stat_statements per-query session deltas;
+- on Windows the 50 ms sample reads the OS-maintained peak working set
+  (peak_wset), so sub-interval children still report a true peak RSS.
+
+After a run times out, the cell's remaining runs are SKIPPED (never retry
+a timeout - bench-methodology section 3); only executed runs appear in the
+raw results and the cell reports TIMEOUT.
+"""
+
+from __future__ import annotations
+
+import csv
+import gzip
+import logging
+import random
+import re
+import shutil
+import statistics
+import subprocess
+import sys
+import tarfile
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import psutil
+import psycopg
+
+from common.monitoring import get_prom_url, host_window_metrics
+from common.pg import get_dsn
+from common.pipeline import (
+	REPO_ROOT,
+	check_dependencies,
+	load_lab_config,
+	parse_task_args,
+	remove_stale_marker,
+	write_marker,
+)
+from common.results import compare, parse_rows
+from common.storage_sizes import update_storage_sizes
+
+logger = logging.getLogger(__name__)
+
+TASK_ID = "09"
+DEPENDS_ON = ["08"]
+
+CELL_TIMEOUT_S = 30 * 60   # hard per-run timeout, spec 09
+SAMPLE_INTERVAL_S = 0.05   # psutil sampling cadence, spec 09
+MEASURED_RUNS = 3          # 1 warm-up + 3 measured per cell, spec 09
+
+FORMATS = ("csv", "json", "sqlite", "pg", "rdf")
+QUERY_NUMBERS = tuple(range(1, 8))
+
+RAW_HEADER = [
+	"format", "query", "run", "wall_s", "peak_rss_mb", "cpu_s",
+	"cpu_util_pct", "read_mb", "rows", "checksum_ok", "status", "exec_order",
+]
+MEDIAN_HEADER = [
+	"format", "query", "wall_s_median", "wall_s_min", "wall_s_max",
+	"peak_rss_mb_median", "cpu_s_median", "cpu_util_pct_median",
+	"read_mb_median", "rows", "checksum_ok", "status",
+]
+
+# storage_sizes.csv completeness gate: every (format, variant) that must
+# exist after consolidation (working + archival footprints, spec 09)
+REQUIRED_SIZE_ROWS = {
+	("csv", "tree"), ("csv", "targz"),
+	("json", "full"), ("json", "hybrid"), ("json", "targz"),
+	("sqlite", "db"), ("sqlite", "db_gz"),
+	("pg", "live"), ("pg", "dump"),
+	("rdf", "nt_gz"), ("rdf", "ttl"), ("rdf", "store"),
+}
+
+
+class CellTimeoutError(Exception):
+	pass
+
+
+@dataclass
+class RunResult:
+	status: str
+	wall_s: float = 0.0
+	peak_rss_mb: float = 0.0
+	cpu_s: float = 0.0
+	cpu_util_pct: float = 0.0
+	read_mb: float = 0.0
+	rows: int | None = None
+	checksum_ok: bool | None = None
+
+	def raw_row(self, fmt: str, qn: int, run: int, order: int) -> list:
+		ok = "" if self.checksum_ok is None else str(self.checksum_ok).lower()
+		return [
+			fmt, f"q{qn}", run, round(self.wall_s, 4), round(self.peak_rss_mb, 1),
+			round(self.cpu_s, 3), round(self.cpu_util_pct, 1), round(self.read_mb, 2),
+			"" if self.rows is None else self.rows, ok, self.status, order,
+		]
+
+
+def _kill_tree(proc: subprocess.Popen) -> None:
+	"""Kill the child and every descendant (the stub's real interpreter)."""
+	try:
+		for child in psutil.Process(proc.pid).children(recursive=True):
+			try:
+				child.kill()
+			except psutil.NoSuchProcess:
+				logger.debug("kill_tree: pid %d already exited", child.pid)
+	except psutil.Error:
+		logger.warning("kill_tree: descendants of pid %d not enumerable", proc.pid, exc_info=True)
+	proc.kill()
+	proc.wait()
+
+
+def _run_child(cmd: list[str], stdout_path: Path, stderr_path: Path) -> tuple[int, RunResult]:
+	"""Spawn one query subprocess and sample its process TREE until exit.
+
+	The venv python.exe on Windows is a launcher stub: the real interpreter
+	runs as its child, so sampling only the spawned PID measures the stub
+	(~4 MB flat, ~0 CPU). Every sample therefore walks the descendant tree
+	and keeps per-PID last-known CPU / read bytes and the OS peak working
+	set; totals are sums over all PIDs ever seen. The flat stub adds ~4 MB
+	to the peak-RSS sum - negligible against real query footprints.
+	"""
+	cores = psutil.cpu_count(logical=True)
+	seen: dict[int, list] = {}  # pid -> [cpu_s, read_bytes, peak_rss_bytes]
+	timed_out = False
+	start = time.perf_counter()
+	with open(stdout_path, "wb") as out_fh, open(stderr_path, "wb") as err_fh:
+		proc = subprocess.Popen(cmd, stdout=out_fh, stderr=err_fh, cwd=REPO_ROOT)
+		try:
+			try:
+				root = psutil.Process(proc.pid)
+			except psutil.Error:
+				root = None  # child already gone: exited faster than one sample
+			while root is not None and proc.poll() is None:
+				try:
+					procs = [root, *root.children(recursive=True)]
+				except psutil.Error:
+					break
+				for p in procs:
+					try:
+						mem = p.memory_info()
+						cpu = p.cpu_times()
+						io = p.io_counters()
+					except psutil.Error:
+						continue  # that process exited between listing and sampling
+					entry = seen.setdefault(p.pid, [0.0, 0, 0])
+					entry[0] = cpu.user + cpu.system
+					entry[1] = io.read_bytes
+					entry[2] = max(entry[2], getattr(mem, "peak_wset", 0) or mem.rss)
+				if time.perf_counter() - start > CELL_TIMEOUT_S:
+					timed_out = True
+					break
+				time.sleep(SAMPLE_INTERVAL_S)
+			if timed_out:
+				_kill_tree(proc)
+				raise CellTimeoutError(f"exceeded {CELL_TIMEOUT_S} s")
+			returncode = proc.wait()
+		finally:
+			if proc.poll() is None:
+				_kill_tree(proc)
+	wall = time.perf_counter() - start
+	cpu_s = sum(entry[0] for entry in seen.values())
+	result = RunResult(
+		status="", wall_s=wall,
+		peak_rss_mb=sum(entry[2] for entry in seen.values()) / 1048576,
+		cpu_s=cpu_s,
+		cpu_util_pct=100.0 * cpu_s / wall / cores if wall > 0 else 0.0,
+		read_mb=sum(entry[1] for entry in seen.values()) / 1048576,
+	)
+	return returncode, result
+
+
+def _run_cell_once(fmt: str, qn: int, cmd: list[str], tmp_dir: Path, expected: list[tuple]) -> RunResult:
+	"""One benchmark run with pattern C isolation (code-error-handling)."""
+	stdout_path = tmp_dir / "stdout.txt"
+	stderr_path = tmp_dir / "stderr.txt"
+	try:
+		returncode, result = _run_child(cmd, stdout_path, stderr_path)
+	except CellTimeoutError:
+		logger.error("cell %s/q%d timed out after %d s", fmt, qn, CELL_TIMEOUT_S)
+		return RunResult(status="TIMEOUT", wall_s=CELL_TIMEOUT_S)
+	except Exception:
+		logger.error("cell %s/q%d failed to execute", fmt, qn, exc_info=True)
+		return RunResult(status="ERROR")
+	if returncode != 0:
+		tail = stderr_path.read_text(encoding="utf-8", errors="replace")[-2000:]
+		logger.error("cell %s/q%d exited %d; stderr tail: %s", fmt, qn, returncode, tail)
+		result.status = "ERROR"
+		return result
+	actual = parse_rows(stdout_path.read_text(encoding="ascii"))
+	diff = compare(expected, actual)
+	if diff:
+		logger.error("cell %s/q%d checksum mismatch: %s", fmt, qn, diff)
+	result.status = "OK"
+	result.rows = len(actual)
+	result.checksum_ok = diff is None
+	return result
+
+
+@dataclass
+class CellOutcome:
+	fmt: str
+	qn: int
+	measured: list[RunResult] = field(default_factory=list)
+	timed_out: bool = False  # any run incl. warm-up: remaining runs skipped
+	window: tuple[float, float] | None = None
+
+	def status(self) -> str:
+		if self.timed_out or any(r.status == "TIMEOUT" for r in self.measured):
+			return "TIMEOUT"
+		if len(self.measured) < MEASURED_RUNS or any(r.status == "ERROR" for r in self.measured):
+			return "ERROR"
+		if not all(r.checksum_ok for r in self.measured):
+			return "INVALID"
+		return "OK"
+
+
+def _median_row(outcome: CellOutcome) -> list:
+	status = outcome.status()
+	fmt, qn, measured = outcome.fmt, outcome.qn, outcome.measured
+	if status != "OK":
+		ok = "false" if status == "INVALID" else ""
+		return [fmt, f"q{qn}", "", "", "", "", "", "", "", "", ok, status]
+	walls = [r.wall_s for r in measured]
+	return [
+		fmt, f"q{qn}",
+		round(statistics.median(walls), 4), round(min(walls), 4), round(max(walls), 4),
+		round(statistics.median(r.peak_rss_mb for r in measured), 1),
+		round(statistics.median(r.cpu_s for r in measured), 3),
+		round(statistics.median(r.cpu_util_pct for r in measured), 1),
+		round(statistics.median(r.read_mb for r in measured), 2),
+		measured[0].rows, "true", status,
+	]
+
+
+def _capture_explain(dsn: str, queries_dir: Path, explain_dir: Path) -> None:
+	"""EXPLAIN (ANALYZE, BUFFERS) per PG query; verify Q5 avoids track_summary."""
+	explain_dir.mkdir(parents=True, exist_ok=True)
+	with psycopg.connect(dsn) as conn:
+		for qn in QUERY_NUMBERS:
+			sql = (queries_dir / "pg" / f"q{qn}.sql").read_text(encoding="ascii")
+			plan = "\n".join(r[0] for r in conn.execute("EXPLAIN (ANALYZE, BUFFERS) " + sql).fetchall())
+			(explain_dir / f"q{qn}.txt").write_text(plan + "\n", encoding="utf-8", newline="\n")
+			if qn == 5 and "track_summary" in plan:
+				raise RuntimeError("q5 plan touches track_summary - the forced full scan is violated (bench-methodology section 4)")
+	logger.info("EXPLAIN plans captured to %s", explain_dir)
+
+
+PG_STAT_QUERY = """
+SELECT query, calls, total_exec_time, rows,
+       shared_blks_hit, shared_blks_read, temp_blks_read, temp_blks_written
+FROM pg_stat_statements
+"""
+PG_STAT_COLUMNS = [
+	"calls", "total_exec_time_ms", "rows",
+	"shared_blks_hit", "shared_blks_read", "temp_blks_read", "temp_blks_written",
+]
+_QTAG = re.compile(r"^-- Q(\d):")
+
+
+def _pg_statements_snapshot(dsn: str) -> dict[int, list[float]] | None:
+	"""Sum pg_stat_statements counters per benchmark query (matched by the
+	leading '-- Q<N>:' comment). Best-effort: None when the extension is
+	unavailable - server-side stats are auxiliary and never fail the task.
+	"""
+	try:
+		sums: dict[int, list[float]] = {}
+		with psycopg.connect(dsn) as conn:
+			for query_text, *values in conn.execute(PG_STAT_QUERY):
+				match = _QTAG.match(query_text)
+				if not match:
+					continue
+				acc = sums.setdefault(int(match.group(1)), [0.0] * len(values))
+				for i, v in enumerate(values):
+					acc[i] += float(v)
+		return sums
+	except Exception:
+		logger.warning("pg_stat_statements snapshot skipped (extension unavailable or not readable)", exc_info=True)
+		return None
+
+
+def _write_pg_server_stats(path: Path, start: dict[int, list[float]], end: dict[int, list[float]]) -> None:
+	with open(path, "w", encoding="ascii", newline="") as fh:
+		writer = csv.writer(fh)
+		writer.writerow(["query"] + PG_STAT_COLUMNS)
+		for qn in QUERY_NUMBERS:
+			before = start.get(qn, [0.0] * len(PG_STAT_COLUMNS))
+			after = end.get(qn)
+			if after is None:
+				logger.warning("pg_stat_statements: q%d missing in the end snapshot (entry evicted?)", qn)
+				writer.writerow([f"q{qn}"] + [""] * len(PG_STAT_COLUMNS))
+				continue
+			delta = [after[i] - before[i] for i in range(len(PG_STAT_COLUMNS))]
+			writer.writerow([f"q{qn}"] + [round(v, 3) for v in delta])
+	logger.info("pg server-side statement deltas written to %s", path)
+
+
+def _tree_bytes(root: Path) -> int:
+	return sum(p.stat().st_size for p in root.rglob("*") if p.is_file())
+
+
+def _consolidate_storage_sizes(out_root: Path, bench_dir: Path) -> None:
+	"""Build archival variants, record all sizes, verify completeness."""
+	archives = bench_dir / "archives"
+	archives.mkdir(parents=True, exist_ok=True)
+
+	csv_tgz = archives / "csv_tree.tar.gz"
+	logger.info("archiving out/csv -> %s", csv_tgz)
+	with tarfile.open(csv_tgz, "w:gz") as tar:
+		tar.add(out_root / "csv", arcname="csv")
+	update_storage_sizes(bench_dir, "csv", [
+		("tree", _tree_bytes(out_root / "csv")), ("targz", csv_tgz.stat().st_size),
+	])
+
+	json_tgz = archives / "json.tar.gz"
+	logger.info("archiving out/json -> %s", json_tgz)
+	with tarfile.open(json_tgz, "w:gz") as tar:
+		tar.add(out_root / "json", arcname="json")
+	update_storage_sizes(bench_dir, "json", [("targz", json_tgz.stat().st_size)])
+
+	db_gz = archives / "tribo.db.gz"
+	logger.info("compressing out/sqlite/tribo.db -> %s", db_gz)
+	with open(out_root / "sqlite" / "tribo.db", "rb") as src, gzip.open(db_gz, "wb") as dst:
+		shutil.copyfileobj(src, dst, 1048576)
+	update_storage_sizes(bench_dir, "sqlite", [("db_gz", db_gz.stat().st_size)])
+
+	# pg_dump -Fc is internally gzip-compressed: it IS the pg archival variant
+	update_storage_sizes(bench_dir, "pg", [("dump", (out_root / "pg" / "tribo.dump").stat().st_size)])
+
+	present = set()
+	with open(bench_dir / "storage_sizes.csv", encoding="ascii") as fh:
+		for row in csv.reader(fh):
+			present.add((row[0], row[1]))
+	missing = REQUIRED_SIZE_ROWS - present
+	if missing:
+		raise ValueError(f"storage_sizes.csv incomplete after consolidation: missing {sorted(missing)}")
+	logger.info("storage_sizes.csv consolidated: all %d required rows present", len(REQUIRED_SIZE_ROWS))
+
+
+def _environment_record(path: Path, dsn: str) -> None:
+	import platform
+
+	lines = [
+		f"benchmark_host_os: {platform.platform()}",
+		f"benchmark_host_cpu: {platform.processor()}",
+		f"benchmark_host_cores_logical: {psutil.cpu_count(logical=True)}",
+		f"benchmark_host_cores_physical: {psutil.cpu_count(logical=False)}",
+		f"benchmark_host_ram_gib: {round(psutil.virtual_memory().total / 1073741824, 1)}",
+		f"python: {platform.python_version()}",
+	]
+	try:
+		disks = subprocess.run(
+			["powershell", "-NoProfile", "-Command",
+				"Get-PhysicalDisk | ForEach-Object { $_.FriendlyName + ' / ' + $_.MediaType + ' / ' + [math]::Round($_.Size / 1GB) + ' GB' }"],
+			capture_output=True, text=True, timeout=60, check=True)
+		lines.append("benchmark_host_disks: " + "; ".join(ln.strip() for ln in disks.stdout.splitlines() if ln.strip()))
+	except Exception:
+		logger.warning("disk info skipped (Get-PhysicalDisk query failed - non-Windows host or no permission)", exc_info=True)
+	import sqlite3
+
+	import ijson
+	import pyoxigraph
+
+	lines += [
+		f"sqlite: {sqlite3.sqlite_version}",
+		f"psycopg: {psycopg.__version__}",
+		f"pyoxigraph: {pyoxigraph.__version__}",
+		f"ijson: {ijson.__version__}",
+		f"psutil: {psutil.__version__}",
+	]
+	with psycopg.connect(dsn) as conn:
+		pg_version = conn.execute("SELECT version()").fetchone()[0]
+		lines.append(f"postgresql_server: {pg_version}")
+		lines.append(f"postgresql_host: {conn.info.host} (remote; 4 cores / 3.8 GiB, see .done/06.ok)")
+	lines += [
+		"cold_cache_pass: skipped (Windows host, no page-cache drop - bench-methodology section 6)",
+		"pg_backend_sampling: not possible for a remote server; Prometheus windows + pg_stat_statements deltas instead",
+	]
+	path.write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n")
+	logger.info("environment record written to %s", path)
+
+
+def _run_matrix(bench_dir: Path, expected: dict[int, list[tuple]], seed: int) -> list[CellOutcome]:
+	cells = [(fmt, qn) for fmt in FORMATS for qn in QUERY_NUMBERS]
+	random.Random(f"{seed}-bench-order").shuffle(cells)
+	tmp_dir = bench_dir / "tmp"
+	tmp_dir.mkdir(parents=True, exist_ok=True)
+	outcomes: list[CellOutcome] = []
+	with open(bench_dir / "results_raw.csv", "w", encoding="ascii", newline="") as raw_fh:
+		writer = csv.writer(raw_fh)
+		writer.writerow(RAW_HEADER)
+		for order, (fmt, qn) in enumerate(cells, start=1):
+			cmd = [sys.executable, str(bench_dir / "queries" / fmt / f"q{qn}.py")]
+			outcome = CellOutcome(fmt, qn)
+			window_start = time.time()
+			for run in range(MEASURED_RUNS + 1):  # run 0 = warm-up
+				result = _run_cell_once(fmt, qn, cmd, tmp_dir, expected[qn])
+				writer.writerow(result.raw_row(fmt, qn, run, order))
+				raw_fh.flush()
+				if run > 0:
+					outcome.measured.append(result)
+				logger.info("cell %s/q%d run %d: %s wall=%.3fs rss=%.1fMB", fmt, qn, run, result.status, result.wall_s, result.peak_rss_mb)
+				if result.status == "TIMEOUT":
+					outcome.timed_out = True
+					logger.error("cell %s/q%d: skipping remaining runs after timeout", fmt, qn)
+					break
+			outcome.window = (window_start, time.time())
+			outcomes.append(outcome)
+			status = outcome.status()
+			walls = [r.wall_s for r in outcome.measured if r.status == "OK"]
+			detail = f"median {statistics.median(walls):.3f} s" if status == "OK" else status
+			print(f"[{order}/{len(cells)}] {fmt}/q{qn}: {detail}")
+	shutil.rmtree(tmp_dir)
+	return outcomes
+
+
+def _write_pg_host_metrics(path: Path, prom_url: str, outcomes: list[CellOutcome]) -> None:
+	with open(path, "w", encoding="ascii", newline="") as fh:
+		writer = csv.writer(fh)
+		writer.writerow([
+			"query", "window_s", "host_cpu_avg_pct", "host_cpu_max_pct",
+			"host_ram_used_baseline_mb", "host_ram_used_peak_mb", "host_ram_used_delta_mb",
+		])
+		for outcome in sorted((o for o in outcomes if o.fmt == "pg"), key=lambda o: o.qn):
+			start, end = outcome.window
+			metrics = host_window_metrics(prom_url, start, end)
+			row = [f"q{outcome.qn}", round(end - start, 1)]
+			if metrics is None:
+				row += [""] * 5
+			else:
+				row += [
+					metrics["host_cpu_avg_pct"], metrics["host_cpu_max_pct"],
+					metrics["host_ram_used_baseline_mb"], metrics["host_ram_used_peak_mb"],
+					metrics["host_ram_used_delta_mb"],
+				]
+			writer.writerow(row)
+	logger.info("per-cell PG host metrics written to %s", path)
+
+
+def main() -> int:
+	args = parse_task_args("Task 09: execute the Q1-Q7 x 5 formats benchmark matrix")
+	out_root: Path = args.out_root
+	bench_dir = out_root / "bench"
+	try:
+		check_dependencies(out_root, DEPENDS_ON)
+		remove_stale_marker(out_root, TASK_ID)
+		dump_path = out_root / "pg" / "tribo.dump"
+		if not dump_path.exists():
+			raise FileNotFoundError(f"{dump_path} missing - produce it with pg_dump -Fc before task 09 (storage consolidation needs it)")
+		dsn = get_dsn()
+		expected = {qn: parse_rows((bench_dir / "expected" / f"q{qn}.rows.csv").read_text(encoding="ascii")) for qn in QUERY_NUMBERS}
+		seed = load_lab_config(out_root)["seed"]
+		logger.warning("cold-cache pass skipped (Windows host - bench-methodology section 6)")
+		_environment_record(bench_dir / "environment.txt", dsn)
+
+		prom_url = get_prom_url()
+		stats_start = _pg_statements_snapshot(dsn)
+		session_start = time.time()
+		outcomes = _run_matrix(bench_dir, expected, seed)
+		session_end = time.time()
+
+		with open(bench_dir / "results_median.csv", "w", encoding="ascii", newline="") as fh:
+			writer = csv.writer(fh)
+			writer.writerow(MEDIAN_HEADER)
+			for outcome in sorted(outcomes, key=lambda o: (o.fmt, o.qn)):
+				writer.writerow(_median_row(outcome))
+
+		# snapshot BEFORE the EXPLAIN pass so its executions cannot pollute the deltas
+		if stats_start is not None:
+			stats_end = _pg_statements_snapshot(dsn)
+			if stats_end is not None:
+				_write_pg_server_stats(bench_dir / "pg_server_stats.csv", stats_start, stats_end)
+		_capture_explain(dsn, bench_dir / "queries", bench_dir / "explain")
+		session_host = None
+		if prom_url:
+			_write_pg_host_metrics(bench_dir / "pg_host_metrics.csv", prom_url, outcomes)
+			session_host = host_window_metrics(prom_url, session_start, session_end)
+		else:
+			logger.warning("TRIBO_PROM_URL unset - PG host metrics skipped")
+
+		_consolidate_storage_sizes(out_root, bench_dir)
+
+		by_status: dict[str, int] = {}
+		for outcome in outcomes:
+			by_status[outcome.status()] = by_status.get(outcome.status(), 0) + 1
+		entries = {
+			"cells_total": len(outcomes),
+			"cells_ok": by_status.get("OK", 0),
+			"cells_invalid": by_status.get("INVALID", 0),
+			"cells_timeout": by_status.get("TIMEOUT", 0),
+			"cells_error": by_status.get("ERROR", 0),
+			"session_wall_s": round(session_end - session_start, 1),
+			"cold_cache_pass": "skipped (Windows host)",
+		}
+		if session_host:
+			entries.update({f"session_{k}": v for k, v in session_host.items()})
+		marker_path = write_marker(out_root, TASK_ID, entries)
+	except Exception:
+		logger.critical("task %s failed", TASK_ID, exc_info=True)
+		return 1
+
+	print(f"task 09 ok: {entries['cells_ok']}/{entries['cells_total']} cells OK "
+		f"(invalid={entries['cells_invalid']}, timeout={entries['cells_timeout']}, error={entries['cells_error']}), "
+		f"session {entries['session_wall_s']} s")
+	print(f"results: {bench_dir / 'results_raw.csv'}, {bench_dir / 'results_median.csv'}")
+	print(f"marker: {marker_path}")
+	return 0
+
+
+if __name__ == "__main__":
+	sys.exit(main())

common/storage_sizes.py

@@ -1,8 +1,10 @@
-"""Idempotent per-format updates to out/bench/storage_sizes.csv.
+"""Idempotent per-(format, variant) updates to out/bench/storage_sizes.csv.
 
 Contract (docs/rules/build-pipeline-tasks.md section 4): columns
-format,variant,bytes; a task re-run REPLACES its own format's rows and never
-touches other formats' rows.
+format,variant,bytes; a task re-run REPLACES its own (format, variant) rows
+and never touches other rows. Replacement is per variant, not per format:
+converters 04-07 own the working-footprint variants while task 09 appends
+archival variants of the same formats.
 """
 
 from __future__ import annotations
@@ -25,12 +27,13 @@ def update_storage_sizes(bench_dir: Path, fmt: str, variants: list[tuple[str, in
 	lock = bench_dir / "storage_sizes.lock"
 	fd = _acquire(lock)
 	try:
+		replaced = {(fmt, variant) for variant, _ in variants}
 		kept: list[str] = []
 		if path.exists():
 			lines = path.read_text(encoding="ascii").splitlines()
 			if lines and lines[0] != HEADER:
 				raise ValueError(f"{path}: unexpected header {lines[0]!r}")
-			kept = [ln for ln in lines[1:] if ln and ln.split(",", 1)[0] != fmt]
+			kept = [ln for ln in lines[1:] if ln and tuple(ln.split(",")[:2]) not in replaced]
 		rows = kept + [f"{fmt},{variant},{nbytes}" for variant, nbytes in variants]
 		with open(path, "w", encoding="ascii", newline="\n") as fh:
 			fh.write(HEADER + "\n" + "\n".join(rows) + "\n")

docs/Initial_Prompt.md

@@ -0,0 +1,90 @@
+# Prompt: Generate a Multi-File Specification for a Tribology Data Storage Evaluation Pipeline
+
+You are an AI assistant generating a specification for execution by Anthropic Claude Code
+in a Cursor development environment (local filesystem, on-premise only, no cloud).
+
+## Objective
+Produce 12 Markdown files — `00_PLAN.md` plus task files `01`–`11` — that specify a complete,
+reproducible pipeline: simulate a tribology laboratory, generate ~150 MB of realistic
+measurement data in CSV, convert it into 4 target storage formats, benchmark 7 retrieval
+scenarios across all 5 formats with resource metering, extrapolate results to 600 GB+,
+and produce a decision report with tables, charts, and weighted scoring.
+
+## File structure requirements
+Each task file must contain: Context/Goal, Dependencies, Input, Processing instructions,
+Output (artifacts + completion marker `./out/.done/<task>.ok`).
+`00_PLAN.md` must contain: goals, execution environment, a Mermaid dependency flowchart,
+parallelism notes, execution order, and global conventions (ISO-8601 UTC timestamps,
+SI-explicit column names, fixed random seed 20260711, Python with tabs for indentation).
+
+## Laboratory configuration (task 01)
+- Material: Ti-6Al-4V coupons 10×10×3 mm, Cr adhesion layer, Pt-Au coating sputtered in a
+  composition gradient; film thickness 0.3–1.1 µm.
+- Instruments: RAPID custom high-throughput parallelized 6-probe tribometer (friction);
+  Bruker TI980 TriboIndenter (nanoindentation); Bruker M4 Tornado micro-XRF (composition
+  mapping); Kurt J. Lesker PVD 200 sputter-down (deposition); AFM (roughness); optical
+  profilometry (thickness); SIMTRA (sputter-transport Monte-Carlo simulation).
+- Deposition matrix, 4 batches (B721–B724) with gun tilt (20°:0°, 20°:20°, 20°:20°, 0°:20°),
+  Pt/Au power (150/50, 100/100, 150/50, 50/150 W) and discharge voltages.
+- Hierarchy and volumes: batch(×4) → wafer(×3) → coupon(×49, 7×7 grid) = 588 coupons;
+  480 friction coupons (4 RAPID runs × 5 plates × 6 probes × 4 coupons/square),
+  108 reserve; 2 environments (Lab Air / Dry N2, 2 runs each); 3 replicate tracks per
+  coupon with counterface rotation → 1,440 tracks × 1,000 cycles = 1.44 M cycle rows;
+  friction loops (200 points every 100th cycle) → 2.88 M loop points.
+- Physical models for simulation: exponential COF run-in with steady-state depending on
+  Au% and environment; hardness/modulus linear in Au% with noise; log-normal roughness;
+  Archard wear; Au gradient across wafer position (batch-dependent center: 8/25/10/60 wt%).
+
+## Process flow diagrams (task 02) — Mermaid
+Coupon assembly (clean → sonicate → Cr adhesive → PVD sputter); characterization &
+batch assembly (SIMTRA, profilometry, ×49/wafer, ×3/batch, batches 721–724); testing tree
+(friction µ_normal/µ_dry_nit, nanoindentation, AFM, XRF); tribometer session sequence
+(request → samples → ball holders, 3 counterfaces/holder → Excel test plan → setup →
+run → software stop → save avg files → teardown); execution loop (5 plates × 6 probes ×
+4 coupons × 3 tracks with counterface rotation); data-hierarchy ERD.
+
+## Data generation (task 03)
+Hierarchical CSV file tree mirroring the lab structure (batches, simtra, per-coupon
+xrf_map/profilometry/nanoindentation/afm, per-track track_info/cof_vs_cycle/
+friction_loops/wear), streaming generation, MANIFEST.csv with row/byte/sha256,
+size guard 120–200 MB.
+
+## Conversions (tasks 04–07), each recording sizes into storage_sizes.csv
+- JSON-LD: vocab + QUDT + PROV-O; two variants — FULL (bulk embedded, per-coupon files,
+  compact) and HYBRID (metadata only, sourceFile links to CSV).
+- SQLite: single file, integer surrogate keys, WITHOUT ROWID bulk tables, precomputed
+  track_summary, indexes for the benchmark queries, load PRAGMAs, VACUUM/ANALYZE.
+- PostgreSQL 16: same logical schema, RANGE partitioning of bulk tables, COPY loading,
+  b-tree + BRIN indexes, materialized track_summary, pg_dump archive.
+- RDF: N-Triples.gz + Turtle serializations; queryable embedded store (oxigraph preferred,
+  Fuseki/TDB2 fallback); document triple-modeling decisions; expect 25–45 M triples.
+
+## Benchmarks (tasks 08–09)
+Seven scenarios: Q1 single-track COF curve; Q2 steady-state COF at Au=10±0.5 wt%;
+Q3 hardness vs COF join; Q4 condition filter (Dry N2, 100 mN, cof>0.20); Q5 mean run-in
+per batch aggregating ALL raw cycles (summary tables forbidden); Q6 wear volume vs load
+join; Q7 Stribeck-style grouped aggregation. Implement per format (direct path read /
+ijson streaming / SQL / SQL+EXPLAIN / SPARQL); identical result sets validated by checksum.
+Harness: each cell in a separate subprocess; psutil metering (wall time, peak RSS, CPU
+time/utilization, read bytes); 1 warm-up + 3 measured runs, medians; 30-min timeout;
+outputs results_raw.csv / results_median.csv.
+
+## Extrapolation (task 10)
+Project storage linearly and query time by access-pattern complexity laws
+(O(1)/O(log n)/O(n)/single-threaded vs parallel partitioned scan) from the 150 MB
+calibration point to 600 GB, 1.2 TB, 6 TB; flag >1 h IMPRACTICAL, >24 h FAIL; RAM models
+per format; hardware sizing and cost table driven by an editable hw_prices.yaml
+(June 2026 defaults: DDR5 RDIMM ~$14/GB, enterprise NVMe ~$250/TB, 16-core 1U ~$6k).
+
+## Reporting (task 11)
+Tables (footprint + coefficients vs CSV, measured metrics, projected times with flags,
+hardware costs, scoring matrix); matplotlib charts (log scales): footprint, RAM,
+per-query bars, degradation curves 150 MB→6 TB, stacked weighted score, cost-vs-performance
+scatter. Weighted scoring: search speed ×5 (0–50), RAM economy ×2 (0–20), disk economy
+×1 (0–10), max 80; search score from inverse geometric mean of Q1–Q7 projected at 600 GB.
+Concise English narrative: executive summary and use-case mapping (analysis, reports,
+search/filter, archiving, inter-lab exchange), limitations.
+
+## Style
+Specification prose in English, concise and directive. No implementation code in the
+spec — only precise instructions an autonomous coding agent can execute.

docs/rules/bench-methodology.md

@@ -82,6 +82,15 @@ invalidates everything downstream.
   pass is therefore SKIPPED and marked as such in results and report -
   never approximated by a fake (e.g. file re-copy tricks) without a
   rules update describing the method.
+- **PostgreSQL is remote on this setup** (dedicated DB host on the LAN),
+  so spec 09's backend-PID sampling via `pg_stat_activity` + psutil is
+  impossible - psutil cannot attribute a remote process. Instead the
+  runner records (a) client-side child metrics exactly like every other
+  format (wall times therefore include LAN round-trips - stated in the
+  report), (b) best-effort Prometheus / node_exporter host CPU/RAM
+  windows per PG cell and for the whole session (`TRIBO_PROM_URL`), and
+  (c) best-effort `pg_stat_statements` per-query session deltas
+  (server-side execution time and buffer traffic).
 - Quiesce the machine during measured runs (no parallel pipeline tasks,
   no background loads); the runner runs cells sequentially by design.
 

make_queries.py

@@ -182,6 +182,55 @@ PY_FORMATS = [
 	("rdf", "rdf_queries", '"rdf" / "oxigraph_store"'),
 ]
 
+SQLITE_WRAPPER = '''"""Benchmark query q{n} for the sqlite format (generated by task 08).
+
+Executes the adjacent q{n}.sql against out/sqlite/tribo.db (read-only) and
+prints the canonical result rows to stdout at full float precision; the
+benchmark runner (task 09) captures and validates them against
+out/bench/expected/q{n}.rows.csv.
+"""
+import sqlite3
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+sys.path.insert(0, str(REPO_ROOT))
+
+from common.results import print_rows
+
+if __name__ == "__main__":
+	sql = Path(__file__).with_suffix(".sql").read_text(encoding="ascii")
+	db = REPO_ROOT / "out" / "sqlite" / "tribo.db"
+	conn = sqlite3.connect(f"file:{{db.as_posix()}}?mode=ro", uri=True)
+	print_rows([tuple(r) for r in conn.execute(sql).fetchall()])
+'''
+
+PG_WRAPPER = '''"""Benchmark query q{n} for the pg format (generated by task 08).
+
+Executes the adjacent q{n}.sql on the PostgreSQL instance addressed by
+TRIBO_PG_DSN and prints the canonical result rows to stdout at full float
+precision; the benchmark runner (task 09) captures and validates them
+against out/bench/expected/q{n}.rows.csv.
+"""
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+sys.path.insert(0, str(REPO_ROOT))
+
+import psycopg
+
+from common.pg import get_dsn
+from common.results import print_rows
+
+if __name__ == "__main__":
+	sql = Path(__file__).with_suffix(".sql").read_text(encoding="ascii")
+	with psycopg.connect(get_dsn()) as conn:
+		print_rows([tuple(r) for r in conn.execute(sql).fetchall()])
+'''
+
+SQL_WRAPPERS = {"sqlite": SQLITE_WRAPPER, "pg": PG_WRAPPER}
+
 
 def write_query_files(bench_dir: Path) -> int:
 	queries_dir = bench_dir / "queries"
@@ -198,7 +247,8 @@ def write_query_files(bench_dir: Path) -> int:
 		fmt_dir.mkdir(parents=True, exist_ok=True)
 		for n, text in sql.items():
 			(fmt_dir / f"q{n}.sql").write_text(text, encoding="ascii", newline="\n")
-			written += 1
+			(fmt_dir / f"q{n}.py").write_text(SQL_WRAPPERS[fmt].format(n=n), encoding="ascii", newline="\n")
+			written += 2
 	rdf_dir = queries_dir / "rdf"
 	for n, keys in RDF_SPARQL_FILES.items():
 		text = "\n\n".join(f"# retrieval query: {key}\n{rdf_queries.SPARQL[key]}" for key in keys)