#!/usr/bin/env python3 """Dependency-free Embedded Semantics integration-readiness self-test. The self-test never contacts a network service and never performs a downstream side effect. It exercises a checked set of resolver-envelope fixtures and proves that only a structurally valid ``resolved`` response receives an exact ConceptCode and registryVersion. Unknown, ambiguous, malformed, request-error, and service-error fixtures remain explicit no-assignment outcomes. """ from __future__ import annotations import argparse import hashlib import json import sys from collections.abc import Mapping, Sequence from copy import deepcopy from pathlib import Path from typing import Any from site_assets import ( INTEGRATION_READINESS_CONTRACT_RELATIVE_PATH, INTEGRATION_READINESS_FIXTURES_RELATIVE_PATH, site_contract_path, ) INTEGRATION_READINESS_SCHEMA = "embedded-semantics.integration-readiness.v1" INTEGRATION_READINESS_REPORT_SCHEMA = "embedded-semantics.integration-readiness-report.v1" INTEGRATION_READINESS_FIXTURES_SCHEMA = "embedded-semantics.integration-readiness-fixtures.v1" REPORT_VERSION = "1.0.0" CANONICAL_RECIPE_IDS = ( "multilingual-memory", "workflow-routing", "api-event-interoperability", "clarification", "audit-provenance", ) ALWAYS_RETAINED_FIELDS = ( "originalExpression", "language", "semanticOutcome", "resolverEvidence", ) RESOLVED_ONLY_FIELDS = ("conceptCode", "registryVersion") REQUIRED_FIXTURE_IDS = ( "resolved-valid", "unknown-expression", "ambiguous-expression", "malformed-resolved", "request-error", "service-error", ) DEFAULT_AUTHORIZATION_POLICY_VERSION = "local-readiness-policy-v1" DEFAULT_ROOT = Path(__file__).resolve().parent DEFAULT_CONTRACT_PATH = site_contract_path(DEFAULT_ROOT, INTEGRATION_READINESS_CONTRACT_RELATIVE_PATH) DEFAULT_FIXTURES_PATH = site_contract_path(DEFAULT_ROOT, INTEGRATION_READINESS_FIXTURES_RELATIVE_PATH) def canonical_json_bytes(value: object) -> bytes: """Return stable UTF-8 JSON bytes for hashing and byte-for-byte comparison.""" return ( json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n" ).encode("utf-8") def pretty_json(value: object) -> str: """Return deterministic human-readable JSON with a trailing newline.""" return json.dumps(value, ensure_ascii=False, sort_keys=True, indent=2) + "\n" def sha256_bytes(data: bytes) -> str: """Return a lowercase SHA-256 digest.""" return hashlib.sha256(data).hexdigest() def sha256_json(value: object) -> str: """Return the SHA-256 of canonical JSON bytes.""" return sha256_bytes(canonical_json_bytes(value)) def load_json_object(path: Path) -> dict[str, Any]: """Load a checked JSON object and reject non-object documents.""" value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise ValueError(f"Expected a JSON object at {path}") return value def _non_empty_string(value: object) -> bool: """Return whether a value is a non-whitespace string without normalizing it.""" return isinstance(value, str) and bool(value.strip()) def _copy_evidence(value: object) -> object: """Return an independent JSON-compatible copy of observed fixture evidence.""" return deepcopy(value) def validate_contract(contract: Mapping[str, object]) -> tuple[str, ...]: """Validate the checked readiness contract without JSON Schema dependencies.""" errors: list[str] = [] if contract.get("schema") != INTEGRATION_READINESS_SCHEMA: errors.append("unexpected readiness contract schema identifier") if contract.get("version") != REPORT_VERSION: errors.append("unexpected readiness contract version") recipe_selection = contract.get("recipeSelection") if not isinstance(recipe_selection, Mapping): errors.append("recipeSelection must be an object") else: if recipe_selection.get("requiredCount") != 1: errors.append("exactly one recipe must be required") recipes = recipe_selection.get("recipes") if not isinstance(recipes, list) or len(recipes) != len(CANONICAL_RECIPE_IDS): errors.append("readiness contract must contain exactly five recipes") else: actual_ids = { str(recipe.get("id")) for recipe in recipes if isinstance(recipe, Mapping) and recipe.get("id") } if actual_ids != set(CANONICAL_RECIPE_IDS): errors.append("readiness contract recipe identifiers do not match the canonical five") for index, recipe in enumerate(recipes, start=1): if not isinstance(recipe, Mapping): errors.append(f"recipe {index} must be an object") continue for field in ( "id", "title", "fitStatement", "bestPlacement", "benefitMetrics", ): if field not in recipe: errors.append(f"recipe {index} missing {field}") record = contract.get("clientRecord") if not isinstance(record, Mapping): errors.append("clientRecord must be an object") else: if tuple(record.get("alwaysRetained", ())) != ALWAYS_RETAINED_FIELDS: errors.append("always-retained fields do not match the canonical record contract") if tuple(record.get("resolvedOnly", ())) != RESOLVED_ONLY_FIELDS: errors.append("resolved-only fields do not match the canonical record contract") if record.get("noAssignmentRule") != ( "Every non-resolved fixture must omit conceptCode and registryVersion." ): errors.append("clientRecord must preserve every non-resolved fixture as no assignment") fixture_set = contract.get("fixtureSet") if not isinstance(fixture_set, Mapping): errors.append("fixtureSet must be an object") else: required = fixture_set.get("requiredFixtureIds") if not isinstance(required, list) or tuple(required) != REQUIRED_FIXTURE_IDS: errors.append("fixtureSet required IDs do not match the canonical six") if fixture_set.get("networkRequired") is not False: errors.append("readiness fixtures must require no network") authorization = contract.get("authorization") if not isinstance(authorization, Mapping): errors.append("authorization must be an object") else: if authorization.get("evaluatedSeparately") is not True: errors.append("authorization must be evaluated separately") if authorization.get("semanticIdentityGrantsPermission") is not False: errors.append("semantic identity must not grant permission") if authorization.get("sideEffectPerformedBySelfTest") is not False: errors.append("self-test must perform no side effect") report = contract.get("report") if not isinstance(report, Mapping): errors.append("report must be an object") else: if report.get("deterministic") is not True: errors.append("readiness report must be deterministic") if report.get("containsCurrentTimestamp") is not False: errors.append("readiness report must not contain a current timestamp") if report.get("containsSecrets") is not False: errors.append("readiness report must contain no secrets") return tuple(errors) def validate_fixture_set(fixtures: Mapping[str, object]) -> tuple[str, ...]: """Validate the six fixed local fixture cases.""" errors: list[str] = [] if fixtures.get("schema") != INTEGRATION_READINESS_FIXTURES_SCHEMA: errors.append("unexpected readiness fixture schema identifier") if fixtures.get("version") != REPORT_VERSION: errors.append("unexpected readiness fixture version") cases = fixtures.get("fixtures") if not isinstance(cases, list) or len(cases) != len(REQUIRED_FIXTURE_IDS): errors.append("fixture set must contain exactly six cases") return tuple(errors) ids: list[str] = [] for index, fixture in enumerate(cases, start=1): if not isinstance(fixture, Mapping): errors.append(f"fixture {index} must be an object") continue fixture_id = fixture.get("id") if not isinstance(fixture_id, str): errors.append(f"fixture {index} missing string id") continue ids.append(fixture_id) for field in ("title", "expression", "language", "envelope", "expected"): if field not in fixture: errors.append(f"fixture {fixture_id} missing {field}") expected = fixture.get("expected") if not isinstance(expected, Mapping): errors.append(f"fixture {fixture_id} expected must be an object") continue if expected.get("semanticOutcome") not in { "resolved", "unknown_expression", "ambiguous_expression", "request_or_service_error", }: errors.append(f"fixture {fixture_id} has unsupported expected semanticOutcome") if expected.get("identityAssigned") not in {True, False}: errors.append(f"fixture {fixture_id} expected.identityAssigned must be boolean") if tuple(ids) != REQUIRED_FIXTURE_IDS: errors.append("fixture IDs or ordering do not match the canonical six") return tuple(errors) def build_client_record( expression: str, language: str | None, envelope: object, ) -> dict[str, Any]: """Convert one observed resolver envelope into the fail-closed client record. Identity fields are added only after the exact resolved acceptance predicate succeeds. All other paths retain evidence and omit both identity fields. """ record: dict[str, Any] = { "originalExpression": expression, "language": language, "semanticOutcome": "request_or_service_error", "resolverEvidence": _copy_evidence(envelope), } if not isinstance(envelope, Mapping): record["errorKind"] = "malformed_or_unsupported" record["error"] = { "code": "invalid_envelope", "message": "resolver evidence must be a JSON object", } return record error = envelope.get("error") if error is not None: error_mapping = error if isinstance(error, Mapping) else {} error_code = error_mapping.get("code") record["errorKind"] = "service_error" if error_code == "service_error" else "request_error" record["error"] = _copy_evidence(error) return record data = envelope.get("data") if not isinstance(data, Mapping): record["errorKind"] = "malformed_or_unsupported" record["error"] = { "code": "invalid_envelope", "message": "resolver envelope is missing a data object", } return record status = data.get("status") if status == "resolved": concept = data.get("concept") code = concept.get("code") if isinstance(concept, Mapping) else None version = concept.get("registryVersion") if isinstance(concept, Mapping) else None if not _non_empty_string(code) or not _non_empty_string(version): record["errorKind"] = "malformed_or_unsupported" record["error"] = { "code": "invalid_resolved_contract", "message": "resolved requires non-empty concept.code and concept.registryVersion", } return record record["semanticOutcome"] = "resolved" record["conceptCode"] = code record["registryVersion"] = version return record if status == "abstained": reason = data.get("reason") if reason == "unknown_expression": record["semanticOutcome"] = "unknown_expression" record["abstentionReason"] = reason return record if reason == "ambiguous_expression": record["semanticOutcome"] = "ambiguous_expression" record["abstentionReason"] = reason return record record["errorKind"] = "malformed_or_unsupported" record["error"] = { "code": "unsupported_abstention_reason", "message": f"unsupported abstention reason: {reason!r}", } return record record["errorKind"] = "malformed_or_unsupported" record["error"] = { "code": "unsupported_status", "message": f"unsupported resolver status: {status!r}", } return record def _recipe_profiles(contract: Mapping[str, object]) -> list[Mapping[str, object]]: """Return recipe profiles from a validated readiness contract.""" selection = contract.get("recipeSelection") if not isinstance(selection, Mapping): return [] recipes = selection.get("recipes") if not isinstance(recipes, list): return [] return [recipe for recipe in recipes if isinstance(recipe, Mapping)] def select_recipe(contract: Mapping[str, object], recipe_id: str) -> Mapping[str, object]: """Select exactly one canonical recipe or fail closed.""" if not isinstance(recipe_id, str) or not recipe_id: raise ValueError("exactly one recipe identifier is required") if "," in recipe_id or " " in recipe_id.strip(): raise ValueError("select exactly one canonical recipe identifier") matches = [recipe for recipe in _recipe_profiles(contract) if recipe.get("id") == recipe_id] if len(matches) != 1: raise ValueError(f"unsupported recipe: {recipe_id}") return matches[0] def _check(check_id: str, passed: bool, detail: str) -> dict[str, object]: """Build one deterministic check result.""" return {"id": check_id, "passed": bool(passed), "detail": detail} def run_fixture_case( fixture: Mapping[str, object], *, authorization_policy_version: str, ) -> dict[str, object]: """Exercise one fixed fixture and return an inspectable case result.""" fixture_id = str(fixture.get("id", "")) expression = str(fixture.get("expression", "")) language_value = fixture.get("language") language = language_value if isinstance(language_value, str) else None envelope = fixture.get("envelope") expected = fixture.get("expected") expected_mapping = expected if isinstance(expected, Mapping) else {} record = build_client_record(expression, language, envelope) expected_outcome = expected_mapping.get("semanticOutcome") expected_identity = expected_mapping.get("identityAssigned") is True actual_identity = all(field in record for field in RESOLVED_ONLY_FIELDS) any_identity_field = any(field in record for field in RESOLVED_ONLY_FIELDS) checks = [ _check( "always-retained-fields", all(field in record for field in ALWAYS_RETAINED_FIELDS), "originalExpression, language, semanticOutcome, and resolverEvidence are present", ), _check( "semantic-outcome", record.get("semanticOutcome") == expected_outcome, f"expected {expected_outcome!r}; observed {record.get('semanticOutcome')!r}", ), _check( "resolver-evidence-preserved", canonical_json_bytes(record.get("resolverEvidence")) == canonical_json_bytes(envelope), "the complete fixed envelope is retained byte-equivalently after canonical JSON serialization", ), _check( "resolved-only-identity-presence", actual_identity is expected_identity and (expected_identity or not any_identity_field), "code/version exist together only when the fixture expects a valid resolved assignment", ), ] if expected_identity: checks.extend( [ _check( "exact-concept-code", record.get("conceptCode") == expected_mapping.get("conceptCode"), "the exact fixture ConceptCode is copied without rewriting", ), _check( "exact-registry-version", record.get("registryVersion") == expected_mapping.get("registryVersion"), "the exact fixture registryVersion is copied without rewriting", ), ] ) else: checks.append( _check( "no-assignment", not any_identity_field, "non-resolved evidence contains neither conceptCode nor registryVersion", ) ) authorization = { "outcome": "not_authorized", "policyVersion": authorization_policy_version, "evaluatedSeparately": True, "semanticIdentityGrantsPermission": False, } side_effect = { "planned": False, "performed": False, "reason": "local fixture self-test only", } checks.extend( [ _check( "authorization-separate", authorization["evaluatedSeparately"] is True and authorization["semanticIdentityGrantsPermission"] is False, "authorization has an independent outcome and policy version", ), _check( "no-side-effect", side_effect["performed"] is False and side_effect["planned"] is False, "the self-test records evidence but performs no downstream action", ), ] ) return { "fixtureId": fixture_id, "title": str(fixture.get("title", fixture_id)), "expected": _copy_evidence(expected_mapping), "record": record, "authorization": authorization, "sideEffect": side_effect, "checks": checks, "passed": all(bool(item["passed"]) for item in checks), } def _report_without_digest(report: Mapping[str, object]) -> dict[str, object]: """Return a deep copy with the self-referential report digest removed.""" value = deepcopy(dict(report)) integrity = value.get("integrity") if isinstance(integrity, dict): integrity.pop("reportContentSha256", None) return value def run_readiness_self_test( contract: Mapping[str, object], fixtures: Mapping[str, object], recipe_id: str, *, authorization_policy_version: str = DEFAULT_AUTHORIZATION_POLICY_VERSION, ) -> dict[str, object]: """Run all six local cases and produce a deterministic readiness report.""" contract_errors = validate_contract(contract) fixture_errors = validate_fixture_set(fixtures) if contract_errors: raise ValueError("invalid readiness contract: " + "; ".join(contract_errors)) if fixture_errors: raise ValueError("invalid readiness fixtures: " + "; ".join(fixture_errors)) if not _non_empty_string(authorization_policy_version): raise ValueError("authorization policy version must be a non-empty string") recipe = select_recipe(contract, recipe_id) fixture_values = fixtures.get("fixtures") assert isinstance(fixture_values, list) cases = [ run_fixture_case( fixture, authorization_policy_version=authorization_policy_version, ) for fixture in fixture_values if isinstance(fixture, Mapping) ] identity_assigned = [ str(case["fixtureId"]) for case in cases if isinstance(case.get("record"), Mapping) and all(field in case["record"] for field in RESOLVED_ONLY_FIELDS) ] no_assignment = [ str(case["fixtureId"]) for case in cases if str(case["fixtureId"]) not in identity_assigned ] failed_case_ids = [str(case["fixtureId"]) for case in cases if case.get("passed") is not True] checks = [ _check( "exactly-one-recipe", recipe.get("id") == recipe_id, f"selected one canonical recipe: {recipe_id}", ), _check( "all-six-fixtures-exercised", [case["fixtureId"] for case in cases] == list(REQUIRED_FIXTURE_IDS), "resolved, unknown, ambiguous, malformed, request-error, and service-error cases ran in canonical order", ), _check( "only-valid-resolved-receives-identity", identity_assigned == ["resolved-valid"], f"identity-assigned fixtures: {identity_assigned!r}", ), _check( "all-other-fixtures-omit-identity", no_assignment == list(REQUIRED_FIXTURE_IDS[1:]), f"no-assignment fixtures: {no_assignment!r}", ), _check( "authorization-remains-independent", all( isinstance(case.get("authorization"), Mapping) and case["authorization"].get("evaluatedSeparately") is True and case["authorization"].get("semanticIdentityGrantsPermission") is False for case in cases ), "every case records an independent, non-authorizing policy result", ), _check( "no-side-effects", all( isinstance(case.get("sideEffect"), Mapping) and case["sideEffect"].get("performed") is False for case in cases ), "no fixture case performs or schedules a downstream side effect", ), ] all_case_checks_pass = not failed_case_ids all_summary_checks_pass = all(bool(item["passed"]) for item in checks) ready = all_case_checks_pass and all_summary_checks_pass report: dict[str, object] = { "$schema": str(contract.get("report", {}).get("schemaUrl", "")) if isinstance(contract.get("report"), Mapping) else "", "schema": INTEGRATION_READINESS_REPORT_SCHEMA, "version": REPORT_VERSION, "reportId": "", "scope": { "mode": "local_fixed_fixtures", "networkCalls": 0, "credentialsRequired": False, "productionDataRead": False, "productionDataWritten": False, "sideEffectsPerformed": 0, "proves": [ "client record construction for six fixed resolver outcomes", "resolved-only ConceptCode and registryVersion assignment", "independent authorization evidence", "deterministic side-effect-free report generation", ], "doesNotProve": [ "live registry coverage", "deployment correctness", "permission to perform a downstream action", "model quality or semantic generalization", ], }, "selectedRecipe": _copy_evidence(recipe), "clientRecordContract": { "alwaysRetained": list(ALWAYS_RETAINED_FIELDS), "resolvedOnly": list(RESOLVED_ONLY_FIELDS), "acceptIdentityOnlyWhen": ( 'data.status == "resolved" and data.concept.code and ' "data.concept.registryVersion are non-empty strings" ), "everyOtherOutcome": "no ConceptCode assignment", }, "authorization": { "outcome": "not_authorized", "policyVersion": authorization_policy_version, "evaluatedSeparately": True, "semanticIdentityGrantsPermission": False, "sideEffectPerformed": False, }, "checks": checks, "cases": cases, "proof": { "identityAssignedFixtureIds": identity_assigned, "noAssignmentFixtureIds": no_assignment, "requiredIdentityAssignedFixtureIds": ["resolved-valid"], "requiredNoAssignmentFixtureIds": list(REQUIRED_FIXTURE_IDS[1:]), }, "benefitMetricPlan": _copy_evidence(recipe.get("benefitMetrics", [])), "summary": { "fixtureCount": len(cases), "passedFixtureCount": len(cases) - len(failed_case_ids), "failedFixtureIds": failed_case_ids, "checkCount": len(checks) + sum(len(case["checks"]) for case in cases), "ready": ready, "disposition": "locally_conformant" if ready else "not_ready", }, "invariants": [ "The original expression, language or null, semantic outcome, and resolver evidence are always retained.", "Only resolved-valid receives the exact ConceptCode and registryVersion.", "Unknown, ambiguity, malformed response, request error, and service error remain no-assignment outcomes.", "Authorization is a separate policy result and semantic identity grants no permission.", "The self-test makes no network call and performs no side effect.", ], "integrity": { "contractSha256": sha256_json(contract), "fixturesSha256": sha256_json(fixtures), "reportContentSha256": "", }, } report_id_material = { "schema": report["schema"], "version": report["version"], "recipe": recipe_id, "contractSha256": report["integrity"]["contractSha256"], "fixturesSha256": report["integrity"]["fixturesSha256"], "authorizationPolicyVersion": authorization_policy_version, } report["reportId"] = f"sha256:{sha256_json(report_id_material)}" report["integrity"]["reportContentSha256"] = sha256_json(_report_without_digest(report)) return report def validate_readiness_report(report: Mapping[str, object]) -> tuple[str, ...]: """Validate report invariants and deterministic self-digest.""" errors: list[str] = [] if report.get("schema") != INTEGRATION_READINESS_REPORT_SCHEMA: errors.append("unexpected readiness report schema identifier") if report.get("version") != REPORT_VERSION: errors.append("unexpected readiness report version") selected = report.get("selectedRecipe") if not isinstance(selected, Mapping) or selected.get("id") not in CANONICAL_RECIPE_IDS: errors.append("readiness report must contain one canonical selected recipe") cases = report.get("cases") if not isinstance(cases, list) or [ case.get("fixtureId") for case in cases if isinstance(case, Mapping) ] != list(REQUIRED_FIXTURE_IDS): errors.append("readiness report must contain the canonical six fixture cases") else: for case in cases: if not isinstance(case, Mapping): errors.append("readiness report case must be an object") continue record = case.get("record") if not isinstance(record, Mapping): errors.append(f"case {case.get('fixtureId')} record must be an object") continue for field in ALWAYS_RETAINED_FIELDS: if field not in record: errors.append(f"case {case.get('fixtureId')} missing always-retained field {field}") has_identity = any(field in record for field in RESOLVED_ONLY_FIELDS) if case.get("fixtureId") == "resolved-valid": if not all(field in record for field in RESOLVED_ONLY_FIELDS): errors.append("resolved-valid must contain both identity fields") elif has_identity: errors.append(f"case {case.get('fixtureId')} must contain no identity fields") if case.get("passed") is not True: errors.append(f"case {case.get('fixtureId')} did not pass") proof = report.get("proof") if not isinstance(proof, Mapping): errors.append("readiness report proof must be an object") else: if proof.get("identityAssignedFixtureIds") != ["resolved-valid"]: errors.append("only resolved-valid may receive identity") if proof.get("noAssignmentFixtureIds") != list(REQUIRED_FIXTURE_IDS[1:]): errors.append("all non-resolved fixtures must remain no assignment") authorization = report.get("authorization") if not isinstance(authorization, Mapping): errors.append("readiness report authorization must be an object") else: if authorization.get("evaluatedSeparately") is not True: errors.append("authorization must be evaluated separately") if authorization.get("semanticIdentityGrantsPermission") is not False: errors.append("semantic identity must grant no permission") if authorization.get("sideEffectPerformed") is not False: errors.append("readiness self-test must perform no side effect") summary = report.get("summary") if not isinstance(summary, Mapping) or summary.get("ready") is not True: errors.append("readiness report summary must be ready") integrity = report.get("integrity") if not isinstance(integrity, Mapping): errors.append("readiness report integrity must be an object") else: expected = sha256_json(_report_without_digest(report)) if integrity.get("reportContentSha256") != expected: errors.append("readiness report content digest mismatch") return tuple(errors) def _build_parser() -> argparse.ArgumentParser: """Create the command-line parser.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--recipe", choices=CANONICAL_RECIPE_IDS, help="Exactly one canonical recipe to test.", ) parser.add_argument( "--contract", type=Path, default=DEFAULT_CONTRACT_PATH, help="Path to readiness-contract.json.", ) parser.add_argument( "--fixtures", type=Path, default=DEFAULT_FIXTURES_PATH, help="Path to fixtures.json.", ) parser.add_argument( "--policy-version", default=DEFAULT_AUTHORIZATION_POLICY_VERSION, help="Deterministic local authorization-policy version to record.", ) parser.add_argument( "--output", type=Path, help="Write the deterministic report to this path; omit for stdout.", ) parser.add_argument( "--compact", action="store_true", help="Emit canonical compact JSON instead of indented JSON.", ) parser.add_argument( "--list-recipes", action="store_true", help="Print canonical recipe IDs and exit.", ) return parser def main(argv: Sequence[str] | None = None) -> int: """Run the local self-test and return a shell-friendly exit code.""" args = _build_parser().parse_args(argv) if args.list_recipes: print("\n".join(CANONICAL_RECIPE_IDS)) return 0 if not args.recipe: print("ERROR: --recipe is required unless --list-recipes is used", file=sys.stderr) return 2 try: contract = load_json_object(args.contract) fixtures = load_json_object(args.fixtures) report = run_readiness_self_test( contract, fixtures, args.recipe, authorization_policy_version=args.policy_version, ) errors = validate_readiness_report(report) except (OSError, ValueError, json.JSONDecodeError) as error: print(f"ERROR: {error}", file=sys.stderr) return 2 if errors: for error in errors: print(f"ERROR: {error}", file=sys.stderr) return 1 output = canonical_json_bytes(report) if args.compact else pretty_json(report).encode("utf-8") if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_bytes(output) else: sys.stdout.buffer.write(output) return 0 if __name__ == "__main__": raise SystemExit(main())