fix(gate): keep quality-gate logs readable (BUG-1230)

Summarize successful steps, print the failing check and error text,
and split the validate job into seven steps. Local npm test stays TAP.
This commit is contained in:
jesse-ux
2026-10-04 22:58:17 +08:00
parent 06bfae92d8
commit 2c64b66423
11 changed files with 729 additions and 40 deletions
+184 -33
View File
@@ -8,9 +8,11 @@ import contextlib
import json
import os
import py_compile
import re
import subprocess
import sys
import tempfile
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
@@ -119,6 +121,8 @@ CORE_PYTEST_TARGETS = [
"tests/test_upstream_import_plan.py",
# Tracked-file privacy must run in staging's explicit quick test list.
"tests/test_repo_privacy_markers.py",
# Summary lines by default; failure keeps the last 200 lines and a full log (BUG-1230).
"tests/test_run_quality_gate_output.py",
"tests/test_upstream_git_import_b9a0ef8f.py",
"tests/test_interpretation_template_registry.py",
"tests/test_vedastro_external_technique_evidence.py",
@@ -331,6 +335,98 @@ def tail_text(text: str, *, limit: int = 2400) -> str:
return f"...\n{text[-limit:]}"
# Default output is one line per successful step. `--verbose` restores the
# previous passthrough so a local debugging run still shows child output.
VERBOSE = False
_step_ordinal = 0
_FAILURE_TAIL_LINES = 200
def set_verbose(enabled: bool) -> None:
global VERBOSE
VERBOSE = enabled
def reset_output_state() -> None:
global _step_ordinal
_step_ordinal = 0
def format_profile_banner(profile_name: str, profile: dict, *, verbose: bool) -> str:
if verbose:
body = json.dumps(profile, ensure_ascii=False, indent=2)
return f"\n== Quality gate profile: {profile_name} ==\n{body}"
flags = " ".join(f"{key}={str(value).lower()}" for key, value in profile.items())
return f"quality gate profile={profile_name} {flags}"
def _step_name(cmd: list[str], step: str | None) -> str:
if step:
return step
if len(cmd) >= 2 and str(cmd[1]).endswith(".py"):
return str(cmd[1]).replace("\\", "/")
if len(cmd) >= 3 and cmd[1] == "-m":
return str(cmd[2])
return " ".join(str(part) for part in cmd[:3])
def _slug(name: str) -> str:
slug = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-")
return slug[:80] or "step"
def _combine_output(stdout: str, stderr: str) -> str:
parts: list[str] = []
if stdout:
parts.append(stdout if stdout.endswith("\n") else f"{stdout}\n")
if stderr:
if stdout:
parts.append("===== stderr =====\n")
parts.append(stderr if stderr.endswith("\n") else f"{stderr}\n")
return "".join(parts)
def last_output_lines(text: str, count: int = _FAILURE_TAIL_LINES) -> str:
lines = text.splitlines()
if len(lines) <= count:
return "\n".join(lines)
return "\n".join(lines[-count:])
def _write_failure_log(name: str, output: str) -> Path:
directory = ROOT / "gate-logs"
directory.mkdir(parents=True, exist_ok=True)
path = directory / f"{_step_ordinal:02d}-{_slug(name)}.log"
path.write_text(output, encoding="utf-8")
return path
def _report_captured_failure(
label: str,
cmd: list[str],
returncode: int,
output: str,
*,
cwd: Path,
optional: bool,
) -> bool:
log_path = _write_failure_log(label, output)
relative = log_path.relative_to(ROOT).as_posix()
print(f"✗ {label} exit={returncode}", file=sys.stderr)
tail = last_output_lines(output)
if tail:
print(tail, file=sys.stderr)
print(f"full log: {relative}", file=sys.stderr)
if optional:
print(f"Optional step failed with exit code {returncode}; continuing.")
return False
print(
format_failure_summary(label, cmd, returncode, stdout="", stderr="", cwd=cwd),
file=sys.stderr,
)
raise SystemExit(returncode)
def extract_json_payload(text: str) -> dict:
text = text.strip()
if not text:
@@ -386,45 +482,92 @@ def format_failure_summary(
return "\n".join(lines)
def run(cmd: list[str], *, optional: bool = False, step: str | None = None, cwd: Path = ROOT) -> bool:
label = step or " ".join(cmd[:2])
print(f"\n$ {' '.join(cmd)}")
completed = subprocess.run(cmd, cwd=cwd, text=True)
def run(cmd: list[str], *, optional: bool = False, step: str | None = None, cwd: Path | None = None) -> bool:
"""Run one gate step.
Compact mode captures stdout and stderr: success prints one line, failure
prints the last 200 lines and writes the full output under ``gate-logs/``.
``--verbose`` inherits the child streams, matching the previous behavior.
Timeouts are unchanged: this wrapper does not add a ``timeout=`` of its own,
and ``--test-timeout`` / ``test_timeout_seconds`` stay profile metadata.
"""
global _step_ordinal
_step_ordinal += 1
if cwd is None:
cwd = ROOT
label = _step_name(cmd, step)
if VERBOSE:
print(f"\n$ {' '.join(cmd)}")
completed = subprocess.run(cmd, cwd=cwd, text=True)
if completed.returncode == 0:
return True
if optional:
print(f"Optional step failed with exit code {completed.returncode}; continuing.")
return False
print(
format_failure_summary(label, cmd, completed.returncode, stdout="", stderr="", cwd=cwd),
file=sys.stderr,
)
raise SystemExit(completed.returncode)
started = time.perf_counter()
completed = subprocess.run(cmd, cwd=cwd, text=True, capture_output=True, errors="replace")
elapsed = time.perf_counter() - started
if completed.returncode == 0:
print(f"✓ {label} {elapsed:.1f}s")
return True
if optional:
print(f"Optional step failed with exit code {completed.returncode}; continuing.")
return False
print(
format_failure_summary(label, cmd, completed.returncode, stdout="", stderr="", cwd=cwd),
file=sys.stderr,
)
raise SystemExit(completed.returncode)
output = _combine_output(completed.stdout or "", completed.stderr or "")
return _report_captured_failure(label, cmd, completed.returncode, output, cwd=cwd, optional=optional)
def run_oracle_collection_queue_and_validator() -> None:
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False, encoding="utf-8") as handle:
queue_path = Path(handle.name)
try:
print(f"\n$ {' '.join(ORACLE_COLLECTION_QUEUE_CMD)}")
completed = subprocess.run(ORACLE_COLLECTION_QUEUE_CMD, cwd=ROOT, text=True, capture_output=True)
if completed.stdout:
print(completed.stdout, end="" if completed.stdout.endswith("\n") else "\n")
if completed.stderr:
print(completed.stderr, end="" if completed.stderr.endswith("\n") else "\n", file=sys.stderr)
if completed.returncode != 0:
print(
format_failure_summary(
"oracle_collection_queue",
global _step_ordinal
_step_ordinal += 1
label = "oracle_collection_queue"
if VERBOSE:
print(f"\n$ {' '.join(ORACLE_COLLECTION_QUEUE_CMD)}")
completed = subprocess.run(ORACLE_COLLECTION_QUEUE_CMD, cwd=ROOT, text=True, capture_output=True)
if completed.stdout:
print(completed.stdout, end="" if completed.stdout.endswith("\n") else "\n")
if completed.stderr:
print(completed.stderr, end="" if completed.stderr.endswith("\n") else "\n", file=sys.stderr)
if completed.returncode != 0:
print(
format_failure_summary(
label,
ORACLE_COLLECTION_QUEUE_CMD,
completed.returncode,
stdout=completed.stdout,
stderr=completed.stderr,
),
file=sys.stderr,
)
raise SystemExit(completed.returncode)
else:
started = time.perf_counter()
completed = subprocess.run(
ORACLE_COLLECTION_QUEUE_CMD,
cwd=ROOT,
text=True,
capture_output=True,
errors="replace",
)
elapsed = time.perf_counter() - started
if completed.returncode != 0:
output = _combine_output(completed.stdout or "", completed.stderr or "")
_report_captured_failure(
label,
ORACLE_COLLECTION_QUEUE_CMD,
completed.returncode,
stdout=completed.stdout,
stderr=completed.stderr,
),
file=sys.stderr,
)
raise SystemExit(completed.returncode)
queue_path.write_text(completed.stdout, encoding="utf-8")
output,
cwd=ROOT,
optional=False,
)
print(f"✓ {label} {elapsed:.1f}s")
queue_path.write_text(completed.stdout or "", encoding="utf-8")
validator_cmd = [part if part != "{queue_file}" else str(queue_path) for part in ORACLE_EVIDENCE_VALIDATOR_CMD]
run(validator_cmd, step="oracle_evidence_validator")
finally:
@@ -506,19 +649,25 @@ def run_vedastro_live_smoke() -> None:
def compile_targets() -> None:
print("\n== Compile core Python files ==")
if VERBOSE:
print("\n== Compile core Python files ==")
targets: list[Path] = []
for directory in COMPILE_DIRS:
targets.extend(sorted(directory.glob("*.py")))
targets.extend(EXTRA_COMPILE_TARGETS)
seen: set[Path] = set()
started = time.perf_counter()
for target in targets:
if target in seen or not target.exists():
continue
seen.add(target)
print(f"compile {target.relative_to(ROOT)}")
if VERBOSE:
print(f"compile {target.relative_to(ROOT)}")
py_compile.compile(str(target), doraise=True)
if not VERBOSE:
elapsed = time.perf_counter() - started
print(f"✓ py_compile {len(seen)} files {elapsed:.1f}s")
def validate_json_files() -> None:
@@ -576,12 +725,14 @@ def main() -> int:
help="Fail an individual pytest call after SECONDS; defaults to the selected profile boundary.",
)
parser.add_argument("--require-external-parity", action="store_true", help="Fail the release gate unless the three-engine raw parity manifest passes.")
parser.add_argument("--verbose", action="store_true", help="Stream every command and its output. The default prints one summary line per successful step.")
args = parser.parse_args()
profile = run_profile(args)
set_verbose(args.verbose)
reset_output_state()
os.environ.setdefault("PYTHONPATH", str(ROOT / "scripts"))
print(f"\n== Quality gate profile: {args.profile} ==")
print(json.dumps(profile, ensure_ascii=False, indent=2))
print(format_profile_banner(args.profile, profile, verbose=args.verbose))
if args.profile == "runtime-truth":
for target in [
ROOT / "scripts" / "jyotish_api_server.py",