Organize package modules under 400 lines

Author: cobycloud

.ssot/registry.json

@@ -20,6 +20,7 @@
     "tools/",
     "profiles/",
     "src/tigrcorn/profiles/",
+    "src/tigrcorn/capabilities/schema/runtime-capability-registry.schema.json",
     "docs/review/performance/",
     "examples/http_entity_static",
     "examples/advanced_protocol_delivery",

MUT.json

@@ -31,8 +31,7 @@ When sources disagree, use this order:
 
 1. **Definitive machine-readable truth**
    - `.ssot/registry.json`
-   - `tools/ssot_sync.py`
-   - normalized `.ssot` scaffold bootstrapped from `ssot-registry init`
+   - normalized `.ssot` scaffold and document rows maintained through the SSOT CLI
 2. **Canonical current-state truth**
    - `docs/review/conformance/state/CURRENT_REPOSITORY_STATE.md`
    - `docs/review/conformance/current_state_chain.current.json`
@@ -106,7 +105,7 @@ When a machine-readable statement exists in `.ssot/registry.json`, prose and der
 ### If you change canonical truth
 
 Update `.ssot/registry.json` first, then regenerate or reconcile downstream docs and JSON views.
-The `.ssot` directory scaffold itself is initialized by `tools/ssot_sync.py` through `ssot-registry init`.
+The `.ssot` directory scaffold itself is initialized and maintained through the SSOT CLI.
 
 ### If you change CLI behavior
 
@@ -145,7 +144,7 @@ Confirm path and naming rules in `docs/gov/tree.md`, add `MUT.json` where needed
 Run, at minimum:
 
 ```bash
-python tools/ssot_sync.py --check
+uv run ssot validate .
 python tools/govchk.py scan
 PYTHONPATH=src python -m compileall -q src benchmarks tools
 PYTHONPATH=src pytest -q

docs/gov/authoring.md

@@ -14,7 +14,7 @@ conformance checks.
 - `feat:contract-examples` - contract-native and ASGI/3 compatibility examples
   live under `examples/contract/`.
 - `feat:ssot-contract-boundary-sync` - boundary, feature, claim, test, and
-  evidence rows are generated by `tools/ssot_sync.py`.
+  evidence rows are maintained in `.ssot/registry.json` through SSOT CLI operations.
 - `feat:contract-release-evidence` - release evidence is rooted at
   `docs/review/conformance/releases/0.3.9/release-0.3.9/`.
 - `feat:asgi3-app-compat-suite` - ASGI/3 compatibility is covered by the

docs/review/conformance/CONTRACT_PROOF_BOUNDARY.md

@@ -38,7 +38,7 @@ This document defines the package-wide current-state chain for the repository.
 ## Canonical registry source
 
 - `.ssot/registry.json` is the canonical machine-readable governance registry for current-state, boundary, release, claim, evidence, and risk linkage.
-- `tools/ssot_sync.py` regenerates `.ssot/registry.json` from the current promoted repository truth.
+- SSOT CLI operations maintain `.ssot/registry.json`; validation gates read the committed registry directly.
 
 ## Scoped current audits that are **not** package-wide current-state sources
 

docs/review/conformance/CURRENT_STATE_CHAIN.md

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from .helpers import *
+from .records import *
+from .bundle import *
+from .status_docs import *
+from .core import *
+
+__all__ = [name for name in globals() if not name.startswith('_')]

pkgs/tigrcorn-certification/src/tigrcorn_certification/aioquic_preflight.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from .imports import *
+from .helpers import *
+
+def _bundle_manifest(*, artifact_root: str, matrix_path: str, scenario_ids: Sequence[str]) -> dict[str, Any]:
+    return {
+        'bundle_kind': 'aioquic_adapter_preflight_bundle',
+        'generated_at': _now(),
+        'release_gate_eligible': False,
+        'artifact_root': artifact_root,
+        'matrix_path': matrix_path,
+        'scenario_ids': list(scenario_ids),
+        'note': 'This bundle proves the third-party aioquic HTTP/3 adapters can execute cleanly before strict-target checkpoint promotion work continues.',
+    }
+
+
+def _bundle_index(*, artifact_root: str, matrix_path: str, scenario_records: Sequence[Mapping[str, Any]], environment: Mapping[str, Any], gate_status: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        'artifact_root': artifact_root,
+        'bundle_kind': 'aioquic_adapter_preflight_bundle',
+        'generated_at': _now(),
+        'matrix_path': matrix_path,
+        'scenario_count': len(scenario_records),
+        'scenario_ids': [str(item['scenario_id']) for item in scenario_records],
+        'all_adapters_passed': all(bool(item['passed']) for item in scenario_records),
+        'no_peer_exit_code_2': all(int(item['peer_exit_code']) != 2 for item in scenario_records),
+        'negotiation_metadata_emitted': all(bool(item['negotiation_metadata_emitted']) for item in scenario_records),
+        'transcript_metadata_emitted': all(bool(item['transcript_emitted']) for item in scenario_records),
+        'all_protocols_h3': all(item.get('protocol') == 'h3' for item in scenario_records),
+        'all_handshakes_complete': all(bool(item['handshake_complete']) for item in scenario_records),
+        'certificate_inputs_ready': all(bool(item['certificate_inputs_ready']) for item in scenario_records),
+        'packet_traces_emitted': all(bool(item['packet_trace_exists']) for item in scenario_records),
+        'qlogs_emitted': all(bool(item['qlog_exists']) for item in scenario_records),
+        'environment': dict(environment),
+        'gate_status_after_preflight': dict(gate_status),
+        'release_gate_eligible': False,
+    }
+
+
+def _bundle_summary(index: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        'artifact_root': index['artifact_root'],
+        'bundle_kind': index['bundle_kind'],
+        'generated_at': index['generated_at'],
+        'scenario_count': index['scenario_count'],
+        'all_adapters_passed': index['all_adapters_passed'],
+        'no_peer_exit_code_2': index['no_peer_exit_code_2'],
+        'all_protocols_h3': index['all_protocols_h3'],
+        'all_handshakes_complete': index['all_handshakes_complete'],
+        'certificate_inputs_ready': index['certificate_inputs_ready'],
+    }
+
+
+def _bundle_readme(index: Mapping[str, Any], scenario_records: Sequence[Mapping[str, Any]]) -> str:
+    lines = [
+        '# aioquic adapter preflight bundle',
+        '',
+        'This bundle preserves the direct third-party aioquic HTTP/3 adapter preflight runs used before strict-target certification checkpoints.',
+        '',
+        '## Exit-criteria status',
+        '',
+        f"- all adapters passed: `{index['all_adapters_passed']}`",
+        f"- no peer exit code 2: `{index['no_peer_exit_code_2']}`",
+        f"- negotiation metadata emitted: `{index['negotiation_metadata_emitted']}`",
+        f"- transcript metadata emitted: `{index['transcript_metadata_emitted']}`",
+        f"- ALPN h3 observed for every run: `{index['all_protocols_h3']}`",
+        f"- QUIC handshakes complete: `{index['all_handshakes_complete']}`",
+        f"- certificate inputs ready: `{index['certificate_inputs_ready']}`",
+        '',
+        '## Scenarios',
+        '',
+    ]
+    for item in scenario_records:
+        lines.extend([
+            f"- `{item['scenario_id']}` → passed=`{item['passed']}`, peer_exit=`{item['peer_exit_code']}`, protocol=`{item['protocol']}`, handshake_complete=`{item['handshake_complete']}`",
+        ])
+    lines.append('')
+    return '\n'.join(lines) + '\n'

pkgs/tigrcorn-certification/src/tigrcorn_certification/aioquic_preflight/__init__.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+from .imports import *
+from .helpers import *
+from .records import *
+from .bundle import *
+from .status_docs import *
+
+def run_aioquic_adapter_preflight(
+    root: str | Path,
+    *,
+    release_root: str = DEFAULT_RELEASE_ROOT,
+    bundle_name: str = DEFAULT_BUNDLE_NAME,
+    matrix_path: str = DEFAULT_MATRIX_PATH,
+    scenario_ids: Sequence[str] = DEFAULT_PRELIGHT_SCENARIOS,
+    bundle_root: str | Path | None = None,
+    require_pass: bool = False,
+) -> dict[str, Any]:
+    repo_root = Path(root)
+    resolved_release_root = repo_root / release_root
+    target_bundle_root = Path(bundle_root) if bundle_root is not None else resolved_release_root / bundle_name
+    if target_bundle_root.exists():
+        shutil.rmtree(target_bundle_root)
+    target_bundle_root.mkdir(parents=True, exist_ok=True)
+
+    with tempfile.TemporaryDirectory(prefix='aioquic-preflight-') as tmpdir:
+        summary = run_external_matrix(
+            repo_root / matrix_path,
+            artifact_root=tmpdir,
+            source_root=repo_root,
+            scenario_ids=list(scenario_ids),
+            strict=True,
+        )
+        generated_root = Path(summary.artifact_root)
+        for source_name, target_name in (
+            ('manifest.json', 'generated_matrix_manifest.json'),
+            ('index.json', 'generated_matrix_index.json'),
+            ('summary.json', 'generated_matrix_summary.json'),
+        ):
+            shutil.copy2(generated_root / source_name, target_bundle_root / target_name)
+        for scenario_id in scenario_ids:
+            shutil.copytree(generated_root / scenario_id, target_bundle_root / scenario_id)
+
+    environment = {
+        'python_version': sys.version,
+        'python_minor_version': f'{sys.version_info.major}.{sys.version_info.minor}',
+        'aioquic_version': _module_version('aioquic'),
+        'wsproto_version': _module_version('wsproto'),
+        'h2_version': _module_version('h2'),
+        'websockets_version': _module_version('websockets'),
+    }
+    gate_status = {
+        'authoritative_boundary_passed': evaluate_release_gates(repo_root).passed,
+        'strict_target_boundary_passed': evaluate_release_gates(repo_root, boundary_path='docs/review/conformance/certification_boundary.strict_target.json').passed,
+        'promotion_target_passed': evaluate_promotion_target(repo_root).passed,
+    }
+    scenario_records = [_extract_scenario_record(repo_root, target_bundle_root, scenario_id) for scenario_id in scenario_ids]
+    index = _bundle_index(
+        artifact_root=str(target_bundle_root.relative_to(repo_root)) if target_bundle_root.is_relative_to(repo_root) else str(target_bundle_root),
+        matrix_path=matrix_path,
+        scenario_records=scenario_records,
+        environment=environment,
+        gate_status=gate_status,
+    )
+    manifest = _bundle_manifest(
+        artifact_root=index['artifact_root'],
+        matrix_path=matrix_path,
+        scenario_ids=scenario_ids,
+    )
+    summary = _bundle_summary(index)
+    _dump_json(target_bundle_root / 'manifest.json', manifest)
+    _dump_json(target_bundle_root / 'index.json', index)
+    _dump_json(target_bundle_root / 'summary.json', summary)
+    _dump_json(target_bundle_root / 'preflight.json', {
+        'generated_at': _now(),
+        'environment': environment,
+        'gate_status_after_preflight': gate_status,
+        'scenario_records': scenario_records,
+    })
+    (target_bundle_root / 'README.md').write_text(_bundle_readme(index, scenario_records), encoding='utf-8')
+
+    snapshot = {
+        'checkpoint': 'aioquic_adapter_preflight',
+        'status': 'aioquic_adapter_preflight_passed' if summary['all_adapters_passed'] else 'aioquic_adapter_preflight_failed',
+        'current_state': {
+            'release_root': release_root,
+            'bundle_root': index['artifact_root'],
+            'matrix_path': matrix_path,
+            'scenario_ids': list(scenario_ids),
+            'scenario_records': scenario_records,
+            'environment': environment,
+            'all_adapters_passed': index['all_adapters_passed'],
+            'no_peer_exit_code_2': index['no_peer_exit_code_2'],
+            'negotiation_metadata_emitted': index['negotiation_metadata_emitted'],
+            'transcript_metadata_emitted': index['transcript_metadata_emitted'],
+            'all_protocols_h3': index['all_protocols_h3'],
+            'all_handshakes_complete': index['all_handshakes_complete'],
+            'certificate_inputs_ready': index['certificate_inputs_ready'],
+            'packet_traces_emitted': index['packet_traces_emitted'],
+            'qlogs_emitted': index['qlogs_emitted'],
+            'gate_status_after_preflight': gate_status,
+        },
+        'remaining_strict_target_blockers': [
+            'websocket-http3-server-aioquic-client-permessage-deflate',
+            'http3-connect-relay-aioquic-client',
+            'http3-trailer-fields-aioquic-client',
+            'http3-content-coding-aioquic-client',
+        ],
+    }
+    if require_pass and not summary['all_adapters_passed']:
+        raise AioquicAdapterPreflightError('one or more aioquic adapter preflight scenarios failed')
+    return snapshot

pkgs/tigrcorn-certification/src/tigrcorn_certification/aioquic_preflight/bundle.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from .imports import *
+
+class AioquicAdapterPreflightError(RuntimeError):
+    """Raised when the aioquic adapter preflight fails and strict pass mode is enabled."""
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _load_json(path: Path) -> dict[str, Any]:
+    return json.loads(path.read_text(encoding='utf-8'))
+
+
+def _dump_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + '\n', encoding='utf-8')
+
+
+def _module_version(name: str) -> str | None:
+    try:
+        return importlib.metadata.version(name)
+    except importlib.metadata.PackageNotFoundError:
+        return None
+
+
+def _command_option(command: Sequence[str], option: str) -> str | None:
+    try:
+        index = list(command).index(option)
+    except ValueError:
+        return None
+    next_index = index + 1
+    if next_index >= len(command):
+        return None
+    return str(command[next_index])
+
+
+def _module_name(command: Sequence[str]) -> str | None:
+    try:
+        index = list(command).index('-m')
+    except ValueError:
+        return None
+    next_index = index + 1
+    if next_index >= len(command):
+        return None
+    return str(command[next_index])
+
+
+def _path_ready(entry: Mapping[str, Any] | None) -> bool:
+    if not isinstance(entry, Mapping):
+        return False
+    return bool(entry.get('exists')) and bool(entry.get('is_file'))
+
+
+def _default_certificate_inputs(repo_root: Path, peer_command: Sequence[str]) -> dict[str, Any]:
+    def _entry(option: str) -> dict[str, Any]:
+        value = _command_option(peer_command, option)
+        if value is None:
+            return {'path': None, 'exists': False, 'is_file': False}
+        candidate = repo_root / value
+        return {
+            'path': value,
+            'exists': candidate.exists(),
+            'is_file': candidate.is_file(),
+        }
+
+    ca = _entry('--cacert')
+    cert = _entry('--client-cert')
+    key = _entry('--client-key')
+    client_material_requested = bool(cert['path'] or key['path'])
+    client_material_ready = (not client_material_requested) or (bool(cert['exists']) and bool(key['exists']))
+    return {
+        'ca_cert': ca,
+        'client_cert': cert,
+        'client_key': key,
+        'client_material_requested': client_material_requested,
+        'client_material_ready': client_material_ready,
+        'ready': bool(ca['exists']) and client_material_ready,
+    }
+
+
+def _scenario_kind(scenario_id: str) -> str:
+    if 'websocket' in scenario_id:
+        return 'http3_websocket_adapter'
+    return 'http3_client_adapter'