fix: address critical audit bugs and improve configuration robustess

docs/bugs.md

@@ -0,0 +1,213 @@
+This is a solid, well-structured repo. The architecture matches the spec perfectly and most of the code is production-quality. Here's the full audit — bugs first, then improvements.
+
+***
+
+## 🔴 Bugs (Will Break at Runtime)
+
+### Bug 1 — `asyncio.run()` inside an already-running event loop crashes
+**File:** `pipelineprobe/connectors/postgres.py` line `get_stats_sync()`
+
+`asyncio.run()` throws `RuntimeError: This event loop is already running` if PipelineProbe is ever called from an async context (FastAPI, Jupyter, any async test runner). This is a latent crash waiting to happen.
+
+**Fix:** Replace the async/sync split entirely. `asyncpg` is overkill for a one-shot audit query. Use `psycopg2` synchronously:
+
+```python
+import logging
+from typing import Any, Dict, List
+
+import psycopg2
+import psycopg2.extras
+
+from pipelineprobe.config import WarehouseConfig
+
+logger = logging.getLogger(__name__)
+
+class PostgresConnector:
+    def __init__(self, config: WarehouseConfig):
+        self.config = config
+
+    def get_stats_sync(self) -> List[Dict[str, Any]]:
+        try:
+            conn = psycopg2.connect(self.config.dsn)
+            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+                cur.execute("""
+                    SELECT
+                        t.schemaname,
+                        t.relname AS tablename,
+                        t.n_live_tup AS row_count,
+                        EXISTS (
+                            SELECT 1 FROM information_schema.columns c
+                            WHERE c.table_schema = t.schemaname
+                              AND c.table_name = t.relname
+                              AND c.column_name IN ('updated_at', 'created_at')
+                        ) AS has_timestamps
+                    FROM pg_stat_user_tables t
+                    ORDER BY t.n_live_tup DESC
+                    LIMIT 50
+                """)
+                return [dict(r) for r in cur.fetchall()]
+        except Exception as e:
+            logger.error("Error connecting to Postgres: %s", e)
+            return []
+        finally:
+            if conn:
+                conn.close()
+```
+
+Also update `pyproject.toml` — remove `asyncpg`, add `psycopg2-binary`.
+
+***
+
+### Bug 2 — `config.py`: `AirflowConfig` field is `base_url` but `init` command writes `url`
+**File:** `pipelineprobe/config.py` line 8 vs `cli.py` `init` command
+
+`AirflowConfig` defines `base_url` as the field name, but the `init` command generates a YAML stub with key `url`. When a user runs `pipelineprobe init` then `pipelineprobe audit`, the `base_url` field gets its default (`http://localhost:8080`) instead of their value because the YAML key doesn't match.
+
+**Fix in `cli.py` init command** — change the generated YAML:
+```yaml
+orchestrator:
+  base_url: "http://localhost:8080"   # ← was "url:"
+  username: "admin"
+```
+
+***
+
+### Bug 3 — `cli.py`: `postgres_tables` hardcoded context key doesn't work for BigQuery/Snowflake rules
+**File:** `pipelineprobe/cli.py` line 66
+
+```python
+"postgres_tables": warehouse_tables,  # Backwards compatible context key
+```
+
+The comment says "backwards compatible" but `postgres_rules.py` reads from `postgres_tables` key while BigQuery and Snowflake also return their data into the same key. This is fine for now — but the `postgres_rules.py` rule will fire on BigQuery/Snowflake data with misleading Postgres-specific recommendations (e.g. "consider partitioning" using Postgres terminology on a BigQuery table).
+
+**Fix:** Use a neutral key and pass warehouse type to the context:
+```python
+context = {
+    "airflow_dags": airflow_dags,
+    "airflow_tasks": airflow_tasks,
+    "dbt_models": dbt_models,
+    "warehouse_tables": warehouse_tables,   # ← neutral key
+    "warehouse_type": cfg.warehouse.type,   # ← pass type for rule conditions
+}
+```
+Update `postgres_rules.py` to read `warehouse_tables` and guard with `if context.get("warehouse_type") == "postgres"`.
+
+***
+
+### Bug 4 — `dbt.py`: manifest path is double-joined
+**File:** `pipelineprobe/connectors/dbt.py` lines 16–17
+
+```python
+manifest_path = project_dir / self.config.manifest_path
+```
+
+`DbtConfig` defaults:
+```python
+project_dir: str = "./analytics"
+manifest_path: str = "target/manifest.json"
+```
+
+So the resolved path becomes `./analytics/target/manifest.json`. But many users set `project_dir` to the dbt project root AND `manifest_path` to a full relative path from CWD like `./analytics/target/manifest.json`. This causes a double-join. The config field name `manifest_path` implies it's already a full path, not relative to `project_dir`.
+
+**Fix:** Make `manifest_path` and `run_results_path` absolute-or-CWD-relative, not relative to `project_dir`:
+```python
+manifest_path = Path(self.config.manifest_path)
+run_results_path = Path(self.config.run_results_path)
+```
+Update defaults in `DbtConfig` to `"./analytics/target/manifest.json"` to match the original intent, and update the generated `init` YAML accordingly.
+
+***
+
+### Bug 5 — `airflow_rules.py`: `check_stale_dags` uses `.days` which truncates — misses sub-day staleness
+**File:** `pipelineprobe/rules/airflow_rules.py` line 93
+
+```python
+days_since = (now - end_time_aware).days
+```
+
+`timedelta.days` is the integer day component only — it **does not round**. A DAG that last succeeded 6 days and 23 hours ago returns `.days == 6`, not 7, so it never triggers the `> 7` check. Use `total_seconds()` instead:
+
+```python
+days_since = (now - end_time_aware).total_seconds() / 86400
+if days_since > stale_threshold_days:
+```
+
+***
+
+## 🟡 Issues That Will Cause Confusion
+
+### Issue 1 — `pyproject.toml` likely missing `asyncpg` / `python-dateutil` / `snowflake-connector-python` as explicit deps
+The connectors import `asyncpg`, `dateutil`, and `snowflake.connector` but without seeing `pyproject.toml` contents fully, these are often missed. Verify the `[project.dependencies]` section includes:
+```toml
+dependencies = [
+  "typer>=0.9",
+  "httpx>=0.27",
+  "pydantic>=2.0",
+  "pydantic-settings>=2.0",
+  "pyyaml>=6.0",
+  "jinja2>=3.1",
+  "python-dateutil>=2.8",
+  "psycopg2-binary>=2.9",       # after fixing Bug 1
+  "snowflake-connector-python>=3.0",
+  "google-cloud-bigquery>=3.0",
+]
+```
+
+***
+
+### Issue 2 — `cli.py`: `fail_on_critical` default is `5` in config but `0` in `init` YAML
+`ReportConfig` defaults `fail_on_critical: int = 5` but the `init` command generates `fail_on_critical: 0`. A user who runs `init`, gets the YAML, and doesn't edit it will find the audit **always fails on any critical issue**. The defaults should be consistent — pick `5` in both places.
+
+***
+
+### Issue 3 — `renderer.py`: `report.html` template missing — `TemplateNotFound` at runtime
+The `renderer.py` loads `report.html` from `pipelineprobe/templates/` but the `templates/` directory exists — if `report.html` is empty or missing, every audit run fails at the last step with a Jinja2 `TemplateNotFound` error. Verify the template file exists and has actual HTML. If it's a stub, even a minimal one like this is enough to unblock users:
+
+```html
+<!DOCTYPE html>
+<html>
+<head><title>PipelineProbe Report</title></head>
+<body>
+  <h1>Score: {{ summary.score }}/100</h1>
+  <p>Critical: {{ summary.critical_count }} | Warnings: {{ summary.warning_count }}</p>
+  {% for issue in issues %}
+  <div class="{{ issue.severity }}">
+    <strong>{{ issue.summary }}</strong>
+    <p>{{ issue.recommendation }}</p>
+  </div>
+  {% endfor %}
+</body>
+</html>
+```
+
+***
+
+## ✅ What's Already Good
+
+| Area | Verdict |
+|------|---------|
+| Overall architecture (connectors / rules / renderer) | Clean, matches spec exactly |
+| Airflow pagination with `offset` loop | Correct |
+| Snowflake CTE approach (avoids correlated subquery restriction) | Smart fix |
+| dbt test count via `depends_on.nodes` traversal | Correct approach |
+| Timezone-aware `datetime.now(timezone.utc)` in rules | Correct |
+| `fail_on_critical` CLI override | Good UX |
+| Partial report on connector failure (returns `[]`) | Resilient |
+| Snowflake missing-credential guard | Correct |
+
+***
+
+## Priority Fix Order
+
+```
+1. Bug 1  — Replace asyncpg/asyncio.run with psycopg2 sync (crash risk)
+2. Bug 2  — Fix init YAML key url → base_url (silent misconfiguration)
+3. Bug 3  — Neutral warehouse_tables context key (wrong rules on wrong warehouse)
+4. Bug 4  — Fix dbt manifest double-join path (FileNotFoundError for all dbt users)
+5. Bug 5  — Use total_seconds() for stale DAG check (off-by-<1-day logic error)
+6. Issue 2 — Align fail_on_critical defaults (unexpected CI failures)
+7. Issue 3 — Confirm report.html template exists and renders
+```
+
+Fix bugs 1–4 before any public share or PyPI publish — they will hit every first-time user.

pipelineprobe/cli.py

@@ -73,7 +73,8 @@ def audit(
         "airflow_dags": airflow_dags,
         "airflow_tasks": airflow_tasks,
         "dbt_models": dbt_models,
-        "postgres_tables": warehouse_tables, # Backwards compatible context key
+        "warehouse_tables": warehouse_tables,
+        "warehouse_type": cfg.warehouse.type,
     }
 
     typer.echo("Running rule engine...")
@@ -123,13 +124,15 @@ def init():
     default_config = """# PipelineProbe Configuration
 
 orchestrator:
-  url: "http://localhost:8080"
+  base_url: "http://localhost:8080"
   username: "admin"
   # Set PIPELINEPROBE_AIRFLOW_PASSWORD in environment instead of hardcoding here
 
 dbt:
   project_dir: "./dbt"
   target: "dev"
+  manifest_path: "./dbt/target/manifest.json"
+  run_results_path: "./dbt/target/run_results.json"
 
 warehouse:
   # Set PIPELINEPROBE_WAREHOUSE_DSN in environment
@@ -138,7 +141,7 @@ def init():
 report:
   output_dir: "./reports"
   format: "both"
-  fail_on_critical: 0
+  fail_on_critical: 5
 """
     with open("pipelineprobe.yml", "w") as f:
         f.write(default_config)

pipelineprobe/config.py

@@ -14,8 +14,8 @@ class AirflowConfig(BaseModel):
 class DbtConfig(BaseModel):
     project_dir: str = "./analytics"
     target: str = "prod"
-    manifest_path: str = "target/manifest.json"
-    run_results_path: str = "target/run_results.json"
+    manifest_path: str = "./analytics/target/manifest.json"
+    run_results_path: str = "./analytics/target/run_results.json"
 
 class WarehouseConfig(BaseModel):
     type: str = "postgres"

pipelineprobe/connectors/dbt.py

@@ -14,8 +14,8 @@ def __init__(self, config: DbtConfig):
 
     def get_models(self) -> List[DbtModel]:
         project_dir = Path(self.config.project_dir)
-        manifest_path = project_dir / self.config.manifest_path
-        run_results_path = project_dir / self.config.run_results_path
+        manifest_path = Path(self.config.manifest_path)
+        run_results_path = Path(self.config.run_results_path)
 
         if not manifest_path.exists():
             logger.warning("dbt manifest not found at %s — skipping dbt checks.", manifest_path)

pipelineprobe/connectors/postgres.py

@@ -1,8 +1,8 @@
-import asyncio
 import logging
 from typing import Any, Dict, List
 
-import asyncpg
+import psycopg2
+import psycopg2.extras
 
 from pipelineprobe.config import WarehouseConfig
 
@@ -12,34 +12,31 @@ class PostgresConnector:
     def __init__(self, config: WarehouseConfig):
         self.config = config
 
-    async def get_table_stats(self) -> List[Dict[str, Any]]:
+    def get_stats_sync(self) -> List[Dict[str, Any]]:
         conn = None
         try:
-            conn = await asyncpg.connect(self.config.dsn)
-            query = """
-                SELECT 
-                    t.schemaname, 
-                    t.relname as tablename, 
-                    t.n_live_tup as row_count,
-                    EXISTS (
-                        SELECT 1 
-                        FROM information_schema.columns c 
-                        WHERE c.table_schema = t.schemaname 
-                          AND c.table_name = t.relname 
-                          AND c.column_name IN ('updated_at', 'created_at')
-                    ) as has_timestamps
-                FROM pg_stat_user_tables t
-                ORDER BY t.n_live_tup DESC
-                LIMIT 50;
-            """
-            rows = await conn.fetch(query)
-            return [dict(r) for r in rows]
+            conn = psycopg2.connect(self.config.dsn)
+            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+                cur.execute("""
+                    SELECT 
+                        t.schemaname, 
+                        t.relname AS tablename, 
+                        t.n_live_tup AS row_count,
+                        EXISTS (
+                            SELECT 1 FROM information_schema.columns c 
+                            WHERE c.table_schema = t.schemaname 
+                              AND c.table_name = t.relname 
+                              AND c.column_name IN ('updated_at', 'created_at')
+                        ) AS has_timestamps
+                    FROM pg_stat_user_tables t
+                    ORDER BY t.n_live_tup DESC
+                    LIMIT 50
+                """)
+                return [dict(r) for r in cur.fetchall()]
         except Exception as e:
             logger.error("Error connecting to Postgres: %s", e)
             return []
         finally:
             if conn:
-                await conn.close()
-
-    def get_stats_sync(self) -> List[Dict[str, Any]]:
-        return asyncio.run(self.get_table_stats())
+                conn.close()
+

pipelineprobe/rules/airflow_rules.py

@@ -96,7 +96,7 @@ def check_stale_dags(context: dict) -> List[Issue]:
             if end_time_aware.tzinfo is None:
                 end_time_aware = end_time_aware.replace(tzinfo=timezone.utc)
 
-            days_since = (now - end_time_aware).days
+            days_since = (now - end_time_aware).total_seconds() / 86400
             if days_since > stale_threshold_days:
                 issues.append(
                     Issue(

pipelineprobe/rules/postgres_rules.py

@@ -3,7 +3,9 @@
 
 def check_large_tables(context: dict) -> List[Issue]:
     issues = []
-    tables: List[Dict[str, Any]] = context.get("postgres_tables", [])
+    if context.get("warehouse_type") != "postgres":
+        return issues
+    tables: List[Dict[str, Any]] = context.get("warehouse_tables", [])
     for table in tables:
         if table.get("row_count", 0) > 10_000_000:
             issues.append(
@@ -20,7 +22,9 @@ def check_large_tables(context: dict) -> List[Issue]:
 
 def check_missing_timestamps(context: dict) -> List[Issue]:
     issues = []
-    tables: List[Dict[str, Any]] = context.get("postgres_tables", [])
+    if context.get("warehouse_type") != "postgres":
+        return issues
+    tables: List[Dict[str, Any]] = context.get("warehouse_tables", [])
     for table in tables:
         # For simplicity, we flag tables > 1,000,000 rows without timestamps
         if table.get("row_count", 0) > 1_000_000 and not table.get("has_timestamps"):

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "pydantic-settings>=2.0.0",
     "httpx>=0.25.0",
     "jinja2>=3.1.2",
-    "asyncpg>=0.28.0",
+    "psycopg2-binary>=2.9.0",
     "pyyaml>=6.0",
     "python-dateutil>=2.8.0",
     "google-cloud-bigquery>=3.11.0",