fix(run6.3): smoke-found bench harness fixes (7 schema/protocol bugs)

scripts/bench.py

@@ -45,6 +45,13 @@
 
 import httpx
 
+# When invoked as `uv run python scripts/bench.py`, sys.path[0] is `scripts/` (script dir),
+# so `from scripts.bench_ingest import ...` fails because the parent (repo root) is not on path.
+# Add it here so the `scripts.*` package imports inside functions resolve correctly.
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
+
 # ---------------------------------------------------------------------------
 # Constants
 # ---------------------------------------------------------------------------
@@ -128,9 +135,13 @@ def _env_summary() -> dict[str, str]:
     return result
 
 
-def _docker_compose(args: list[str], timeout: int = 120) -> subprocess.CompletedProcess[str]:
+def _docker_compose(
+    args: list[str],
+    timeout: int = 120,
+    env: dict[str, str] | None = None,
+) -> subprocess.CompletedProcess[str]:
     cmd = ["docker", "compose", *args]
-    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
+    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, env=env)
 
 
 # ---------------------------------------------------------------------------
@@ -195,9 +206,16 @@ def phase_preflight_reuse() -> int:
 def phase_up() -> int:
     """Start required docker compose services."""
     _log("Phase 2: docker compose up")
+    # Override compose-host env vars to compose-internal hostnames; .env's localhost
+    # values would otherwise leak into containers via compose variable interpolation.
+    compose_env = os.environ.copy()
+    compose_env["ALAYA_DATABASE_URL"] = "postgresql+asyncpg://alaya:alaya@postgres:5432/alaya"
+    compose_env["ALAYA_REDIS_URL"] = "redis://redis:6379/0"
+    compose_env["ALAYA_ENV"] = "dev"
     result = _docker_compose(
         ["up", "-d", "postgres", "redis", "migrations", "api", "worker"],
         timeout=120,
+        env=compose_env,
     )
     if result.returncode != 0:
         _log(f"FAIL: docker compose up failed:\n{result.stderr}")
@@ -253,10 +271,11 @@ def phase_bootstrap(deadline: float) -> tuple[int, str, str]:
 
     with httpx.Client(timeout=30.0) as client:
         # Create workspace via HTTP API (requires bootstrap key)
+        ts = datetime.now(UTC).strftime("%Y%m%d%H%M%S")
         r = client.post(
             f"{API_URL}/workspaces",
             headers={"X-Api-Key": bootstrap_key, "Content-Type": "application/json"},
-            json={"name": f"bench-{datetime.now(UTC).strftime('%Y%m%d%H%M%S')}"},
+            json={"name": f"bench-{ts}", "slug": f"bench-{ts}"},
         )
         if r.status_code not in (200, 201):
             _log(f"FAIL: workspace create returned {r.status_code}: {r.text[:200]}")

scripts/bench_report.py

@@ -26,6 +26,7 @@
 import statistics
 import uuid
 from datetime import datetime
+from decimal import Decimal
 from typing import Any
 
 from sqlalchemy import Connection, text
@@ -59,23 +60,26 @@ def _scalar(conn: Connection, sql: str, **params: Any) -> Any:
 
 
 # ---------------------------------------------------------------------------
-# Data queries (all scoped to workspace + created_at >= started_at)
+# Data queries (all scoped to workspace_id; temporal filter only on
+# pipeline_traces.created_at and integrator_runs.started_at where columns
+# are always populated)
 # ---------------------------------------------------------------------------
 
 
 def _get_extraction_runs(conn: Connection, workspace_id: uuid.UUID, started_at: str) -> list[dict[str, Any]]:
+    # No temporal filter: extraction_runs.started_at is nullable (never set in practice).
+    # Workspace isolation is sufficient — bench creates a fresh workspace per run.
     rows = _query(
         conn,
         """
         SELECT id, workspace_id, status, tokens_in, tokens_out, tokens_cached,
-               cost_usd, cortex_cost_usd, crystallizer_cost_usd, created_at, completed_at
+               cost_usd, entities_created, entities_merged, relations_created,
+               claims_created, claims_superseded, started_at, completed_at
         FROM extraction_runs
         WHERE workspace_id = :wid
-          AND created_at >= :started_at
-        ORDER BY created_at
+        ORDER BY completed_at NULLS LAST, id
         """,
         wid=str(workspace_id),
-        started_at=started_at,
     )
     return [_serialize_row(r) for r in rows]
 
@@ -85,12 +89,13 @@ def _get_integrator_runs(conn: Connection, workspace_id: uuid.UUID, started_at:
         conn,
         """
         SELECT id, workspace_id, status, tokens_used, cost_usd, duration_ms,
-               entities_scanned, entities_deduplicated, entities_merged,
-               entities_enriched, created_at, completed_at
+               entities_scanned, entities_deduplicated, entities_enriched,
+               relations_created, claims_updated, noise_removed,
+               started_at, completed_at
         FROM integrator_runs
         WHERE workspace_id = :wid
-          AND created_at >= :started_at
-        ORDER BY created_at
+          AND started_at >= :started_at
+        ORDER BY started_at
         """,
         wid=str(workspace_id),
         started_at=started_at,
@@ -118,13 +123,19 @@ def _get_pipeline_traces(conn: Connection, workspace_id: uuid.UUID, started_at:
 
 
 def _serialize_row(row: dict[str, Any]) -> dict[str, Any]:
-    """Convert non-JSON-serialisable types (UUID, datetime) to strings."""
+    """Convert non-JSON-serialisable types (UUID, datetime, Decimal) to native JSON types.
+
+    Postgres NUMERIC columns (cost_usd) come back as Decimal; convert to float for
+    manifest.json serialisation.
+    """
     out: dict[str, Any] = {}
     for k, v in row.items():
         if isinstance(v, uuid.UUID):
             out[k] = str(v)
         elif isinstance(v, datetime):
             out[k] = v.isoformat()
+        elif isinstance(v, Decimal):
+            out[k] = float(v)
         else:
             out[k] = v
     return out
@@ -141,29 +152,25 @@ def _description_rate(conn: Connection, workspace_id: uuid.UUID, started_at: str
     total = _scalar(
         conn,
         """
-        SELECT COUNT(*) FROM claims c
+        SELECT COUNT(*) FROM l2_claims c
         JOIN extraction_runs er ON er.id = c.extraction_run_id
         WHERE c.workspace_id = :wid
-          AND er.created_at >= :started_at
         """,
         wid=str(workspace_id),
-        started_at=started_at,
     )
     if not total:
         return None
 
     desc_count = _scalar(
         conn,
         f"""
-        SELECT COUNT(*) FROM claims c
-        JOIN predicates p ON p.id = c.predicate_id AND p.workspace_id = c.workspace_id
+        SELECT COUNT(*) FROM l2_claims c
+        JOIN predicate_definitions p ON p.id = c.predicate_id AND p.workspace_id = c.workspace_id
         JOIN extraction_runs er ON er.id = c.extraction_run_id
         WHERE c.workspace_id = :wid
-          AND er.created_at >= :started_at
           AND p.slug IN ({slugs_list})
         """,
         wid=str(workspace_id),
-        started_at=started_at,
     )
     return round((desc_count or 0) / total, 4)
 
@@ -173,18 +180,16 @@ def _claims_per_entity(conn: Connection, workspace_id: uuid.UUID, started_at: st
     claims_total = _scalar(
         conn,
         """
-        SELECT COUNT(*) FROM claims c
+        SELECT COUNT(*) FROM l2_claims c
         JOIN extraction_runs er ON er.id = c.extraction_run_id
         WHERE c.workspace_id = :wid
-          AND er.created_at >= :started_at
         """,
         wid=str(workspace_id),
-        started_at=started_at,
     )
     entities_total = _scalar(
         conn,
         """
-        SELECT COUNT(*) FROM entities
+        SELECT COUNT(*) FROM l1_entities
         WHERE workspace_id = :wid
           AND created_at >= :started_at
           AND is_deleted = false
@@ -204,13 +209,11 @@ def _claims_per_event_stddev(conn: Connection, workspace_id: uuid.UUID, started_
         """
         SELECT er.id AS run_id, COUNT(c.id) AS claim_count
         FROM extraction_runs er
-        LEFT JOIN claims c ON c.extraction_run_id = er.id
+        LEFT JOIN l2_claims c ON c.extraction_run_id = er.id
         WHERE er.workspace_id = :wid
-          AND er.created_at >= :started_at
         GROUP BY er.id
         """,
         wid=str(workspace_id),
-        started_at=started_at,
     )
     counts = [r["claim_count"] for r in rows]
     if len(counts) < 2:
@@ -224,9 +227,9 @@ def _dedup_actions(conn: Connection, workspace_id: uuid.UUID, started_at: str) -
         conn,
         """
         SELECT COUNT(*) FROM integrator_actions ia
-        JOIN integrator_runs ir ON ir.id = ia.integrator_run_id
+        JOIN integrator_runs ir ON ir.id = ia.run_id
         WHERE ia.workspace_id = :wid
-          AND ir.created_at >= :started_at
+          AND ir.started_at >= :started_at
           AND ia.action_type = 'merge'
         """,
         wid=str(workspace_id),
@@ -242,11 +245,9 @@ def _run_failure_count(conn: Connection, workspace_id: uuid.UUID, started_at: st
         """
         SELECT COUNT(*) FROM extraction_runs
         WHERE workspace_id = :wid
-          AND created_at >= :started_at
           AND status = 'failed'
         """,
         wid=str(workspace_id),
-        started_at=started_at,
     )
     return int(val or 0)