#!/usr/bin/env python3 from __future__ import annotations import argparse import datetime as dt import hashlib import json import os import re import shutil import subprocess import sys import time from pathlib import Path from typing import Any, Mapping from urllib.error import HTTPError from urllib.parse import urlencode from urllib.request import Request, urlopen try: from lib.test1_runtime_evidence import ( EvidenceBlocked as Test1EvidenceBlocked, build_test1_connected_work_evidence, sensitive_values_for_mapping, write_test1_connected_work_evidence, ) except ModuleNotFoundError: from scripts.lib.test1_runtime_evidence import ( EvidenceBlocked as Test1EvidenceBlocked, build_test1_connected_work_evidence, sensitive_values_for_mapping, write_test1_connected_work_evidence, ) try: from hubspot_pressure_matrix_contract import ( build_scale_gate, load_pressure_matrix_contract, ) except ModuleNotFoundError: from scripts.hubspot_pressure_matrix_contract import ( build_scale_gate, load_pressure_matrix_contract, ) DEFAULT_ENV_ROOT = Path(os.environ.get("ENV_ROOT", ".")) DEFAULT_BASE_URL = os.environ.get("PUBLIC_BASE_URL", "") DEFAULT_TARGET_ENV = os.environ.get("ENV_NAME", "staging") DEFAULT_NAMESPACE = "hubspot-crm-connector" DEFAULT_WORKITEM_INDEX = "idx_connector_work_item_source_lane_gen" DEFAULT_RUNTIME_CONNECTORS_PATH = ".runtime/nexuscrm-connectors-pr43" DEFAULT_PREFLIGHT_SCRIPT = Path("scripts/verify_staging_hubspot_webhook.py") DEFAULT_PRESSURE_RUNNER = Path("scripts/hubspot_workitem_pressure_gate.py") DEFAULT_MATERIALIZED_RUNNER = Path("scripts/hubspot_materialized_workitem_gate.py") DEFAULT_DEPLOYMENT_CONFIG = Path(__file__).resolve().parents[1] / "deployments" / "deployment-config.json" DEFAULT_CLIENT_SECRET_ENV = "EXP_HUBSPOT_CLIENT_SECRET" DEFAULT_ACCESS_TOKEN_ENV = "EXP_HUBSPOT_ACCESS_TOKEN" MATRIX_VERSION = "connected-work-matrix-v1" GATE_DESCRIPTIONS = { "runtime": "runtime refs, health, WorkItem claim flags, and schema/index authority", "secret": "runtime env sourcing plus signed admission preflight without secret disclosure", "hubspot-read": "read-only HubSpot discovery and batch/read capacity", "workitem-pilot": "small clean/materialized WorkItem queue drain without business-flow pressure", "normal-pilot": "small signed normal inbound end-to-end pilot", "coalescer-log": "canonical alias coalescer/fallback log sanity from the backend log tail", "workitem-materialized": "scoped materialized Event+WorkItem 800/120 queue gate at the requested worker count", "workitem-clean": "signed admission clean/no-op 800/120 gate at the requested worker count", "normal800": "full normal inbound 800/120 end-to-end gate at the requested worker count", "workitem-scale": "candidate-scoped 2k/4k/6k inbound source-projection gate at exactly 8/16/32 deployed workers", } NORMAL800_REQUIRED_GATES = [ "runtime", "secret", "hubspot-read", "workitem-pilot", "normal-pilot", "workitem-materialized", "workitem-clean", ] GATE_DEPENDENCIES = { "normal800": NORMAL800_REQUIRED_GATES, "workitem-scale": ["runtime"], } PRESSURE_FIXTURE_GATES = {"workitem-pilot", "normal-pilot", "workitem-clean", "normal800", "workitem-scale"} TEST1_FIXTURELESS_EVIDENCE_GATES = frozenset({"runtime", "workitem-pilot"}) SECRET_NAME_RE = re.compile(r"(SECRET|TOKEN|PASSWORD|PASSWD|PRIVATE|KEY)", re.IGNORECASE) REF_ENV_KEYS = [ "ENV_NAME", "DEPLOY_ID", "UPSTREAM_REF", "OFFICIAL_APP_PACKS_REF", "DOWNSTREAM_REF", "CONNECTORS_EXPECTED_REF", "CONNECTORS_REF", "CONNECTORS_MODULE_VERSION", "CONNECTORS_MODULE_READBACK_REF", "NEXUSCRM_CONNECTOR_WORKITEM_CLAIM_ENABLED", "NEXUSCRM_CONNECTOR_INBOUND_WORKERS_PER_PACK", "NEXUSCRM_CONNECTOR_INBOUND_MAX_PROCESSING_ROWS", "NEXUSCRM_CONNECTOR_INBOUND_QUEUE_PROFILE", "NEXUSCRM_CONNECTOR_VERIFIED_INGRESS_RECEIPTS", "PUBLIC_BASE_URL", "TIDB_HOST", "TIDB_PORT", "TIDB_DATABASE", "TIDB_USER", ] SECRET_ENV_KEYS = [ DEFAULT_CLIENT_SECRET_ENV, DEFAULT_ACCESS_TOKEN_ENV, "TIDB_PASSWORD", ] class GateFailure(Exception): def __init__(self, status: str, message: str) -> None: super().__init__(message) self.status = status self.message = message def utc_stamp() -> str: return dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ") def short_sha(value: str) -> str: value = (value or "").strip() return value[:7] if value else "unknown" def requires_fixture_manifest(gates: list[str]) -> bool: return any(gate in PRESSURE_FIXTURE_GATES for gate in gates) def validate_fixture_manifest_requirement( gates: list[str], fixture_manifest: str, candidate_id: str, target_env: str, *, test_slot_identity_verified: bool = False, ) -> None: if "workitem-scale" in gates and str(target_env).strip().lower() != "test1": raise GateFailure( "BLOCK_PRESSURE_SCALE_TEST1_ONLY", "the 2k/4k/6k workitem-scale gate is restricted to Test1", ) if not requires_fixture_manifest(gates): return if ( target_env == "test1" and not fixture_manifest and not candidate_id and test_slot_identity_verified and len(gates) == len(TEST1_FIXTURELESS_EVIDENCE_GATES) and frozenset(gates) == TEST1_FIXTURELESS_EVIDENCE_GATES ): return if not fixture_manifest: raise GateFailure( "BLOCK_HUBSPOT_FIXTURE_MANIFEST_REQUIRED", "candidate-scoped HubSpot fixture manifest is required for pressure gates", ) if not candidate_id: raise GateFailure( "BLOCK_HUBSPOT_FIXTURE_CANDIDATE_REQUIRED", "candidate id is required with a HubSpot fixture manifest", ) def sha256_text(value: str) -> str: return hashlib.sha256(value.encode()).hexdigest() def parse_env_file(path: Path) -> dict[str, str]: env: dict[str, str] = {} if not path.exists(): raise GateFailure( "BLOCK_INPUT_SECRET_GATE_runtime_env_missing", f"runtime env missing: {path}", ) for raw_line in path.read_text().splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'): value = value[1:-1] env[key] = value return env def make_child_env(runtime_env: dict[str, str]) -> dict[str, str]: merged = os.environ.copy() merged.update(runtime_env) merged["RUNTIME_ENV_SOURCE_FILE"] = runtime_env.get("RUNTIME_ENV_SOURCE_FILE", "") return merged def bind_runtime_inputs(args: argparse.Namespace, runtime_env: Mapping[str, str]) -> None: """Resolve non-secret matrix identity from the owning runtime env.""" args.base_url = str(args.base_url or "").strip() or runtime_env.get( "PUBLIC_BASE_URL", "" ).strip() args.expected_sha = str(args.expected_sha or "").strip() or ( runtime_env.get("CONNECTORS_MODULE_READBACK_REF", "").strip() or runtime_env.get("CONNECTORS_EXPECTED_REF", "").strip() or runtime_env.get("CONNECTORS_REF", "").strip() ) if args.app_id == 0: args.app_id = int(runtime_env.get("EXP_HUBSPOT_APP_ID", "0") or "0") if args.portal_id == 0: args.portal_id = int(runtime_env.get("EXP_HUBSPOT_PORTAL_ID", "0") or "0") def validate_test_slot_runtime_identity( args: argparse.Namespace, runtime_env: Mapping[str, str], ) -> bool: """Fail before network or database access when a Test slot is misbound.""" if not args.require_test_slot_identity: return False if args.target_env != "test1": raise GateFailure( "BLOCK_TEST_SLOT_IDENTITY_TARGET", "strict Test slot identity is currently defined only for test1", ) try: config = json.loads(Path(args.deployment_config).read_text(encoding="utf-8")) target = config["environments"]["test"]["topology"]["target_config"][args.target_env] tidb = target["tidb"] hubspot = target["hubspot"] expected_backend_port = str(target["backend_port"]).strip() expected_portal_id = int(str(hubspot["portal_id"]).strip()) if expected_portal_id <= 0: raise ValueError("HubSpot portal id must be positive") expected_tidb = { "TIDB_HOST": str(tidb["host"]).strip(), "TIDB_PORT": str(tidb["port"]).strip(), "TIDB_USER": str(tidb["user"]).strip(), "TIDB_CLOUD_CLUSTER_ID": str(tidb["cluster_id"]).strip(), "TIDB_CLOUD_BRANCH_ID": str(tidb["branch_id"]).strip(), } except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: raise GateFailure( "BLOCK_TEST_SLOT_IDENTITY_AUTHORITY", "authoritative Test slot topology is unavailable or invalid", ) from exc env_root = Path(args.env_root).expanduser().resolve() runtime_env_path = Path(args.runtime_env).expanduser().resolve() expected_server_name = f"nexuscrm-new-{args.target_env}.pingcap.com" expected = { "ENV_NAME": f"test-{args.target_env}", "ENV_ROOT": str(env_root), "DEPLOY_TARGET": args.target_env, "BACKEND_PORT": expected_backend_port, "SERVER_NAME": expected_server_name, "PUBLIC_BASE_URL": f"https://{expected_server_name}", "EXP_HUBSPOT_PORTAL_ID": str(expected_portal_id), "TIDB_DATABASE": "nexuscrm", **expected_tidb, } actual = {key: str(runtime_env.get(key, "")).strip() for key in expected} if actual.get("ENV_ROOT"): actual["ENV_ROOT"] = str(Path(actual["ENV_ROOT"]).expanduser().resolve()) mismatches = sorted(key for key, value in expected.items() if actual.get(key) != value) if runtime_env_path != env_root / "env" / "runtime.env": mismatches.append("RUNTIME_ENV_FILE") if str(args.base_url).strip() != expected["PUBLIC_BASE_URL"]: mismatches.append("MATRIX_BASE_URL") if args.portal_id != expected_portal_id: mismatches.append("MATRIX_HUBSPOT_PORTAL_ID") try: runtime_app_id = int(str(runtime_env.get("EXP_HUBSPOT_APP_ID", "")).strip()) except ValueError: runtime_app_id = 0 if runtime_app_id <= 0: mismatches.append("EXP_HUBSPOT_APP_ID") if args.app_id != runtime_app_id: mismatches.append("MATRIX_HUBSPOT_APP_ID") if mismatches: raise GateFailure( "BLOCK_TEST_SLOT_IDENTITY_MISMATCH", "runtime environment does not match authoritative Test1 identity: " + ", ".join(sorted(set(mismatches))), ) return True def safe_env_readback(env: dict[str, str], runtime_env_path: Path) -> dict[str, Any]: refs = {key: env.get(key, "") for key in REF_ENV_KEYS if key in env} presence = {key: bool(env.get(key, "").strip()) for key in SECRET_ENV_KEYS} return { "runtimeEnvSourceFile": str(runtime_env_path), "refs": refs, "secretPresence": presence, } def redact(text: str, env: dict[str, str]) -> str: redacted = text for key, value in env.items(): if not value or not SECRET_NAME_RE.search(key): continue redacted = redacted.replace(value, "[REDACTED_SECRET]") return redacted def write_json(path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2, sort_keys=True, default=str) + "\n") def read_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text()) def run_cmd( cmd: list[str], *, env: dict[str, str], cwd: Path | None, timeout: int, log_path: Path, ) -> subprocess.CompletedProcess[str]: started = time.perf_counter() proc = subprocess.run( cmd, cwd=str(cwd) if cwd else None, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, check=False, ) elapsed = time.perf_counter() - started log_path.parent.mkdir(parents=True, exist_ok=True) log_path.write_text( json.dumps( { "cmd": scrub_cmd(cmd), "returncode": proc.returncode, "elapsedSeconds": round(elapsed, 3), "stdout": redact(proc.stdout, env), "stderr": redact(proc.stderr, env), }, indent=2, sort_keys=True, ) + "\n" ) return proc def scrub_cmd(cmd: list[str]) -> list[str]: scrubbed: list[str] = [] skip_next = False for item in cmd: if skip_next: scrubbed.append("[REDACTED_VALUE]") skip_next = False continue scrubbed.append(item) if SECRET_NAME_RE.search(item) and item.startswith("--"): skip_next = True return scrubbed def http_request_json(method: str, url: str, body: Any | None, headers: dict[str, str], timeout: int) -> tuple[int, str, float]: raw = b"" if body is None else json.dumps(body, separators=(",", ":"), sort_keys=True).encode() req = Request(url, data=raw if body is not None else None, headers=headers, method=method) started = time.perf_counter() try: with urlopen(req, timeout=timeout) as resp: return resp.status, resp.read().decode(errors="replace"), (time.perf_counter() - started) * 1000.0 except HTTPError as exc: return exc.code, exc.read().decode(errors="replace"), (time.perf_counter() - started) * 1000.0 def evaluate_hubspot_scale_route(rows: list[list[str]], contract: dict[str, Any]) -> dict[str, Any]: expected = { "eventKind": str(contract["eventKind"]), "flowDeveloperName": str(contract["flowDeveloperName"]), "invocationMode": str(contract["invocationMode"]), "maxBatchSize": int(contract["webhookBatchSize"]), "queueClass": str(contract["queueClass"]), } observed_rows = [ { "eventKind": row[0], "flowDeveloperName": row[1], "invocationMode": row[2], "maxBatchSize": int(row[3]) if len(row) > 3 and str(row[3]).isdigit() else None, "queueClass": row[4] if len(row) > 4 else "", } for row in rows if len(row) >= 5 ] return { "expected": expected, "observed": observed_rows, "activeRouteCount": len(observed_rows), "matches": len(observed_rows) == 1 and observed_rows[0] == expected, } def health_gate(args: argparse.Namespace, env: dict[str, str], gate_dir: Path) -> dict[str, Any]: status, body, latency_ms = http_request_json("GET", args.base_url.rstrip("/") + "/health", None, {}, 30) runtime_sha = "" connectors_ref = env.get("CONNECTORS_REF", "").strip() explicit_override = bool(connectors_ref) runtime_path = args.env_root / args.connectors_runtime_path if explicit_override and runtime_path.exists(): proc = run_cmd( ["git", "rev-parse", "HEAD"], env=env, cwd=runtime_path, timeout=15, log_path=gate_dir / "git-rev-parse.json", ) runtime_sha = proc.stdout.strip() if proc.returncode == 0 else "" schema = schema_readback(env, gate_dir) scale_route: dict[str, Any] = {"required": False, "matches": True} if getattr(args, "require_hubspot_scale_route", False): try: scale_route = hubspot_scale_route_readback(env, args.pressure_contract_data) scale_route["required"] = True except Exception as exc: scale_route = { "required": True, "matches": False, "error": f"{type(exc).__name__}: {exc}", } expected = args.expected_sha.strip() expected_ref = env.get("CONNECTORS_EXPECTED_REF", "").strip() module_ref = env.get("CONNECTORS_MODULE_READBACK_REF", "").strip() observed = { "workItemClaimEnabled": env.get("NEXUSCRM_CONNECTOR_WORKITEM_CLAIM_ENABLED") == "true", "workerCountMatches": env.get("NEXUSCRM_CONNECTOR_INBOUND_WORKERS_PER_PACK") == str(args.workers), "maxProcessingRowsMatches": env.get("NEXUSCRM_CONNECTOR_INBOUND_MAX_PROCESSING_ROWS") == str(args.max_processing_rows), "queueProfileMatches": env.get("NEXUSCRM_CONNECTOR_INBOUND_QUEUE_PROFILE") == args.queue_profile, "verifiedIngressReceiptsEnabled": env.get("NEXUSCRM_CONNECTOR_VERIFIED_INGRESS_RECEIPTS") == "true", "schemaAuthorityPresent": bool(schema.get("hasRequiredColumns") and schema.get("hasLaneHeadIndex")), } checks = { "health200": status == 200, "expectedRefMatches": not expected or not expected_ref or expected_ref == expected, "runtimeShaMatches": not explicit_override or runtime_sha == expected, "connectorsRefMatches": not explicit_override or connectors_ref == expected, "moduleReadbackMatches": not expected or module_ref == expected, "schemaAuthorityPresent": observed["schemaAuthorityPresent"] if args.require_schema_authority else True, "workItemClaimEnabled": observed["workItemClaimEnabled"] if args.require_workitem_claim else True, "workerCountExpected": observed["workerCountMatches"] if args.require_worker_count else True, "maxProcessingRowsExpected": ( observed["maxProcessingRowsMatches"] if args.require_max_processing_rows else True ), "queueProfileExpected": observed["queueProfileMatches"] if args.require_queue_profile else True, "hubspotScaleRouteExpected": bool(scale_route.get("matches")), "verifiedIngressReceiptsExpected": ( observed["verifiedIngressReceiptsEnabled"] if getattr(args, "require_hubspot_scale_route", False) else True ), } status_name = "PASS_RUNTIME_REF_SCHEMA_GATE" if all(checks.values()) else "BLOCK_RUNTIME_REF_SCHEMA_GATE" report = { "gate": "A1_RUNTIME_REF_SCHEMA", "status": status_name, "checks": checks, "health": {"statusCode": status, "bodyHash": sha256_text(body), "latencyMs": round(latency_ms, 3)}, "runtime": { "connectorsRuntimePath": str(runtime_path), "runtimeSha": runtime_sha, "expectedSha": expected, "explicitOverride": explicit_override, "readback": safe_env_readback(env, args.runtime_env), "observed": observed, "requirements": { "requireSchemaAuthority": args.require_schema_authority, "requireWorkItemClaim": args.require_workitem_claim, "requireWorkerCount": args.require_worker_count, "requireMaxProcessingRows": args.require_max_processing_rows, "requireQueueProfile": args.require_queue_profile, "workers": args.workers, "maxProcessingRows": args.max_processing_rows, "queueProfile": args.queue_profile, }, }, "schema": schema, "hubspotScaleRoute": scale_route, } write_json(gate_dir / "report.json", report) return report def mysql_query_rows(env: dict[str, str], sql: str) -> list[list[str]]: client = shutil.which("mysql") or shutil.which("mariadb") if not client: raise GateFailure("BLOCK_DB_CLIENT_MISSING_mysql_cli", "mysql or mariadb CLI is required on the runner host") client_env = os.environ.copy() client_env.update(env) client_env["MYSQL_PWD"] = env.get("TIDB_PASSWORD", "") proc = subprocess.run( [ client, "--ssl", "--batch", "--skip-column-names", "--raw", "-h", env.get("TIDB_HOST", ""), "-P", env.get("TIDB_PORT", "4000"), "-u", env.get("TIDB_USER", ""), env.get("TIDB_DATABASE", "nexuscrm"), "-e", sql, ], env=client_env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=60, check=False, ) if proc.returncode != 0: detail = (proc.stderr or proc.stdout or "mysql CLI query failed").strip() raise GateFailure("BLOCK_DB_READBACK_QUERY_FAILED", detail[:500]) return [line.split("\t") for line in proc.stdout.splitlines() if line.strip()] def hubspot_scale_route_readback(env: dict[str, str], contract: dict[str, Any]) -> dict[str, Any]: namespace = DEFAULT_NAMESPACE event_kind = str(contract["eventKind"]) if "'" in namespace or "'" in event_kind: raise GateFailure("BLOCK_HUBSPOT_SCALE_ROUTE_CONTRACT_INVALID", "route readback contract contains an unsafe literal") rows = mysql_query_rows( env, f""" SELECT EventKind, FlowDeveloperName, InvocationMode, MaxBatchSize, QueueClass FROM `_System_ConnectorInboundRoute` WHERE Namespace = '{namespace}' AND EventKind = '{event_kind}' AND IsCurrent = 1 AND COALESCE(IsDeleted, 0) = 0 ORDER BY Version DESC, Id """, ) return evaluate_hubspot_scale_route(rows, contract) def schema_readback(env: dict[str, str], gate_dir: Path) -> dict[str, Any]: required_columns = {"LaneOrdinal", "Generation", "SourceLaneKey", "RunAfter"} try: column_rows = mysql_query_rows(env, "SHOW COLUMNS FROM `_System_ConnectorWorkItem`") index_rows = mysql_query_rows(env, "SHOW INDEX FROM `_System_ConnectorWorkItem`") columns = [row[0] for row in column_rows if row] indexes = [row[2] for row in index_rows if len(row) >= 3] except Exception as exc: # DB error is a gate result, not a script crash. report = {"error": f"{type(exc).__name__}: {exc}"} write_json(gate_dir / "schema-error.json", report) return {"hasRequiredColumns": False, "hasLaneHeadIndex": False, "error": report["error"]} safe = { "columns": columns, "indexes": sorted(set(indexes)), "requiredColumns": sorted(required_columns), "hasRequiredColumns": required_columns.issubset(set(columns)), "laneHeadIndex": DEFAULT_WORKITEM_INDEX, "hasLaneHeadIndex": DEFAULT_WORKITEM_INDEX in set(indexes), } write_json(gate_dir / "schema-readback.json", safe) return safe def preflight_gate(args: argparse.Namespace, env: dict[str, str], gate_dir: Path) -> dict[str, Any]: cmd = [ sys.executable, str(args.preflight_script), "--base-url", args.base_url, "--client-secret-env", args.client_secret_env, "--target-env", args.target_env, "--runtime-env-source", str(args.runtime_env), "--artifact-dir", str(gate_dir / "child"), "--report", str(gate_dir / "child" / "report.md"), ] proc = run_cmd(cmd, env=env, cwd=None, timeout=args.preflight_timeout, log_path=gate_dir / "preflight-subprocess.json") pass_files = sorted((gate_dir / "child").glob("PASS_INBOUND_SECRET_SIGNATURE_PREFLIGHT*.json")) block_files = sorted((gate_dir / "child").glob("BLOCK_*.json")) suffix = short_sha(args.expected_sha) if args.expected_sha.strip() else args.target_env status_name = "PASS_INBOUND_SECRET_SIGNATURE_PREFLIGHT_" + suffix if proc.returncode == 0 and pass_files else "BLOCK_INBOUND_SECRET_SIGNATURE_PREFLIGHT" report = { "gate": "A2_INPUT_SECRET_SIGNATURE", "status": status_name, "returncode": proc.returncode, "passArtifact": str(pass_files[-1]) if pass_files else "", "blockArtifact": str(block_files[-1]) if block_files else "", "runtimeEnvSourceFile": str(args.runtime_env), "secretPresence": {key: bool(env.get(key, "").strip()) for key in [args.client_secret_env, args.hubspot_token_env]}, } write_json(gate_dir / "report.json", report) return report def hubspot_read_gate(args: argparse.Namespace, env: dict[str, str], gate_dir: Path) -> dict[str, Any]: token = env.get(args.hubspot_token_env, "").strip() if not token: report = { "gate": "B1_HUBSPOT_READ_ONLY", "status": f"BLOCK_HUBSPOT_READ_ONLY_MISSING_{args.hubspot_token_env}", "tokenPresence": False, } write_json(gate_dir / "report.json", report) return report headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} contacts: list[str] = [] after = "" while len(contacts) < args.hubspot_required_ids: params: dict[str, Any] = { "limit": 100, "archived": "false", "properties": "email,firstname,lastname,hs_object_id", } if after: params["after"] = after url = "https://api.hubapi.com/crm/v3/objects/contacts?" + urlencode(params) status, body, latency = http_request_json("GET", url, None, headers, 60) if not (200 <= status < 300): report = { "gate": "B1_HUBSPOT_READ_ONLY", "status": f"BLOCK_HUBSPOT_DISCOVERY_HTTP_{status}", "tokenPresence": True, "latencyMs": round(latency, 3), } write_json(gate_dir / "report.json", report) return report data = json.loads(body or "{}") for item in data.get("results") or []: cid = str(item.get("id") or "").strip() if cid and cid not in contacts: contacts.append(cid) if len(contacts) >= args.hubspot_required_ids: break after = str(((data.get("paging") or {}).get("next") or {}).get("after") or "") if not after: break selected = contacts[: args.hubspot_required_ids] chunk_counts: list[dict[str, Any]] = [] read_total = 0 for offset in range(0, len(selected), args.hubspot_batch_size): chunk = selected[offset : offset + args.hubspot_batch_size] if not chunk: continue body = { "properties": ["email", "firstname", "lastname", "hs_object_id"], "inputs": [{"id": value} for value in chunk], } status, text, latency = http_request_json("POST", "https://api.hubapi.com/crm/v3/objects/contacts/batch/read", body, headers, 60) count = 0 if 200 <= status < 300: count = len((json.loads(text or "{}").get("results") or [])) read_total += count chunk_counts.append({"requestCount": len(chunk), "statusCode": status, "responseCount": count, "latencyMs": round(latency, 3)}) pass_gate = len(selected) >= args.hubspot_required_ids and read_total >= len(selected) report = { "gate": "B1_HUBSPOT_READ_ONLY", "status": "PASS_HUBSPOT_READ_ONLY_GATE" if pass_gate else "BLOCK_HUBSPOT_READ_ONLY_GATE", "tokenPresence": True, "requiredIds": args.hubspot_required_ids, "discoveredIds": len(contacts), "selectedIds": len(selected), "batchReadReturned": read_total, "chunkCounts": chunk_counts, } write_json(gate_dir / "report.json", report) return report def build_pressure_runner_command( args: argparse.Namespace, *, report_path: Path, run_id: str, profile: str, workers: int, events: int = 0, lanes: int = 0, timeout_seconds: int = 0, ) -> list[str]: cmd = [ sys.executable, str(args.pressure_runner), "--target-env", args.target_env, "--base-url", args.base_url, "--client-secret-env", args.client_secret_env, "--hubspot-token-env", args.hubspot_token_env, "--portal-id", str(args.portal_id), "--app-id", str(args.app_id), "--output", str(report_path), "--run-id", run_id, "--profile", profile, "--workers", str(workers), ] pressure_contract = str(getattr(args, "pressure_contract", "")).strip() if pressure_contract: cmd.extend(["--pressure-contract", pressure_contract]) if events: cmd.extend(["--events", str(events)]) if lanes: cmd.extend(["--lanes", str(lanes)]) if timeout_seconds: cmd.extend(["--timeout-seconds", str(timeout_seconds)]) if args.fixture_manifest: cmd.extend(["--fixture-manifest", str(args.fixture_manifest), "--candidate-id", args.candidate_id]) return cmd def pressure_gate( args: argparse.Namespace, env: dict[str, str], gate_dir: Path, profile: str, single_gate: str, label: str, timeout: int, *, requested_gate: str = "", events: int = 0, lanes: int = 0, ) -> dict[str, Any]: run_id = f"{args.run_id}-{label}-{short_sha(args.expected_sha)}-{utc_stamp()}" report_path = gate_dir / "pressure-report.json" pressure_env = dict(env) pressure_env["XIAOPENG_PRESSURE_GATE_TIMEOUT_SECONDS"] = str(timeout) pressure_env["XIAOPENG_PRESSURE_DB_READ_TIMEOUT_SECONDS"] = str(max(180, timeout)) cmd = build_pressure_runner_command( args, report_path=report_path, run_id=run_id, profile=profile, workers=args.workers, events=events, lanes=lanes, timeout_seconds=timeout, ) if single_gate: cmd.extend(["--single-gate", single_gate]) proc = run_cmd(cmd, env=pressure_env, cwd=None, timeout=timeout + 90, log_path=gate_dir / "pressure-subprocess.json") child = read_json(report_path) if report_path.exists() else {} child_verdict = str(child.get("verdict") or "") results = child.get("results") or [] result_verdicts = [str(row.get("verdict") or "") for row in results if isinstance(row, dict)] passed = proc.returncode == 0 and child_verdict.startswith("PASS") and all(v == "PASS" for v in result_verdicts) report = { "gate": label, "requestedGate": requested_gate, "status": f"PASS_{label}" if passed else f"BLOCK_{label}", "returncode": proc.returncode, "runId": run_id, "profile": profile, "singleGate": single_gate, "requestedWorkers": args.workers, "requestedEvents": events, "requestedLanes": lanes, "pressureTimeoutSeconds": timeout, "pressureReport": str(report_path), "childVerdict": child_verdict, "resultVerdicts": result_verdicts, "results": summarize_pressure_results(results), } write_json(gate_dir / "report.json", report) return report def materialized_gate(args: argparse.Namespace, env: dict[str, str], gate_dir: Path, label: str, events: int, lanes: int, workers: int, timeout: int) -> dict[str, Any]: # Keep it short enough for current schema limits while preserving uniqueness. run_id = f"m-{short_sha(args.expected_sha)}-{utc_stamp()}" report_path = gate_dir / "materialized-report.json" cmd = [ sys.executable, str(args.materialized_runner), "--out", str(report_path), "--run-id", run_id, "--repo", str(args.env_root / args.connectors_runtime_path), "--events", str(events), "--lanes", str(lanes), "--workers", str(workers), "--queue-class", args.queue_class, "--timeout", str(timeout), "--cleanup", ] proc = run_cmd(cmd, env=env, cwd=None, timeout=timeout + 90, log_path=gate_dir / "materialized-subprocess.json") child = read_json(report_path) if report_path.exists() else {} child_verdict = str(child.get("verdict") or "") passed = proc.returncode == 0 and child_verdict == "PASS" report = { "gate": label, "status": f"PASS_{label}" if passed else f"BLOCK_{label}", "returncode": proc.returncode, "runId": run_id, "materializedReport": str(report_path), "childVerdict": child_verdict, "result": summarize_materialized_result(child), } write_json(gate_dir / "report.json", report) return report def summarize_pressure_results(results: list[Any]) -> list[dict[str, Any]]: keys = [ "accepted", "rejected", "unbound_accepted", "verified_receipt_count", "inline_event_id_count", "event_primary_key_count", "receipt_status_counts", "receipt_materialization_seconds", "materialized", "work_materialized", "processed", "skipped", "claimed", "failed", "retry", "dead", "nonterminal", "first_claim_latency_ms", "p50_claim_latency_ms", "p95_claim_latency_ms", "max_claim_latency_ms", "drain_seconds", "terminal_elapsed_seconds", "post_admission_drain_seconds", "admission_seconds", "admission_concurrency", "admission_events_per_second", "admission_request_p95_ms", "admission_request_max_ms", "terminal_events_per_second", "terminal_events_per_minute", "business_oracle_ok", "business_expected_source_count", "business_observed_record_count", "business_matching_source_count", "missing_source_count", "duplicate_source_count", "mismatched_source_count", "max_active_workers", "max_valid_simultaneous_leases", "max_observed_leased_work_items", "max_observed_leased_lanes", "max_observed_lease_tokens", "lease_invariant_violations", "claim_eligible_count", "lane_ordering_ok", "lane_order_violations", "explain_has_broad_scan", "explain_available", "active_pack_match", "candidate_claim_plan", "scheduler_lease_read_path_used", "verdict", ] summary: list[dict[str, Any]] = [] for row in results: if not isinstance(row, dict): continue gate = row.get("gate") if isinstance(row.get("gate"), dict) else {} item = { "name": gate.get("name") if isinstance(gate, dict) else "", "events": gate.get("events") if isinstance(gate, dict) else None, "lanes": gate.get("lanes") if isinstance(gate, dict) else None, "workers": gate.get("workers") if isinstance(gate, dict) else None, "expectedBulkEligibleWorkers": gate.get("expected_bulk_eligible_workers") if isinstance(gate, dict) else None, "expectation": gate.get("expectation") if isinstance(gate, dict) else None, "releaseBlocking": gate.get("release_blocking") if isinstance(gate, dict) else None, "correctnessRequired": gate.get("correctness_required") if isinstance(gate, dict) else None, "description": gate.get("description") if isinstance(gate, dict) else None, "nonPassPolicy": gate.get("non_pass_policy") if isinstance(gate, dict) else None, } item.update({key: row.get(key) for key in keys if key in row}) claim_scope = row.get("claim_scope") if isinstance(claim_scope, dict): item.update( { "installed_pack_id_sha256": claim_scope.get( "installed_pack_id_sha256" ), "generation": claim_scope.get("generation"), "claim_scope_workitem_count": claim_scope.get("workitem_count"), } ) summary.append(item) return summary def summarize_materialized_result(report: dict[str, Any]) -> dict[str, Any]: final = report.get("final") if isinstance(report.get("final"), dict) else {} width = report.get("max_lease_width") if isinstance(report.get("max_lease_width"), dict) else {} return { "events": report.get("events"), "lanes": report.get("lanes"), "workers_expected": report.get("workers_expected"), "queue_class": report.get("queue_class"), "duration_seconds": report.get("duration_seconds"), "first_claim_latency_ms": report.get("first_claim_latency_ms"), "max_processing_observed": report.get("max_processing_observed"), "max_lease_width": width, "event_terminal": final.get("event_terminal"), "work_terminal": final.get("work_terminal"), "eligible_heads": final.get("eligible_heads"), "claimed_or_terminal": final.get("claimed_or_terminal"), "split_brain": final.get("split_brain"), "explain_has_broad_scan": report.get("explain_has_broad_scan"), "verdict": report.get("verdict"), } def coalescer_log_gate(args: argparse.Namespace, env: dict[str, str], gate_dir: Path) -> dict[str, Any]: log_path = args.backend_log if not log_path.exists(): report = {"gate": "B2_CANONICAL_ALIAS_COALESCER_LOG", "status": "BLOCK_COALESCER_LOG_MISSING", "backendLog": str(log_path)} write_json(gate_dir / "report.json", report) return report text = read_file_tail(log_path, args.backend_log_tail_bytes) fallback_count = len(re.findall(r"fallback_single_reads=[1-9]", text)) old_global_error_count = len(re.findall(r"canonical alias demux requires one response record per requested ID", text)) coalesced_count = len(re.findall(r"connector_inbound_batch_read_coalesced", text)) report = { "gate": "B2_CANONICAL_ALIAS_COALESCER_LOG", "status": "PASS_COALESCER_LOG_GATE" if coalesced_count > 0 else "BLOCK_COALESCER_LOG_NO_COALESCED_READS", "backendLog": str(log_path), "tailBytes": args.backend_log_tail_bytes, "coalescedReadLogCount": coalesced_count, "fallbackSingleReadLogCount": fallback_count, "canonicalAliasGlobalErrorMentionsInTail": old_global_error_count, } write_json(gate_dir / "report.json", report) return report def read_file_tail(path: Path, max_bytes: int) -> str: if max_bytes <= 0: max_bytes = 200_000 with path.open("rb") as fh: fh.seek(0, os.SEEK_END) size = fh.tell() fh.seek(max(0, size - max_bytes), os.SEEK_SET) return fh.read(max_bytes).decode(errors="replace") def gate_passed(report: dict[str, Any]) -> bool: status = str(report.get("status") or "") return status.startswith("PASS_") def resolve_scale_gate(args: argparse.Namespace) -> tuple[dict[str, Any], str]: contract = getattr(args, "pressure_contract_data", None) or load_pressure_matrix_contract( getattr(args, "pressure_contract", None) ) spec = build_scale_gate( workers=args.workers, events=args.pressure_events, source_lanes=args.pressure_lanes or None, timeout_seconds=args.pressure_timeout or None, contract=contract, ) label = ( f"S1_INBOUND_{spec['events']}_EVENTS_{spec['lanes']}_LANES_" f"WORKER{spec['workers']}" ) return spec, label def worker_bound_label(base: str, workers: int) -> str: return f"{base}_16" if workers == 16 else f"{base}_WORKER{workers}" def run_matrix(args: argparse.Namespace) -> dict[str, Any]: args.env_root = Path(args.env_root) args.runtime_env = Path(args.runtime_env) if args.runtime_env else args.env_root / "env" / "runtime.env" args.preflight_script = Path(args.preflight_script) args.pressure_runner = Path(args.pressure_runner) args.materialized_runner = Path(args.materialized_runner) args.backend_log = Path(args.backend_log) if args.backend_log else args.env_root / "logs" / "local-20260708T142941Z" / "backend.stderr.log" args.connectors_runtime_path = Path(args.connectors_runtime_path) runtime_env = parse_env_file(args.runtime_env) bind_runtime_inputs(args, runtime_env) test_slot_identity_verified = validate_test_slot_runtime_identity(args, runtime_env) default_day = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%d") args.output_dir = ( Path(args.output_dir) if args.output_dir else args.env_root / "artifacts" / "hubspot-workitem-pressure" / f"{args.target_env}-{default_day}" / f"connected-work-matrix-{short_sha(args.expected_sha)}-{utc_stamp()}" ) args.output_dir.mkdir(parents=True, exist_ok=True) runtime_env["RUNTIME_ENV_SOURCE_FILE"] = str(args.runtime_env) child_env = make_child_env(runtime_env) requested = [item.strip() for item in args.gates.split(",") if item.strip()] pressure_gates = PRESSURE_FIXTURE_GATES.intersection(requested) if pressure_gates and not args.require_worker_count: raise GateFailure( "BLOCK_PRESSURE_WORKER_READBACK_REQUIRED", "pressure gates require exact deployed worker-count readback", ) validate_fixture_manifest_requirement( requested, args.fixture_manifest, args.candidate_id, args.target_env, test_slot_identity_verified=test_slot_identity_verified, ) args.require_hubspot_scale_route = "workitem-scale" in requested args.pressure_contract_data = ( load_pressure_matrix_contract(args.pressure_contract) if args.require_hubspot_scale_route else {} ) scale_spec, scale_label = resolve_scale_gate(args) if args.require_hubspot_scale_route else ({}, "") worker = args.workers pilot_name = "pilot_20_events_5_lanes" if worker == 16 else f"pilot_20_events_5_lanes_worker{worker}" noop_pilot_name = "noop_pilot_20_events_20_lanes" if worker == 16 else f"noop_pilot_20_events_20_lanes_worker{worker}" gate_specs = { "runtime": lambda d: health_gate(args, child_env, d), "secret": lambda d: preflight_gate(args, child_env, d), "hubspot-read": lambda d: hubspot_read_gate(args, child_env, d), "workitem-pilot": lambda d: pressure_gate(args, child_env, d, "noop-pilot", noop_pilot_name, worker_bound_label("E0_WORKITEM_CLEAN_PILOT_20_20", worker), args.pilot_timeout, requested_gate="workitem-pilot"), "normal-pilot": lambda d: pressure_gate(args, child_env, d, "pilot", pilot_name, worker_bound_label("D1F1_NORMAL_INBOUND_PILOT_20_5", worker), args.pilot_timeout, requested_gate="normal-pilot"), "coalescer-log": lambda d: coalescer_log_gate(args, child_env, d), "workitem-materialized": lambda d: materialized_gate(args, child_env, d, worker_bound_label("E1_WORKITEM_MATERIALIZED_800_120", worker), 800, 120, worker, args.materialized_timeout), "workitem-clean": lambda d: pressure_gate(args, child_env, d, "noop-full", f"noop_worker{worker}_800_events_120_lanes", worker_bound_label("E1B_ADMISSION_CLEAN_800_120", worker), args.workitem_timeout, requested_gate="workitem-clean"), "normal800": lambda d: pressure_gate(args, child_env, d, "full", f"worker{worker}_800_events_120_lanes", worker_bound_label("G1_NORMAL_INBOUND_800_120", worker), args.normal800_timeout, requested_gate="normal800"), "workitem-scale": lambda d: pressure_gate( args, child_env, d, "scale", "", scale_label, int(scale_spec["timeout_seconds"]), requested_gate="workitem-scale", events=int(scale_spec["events"]), lanes=int(scale_spec["lanes"]), ), } reports: list[dict[str, Any]] = [] pass_by_gate: dict[str, bool] = {} for gate_name in requested: if gate_name not in gate_specs: raise GateFailure("BLOCK_MATRIX_UNKNOWN_GATE", f"unknown gate: {gate_name}") required_gates = GATE_DEPENDENCIES.get(gate_name, []) dependencies_enforced = gate_name != "normal800" or not args.force_normal800 if required_gates and dependencies_enforced: missing = [name for name in required_gates if not pass_by_gate.get(name)] if missing: gate_label = scale_label if gate_name == "workitem-scale" else f"G1_NORMAL_INBOUND_800_120_WORKER{worker}" report = { "gate": gate_label, "requestedGate": gate_name, "status": f"BLOCK_{gate_name.upper().replace('-', '_')}_PREREQUISITE_GATES", "requiredPassGates": required_gates, "missingPassGates": missing, "omittedRequiredGates": [name for name in required_gates if name not in requested], } write_json(args.output_dir / gate_name / "report.json", report) reports.append(report) if args.stop_on_fail: break continue gate_dir = args.output_dir / gate_name gate_dir.mkdir(parents=True, exist_ok=True) try: report = gate_specs[gate_name](gate_dir) except subprocess.TimeoutExpired as exc: report = {"gate": gate_name, "status": f"BLOCK_{gate_name.upper().replace('-', '_')}_TIMEOUT", "timeoutSeconds": exc.timeout} write_json(gate_dir / "report.json", report) except GateFailure as exc: report = {"gate": gate_name, "status": exc.status, "message": exc.message} write_json(gate_dir / "report.json", report) report.setdefault("requestedGate", gate_name) write_json(gate_dir / "report.json", report) reports.append(report) pass_by_gate[gate_name] = gate_passed(report) write_running_summary(args, reports) if args.stop_on_fail and not gate_passed(report): break all_pass = bool(reports) and all(gate_passed(report) for report in reports) summary = { "matrix": "CONNECTED_WORK_MATRIX", "matrixVersion": MATRIX_VERSION, "status": f"PASS_CONNECTED_WORK_MATRIX_{short_sha(args.expected_sha)}" if all_pass else f"BLOCK_CONNECTED_WORK_MATRIX_{short_sha(args.expected_sha)}", "generatedAt": dt.datetime.now(dt.timezone.utc).isoformat(), "outputDir": str(args.output_dir), "expectedSha": args.expected_sha, "baseUrl": args.base_url, "targetEnv": args.target_env, "runtimeEnvSourceFile": str(args.runtime_env), "forceNormal800": args.force_normal800, "gateDependencies": GATE_DEPENDENCIES, "fixtureManifest": str(args.fixture_manifest) if args.fixture_manifest else "", "candidateId": args.candidate_id, "requestedGates": requested, "pressureCell": scale_spec if "workitem-scale" in requested else {}, "gates": reports, } test1_evidence_path = emit_test1_connected_work_evidence( args, summary, runtime_env, ) if test1_evidence_path is not None: summary["test1Evidence"] = str(test1_evidence_path) write_json(args.output_dir / "matrix_summary.json", summary) write_markdown_summary(args.output_dir / "README.md", summary) return summary def emit_test1_connected_work_evidence( args: argparse.Namespace, summary: dict[str, Any], runtime_env: Mapping[str, str], ) -> Path | None: """Persist the strict certificate only for the Test1 execution boundary.""" if args.target_env != "test1": return None requested_gates = frozenset(summary.get("requestedGates") or []) if requested_gates != TEST1_FIXTURELESS_EVIDENCE_GATES: return None path = Path(args.output_dir) / "test1-connected-work-evidence.json" try: report = build_test1_connected_work_evidence(summary) write_test1_connected_work_evidence( path, report, sensitive_values=sensitive_values_for_mapping(runtime_env), ) except Test1EvidenceBlocked as exc: raise GateFailure(exc.verdict, exc.message) from exc return path def write_running_summary(args: argparse.Namespace, reports: list[dict[str, Any]]) -> None: summary = { "matrix": "CONNECTED_WORK_MATRIX", "matrixVersion": MATRIX_VERSION, "status": f"RUNNING_CONNECTED_WORK_MATRIX_{short_sha(args.expected_sha)}", "generatedAt": dt.datetime.now(dt.timezone.utc).isoformat(), "outputDir": str(args.output_dir), "expectedSha": args.expected_sha, "baseUrl": args.base_url, "targetEnv": args.target_env, "runtimeEnvSourceFile": str(args.runtime_env), "forceNormal800": args.force_normal800, "gateDependencies": GATE_DEPENDENCIES, "fixtureManifest": str(args.fixture_manifest) if args.fixture_manifest else "", "candidateId": args.candidate_id, "gates": reports, } write_json(args.output_dir / "matrix_summary.partial.json", summary) write_markdown_summary(args.output_dir / "README.partial.md", summary) def write_markdown_summary(path: Path, summary: dict[str, Any]) -> None: lines = [ f"# {summary['matrix']} {summary['status']}", "", f"- generatedAt: {summary['generatedAt']}", f"- expectedSha: {summary['expectedSha']}", f"- targetEnv: {summary['targetEnv']}", f"- baseUrl: {summary['baseUrl']}", f"- runtimeEnvSourceFile: {summary['runtimeEnvSourceFile']}", "", "| Gate | Status | Artifact |", "|---|---|---|", ] for report in summary.get("gates") or []: gate = str(report.get("gate") or "") status = str(report.get("status") or "") artifact = "" if summary.get("outputDir"): artifact_dir = str(report.get("requestedGate") or gate_name_to_dir(gate)) artifact = str(Path(summary["outputDir"]) / artifact_dir / "report.json") lines.append(f"| `{gate}` | `{status}` | `{artifact}` |") path.write_text("\n".join(lines) + "\n") def gate_name_to_dir(gate: str) -> str: mapping = { "A1_RUNTIME_REF_SCHEMA": "runtime", "A2_INPUT_SECRET_SIGNATURE": "secret", "B1_HUBSPOT_READ_ONLY": "hubspot-read", "B2_CANONICAL_ALIAS_COALESCER_LOG": "coalescer-log", "E0_WORKITEM_CLEAN_PILOT_20_20_16": "workitem-pilot", "D1F1_NORMAL_INBOUND_PILOT_20_5_16": "normal-pilot", "E1_WORKITEM_MATERIALIZED_800_120_16": "workitem-materialized", "E1B_ADMISSION_CLEAN_800_120_16": "workitem-clean", "G1_NORMAL_INBOUND_800_120_16": "normal800", } return mapping.get(gate, gate.lower().replace("_", "-")) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Reusable Connected Work matrix gate runner for deployed environment validation.") parser.add_argument("--env-root", default=str(DEFAULT_ENV_ROOT)) parser.add_argument("--runtime-env", default="") parser.add_argument("--base-url", default=DEFAULT_BASE_URL) parser.add_argument("--target-env", default=DEFAULT_TARGET_ENV) parser.add_argument("--expected-sha", default="") parser.add_argument("--connectors-runtime-path", default=DEFAULT_RUNTIME_CONNECTORS_PATH) parser.add_argument("--preflight-script", default=str(DEFAULT_PREFLIGHT_SCRIPT)) parser.add_argument("--pressure-runner", default=str(DEFAULT_PRESSURE_RUNNER)) parser.add_argument("--materialized-runner", default=str(DEFAULT_MATERIALIZED_RUNNER)) parser.add_argument("--deployment-config", default=str(DEFAULT_DEPLOYMENT_CONFIG)) parser.add_argument("--backend-log", default="") parser.add_argument("--output-dir", default="") parser.add_argument("--run-id", default="cw-matrix") parser.add_argument("--fixture-manifest", default="") parser.add_argument("--candidate-id", default="") parser.add_argument("--gates", default="runtime,secret,hubspot-read,workitem-pilot,normal-pilot,coalescer-log,workitem-materialized,workitem-clean,normal800") parser.add_argument("--list-gates", action="store_true", help="Print reusable gate names and descriptions, then exit.") parser.add_argument("--stop-on-fail", action=argparse.BooleanOptionalAction, default=True) parser.add_argument("--force-normal800", action="store_true") parser.add_argument("--require-schema-authority", action=argparse.BooleanOptionalAction, default=True) parser.add_argument("--require-workitem-claim", action=argparse.BooleanOptionalAction, default=False) parser.add_argument("--require-worker-count", action=argparse.BooleanOptionalAction, default=True) parser.add_argument("--require-max-processing-rows", action=argparse.BooleanOptionalAction, default=False) parser.add_argument("--require-queue-profile", action=argparse.BooleanOptionalAction, default=False) parser.add_argument("--require-test-slot-identity", action=argparse.BooleanOptionalAction, default=False) parser.add_argument("--client-secret-env", default=DEFAULT_CLIENT_SECRET_ENV) parser.add_argument("--hubspot-token-env", default=DEFAULT_ACCESS_TOKEN_ENV) parser.add_argument("--portal-id", type=int, default=0) parser.add_argument("--app-id", type=int, default=0) parser.add_argument("--workers", type=int, default=16) parser.add_argument("--pressure-contract", default=os.environ.get("HUBSPOT_PRESSURE_MATRIX_CONTRACT", "")) parser.add_argument("--pressure-events", type=int, default=0) parser.add_argument("--pressure-lanes", type=int, default=0, help="0 uses the pressure matrix contract default") parser.add_argument("--pressure-timeout", type=int, default=0, help="0 uses the pressure matrix contract default") parser.add_argument("--max-processing-rows", type=int, default=32) parser.add_argument("--queue-profile", default="default-heavy") parser.add_argument("--queue-class", default="bulk") parser.add_argument("--hubspot-required-ids", type=int, default=120) parser.add_argument("--hubspot-batch-size", type=int, default=100) parser.add_argument("--preflight-timeout", type=int, default=120) parser.add_argument("--pilot-timeout", type=int, default=240) parser.add_argument("--materialized-timeout", type=int, default=720) parser.add_argument("--workitem-timeout", type=int, default=900) parser.add_argument("--normal800-timeout", type=int, default=1200) parser.add_argument("--backend-log-tail-bytes", type=int, default=2_000_000) return parser.parse_args() def main() -> int: args = parse_args() if args.list_gates: print(json.dumps({"matrixVersion": MATRIX_VERSION, "gates": GATE_DESCRIPTIONS, "dependencies": GATE_DEPENDENCIES}, indent=2, sort_keys=True)) return 0 try: summary = run_matrix(args) except GateFailure as exc: print(json.dumps({"status": exc.status, "message": exc.message}, sort_keys=True)) return 1 print(json.dumps({"status": summary["status"], "outputDir": summary["outputDir"], "summary": str(Path(summary["outputDir"]) / "matrix_summary.json")}, sort_keys=True)) return 0 if str(summary["status"]).startswith("PASS_") else 1 if __name__ == "__main__": raise SystemExit(main())