diff --git a/CLAUDE.md b/CLAUDE.md index 184b2a2c..73bda378 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -73,7 +73,7 @@ The dependency direction is roughly `utils/` → `github/` + `docker/` + `agents 4. **resolve_packages** — Emit one dependency **seed** per commit, and the story of how it was reached. Six units compose it: `discover` picks the packaging root, `declare` reads only what the project states it needs, `interpreter` walks a declared ladder (`requires-python` → trove classifiers → `asv.conf.json` `pythons` → newest release at commit date) and records the rung in `interpreter_source`, `pin` runs one `uv pip compile` with the commit date as `--exclude-newer`, `probe` dry-runs the result, and the row is written. What it deliberately does not read: `requirements*.txt` globs, `environment.yml`, and import statements — so a project that declares nothing gets an empty seed and says so, rather than a list of invented PyPI names. Benchmark tooling (`asv`, `pytest`, `hypothesis`, `setuptools`, `wheel`, `pip`, `versioneer`) is stripped from both the declared set and the compiled one: the base image owns it, and a second owner only starts a version fight. **The stage gates nothing.** `can_install` is retained, nullable, and no longer read or written; `probe_status` (`installable` → `unresolved` → `failed` → `empty`) orders the stage 5 queue best-first and excludes nobody. Stage 6 is the sole arbiter of buildability, because it is the only stage that builds in the real container. 5. **render_problems** — Scrape linked issues and render deconstructed problem contexts 6. **synthesize_images** — Agent-based Docker build context synthesis (uses env_payload/python_version from stage 4) -7. **harbor_healthcheck** — Run every synthesized container through Harbor's oracle agent, record per-benchmark speedups to `harbor_runs`. Supports local Docker and Daytona via `--harbor-environment`; the row records which one in `harbor_runs.environment`. Local runs are useful for iteration; only Daytona runs gate stage 8. +7. **harbor_healthcheck** — Run every synthesized container through Harbor's oracle agent, record per-benchmark speedups to `harbor_runs`. Supports local Docker and Daytona via `--harbor-environment`; the row records which one in `harbor_runs.environment`. Local runs are useful for iteration; only Daytona runs gate stage 8. An **LSV cache** (`DATASMITH_LSV_CACHE_ENABLED`, default on) lets a repeat trial skip LSV's two expensive passes: the runner bakes the cached survey (`lsv_deps_cache`) into the image and injects datasmith Supabase creds + an 11-column resource key so `lsv_init.py` fetches the cached baselines (`lsv_baseline_cache`) and passes `force=False` only on a full hit; the oracle trial writes both back. It is a pure cost optimization — every miss or error degrades to today's `force=True`, changing no reward or gate — and requires `SUPABASE_URL` to be the `db.formulacode.org` tunnel so the trial container can reach it. 8. **publish** — Build, verify, and publish Docker images to DockerHub. Two gates, and a PR must clear both: it needs at least one successful `harbor_runs` row whose `max_speedup >= 1.05` in an admitted environment (`DATASMITH_PUBLISH_ENVIRONMENTS`, default **daytona** only), *and* its `candidate_containers` row must be `verification_state = 'verified'`. The harbor row says the container is fast; `verification_state` says it is honest, and neither substitutes for the other — `harbor_runs` outlives the container generation that produced it, so a pre-honesty-gate row can carry a fast trial. That second gate is deliberately **not** a knob. 9. **scrape_benchmark_source** — For each `(owner, repo)` in `candidate_containers`, check out the repo at its container SHA, AST-parse every ASV-style benchmark function under the repo's `benchmark_dir`, and upsert one row per `(owner, repo, benchmark_without_params)` into `benchmark_codes` for the FormulaCode website's data sync. @@ -271,6 +271,8 @@ Note that `00016` and `00017` used `GRANT ALL ... TO anon` rather than `GRANT SE | `packages` | One seed per `(owner, repo, sha)`: `env_payload` (pinned deps) and `python_version`, plus `interpreter_source` (which ladder rung chose that interpreter), `primary_root`, `requires_python`, the advisory `probe_status` / `probe_log`, `dropped_requirements` (JSON-encoded text, like `env_payload` — every requirement that was refused, with its reason), and provenance: `resolver_version`, `uv_version`, `resolved_at`, `cutoff_used` (null when the commit-date cutoff had to be relaxed). `can_install` is deprecated — nullable, no longer read or written; `resolver_version = 'legacy'` marks the rows the predecessor wrote. | Stage 4 | | `candidate_containers` | Successful agent-generated `build_pkg_sh` / `build_run_sh` per SHA, plus `build_manifest` (sealed build facts merged with verify observations), `manifest_warnings` (non-fatal invariant ids), and `verification_state` (`unverified` / `verified`, with `verified_at`). `build_manifest IS NULL` identifies rows built before manifests existed; `verification_state = 'unverified'` identifies rows built before the honesty gate applied to them. | Stage 6 (on success) | | `harbor_runs` | One row per Harbor oracle trial for a synthesized container: `max_speedup`, `geomean_speedup`, `n_benchmarks`, `wallclock_sec`, `reward_payload`, `status`. One-to-many FK on `candidate_containers(owner, repo, sha)`. | Stage 7 | +| `lsv_baseline_cache` | Resource-keyed cache of LSV base-commit baseline timings so stage 7 skips the timing pass on a repeat trial. 11-column PK pins every fact that moves a timing (task + env + image + host/machine_class + cgroup pins + in-sandbox `detected_cpu_model`); `baselines` JSONB is `session.export_baselines()`. Advisory, no FK. Oracle trials write it; every trial reads it. A miss falls back to `force=True`. | Stage 7 (oracle writeback) | +| `lsv_deps_cache` | Task-keyed (`owner, repo, issue_number` PK) cache of the LSV coverage **survey** — the `lightspeed_deps.db` SQLite file, baselines stripped, as `deps_db` BYTEA — so stage 7 skips the survey pass. Resource-independent (survey depends only on code, not CPU), hence one row per task. Required for the baseline cache to load at all (`load_baselines` needs the surveyed DB on disk first). | Stage 7 (oracle writeback) | | `benchmark_information` | Per-benchmark speedup measurements from terminal-bench eval runs: one row per (run, owner/repo/issue, benchmark, agent, model). `speedup` is `(agent/nop)/(oracle/nop)` so 1.0 = parity with the human expert. `benchmark_type` (`time`/`mem`/`peakmem`/`track`) is a generated column derived from the ASV naming convention. Loaded out-of-band via `scripts/load_benchmark_information.py`. | (manual) | | `benchmark_codes` | One row per `(owner, repo, benchmark_without_params)` carrying the Python source of each ASV benchmark function plus its setup. Joined to `benchmark_information` on `(owner, repo, benchmark_name)` by the FormulaCode website. | Stage 9 | | `error_logs` | Per-attempt synthesis results: agent output, failure stage/return code, error messages | Stage 6 (`Synthesizer._log_attempt`) | @@ -285,7 +287,7 @@ Note that `00016` and `00017` used `GRANT ALL ... TO anon` rather than `GRANT SE ### Migrations -SQL migrations live in `supabase/migrations/`, numbered `00001_` upward (currently through `00029_`). The sequence has gaps because numbers get claimed on branches before they land: `00018_lsv_cache_drop_cpu_model.sql` lives on `origin/lsv-cache-integration`, and `00024` is authored in a separate working tree (per `00025`'s header) — `00026` re-lands that same table under a number that is free here. So check other branches before claiming a number, and record in the file header why you skipped one. `00027_pull_requests_window_indexes.sql` adds the two indexes the stage 2–5 window predicates need on `pull_requests` — `merged_at` for the stage-wide scan, `(owner, repo, merged_at)` for the per-repository skip set — and deliberately grants nothing to `anon`. `00028_packages_resolution_v2.sql` carries the stage 4 redesign's provenance columns, and `00029_candidate_containers_verification_state.sql` adds `verification_state` — every pre-existing row defaults to `unverified`, because the corpus predates the honesty gate and has not earned the label. +SQL migrations live in `supabase/migrations/`, numbered `00001_` upward (currently through `00032_`). The sequence has gaps because numbers get claimed on branches before they land: `00018_lsv_cache_drop_cpu_model.sql` lives on `origin/lsv-cache-integration`, and `00024` is authored in a separate working tree (per `00025`'s header) — `00026` re-lands that same table under a number that is free here. So check other branches before claiming a number, and record in the file header why you skipped one. `00027_pull_requests_window_indexes.sql` adds the two indexes the stage 2–5 window predicates need on `pull_requests` — `merged_at` for the stage-wide scan, `(owner, repo, merged_at)` for the per-repository skip set — and deliberately grants nothing to `anon`. `00028_packages_resolution_v2.sql` carries the stage 4 redesign's provenance columns, and `00029_candidate_containers_verification_state.sql` adds `verification_state` — every pre-existing row defaults to `unverified`, because the corpus predates the honesty gate and has not earned the label. `00031_lsv_baseline_cache.sql` and `00032_lsv_deps_cache.sql` add the stage-7 LSV cache tables (baseline timings and the survey deps DB); both are private (`GRANT SELECT ... TO grafana_ro`, no anon) and squash the old `origin/lsv-cache-integration` cpu-model churn (`00016`→`00019`) into one clean pair. To apply a new migration against the local instance: diff --git a/src/datasmith/harbor_adapter/adapter.py b/src/datasmith/harbor_adapter/adapter.py index f68bedb6..724b4104 100644 --- a/src/datasmith/harbor_adapter/adapter.py +++ b/src/datasmith/harbor_adapter/adapter.py @@ -98,6 +98,7 @@ def _copy_template_files(self, paths: HarborTaskPaths) -> None: "entrypoint.sh", "lsv_init.py", "lsv_measure.py", + "lsv_cache_writeback.py", "parser.py", "upload.py", "pytest_runner.py", @@ -105,6 +106,24 @@ def _copy_template_files(self, paths: HarborTaskPaths) -> None: ]: copy2(self.template_dir / name, paths.environment_dir / name) + def _write_cache_files(self, paths: HarborTaskPaths, deps_db: bytes | None) -> None: + """Populate ``environment/cache/`` with the pre-surveyed LSV deps DB. + + The directory is created unconditionally -- even on a cache miss -- and + seeded with a ``.gitkeep`` so the Dockerfile's ``COPY cache/`` directive + always has a source and the build never fails. When ``deps_db`` is + present (a hit staged by the runner), it is written as + ``lightspeed_deps.db``; the Dockerfile bakes it to + ``/opt/lsv/cache/lightspeed_deps.db``, where lsv_init.py stages it before + ``load_baselines`` (which requires the surveyed DB on disk first). A miss + leaves only ``.gitkeep``, so lsv_init falls through to force=True. + """ + cache_dir = paths.environment_dir / "cache" + cache_dir.mkdir(parents=True, exist_ok=True) + (cache_dir / ".gitkeep").write_text("") + if deps_db: + (cache_dir / "lightspeed_deps.db").write_bytes(deps_db) + def _write_instruction_md(self, rec: FormulaCodeRecord, paths: HarborTaskPaths) -> None: """Generate instruction.md file.""" instruction_content = render_instruction_md(rec.instructions) @@ -146,6 +165,7 @@ def _write_test_files( paths: HarborTaskPaths, run_pytest: bool = True, rounds: int = 1, + render_env: dict[str, str] | None = None, ) -> None: """Generate test files (test.sh, config.json).""" # test.sh @@ -156,6 +176,7 @@ def _write_test_files( issue_number=rec.issue_number, run_pytest=run_pytest, rounds=rounds, + render_env=render_env, ) test_sh_path = paths.tests_dir / "test.sh" test_sh_path.write_text(test_sh_content) @@ -179,6 +200,7 @@ def _write_solution_files( rec: FormulaCodeRecord, paths: HarborTaskPaths, rounds: int = 1, + render_env: dict[str, str] | None = None, ) -> None: """Generate solution files (solve.sh, setup.sh).""" # solve.sh @@ -195,6 +217,7 @@ def _write_solution_files( issue_number=rec.issue_number, rounds=rounds, extra_setup_commands=extra_setup_commands, + render_env=render_env, ) setup_sh_path = paths.tests_dir / "setup.sh" setup_sh_path.write_text(setup_sh_content) @@ -211,9 +234,16 @@ def generate_task( rounds: int = DATASMITH_LSV_ROUNDS, verifier_env: dict[str, str] | None = None, expected_n: int | None = None, + render_env: dict[str, str] | None = None, + deps_db: bytes | None = None, ) -> Path: """Generate a complete Harbor task directory for the given FormulaCodeRecord. + ``render_env`` is baked as ``export K=V`` lines into setup.sh and test.sh + (see ``render_run_setup_sh``). Stage 7 uses it to hand the LSV baseline + cache its datasmith Supabase creds and resource key; ``None`` bakes + nothing, leaving both scripts byte-identical to the pre-cache output. + ``expected_n`` is the operator-declared count of benchmarks this PR should impact, read from ``formulacode_task_overrides``. It is injected into the trial container as ``FORMULACODE_EXPECTED_N`` -- the producer @@ -226,6 +256,11 @@ def generate_task( That is the common case (the column is hand-declared and usually NULL), and the invariant then skips. Emitting a key that is always empty would make the wiring look live when it is not. + + ``deps_db`` is the pre-surveyed LSV deps DB the runner fetched from + ``lsv_deps_cache`` for this task; it is baked into the image so lsv_init + can skip the survey pass. ``None`` bakes only an empty ``cache/`` + placeholder, leaving lsv_init on force=True. """ out_dir = self.out_root / rec.task_dir_name out_dir.mkdir(parents=True, exist_ok=True) @@ -235,6 +270,7 @@ def generate_task( # Copy static template files self._copy_template_files(paths) + self._write_cache_files(paths, deps_db) # Generate all task files self._write_instruction_md(rec, paths) @@ -242,7 +278,7 @@ def generate_task( verifier_env = {**(verifier_env or {}), "FORMULACODE_EXPECTED_N": str(expected_n)} self._write_task_toml(rec, paths, timeout_sec, cpus, memory, storage, verifier_env=verifier_env) self._write_environment_files(rec, paths) - self._write_test_files(rec, paths, run_pytest, rounds) - self._write_solution_files(rec, paths, rounds) + self._write_test_files(rec, paths, run_pytest, rounds, render_env=render_env) + self._write_solution_files(rec, paths, rounds, render_env=render_env) return out_dir diff --git a/src/datasmith/harbor_adapter/template/Dockerfile b/src/datasmith/harbor_adapter/template/Dockerfile index d51aea51..09589c42 100644 --- a/src/datasmith/harbor_adapter/template/Dockerfile +++ b/src/datasmith/harbor_adapter/template/Dockerfile @@ -3,19 +3,24 @@ FROM {{ base_image }} # Install required system packages RUN apt-get update && apt-get install -y git tmux asciinema RUN curl -LsSf https://astral.sh/uv/0.7.13/install.sh | sh -RUN mkdir -p /logs /opt/lsv /tests +RUN mkdir -p /logs /opt/lsv /opt/lsv/cache /tests # Copy environment setup scripts COPY entrypoint.sh /entrypoint.sh # Bake LSV helper scripts into the image so setup.sh can find them before # Harbor's verifier mounts /tests/ (verifier runs AFTER setup.sh). -COPY lsv_init.py lsv_measure.py parser.py upload.py pytest_runner.py jinja_patch_plugin_pandas.py /opt/lsv/ +COPY lsv_init.py lsv_measure.py lsv_cache_writeback.py parser.py upload.py pytest_runner.py jinja_patch_plugin_pandas.py /opt/lsv/ # lsv_init.py reads /tests/config.json during setup.sh — bake it in too. # Harbor's verifier later overwrites this with identical content. COPY config.json /tests/config.json +# Pre-surveyed LSV deps DB (lsv_deps_cache), when the runner staged one. The +# adapter always creates environment/cache/ (with a .gitkeep) so this COPY has a +# source even on a cache miss; lsv_init reads lightspeed_deps.db from here. +COPY cache/ /opt/lsv/cache/ + RUN chmod +x /entrypoint.sh ENV PYTHONPATH=/opt/lsv:${PYTHONPATH:-} diff --git a/src/datasmith/harbor_adapter/template/lsv_cache_writeback.py b/src/datasmith/harbor_adapter/template/lsv_cache_writeback.py new file mode 100644 index 00000000..5cfede55 --- /dev/null +++ b/src/datasmith/harbor_adapter/template/lsv_cache_writeback.py @@ -0,0 +1,386 @@ +"""LSV cache writeback — runs from test.sh after lsv_measure. + +Reads ``lsv_cache_state.json`` (written by lsv_init.py this trial) and, on a +MISS, persists the freshly measured facts to datasmith's own Supabase so the +next trial on the same task/hardware hits and skips a pass: + +* baselines (resource-keyed) -> ``lsv_baseline_cache`` (00031), on a baseline + miss. Skips the base-commit timing pass. +* the coverage survey (resource-independent) -> ``lsv_deps_cache`` (00032), on a + deps miss. Skips the survey pass. Uploaded baselines-stripped so no per-host + timing rows leak across hardware. + +Oracle-only: only the oracle trial produces canonical facts. An agent run +measures post-patch code, so letting it write would corrupt the very baseline +future agent runs compare against. This self-gates on ``HARBOR_AGENT_NAME``, and +test.sh only invokes it on the oracle branch. + +Stdlib only (``urllib`` + ``sqlite3``) so nothing extra is baked into the image. + +Usage (invoked from test.sh; all inputs from env): + python /opt/lsv/lsv_cache_writeback.py +""" + +from __future__ import annotations + +import json +import os +import shutil +import sqlite3 +import sys +import tempfile +import urllib.error +import urllib.request +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +LSV_DIR = Path(os.environ.get("LSV_OUTPUT_DIR", "/logs/artifacts/lsv")) +CACHE_STATE_PATH = LSV_DIR / "lsv_cache_state.json" +_RESOURCE_ATTRS_PATH = LSV_DIR / "lsv_resource_attrs.json" + +# Order matters: the upsert enumerates these as the on_conflict target, so it +# must match the primary key declared in migration 00031 exactly. +_BASELINE_PK_COLS = ( + "owner", "repo", "issue_number", + "env", "container_name", "image_digest", + "machine_class", "docker_host_id", + "cpu_count", "mem_bytes", + "detected_cpu_model", +) + + +def _ts() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _service_headers() -> dict[str, str]: + """Auth headers for datasmith's own Supabase (NOT Harbor's project). + + The two are separate Postgres projects: Harbor's holds run uploads and + snapshots, datasmith's holds ``lsv_baseline_cache``. The User-Agent override + matters -- Cloudflare's bot-fight rule blocks the default ``Python-urllib`` + UA before Access sees the request. CF-Access headers are injected when + ``db.formulacode.org`` is the target (see CLAUDE.md remote-access). + """ + key = os.environ.get("DATASMITH_SUPABASE_SERVICE_KEY", "") + if not key: + raise RuntimeError("DATASMITH_SUPABASE_SERVICE_KEY not set") + headers = { + "Authorization": f"Bearer {key}", + "apikey": key, + "User-Agent": "datasmith-lsv-cache/1.0", + } + cf_id = os.environ.get("DATASMITH_CF_ACCESS_CLIENT_ID", "") + cf_secret = os.environ.get("DATASMITH_CF_ACCESS_CLIENT_SECRET", "") + if cf_id and cf_secret: + headers["CF-Access-Client-Id"] = cf_id + headers["CF-Access-Client-Secret"] = cf_secret + return headers + + +def _request( + url: str, + method: str, + headers: dict[str, str], + data: bytes | None = None, + content_type: str | None = None, +) -> bytes: + if content_type: + headers = {**headers, "Content-Type": content_type} + req = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=120) as resp: # noqa: S310 + return resp.read() + except urllib.error.HTTPError as e: + body = e.read().decode(errors="replace") + print(f"[{_ts()}] [cache_writeback] ERROR: {method} {url} -> {e.code}: {body}") + raise + + +def _parse_task_id(task_id: str) -> tuple[str, str, int] | None: + """Parse ``owner__repo__`` (double-underscore, matching + ``adapter.FormulaCodeRecord.task_dir_name``). Returns None if malformed.""" + try: + owner, repo, issue_str = task_id.rsplit("__", 2) + if not owner or not repo: + return None + return owner, repo, int(issue_str) + except (ValueError, IndexError): + return None + + +def _read_baselines(deps_db_path: Path) -> dict[str, dict[str, float | int]]: + """Pull every baseline row out of the deps DB as a JSON-serializable dict. + Shape matches ``session.export_baselines`` so a future ``load_baselines`` + accepts it verbatim.""" + if not deps_db_path.exists(): + return {} + with sqlite3.connect(deps_db_path) as con: + rows = con.execute( + "SELECT benchmark_id, median, ci_99_a, ci_99_b, q_25, q_75, repeat, number " + "FROM baseline" + ).fetchall() + out: dict[str, dict[str, float | int]] = {} + for bid, median, ci_a, ci_b, q25, q75, rep, num in rows: + out[bid] = { + "median": median, + "ci_99_a": ci_a, + "ci_99_b": ci_b, + "q_25": q25, + "q_75": q75, + "repeat": rep, + "number": num, + } + return out + + +def _survey_bytes(deps_db_path: Path) -> bytes | None: + """Return the survey-only deps DB as bytes: a copy with every baseline row + deleted and the WAL checkpointed into the main file. + + The DELETE strips the resource-DEPENDENT timings (they live in + lsv_baseline_cache under a hardware key); the survey that remains is + resource-independent and safe to reuse on any host. ``PRAGMA + wal_checkpoint(TRUNCATE)`` is load-bearing: without it the deleted rows + survive in the WAL and get uploaded anyway, reintroducing exactly the + cross-host timing pollution the DELETE removed. Operates on a copy so the + live DB the trial still uses is untouched. Returns None if absent/unreadable. + """ + if not deps_db_path.exists() or deps_db_path.stat().st_size == 0: + return None + tmp_dir = tempfile.mkdtemp(prefix="lsv_deps_") + try: + tmp_db = Path(tmp_dir) / "survey.db" + shutil.copy(deps_db_path, tmp_db) + with sqlite3.connect(tmp_db) as con: + con.execute("DELETE FROM baseline") + con.commit() + con.execute("PRAGMA wal_checkpoint(TRUNCATE)") + return tmp_db.read_bytes() + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +def _upsert_deps_cache_row( + base_url: str, + *, + owner: str, + repo: str, + issue_number: int, + survey: bytes, +) -> None: + """Upsert the survey blob into ``lsv_deps_cache`` keyed by the task alone. + BYTEA moves through PostgREST as PostgreSQL's hex form: send ``\\x`` and + Postgres parses it back to the raw bytes (the runner reverses this on read).""" + row = { + "owner": owner, + "repo": repo, + "issue_number": issue_number, + "deps_db": "\\x" + survey.hex(), + "updated_at": datetime.now(timezone.utc).isoformat(), + } + headers = { + **_service_headers(), + "Prefer": "resolution=merge-duplicates,return=minimal", + } + _request( + f"{base_url}/rest/v1/lsv_deps_cache?on_conflict=owner,repo,issue_number", + "POST", + headers, + data=json.dumps(row).encode(), + content_type="application/json", + ) + print( + f"[{_ts()}] [cache_writeback] upserted lsv_deps_cache for " + f"{owner}/{repo}#{issue_number} ({len(survey)} bytes)" + ) + + +def _upsert_baseline_cache_row( + base_url: str, + *, + owner: str, + repo: str, + issue_number: int, + attrs: dict[str, Any], + baselines: dict[str, Any], +) -> None: + """Upsert one ``lsv_baseline_cache`` row keyed on the full resource tuple. + ``attrs`` supplies every non-task PK column; on_conflict enumerates the + whole PK so a re-measure on identical hardware replaces the row.""" + row = { + "owner": owner, + "repo": repo, + "issue_number": issue_number, + "baselines": baselines, + "updated_at": datetime.now(timezone.utc).isoformat(), + **attrs, + } + headers = { + **_service_headers(), + "Prefer": "resolution=merge-duplicates,return=minimal", + } + on_conflict = ",".join(_BASELINE_PK_COLS) + _request( + f"{base_url}/rest/v1/lsv_baseline_cache?on_conflict={on_conflict}", + "POST", + headers, + data=json.dumps(row).encode(), + content_type="application/json", + ) + print( + f"[{_ts()}] [cache_writeback] upserted lsv_baseline_cache for " + f"{owner}/{repo}#{issue_number} env={attrs['env']} " + f"digest={str(attrs['image_digest'])[:24]} ({len(baselines)} baselines)" + ) + + +def _read_resource_attrs() -> dict[str, Any]: + """Load the cache key from the JSON lsv_init.py wrote — the single source of + truth, so lookup and upsert never disagree on '' vs "". Falls back to the + LSV_* env vars if the file is missing (lsv_init did not complete), which + loses only ``detected_cpu_model`` (no env equivalent — it is /proc-derived).""" + if not _RESOURCE_ATTRS_PATH.exists(): + print( + f"[{_ts()}] [cache_writeback] WARNING: {_RESOURCE_ATTRS_PATH} missing — " + "falling back to LSV_* env vars" + ) + return _read_resource_attrs_from_env() + try: + attrs: dict[str, Any] = json.loads(_RESOURCE_ATTRS_PATH.read_text()) + except (OSError, json.JSONDecodeError) as exc: + print(f"[{_ts()}] [cache_writeback] WARNING: {_RESOURCE_ATTRS_PATH} unreadable ({exc})") + return _read_resource_attrs_from_env() + for k in ("cpu_count", "mem_bytes"): + try: + attrs[k] = int(attrs.get(k, 0)) + except (TypeError, ValueError): + attrs[k] = 0 + return attrs + + +def _read_resource_attrs_from_env() -> dict[str, Any]: + """Fallback cache key from env. cpu_count/mem_bytes are never read from /proc + (inside the sandbox they show the host, not the cgroup pin); detected_cpu_model + has no env source, so it defaults to '' here.""" + + def _int(v: str) -> int: + try: + return int(v) + except (TypeError, ValueError): + return 0 + + return { + "env": os.environ.get("LSV_ENV", ""), + "container_name": os.environ.get("LSV_CONTAINER_NAME", ""), + "image_digest": os.environ.get("LSV_IMAGE_DIGEST", ""), + "machine_class": os.environ.get("LSV_MACHINE_CLASS", ""), + "docker_host_id": os.environ.get("LSV_DOCKER_HOST_ID", ""), + "cpu_count": _int(os.environ.get("LSV_CPU_COUNT", "")), + "mem_bytes": _int(os.environ.get("LSV_MEM_BYTES", "")), + "detected_cpu_model": "", + } + + +def _env_summary() -> str: + """One-line ``NAME=set|unset`` summary (never values — the key is secret) so + silent-skip failures are diagnosable.""" + keys = [ + "DATASMITH_SUPABASE_URL", + "DATASMITH_SUPABASE_SERVICE_KEY", + "DATASMITH_CF_ACCESS_CLIENT_ID", + "DATASMITH_CF_ACCESS_CLIENT_SECRET", + "LSV_TASK_ID", + "HARBOR_AGENT_NAME", + "FORMULACODE_NO_UPLOAD", + ] + return ", ".join(f"{k}={'set' if os.environ.get(k) else 'unset'}" for k in keys) + + +def main() -> int: + print(f"[{_ts()}] [cache_writeback] starting; env: {_env_summary()}") + + # Oracle-only: bail before any I/O so an agent run can never overwrite the + # oracle baseline it reads. + agent_name = os.environ.get("HARBOR_AGENT_NAME", "oracle").lower() + if agent_name != "oracle": + print(f"[{_ts()}] [cache_writeback] SKIP: HARBOR_AGENT_NAME={agent_name!r} (not oracle)") + return 0 + + base_url = os.environ.get("DATASMITH_SUPABASE_URL") + if not base_url: + print(f"[{_ts()}] [cache_writeback] SKIP: DATASMITH_SUPABASE_URL not set") + return 0 + if os.environ.get("FORMULACODE_NO_UPLOAD"): + print(f"[{_ts()}] [cache_writeback] SKIP: FORMULACODE_NO_UPLOAD is set") + return 0 + if not os.environ.get("DATASMITH_SUPABASE_SERVICE_KEY"): + print(f"[{_ts()}] [cache_writeback] SKIP: DATASMITH_SUPABASE_SERVICE_KEY not set") + return 0 + if not CACHE_STATE_PATH.exists(): + print(f"[{_ts()}] [cache_writeback] SKIP: no {CACHE_STATE_PATH} (lsv_init likely failed)") + return 0 + + state = json.loads(CACHE_STATE_PATH.read_text()) + deps_was_cached = bool(state.get("deps_was_cached")) + baselines_was_cached = bool(state.get("baselines_was_cached")) + deps_db_path = Path(state.get("deps_db_path") or "") + print( + f"[{_ts()}] [cache_writeback] state: deps_was_cached={deps_was_cached} " + f"baselines_was_cached={baselines_was_cached} deps_db_path={deps_db_path}" + ) + if deps_was_cached and baselines_was_cached: + print(f"[{_ts()}] [cache_writeback] full cache hit, nothing to write back") + return 0 + + task_id = os.environ.get("LSV_TASK_ID", "") + parsed = _parse_task_id(task_id) + if parsed is None: + print(f"[{_ts()}] [cache_writeback] SKIP: cannot parse LSV_TASK_ID={task_id!r}") + return 0 + owner, repo, issue_number = parsed + + rc = 0 + + # Deps (survey) layer: task-keyed, resource-independent. Upload only on a + # miss; a hit means the staged survey is already canonical. + if not deps_was_cached: + try: + survey = _survey_bytes(deps_db_path) + if survey is None: + print(f"[{_ts()}] [cache_writeback] no deps DB at {deps_db_path}, nothing to upload") + else: + _upsert_deps_cache_row( + base_url, owner=owner, repo=repo, issue_number=issue_number, survey=survey + ) + except Exception as exc: # noqa: BLE001 -- best-effort, never fails the trial + print(f"[{_ts()}] [cache_writeback] deps writeback FAILED: {exc}") + rc = 1 + + # Baseline layer: resource-keyed. Needs the full hardware key (env etc.). + if not baselines_was_cached: + attrs = _read_resource_attrs() + if not attrs["env"]: + print(f"[{_ts()}] [cache_writeback] SKIP baselines: missing LSV_ENV (resource columns not provided)") + else: + try: + baselines = _read_baselines(deps_db_path) + if not baselines: + print(f"[{_ts()}] [cache_writeback] no baselines in {deps_db_path}, nothing to write") + else: + _upsert_baseline_cache_row( + base_url, + owner=owner, repo=repo, issue_number=issue_number, + attrs=attrs, baselines=baselines, + ) + except Exception as exc: # noqa: BLE001 -- best-effort, never fails the trial + print(f"[{_ts()}] [cache_writeback] baselines writeback FAILED: {exc}") + rc = 1 + + print(f"[{_ts()}] [cache_writeback] done") + return rc + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/datasmith/harbor_adapter/template/lsv_init.py b/src/datasmith/harbor_adapter/template/lsv_init.py index 5079c759..8fd36de9 100644 --- a/src/datasmith/harbor_adapter/template/lsv_init.py +++ b/src/datasmith/harbor_adapter/template/lsv_init.py @@ -4,6 +4,27 @@ runs initialize_diffcheck to build the dependency database and baseline timing, then captures a snapshot of the benchmark environment. +Baseline cache +-------------- +``initialize_diffcheck`` runs two passes -- a coverage survey and a +base-commit timing pass -- and the timing pass dominates. Both are skipped +when the deps DB already holds the survey AND a baseline for every benchmark +(see the LSV fork, ``asv/contrib/lightspeed/session.py``). This module reuses +that cache across trials: + +* deps DB (survey, resource-independent) -- staged by the runner into + ``/opt/lsv/cache/lightspeed_deps.db`` at build time, if available. +* baselines (resource-keyed) -- fetched at runtime from datasmith's own + Supabase (``lsv_baseline_cache``), because only inside the sandbox can we + read the CPU model that keys them. The key blends runner-supplied attrs + (``cpu_count``/``mem_bytes`` -- the cgroup limits the runner asked Harbor to + pin) with ``detected_cpu_model`` read from ``/proc/cpuinfo`` here. + +A miss on either layer degrades to today's behaviour: ``force=True`` re-runs +both passes. The hit/miss decision is recorded in ``lsv_cache_state.json`` for +the post-trial writeback, and the resolved key in ``lsv_resource_attrs.json`` +so the writeback reuses identical values. + Usage: python /tests/lsv_init.py [--rounds N] """ @@ -14,11 +35,16 @@ import json import os import re +import shutil import subprocess import sys +import urllib.error +import urllib.parse +import urllib.request from datetime import datetime, timezone from glob import glob from pathlib import Path +from typing import Any def _ts() -> str: @@ -31,6 +57,148 @@ def _ts() -> str: SNAPSHOT_FILTER = os.environ.get("FORMULACODE_SNAPSHOT_FILTER", r".*") SNAPSHOT_TIMEOUT = os.environ.get("FORMULACODE_SNAPSHOT_TIMEOUT", "30") +# The runner bakes a cached, baseline-stripped deps DB (the coverage survey) +# to this path at build time when one is available. Absent on the first run of +# a task and whenever the cache is disabled -- both fall through to force=True. +CACHED_DEPS_DB = Path(os.environ.get("LSV_CACHED_DEPS_DB", "/opt/lsv/cache/lightspeed_deps.db")) + + +# /proc/cpuinfo and /proc/meminfo inside the sandbox report the underlying +# physical host (Daytona's full hardware), not the cgroup limit -- so cpu_count +# and mem_bytes must come from the runner (what it asked Harbor to pin), never +# from /proc. detected_cpu_model is the ONE attr we do read from /proc here: it +# is exactly the host-level fact we want, because Daytona's `default` +# machine_class fans one (cpu, mem) request across several EPYC SKUs +# (9254/9334/9354P) that benchmark differently, and a baseline must not replay +# across them. +_MODEL_RE = re.compile(r"^model name\s*:\s*(.+)$", re.MULTILINE) + + +def _detect_cpu_model() -> str: + """First ``model name`` from /proc/cpuinfo, or '' if unreadable.""" + try: + text = Path("/proc/cpuinfo").read_text() + except OSError: + return "" + m = _MODEL_RE.search(text) + return m.group(1).strip() if m else "" + + +def _read_resource_attrs() -> dict[str, Any]: + """Build the 11-column cache key for this trial: seven runner-supplied + attrs (via setup.sh's render_env block), ``cpu_count``/``mem_bytes`` (the + pinned cgroup limits, also runner-supplied), and ``detected_cpu_model`` + read from /proc/cpuinfo inside this sandbox.""" + + def _int(v: str) -> int: + try: + return int(v) + except (TypeError, ValueError): + return 0 + + return { + "env": os.environ.get("LSV_ENV", ""), + "container_name": os.environ.get("LSV_CONTAINER_NAME", ""), + "image_digest": os.environ.get("LSV_IMAGE_DIGEST", ""), + "machine_class": os.environ.get("LSV_MACHINE_CLASS", ""), + "docker_host_id": os.environ.get("LSV_DOCKER_HOST_ID", ""), + "cpu_count": _int(os.environ.get("LSV_CPU_COUNT", "")), + "mem_bytes": _int(os.environ.get("LSV_MEM_BYTES", "")), + "detected_cpu_model": _detect_cpu_model(), + } + + +def _datasmith_headers() -> dict[str, str] | None: + """Auth headers for datasmith's own Supabase (not Harbor's project). + + Returns None when the service key is absent, which disables the cache. The + User-Agent override matters: Cloudflare's bot-fight rule blocks the default + ``Python-urllib`` UA before Access sees the request. Mirrors + ``lsv_cache_writeback._service_headers``. + """ + key = os.environ.get("DATASMITH_SUPABASE_SERVICE_KEY", "") + if not key: + return None + headers = { + "Authorization": f"Bearer {key}", + "apikey": key, + "Accept": "application/json", + "User-Agent": "datasmith-lsv-init/1.0", + } + cf_id = os.environ.get("DATASMITH_CF_ACCESS_CLIENT_ID", "") + cf_secret = os.environ.get("DATASMITH_CF_ACCESS_CLIENT_SECRET", "") + if cf_id and cf_secret: + headers["CF-Access-Client-Id"] = cf_id + headers["CF-Access-Client-Secret"] = cf_secret + return headers + + +def _parse_task_id(task_id: str) -> tuple[str, str, int] | None: + """Parse ``owner__repo__`` (double-underscore, matching the runner's + ``rec.task_dir_name``). Returns None on a malformed id.""" + try: + owner, repo, issue_str = task_id.rsplit("__", 2) + if not owner or not repo: + return None + return owner, repo, int(issue_str) + except (ValueError, IndexError): + return None + + +def _fetch_baselines_from_cache(attrs: dict[str, Any]) -> dict[str, Any] | None: + """Look up an exact-match ``lsv_baseline_cache`` row on datasmith Supabase. + + Returns the cached ``baselines`` JSONB, or None on miss / missing creds / + network failure -- every None path falls through to a fresh measure. + """ + base_url = os.environ.get("DATASMITH_SUPABASE_URL", "") + headers = _datasmith_headers() + task_id = os.environ.get("LSV_TASK_ID", "") + if not base_url or not headers or not task_id: + print(f"[{_ts()}] [lsv_init] cache: skip baseline lookup (creds/task_id missing)") + return None + parsed = _parse_task_id(task_id) + if parsed is None: + print(f"[{_ts()}] [lsv_init] cache: could not parse LSV_TASK_ID={task_id!r}") + return None + owner, repo, issue = parsed + + qs = [ + "select=baselines", + f"owner=eq.{urllib.parse.quote(owner)}", + f"repo=eq.{urllib.parse.quote(repo)}", + f"issue_number=eq.{issue}", + ] + for k, v in attrs.items(): + qs.append(f"{k}=eq.{urllib.parse.quote(str(v))}") + url = f"{base_url}/rest/v1/lsv_baseline_cache?" + "&".join(qs) + if not url.startswith(("https://", "http://")): + return None + req = urllib.request.Request(url, headers=headers, method="GET") + try: + with urllib.request.urlopen(req, timeout=30) as resp: # noqa: S310 + rows = json.loads(resp.read().decode()) + except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError) as exc: + print(f"[{_ts()}] [lsv_init] cache: baseline lookup failed ({exc})") + return None + if not rows: + return None + return rows[0].get("baselines") or None + + +def _append_profile_export(name: str, value: str) -> None: + """Append ``export NAME=value`` to the shared asv profile file so a later + process in the trial (test.sh sources it) sees the variable. Idempotent: + an existing entry for ``name`` is left as-is.""" + line = f"export {name}={value}\n" + profile_path = Path("/etc/profile.d/asv_build_vars.sh") + if profile_path.exists(): + if name not in profile_path.read_text(): + with open(profile_path, "a") as f: + f.write(line) + else: + profile_path.write_text(line) + def detect_source_root() -> Path: """Derive the source package root from /tests/config.json patch headers. @@ -377,14 +545,55 @@ def main() -> None: print(f"[{_ts()}] [lsv_init] benchmark_dir={session.benchmark_dir}") + # ── Resolve the cache key and persist it for the writeback ─────────── + # Single source of truth: lsv_cache_writeback.py reads the same file, so + # lookup and upsert can never disagree on '' vs "" for empty fields. + resource_attrs = _read_resource_attrs() + (OUTPUT_DIR / "lsv_resource_attrs.json").write_text(json.dumps(resource_attrs, indent=2)) + print( + f"[{_ts()}] [lsv_init] resource_attrs: env={resource_attrs['env']!r} " + f"machine_class={resource_attrs['machine_class']!r} " + f"docker_host_id={resource_attrs['docker_host_id']!r} " + f"cpu_count={resource_attrs['cpu_count']} mem_bytes={resource_attrs['mem_bytes']} " + f"detected_cpu_model={resource_attrs['detected_cpu_model']!r}" + ) + + # ── Cache layer 1: deps DB (survey), pre-staged by the runner ──────── + deps_was_cached = False + baselines_was_cached = False + if CACHED_DEPS_DB.exists() and CACHED_DEPS_DB.stat().st_size > 0: + session.deps_db_path.parent.mkdir(parents=True, exist_ok=True) + shutil.copy(CACHED_DEPS_DB, session.deps_db_path) + deps_was_cached = True + print(f"[{_ts()}] [lsv_init] cache: staged deps DB from {CACHED_DEPS_DB}") + + # ── Cache layer 2: baselines (resource-keyed, fetched at runtime) ──── + # load_baselines needs the deps DB present first (it raises otherwise), so + # only attempt it once the survey has been staged. + if deps_was_cached: + cached_baselines = _fetch_baselines_from_cache(resource_attrs) + if cached_baselines: + try: + n = session.load_baselines(cached_baselines) + baselines_was_cached = True + print(f"[{_ts()}] [lsv_init] cache: loaded {n} baselines from lsv_baseline_cache") + except Exception as exc: # noqa: BLE001 -- any failure falls through to a fresh measure + print(f"[{_ts()}] [lsv_init] cache: load_baselines failed ({exc}); measuring fresh") + + # A full hit (survey + every baseline present) lets initialize_diffcheck + # short-circuit both passes. Anything less: force=True guarantees a fresh + # measure rather than trusting a partially populated DB -- identical to the + # historical unconditional force=True behaviour. + full_hit = deps_was_cached and baselines_was_cached + # Run initialize_diffcheck print("=" * 64) - print(f"[{_ts()}] [phase] LSV initialize_diffcheck") + print(f"[{_ts()}] [phase] LSV initialize_diffcheck (force={not full_hit})") print("=" * 64) init_result = session.initialize_diffcheck( source_root=source_root, - force=True, + force=not full_hit, rounds=args.rounds, repeat=args.repeat, warmup_time=args.warmup_time, @@ -427,6 +636,24 @@ def main() -> None: } (OUTPUT_DIR / "lsv_init_results.json").write_text(json.dumps(init_data, indent=2)) + # ── Record cache hit/miss for the post-trial writeback ────────────── + # Any miss tells lsv_cache_writeback.py which layer to upload so the next + # trial on the same (task, hardware) hits. + cache_state = { + "deps_was_cached": deps_was_cached, + "baselines_was_cached": baselines_was_cached, + "deps_db_path": str(session.deps_db_path), + } + (OUTPUT_DIR / "lsv_cache_state.json").write_text(json.dumps(cache_state, indent=2)) + + # Surface the baseline hit to parser.py (which runs later in test.sh, a + # separate process) via the sourced profile file. Invariant #20 + # (baseline_from_cache) reads it: a cached oracle baseline replayed to a + # warm agent run is flagged advisory, gating nothing. + _append_profile_export( + "FORMULACODE_BASELINE_FROM_CACHE", "1" if baselines_was_cached else "0" + ) + # Snapshot capture (oracle only — production runs download pre-built snapshots) agent_name = os.environ.get("HARBOR_AGENT_NAME", "").lower() if agent_name == "oracle": @@ -437,16 +664,7 @@ def main() -> None: print(f"[{_ts()}] [lsv_init] Skipping snapshot capture (non-oracle run)") # Export BENCHMARK_DIR for downstream scripts - benchmark_dir_str = str(session.benchmark_dir) - profile_line = f"export BENCHMARK_DIR={benchmark_dir_str}\n" - profile_path = Path("/etc/profile.d/asv_build_vars.sh") - if profile_path.exists(): - existing = profile_path.read_text() - if "BENCHMARK_DIR" not in existing: - with open(profile_path, "a") as f: - f.write(profile_line) - else: - profile_path.write_text(profile_line) + _append_profile_export("BENCHMARK_DIR", str(session.benchmark_dir)) print(f"[{_ts()}] [lsv_init] Complete. Results at {OUTPUT_DIR}") diff --git a/src/datasmith/harbor_adapter/template/setup.sh b/src/datasmith/harbor_adapter/template/setup.sh index b9a38d41..787698cc 100644 --- a/src/datasmith/harbor_adapter/template/setup.sh +++ b/src/datasmith/harbor_adapter/template/setup.sh @@ -17,6 +17,14 @@ export OWNER REPO ISSUE_NUMBER TASK_ID # reads this to decide whether to capture the oracle snapshot baseline. export HARBOR_AGENT_NAME="oracle" +# LSV baseline cache: creds for datasmith's own Supabase + the resource key +# fields, baked here for the same reason as HARBOR_AGENT_NAME ([verifier.env] +# does not reach setup.sh on Daytona). Empty unless the runner enabled the +# cache, in which case lsv_init.py reads these to look baselines up. +{%- for k, v in render_env.items() %} +export {{ k }}={{ v | shell_quote }} +{%- endfor %} + setup_start=$(date +%s) # Track which phase we're currently in so the exit trap can report WHERE # setup crashed, not just the exit code. Updated at every phase boundary diff --git a/src/datasmith/harbor_adapter/template/test.sh b/src/datasmith/harbor_adapter/template/test.sh index 8dddce5f..86b438b9 100644 --- a/src/datasmith/harbor_adapter/template/test.sh +++ b/src/datasmith/harbor_adapter/template/test.sh @@ -23,6 +23,14 @@ ISSUE_NUMBER="{{ issue_number }}" TASK_ID="${ISSUE_NUMBER}" export OWNER REPO ISSUE_NUMBER TASK_ID +# LSV baseline cache creds + resource key (datasmith's own Supabase). Baked at +# render time; empty unless the runner enabled the cache. lsv_cache_writeback.py +# below reads these. Harbor's [verifier.env] also carries them into test.sh, but +# baking keeps setup.sh and test.sh consistent. +{%- for k, v in render_env.items() %} +export {{ k }}={{ v | shell_quote }} +{%- endfor %} + AGENT_MODEL_NAME_INPUT="${1:-agent}" AGENT_KEY="$(printf '%s' "${AGENT_MODEL_NAME_INPUT}" | tr '[:upper:]' '[:lower:]' | sed -E 's/[^a-z0-9_-]+/-/g; s/^-+//; s/-+$//')" if [ -z "${AGENT_KEY}" ]; then @@ -139,6 +147,16 @@ PYEOF echo "[$(ts)] [test] Computing reward..." python /opt/lsv/parser.py --owner "${OWNER}" --repo "${REPO}" --issue-number "${ISSUE_NUMBER}" --agent-key "${AGENT_KEY}" --base-commit "{{ base_commit }}" +# ── LSV baseline cache writeback (oracle only) ─────────────────────────── +# Persist the freshly measured baselines to datasmith's Supabase so the next +# trial on the same (task, hardware) hits and skips the timing pass. The +# script self-gates on HARBOR_AGENT_NAME and the DATASMITH_* creds; the shell +# guard mirrors the oracle_flag pattern below. Non-fatal. +if [ "${AGENT_KEY}" = "oracle" ]; then + echo "[$(ts)] [test] LSV cache writeback..." + python /opt/lsv/lsv_cache_writeback.py || echo "WARNING: lsv cache writeback failed" +fi + # ── Upload to Supabase (if configured) ─────────────────────────────────── if [ -n "${SUPABASE_URL:-}" ] && [ -n "${SUPABASE_ANON_KEY:-}" ] && [ -z "${FORMULACODE_NO_UPLOAD:-}" ]; then echo "[$(ts)] [test] Uploading to Supabase..." diff --git a/src/datasmith/harbor_adapter/utils.py b/src/datasmith/harbor_adapter/utils.py index bb2bc879..0b61f616 100644 --- a/src/datasmith/harbor_adapter/utils.py +++ b/src/datasmith/harbor_adapter/utils.py @@ -2,6 +2,7 @@ import json import os +import shlex from pathlib import Path from textwrap import dedent @@ -19,6 +20,10 @@ autoescape=False, keep_trailing_newline=True, ) +# Used by the render_env export blocks in setup.sh/test.sh: bake host-supplied +# values (cache creds + LSV_* key fields) as shell-safe `export K=V` lines, +# because Harbor's [verifier.env] does not reach setup.sh on Daytona. +_TEMPLATE_ENV.filters["shell_quote"] = shlex.quote # ========== Harbor-style template helpers ========== @@ -151,8 +156,13 @@ def render_test_sh( issue_number: int, run_pytest: bool = True, rounds: int = DATASMITH_LSV_ROUNDS, + render_env: dict[str, str] | None = None, ) -> str: - """Render test.sh for phase orchestration and verifier execution.""" + """Render test.sh for phase orchestration and verifier execution. + + ``render_env`` is baked as ``export K=V`` lines so the LSV cache creds and + key fields reach the verifier process; empty by default (no exports). + """ return _render_template( "test.sh", base_commit=base_commit, @@ -161,6 +171,7 @@ def render_test_sh( issue_number=issue_number, run_pytest=run_pytest, rounds=rounds, + render_env=render_env or {}, ) @@ -176,8 +187,14 @@ def render_run_setup_sh( issue_number: int, rounds: int = DATASMITH_LSV_ROUNDS, extra_setup_commands: str = "", + render_env: dict[str, str] | None = None, ) -> str: - """Render setup.sh with task metadata and extra setup commands.""" + """Render setup.sh with task metadata and extra setup commands. + + ``render_env`` is baked as ``export K=V`` lines so lsv_init.py (which runs + from setup.sh, where Harbor's [verifier.env] does not reach on Daytona) sees + the LSV cache creds and key fields; empty by default (no exports). + """ return _render_template( "setup.sh", owner=owner, @@ -185,4 +202,5 @@ def render_run_setup_sh( issue_number=issue_number, rounds=rounds, extra_setup_commands=extra_setup_commands or "", + render_env=render_env or {}, ) diff --git a/src/datasmith/resolution/git_utils.py b/src/datasmith/resolution/git_utils.py index f212eddc..af8731a6 100644 --- a/src/datasmith/resolution/git_utils.py +++ b/src/datasmith/resolution/git_utils.py @@ -12,9 +12,8 @@ from pathlib import Path from typing import Any, cast -from git import Commit, Repo - from datasmith.utils import get_logger +from git import Commit, Repo from .constants import ASV_REGEX, GIT_CACHE_DIR diff --git a/src/datasmith/resolution/orchestrator.py b/src/datasmith/resolution/orchestrator.py index b84e9962..81496426 100644 --- a/src/datasmith/resolution/orchestrator.py +++ b/src/datasmith/resolution/orchestrator.py @@ -31,9 +31,9 @@ from typing import Any import json5 -from git import Commit from datasmith.utils import get_logger +from git import Commit from .cache import cache_completion from .constants import CACHE_LOCATION diff --git a/src/datasmith/runners/harbor_healthcheck.py b/src/datasmith/runners/harbor_healthcheck.py index 45399ff2..e6a0883a 100644 --- a/src/datasmith/runners/harbor_healthcheck.py +++ b/src/datasmith/runners/harbor_healthcheck.py @@ -14,6 +14,8 @@ import json import os +import socket +import subprocess import uuid from pathlib import Path from typing import Any @@ -27,6 +29,41 @@ MIN_SPEEDUP_GATE = 1.05 # mirrored by publish/records.py +# ── LSV baseline cache knobs ──────────────────────────────────────────────── +# Master switch. Off => the runner injects no cache creds/key, so lsv_init falls +# back to force=True and behaves exactly as before the cache existed. +DATASMITH_LSV_CACHE_ENABLED: bool = os.environ.get("DATASMITH_LSV_CACHE_ENABLED", "1").strip().lower() not in ( + "0", + "false", + "no", +) + +# Cache-key fields for where a trial ran. docker keys on the host id (baselines +# don't transfer between dev hosts); daytona keys on the machine_class (plus the +# in-sandbox detected_cpu_model, since one class spans several EPYC SKUs). +DATASMITH_DOCKER_HOST_ID: str = os.environ.get("DATASMITH_DOCKER_HOST_ID", socket.gethostname()) +DATASMITH_DAYTONA_MACHINE_CLASS: str = os.environ.get("DATASMITH_DAYTONA_MACHINE_CLASS", "default") + +# Trial cgroup pins. These are BOTH written to the trial (task.toml cpus + +# EnvironmentConfig override_memory_mb) AND recorded as the cache-key +# cpu_count/mem_bytes, so the key describes the hardware the trial actually got. +# Memory defaults to 32 GB on both environments -- matching the pre-cache +# hardcoded pin (lsv_init OOMs large repos like sklearn under 4 GB); lower it +# deliberately per environment if needed. +DATASMITH_HARBOR_TRIAL_CPUS_DOCKER: int = int(os.environ.get("DATASMITH_HARBOR_TRIAL_CPUS_DOCKER", "2")) +DATASMITH_HARBOR_TRIAL_CPUS_DAYTONA: int = int(os.environ.get("DATASMITH_HARBOR_TRIAL_CPUS_DAYTONA", "2")) +DATASMITH_HARBOR_TRIAL_MEMORY_MB_DOCKER: int = int(os.environ.get("DATASMITH_HARBOR_TRIAL_MEMORY_MB_DOCKER", "32768")) +DATASMITH_HARBOR_TRIAL_MEMORY_MB_DAYTONA: int = int(os.environ.get("DATASMITH_HARBOR_TRIAL_MEMORY_MB_DAYTONA", "32768")) + + +def _trial_pins(use_daytona: bool) -> tuple[int, int]: + """(cpu_count, mem_mb) the runner pins for a trial in this environment. The + single authority for both the actual pin and the cache key, so they cannot + drift apart.""" + if use_daytona: + return DATASMITH_HARBOR_TRIAL_CPUS_DAYTONA, DATASMITH_HARBOR_TRIAL_MEMORY_MB_DAYTONA + return DATASMITH_HARBOR_TRIAL_CPUS_DOCKER, DATASMITH_HARBOR_TRIAL_MEMORY_MB_DOCKER + def _patch_harbor_trial_name() -> None: """Suppress the 7-char random suffix Harbor appends to every trial_name. @@ -85,11 +122,140 @@ def _build_verifier_env() -> dict[str, str]: return env +def _build_base_verifier_env() -> dict[str, str]: + """Creds for datasmith's OWN Supabase (the LSV cache's home), prefixed + ``DATASMITH_`` so they never collide with the Harbor ``SUPABASE_*`` names the + container already carries for upload.py. + + Returns ``{}`` when the service key is absent, which disables the cache. The + URL must be reachable from inside the trial container: a Daytona (or remote + docker) sandbox cannot reach a ``127.0.0.1`` instance, so the cache only + works when ``SUPABASE_URL`` is the ``db.formulacode.org`` tunnel (see + CLAUDE.md remote-access); a localhost URL simply yields cache misses. + """ + url = os.environ.get("SUPABASE_URL", "") + key = os.environ.get("SUPABASE_KEY", "") + if not url or not key: + return {} + env = { + "DATASMITH_SUPABASE_URL": url, + "DATASMITH_SUPABASE_SERVICE_KEY": key, + } + for k in ("DATASMITH_CF_ACCESS_CLIENT_ID", "DATASMITH_CF_ACCESS_CLIENT_SECRET"): + v = os.environ.get(k) + if v: + env[k] = v + return env + + +def _resolve_image_digest(container_name: str | None) -> str: + """Best-effort stable digest for the task image, so a rebuild under the same + name (different deps -> different timings) does not reuse a stale baseline. + Falls back to the container_name (still stable per image); never raises.""" + if not container_name: + return "" + try: + out = subprocess.run( + ["docker", "image", "inspect", "--format", "{{index .RepoDigests 0}}", container_name], + capture_output=True, + text=True, + timeout=30, + ) + digest = (out.stdout or "").strip() + if "@" in digest: + return "manifest:" + digest.split("@", 1)[1][:32] + except Exception as exc: + logger.debug("image digest lookup failed for %s (%s); using container_name", container_name, exc) + return container_name + + +def _compute_resource_attrs( + *, + use_daytona: bool, + container_name: str | None, + image_digest: str, + cpu_count: int, + mem_mb: int, +) -> dict[str, str]: + """The runner-supplied portion of the lsv_baseline_cache key -- every column + except ``detected_cpu_model``, which lsv_init reads from /proc inside the + sandbox. Values are strings for baking into shell exports; the in-container + code coerces cpu_count/mem_bytes back to int.""" + return { + "env": "daytona" if use_daytona else "docker", + "container_name": container_name or "", + "image_digest": image_digest, + "machine_class": DATASMITH_DAYTONA_MACHINE_CLASS if use_daytona else "", + "docker_host_id": "" if use_daytona else DATASMITH_DOCKER_HOST_ID, + "cpu_count": str(cpu_count), + "mem_bytes": str(mem_mb * 1024 * 1024), + } + + +def _decode_bytea(value: Any) -> bytes | None: + """Decode a PostgREST bytea payload (PostgreSQL hex form ``\\x...``) to raw + bytes. Returns None on anything unexpected -- the caller degrades to a miss.""" + if not isinstance(value, str) or not value.startswith("\\x"): + return None + try: + return bytes.fromhex(value[2:]) + except ValueError: + return None + + +def _fetch_deps_db(owner: str, repo: str, issue_number: int) -> bytes | None: + """Fetch the pre-surveyed LSV deps DB for a task from ``lsv_deps_cache`` on + datasmith's own Supabase, or None on miss / any error. + + Host-side over ``get_client()`` (not the in-container urllib path): the survey + is resource-independent, so it is fetched once here and baked into the image, + which works even when SUPABASE_URL is localhost (the container's baseline + fetch still needs the tunnel, but that is a separate layer). Best-effort: + every failure returns None and the trial re-runs the survey under force=True. + """ + try: + resp = ( + get_client() + .table("lsv_deps_cache") + .select("deps_db") + .eq("owner", owner) + .eq("repo", repo) + .eq("issue_number", issue_number) + .limit(1) + .execute() + ) + except Exception as exc: + logger.debug("lsv_deps_cache lookup failed for %s/%s#%d (%s)", owner, repo, issue_number, exc) + return None + rows = resp.data or [] + if not rows or not isinstance(rows[0], dict): + return None + return _decode_bytea(rows[0].get("deps_db")) + + +def _lsv_render_env(base_env: dict[str, str], *, task_id: str, attrs: dict[str, str]) -> dict[str, str]: + """Combine the datasmith creds with the LSV_* cache key into the dict baked + into setup.sh/test.sh. lsv_init reads LSV_* + creds to look baselines up; + lsv_cache_writeback reads them to upsert.""" + return { + **base_env, + "LSV_TASK_ID": task_id, + "LSV_ENV": attrs["env"], + "LSV_CONTAINER_NAME": attrs["container_name"], + "LSV_IMAGE_DIGEST": attrs["image_digest"], + "LSV_MACHINE_CLASS": attrs["machine_class"], + "LSV_DOCKER_HOST_ID": attrs["docker_host_id"], + "LSV_CPU_COUNT": attrs["cpu_count"], + "LSV_MEM_BYTES": attrs["mem_bytes"], + } + + def _materialize_tasks( items: list[dict[str, Any]], task_dir: Path, *, rounds: int, + use_daytona: bool, ) -> dict[str, dict[str, Any]]: """Write one Harbor task directory per PR. Returns a mapping from the Harbor task directory name back to the datasmith row metadata we need @@ -97,6 +263,17 @@ def _materialize_tasks( adapter = FormulaCodeAdapter(harbor_tasks_root=task_dir, force=True) verifier_env = _build_verifier_env() or None + # LSV baseline cache: on only when enabled AND datasmith creds are present. + # cpu_count/mem_mb are the pins the trial gets (see _build_job_config) and + # double as the cache-key hardware fields. + base_env = _build_base_verifier_env() + cache_on = DATASMITH_LSV_CACHE_ENABLED and bool(base_env) + cpu_count, mem_mb = _trial_pins(use_daytona) + if cache_on: + logger.info("LSV baseline cache enabled (env=%s)", "daytona" if use_daytona else "docker") + elif DATASMITH_LSV_CACHE_ENABLED: + logger.info("LSV baseline cache idle: SUPABASE_URL/SUPABASE_KEY not set") + # Per-task operator declarations. expected_n is the producer for the # dilution_ratio invariant; the trial container cannot read the table # itself (RLS-locked, no anon grant), so it is injected per task via @@ -123,13 +300,37 @@ def _materialize_tasks( pr.get("issue_number"), ) continue + render_env: dict[str, str] | None = None + deps_db: bytes | None = None + if cache_on: + attrs = _compute_resource_attrs( + use_daytona=use_daytona, + container_name=pr.get("container_name"), + image_digest=_resolve_image_digest(pr.get("container_name")), + cpu_count=cpu_count, + mem_mb=mem_mb, + ) + render_env = _lsv_render_env(base_env, task_id=rec.task_dir_name, attrs=attrs) + deps_db = _fetch_deps_db(rec.owner, rec.repo, rec.issue_number) + if deps_db: + logger.info( + "LSV survey cache hit for %s/%s#%d (%d bytes)", + rec.owner, + rec.repo, + rec.issue_number, + len(deps_db), + ) try: adapter.generate_task( rec, run_pytest=True, rounds=rounds, + cpus=cpu_count, + memory=f"{mem_mb}M", verifier_env=verifier_env, expected_n=expected_n_for(overrides, (rec.owner, rec.repo, rec.issue_number)), + render_env=render_env, + deps_db=deps_db, ) except Exception: logger.exception("generate_task failed for %s/%s#%d", rec.owner, rec.repo, rec.issue_number) @@ -164,10 +365,10 @@ def _build_job_config( # Harbor defaults to 4 GB per trial container (per the task.toml template) # which is too tight for lsv_init on mid/large Python repos — sklearn's - # dep-graph walk alone exceeds 4 GB and gets OOM-killed with exit 137. - # Bump to 32 GB across the board; the host has 500 GB so there's plenty - # of headroom, and smaller repos won't actually use more than they need. - MEMORY_MB = 32 * 1024 + # dep-graph walk alone exceeds 4 GB and gets OOM-killed with exit 137. The + # pin comes from _trial_pins so it stays identical to the cache-key + # mem_bytes; the default is still 32 GB. + _, MEMORY_MB = _trial_pins(use_daytona) if use_daytona: environment = EnvironmentConfig( type=EnvironmentType.DAYTONA, @@ -427,7 +628,7 @@ async def run_harbor_healthcheck( if job_name is None: job_name = f"fc-healthcheck-{uuid.uuid4().hex[:8]}" - task_id_map = _materialize_tasks(items, task_dir, rounds=rounds) + task_id_map = _materialize_tasks(items, task_dir, rounds=rounds, use_daytona=use_daytona) if not task_id_map: logger.warning("No tasks materialized — skipping Harbor dispatch") return [] diff --git a/supabase/migrations/00031_lsv_baseline_cache.sql b/supabase/migrations/00031_lsv_baseline_cache.sql new file mode 100644 index 00000000..f6826ea0 --- /dev/null +++ b/supabase/migrations/00031_lsv_baseline_cache.sql @@ -0,0 +1,75 @@ +-- Resource-keyed cache of LSV baseline timings, so stage 7 (harbor_healthcheck) +-- stops re-measuring the base-commit baseline on every trial. +-- +-- WHY THIS EXISTS +-- +-- Stage 7 measures a speedup with LSV (`asv.contrib.lightspeed`). Every trial +-- calls `initialize_diffcheck`, which runs two passes: a coverage SURVEY (which +-- source files each benchmark touches) and a BASELINE TIMING pass (how fast each +-- impacted benchmark runs at the base commit). Today `lsv_init.py` passes +-- `force=True`, so both passes run from scratch on every trial -- and a task is +-- benchmarked many times (oracle re-runs x agents x models x concurrency). The +-- timing pass dominates that cost. +-- +-- `initialize_diffcheck(force=False)` short-circuits both passes when the deps DB +-- already holds the survey AND a baseline row for every benchmark (verified in +-- the LSV fork, asv/contrib/lightspeed/session.py). This table is the durable, +-- cross-trial home for those baseline rows. The oracle trial populates it; every +-- trial reads it. `session.export_baselines()` produces the exact JSON stored in +-- `baselines` -> `{benchmark_id: {median, ci_99_a, ci_99_b, q_25, q_75, repeat, +-- number}}`. +-- +-- WHY THE KEY IS THIS WIDE +-- +-- A baseline timing is only valid on the hardware it was measured on: replaying a +-- fast-CPU baseline against a slow-CPU patched run fabricates a speedup and +-- silently corrupts the signal. The primary key therefore pins every resource +-- fact that moves a timing: +-- * (owner, repo, issue_number) -- the task +-- * env -- 'docker' | 'daytona' +-- * container_name, image_digest -- the exact image (a rebuild changes deps -> timings) +-- * machine_class, docker_host_id -- where it ran (daytona class / docker host) +-- * cpu_count, mem_bytes -- the cgroup pins the runner REQUESTED (not /proc) +-- * detected_cpu_model -- the CPU model read from /proc/cpuinfo INSIDE +-- the sandbox; daytona's 'default' class fans one +-- (cpu,mem) request across several EPYC SKUs, so +-- this is the only field that distinguishes them. +-- Matching is exact equality on all 11 columns, so an over-coarse key is +-- impossible; the only failure mode is under-hitting, which is safe (the trial +-- falls back to a fresh `force=True` measure). Every key column is NOT NULL with +-- an empty-string / zero default so docker rows (machine_class='') and daytona +-- rows (docker_host_id='') coexist without NULL-PK gaps, and lookup vs upsert can +-- never disagree on '' vs NULL. +-- +-- No FK to pull_requests: this cache is advisory and is also written from the +-- stage-6 measure.sh context, where no PR row is guaranteed to exist. A missing +-- task simply never gets a cache hit. + +CREATE TABLE IF NOT EXISTS lsv_baseline_cache ( + owner TEXT NOT NULL, + repo TEXT NOT NULL, + issue_number INT NOT NULL, + env TEXT NOT NULL DEFAULT '', -- 'docker' | 'daytona' + container_name TEXT NOT NULL DEFAULT '', + image_digest TEXT NOT NULL DEFAULT '', + machine_class TEXT NOT NULL DEFAULT '', + docker_host_id TEXT NOT NULL DEFAULT '', + cpu_count INT NOT NULL DEFAULT 0, + mem_bytes BIGINT NOT NULL DEFAULT 0, + detected_cpu_model TEXT NOT NULL DEFAULT '', + baselines JSONB NOT NULL, + created_at TIMESTAMPTZ DEFAULT NOW(), + updated_at TIMESTAMPTZ DEFAULT NOW(), + PRIMARY KEY (owner, repo, issue_number, env, container_name, image_digest, + machine_class, docker_host_id, cpu_count, mem_bytes, detected_cpu_model) +); + +-- The read path filters by the task first, so index the task prefix. The full PK +-- already backs exact-match lookups on every column. +CREATE INDEX IF NOT EXISTS idx_lsv_baseline_cache_task + ON lsv_baseline_cache (owner, repo, issue_number); + +-- Read-only role for Grafana, following 00009. No anon grant: this is internal +-- pipeline state, private by default per 00015. Pipeline writes use the +-- service-role key, which bypasses grants and RLS. +GRANT SELECT ON lsv_baseline_cache TO grafana_ro; diff --git a/supabase/migrations/00032_lsv_deps_cache.sql b/supabase/migrations/00032_lsv_deps_cache.sql new file mode 100644 index 00000000..489cdfaf --- /dev/null +++ b/supabase/migrations/00032_lsv_deps_cache.sql @@ -0,0 +1,51 @@ +-- Task-keyed cache of the LSV coverage SURVEY (the `.lightspeed_deps.db` +-- SQLite file, baselines stripped), so stage 7 (harbor_healthcheck) stops +-- re-running the survey pass on every trial. +-- +-- WHY THIS EXISTS, AND WHY IT IS SEPARATE FROM lsv_baseline_cache (00031) +-- +-- `initialize_diffcheck` runs two passes: a coverage SURVEY (which source file +-- each benchmark touches) and a BASELINE TIMING pass. It short-circuits BOTH +-- only when the deps DB already holds the survey AND a baseline row for every +-- benchmark (verified in the LSV fork, asv/contrib/lightspeed/session.py) -- +-- and `load_baselines` REQUIRES the surveyed deps DB to already exist on disk, +-- so the baseline cache (00031) is useless without the survey staged first. +-- +-- The two facts have different lifetimes and different keys, hence two tables: +-- * The SURVEY is resource-INDEPENDENT -- it depends only on the code and the +-- benchmark suite at a given commit, not on the CPU. So it is keyed by the +-- task alone, (owner, repo, issue_number), and one row serves docker and +-- every daytona SKU alike. +-- * The BASELINE timings are resource-DEPENDENT and live in lsv_baseline_cache +-- under an 11-column hardware key. +-- Keeping the multi-MB survey blob in its own table also keeps it off +-- pull_requests and clear of the DATASMITH_LARGE_* unfiltered-read guard. +-- +-- TRANSPORT. The blob is a binary SQLite file, stored as BYTEA. Both the +-- in-container writeback (stdlib urllib) and the host-side runner (supabase-py) +-- move it through PostgREST as PostgreSQL's hex form (`\x...`): the writeback +-- POSTs `"\\x" + data.hex()`, the runner reads the `\x`-prefixed string back and +-- `bytes.fromhex`es it. No Storage bucket -- datasmith host-side uses none, and a +-- bucket would add client.storage plumbing and bucket RLS for no gain here. +-- +-- The oracle trial writes the survey (baselines DELETEd + `PRAGMA +-- wal_checkpoint(TRUNCATE)` so no per-host timing rows leak across hardware); +-- the runner reads it and bakes it into the next build. A miss anywhere degrades +-- to today's `force=True` full measure. No FK to pull_requests: advisory cache, +-- also written from the stage-6 measure.sh context where no PR row is +-- guaranteed. A missing task simply never gets a hit. + +CREATE TABLE IF NOT EXISTS lsv_deps_cache ( + owner TEXT NOT NULL, + repo TEXT NOT NULL, + issue_number INT NOT NULL, + deps_db BYTEA NOT NULL, + created_at TIMESTAMPTZ DEFAULT NOW(), + updated_at TIMESTAMPTZ DEFAULT NOW(), + PRIMARY KEY (owner, repo, issue_number) +); + +-- Read-only role for Grafana, following 00009/00031. No anon grant: internal +-- pipeline state, private by default per 00015. Pipeline writes use the +-- service-role key, which bypasses grants and RLS. +GRANT SELECT ON lsv_deps_cache TO grafana_ro; diff --git a/tests/docker/test_lsv_cache.py b/tests/docker/test_lsv_cache.py new file mode 100644 index 00000000..054487e1 --- /dev/null +++ b/tests/docker/test_lsv_cache.py @@ -0,0 +1,180 @@ +"""Unit tests for the stage-7 LSV cache (baseline + survey) wiring. + +These exercise the pure host-side/in-container glue that does not need Harbor, +Docker, or a live Supabase: the resource-key mapping, the bytea transport, the +render_env shell escaping, the task-id parsing shared by reader and writer, and +-- most importantly -- that the on_conflict column lists in the in-container +writeback match the primary keys the migrations actually declare. A drift there +silently disables every upsert (PostgREST 400s, writeback swallows it), so it is +asserted directly against the SQL. +""" + +from __future__ import annotations + +import importlib.util +import re +import shlex +from pathlib import Path +from types import ModuleType + +import pytest + +from datasmith.harbor_adapter.adapter import FormulaCodeAdapter, HarborTaskPaths +from datasmith.harbor_adapter.utils import render_run_setup_sh, render_test_sh +from datasmith.runners.harbor_healthcheck import ( + _compute_resource_attrs, + _decode_bytea, + _lsv_render_env, +) + +_ROOT = Path(__file__).parents[2] +_TEMPLATE = _ROOT / "src" / "datasmith" / "harbor_adapter" / "template" +_MIGRATIONS = _ROOT / "supabase" / "migrations" + + +def _load_template_module(name: str) -> ModuleType: + """Import a stdlib-only template script by path (the template dir is not a + package).""" + spec = importlib.util.spec_from_file_location(f"_tmpl_{name}", _TEMPLATE / f"{name}.py") + assert spec and spec.loader + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def _pk_columns(migration_filename: str) -> list[str]: + """Extract the PRIMARY KEY column list from a migration's CREATE TABLE.""" + sql = (_MIGRATIONS / migration_filename).read_text() + m = re.search(r"PRIMARY KEY\s*\(([^)]*)\)", sql, re.IGNORECASE | re.DOTALL) + assert m, f"no PRIMARY KEY found in {migration_filename}" + return [c.strip() for c in m.group(1).split(",") if c.strip()] + + +# ── resource key ──────────────────────────────────────────────────────────── + + +def test_compute_resource_attrs_daytona_keys_on_machine_class() -> None: + attrs = _compute_resource_attrs( + use_daytona=True, container_name="img:1", image_digest="manifest:ab", cpu_count=2, mem_mb=32768 + ) + assert attrs["env"] == "daytona" + assert attrs["machine_class"] == "default" + assert attrs["docker_host_id"] == "" # daytona keys on class, not host + assert attrs["cpu_count"] == "2" + assert attrs["mem_bytes"] == str(32768 * 1024 * 1024) + + +def test_compute_resource_attrs_docker_keys_on_host(monkeypatch: pytest.MonkeyPatch) -> None: + attrs = _compute_resource_attrs( + use_daytona=False, container_name="img:1", image_digest="", cpu_count=4, mem_mb=8192 + ) + assert attrs["env"] == "docker" + assert attrs["machine_class"] == "" # docker keys on host, not class + assert attrs["docker_host_id"] # gethostname() default, non-empty + assert attrs["cpu_count"] == "4" + + +def test_lsv_render_env_carries_key_and_creds() -> None: + base = {"DATASMITH_SUPABASE_URL": "https://db", "DATASMITH_SUPABASE_SERVICE_KEY": "k"} + attrs = _compute_resource_attrs(use_daytona=True, container_name="c", image_digest="d", cpu_count=2, mem_mb=1024) + env = _lsv_render_env(base, task_id="o__r__7", attrs=attrs) + assert env["DATASMITH_SUPABASE_URL"] == "https://db" + assert env["LSV_TASK_ID"] == "o__r__7" + assert env["LSV_ENV"] == "daytona" + assert env["LSV_MEM_BYTES"] == str(1024 * 1024 * 1024) + + +# ── bytea transport ───────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("payload", [b"", b"hello", bytes(range(256)), b"\x00\x01\x02sqlite"]) +def test_decode_bytea_roundtrip(payload: bytes) -> None: + assert _decode_bytea("\\x" + payload.hex()) == payload + + +@pytest.mark.parametrize("bad", [None, "", "deadbeef", 42, "\\xzz", b"\\x00"]) +def test_decode_bytea_rejects_non_hex(bad: object) -> None: + assert _decode_bytea(bad) is None + + +# ── render_env shell escaping ─────────────────────────────────────────────── + + +def _exports_in(script: str) -> dict[str, str]: + """Parse `export K=V` lines the render_env block emits, unquoting V.""" + out: dict[str, str] = {} + for line in script.splitlines(): + m = re.match(r"^export (LSV_[A-Z_]+|DATASMITH_[A-Z_]+)=(.*)$", line) + if m: + out[m.group(1)] = "".join(shlex.split(m.group(2))) + return out + + +def test_render_env_roundtrips_through_shell_quote() -> None: + tricky = { + "LSV_CONTAINER_NAME": "repo name with spaces", + "LSV_IMAGE_DIGEST": "manifest:a'b$c;rm -rf /", + "DATASMITH_SUPABASE_URL": "https://db.formulacode.org", + } + for script in ( + render_run_setup_sh(owner="o", repo="r", issue_number=1, render_env=tricky), + render_test_sh(base_commit="abc", owner="o", repo="r", issue_number=1, render_env=tricky), + ): + parsed = _exports_in(script) + for k, v in tricky.items(): + assert parsed[k] == v, f"{k} did not survive shell_quote round-trip" + + +def test_render_env_none_and_empty_are_identical_and_bare() -> None: + none_setup = render_run_setup_sh(owner="o", repo="r", issue_number=1, render_env=None) + empty_setup = render_run_setup_sh(owner="o", repo="r", issue_number=1, render_env={}) + assert none_setup == empty_setup + assert "LSV_TASK_ID" not in none_setup # no cache exports when disabled + + +# ── on_conflict / PK agreement (the drift that silently kills upserts) ─────── + + +def test_baseline_writeback_on_conflict_matches_migration_pk() -> None: + writeback = _load_template_module("lsv_cache_writeback") + assert list(writeback._BASELINE_PK_COLS) == _pk_columns("00031_lsv_baseline_cache.sql") + + +def test_deps_cache_on_conflict_matches_migration_pk() -> None: + writeback = _load_template_module("lsv_cache_writeback") + src = (_TEMPLATE / "lsv_cache_writeback.py").read_text() + m = re.search(r"lsv_deps_cache\?on_conflict=([a-z_,]+)", src) + assert m, "deps_cache upsert on_conflict not found" + assert m.group(1).split(",") == _pk_columns("00032_lsv_deps_cache.sql") + # touch the module so an import error here fails loudly too + assert hasattr(writeback, "_upsert_deps_cache_row") + + +# ── task-id parsing is shared by reader (lsv_init) and writer (writeback) ───── + + +@pytest.mark.parametrize("mod_name", ["lsv_init", "lsv_cache_writeback"]) +def test_parse_task_id_double_underscore(mod_name: str) -> None: + mod = _load_template_module(mod_name) + assert mod._parse_task_id("pandas-dev__pandas__123") == ("pandas-dev", "pandas", 123) + assert mod._parse_task_id("o__r__notanint") is None + assert mod._parse_task_id("nounderscores") is None + + +# ── adapter bakes cache/ so the Dockerfile COPY never fails ────────────────── + + +def test_write_cache_files_miss_leaves_only_gitkeep(tmp_path: Path) -> None: + adapter = FormulaCodeAdapter(harbor_tasks_root=tmp_path) + paths = HarborTaskPaths(tmp_path / "task") + adapter._write_cache_files(paths, None) + cache = paths.environment_dir / "cache" + assert (cache / ".gitkeep").exists() + assert not (cache / "lightspeed_deps.db").exists() + + +def test_write_cache_files_hit_writes_deps_db(tmp_path: Path) -> None: + adapter = FormulaCodeAdapter(harbor_tasks_root=tmp_path) + paths = HarborTaskPaths(tmp_path / "task") + adapter._write_cache_files(paths, b"SQLite format 3\x00survey") + assert (paths.environment_dir / "cache" / "lightspeed_deps.db").read_bytes() == b"SQLite format 3\x00survey" diff --git a/tests/resolution/test_no_dependency_harvest.py b/tests/resolution/test_no_dependency_harvest.py index 38a654c9..07fd7977 100644 --- a/tests/resolution/test_no_dependency_harvest.py +++ b/tests/resolution/test_no_dependency_harvest.py @@ -19,12 +19,12 @@ from pathlib import Path import pytest -from git import Actor, Repo from datasmith.resolution import metadata_parser from datasmith.resolution.declare import declare from datasmith.resolution.metadata_parser import analyze_candidate_meta, discover_candidates from datasmith.resolution.models import Candidate +from git import Actor, Repo PYPROJECT = """\ [project]