fix(scene): verify runtime routes and release flags

This commit is contained in:
2026-10-06 12:13:02 +08:00
parent d405c1ba65
commit 2078eeb56b
5 changed files with 64 additions and 15 deletions
@@ -7,6 +7,7 @@ from yuqei_sdk.review_routing import (
SCENE_REVIEW_ROUTES,
REVIEW_ROUTING_VERSION,
scene_review_evaluation_profile,
scene_review_route,
)
from .authoritative_legal_sources import load_authoritative_legal_source_manifest
@@ -29,7 +30,13 @@ def _identity(snapshot):
def build_scene_runtime_manifest(store, settings, *, template_id, contract_type):
route = SCENE_REVIEW_ROUTES.get(template_id)
if route is None or route[0] != contract_type:
if route is None:
raise ValueError("场景与合同类型未在服务端登记,无法核对运行路线。")
try:
requested_review_route = scene_review_route(contract_type, template_id)
except ValueError as error:
raise ValueError("场景与合同类型未在服务端登记,无法核对运行路线。") from error
if requested_review_route != route[1]:
raise ValueError("场景与合同类型未在服务端登记,无法核对运行路线。")
request = SimpleNamespace(contract_type=contract_type, scene_template_id=template_id,
review_profile=scene_review_evaluation_profile(template_id))
@@ -43,7 +50,7 @@ def build_scene_runtime_manifest(store, settings, *, template_id, contract_type)
if not ordinary:
issues.append("review_prompt_missing")
else:
if ordinary["name"] != route[1]:
if ordinary["name"] != requested_review_route:
issues.append("review_runtime_fallback")
if ordinary["status"] != "published" or not ordinary["enabled"]:
issues.append("review_prompt_not_published")
@@ -54,7 +61,7 @@ def build_scene_runtime_manifest(store, settings, *, template_id, contract_type)
sources = load_authoritative_legal_source_manifest()
payload = {"schema_version": "scene-runtime-manifest.v1", "tenant_id": current_tenant_id(),
"template_id": template_id, "contract_type": contract_type, "routing_version": REVIEW_ROUTING_VERSION,
"requested_review_route": route[1], "draft_prompt": drafting,
"requested_review_route": requested_review_route, "draft_prompt": drafting,
"prompt_storage_scope": "platform_shared",
"trial_review_prompt": trial, "runtime_review_prompt": ordinary,
"provider_mode": settings.provider_call_mode,
@@ -13,6 +13,7 @@ from yuqei_ai_platform_api.contract_drafting import capture_drafting_prompt_snap
from yuqei_ai_platform_api.main import create_app
from yuqei_ai_platform_api.review_scene_prompts import scene_review_template
from yuqei_ai_platform_api.scene_runtime_manifest import build_scene_runtime_manifest
from yuqei_contract_api.contract_templates import get_contract_template
from yuqei_ai_platform_api.scene_evaluation_gate import (
BATCH_B_DRAFTING_FAILURE_CASES,
BATCH_B_SECOND_DRAFTING_FAILURE_CASES,
@@ -47,8 +48,9 @@ def fixture_store(template_id=SCENE):
@pytest.mark.parametrize("template_id", SCENE_REVIEW_ROUTES)
def test_manifest_matches_actual_routes_without_returning_prompt_content(template_id):
_, store = fixture_store(template_id)
template = get_contract_template(template_id)
manifest = build_scene_runtime_manifest(store, AiPlatformSettings(provider_call_mode="codex_cli"),
template_id=template_id, contract_type=SCENE_REVIEW_ROUTES[template_id][0])
template_id=template_id, contract_type=template.contract_type)
assert manifest["alignment_errors"] == []
assert manifest["published_prompt_alignment_verified"] is True
assert manifest["model"] == "gpt-5.6-sol" and manifest["model_calls"] == 0
@@ -667,15 +667,23 @@ class DraftingWorkspaceStore:
"review_coverage_by_prompt": [{**group, "complete_variants": sorted(group["complete_variants"]),
"passed_variants": sorted(group["passed_variants"]),
"missing_variants": sorted(set(VARIANTS) - group["passed_variants"])} for group in review_groups.values()],
"truncated": len(data) > limit or record_limit,
"source_applicability_verified": False, "published_prompt_alignment_verified": False,
"authorizes_publication": False}
"truncated": len(data) > limit or record_limit, "authorizes_publication": False}
from .offline_scene_evidence_registry import merge_offline_scene_evidence, scene_offline_evidence
summary = merge_offline_scene_evidence(summary, scene_offline_evidence(
template_id, current_template_content_sha256=current.content_sha256))
from .holdout_scene_evidence_registry import merge_holdout_scene_evidence, scene_holdout_evidence
return merge_holdout_scene_evidence(summary, scene_holdout_evidence(
summary = merge_holdout_scene_evidence(summary, scene_holdout_evidence(
template_id, current_template_content_sha256=current.content_sha256))
runtime_reader = getattr(self.repository, "template_runtime_manifest_reader", None)
runtime = runtime_reader(current, tenant) if runtime_reader else None
from .template_release_readiness import assess_release_readiness
readiness = assess_release_readiness(current, summary, runtime, tenant=tenant)
summary.update({
"source_applicability_verified": readiness.get("source_applicability_verified") is True,
"published_prompt_alignment_verified": readiness.get("published_prompt_alignment_verified") is True,
"release_readiness": readiness,
})
return summary
def evaluation_source(self, template_id):
"""Resolve an editable snapshot without publishing it or changing history."""
@@ -19,13 +19,7 @@ class ContractRepository:
if store is None:
return None
tenant = get_tenant_context() or "default"
summary = store.template_evaluation_summary(template_id, tenant=tenant)
template, _ = store.evaluation_source(template_id)
reader = getattr(self, "template_runtime_manifest_reader", None)
runtime = reader(template, tenant) if reader else None
from .template_release_readiness import assess_release_readiness
summary["release_readiness"] = assess_release_readiness(template, summary, runtime, tenant=tenant)
return summary
return store.template_evaluation_summary(template_id, tenant=tenant)
def create_contract(self, request: ContractCreateRequest) -> ContractDetail:
raise NotImplementedError
@@ -87,6 +87,44 @@ def test_passing_synthetic_model_records_do_not_approve_unverified_sources():
assert "SCENE_SOURCE_APPLICABILITY_PENDING" in codes(assess_release_readiness(altered, summary, runtime, tenant="default"))
def test_summary_top_level_release_flags_follow_only_verified_server_readiness(monkeypatch):
template, _, runtime = evidence()
repository = InMemoryContractRepository(contracts=[])
repository.template_runtime_manifest_reader = lambda current, tenant: runtime
trial = DraftingWorkspaceStore(repository, mode="scene_evaluation")
current, _ = trial.evaluation_source(SCENE)
assert current.content_sha256 == template.content_sha256
verified_canonical = get_contract_template(SCENE).model_copy(deep=True)
for source in verified_canonical.input_schema.scene["source_references"]:
source["current_effectiveness_verified"] = True
monkeypatch.setattr(
"yuqei_contract_api.template_release_readiness.get_contract_template",
lambda template_id: verified_canonical,
)
summary = trial.template_evaluation_summary(SCENE, tenant="default")
assert summary["source_applicability_verified"] is True
assert summary["published_prompt_alignment_verified"] is True
assert summary["source_applicability_verified"] is summary["release_readiness"]["source_applicability_verified"]
assert summary["published_prompt_alignment_verified"] is summary["release_readiness"]["published_prompt_alignment_verified"]
def test_summary_top_level_release_flags_reject_cached_claims_and_bad_runtime_hash():
_, _, runtime = evidence()
runtime["source_applicability_verified"] = True
runtime["published_prompt_alignment_verified"] = True
runtime["manifest_sha256"] = "0" * 64
repository = InMemoryContractRepository(contracts=[])
repository.template_runtime_manifest_reader = lambda current, tenant: runtime
trial = DraftingWorkspaceStore(repository, mode="scene_evaluation")
summary = trial.template_evaluation_summary(SCENE, tenant="default")
assert summary["source_applicability_verified"] is False
assert summary["published_prompt_alignment_verified"] is False
assert "SCENE_RUNTIME_UNAVAILABLE" in codes(summary["release_readiness"])
@pytest.mark.parametrize("change", ["tenant", "hash", "time", "naive_time", "shape", "route", "missing"])
def test_invalid_runtime_never_reuses_cached_alignment(change):
template, summary, runtime = evidence()