fix(report): bind professional report hash to the delivered packet (BUG-693, BUG-694)
Independent Staging Quality Gate / validate (push) Successful in 9m45s
Independent Staging Quality Gate / publish (push) Successful in 2m14s

Rebind after sanitize, write an explicit binding_scope, and make the quality gate recompute coverage. Wall-clock fields stay in the packet but out of the hash.
This commit is contained in:
jesse-ux
2026-09-15 10:09:26 +08:00
parent 039b0a2608
commit 1d2aeffaef
8 changed files with 403 additions and 32 deletions
+98 -14
View File
@@ -70,6 +70,27 @@ EPHEMERIS_FIELDS = (
POSITION_MODES = ("legacy", "mean", "apparent")
NODE_MODES = ("mean", "true")
# Envelope keys never enter the result hash (self-describing binding metadata).
RESULT_BINDING_ENVELOPE_KEYS = frozenset({
"result_hash",
"result_binding",
"calculation_profile",
"calculation_profile_id",
})
# D2: only these three classes may be excluded from the delivered-object hash.
ALLOWED_BINDING_EXCLUDED_TOP_KEYS = (
"generated_at",
"report_quality_gate",
"shared_full_report_authority",
)
ALLOWED_BINDING_EXCLUDED_PATHS = (
"ai_and_audit.summary.elapsed_seconds",
"**.elapsed_seconds",
"**.called_at",
"**.cache_created_at",
"**.cache_expires_at",
)
class CalculationProfileError(ValueError):
"""Raised for inputs that cannot be normalized without data loss."""
@@ -431,6 +452,76 @@ def attach_calculation_profile(result: dict[str, Any], args: Any) -> dict[str, A
return bind_result_to_profile(result, profile)
def default_result_binding_scope() -> dict[str, list[str]]:
"""Explicit exclusion set written onto every result_binding receipt."""
return {
"excluded_top_keys": list(ALLOWED_BINDING_EXCLUDED_TOP_KEYS),
"excluded_paths": list(ALLOWED_BINDING_EXCLUDED_PATHS),
}
def _drop_key_recursive(payload: Any, key_name: str) -> Any:
if isinstance(payload, dict):
return {
key: _drop_key_recursive(value, key_name)
for key, value in payload.items()
if key != key_name
}
if isinstance(payload, list):
return [_drop_key_recursive(item, key_name) for item in payload]
return payload
def _drop_dotted_path(payload: Any, path: tuple[str, ...]) -> Any:
if len(path) == 2 and path[0] == "**":
return _drop_key_recursive(payload, path[1])
if not path or not isinstance(payload, dict) or path[0] not in payload:
return payload
cloned = dict(payload)
key, *rest = path
if not rest:
cloned.pop(key, None)
return cloned
cloned[key] = _drop_dotted_path(payload[key], tuple(rest))
return cloned
def result_payload_for_binding(
result: dict[str, Any],
scope: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Business payload that participates in result_hash."""
binding_scope = scope if isinstance(scope, dict) else default_result_binding_scope()
excluded_top = RESULT_BINDING_ENVELOPE_KEYS | {
str(key) for key in (binding_scope.get("excluded_top_keys") or ())
}
payload = {key: value for key, value in result.items() if key not in excluded_top}
for raw_path in binding_scope.get("excluded_paths") or ():
parts = tuple(part for part in str(raw_path).split(".") if part)
if parts:
payload = _drop_dotted_path(payload, parts)
return payload
def hash_bound_result(
result: dict[str, Any],
input_hash: str,
scope: dict[str, Any] | None = None,
) -> str:
"""SHA-256 of the canonical bound payload. Does not mutate ``result``."""
encoded = json.dumps(
canonicalize_result_payload({
"input_hash": input_hash,
"result": result_payload_for_binding(result, scope),
}),
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
default=str,
)
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
def bind_result_to_profile(result: dict[str, Any], profile: dict[str, Any]) -> dict[str, Any]:
"""Bind one concrete calculation result to its normalized input profile."""
if not isinstance(result, dict) or not isinstance(profile, dict):
@@ -438,21 +529,14 @@ def bind_result_to_profile(result: dict[str, Any], profile: dict[str, Any]) -> d
input_hash = profile.get("input_hash")
if not isinstance(input_hash, str) or len(input_hash) != 64:
raise ValueError("calculation profile is missing a valid input_hash")
result_payload = {
key: value
for key, value in result.items()
if key not in {"result_hash", "result_binding", "calculation_profile", "calculation_profile_id"}
}
encoded = json.dumps(
canonicalize_result_payload({"input_hash": input_hash, "result": result_payload}),
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
default=str,
)
result_hash = hashlib.sha256(encoded.encode("utf-8")).hexdigest()
scope = default_result_binding_scope()
result_hash = hash_bound_result(result, input_hash, scope)
result["result_hash"] = result_hash
result["result_binding"] = {"input_hash": input_hash, "result_hash": result_hash}
result["result_binding"] = {
"input_hash": input_hash,
"result_hash": result_hash,
"binding_scope": scope,
}
return result
+52
View File
@@ -9,6 +9,19 @@ import re
from pathlib import Path
from typing import Any
try:
from scripts.calculation_profile_contract import (
ALLOWED_BINDING_EXCLUDED_PATHS,
ALLOWED_BINDING_EXCLUDED_TOP_KEYS,
hash_bound_result,
)
except ImportError: # pragma: no cover - direct-script execution
from calculation_profile_contract import (
ALLOWED_BINDING_EXCLUDED_PATHS,
ALLOWED_BINDING_EXCLUDED_TOP_KEYS,
hash_bound_result,
)
SCHEMA_VERSION = "jyotish.full_report_quality_gate.v1"
REPORT_SCHEMA = "pl9_style_professional_export_v1"
@@ -170,6 +183,45 @@ def evaluate_full_report(packet: dict[str, Any], rendered_markdown: str | None =
else:
checks.append(_check("provenance:result_binding", "passed", "input/result hashes match"))
scope = _as_dict(binding.get("binding_scope"))
if "excluded_top_keys" not in scope or "excluded_paths" not in scope:
blocking_reasons.append("provenance_binding_scope_absent")
checks.append(_check(
"provenance:result_binding_scope",
"blocked",
"provenance_binding_scope_absent",
))
else:
extra_top = {
str(item) for item in (scope.get("excluded_top_keys") or ())
} - set(ALLOWED_BINDING_EXCLUDED_TOP_KEYS)
extra_paths = {
str(item) for item in (scope.get("excluded_paths") or ())
} - set(ALLOWED_BINDING_EXCLUDED_PATHS)
if extra_top or extra_paths:
blocking_reasons.append("provenance_binding_scope_unexpected_exclusion")
checks.append(_check(
"provenance:result_binding_scope",
"blocked",
"provenance_binding_scope_unexpected_exclusion",
))
elif (
isinstance(expected_input_hash, str)
and hash_bound_result(packet, expected_input_hash, scope) != expected_result_hash
):
blocking_reasons.append("provenance_binding_scope_mismatch")
checks.append(_check(
"provenance:result_binding_scope",
"blocked",
"provenance_binding_scope_mismatch",
))
else:
checks.append(_check(
"provenance:result_binding_scope",
"passed",
"clipped payload matches result_hash",
))
chart_identity = _as_dict(packet.get("chart_identity"))
for key in (
"chart_profile_id",
+25 -14
View File
@@ -109,18 +109,23 @@ if SCRIPT_DIR not in sys.path:
try:
from calculation_profile_contract import (
attach_calculation_profile,
bind_result_to_profile,
build_calculation_profile,
)
except ImportError: # pragma: no cover - package import
try:
from scripts.calculation_profile_contract import (
attach_calculation_profile,
bind_result_to_profile,
build_calculation_profile,
)
except ImportError:
def attach_calculation_profile(payload, args=None):
return payload
def bind_result_to_profile(result, profile=None):
return result
def build_calculation_profile(args=None):
return {"status": "blocked", "reason": "calculation_profile_contract_absent"}
@@ -12398,8 +12403,8 @@ def cmd_yoga(args):
yoga_context = None
ai = SIGNS.index(asc) if asc in SIGNS else 0
kl = list(set([SIGN_LORDS[SIGNS[(ai + h - 1) % 12]] for h in [1, 4, 7, 10]]))
tl = list(set([SIGN_LORDS[SIGNS[(ai + h - 1) % 12]] for h in [1, 5, 9]]))
kl = sorted({SIGN_LORDS[SIGNS[(ai + h - 1) % 12]] for h in [1, 4, 7, 10]})
tl = sorted({SIGN_LORDS[SIGNS[(ai + h - 1) % 12]] for h in [1, 5, 9]})
# 调用数据驱动引擎(yoga_engine.py
yogas = detect_yogas(planets, asc, context=yoga_context)
@@ -17775,27 +17780,33 @@ def build_professional_report_reference_packet(
selected = _apply_pl9_pack_selection(packet, selected_pack_ids or [])
final_packet = _attach_full_report_pack(selected, args)
final_packet = _attach_startrack_language_bridge(final_packet, args)
final_packet = attach_calculation_profile(final_packet, args)
final_packet = _attach_default_chart_identity(final_packet)
final_packet = _attach_report_governance_contracts(final_packet, args)
final_packet['reader_engine_boundary_notice'] = _professional_report_reference_boundary_notice()
final_packet['generated_at'] = datetime.utcnow().replace(microsecond=0).isoformat() + 'Z'
if final_packet.get('selected_report_scope') == 'full':
delivered = sanitize_professional_report_reference(final_packet)
profile = delivered.get('calculation_profile') if isinstance(delivered, dict) else None
if (
isinstance(profile, dict)
and isinstance(profile.get('input_hash'), str)
and len(profile['input_hash']) == 64
):
delivered = bind_result_to_profile(delivered, profile)
if isinstance(delivered, dict) and delivered.get('selected_report_scope') == 'full':
evaluate_full_report = _try_attr_import('full_report_quality_gate', 'evaluate_full_report')
if evaluate_full_report is None:
final_packet['report_quality_gate'] = {
delivered['report_quality_gate'] = {
'status': 'blocked',
'reason': 'full_report_quality_gate_absent',
}
else:
try:
public_quality_input = sanitize_professional_report_reference(final_packet)
final_packet['report_quality_gate'] = evaluate_full_report(
public_quality_input,
render_pl9_markdown(public_quality_input),
delivered['report_quality_gate'] = evaluate_full_report(
delivered,
render_pl9_markdown(delivered),
)
except Exception:
final_packet['report_quality_gate'] = {
delivered['report_quality_gate'] = {
'status': 'blocked',
'reason': 'full_report_quality_gate_failed',
}
@@ -17803,7 +17814,7 @@ def build_professional_report_reference_packet(
'shared_full_report_authority', 'build_shared_full_report_authority'
)
if build_shared_full_report_authority is None:
final_packet['shared_full_report_authority'] = {
delivered['shared_full_report_authority'] = {
'schema_version': 'jyotish.shared_full_report_authority.v1',
'status': 'blocked',
'reason': 'shared_full_report_authority_absent',
@@ -17811,15 +17822,15 @@ def build_professional_report_reference_packet(
}
else:
try:
final_packet['shared_full_report_authority'] = build_shared_full_report_authority(final_packet)
delivered['shared_full_report_authority'] = build_shared_full_report_authority(delivered)
except Exception:
final_packet['shared_full_report_authority'] = {
delivered['shared_full_report_authority'] = {
'schema_version': 'jyotish.shared_full_report_authority.v1',
'status': 'blocked',
'reason': 'shared_full_report_authority_failed',
'read_only': True,
}
return sanitize_professional_report_reference(final_packet)
return delivered
def cmd_pl9_export(args):
full = cmd_full_reading(args)