295 lines
20 KiB
Python
295 lines
20 KiB
Python
"""Read-only asset audit and isolated defect probes; never calls a model or publishes assets.
|
|
|
|
Run from the repository root. --runtime adds PostgreSQL READ ONLY transactions
|
|
using existing service settings. Business contract text, credentials and user IDs
|
|
are not collected. The only writes are an explicitly requested local JSON report
|
|
and transient SQLite :memory: data used by a publication-isolation probe.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections import Counter
|
|
from datetime import UTC, datetime
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
CONTRACT = ROOT / "yuqei-contract-platform/services/contract-api"
|
|
AI = ROOT / "yuqei-ai-platform/services/ai-platform-api"
|
|
sys.path[:0] = [str(CONTRACT / "src"), str(AI / "src")]
|
|
|
|
from yuqei_contract_api.contract_templates import ( # noqa: E402
|
|
ContractTemplateEditDraft, ContractTemplateItem, ContractTemplateVariable,
|
|
TemplateRenderError, get_contract_template_body, list_contract_templates,
|
|
render_contract_template, render_template_body,
|
|
validate_contract_template_inputs,
|
|
)
|
|
from yuqei_contract_api.template_inputs import input_schema_hash, template_sample_values # noqa: E402
|
|
from yuqei_contract_api.contract_body_quality import inspect_contract_body_quality # noqa: E402
|
|
from yuqei_contract_api.lifecycle_helpers import validate_template_edit_draft # noqa: E402
|
|
from yuqei_ai_platform_api.prompts import ( # noqa: E402
|
|
_render_packaged_prompt_asset, _render_template, contract_review_prompt_candidates,
|
|
)
|
|
from yuqei_ai_platform_api.prompt_compiler import PromptVersionConflict # noqa: E402
|
|
|
|
PROMPT_FILE = AI / "src/yuqei_ai_platform_api/prompt_assets/v1_phase15/default-prompt-assets.json"
|
|
TOKEN = re.compile(r"{{\s*([^{}]+?)\s*}}")
|
|
|
|
|
|
def sha(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def source_inventory() -> dict:
|
|
templates = []
|
|
for item in list_contract_templates().items:
|
|
body = get_contract_template_body(item.id)
|
|
required = {v.name for v in item.required_variables}
|
|
optional = {v.name for v in item.optional_variables}
|
|
placeholders = set(TOKEN.findall(body))
|
|
checks = []
|
|
for mode, variables in (("required_only", item.required_variables),
|
|
("all_fields", [*item.required_variables, *item.optional_variables])):
|
|
values = {v.name: f"AUDIT-VALUE-{i:02d}" for i, v in enumerate(variables)}
|
|
if item.input_schema:
|
|
values = template_sample_values(item.input_schema, [v.name for v in item.required_variables], [v.name for v in item.optional_variables], with_optional=mode == "all_fields")
|
|
rendered = render_contract_template(item.id, values)
|
|
validation = validate_contract_template_inputs(item, values)
|
|
body_values = {name: value for name, value in validation.normalized_values.items() if not item.input_schema or item.input_schema.fields[name].role == "body_fact"}
|
|
checks.append({
|
|
"mode": mode, "body_length": len(rendered),
|
|
"quality_codes": [i.code for i in inspect_contract_body_quality(rendered)],
|
|
"missing_required_values": sorted(v for k, v in body_values.items() if k in validation.required_fields and v not in rendered),
|
|
"instruction_residue_markers": [m for m in ("不得默认生成", "不得自动生成", "必须转专项", "需法务", "必须由业务和法务确认") if m in rendered],
|
|
})
|
|
templates.append({
|
|
"id": item.id, "name": item.name, "type": item.contract_type,
|
|
"version": item.version, "body_sha256": sha(body),
|
|
"required_count": len(required), "optional_count": len(optional),
|
|
"required_fields": sorted(required), "optional_fields": sorted(optional),
|
|
"undeclared_placeholders": sorted(placeholders - required - optional),
|
|
"unused_required_fields": sorted(required - placeholders),
|
|
"unused_optional_fields_in_body": sorted(optional - placeholders),
|
|
"outside_body_roles": {name: spec.role for name, spec in item.input_schema.fields.items() if spec.role != "body_fact"} if item.input_schema else {},
|
|
"input_schema_sha256": input_schema_hash(item.input_schema),
|
|
"executable_rule_ids": [rule.id for rule in item.input_schema.rules] if item.input_schema else [],
|
|
"input_fixture_count": len(item.input_schema.fixtures) if item.input_schema else 0,
|
|
"review_prompt_candidates": contract_review_prompt_candidates(item.contract_type),
|
|
"render_checks": checks,
|
|
})
|
|
assets = json.loads(PROMPT_FILE.read_text(encoding="utf-8"))["assets"]
|
|
prompts = []
|
|
for asset in assets:
|
|
schema = asset.get("output_schema") or {}
|
|
prompts.append({
|
|
"key": asset["prompt_key"], "name": asset["name"], "type": asset["prompt_type"],
|
|
"content_sha256": sha(asset["prompt_content"]),
|
|
"content_length": len(asset["prompt_content"]),
|
|
"system_sha256": sha(asset.get("system_prompt", "")),
|
|
"required_variables": [v["name"] for v in asset.get("variables", []) if v.get("required")],
|
|
"schema_required": schema.get("required", []),
|
|
"untyped_array_fields": [k for k, v in schema.get("properties", {}).items()
|
|
if isinstance(v, dict) and v.get("type") == "array" and "items" not in v],
|
|
"embedded_article_numbers": bool(re.search(r"第\d+", asset["prompt_content"])),
|
|
})
|
|
return {
|
|
"template_count": len(templates), "required_field_count": sum(t["required_count"] for t in templates),
|
|
"optional_field_count": sum(t["optional_count"] for t in templates),
|
|
"prompt_count": len(prompts), "prompt_types": dict(Counter(p["type"] for p in prompts)),
|
|
"template_types_using_generic_review": [t["type"] for t in templates if len(t["review_prompt_candidates"]) == 2],
|
|
"templates": templates, "prompts": prompts,
|
|
}
|
|
|
|
|
|
def isolated_probes() -> list[dict]:
|
|
results: list[dict] = []
|
|
|
|
def check(key: str, expectation: str, passed: bool, observed: object) -> None:
|
|
results.append({"id": key, "expectation": expectation,
|
|
"status": "pass" if passed else "fail", "observed": observed})
|
|
|
|
variables = {"contract_text": "原文保留 {counterparty}", "counterparty": "被替换的公司"}
|
|
output = _render_template("专项约束\n{{contract_text}}", "通用\n{contract_text}", variables)
|
|
check("prompt_recursive_substitution", "用户数据中的变量形状原样保留", variables["contract_text"] in output, output)
|
|
variables["contract_text"] = '接口格式 {"amount": 100}'
|
|
output = _render_template("专项约束\n{{contract_text}}", "通用\n{contract_text}", variables)
|
|
check("prompt_json_fallback", "正文包含 JSON 不应丢弃专项约束", "专项约束" in output, output)
|
|
variables["contract_text"] = "原文保留 {counterparty}"
|
|
imported = json.dumps({"asset": {"prompt_content": "原始正文：{{contract_text}}"}})
|
|
output = _render_packaged_prompt_asset(imported, variables)
|
|
check("imported_prompt_recursive_substitution", "导入资产也不能二次解释用户变量", variables["contract_text"] in output, output)
|
|
imported = json.dumps({"asset": {"prompt_content": "正文：{{contract_text}}", "variables": [{"name": "contract_text", "required": True}]}})
|
|
try:
|
|
output = _render_packaged_prompt_asset(imported, {})
|
|
rejected = False
|
|
except (ValueError, KeyError):
|
|
output, rejected = "rejected", True
|
|
check("prompt_required_input", "缺少声明的必填输入应明确报错", rejected, output)
|
|
|
|
template = ContractTemplateItem(id="AUDIT-ONLY", name="隔离测试", version="draft", contract_type="test", body_preview="",
|
|
required_variables=[ContractTemplateVariable(name="甲方", label="甲方")],
|
|
optional_variables=[ContractTemplateVariable(name="补充说明", label="补充说明", required=False)])
|
|
output = render_template_body(template, "甲方为{{甲方}}（补充{{补充说明}}）提供服务。", {"甲方": "隔离甲方"})
|
|
check("optional_parenthetical_baseline", "独立可选括号不应删除必填信息", "隔离甲方" in output, output)
|
|
try:
|
|
output = render_template_body(template, "甲方{{甲方}}备注{{补充说明}}。双方同意按约履行。", {"甲方": "隔离甲方"})
|
|
rejected = False
|
|
except TemplateRenderError as error:
|
|
output, rejected = str(error), True
|
|
check("optional_segment_deletes_required", "混合句段中仍应保留必填值或拒绝渲染", rejected or "隔离甲方" in output, {"rejected": rejected, "output": output})
|
|
output = render_template_body(template, "服务约定：{{甲方}}。其余条款继续有效。", {"甲方": "业务系统不得泄露商业秘密"})
|
|
check("internal_cleanup_deletes_user_content", "用户约定不能被内部说明清理规则静默删除", "业务系统不得泄露商业秘密" in output, output)
|
|
codes = [i.code for i in inspect_contract_body_quality("【特别提示】付款后五个工作日交货。")]
|
|
check("bracket_emphasis_false_positive", "普通强调标题不应被当作未填占位符", not codes, codes)
|
|
codes = [i.code for i in inspect_contract_body_quality("甲方承诺按时交货。补充约定{{尚未闭合")]
|
|
check("unclosed_placeholder_missed", "未闭合的模板标记应被识别", bool(codes), codes)
|
|
|
|
def draft(body: str) -> ContractTemplateEditDraft:
|
|
return ContractTemplateEditDraft(template_id=template.id, base_version="draft", draft_version="audit",
|
|
body_content=body, required_variables=["甲方"], optional_variables=[],
|
|
clause_ids=["CLAUSE-AUDIT"], review_rule_ids=["RULE-AUDIT"], actor_name="isolated-audit", updated_at=datetime.now(UTC))
|
|
|
|
unknown_body = "甲方{{甲方}}。" + "双方依法履行本合同的约定义务。" * 8 + "补充{{未声明}}。"
|
|
validation = validate_template_edit_draft(template, draft(unknown_body), clause_ids={"CLAUSE-AUDIT"}, review_rule_ids={"RULE-AUDIT"})
|
|
try:
|
|
render_template_body(template, unknown_body, {"甲方": "隔离甲方"})
|
|
render_failed = False
|
|
except TemplateRenderError:
|
|
render_failed = True
|
|
check("publish_unknown_placeholder", "无法渲染的未声明变量不能通过发布校验", not validation.valid,
|
|
{"valid": validation.valid, "issues": [i.model_dump() for i in validation.issues], "render_failed": render_failed})
|
|
|
|
purchase = next(t for t in list_contract_templates().items if t.contract_type == "purchase")
|
|
baseline = template_sample_values(purchase.input_schema, [v.name for v in purchase.required_variables], [v.name for v in purchase.optional_variables], with_optional=False)
|
|
baseline_validation = validate_contract_template_inputs(purchase, baseline)
|
|
cases = []
|
|
for name, value, expected_code in (
|
|
("合同价款", "1元", "RULE_EQUAL_PRODUCT"),
|
|
("签署日期", "2026-02-30", "FIELD_TYPE"),
|
|
("争议解决方式", "提交某仲裁委员会仲裁，同时向某人民法院起诉", "FIELD_TYPE"),
|
|
):
|
|
try:
|
|
render_contract_template(purchase.id, {**baseline, name: value})
|
|
rejected, issue_codes = False, []
|
|
except TemplateRenderError as error:
|
|
issue_codes = [issue.code for issue in error.issues if name in issue.fields]
|
|
rejected = expected_code in issue_codes
|
|
cases.append({"field": name, "value": value, "rejected_for_expected_reason": rejected, "issue_codes": issue_codes})
|
|
check("template_semantic_rules_not_executed", "有效基准可生成；独立注入矛盾金额、无效日期和双选争议时，各自应准确拦截", baseline_validation.valid and all(case["rejected_for_expected_reason"] for case in cases),
|
|
{"valid_baseline": baseline_validation.valid, "independent_cases": cases})
|
|
purchase_output = render_contract_template(purchase.id, baseline)
|
|
residues = [marker for marker in ("不得默认生成", "不得自动生成", "必须由业务和法务确认") if marker in purchase_output]
|
|
check("template_instruction_residue", "给生成系统的操作说明不应进入合同正文", not residues, residues)
|
|
|
|
from sqlalchemy import create_engine, func, select
|
|
from sqlalchemy.orm import sessionmaker
|
|
from yuqei_ai_platform_api.models import PromptTemplateModel, PromptTemplateChangeModel
|
|
from yuqei_ai_platform_api.persistence.prompts import SqlAlchemyPromptTemplateRepository, PromptTemplateCreate, PromptTemplatePublishRequest
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
try:
|
|
PromptTemplateModel.__table__.create(engine)
|
|
PromptTemplateChangeModel.__table__.create(engine)
|
|
factory = sessionmaker(bind=engine, expire_on_commit=False)
|
|
repo = SqlAlchemyPromptTemplateRepository(factory)
|
|
repo.upsert_prompt_template(PromptTemplateCreate(name="audit", version="v1", content="已发布原文"))
|
|
repo.publish_prompt_template("audit", PromptTemplatePublishRequest(version="v1"))
|
|
try:
|
|
repo.upsert_prompt_template(PromptTemplateCreate(name="audit", version="v1", content="保存草稿时的新内容"))
|
|
except PromptVersionConflict:
|
|
pass # Rejection is expected; independently verify unchanged runtime below.
|
|
current = repo.get_prompt_template("audit")
|
|
with factory() as session:
|
|
history_count = session.scalar(select(func.count()).select_from(PromptTemplateChangeModel))
|
|
check("published_prompt_mutable_via_upsert", "保存同版本草稿不能直接更改正在运行的发布版本", current.content == "已发布原文",
|
|
{"current_content": current.content, "status": current.status, "publish_history_count": history_count})
|
|
try:
|
|
repo.upsert_prompt_template(PromptTemplateCreate(name="audit", version="v1", content="模拟前端保存草稿", enabled=False))
|
|
except PromptVersionConflict:
|
|
pass
|
|
current = repo.get_prompt_template("audit")
|
|
check("frontend_save_disables_published_prompt", "前端保存草稿不能下线同版本发布资产", current is not None,
|
|
{"runtime_prompt_missing": current is None})
|
|
finally:
|
|
engine.dispose()
|
|
return results
|
|
|
|
|
|
def runtime_inventory(tenant: str, sources: dict) -> dict:
|
|
from sqlalchemy import create_engine, text
|
|
from yuqei_ai_platform_api.config import AiPlatformSettings
|
|
from yuqei_contract_api.config import ContractApiSettings
|
|
source_hashes = {p["key"]: p["content_sha256"] for p in sources["prompts"]}
|
|
result = {"tenant_scope": tenant, "database_access": "PostgreSQL READ ONLY", "business_contracts_read": False}
|
|
settings = [("ai", AiPlatformSettings(_env_file=AI / ".env")), ("contract", ContractApiSettings(_env_file=CONTRACT / ".env"))]
|
|
for name, config in settings:
|
|
engine = create_engine(config.database_url, connect_args={"connect_timeout": 5})
|
|
try:
|
|
if engine.dialect.name != "postgresql":
|
|
raise ValueError("Runtime audit requires PostgreSQL read-only transactions")
|
|
with engine.connect() as connection:
|
|
connection.execute(text("SET TRANSACTION READ ONLY"))
|
|
connection.execute(text("SET LOCAL statement_timeout = '10000'"))
|
|
if name == "ai":
|
|
result["model_runtime_settings"] = {k: getattr(config, k) for k in
|
|
("provider_call_mode", "codex_cli_model", "provider_config_source", "model_job_dispatch_mode")}
|
|
prompts = []
|
|
for row in connection.execute(text("SELECT name, version, content, status, enabled FROM prompt_templates ORDER BY name, version")).mappings():
|
|
try:
|
|
payload = json.loads(row["content"])
|
|
except ValueError:
|
|
payload = {}
|
|
asset = payload.get("asset", {}) if isinstance(payload, dict) else {}
|
|
content = asset.get("prompt_content", row["content"])
|
|
prompts.append({"name": row["name"], "version": row["version"], "status": row["status"], "enabled": row["enabled"],
|
|
"format": "v1_asset" if asset else "plain_text", "content_length": len(row["content"]),
|
|
"content_sha256": sha(content), "matches_packaged_prompt_content": source_hashes.get(row["name"]) == sha(content),
|
|
"requires_legacy_review_schema": "risk_summary" in content and "rendered_report" in content,
|
|
"mentions_party_position": "party_position" in content})
|
|
result["prompts"] = prompts
|
|
result["providers"] = [dict(r) for r in connection.execute(text("SELECT provider, model, enabled, is_default FROM ai_provider_configs WHERE enabled = true")).mappings()]
|
|
result["trace_counts"] = [dict(r) for r in connection.execute(text("SELECT operation, count(*) AS count FROM ai_run_audits WHERE tenant_id = :tenant GROUP BY operation"), {"tenant": tenant}).mappings()]
|
|
result["trace_prompt_counts"] = [dict(r) for r in connection.execute(text("SELECT p.name, count(*) AS count FROM ai_run_audits a LEFT JOIN prompt_templates p ON p.id = a.prompt_template_id WHERE a.tenant_id = :tenant GROUP BY p.name ORDER BY count(*) DESC"), {"tenant": tenant}).mappings()]
|
|
else:
|
|
result["template_database_counts"] = {}
|
|
for table in ("template_lifecycle_events", "template_edit_drafts", "template_published_snapshots"):
|
|
result["template_database_counts"][table] = connection.scalar(text(f"SELECT count(*) FROM {table} WHERE tenant_id = :tenant"), {"tenant": tenant})
|
|
connection.rollback()
|
|
finally:
|
|
engine.dispose()
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--runtime", action="store_true", help="Read aggregate runtime evidence with READ ONLY transactions")
|
|
parser.add_argument("--tenant", help="Explicit tenant scope required with --runtime")
|
|
parser.add_argument("--output", type=Path, help="Write local JSON evidence; never writes to a service")
|
|
parser.add_argument("--strict", action="store_true", help="Exit 1 when a desired behavior probe fails")
|
|
args = parser.parse_args()
|
|
if args.runtime and not args.tenant:
|
|
parser.error("--runtime requires an explicit --tenant")
|
|
evidence = {"schema_version": "template-prompt-audit.v1", "created_at": datetime.now(UTC).isoformat(),
|
|
"model_calls": 0, "live_writes": 0, "sources": source_inventory(), "probes": isolated_probes()}
|
|
if args.runtime:
|
|
evidence["runtime"] = runtime_inventory(args.tenant, evidence["sources"])
|
|
evidence["probe_summary"] = dict(Counter(p["status"] for p in evidence["probes"]))
|
|
if args.output:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(evidence, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
runtime = evidence.get("runtime", {})
|
|
print(json.dumps({"template_count": evidence["sources"]["template_count"], "prompt_count": evidence["sources"]["prompt_count"],
|
|
"probe_summary": evidence["probe_summary"], "failed_probes": [p for p in evidence["probes"] if p["status"] == "fail"],
|
|
"runtime_prompt_count": len(runtime.get("prompts", [])),
|
|
"model_runtime_settings": runtime.get("model_runtime_settings"),
|
|
"template_database_counts": runtime.get("template_database_counts")}, ensure_ascii=False, indent=2))
|
|
return 1 if args.strict and evidence["probe_summary"].get("fail") else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|