Files
khadhroony-bot3/scripts/audit_khadhroony_workspace_rules.py
2026-07-28 18:41:30 +02:00

623 lines
23 KiB
Python
Executable File

#!/usr/bin/env python3
# file: scripts/audit_khadhroony_workspace_rules.py
# version: 10
"""Audit mechanically verifiable rules specific to khadhroony-bot3."""
from __future__ import annotations
import argparse
import dataclasses
import pathlib
import re
import sys
import tomllib
@dataclasses.dataclass(frozen=True)
class Violation:
"""One project-specific workspace rule violation."""
code: str
path: str
line: int
message: str
def package_name(crate_dir: pathlib.Path) -> str | None:
"""Return a crate package name."""
cargo = crate_dir / "Cargo.toml"
if not cargo.exists():
return None
value = tomllib.loads(cargo.read_text(encoding="utf-8")).get("package", {}).get("name")
return value if isinstance(value, str) else None
def relevant_tracing_crate(crate_dir: pathlib.Path) -> bool:
"""Return whether the crate owns a tracing target contract."""
cargo = crate_dir / "Cargo.toml"
data = tomllib.loads(cargo.read_text(encoding="utf-8"))
if "tracing" in data.get("dependencies", {}):
return True
return any("TRACING_TARGET" in path.read_text(encoding="utf-8") for path in crate_dir.glob("src/**/*.rs"))
def audit_tracing(root: pathlib.Path) -> list[Violation]:
"""Audit the project tracing target contract."""
violations: list[Violation] = []
for cargo in sorted(root.glob("*/Cargo.toml")):
crate = cargo.parent
if crate.name == "kb-lib":
continue
if not relevant_tracing_crate(crate):
continue
name = package_name(crate)
if name is None:
continue
constants = crate / "src/constants.rs"
expected = f'pub(crate) const TRACING_TARGET: &str = "{name}";'
relative_constants = constants.relative_to(root).as_posix()
if not constants.exists() or expected not in constants.read_text(encoding="utf-8"):
violations.append(Violation("KH_TRACE001", relative_constants, 1, f"expected `{expected}`"))
crate_root = crate / "src/lib.rs"
if not crate_root.exists():
crate_root = crate / "src/main.rs"
relative_root = crate_root.relative_to(root).as_posix()
if not crate_root.exists() or "pub(crate) use self::constants::TRACING_TARGET;" not in crate_root.read_text(encoding="utf-8"):
violations.append(Violation("KH_TRACE002", relative_root, 1, "missing crate-root TRACING_TARGET re-export"))
for path in crate.glob("src/**/*.rs"):
relative = path.relative_to(root).as_posix()
for index, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
if "khbot." in line:
violations.append(Violation("KH_TRACE003", relative, index, "legacy khbot tracing target is forbidden"))
if "crate::constants::TRACING_TARGET" in line:
violations.append(Violation("KH_TRACE004", relative, index, "use crate::TRACING_TARGET outside crate root"))
if re.search(r"(?:crate::[A-Za-z0-9_]+::|(?<!crate::))TRACING_TARGET", line) and "const TRACING_TARGET" not in line and "use self::constants::TRACING_TARGET" not in line:
if "crate::TRACING_TARGET" not in line:
violations.append(Violation("KH_TRACE005", relative, index, "non-canonical TRACING_TARGET path"))
return violations
def audit_kb_lib_tracing(root: pathlib.Path) -> list[Violation]:
"""Audit hierarchical tracing targets owned by operational kb-lib components."""
violations: list[Violation] = []
crate = root / "kb-lib"
crate_root = crate / "src/lib.rs"
if not crate_root.exists():
return violations
root_text = crate_root.read_text(encoding="utf-8")
re_exports = set()
re_export_pattern = re.compile(
r"^pub\(crate\) use self::[A-Za-z0-9_:]+::"
r"([A-Z][A-Z0-9_]+);$",
re.MULTILINE,
)
for match in re_export_pattern.finditer(root_text):
if match.group(1).startswith("TRACING_TARGET_"):
re_exports.add(match.group(1))
target_pattern = re.compile(
r'^pub\(crate\) const (TRACING_TARGET_[A-Z0-9_]+): &str = "([^"]+)";$',
re.MULTILINE,
)
for constants in sorted((crate / "src").rglob("constants.rs")):
text = constants.read_text(encoding="utf-8")
match = target_pattern.search(text)
if match is None:
continue
component = constants.parent.relative_to(crate / "src")
expected = "kb-lib." + ".".join(component.parts)
relative = constants.relative_to(root).as_posix()
if match.group(2) != expected:
violations.append(
Violation(
"KH_TRACE101",
relative,
text[: match.start()].count("\n") + 1,
f"`{match.group(1)}` must use hierarchical target `{expected}`",
)
)
macro_pattern = re.compile(
r"tracing::(?:debug|error|info|trace|warn)!\(\s*target:\s*crate::(TRACING_TARGET_[A-Z0-9_]+)"
)
for path in sorted((crate / "src").rglob("*.rs")):
relative = path.relative_to(root).as_posix()
text = path.read_text(encoding="utf-8")
for index, line in enumerate(text.splitlines(), 1):
if "let _target" in line:
violations.append(
Violation(
"KH_TRACE102",
relative,
index,
"fake tracing target consumption is forbidden",
)
)
for match in macro_pattern.finditer(text):
if match.group(1) not in re_exports:
violations.append(
Violation(
"KH_TRACE103",
relative,
text[: match.start()].count("\n") + 1,
f"missing crate-root re-export for `{match.group(1)}`",
)
)
return violations
def audit_solana_types(root: pathlib.Path) -> list[Violation]:
"""Reject non-canonical Solana address types in workspace Rust code."""
violations: list[Violation] = []
for path in sorted(root.rglob("*.rs")):
if "target" in path.parts or ".git" in path.parts:
continue
relative = path.relative_to(root).as_posix()
for index, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
if "solana_address::Address" in line:
violations.append(Violation("KH_SOL001", relative, index, "use solana_pubkey::Pubkey instead of solana_address::Address"))
return violations
def audit_kb_lib_symbol_prefixes(root: pathlib.Path) -> list[Violation]:
"""Require stable family prefixes on non-method kb-lib declarations."""
violations: list[Violation] = []
families = {
"decoder": ("DC_", "Dc", "decoder_"),
"executor": ("EX_", "Ex", "executor_"),
"materializer": ("MT_", "Mt", "materializer_"),
"model": ("MD_", "Md", "model_"),
}
declaration = re.compile(
r"^pub(?:\(crate\))?\s+"
r"(?:(?:async|const|unsafe)\s+)*"
r"(const|static|type|struct|enum|trait|union|fn)\s+"
r"([A-Za-z_][A-Za-z0-9_]*)"
)
for family, prefixes in families.items():
family_root = root / "kb-lib/src" / family
paths = [family_root.with_suffix(".rs")]
if family_root.is_dir():
paths.extend(sorted(family_root.rglob("*.rs")))
for path in paths:
if not path.is_file():
continue
relative = path.relative_to(root).as_posix()
for index, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
match = declaration.match(line)
if match is None:
continue
kind = match.group(1)
name = match.group(2)
expected = prefixes[0] if kind in {"const", "static"} else prefixes[2] if kind == "fn" else prefixes[1]
if kind in {"const", "static"} and name.startswith("TRACING_TARGET_"):
continue
if not name.startswith(expected):
violations.append(
Violation(
"KH_NAME001",
relative,
index,
f"`{name}` must start with `{expected}` for the {family} family",
)
)
return violations
def audit_reserved_decoder_scaffolds(root: pathlib.Path) -> list[Violation]:
"""Reject temporary decoder migration markers and incomplete reserved decoders."""
violations: list[Violation] = []
decoder_root = root / "kb-lib/src/decoder"
forbidden = (
"MIGRATION_BOUNDARIES",
"LEGACY_CRATE",
"MIGRATION_STATUS",
"source-preserved-pending-port",
)
struct_pattern = re.compile(r"^pub struct (Dc[A-Za-z0-9]+Decoder);$", re.MULTILINE)
for path in sorted(decoder_root.rglob("*.rs")):
relative = path.relative_to(root).as_posix()
text = path.read_text(encoding="utf-8")
for marker in forbidden:
offset = text.find(marker)
if offset >= 0:
violations.append(
Violation(
"KH_DC001",
relative,
text[:offset].count("\n") + 1,
f"temporary decoder migration marker `{marker}` is forbidden",
)
)
if "Reserved protocol decoder for `" not in text:
continue
struct_match = struct_pattern.search(text)
if struct_match is None:
violations.append(
Violation(
"KH_DC002",
relative,
1,
"reserved decoder must expose one `Dc*Decoder` unit struct",
)
)
continue
type_name = struct_match.group(1)
implementation = f"impl crate::DcApiProtocolDecoder for crate::{type_name} {{"
if implementation not in text:
violations.append(
Violation(
"KH_DC003",
relative,
text[: struct_match.start()].count("\n") + 1,
f"`{type_name}` must implement `DcApiProtocolDecoder`",
)
)
if "fn program_ids(&self) -> &'static [&'static str]" not in text:
violations.append(
Violation(
"KH_DC004",
relative,
1,
"reserved decoder must declare its exact Program ID boundary",
)
)
if "crate::DcApiDecoderSupport::Maybe" not in text:
violations.append(
Violation(
"KH_DC005",
relative,
1,
"reserved decoder must remain explicitly `Maybe` until implemented",
)
)
return violations
def audit_reserved_materializer_scaffolds(root: pathlib.Path) -> list[Violation]:
"""Reject temporary materializer markers and incomplete reserved materializers."""
violations: list[Violation] = []
materializer_root = root / "kb-lib/src/materializer"
forbidden = (
"MIGRATION_BOUNDARIES",
"LEGACY_CRATE",
"MIGRATION_STATUS",
"source-preserved-pending-port",
)
struct_pattern = re.compile(r"^pub struct (Mt[A-Za-z0-9]+Materializer);$", re.MULTILINE)
for path in sorted(materializer_root.rglob("*.rs")):
relative = path.relative_to(root).as_posix()
text = path.read_text(encoding="utf-8")
for marker in forbidden:
offset = text.find(marker)
if offset >= 0:
violations.append(
Violation(
"KH_MT001",
relative,
text[:offset].count("\n") + 1,
f"temporary materializer migration marker `{marker}` is forbidden",
)
)
if not text.startswith("// file:") or "\n//! Reserved " not in text:
continue
struct_match = struct_pattern.search(text)
if struct_match is None:
violations.append(
Violation(
"KH_MT002",
relative,
1,
"reserved materializer must expose one `Mt*Materializer` unit struct",
)
)
continue
type_name = struct_match.group(1)
legacy_implementation = f"impl crate::MtMaterializer for crate::{type_name} {{"
if legacy_implementation not in text:
violations.append(
Violation(
"KH_MT003",
relative,
text[: struct_match.start()].count("\n") + 1,
f"`{type_name}` must implement `MtMaterializer`",
)
)
api_implementation = f"impl crate::MtApiEventMaterializer for crate::{type_name} {{"
if api_implementation not in text:
violations.append(
Violation(
"KH_MT004",
relative,
text[: struct_match.start()].count("\n") + 1,
f"`{type_name}` must implement `MtApiEventMaterializer`",
)
)
if "fn accepted_families(&self) -> &'static [crate::MdEventFamily]" not in text:
violations.append(
Violation(
"KH_MT005",
relative,
1,
"reserved materializer must declare its inactive family boundary",
)
)
if "crate::MtApiMaterializerExecutionResult::ignored()" not in text:
violations.append(
Violation(
"KH_MT006",
relative,
1,
"reserved materializer must return an ignored API result",
)
)
return violations
def audit_reserved_executor_scaffolds(root: pathlib.Path) -> list[Violation]:
"""Reject temporary executor markers and incomplete reserved executors."""
violations: list[Violation] = []
executor_root = root / "kb-lib/src/executor"
forbidden = (
"MIGRATION_BOUNDARIES",
"LEGACY_CRATE",
"MIGRATION_STATUS",
"source-preserved-pending-port",
)
struct_pattern = re.compile(r"^pub struct (Ex[A-Za-z0-9]+Executor);$", re.MULTILINE)
reserved_count = 0
for path in sorted(executor_root.rglob("*.rs")):
relative = path.relative_to(root).as_posix()
text = path.read_text(encoding="utf-8")
for marker in forbidden:
offset = text.find(marker)
if offset >= 0:
violations.append(
Violation(
"KH_EX001",
relative,
text[:offset].count("\n") + 1,
f"temporary executor migration marker `{marker}` is forbidden",
)
)
if "\n//! Reserved executor for `" not in text:
continue
reserved_count += 1
struct_match = struct_pattern.search(text)
if struct_match is None:
violations.append(
Violation(
"KH_EX002",
relative,
1,
"reserved executor must expose one `Ex*Executor` unit struct",
)
)
continue
type_name = struct_match.group(1)
implementation = f"impl crate::ExApiInstructionExecutor for crate::{type_name} {{"
if implementation not in text:
violations.append(
Violation(
"KH_EX003",
relative,
text[: struct_match.start()].count("\n") + 1,
f"`{type_name}` must implement `ExApiInstructionExecutor`",
)
)
if "fn program_ids(&self) -> &'static [&'static str]" not in text:
violations.append(
Violation(
"KH_EX004",
relative,
1,
"reserved executor must declare its exact Program ID boundary",
)
)
if "crate::ExApiExecutionSupport::Maybe" not in text:
violations.append(
Violation(
"KH_EX005",
relative,
1,
"reserved executor must remain explicitly `Maybe` until implemented",
)
)
if '"status":"reserved_executor"' not in text or "instruction_count: 0" not in text:
violations.append(
Violation(
"KH_EX006",
relative,
1,
"reserved executor must build only an explicit zero-instruction plan",
)
)
if reserved_count != 98:
violations.append(
Violation(
"KH_EX007",
"kb-lib/src/executor",
1,
f"expected 98 reserved executor boundaries, found {reserved_count}",
)
)
return violations
def audit_token_2022_naming(root: pathlib.Path) -> list[Violation]:
"""Reject obsolete internal Token2022 spellings while preserving legacy evidence."""
violations: list[Violation] = []
obsolete = (
"kb-lib.decoder.spl.token_2022",
"docs/SPL_TOKEN_2022_",
"`core_spl_token_2022",
"`spl_token_2022`",
)
candidates = [
root / "README.md",
root / "ROADMAP.md",
root / "RULES.md",
root / "RULES_GENERAL.md",
root / "RULES_RUST.md",
root / "RULES_SPECIFIC_KHADHROONY.md",
root / "kb-lib/README.md",
]
candidates.extend(sorted((root / "docs").glob("*.md")))
for path in candidates:
if not path.is_file() or path.name == "RUST_WORKSPACE_RULE_AUDIT.md":
continue
relative = path.relative_to(root).as_posix()
for index, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
for spelling in obsolete:
if spelling in line:
violations.append(
Violation(
"KH_NAME002",
relative,
index,
f"obsolete internal Token2022 spelling `{spelling}`",
)
)
return violations
def audit_private_kb_lib_paths_in_active_docs(root: pathlib.Path) -> list[Violation]:
"""Reject documentation that presents private kb-lib modules as public APIs."""
violations: list[Violation] = []
candidates = [
root / "README.md",
root / "ROADMAP.md",
root / "RULES.md",
root / "RULES_GENERAL.md",
root / "RULES_RUST.md",
root / "RULES_SPECIFIC_KHADHROONY.md",
root / "kb-lib/README.md",
root / "kb-program-ids/README.md",
]
candidates.extend(sorted((root / "docs").glob("*.md")))
candidates.extend(sorted((root / "docs").glob("*.json")))
pattern = re.compile(r"kb_lib::(?:decoder|executor|materializer|model)::")
for path in candidates:
if not path.is_file():
continue
relative = path.relative_to(root).as_posix()
for index, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
match = pattern.search(line)
if match is not None:
violations.append(
Violation(
"KH_API001",
relative,
index,
f"private kb-lib module path `{match.group(0)}` is forbidden in active documentation",
)
)
return violations
def audit_wincode_resolution(root: pathlib.Path) -> list[Violation]:
"""Require the currently compatible Solana wincode dependency family."""
violations: list[Violation] = []
cargo_path = root / "Cargo.toml"
lock_path = root / "Cargo.lock"
if not cargo_path.exists():
return violations
cargo = tomllib.loads(cargo_path.read_text(encoding="utf-8"))
workspace_wincode = cargo.get("workspace", {}).get("dependencies", {}).get("wincode")
version = workspace_wincode.get("version") if isinstance(workspace_wincode, dict) else None
if version != "^0.5":
violations.append(
Violation(
"KH_DEP001",
"Cargo.toml",
1,
"workspace wincode must remain constrained to `^0.5`",
)
)
if not lock_path.exists():
violations.append(
Violation(
"KH_DEP002",
"Cargo.lock",
1,
"local workspace Cargo.lock is required to verify the pinned wincode resolution",
)
)
return violations
lock = tomllib.loads(lock_path.read_text(encoding="utf-8"))
packages = lock.get("package", [])
resolved_wincode = [
package.get("version")
for package in packages
if package.get("name") == "wincode"
]
resolved_varint = [
package.get("version")
for package in packages
if package.get("name") == "solana-wincode-varint"
]
if resolved_wincode != ["0.5.5"]:
violations.append(
Violation(
"KH_DEP003",
"Cargo.lock",
1,
f"expected only wincode 0.5.5, resolved {resolved_wincode}",
)
)
if resolved_varint != ["1.0.0"]:
violations.append(
Violation(
"KH_DEP004",
"Cargo.lock",
1,
f"expected only solana-wincode-varint 1.0.0, resolved {resolved_varint}",
)
)
return violations
def main() -> int:
"""Run the khadhroony-specific audit."""
parser = argparse.ArgumentParser()
parser.add_argument("--root", default=".")
parser.add_argument("--report-only", action="store_true")
arguments = parser.parse_args()
root = pathlib.Path(arguments.root).resolve()
violations = (
audit_tracing(root)
+ audit_kb_lib_tracing(root)
+ audit_kb_lib_symbol_prefixes(root)
+ audit_reserved_decoder_scaffolds(root)
+ audit_reserved_materializer_scaffolds(root)
+ audit_reserved_executor_scaffolds(root)
+ audit_token_2022_naming(root)
+ audit_private_kb_lib_paths_in_active_docs(root)
+ audit_solana_types(root)
+ audit_wincode_resolution(root)
)
violations.sort(key=lambda item: (item.code, item.path, item.line, item.message))
if not violations:
sys.stdout.write("Khadhroony workspace rule audit: clean\n")
return 0
sys.stdout.write(f"Khadhroony workspace rule audit: {len(violations)} violation(s)\n")
for item in violations:
sys.stdout.write(f"{item.code} {item.path}:{item.line}: {item.message}\n")
return 0 if arguments.report_only else 1
if __name__ == "__main__":
raise SystemExit(main())