#!/usr/bin/env python3 from __future__ import annotations import argparse import json import re import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Any import yaml try: import psycopg except ImportError: psycopg = None def parse_dotnet_connection_string(s: str) -> dict[str, str | None]: """Parse .NET connection string format to psycopg-compatible dict. Input: "Host=x;Port=n;Database=d;Username=u;Password=p;Search Path=s;" Output: {host, port, dbname, user, password, options (if Search Path present)} Keys are case-insensitive; unknown keys ignored. """ result: dict[str, str | None] = {} parts = s.split(";") search_path_value = None for part in parts: part = part.strip() if not part: continue if "=" not in part: continue key, value = part.split("=", 1) key_lower = key.strip().lower() value = value.strip() if key_lower == "host": result["host"] = value elif key_lower == "port": result["port"] = value elif key_lower == "database": result["dbname"] = value elif key_lower == "username": result["user"] = value elif key_lower == "password": result["password"] = value elif key_lower == "search path": search_path_value = value if search_path_value: result["options"] = f"-c search_path={search_path_value}" return result def load_spec(spec_path: Path) -> dict[str, Any]: """Load and parse the WBS spec YAML.""" if not spec_path.exists(): raise FileNotFoundError(f"Spec file not found: {spec_path}") return yaml.safe_load(spec_path.read_text(encoding="utf-8")) def resolve_db_connection(root: Path, spec: dict[str, Any]) -> str | None: """Resolve PostgreSQL connection string. Resolution order: 1. env QE_WBS_PG_DSN (psycopg DSN format) 2. env ConnectionStrings__DefaultConnection (.NET format, convert to psycopg) 3. appsettings.Development.json ConnectionStrings.DefaultConnection (.NET format, convert) """ import os # Try env QE_WBS_PG_DSN dsn = os.environ.get("QE_WBS_PG_DSN") if dsn: return dsn # Try env ConnectionStrings__DefaultConnection (.NET format) dotnet_str = os.environ.get("ConnectionStrings__DefaultConnection") if dotnet_str: parsed = parse_dotnet_connection_string(dotnet_str) return _build_psycopg_dsn(parsed) # Try appsettings.Development.json appsettings_path = root / spec.get("meta", {}).get("db_connection", {}).get("dotnet_appsettings", "") if appsettings_path and appsettings_path.is_relative_to(root): full_path = root / appsettings_path if full_path.exists(): try: appsettings = json.loads(full_path.read_text(encoding="utf-8")) conn_str = appsettings.get("ConnectionStrings", {}).get("DefaultConnection", "") if conn_str: parsed = parse_dotnet_connection_string(conn_str) return _build_psycopg_dsn(parsed) except Exception: pass return None def _build_psycopg_dsn(parsed: dict[str, str | None]) -> str: """Build psycopg DSN from parsed dict.""" parts = [] for key in ["host", "port", "dbname", "user", "password"]: val = parsed.get(key) if val: parts.append(f"{key}={val}") # The evidence queries use fully-qualified schema names. Do not emit the # parsed Search Path as a libpq DSN option: unquoted values are interpreted # as connection keywords by psycopg (for example, "search_path"), which # makes an otherwise valid .NET connection string fail to connect. return " ".join(parts) def check_pg_query(root: Path, check: dict[str, Any], dsn: str | None) -> tuple[bool, dict[str, Any]]: """Verify pg_query check type.""" if psycopg is None: return False, {"error": "psycopg not installed"} if not dsn: return False, {"error": "No PostgreSQL connection available"} sql = check.get("sql", "") expect = check.get("expect", {}) try: conn = psycopg.connect(dsn) try: cursor = conn.cursor() cursor.execute(sql) row = cursor.fetchone() observed = row[0] if row else None # Try to coerce to numeric for comparison if observed is not None: try: observed = float(observed) except (ValueError, TypeError): pass # Check expectations passed = True if "min" in expect: min_val = expect["min"] if observed is None or float(observed) < float(min_val): passed = False if "max" in expect and passed: max_val = expect["max"] if observed is None or float(observed) > float(max_val): passed = False if "equals" in expect and passed: eq_val = expect["equals"] if observed != eq_val: passed = False cursor.close() conn.close() return passed, { "sql": sql, "observed": observed, "expected": expect, "connected_as_host": dsn.split("host=")[-1].split()[0] if "host=" in dsn else "unknown" } except Exception as e: conn.close() return False, {"error": str(e), "sql": sql} except Exception as e: return False, {"error": str(e), "sql": sql} def check_log_pattern(root: Path, check: dict[str, Any]) -> tuple[bool, dict[str, Any]]: """Verify log_pattern check type.""" from pathlib import Path as PathlibPath file_glob = check.get("file_glob", "") pattern_str = check.get("pattern", "") expect = check.get("expect", {}) min_matches = expect.get("min_matches") max_matches = expect.get("max_matches") max_age_hours = expect.get("max_age_hours") if not file_glob or not pattern_str: return False, {"error": "Missing file_glob or pattern"} try: pattern = re.compile(pattern_str) except re.error as e: return False, {"error": f"Invalid regex: {e}"} # Glob for files glob_path = root / file_glob matched_files = list(glob_path.parent.glob(glob_path.name)) if "*" in file_glob else ( [glob_path] if glob_path.exists() else [] ) # Handle ** in glob if "**" in file_glob: parts = file_glob.split("**") base = root / parts[0] if parts[0] else root suffix = parts[-1] if len(parts) > 1 else "*" matched_files = list(base.glob(f"**/{suffix}")) # Filter by age if needed import time as time_module now = time_module.time() if max_age_hours: max_age_seconds = max_age_hours * 3600 matched_files = [f for f in matched_files if f.is_file() and (now - f.stat().st_mtime) <= max_age_seconds] # Count matches match_count = 0 matched_lines = [] for file_path in matched_files: try: content = file_path.read_text(encoding="utf-8", errors="replace") for line in content.splitlines(): if pattern.search(line): match_count += 1 if len(matched_lines) < 200: matched_lines.append(f"{file_path.name}: {line}") except Exception: pass # Check expectations passed = True if min_matches is not None and match_count < min_matches: passed = False if max_matches is not None and match_count > max_matches: passed = False # Special case: if neither bound given, default min_matches=1 if min_matches is None and max_matches is None: if match_count < 1: passed = False return passed, { "file_glob": file_glob, "pattern": pattern_str, "observed": match_count, "expected": expect, "matched_lines": matched_lines[:200] } def check_json_gate(root: Path, check: dict[str, Any]) -> tuple[bool, dict[str, Any]]: """Verify json_gate check type.""" path_str = check.get("path", "") expect = check.get("expect", {}) if not path_str: return False, {"error": "Missing path"} json_path = root / path_str if not json_path.exists(): return False, {"error": f"JSON file not found: {json_path}", "path": path_str} try: payload = json.loads(json_path.read_text(encoding="utf-8")) except Exception as e: return False, {"error": f"Failed to parse JSON: {e}", "path": path_str} # Check each expectation key passed = True details = {} for key, expected_val in expect.items(): # Dot-notation path support observed_val = payload for part in key.split("."): if isinstance(observed_val, dict): observed_val = observed_val.get(part) else: observed_val = None break # Check if expected_val is a comparison operator string if isinstance(expected_val, str): # Check two-character operators first for op, op_str in [(">=", ">="), ("<=", "<="), (">", ">"), ("<", "<")]: if expected_val.startswith(op_str): try: expected_num = float(expected_val[len(op_str):]) observed_num = float(observed_val) if observed_val is not None else None if observed_num is None: passed = False details[key] = f"Expected {expected_val}, got {observed_val}" elif op == ">=" and observed_num < expected_num: passed = False details[key] = f"Expected {expected_val}, got {observed_num}" elif op == "<=" and observed_num > expected_num: passed = False details[key] = f"Expected {expected_val}, got {observed_num}" elif op == ">" and observed_num <= expected_num: passed = False details[key] = f"Expected {expected_val}, got {observed_num}" elif op == "<" and observed_num >= expected_num: passed = False details[key] = f"Expected {expected_val}, got {observed_num}" except (ValueError, TypeError): passed = False details[key] = f"Failed to parse comparison: {expected_val}" break else: # Plain equality if observed_val != expected_val: passed = False details[key] = f"Expected {expected_val}, got {observed_val}" else: # Plain equality if observed_val != expected_val: passed = False details[key] = f"Expected {expected_val}, got {observed_val}" return passed, { "path": path_str, "expected": expect, "details": details if details else "All checks passed" } def check_file_exists(root: Path, check: dict[str, Any]) -> tuple[bool, dict[str, Any]]: """Verify file_exists check type.""" paths = check.get("paths", []) expect = check.get("expect", {}) min_bytes = expect.get("min_bytes") if not paths: return False, {"error": "Missing paths"} details = {} passed = True for path_str in paths: file_path = root / path_str exists = file_path.exists() details[path_str] = {"exists": exists} if not exists: passed = False continue if file_path.is_file() and min_bytes is not None: size = file_path.stat().st_size details[path_str]["size"] = size if size < min_bytes: passed = False return passed, { "paths": paths, "expected": expect, "details": details } def check_playwright_report(root: Path, check: dict[str, Any]) -> tuple[bool, dict[str, Any]]: """Verify playwright_report check type.""" report_path_str = check.get("report", "") spec_file = check.get("spec_file", "") expect = check.get("expect", {}) passed_min = expect.get("passed_min", 1) failed_expected = expect.get("failed", 0) if not report_path_str or not spec_file: return False, {"error": "Missing report or spec_file"} report_path = root / report_path_str if not report_path.exists(): return False, {"error": f"Playwright report not found: {report_path}"} try: report = json.loads(report_path.read_text(encoding="utf-8")) except Exception as e: return False, {"error": f"Failed to parse report: {e}"} # Walk suites to find specs matching spec_file passed_count = 0 failed_count = 0 def walk_suites(suites_list): nonlocal passed_count, failed_count if not suites_list: return for suite in suites_list: # Recurse into nested suites if "suites" in suite: walk_suites(suite["suites"]) # Check specs if "specs" in suite: for spec in suite["specs"]: if spec_file in spec.get("file", ""): if spec.get("ok"): passed_count += 1 else: failed_count += 1 suites = report.get("suites", []) walk_suites(suites) passed = (passed_count >= passed_min) and (failed_count == failed_expected) return passed, { "report": report_path_str, "spec_file": spec_file, "passed_count": passed_count, "failed_count": failed_count, "expected": expect } def run_verification_commands(root: Path, task_id: str, commands: list[str]) -> list[dict[str, Any]]: """Execute verification commands (excluding self-references).""" results = [] evidence_dir = root / "Temp" / "evidence" / task_id evidence_dir.mkdir(parents=True, exist_ok=True) for idx, cmd in enumerate(commands): # Skip commands that reference this script to avoid recursion if "verify_wbs_task_v1.py" in cmd: continue try: # Write to command log file log_file = evidence_dir / f"command_{idx}.log" result = subprocess.run( cmd, shell=True, cwd=root, capture_output=True, text=True, encoding="utf-8", errors="replace" ) output = result.stdout + result.stderr log_file.write_text(output, encoding="utf-8") results.append({ "index": idx, "command": cmd, "returncode": result.returncode, "log_file": str(log_file.relative_to(root)) }) except Exception as e: results.append({ "index": idx, "command": cmd, "error": str(e) }) return results def main(argv: list[str] | None = None) -> int: """Main entry point.""" parser = argparse.ArgumentParser(description="Verify a single WBS task evidence") parser.add_argument("--task", required=True, help="Task ID (e.g., QE-M0-01)") parser.add_argument("--repo-root", default=None, help="Repository root path") parser.add_argument("--spec", default=None, help="Spec YAML file path") parser.add_argument("--run-commands", action="store_true", help="Execute verification commands") args = parser.parse_args(argv) task_id = args.task # Resolve root if args.repo_root: root = Path(args.repo_root).resolve() else: root = Path(__file__).resolve().parents[1] # Resolve spec path if args.spec: spec_path = Path(args.spec).resolve() else: spec_path = root / "spec" / "60_quant_engine_wbs.yaml" # Load spec try: spec = load_spec(spec_path) except FileNotFoundError as e: print(f"ERROR: {e}") return 2 # Find task tasks = spec.get("tasks", {}) if task_id not in tasks: print(f"ERROR: Task {task_id} not found in spec") return 2 task = tasks[task_id] evidence_checks = task.get("evidence_checks", []) verification_commands = task.get("success_criteria", {}).get("verification_commands", []) # Create evidence directory evidence_dir = root / "Temp" / "evidence" / task_id evidence_dir.mkdir(parents=True, exist_ok=True) # Resolve DB connection dsn = resolve_db_connection(root, spec) # Run checks check_results = [] gate = "PASS" for check_idx, check in enumerate(evidence_checks): check_type = check.get("type") passed = False detail = {} try: if check_type == "pg_query": passed, detail = check_pg_query(root, check, dsn) elif check_type == "log_pattern": passed, detail = check_log_pattern(root, check) elif check_type == "json_gate": passed, detail = check_json_gate(root, check) elif check_type == "file_exists": passed, detail = check_file_exists(root, check) elif check_type == "playwright_report": passed, detail = check_playwright_report(root, check) else: passed = False detail = {"error": f"Unknown check type: {check_type}"} except Exception as e: passed = False detail = {"error": str(e)} if not passed: gate = "FAIL" check_results.append({ "index": check_idx, "type": check_type, "gate": "PASS" if passed else "FAIL", "detail": detail }) # Run commands if requested if args.run_commands: run_verification_commands(root, task_id, verification_commands) # Prepare verdict verdict = { "task_id": task_id, "formula_id": "QUANT_ENGINE_WBS_TASK_V1", "gate": gate, "generated_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), "checks": check_results, "input_hashes": {} } # Save verdict verdict_path = evidence_dir / "verdict.json" verdict_path.write_text(json.dumps(verdict, ensure_ascii=False, indent=2), encoding="utf-8") # Append lineage event lineage_log = root / "runtime" / "lineage_events.jsonl" lineage_log.parent.mkdir(parents=True, exist_ok=True) event = { "node_id": f"wbs_{task_id}", "command": f"python tools/verify_wbs_task_v1.py --task {task_id}", "returncode": 0 if gate == "PASS" else 1, "elapsed_sec": 0, "gate": gate, "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") } lineage_log.write_text(json.dumps(event, ensure_ascii=False) + "\n", encoding="utf-8", errors="append") # Print summary (ASCII only) print(f"Task: {task_id}") for check_result in check_results: check_gate = check_result["gate"] check_type = check_result["type"] status = "PASS" if check_gate == "PASS" else "FAIL" print(f" [{status}] {check_type}") print(f"Gate: {gate}") return 0 if gate == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())