488 lines
24 KiB
Python
488 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
"""Adversarial regression tests for quality, data, evidence and ledger hardening."""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
|
|
import yaml
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
HOOKS = os.path.join(ROOT, ".claude", "hooks")
|
|
WS = os.path.join(ROOT, ".claude", "tests", "fixtures", "quality-data-hardening-ws")
|
|
os.environ["CLAUDE_PROJECT_DIR"] = ROOT
|
|
os.environ["ORGOS_WORKSPACE"] = WS
|
|
sys.path.insert(0, HOOKS)
|
|
|
|
import doctor as DOCTOR # noqa: E402
|
|
import state_engine as SE # noqa: E402
|
|
|
|
shutil.rmtree(WS, ignore_errors=True)
|
|
os.makedirs(WS, exist_ok=True)
|
|
|
|
passed = failed = 0
|
|
|
|
|
|
def check(name, condition):
|
|
global passed, failed
|
|
if condition:
|
|
passed += 1
|
|
print(f" PASS {name}")
|
|
else:
|
|
failed += 1
|
|
print(f" FAIL {name}")
|
|
|
|
|
|
def sha(path):
|
|
with open(path, "rb") as fh:
|
|
return hashlib.sha256(fh.read()).hexdigest()
|
|
|
|
|
|
def write_report(wf, artifact_id, kind, producer, payload, stage):
|
|
directory = os.path.join(WS, "completion-records", wf)
|
|
os.makedirs(directory, exist_ok=True)
|
|
path = os.path.join(directory, f"{artifact_id}.report.yaml")
|
|
report = {
|
|
"report-type": "workflow-artifact",
|
|
"artifact-kind": kind,
|
|
"artifact-version": 1,
|
|
"tier": (SE.read_ledger(wf) or {}).get("tier", "light"),
|
|
"identity": {
|
|
"artifact-id": artifact_id,
|
|
"workflow-id": wf,
|
|
"stage": stage,
|
|
"producer-role-id": producer,
|
|
},
|
|
"payload": payload,
|
|
"report-header": {
|
|
"bottom-line": f"{kind} adversarial fixture",
|
|
"decision-needed": {"needed": False},
|
|
"confidence": {"value": "Med", "derived-from": "evidence"},
|
|
"risks": [],
|
|
"evidence": [{"source-uri": "README.md", "grade": "E3"}],
|
|
},
|
|
}
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
yaml.safe_dump(report, fh, allow_unicode=True, sort_keys=False)
|
|
return path
|
|
|
|
|
|
def open_stage(wf, stage):
|
|
current = SE.read_ledger(wf).get("stage")
|
|
event = {
|
|
"state-event-id": f"fixture-{wf}-{stage}-{len(SE.read_workflow_events(wf))}",
|
|
"event-type": "state-transition", "workflow-id": wf,
|
|
"from": current, "to": stage, "actor": "OPS-ORCH",
|
|
"effective-at": SE._now(),
|
|
}
|
|
return SE._atomic_event_transaction(wf, workflow_event=event)[0]
|
|
|
|
|
|
def add_receipt(receipt_id, wf=None, *, exit_code=0, ts=None, scoped=True,
|
|
subject=None, source_revision_sha256=None):
|
|
directory = os.path.join(WS, "evidence")
|
|
os.makedirs(directory, exist_ok=True)
|
|
row = {
|
|
"tool_use_id": receipt_id, "tool_name": "Bash", "command": "fixture-check",
|
|
"exit_code": exit_code, "ts": ts or SE._now(),
|
|
"receipt_type": "verification-run", "verification_category": "test",
|
|
"verification_subject": subject or f"check-{receipt_id}",
|
|
"assertion_status": "passed" if exit_code == 0 else "failed",
|
|
}
|
|
if source_revision_sha256 is not None:
|
|
row["source_revision_sha256"] = source_revision_sha256
|
|
if scoped:
|
|
row.update({"workflow_id": wf, "session_id": "fixture-session", "agent_id": "fixture-agent"})
|
|
with open(os.path.join(directory, "ledger.jsonl"), "a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(row) + "\n")
|
|
|
|
|
|
def completion_payload(path, receipt_id):
|
|
return {
|
|
"summary": "verified implementation",
|
|
"source-revision": {"kind": "workspace-tree", "sha256": "a" * 64},
|
|
"primary-artifacts": [{"path": path, "kind": "code", "sha256": sha(path)}],
|
|
"acceptance-criteria-coverage": [{
|
|
"criterion-id": "AC-1", "status": "Passed",
|
|
"evidence-receipt-ids": [receipt_id],
|
|
}],
|
|
"verification-receipt-ids": [receipt_id],
|
|
"remaining-risks": [],
|
|
}
|
|
|
|
|
|
def quality_payload(target, receipt_id, *, status="Passed"):
|
|
return {
|
|
"quality-gate": {"status": status}, "blocker-open": False,
|
|
"reviewed-artifact-id": target["artifact-id"],
|
|
"reviewed-artifact-sha256": target["artifact-sha256"],
|
|
"checks": [{
|
|
"check-id": f"check-{receipt_id}", "category": "test", "status": status,
|
|
"evidence-receipt-ids": [receipt_id],
|
|
}],
|
|
"findings": [],
|
|
}
|
|
|
|
|
|
print("== sanctioned verification runner emits typed receipts ==")
|
|
verify_cmd = [
|
|
sys.executable, os.path.join(HOOKS, "verify_run.py"),
|
|
"--workflow", "verify-run-wf", "--agent", "QA", "--session", "fixture-session",
|
|
"--category", "test", "--subject", "verify-run-pycompile", "--",
|
|
sys.executable, "-m", "py_compile", os.path.join(HOOKS, "verify_run.py"),
|
|
]
|
|
verify_result = subprocess.run(
|
|
verify_cmd, cwd=ROOT, capture_output=True, text=True, env=dict(os.environ))
|
|
check("verify_run returns the verifier's successful exit", verify_result.returncode == 0)
|
|
verify_rows = [json.loads(line) for line in open(os.path.join(WS, "evidence", "ledger.jsonl"))
|
|
if line.strip()]
|
|
verify_receipt = verify_rows[-1] if verify_rows else {}
|
|
check("verify_run receipt is typed and context-bound",
|
|
verify_receipt.get("receipt_type") == "verification-run"
|
|
and verify_receipt.get("workflow_id") == "verify-run-wf"
|
|
and verify_receipt.get("verification_category") == "test"
|
|
and verify_receipt.get("assertion_status") == "passed"
|
|
and verify_receipt.get("exit_code") == 0
|
|
and verify_receipt.get("command_argv_sha256"))
|
|
trivial_result = subprocess.run([
|
|
sys.executable, os.path.join(HOOKS, "verify_run.py"),
|
|
"--workflow", "verify-run-wf", "--agent", "QA", "--session", "fixture-session",
|
|
"--category", "test", "--subject", "fake-proof", "--", "true",
|
|
], cwd=ROOT, capture_output=True, text=True, env=dict(os.environ))
|
|
check("verify_run rejects trivial commands as verification proof",
|
|
trivial_result.returncode == 2 and "trivial" in trivial_result.stderr)
|
|
|
|
|
|
print("== fail-closed identity, stage and risk ==")
|
|
try:
|
|
SE.init_ledger("bad-tier", tier="premium")
|
|
unknown_tier_rejected = False
|
|
except ValueError:
|
|
unknown_tier_rejected = True
|
|
check("unknown tier cannot initialize a workflow", unknown_tier_rejected)
|
|
|
|
SE.init_ledger("scope-wf", tier="light")
|
|
future = write_report("scope-wf", "future-brief", "decision-brief", "EXEC-CEO",
|
|
{"mode": "converge", "tier": "light",
|
|
"candidate-families": ["FAM-CPO", "FAM-CTO", "FAM-CFO"]}, "discovery")
|
|
check("future-stage artifact is rejected", not SE.submit_report("scope-wf", future, "EXEC-CEO")[0])
|
|
|
|
high_risk = write_report("scope-wf", "high-risk", "workload-profile", "EXEC-CEO", {
|
|
"surfaces": {"ui": False, "public-api": False, "persistence": True, "infrastructure": False},
|
|
"risk": {"security-bearing": True, "data-migration": True, "external-side-effect": True,
|
|
"risk-level": "High", "reversibility": "one-way-door",
|
|
"blast-radius": "production-customer-revenue", "privacy": True,
|
|
"regulatory": False, "slo-impact": True},
|
|
"required-capabilities": ["data"], "product-feature": False,
|
|
}, "intake")
|
|
check("risk hard floor prevents a light-tier declaration", not SE.submit_report("scope-wf", high_risk, "EXEC-CEO")[0])
|
|
|
|
|
|
print("== completion, receipt scope and quality freshness ==")
|
|
SE.init_ledger("quality-wf", tier="light")
|
|
check("open build", open_stage("quality-wf", "build"))
|
|
implementation = os.path.join(WS, "implementation.txt")
|
|
with open(implementation, "w", encoding="utf-8") as fh:
|
|
fh.write("implementation\n")
|
|
|
|
add_receipt("unscoped-completion", scoped=False)
|
|
unscoped_completion = write_report(
|
|
"quality-wf", "completion-unscoped", "completion-record", "ENG-BE",
|
|
completion_payload(implementation, "unscoped-completion"), "build")
|
|
check("unscoped completion receipt is rejected",
|
|
not SE.submit_report("quality-wf", unscoped_completion, "ENG-BE")[0])
|
|
|
|
add_receipt("completion-ok", "quality-wf")
|
|
completion = write_report("quality-wf", "completion-1", "completion-record", "ENG-BE",
|
|
completion_payload(implementation, "completion-ok"), "build")
|
|
check("scoped completion exact revision is accepted", SE.submit_report("quality-wf", completion, "ENG-BE")[0])
|
|
target = next(a for a in SE._trusted_artifacts("quality-wf") if a["artifact-id"] == "completion-1")
|
|
|
|
check("open verification", open_stage("quality-wf", "verification"))
|
|
no_failure_ok, no_failure_reasons = SE.can_transition(
|
|
"quality-wf", "build", actor="OPS-ORCH")
|
|
check("verification rework requires a current trusted Failed quality event",
|
|
not no_failure_ok
|
|
and any("quality_gate_status != Failed" in reason for reason in no_failure_reasons))
|
|
check("caller fact injection cannot authorize verification rework",
|
|
not SE.can_transition(
|
|
"quality-wf", "build", actor="OPS-ORCH",
|
|
ctx={"facts": {"quality-gate-failed": True}})[0])
|
|
with open(os.path.join(WS, "evidence", "ledger.jsonl"), "a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps({
|
|
"tool_use_id": "raw-bash-pass", "tool_name": "Bash", "command": "pytest",
|
|
"exit_code": 0, "ts": SE._now(), "workflow_id": "quality-wf",
|
|
"session_id": "fixture-session", "agent_id": "fixture-agent",
|
|
}) + "\n")
|
|
raw_quality = write_report(
|
|
"quality-wf", "raw-quality", "quality-gate-review", "QA",
|
|
quality_payload(target, "raw-bash-pass"), "verification")
|
|
raw_quality_ok, raw_quality_error = SE.record_quality_gate("quality-wf", raw_quality, "QA")
|
|
check("new workflow rejects untyped raw Bash receipt as a Passed quality proof",
|
|
not raw_quality_ok and "verification-run" in str(raw_quality_error))
|
|
|
|
add_receipt("stale-quality", "quality-wf", ts="2000-01-01T00:00:00Z")
|
|
stale_quality = write_report(
|
|
"quality-wf", "quality-stale", "quality-gate-review", "QA",
|
|
quality_payload(target, "stale-quality"), "verification")
|
|
check("receipt from before the verification stage is rejected",
|
|
not SE.record_quality_gate("quality-wf", stale_quality, "QA")[0])
|
|
|
|
add_receipt("qa-pass-1", "quality-wf")
|
|
qa_pass = write_report("quality-wf", "quality-qa-1", "quality-gate-review", "QA",
|
|
quality_payload(target, "qa-pass-1"), "verification")
|
|
check("fresh QA pass is recorded", SE.record_quality_gate("quality-wf", qa_pass, "QA")[0])
|
|
check("Passed quality cannot authorize verification rework",
|
|
not SE.can_transition("quality-wf", "build", actor="OPS-ORCH")[0])
|
|
|
|
add_receipt("sec-fail", "quality-wf", exit_code=1)
|
|
sec_fail = write_report("quality-wf", "quality-sec-1", "quality-gate-review", "SEC-ENGINEER",
|
|
quality_payload(target, "sec-fail", status="Failed"), "verification")
|
|
check("independent SEC failure is recorded", SE.record_quality_gate("quality-wf", sec_fail, "SEC-ENGINEER")[0])
|
|
add_receipt("qa-pass-2", "quality-wf")
|
|
qa_pass_2 = write_report("quality-wf", "quality-qa-2", "quality-gate-review", "QA",
|
|
quality_payload(target, "qa-pass-2"), "verification")
|
|
check("later QA pass is recorded", SE.record_quality_gate("quality-wf", qa_pass_2, "QA")[0])
|
|
check("later QA pass cannot overwrite another auditor's failure",
|
|
SE.read_ledger("quality-wf").get("quality_gate_status") == "Failed")
|
|
check("trusted current Failed quality authorizes verification rework",
|
|
SE.can_transition("quality-wf", "build", actor="OPS-ORCH")[0])
|
|
check("non-executor cannot perform verification rework",
|
|
not SE.can_transition("quality-wf", "build", actor="ENG-BE")[0])
|
|
check("release decision cannot be recorded before acceptance stage",
|
|
not SE.record_release_decision("quality-wf", qa_pass_2, "HUMAN-001")[0])
|
|
|
|
rework_ok, rework_reasons = SE.transition("quality-wf", "build", actor="OPS-ORCH")
|
|
rework_events = [
|
|
event for event in SE.read_workflow_events("quality-wf")
|
|
if event.get("event-type") == "state-transition"
|
|
]
|
|
check("Failed verification performs an audited rework transition",
|
|
rework_ok and not rework_reasons
|
|
and SE.current_stage("quality-wf") == "build"
|
|
and rework_events[-1].get("from") == "verification"
|
|
and rework_events[-1].get("to") == "build"
|
|
and rework_events[-1].get("actor") == "OPS-ORCH")
|
|
add_receipt("completion-2-ok", "quality-wf")
|
|
completion_2 = write_report("quality-wf", "completion-2", "completion-record", "ENG-BE",
|
|
completion_payload(implementation, "completion-2-ok"), "build")
|
|
check("new completion revision is accepted", SE.submit_report("quality-wf", completion_2, "ENG-BE")[0])
|
|
check("new completion invalidates all quality events bound to the old revision",
|
|
SE.read_ledger("quality-wf").get("quality_gate_status") is None)
|
|
|
|
|
|
print("== standard Passed quality receipts bind exact check and source revision ==")
|
|
binding_wf = "quality-binding-wf"
|
|
SE.init_ledger(binding_wf, tier="standard")
|
|
check("open standard build", open_stage(binding_wf, "build"))
|
|
binding_implementation = os.path.join(WS, "binding-implementation.txt")
|
|
with open(binding_implementation, "w", encoding="utf-8") as fh:
|
|
fh.write("standard implementation\n")
|
|
binding_source_sha = "d" * 64
|
|
binding_completion = write_report(
|
|
binding_wf, "binding-completion", "completion-record", "ENG-BE",
|
|
completion_payload(binding_implementation, "fixture-only"), "build")
|
|
with open(binding_completion, encoding="utf-8") as fh:
|
|
binding_completion_report = yaml.safe_load(fh)
|
|
binding_completion_report["payload"]["source-revision"]["sha256"] = binding_source_sha
|
|
with open(binding_completion, "w", encoding="utf-8") as fh:
|
|
yaml.safe_dump(binding_completion_report, fh, allow_unicode=True, sort_keys=False)
|
|
binding_completion_sha = sha(binding_completion)
|
|
check("seed trusted standard completion revision", SE._atomic_event_transaction(
|
|
binding_wf, artifact_event={
|
|
"artifact-event-id": "fixture-binding-completion",
|
|
"event-type": "artifact-submitted", "artifact-id": "binding-completion",
|
|
"report-id": "binding-completion", "workflow-id": binding_wf,
|
|
"artifact-kind": "completion-record", "artifact-version": 1, "stage": "build",
|
|
"producer-role-id": "ENG-BE", "path": os.path.relpath(binding_completion, WS),
|
|
"artifact-sha256": binding_completion_sha, "report-sha256": binding_completion_sha,
|
|
"effective-at": SE._now(),
|
|
})[0])
|
|
binding_target = next(a for a in SE._trusted_artifacts(binding_wf)
|
|
if a["artifact-id"] == "binding-completion")
|
|
check("open standard verification", open_stage(binding_wf, "verification"))
|
|
|
|
add_receipt("wrong-revision", binding_wf, subject="check-wrong-revision",
|
|
source_revision_sha256="e" * 64)
|
|
wrong_revision = write_report(
|
|
binding_wf, "quality-wrong-revision", "quality-gate-review", "QA",
|
|
quality_payload(binding_target, "wrong-revision"), "verification")
|
|
wrong_revision_ok, wrong_revision_error = SE.record_quality_gate(
|
|
binding_wf, wrong_revision, "QA")
|
|
check("64-hex receipt from another source revision is rejected",
|
|
not wrong_revision_ok and "최신 completion-record" in str(wrong_revision_error))
|
|
|
|
add_receipt("missing-revision", binding_wf, subject="check-missing-revision")
|
|
missing_revision = write_report(
|
|
binding_wf, "quality-missing-revision", "quality-gate-review", "QA",
|
|
quality_payload(binding_target, "missing-revision"), "verification")
|
|
missing_revision_ok, missing_revision_error = SE.record_quality_gate(
|
|
binding_wf, missing_revision, "QA")
|
|
check("standard Passed receipt without source revision is rejected",
|
|
not missing_revision_ok and "source_revision_sha256" in str(missing_revision_error))
|
|
|
|
add_receipt("wrong-subject", binding_wf, subject="another-check",
|
|
source_revision_sha256=binding_source_sha)
|
|
wrong_subject = write_report(
|
|
binding_wf, "quality-wrong-subject", "quality-gate-review", "QA",
|
|
quality_payload(binding_target, "wrong-subject"), "verification")
|
|
wrong_subject_ok, wrong_subject_error = SE.record_quality_gate(
|
|
binding_wf, wrong_subject, "QA")
|
|
check("Passed receipt subject must equal the quality check id",
|
|
not wrong_subject_ok and "verification_subject" in str(wrong_subject_error))
|
|
|
|
add_receipt("shared-pass", binding_wf, subject="check-shared-pass",
|
|
source_revision_sha256=binding_source_sha)
|
|
reused_payload = quality_payload(binding_target, "shared-pass")
|
|
reused_payload["checks"].append({
|
|
"check-id": "check-second", "category": "test", "status": "Passed",
|
|
"evidence-receipt-ids": ["shared-pass"],
|
|
})
|
|
reused_report = write_report(
|
|
binding_wf, "quality-reused-receipt", "quality-gate-review", "QA",
|
|
reused_payload, "verification")
|
|
reused_ok, reused_error = SE.record_quality_gate(binding_wf, reused_report, "QA")
|
|
check("one Passed receipt cannot prove multiple checks in one report",
|
|
not reused_ok and "재사용 금지" in str(reused_error))
|
|
|
|
add_receipt("bound-one", binding_wf, subject="check-bound-one",
|
|
source_revision_sha256=binding_source_sha)
|
|
add_receipt("bound-two", binding_wf, subject="check-bound-two",
|
|
source_revision_sha256=binding_source_sha)
|
|
bound_payload = quality_payload(binding_target, "bound-one")
|
|
bound_payload["checks"].append({
|
|
"check-id": "check-bound-two", "category": "test", "status": "Passed",
|
|
"evidence-receipt-ids": ["bound-two"],
|
|
})
|
|
bound_report = write_report(
|
|
binding_wf, "quality-exact-bindings", "quality-gate-review", "QA",
|
|
bound_payload, "verification")
|
|
check("distinct exact-subject receipts for the latest revision are accepted",
|
|
SE.record_quality_gate(binding_wf, bound_report, "QA")[0])
|
|
|
|
add_receipt("shared-failure", binding_wf, exit_code=1, subject="diagnostic-run")
|
|
failed_payload = quality_payload(binding_target, "shared-failure", status="Failed")
|
|
failed_payload["checks"].append({
|
|
"check-id": "check-another-failure", "category": "test", "status": "Failed",
|
|
"evidence-receipt-ids": ["shared-failure"],
|
|
})
|
|
failed_report = write_report(
|
|
binding_wf, "quality-failed-diagnostics", "quality-gate-review", "QA",
|
|
failed_payload, "verification")
|
|
check("Failed diagnostic reviews retain legacy subject/reuse behavior",
|
|
SE.record_quality_gate(binding_wf, failed_report, "QA")[0])
|
|
|
|
|
|
print("== typed data artifacts and exact lineage ==")
|
|
SE.init_ledger("data-wf", tier="light")
|
|
check("open decide for basis artifact", open_stage("data-wf", "decide"))
|
|
packet = write_report("data-wf", "packet-1", "executive-decision-packet", "EXEC-CEO",
|
|
{"recommendation": "proceed"}, "decide")
|
|
check("data basis packet submitted", SE.submit_report("data-wf", packet, "EXEC-CEO")[0])
|
|
packet_event = next(a for a in SE._trusted_artifacts("data-wf") if a["artifact-id"] == "packet-1")
|
|
check("open design for data model", open_stage("data-wf", "design"))
|
|
|
|
empty_model = write_report("data-wf", "model-empty", "data-model", "ARCH-DATA", {
|
|
"basis-artifact-id": "packet-1", "basis-artifact-sha256": packet_event["artifact-sha256"],
|
|
"conceptual": {}, "logical": {}, "physical": {}, "ownership": {},
|
|
"classification": {"pii": False, "sensitivity": "internal"},
|
|
"lineage": [], "retention": {}, "compatibility": {}, "data-quality-thresholds": [],
|
|
}, "design")
|
|
check("empty data-model placeholders are rejected", not SE.submit_report("data-wf", empty_model, "ARCH-DATA")[0])
|
|
|
|
valid_model_payload = {
|
|
"basis-artifact-id": "packet-1", "basis-artifact-sha256": packet_event["artifact-sha256"],
|
|
"conceptual": {"entities": ["Account"]},
|
|
"logical": {"tables": ["accounts"]},
|
|
"physical": {"engine": "postgresql"},
|
|
"ownership": {"owner": "DATA-ENGINEER"},
|
|
"classification": {"pii": False, "sensitivity": "internal"},
|
|
"lineage": [{"from": "source.accounts", "to": "warehouse.accounts"}],
|
|
"retention": {"policy": "delete", "duration": "365d"},
|
|
"compatibility": {"schema-version": "1", "evolution-policy": "backward"},
|
|
"data-quality-thresholds": [{"metric": "primary-key-null-rate", "max": 0}],
|
|
}
|
|
bad_basis = dict(valid_model_payload)
|
|
bad_basis["basis-artifact-sha256"] = "f" * 64
|
|
bad_basis_report = write_report("data-wf", "model-bad-basis", "data-model", "ARCH-DATA",
|
|
bad_basis, "design")
|
|
check("data-model cannot cite a fabricated basis SHA",
|
|
not SE.submit_report("data-wf", bad_basis_report, "ARCH-DATA")[0])
|
|
model = write_report("data-wf", "model-1", "data-model", "ARCH-DATA",
|
|
valid_model_payload, "design")
|
|
check("complete data-model with exact lineage basis is accepted",
|
|
SE.submit_report("data-wf", model, "ARCH-DATA")[0])
|
|
|
|
snapshot = os.path.join(WS, "dataset.csv")
|
|
with open(snapshot, "w", encoding="utf-8") as fh:
|
|
fh.write("account_id,active\n1,true\n")
|
|
empty_metrics = write_report("data-wf", "metrics-empty", "metrics-analysis", "DATA-ANALYST",
|
|
{}, "design")
|
|
check("empty metrics analysis is rejected", not SE.submit_report("data-wf", empty_metrics, "DATA-ANALYST")[0])
|
|
|
|
add_receipt("metrics-unscoped", scoped=False)
|
|
metrics_payload = {
|
|
"metric-contract": {
|
|
"metric-id": "active-account-rate", "version": "1", "grain": "account",
|
|
"numerator": "active accounts", "denominator": "all accounts", "unit": "ratio",
|
|
"timezone": "UTC", "observation-window": "2026-07-01/2026-07-16",
|
|
"source-fields": ["account_id", "active"],
|
|
},
|
|
"dataset-snapshot": {
|
|
"snapshot-id": "snapshot-1", "path": snapshot, "sha256": sha(snapshot),
|
|
"schema-version": "1", "row-count": 1, "as-of": "2026-07-16T00:00:00Z",
|
|
"classification": "internal",
|
|
},
|
|
"analysis-run": {
|
|
"query-sha256": "b" * 64, "environment": "fixture",
|
|
"result-sha256": "c" * 64, "evidence-receipt-ids": ["metrics-unscoped"],
|
|
},
|
|
"findings": [{"finding-id": "F-1", "summary": "one active account", "value": 1.0}],
|
|
"limitations": ["fixture sample"],
|
|
}
|
|
metrics_unscoped = write_report("data-wf", "metrics-unscoped", "metrics-analysis", "DATA-ANALYST",
|
|
metrics_payload, "design")
|
|
check("unscoped analysis execution receipt is rejected",
|
|
not SE.submit_report("data-wf", metrics_unscoped, "DATA-ANALYST")[0])
|
|
add_receipt("metrics-ok", "data-wf")
|
|
metrics_payload["analysis-run"]["evidence-receipt-ids"] = ["metrics-ok"]
|
|
metrics = write_report("data-wf", "metrics-1", "metrics-analysis", "DATA-ANALYST",
|
|
metrics_payload, "design")
|
|
check("snapshot-bound metrics analysis with scoped execution receipt is accepted",
|
|
SE.submit_report("data-wf", metrics, "DATA-ANALYST")[0])
|
|
|
|
|
|
print("== token/KPI/JSONL ledgers reject fabricated inputs ==")
|
|
env = {**os.environ, "CLAUDE_PROJECT_DIR": ROOT, "ORGOS_WORKSPACE": WS}
|
|
token_cli = os.path.join(HOOKS, "token_ledger.py")
|
|
kpi_cli = os.path.join(HOOKS, "kpi_ledger.py")
|
|
negative_token = subprocess.run(
|
|
[sys.executable, token_cli, "log", "--workflow", "data-wf", "--role", "DATA-ANALYST",
|
|
"--tokens", "-1", "--tier", "light"], env=env, capture_output=True, text=True)
|
|
check("negative token usage is rejected", negative_token.returncode == 2)
|
|
unknown_token_tier = subprocess.run(
|
|
[sys.executable, token_cli, "log", "--workflow", "data-wf", "--role", "DATA-ANALYST",
|
|
"--tokens", "10", "--tier", "premium"], env=env, capture_output=True, text=True)
|
|
check("unknown token tier is rejected", unknown_token_tier.returncode == 2)
|
|
fake_kpi = subprocess.run(
|
|
[sys.executable, kpi_cli, "log", "--metric", "looks-good-rate", "--value", "1",
|
|
"--window-start", "2026-07-01", "--window-end", "2026-07-16"],
|
|
env=env, capture_output=True, text=True)
|
|
check("unregistered KPI cannot be injected", fake_kpi.returncode == 2)
|
|
|
|
broken = os.path.join(WS, "state", "broken.jsonl")
|
|
os.makedirs(os.path.dirname(broken), exist_ok=True)
|
|
with open(broken, "w", encoding="utf-8") as fh:
|
|
fh.write('{"event": "torn"\n')
|
|
doctor_report = DOCTOR.Report()
|
|
DOCTOR.check_jsonl_integrity(doctor_report)
|
|
check("doctor reports malformed JSONL as a hard failure", doctor_report.n_fail > 0)
|
|
os.remove(broken)
|
|
|
|
print(f"\n{passed} passed, {failed} failed")
|
|
exit_code = 1 if failed else 0
|
|
shutil.rmtree(WS, ignore_errors=True)
|
|
sys.exit(exit_code)
|