From 028dc13b35a24169cf44351da18258bb8fa3c861 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 11 Jun 2026 15:29:39 +0000 Subject: [PATCH] feat(tooling): JSON-output sample gate (G9) + --sample fixtures (#654) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements issue #654 Option A (embedded-sample convention) plus the verification harness the issue asked for: - scripts/smoke_json_output.py — new advisory gate (G9) that discovers every tool whose --help advertises JSON output, runs --sample , and asserts the stdout parses as JSON. Tools advertising JSON without --sample are reported as 'uncovered' (a backlog, not a failure) so the gate can be adopted incrementally; --strict flips that to a hard failure once coverage is high. Wired into ci-quality-gate.yml alongside G8. - Added --sample embedded fixtures to the 5 tools named in #654: error_budget_calculator, slo_review, blast_radius_calculator, audit_log_analyzer, api_linter. Their required args are now optional when --sample is passed; missing-arg behavior is unchanged otherwise. - Fixed 4 tools the new gate surfaced (prompt_rater, coach_tip_classifier, cheat_code_filter, redaction_linter): their --sample path printed human text and ignored --json; it now honors the JSON flag. - Synced the 3 dual-published standalone copies (slo-architect x2, chaos-engineering) so the drift guard stays green. Gate now reports 16 tools covered, 16 verified, 0 failures. https://claude.ai/code/session_01CUWsrUNZP9jpxvAwq67UiT --- .github/workflows/ci-quality-gate.yml | 5 + .../scripts/blast_radius_calculator.py | 17 +- .../claude-coach/scripts/cheat_code_filter.py | 10 +- .../scripts/coach_tip_classifier.py | 11 +- .../claude-coach/scripts/prompt_rater.py | 11 +- .../api-design-reviewer/scripts/api_linter.py | 66 ++++- .../scripts/blast_radius_calculator.py | 17 +- .../scripts/audit_log_analyzer.py | 50 +++- .../scripts/error_budget_calculator.py | 12 +- .../slo-architect/scripts/slo_review.py | 33 ++- .../scripts/error_budget_calculator.py | 12 +- .../slo-architect/scripts/slo_review.py | 33 ++- .../handoff/scripts/redaction_linter.py | 47 ++-- scripts/smoke_json_output.py | 227 ++++++++++++++++++ 14 files changed, 473 insertions(+), 78 deletions(-) create mode 100755 scripts/smoke_json_output.py diff --git a/.github/workflows/ci-quality-gate.yml b/.github/workflows/ci-quality-gate.yml index 897f847b..918c7026 100644 --- a/.github/workflows/ci-quality-gate.yml +++ b/.github/workflows/ci-quality-gate.yml @@ -113,6 +113,11 @@ jobs: run: | python3 scripts/smoke_scripts.py + - name: JSON-output sample gate (gate G9 — advisory) + continue-on-error: true + run: | + python3 scripts/smoke_json_output.py + - name: Counter derivation check (gate G3 — advisory) continue-on-error: true run: | diff --git a/engineering/chaos-engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py b/engineering/chaos-engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py index 1ab87664..1589ad6f 100755 --- a/engineering/chaos-engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py +++ b/engineering/chaos-engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py @@ -70,9 +70,11 @@ def render_text(result): def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - ap.add_argument("--traffic-share", type=float, required=True, help="Fraction (0-1) of traffic affected") - ap.add_argument("--user-pop", type=int, required=True, help="Total user population") - ap.add_argument("--duration-min", type=int, required=True, help="Experiment duration in minutes") + ap.add_argument("--traffic-share", type=float, help="Fraction (0-1) of traffic affected") + ap.add_argument("--user-pop", type=int, help="Total user population") + ap.add_argument("--duration-min", type=int, help="Experiment duration in minutes") + ap.add_argument("--sample", action="store_true", + help="Run with embedded sample inputs (5%% traffic, 100k users, 30 min)") ap.add_argument("--baseline-availability", type=float, default=0.999, help="Baseline availability (default: 0.999)") ap.add_argument("--expected-impact-availability", type=float, default=0.95, dest="impact_avail", help="Availability under fault (default: 0.95)") @@ -81,9 +83,16 @@ def main(): ap.add_argument("--format", choices=["text", "json"], default="text") args = ap.parse_args() + if args.sample: + traffic_share, user_pop, duration_min = 0.05, 100000, 30 + elif None not in (args.traffic_share, args.user_pop, args.duration_min): + traffic_share, user_pop, duration_min = args.traffic_share, args.user_pop, args.duration_min + else: + ap.error("--traffic-share, --user-pop and --duration-min are required (or use --sample)") + try: result = calculate( - args.traffic_share, args.user_pop, args.duration_min, + traffic_share, user_pop, duration_min, args.baseline_availability, args.impact_avail, args.monthly_budget_min, ) except ValueError as e: diff --git a/engineering/claude-coach/skills/claude-coach/scripts/cheat_code_filter.py b/engineering/claude-coach/skills/claude-coach/scripts/cheat_code_filter.py index 976b62cc..c447c5f9 100644 --- a/engineering/claude-coach/skills/claude-coach/scripts/cheat_code_filter.py +++ b/engineering/claude-coach/skills/claude-coach/scripts/cheat_code_filter.py @@ -118,12 +118,16 @@ def render_human(picks: list[Technique]) -> str: return "\n".join(out) -def sample_run() -> int: +def sample_run(as_json: bool = False) -> int: sample_path = DEFAULT_GLOSSARY if not sample_path.exists(): print("Sample glossary not found; place references/cheat-codes.md alongside this script.", file=sys.stderr) return 1 - picks = rank(parse_glossary(sample_path), ["writing", "coding"], 5) + use_cases = ["writing", "coding"] + picks = rank(parse_glossary(sample_path), use_cases, 5) + if as_json: + print(json.dumps({"use_cases": use_cases, "picks": [asdict(t) for t in picks]}, indent=2)) + return 0 print(render_human(picks)) return 0 @@ -138,7 +142,7 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) if args.sample: - return sample_run() + return sample_run(args.json) if not args.use_cases: parser.error("--use-cases is required unless --sample is passed") diff --git a/engineering/claude-coach/skills/claude-coach/scripts/coach_tip_classifier.py b/engineering/claude-coach/skills/claude-coach/scripts/coach_tip_classifier.py index 997a8891..c567457a 100644 --- a/engineering/claude-coach/skills/claude-coach/scripts/coach_tip_classifier.py +++ b/engineering/claude-coach/skills/claude-coach/scripts/coach_tip_classifier.py @@ -164,7 +164,7 @@ def render_human(d: Decision) -> str: return "\n".join(out) -def sample_run() -> int: +def sample_run(as_json: bool = False) -> int: cases = [ ("Can you help me with my email?", False), ("Write a 200-word product description for a noise-cancelling headphone targeting remote workers, focused on the focus-time benefit, no marketing fluff.", False), @@ -172,8 +172,11 @@ def sample_run() -> int: ("Can you make this better?", True), ("stop with the tips, just rewrite it", False), ] - for prompt, prev in cases: - d = classify(prompt, previous_tip_given=prev) + decisions = [classify(prompt, previous_tip_given=prev) for prompt, prev in cases] + if as_json: + print(json.dumps([asdict(d) for d in decisions], indent=2)) + return 0 + for d in decisions: print(render_human(d)) print("-" * 60) return 0 @@ -188,7 +191,7 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) if args.sample: - return sample_run() + return sample_run(args.json) if not args.prompt: parser.error("--prompt is required unless --sample is passed") diff --git a/engineering/claude-coach/skills/claude-coach/scripts/prompt_rater.py b/engineering/claude-coach/skills/claude-coach/scripts/prompt_rater.py index 02091683..11e5f49d 100644 --- a/engineering/claude-coach/skills/claude-coach/scripts/prompt_rater.py +++ b/engineering/claude-coach/skills/claude-coach/scripts/prompt_rater.py @@ -138,14 +138,17 @@ def render_human(r: Rating) -> str: ) -def sample_run() -> int: +def sample_run(as_json: bool = False) -> int: samples = [ "Can you help me with my email?", "Write a 200-word product description for a noise-cancelling headphone targeting remote workers, focused on the focus-time benefit, no marketing fluff.", "thoughts?", ] - for s in samples: - r = rate(s) + ratings = [rate(s) for s in samples] + if as_json: + print(json.dumps([asdict(r) for r in ratings], indent=2)) + return 0 + for r in ratings: print(render_human(r)) print("-" * 60) return 0 @@ -159,7 +162,7 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) if args.sample: - return sample_run() + return sample_run(args.json) if not args.prompt: parser.error("--prompt is required unless --sample is passed") diff --git a/engineering/skills/api-design-reviewer/scripts/api_linter.py b/engineering/skills/api-design-reviewer/scripts/api_linter.py index 53637d5e..6bd4c919 100644 --- a/engineering/skills/api-design-reviewer/scripts/api_linter.py +++ b/engineering/skills/api-design-reviewer/scripts/api_linter.py @@ -826,6 +826,35 @@ class APILinter: return "\n".join(report_lines) +# Embedded sample OpenAPI spec — intentionally imperfect (a verb in a URL, a +# snake_case property) so --sample produces a representative report. +SAMPLE_OPENAPI_SPEC = { + "openapi": "3.0.0", + "info": {"title": "Sample API", "version": "1.0.0"}, + "servers": [{"url": "https://api.example.com"}], + "paths": { + "/user-profiles/{userId}": { + "get": { + "summary": "Get a user profile", + "responses": {"200": {"description": "OK"}, "404": {"description": "Not found"}}, + "parameters": [{"name": "userId", "in": "path", "required": True}], + } + }, + "/user-profiles/create": { + "post": { + "summary": "Create a user profile (verb-in-URL anti-pattern)", + "responses": {"201": {"description": "Created"}}, + } + }, + }, + "components": { + "schemas": { + "UserProfile": {"properties": {"first_name": {"type": "string"}}} + } + }, +} + + def main(): """Main CLI entry point.""" parser = argparse.ArgumentParser( @@ -836,13 +865,21 @@ Examples: python api_linter.py openapi.json python api_linter.py --format json openapi.json > report.json python api_linter.py --raw-endpoints endpoints.json + python api_linter.py --sample --format json """ ) - + parser.add_argument( 'input_file', + nargs='?', help='Input file: OpenAPI/Swagger JSON file or raw endpoints JSON' ) + + parser.add_argument( + '--sample', + action='store_true', + help='Lint an embedded sample OpenAPI spec (no input file needed)' + ) parser.add_argument( '--format', @@ -863,17 +900,22 @@ Examples: ) args = parser.parse_args() - - # Load input file - try: - with open(args.input_file, 'r') as f: - input_data = json.load(f) - except FileNotFoundError: - print(f"Error: Input file '{args.input_file}' not found.", file=sys.stderr) - return 1 - except json.JSONDecodeError as e: - print(f"Error: Invalid JSON in '{args.input_file}': {e}", file=sys.stderr) - return 1 + + # Load input data — from the embedded sample or the input file + if args.sample: + input_data = SAMPLE_OPENAPI_SPEC + else: + if not args.input_file: + parser.error("input_file is required (or use --sample)") + try: + with open(args.input_file, 'r') as f: + input_data = json.load(f) + except FileNotFoundError: + print(f"Error: Input file '{args.input_file}' not found.", file=sys.stderr) + return 1 + except json.JSONDecodeError as e: + print(f"Error: Invalid JSON in '{args.input_file}': {e}", file=sys.stderr) + return 1 # Initialize linter and run analysis linter = APILinter() diff --git a/engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py b/engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py index 1ab87664..1589ad6f 100755 --- a/engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py +++ b/engineering/skills/chaos-engineering/scripts/blast_radius_calculator.py @@ -70,9 +70,11 @@ def render_text(result): def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - ap.add_argument("--traffic-share", type=float, required=True, help="Fraction (0-1) of traffic affected") - ap.add_argument("--user-pop", type=int, required=True, help="Total user population") - ap.add_argument("--duration-min", type=int, required=True, help="Experiment duration in minutes") + ap.add_argument("--traffic-share", type=float, help="Fraction (0-1) of traffic affected") + ap.add_argument("--user-pop", type=int, help="Total user population") + ap.add_argument("--duration-min", type=int, help="Experiment duration in minutes") + ap.add_argument("--sample", action="store_true", + help="Run with embedded sample inputs (5%% traffic, 100k users, 30 min)") ap.add_argument("--baseline-availability", type=float, default=0.999, help="Baseline availability (default: 0.999)") ap.add_argument("--expected-impact-availability", type=float, default=0.95, dest="impact_avail", help="Availability under fault (default: 0.95)") @@ -81,9 +83,16 @@ def main(): ap.add_argument("--format", choices=["text", "json"], default="text") args = ap.parse_args() + if args.sample: + traffic_share, user_pop, duration_min = 0.05, 100000, 30 + elif None not in (args.traffic_share, args.user_pop, args.duration_min): + traffic_share, user_pop, duration_min = args.traffic_share, args.user_pop, args.duration_min + else: + ap.error("--traffic-share, --user-pop and --duration-min are required (or use --sample)") + try: result = calculate( - args.traffic_share, args.user_pop, args.duration_min, + traffic_share, user_pop, duration_min, args.baseline_availability, args.impact_avail, args.monthly_budget_min, ) except ValueError as e: diff --git a/engineering/skills/secrets-vault-manager/scripts/audit_log_analyzer.py b/engineering/skills/secrets-vault-manager/scripts/audit_log_analyzer.py index b31e4e66..34b8068e 100644 --- a/engineering/skills/secrets-vault-manager/scripts/audit_log_analyzer.py +++ b/engineering/skills/secrets-vault-manager/scripts/audit_log_analyzer.py @@ -278,6 +278,32 @@ def print_human(result, threshold): print(" (* = off-hours)") +# Embedded synthetic audit log — exercises volume-spike + off-hours + failed-access +# detectors so --sample produces a non-trivial report without a real log file. +SAMPLE_ENTRIES = [ + {"timestamp": "2026-03-20T03:14:00Z", "type": "request", + "auth": {"display_name": "approle-payment-svc"}, + "request": {"path": "secret/data/production/payment/api-keys", "operation": "read"}, + "response": {"status_code": 200}, "remote_address": "10.0.1.15"}, + {"timestamp": "2026-03-20T03:15:00Z", "type": "request", + "auth": {"display_name": "approle-payment-svc"}, + "request": {"path": "secret/data/production/payment/db", "operation": "read"}, + "response": {"status_code": 200}, "remote_address": "10.0.1.99"}, + {"timestamp": "2026-03-20T03:16:00Z", "type": "request", + "auth": {"display_name": "approle-payment-svc"}, + "request": {"path": "secret/data/production/payment/jwt", "operation": "read"}, + "response": {"status_code": 403}, "remote_address": "203.0.113.7"}, + {"timestamp": "2026-03-20T03:17:00Z", "type": "request", + "auth": {"display_name": "approle-payment-svc"}, + "request": {"path": "secret/data/production/payment/jwt", "operation": "read"}, + "response": {"status_code": 403}, "remote_address": "203.0.113.7"}, + {"timestamp": "2026-03-20T14:00:00Z", "type": "request", + "auth": {"display_name": "ci-runner"}, + "request": {"path": "secret/data/ci/tokens", "operation": "read"}, + "response": {"status_code": 200}, "remote_address": "10.0.2.20"}, +] + + def main(): parser = argparse.ArgumentParser( description="Analyze Vault/cloud secret manager audit logs for anomalies.", @@ -299,7 +325,7 @@ def main(): %(prog)s --log-file audit.json --threshold 3 --json """), ) - parser.add_argument("--log-file", required=True, help="Path to audit log file (JSON lines or JSON array)") + parser.add_argument("--log-file", help="Path to audit log file (JSON lines or JSON array)") parser.add_argument( "--threshold", type=int, @@ -307,23 +333,35 @@ def main(): help="Anomaly sensitivity threshold — lower = more sensitive (default: 5)", ) parser.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON") + parser.add_argument("--sample", action="store_true", + help="Analyze an embedded synthetic audit log") args = parser.parse_args() - entries = load_logs(args.log_file) + if args.sample: + entries = SAMPLE_ENTRIES + log_file = "" + threshold = 2 + else: + if not args.log_file: + parser.error("--log-file is required (or use --sample)") + entries = load_logs(args.log_file) + log_file = args.log_file + threshold = args.threshold + if not entries: print("No log entries found in file.", file=sys.stderr) sys.exit(1) - result = analyze(entries, args.threshold) - result["log_file"] = args.log_file - result["threshold"] = args.threshold + result = analyze(entries, threshold) + result["log_file"] = log_file + result["threshold"] = threshold result["analyzed_at"] = datetime.now().isoformat() if args.json_output: print(json.dumps(result, indent=2)) else: - print_human(result, args.threshold) + print_human(result, threshold) if __name__ == "__main__": diff --git a/engineering/skills/slo-architect/scripts/error_budget_calculator.py b/engineering/skills/slo-architect/scripts/error_budget_calculator.py index 1fd9db0c..55ddedcc 100755 --- a/engineering/skills/slo-architect/scripts/error_budget_calculator.py +++ b/engineering/skills/slo-architect/scripts/error_budget_calculator.py @@ -125,13 +125,21 @@ def render_text(result): def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - ap.add_argument("--target", type=float, required=True, help="Target percent (e.g., 99.9)") + ap.add_argument("--target", type=float, help="Target percent (e.g., 99.9)") ap.add_argument("--window-days", type=int, default=28, help="Window in days (default: 28)") ap.add_argument("--format", choices=["text", "json"], default="text") + ap.add_argument("--sample", action="store_true", help="Run with embedded sample inputs (99.9%% / 28d)") args = ap.parse_args() + if args.sample: + target, window_days = 99.9, 28 + elif args.target is not None: + target, window_days = args.target, args.window_days + else: + ap.error("--target is required (or use --sample)") + try: - result = compute(args.target, args.window_days) + result = compute(target, window_days) except ValueError as e: print(f"ERROR: {e}", file=sys.stderr) return 2 diff --git a/engineering/skills/slo-architect/scripts/slo_review.py b/engineering/skills/slo-architect/scripts/slo_review.py index 83a19532..afb70fa6 100755 --- a/engineering/skills/slo-architect/scripts/slo_review.py +++ b/engineering/skills/slo-architect/scripts/slo_review.py @@ -64,8 +64,16 @@ def _has_cpu_as_sli(text): return False -def audit_one(path): - text = _read(path) +# Embedded sample SLO doc — intentionally flawed (target too high, CPU-as-SLI, +# no error budget policy) so --sample exercises several finding paths. +SAMPLE_SLO_DOC = """# Checkout API SLO +target: 99.995% +window_days: 28 +sli: cpu_usage below 80% +""" + + +def audit_text(text): findings = [] target = _parse_target(text) window_days = _parse_window_days(text) @@ -103,6 +111,10 @@ def audit_one(path): return findings +def audit_one(path): + return audit_text(_read(path)) + + def _walk(target): if os.path.isfile(target): yield target @@ -140,15 +152,20 @@ def render_text(results): def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - ap.add_argument("--slo-doc", required=True, help="Path to SLO doc or directory of docs") + ap.add_argument("--slo-doc", help="Path to SLO doc or directory of docs") ap.add_argument("--format", choices=["text", "json"], default="text") + ap.add_argument("--sample", action="store_true", help="Audit an embedded sample SLO doc") args = ap.parse_args() - if not os.path.exists(args.slo_doc): - print(f"ERROR: not found: {args.slo_doc}", file=sys.stderr) - return 2 - - results = audit(args.slo_doc) + if args.sample: + results = [{"path": "", "findings": audit_text(SAMPLE_SLO_DOC)}] + else: + if not args.slo_doc: + ap.error("--slo-doc is required (or use --sample)") + if not os.path.exists(args.slo_doc): + print(f"ERROR: not found: {args.slo_doc}", file=sys.stderr) + return 2 + results = audit(args.slo_doc) if args.format == "json": print(json.dumps(results, indent=2)) return 1 if any(f[0] == "FAIL" for r in results for f in r["findings"]) else 0 diff --git a/engineering/slo-architect/skills/slo-architect/scripts/error_budget_calculator.py b/engineering/slo-architect/skills/slo-architect/scripts/error_budget_calculator.py index 1fd9db0c..55ddedcc 100755 --- a/engineering/slo-architect/skills/slo-architect/scripts/error_budget_calculator.py +++ b/engineering/slo-architect/skills/slo-architect/scripts/error_budget_calculator.py @@ -125,13 +125,21 @@ def render_text(result): def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - ap.add_argument("--target", type=float, required=True, help="Target percent (e.g., 99.9)") + ap.add_argument("--target", type=float, help="Target percent (e.g., 99.9)") ap.add_argument("--window-days", type=int, default=28, help="Window in days (default: 28)") ap.add_argument("--format", choices=["text", "json"], default="text") + ap.add_argument("--sample", action="store_true", help="Run with embedded sample inputs (99.9%% / 28d)") args = ap.parse_args() + if args.sample: + target, window_days = 99.9, 28 + elif args.target is not None: + target, window_days = args.target, args.window_days + else: + ap.error("--target is required (or use --sample)") + try: - result = compute(args.target, args.window_days) + result = compute(target, window_days) except ValueError as e: print(f"ERROR: {e}", file=sys.stderr) return 2 diff --git a/engineering/slo-architect/skills/slo-architect/scripts/slo_review.py b/engineering/slo-architect/skills/slo-architect/scripts/slo_review.py index 83a19532..afb70fa6 100755 --- a/engineering/slo-architect/skills/slo-architect/scripts/slo_review.py +++ b/engineering/slo-architect/skills/slo-architect/scripts/slo_review.py @@ -64,8 +64,16 @@ def _has_cpu_as_sli(text): return False -def audit_one(path): - text = _read(path) +# Embedded sample SLO doc — intentionally flawed (target too high, CPU-as-SLI, +# no error budget policy) so --sample exercises several finding paths. +SAMPLE_SLO_DOC = """# Checkout API SLO +target: 99.995% +window_days: 28 +sli: cpu_usage below 80% +""" + + +def audit_text(text): findings = [] target = _parse_target(text) window_days = _parse_window_days(text) @@ -103,6 +111,10 @@ def audit_one(path): return findings +def audit_one(path): + return audit_text(_read(path)) + + def _walk(target): if os.path.isfile(target): yield target @@ -140,15 +152,20 @@ def render_text(results): def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - ap.add_argument("--slo-doc", required=True, help="Path to SLO doc or directory of docs") + ap.add_argument("--slo-doc", help="Path to SLO doc or directory of docs") ap.add_argument("--format", choices=["text", "json"], default="text") + ap.add_argument("--sample", action="store_true", help="Audit an embedded sample SLO doc") args = ap.parse_args() - if not os.path.exists(args.slo_doc): - print(f"ERROR: not found: {args.slo_doc}", file=sys.stderr) - return 2 - - results = audit(args.slo_doc) + if args.sample: + results = [{"path": "", "findings": audit_text(SAMPLE_SLO_DOC)}] + else: + if not args.slo_doc: + ap.error("--slo-doc is required (or use --sample)") + if not os.path.exists(args.slo_doc): + print(f"ERROR: not found: {args.slo_doc}", file=sys.stderr) + return 2 + results = audit(args.slo_doc) if args.format == "json": print(json.dumps(results, indent=2)) return 1 if any(f[0] == "FAIL" for r in results for f in r["findings"]) else 0 diff --git a/productivity/handoff/skills/handoff/scripts/redaction_linter.py b/productivity/handoff/skills/handoff/scripts/redaction_linter.py index f09cc2e6..33bd038f 100644 --- a/productivity/handoff/skills/handoff/scripts/redaction_linter.py +++ b/productivity/handoff/skills/handoff/scripts/redaction_linter.py @@ -209,6 +209,27 @@ def _format_human(report: Report, mode: str, path: Path) -> str: return "\n".join(lines) +def _report_json(report, mode: str, file_label: str) -> str: + return json.dumps( + { + "file": file_label, + "mode": mode, + "findings": [ + { + "line": f.line_number, + "pattern": f.pattern_name, + "severity": f.severity, + "match": f.match, + "suggestion": f.suggestion, + } + for f in report.findings + ], + "counts": report.by_severity(), + }, + indent=2, + ) + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Scan a handoff draft for secrets and PII.") parser.add_argument("file", nargs="?", help="Path to the handoff markdown file.") @@ -237,7 +258,10 @@ def main(argv: list[str] | None = None) -> int: "Allowed: AKIAIOSFODNN7EXAMPLE \n" ) report = scan_text(fixture) - print(_format_human(report, "strict", Path(""))) + if args.json: + print(_report_json(report, "strict", "")) + else: + print(_format_human(report, "strict", Path(""))) return 1 if report.findings else 0 if args.mode == "off": @@ -255,26 +279,7 @@ def main(argv: list[str] | None = None) -> int: report = scan_file(path) if args.json: - print( - json.dumps( - { - "file": str(path), - "mode": args.mode, - "findings": [ - { - "line": f.line_number, - "pattern": f.pattern_name, - "severity": f.severity, - "match": f.match, - "suggestion": f.suggestion, - } - for f in report.findings - ], - "counts": report.by_severity(), - }, - indent=2, - ) - ) + print(_report_json(report, args.mode, str(path))) else: print(_format_human(report, args.mode, path)) diff --git a/scripts/smoke_json_output.py b/scripts/smoke_json_output.py new file mode 100755 index 00000000..77ce7ec1 --- /dev/null +++ b/scripts/smoke_json_output.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +"""JSON-output verification gate for Python tools (audit gate G9). + +Companion to smoke_scripts.py (which only asserts `--help` exits 0). Many tools +advertise `--json` or `--format json` in their help text but require positional +or required arguments before they can emit anything — so a bare-flag smoke test +reports false failures (see issue #654). + +This harness verifies JSON output the way the tools are actually meant to run: + + 1. Discover every tool whose `--help` advertises JSON output AND an embedded + `--sample` fixture (the chosen convention — issue #654 Option A). + 2. Run ` --sample ` and assert stdout parses as JSON. + +Tools that advertise JSON output but do NOT yet expose `--sample` are reported +as "uncovered" — a to-do list for backporting the convention, not a failure +(so the gate can be adopted incrementally without going red on day one). +Pass --strict to treat uncovered JSON tools as failures once coverage is high. + +Exit codes: + 0 every --sample JSON tool produced valid JSON (and, with --strict, every + JSON-advertising tool exposes --sample) + 1 one or more --sample JSON runs produced invalid JSON / errored + 2 harness error + +Usage: + python3 scripts/smoke_json_output.py # human-readable report + python3 scripts/smoke_json_output.py --json # machine-readable report + python3 scripts/smoke_json_output.py --strict # uncovered JSON tools fail +""" +from __future__ import annotations + +import argparse +import concurrent.futures +import json +import os +import re +import subprocess +import sys + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +TIMEOUT_SECONDS = 20 + +# Reuse the same exclusion set + exceptions file as the --help gate. +EXCLUDE_DIRS = { + ".git", ".codex", ".gemini", ".hermes", ".vibe", "docs", "audit", + "node_modules", "integrations", "eval-workspace", "site", + "__pycache__", ".venv", "venv", +} +EXCEPTIONS_FILE = os.path.join(REPO_ROOT, "scripts", "smoke_exceptions.txt") + +# The smoke harnesses describe `--sample`/`--json` in their own help text but +# are gate runners, not analysis tools — never classify them as JSON tools. +SELF_SKIP = {"scripts/smoke_json_output.py", "scripts/smoke_scripts.py"} + +# `--format json` is only a valid invocation when help shows json as a choice, +# e.g. `--format {text,json}`. A bare mention of the word "format" elsewhere in +# help must not trigger it (that misfires on tools that only accept `--json`). +_FORMAT_JSON_RE = re.compile(r"--format[ =]?\{[^}]*\bjson\b[^}]*\}") + + +def load_exceptions(path): + exceptions = {} + if not os.path.isfile(path): + return exceptions + with open(path, "r", encoding="utf-8") as f: + for raw in f: + line = raw.strip() + if not line or line.startswith("#"): + continue + rel = line.split("#", 1)[0].strip() + if rel: + exceptions[rel] = True + return exceptions + + +def find_python_files(root): + files = [] + for dirpath, dirnames, filenames in os.walk(root): + dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDE_DIRS) + for name in sorted(filenames): + if name.endswith(".py"): + files.append(os.path.relpath( + os.path.join(dirpath, name), root).replace(os.sep, "/")) + return files + + +def _help_text(abs_path): + try: + proc = subprocess.run( + [sys.executable, abs_path, "--help"], + stdin=subprocess.DEVNULL, capture_output=True, text=True, + timeout=TIMEOUT_SECONDS, cwd=os.path.dirname(abs_path), + ) + except (subprocess.TimeoutExpired, OSError): + return "" + return (proc.stdout or "") + (proc.stderr or "") if proc.returncode == 0 else "" + + +def classify(rel_path): + """Return (advertises_json, json_flag, has_sample) for one tool.""" + if rel_path in SELF_SKIP: + return False, None, False + abs_path = os.path.join(REPO_ROOT, rel_path) + help_text = _help_text(abs_path) + if not help_text: + return False, None, False + low = help_text.lower() + # Prefer `--format json` only when help shows json as an actual choice; + # otherwise fall back to a plain `--json` flag. + json_flag = None + if _FORMAT_JSON_RE.search(low): + json_flag = ["--format", "json"] + elif re.search(r"(? --sample ` and check stdout parses as JSON.""" + abs_path = os.path.join(REPO_ROOT, rel_path) + try: + proc = subprocess.run( + [sys.executable, abs_path, "--sample", *json_flag], + stdin=subprocess.DEVNULL, capture_output=True, text=True, + timeout=TIMEOUT_SECONDS, cwd=os.path.dirname(abs_path), + ) + except subprocess.TimeoutExpired: + return rel_path, False, f"timeout after {TIMEOUT_SECONDS}s" + except OSError as exc: + return rel_path, False, f"could not execute: {exc}" + # A non-zero exit is acceptable only if the tool intentionally signals a + # finding through its exit code (e.g. blast_radius RED) — but it must still + # have emitted valid JSON on stdout. + out = (proc.stdout or "").strip() + if not out: + tail = (proc.stderr or "").strip().splitlines() + return rel_path, False, f"no stdout (exit {proc.returncode}): {tail[-1] if tail else ''}"[:200] + try: + json.loads(out) + except json.JSONDecodeError as exc: + return rel_path, False, f"stdout is not valid JSON: {exc}" + return rel_path, True, "" + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--json", action="store_true", + help="emit a JSON report instead of human-readable output") + parser.add_argument("--strict", action="store_true", + help="treat JSON-advertising tools without --sample as failures") + parser.add_argument("--jobs", type=int, default=os.cpu_count() or 4, + help="parallel workers (default: CPU count)") + parser.add_argument("--root", default=REPO_ROOT, help="repo root") + args = parser.parse_args(argv) + + try: + exceptions = load_exceptions(EXCEPTIONS_FILE) + except OSError as exc: + print(f"ERROR: cannot read exceptions file: {exc}", file=sys.stderr) + return 2 + + all_files = [f for f in find_python_files(args.root) if f not in exceptions] + + # Phase 1: classify in parallel. + json_tools = {} # rel_path -> json_flag + uncovered = [] # advertises json but no --sample + with concurrent.futures.ThreadPoolExecutor(max_workers=args.jobs) as pool: + results = pool.map(lambda f: (f, *classify(f)), all_files) + for rel_path, advertises, json_flag, has_sample in results: + if not advertises: + continue + if has_sample: + json_tools[rel_path] = json_flag + else: + uncovered.append(rel_path) + uncovered.sort() + + # Phase 2: verify covered tools in parallel. + failures = [] + with concurrent.futures.ThreadPoolExecutor(max_workers=args.jobs) as pool: + for rel_path, ok, detail in pool.map( + lambda item: verify_one(item[0], item[1]), sorted(json_tools.items())): + if not ok: + failures.append({"file": rel_path, "detail": detail}) + failures.sort(key=lambda f: f["file"]) + + covered = len(json_tools) + total_json = covered + len(uncovered) + coverage_pct = round(100 * covered / total_json, 1) if total_json else 100.0 + + if args.json: + print(json.dumps({ + "json_advertising_tools": total_json, + "covered_by_sample": covered, + "coverage_pct": coverage_pct, + "verified_ok": covered - len(failures), + "failed": failures, + "uncovered": uncovered, + }, indent=2)) + else: + print(f"JSON-advertising tools: {total_json}") + print(f"Covered by --sample: {covered} ({coverage_pct}%)") + print(f"Verified valid JSON: {covered - len(failures)}") + print(f"Failed: {len(failures)}") + print(f"Uncovered (no --sample): {len(uncovered)}") + if failures: + print("\nFAILURES:") + for f in failures: + print(f" {f['file']}\n {f['detail']}") + if uncovered: + print("\nUNCOVERED (advertise JSON but lack --sample — backport target):") + for f in uncovered: + print(f" {f}") + + if failures: + return 1 + if args.strict and uncovered: + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main())