feat(procore): write metadata-only audit records

api/main.py

@@ -1332,7 +1332,12 @@ def _verify_procore_request(headers, body: bytes) -> bool:
     return False
 
 
-async def _process_procore_v1_webhook(payload: dict, event, correlation_id: str) -> dict:
+async def _process_procore_v1_webhook(
+    payload: dict,
+    event,
+    correlation_id: str,
+    audit_delivery_key: str = "",
+) -> dict:
     """Background pipeline: gate -> fetch -> extract -> review -> store ->
     compare -> comment. Returns a result dict. Failures are recorded and
     alerted, never raised to the webhook caller (which already received 202).
@@ -1432,6 +1437,9 @@ async def _process_procore_v1_webhook(payload: dict, event, correlation_id: str)
 
         # Phase 1B: post comment back to Procore if live and configured
         comment_posted = False
+        writeback_enabled = (
+            _os.getenv("PROCORE_LIVE_WRITEBACK_ENABLED", "false").strip().lower() == "true"
+        )
         try:
             from core.procore.api_client import (
                 format_review_as_comment,
@@ -1443,9 +1451,6 @@ async def _process_procore_v1_webhook(payload: dict, event, correlation_id: str)
             # PROCORE_LIVE_WRITEBACK_ENABLED=true. The write-back resource is
             # UNVERIFIED (deprecated submittal_logs path) pending Procore
             # confirmation — see docs/procore/action_06_submittal_logs_terminology.md.
-            writeback_enabled = (
-                _os.getenv("PROCORE_LIVE_WRITEBACK_ENABLED", "false").strip().lower() == "true"
-            )
             if writeback_enabled and is_live_configured() and retrieval_mode == "live_api":
                 comment_text = format_review_as_comment(review_artifact, att.filename)
                 post_submittal_comment(event.project_id, event.resource_id, comment_text)
@@ -1465,6 +1470,22 @@ async def _process_procore_v1_webhook(payload: dict, event, correlation_id: str)
         if comparison:
             result["comparison"] = comparison
 
+        # Metadata-only audit row (no raw SWMS text/comment body). No-op if
+        # Supabase is unconfigured or migration 008 has not been applied yet.
+        try:
+            from core.procore.audit import record_review_audit
+            record_review_audit(
+                event=event,
+                review_artifact=review_artifact,
+                delivery_key=audit_delivery_key,
+                correlation_id=correlation_id,
+                retrieval_mode=retrieval_mode,
+                writeback_enabled=writeback_enabled,
+                comment_posted=comment_posted,
+            )
+        except Exception as audit_err:
+            logger.warning(f"Procore audit write skipped: {audit_err}")
+
         # Durable status row on completion (no-op when Supabase unconfigured).
         try:
             from core.job_state import record_state
@@ -1529,7 +1550,8 @@ async def procore_webhook_endpoint(request: Request, background_tasks: Backgroun
     # present so every delivery is de-duplicated; durable via Supabase when
     # configured, in-memory otherwise.
     correlation_id = f"procore-{event.delivery_id}" if event.delivery_id else "procore"
-    if not reserve_delivery(delivery_key(event.delivery_id, body), correlation_id):
+    audit_delivery_key = delivery_key(event.delivery_id, body)
+    if not reserve_delivery(audit_delivery_key, correlation_id):
         return JSONResponse(content={"status": "already_processed", "delivery_id": event.delivery_id})
 
     # Log payload for replay/debugging (after reservation).
@@ -1538,7 +1560,13 @@ async def procore_webhook_endpoint(request: Request, background_tasks: Backgroun
     # Heavy pipeline (fetch -> review -> store -> compare -> comment) runs
     # asynchronously: respond 202 fast per Procore webhook guidance. The
     # outcome and any failure are recorded + alerted inside the background task.
-    background_tasks.add_task(_process_procore_v1_webhook, payload, event, correlation_id)
+    background_tasks.add_task(
+        _process_procore_v1_webhook,
+        payload,
+        event,
+        correlation_id,
+        audit_delivery_key,
+    )
     return JSONResponse(
         content={"status": "accepted", "delivery_id": event.delivery_id,
                  "correlation_id": correlation_id},

core/procore/audit.py

@@ -0,0 +1,155 @@
+"""Minimal Supabase audit writer for the Procore review pipeline.
+
+The audit row is metadata-only: hashes, versions, statuses, counts, and
+write-back state. It must not persist raw SWMS text, review prose, comment
+body, attachment bytes, or amendment reasons.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+
+import httpx
+
+from core.procore.webhook_handler import WebhookEvent
+
+log = logging.getLogger(__name__)
+
+_TIMEOUT = 10.0
+_DEFAULT_RETENTION_DAYS = 365
+_WRITEBACK_RESOURCE_STATUS = "unverified"
+
+
+def _supabase_configured() -> bool:
+    return bool(os.getenv("SUPABASE_URL", "") and os.getenv("SUPABASE_SERVICE_ROLE_KEY", ""))
+
+
+def _positive_int_or_none(value: Any) -> int | None:
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return None
+    return parsed if parsed > 0 else None
+
+
+def _retention_days() -> int:
+    raw = os.getenv("PROCORE_AUDIT_RETENTION_DAYS", str(_DEFAULT_RETENTION_DAYS))
+    try:
+        parsed = int(raw)
+    except ValueError:
+        return _DEFAULT_RETENTION_DAYS
+    return parsed if parsed > 0 else _DEFAULT_RETENTION_DAYS
+
+
+def _hard_fail_count(review_artifact: dict[str, Any]) -> int:
+    amendments = review_artifact.get("required_amendments", [])
+    if not isinstance(amendments, list):
+        return 0
+    hard_severities = {"hard_fail", "mandatory", "critical"}
+    return sum(
+        1
+        for item in amendments
+        if isinstance(item, dict)
+        and str(item.get("severity", "")).strip().lower() in hard_severities
+    )
+
+
+def _finding_count(review_artifact: dict[str, Any]) -> int:
+    amendments = review_artifact.get("required_amendments", [])
+    if isinstance(amendments, list):
+        return len(amendments)
+    return 0
+
+
+def build_procore_audit_record(
+    *,
+    event: WebhookEvent,
+    review_artifact: dict[str, Any],
+    delivery_key: str = "",
+    correlation_id: str = "",
+    retrieval_mode: str = "",
+    writeback_enabled: bool = False,
+    comment_posted: bool = False,
+) -> dict[str, Any]:
+    """Build the metadata-only payload for public.record_procore_audit(jsonb)."""
+    return {
+        "record_type": "review",
+        "review_run_id": review_artifact.get("review_run_id", ""),
+        "delivery_key": delivery_key or event.delivery_id,
+        "correlation_id": correlation_id,
+        "company_id": _positive_int_or_none(event.company_id),
+        "project_id": _positive_int_or_none(event.project_id),
+        "document_hash": review_artifact.get("document_hash", ""),
+        "rule_pack_version": review_artifact.get("rule_pack_version", ""),
+        "rule_library_version": review_artifact.get("rule_library_version", ""),
+        "project_review_status": review_artifact.get("project_review_status", ""),
+        "status_recommendation": review_artifact.get("status_recommendation", ""),
+        "workflow_state": review_artifact.get("workflow_state", ""),
+        "review_confidence": review_artifact.get("review_confidence", ""),
+        "finding_count": _finding_count(review_artifact),
+        "hard_fail_count": _hard_fail_count(review_artifact),
+        "writeback": {
+            "enabled": bool(writeback_enabled),
+            "posted": bool(comment_posted),
+            "retrieval_mode": retrieval_mode,
+            "resource_surface": "submittals",
+            "resource_id": _positive_int_or_none(event.resource_id),
+            "resource_status": _WRITEBACK_RESOURCE_STATUS,
+        },
+        "retention_days": _retention_days(),
+    }
+
+
+def _post_supabase_audit(record: dict[str, Any]) -> int | None:
+    base = os.getenv("SUPABASE_URL", "")
+    service_key = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "")
+    url = f"{base}/rest/v1/rpc/record_procore_audit"
+    headers = {
+        "apikey": service_key,
+        "Authorization": f"Bearer {service_key}",
+        "Content-Type": "application/json",
+    }
+    with httpx.Client(timeout=_TIMEOUT) as client:
+        resp = client.post(url, headers=headers, json={"p_record": record})
+        resp.raise_for_status()
+        value = resp.json()
+    if isinstance(value, int):
+        return value
+    return _positive_int_or_none(value)
+
+
+def record_procore_audit(record: dict[str, Any]) -> int | None:
+    """Write one Procore audit row, degrading to no-op if Supabase/RPC is absent."""
+    if not _supabase_configured():
+        log.info("Supabase not configured; skipping Procore audit write")
+        return None
+    try:
+        return _post_supabase_audit(record)
+    except Exception as exc:  # pragma: no cover - network/remote schema errors
+        log.warning("Procore audit write failed; continuing without audit row: %s", exc)
+        return None
+
+
+def record_review_audit(
+    *,
+    event: WebhookEvent,
+    review_artifact: dict[str, Any],
+    delivery_key: str = "",
+    correlation_id: str = "",
+    retrieval_mode: str = "",
+    writeback_enabled: bool = False,
+    comment_posted: bool = False,
+) -> int | None:
+    """Build and write the metadata-only review audit record."""
+    record = build_procore_audit_record(
+        event=event,
+        review_artifact=review_artifact,
+        delivery_key=delivery_key,
+        correlation_id=correlation_id,
+        retrieval_mode=retrieval_mode,
+        writeback_enabled=writeback_enabled,
+        comment_posted=comment_posted,
+    )
+    return record_procore_audit(record)