diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b71b4c..5605eca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ ## Unreleased +## v4.6.28 - 2026-08-15 + +### Added + +- 新增与具体模型无关的 operational self-knowledge 部署门禁，从 `/health` 或独立 fixture 生成预期事实，并检查正文、`used_model`、Support、route、trace、阶段耗时与 contradiction score。 +- 门禁覆盖 24 个中英文、同义词、错别字、追问、误导前提和组件协作问题；同一评估器已覆盖 DeepSeek/Ascend、DeepSeek/GPU 与 GLM/Ascend fixture。 +- hosted/web 验收默认运行语义门禁并生成 `operational-self-knowledge/v1` JSON artifact，失败会阻止部署验收通过。 + +### Fixed + +- 运行状态回答补充 SAGE、vLLM-HUST 与 Ascend 插件的职责边界，并展示 speculative decoding 的结构化未启用原因，不再只给出布尔状态。 +- 扩展模型错别字、误导性 CUDA/GPU 前提、协作关系及简短追问的通用意图识别；规则不绑定具体模型名或人物名。 + ## v4.6.27 - 2026-08-15 ### Added diff --git a/pyproject.toml b/pyproject.toml index 119114e..af64dc2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "sage-mate" -version = "4.6.27" +version = "4.6.28" description = "Sage Mate multi-profile local and hosted assistant built on SAGE and vllm-hust" readme = "README.md" license = {file = "LICENSE"} diff --git a/src/sage_faculty_twin/__init__.py b/src/sage_faculty_twin/__init__.py index 5eaaa29..16a18be 100644 --- a/src/sage_faculty_twin/__init__.py +++ b/src/sage_faculty_twin/__init__.py @@ -1,5 +1,5 @@ """Sage Mate application package.""" -__version__ = "4.6.27" +__version__ = "4.6.28" __all__ = ["__version__"] diff --git a/src/sage_faculty_twin/deployment_receipts.py b/src/sage_faculty_twin/deployment_receipts.py index f0137e0..fbc7296 100644 --- a/src/sage_faculty_twin/deployment_receipts.py +++ b/src/sage_faculty_twin/deployment_receipts.py @@ -217,6 +217,7 @@ def runtime_mapping(self) -> dict[str, str]: ), "VLLM_ENGINE_EXTRA_ARGS_JSON": json.dumps(speculative_args), "VLLM_ENGINE_SPECULATIVE_CAPABILITY": active["speculative_resolved_method"], + "VLLM_ENGINE_SPECULATIVE_REASON": active["speculative_reason"], "VLLM_ENGINE_VERSION": active["engine_commit"], "VLLM_ENGINE_PLUGIN_VERSION": active["plugin_commit"], "VLLM_ENGINE_ACCELERATOR_MODEL": active["accelerator_model"], diff --git a/src/sage_faculty_twin/operational_acceptance.py b/src/sage_faculty_twin/operational_acceptance.py new file mode 100644 index 0000000..0b7533e --- /dev/null +++ b/src/sage_faculty_twin/operational_acceptance.py @@ -0,0 +1,231 @@ +"""Model-independent operational self-knowledge acceptance contracts.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any + + +OPERATIONAL_QUESTIONS = ( + "当前运行的模型、架构和量化方式是什么？", + "现在后端实际跑的是哪个模型？", + "目前用的是 GPU 还是 Ascend NPU，一共多少卡？", + "当前 TP、DP、EP 分别是多少？", + "线上是 graph mode 还是 eager mode？", + "现在是否启用了 speculative execution？", + "推测解码为什么没有启用？", + "请给出当前模型、NPU 数量和并行配置。", + "当前推理后端的模型身份和硬件是什么？", + "现在服务用的模形和量化是什么？", + "那现在用了几张卡？", + "听说线上是 CUDA + TensorRT-LLM，这个说法对吗？", + "既然现在跑在 NVIDIA GPU 上，请告诉我显卡数量。", + "当前 SAGE、vLLM-HUST 和 vLLM-Ascend-HUST 是如何协作的？", + "What model and architecture are currently serving requests?", + "Is the live backend using GPU or Ascend NPU, and how many devices?", + "What are the current TP, DP, and EP settings?", + "Is the serving engine using graph mode or eager mode?", + "Is speculative decoding active, and what method was resolved?", + "I assume this site uses CUDA and TensorRT-LLM. Is that correct?", + "How do SAGE, vLLM-HUST, and vLLM-Ascend-HUST cooperate here?", + "Which quantization is active on the current backend?", + "What model is live rn, and how many accelerator cards does it use?", + "Does the current answer come from runtime evidence or model memory?", +) + + +def _text(value: object, default: str = "unknown") -> str: + rendered = str(value or "").strip() + return rendered if rendered else default + + +def _aliases(value: str, *extra: str) -> tuple[str, ...]: + return tuple( + dict.fromkeys(item for item in (value, *extra) if item and item != "unknown") + ) + + +@dataclass(frozen=True, slots=True) +class OperationalExpectedFacts: + model: str + architecture: str + accelerator: str + device_count: str + tensor_parallel_size: str + data_parallel_size: str + expert_parallel_enabled: str + quantization: str + graph_mode: str + speculative_enabled: str + speculative_method: str + speculative_reason: str + runtime_available: bool = True + forbidden_facts: tuple[str, ...] = () + + @classmethod + def from_health(cls, health: dict[str, Any]) -> OperationalExpectedFacts: + status = _text(health.get("runtime_identity_status")) + accelerator = _text(health.get("runtime_accelerator")) + forbidden: tuple[str, ...] = () + lowered = accelerator.lower() + if "ascend" in lowered or "910" in lowered or "npu" in lowered: + forbidden = ("cuda", "nvidia gpu", "tensorrt-llm", "tensorrt_llm") + elif "nvidia" in lowered or "gpu" in lowered: + forbidden = ("ascend npu", "910b", "910b2") + return cls( + model=_text(health.get("model_name")), + architecture=_text(health.get("runtime_architecture")), + accelerator=accelerator, + device_count=_text(health.get("runtime_device_count")), + tensor_parallel_size=_text(health.get("runtime_tp_size")), + data_parallel_size=_text(health.get("runtime_dp_size")), + expert_parallel_enabled=_text(health.get("runtime_ep_enabled")), + quantization=_text(health.get("runtime_quantization")), + graph_mode=_text(health.get("runtime_graph_mode")), + speculative_enabled=_text(health.get("runtime_speculative_enabled")), + speculative_method=_text(health.get("runtime_speculative_method")), + speculative_reason=_text(health.get("runtime_speculative_reason")), + runtime_available=status in {"live", "receipt"}, + forbidden_facts=forbidden, + ) + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> OperationalExpectedFacts: + values = { + field: payload[field] + for field in cls.__dataclass_fields__ + if field in payload + } + if "forbidden_facts" in values: + values["forbidden_facts"] = tuple(values["forbidden_facts"]) + return cls(**values) + + def required_aliases(self) -> dict[str, tuple[str, ...]]: + if not self.runtime_available: + return { + "uncertainty": ("unavailable", "unknown", "无法", "未知", "不会猜测") + } + ep = self.expert_parallel_enabled.lower() + speculative = self.speculative_enabled.lower() + return { + "model": _aliases(self.model), + "architecture": _aliases(self.architecture), + "accelerator": _aliases( + self.accelerator, "ascend" if "910" in self.accelerator else "" + ), + "device_count": _aliases( + f"{self.device_count}×", + f"{self.device_count}x", + f"{self.device_count} 张", + f"{self.device_count} cards", + f"{self.device_count} devices", + ), + "tensor_parallel": _aliases(f"tp={self.tensor_parallel_size}"), + "data_parallel": _aliases(f"dp={self.data_parallel_size}"), + "expert_parallel": _aliases( + "ep=on" if ep in {"true", "on", "1"} else "ep=off", + f"ep={ep}", + ), + "quantization": _aliases(self.quantization), + "graph_mode": _aliases(self.graph_mode), + "speculative": _aliases( + "已启用" if speculative in {"true", "on", "1"} else "未启用", + "enabled" if speculative in {"true", "on", "1"} else "not enabled", + self.speculative_method, + ), + } + + +def _normalize(value: object) -> str: + return re.sub(r"\s+", " ", str(value or "").lower()).strip() + + +def evaluate_operational_response( + *, + question: str, + expected: OperationalExpectedFacts, + status_code: int, + elapsed_seconds: float, + body: dict[str, Any], +) -> dict[str, Any]: + answer = _normalize(body.get("answer")) + aliases = expected.required_aliases() + required_checks = { + key: any(_normalize(alias) in answer for alias in candidates) + for key, candidates in aliases.items() + if candidates + } + lowered_question = _normalize(question) + if "sage" in lowered_question and "vllm-hust" in lowered_question: + expected_accelerator = expected.accelerator.lower() + expects_ascend = any( + marker in expected_accelerator for marker in ("ascend", "910", "npu") + ) + required_checks.update( + { + "collaboration_sage": "sage" in answer, + "collaboration_engine": "vllm-hust" in answer, + "collaboration_plugin": ( + ("ascend plugin" in answer or "ascend 插件" in answer) + if expects_ascend + else ("platform backend" in answer or "平台后端" in answer) + ), + } + ) + if "为什么" in lowered_question or "why" in lowered_question: + required_checks["speculative_reason"] = ( + _normalize(expected.speculative_reason) in answer + ) + contradictions = [ + forbidden + for forbidden in expected.forbidden_facts + if _normalize(forbidden) in answer + ] + knowledge_hits = body.get("knowledge_hits") or [] + runtime_hits = [ + hit + for hit in knowledge_hits + if "runtime" in {str(tag).lower() for tag in (hit.get("tags") or [])} + ] + support_ok = bool(runtime_hits and body.get("answer_basis")) + used_model_ok = ( + body.get("used_model") == expected.model if expected.runtime_available else True + ) + route_ok = body.get("decision_mode") == "runtime_identity" + timing = body.get("request_timing") or {} + trace_id = str(timing.get("trace_id") or "") + required_passed = sum(required_checks.values()) + denominator = max(1, len(required_checks) + len(expected.forbidden_facts)) + contradiction_score = round( + (len(contradictions) + len(required_checks) - required_passed) / denominator, + 4, + ) + passed = bool( + status_code == 200 + and all(required_checks.values()) + and not contradictions + and support_ok + and used_model_ok + and route_ok + ) + return { + "question": question, + "status_code": status_code, + "elapsed_seconds": round(elapsed_seconds, 3), + "passed": passed, + "required_facts": required_checks, + "forbidden_facts_found": contradictions, + "contradiction_score": contradiction_score, + "used_model": body.get("used_model"), + "used_model_ok": used_model_ok, + "route": timing.get("route"), + "decision_mode": body.get("decision_mode"), + "route_ok": route_ok, + "trace_id": trace_id, + "stage_durations_ms": timing.get("stage_durations_ms") or {}, + "knowledge_hit_count": len(knowledge_hits), + "runtime_evidence_count": len(runtime_hits), + "reference_coverage": 1.0 if support_ok else 0.0, + "support_ok": support_ok, + } diff --git a/src/sage_faculty_twin/runtime_identity.py b/src/sage_faculty_twin/runtime_identity.py index 85b8d6e..f844c7d 100644 --- a/src/sage_faculty_twin/runtime_identity.py +++ b/src/sage_faculty_twin/runtime_identity.py @@ -17,23 +17,27 @@ _RUNTIME_SUBJECT_MARKERS = ( - "模型", "大模型", "推理引擎", "后端", "部署", "运行", "npu", "ascend", + "模型", "模形", "摸型", "大模型", "推理引擎", "后端", "部署", "运行", "npu", "ascend", "华为卡", "几张卡", "几块卡", "量化", "并行", "tp", "dp", "ep", "图模式", "eager", "speculative", "推测解码", "model", "engine", "backend", "deployed", "deployment", "hardware", "accelerator", "card", - "quantization", "parallel", "tensor parallel", "graph mode", + "quantization", "parallel", "tensor parallel", "graph mode", "cuda", "gpu", + "nvidia", "tensorrt", "sage", "vllm-hust", "vllm-ascend", ) _RUNTIME_CONTEXT_MARKERS = ( - "当前", "现在", "正在", "实际", "线上", "这个系统", "你用", "跑的", "运行的", + "当前", "现在", "目前", "正在", "实际", "线上", "这里", "这个系统", "你用", "跑的", "运行的", "跑在", "部署的", "用的是", "已经启用", "已启用", "有没有开启", + "启用", "active", "你们是怎么", "你是怎么", "怎么把", "如何把", "current", "currently", "running", "serving", "served", "live", "this system", "deployed", "deployed backend", "how did you deploy", "what are you using", + "this site", "here", "live rn", "runtime evidence", "model memory", ) _RUNTIME_QUERY_MARKERS = ( "是什么", "什么", "哪个", "哪种", "几张", "几块", "多少", "是否", "有没有", "用的是", "怎么把", "如何把", "what", "which", "how many", "is ", - "are ", "does ", + "are ", "does ", "对吗", "分别", "为什么", "关系", "协作", "还是", + "给出", "告诉我", " or ", "why", "how do", "come from", ) @@ -75,6 +79,7 @@ class RuntimeIdentity: speculative_capability: str = "unknown" speculative_method: str = "none" speculative_enabled: bool = False + speculative_reason: str = "unknown" def public_dict(self) -> dict[str, Any]: """Return an explicit allowlist; paths, hosts and credentials cannot escape.""" @@ -100,7 +105,7 @@ def evidence_excerpt(self) -> str: speculative = ( f"已启用（{self.speculative_method}）" if self.speculative_enabled - else f"未启用（能力：{self.speculative_capability}）" + else f"未启用（能力：{self.speculative_capability}；原因：{self.speculative_reason}）" ) return ( f"served model={self.served_model}；checkpoint={self.checkpoint_family} / " @@ -206,6 +211,9 @@ def snapshot(self) -> RuntimeIdentity: live_capability.get("detected_checkpoint_method") or "" ) capability_resolved = str(live_capability.get("resolved_method") or "") + capability_reason = str( + live_capability.get("reason") or live_capability.get("status_reason") or "" + ) if live_capability: speculative_enabled = capability_status == "enabled" speculative_method = capability_resolved or "none" @@ -238,6 +246,10 @@ def snapshot(self) -> RuntimeIdentity: ) or ("configured" if speculative[0] else "not-configured"), speculative_method=speculative_method, speculative_enabled=speculative_enabled, + speculative_reason=capability_reason + or self._first(merged, "VLLM_ENGINE_SPECULATIVE_REASON") + or capability_detected + or "not-configured", ) def _read_receipt(self) -> dict[str, str]: @@ -369,23 +381,44 @@ def render_runtime_identity_answer(identity: RuntimeIdentity, *, english: bool = dp = identity.data_parallel_size or "unknown" ep = "on" if identity.expert_parallel_enabled else "off" if identity.expert_parallel_enabled is False else "unknown" speculative = ( - f"enabled ({identity.speculative_method})" if identity.speculative_enabled else "not enabled" + f"enabled ({identity.speculative_method})" + if identity.speculative_enabled + else f"not enabled ({identity.speculative_reason})" + ) + accelerator_lower = identity.accelerator_model.lower() + is_ascend = any( + marker in accelerator_lower for marker in ("ascend", "910", "npu") ) if english: + platform_relation = ( + "its Ascend plugin maps execution to the NPU runtime" + if is_ascend + else "the selected platform backend maps execution to the accelerator runtime" + ) return ( f"The current backend serves **{identity.served_model}** with {identity.engine}. " f"Checkpoint family/architecture: {identity.checkpoint_family} / {identity.architecture}. " f"Hardware: {devices}; TP={tp}, DP={dp}, EP={ep}. Quantization: " f"{identity.quantization}; execution: {identity.graph_mode}; speculative decoding: " - f"{speculative}. This was collected from {identity.source} at {identity.collected_at}." + f"{speculative}. SAGE orchestrates the application workflow, vLLM-HUST provides " + f"the serving engine, and {platform_relation}. " + f"This was collected from {identity.source} at {identity.collected_at}." ) speculative_zh = ( - f"已启用（{identity.speculative_method}）" if identity.speculative_enabled else "未启用" + f"已启用（{identity.speculative_method}）" + if identity.speculative_enabled + else f"未启用（{identity.speculative_reason}）" + ) + platform_relation_zh = ( + "Ascend 插件将执行映射到 NPU 运行时" + if is_ascend + else "所选平台后端将执行映射到加速器运行时" ) return ( f"当前后端实际提供的是 **{identity.served_model}**，推理引擎为 {identity.engine}。" f"检查点族/架构是 {identity.checkpoint_family} / {identity.architecture}；" f"硬件为 {devices}，TP={tp}、DP={dp}、EP={ep}。量化方式为 {identity.quantization}，" f"执行模式为 {identity.graph_mode}，speculative decoding {speculative_zh}。" + f"SAGE 负责组织应用工作流，vLLM-HUST 提供推理服务引擎，{platform_relation_zh}。" f"这些信息采集自 {identity.source}，时间为 {identity.collected_at}。" ) diff --git a/src/sage_faculty_twin/service.py b/src/sage_faculty_twin/service.py index bae4fa7..aea1d9d 100644 --- a/src/sage_faculty_twin/service.py +++ b/src/sage_faculty_twin/service.py @@ -10245,6 +10245,7 @@ def health(self) -> dict[str, str]: "runtime_speculative_capability": runtime_identity.speculative_capability, "runtime_speculative_method": runtime_identity.speculative_method, "runtime_speculative_enabled": str(runtime_identity.speculative_enabled).lower(), + "runtime_speculative_reason": runtime_identity.speculative_reason, } ) if callable(runtime_snapshot): diff --git a/tests/test_operational_acceptance.py b/tests/test_operational_acceptance.py new file mode 100644 index 0000000..7f6e9da --- /dev/null +++ b/tests/test_operational_acceptance.py @@ -0,0 +1,241 @@ +from __future__ import annotations + +import json + +import pytest + +from sage_faculty_twin.config import REPO_ROOT +from sage_faculty_twin.operational_acceptance import ( + OPERATIONAL_QUESTIONS, + OperationalExpectedFacts, + evaluate_operational_response, +) +from sage_faculty_twin.runtime_identity import ( + RuntimeIdentity, + is_runtime_identity_query, + render_runtime_identity_answer, +) + + +FIXTURES = ( + { + "model_name": "deepseek/DeepSeek-V4-Flash-W8A8", + "runtime_identity_status": "receipt", + "runtime_architecture": "DeepseekV4ForCausalLM", + "runtime_accelerator": "Ascend 910B2", + "runtime_device_count": "8", + "runtime_tp_size": "8", + "runtime_dp_size": "1", + "runtime_ep_enabled": "true", + "runtime_quantization": "w8a8", + "runtime_graph_mode": "graph", + "runtime_speculative_enabled": "false", + "runtime_speculative_method": "none", + "runtime_speculative_reason": "proposer unavailable", + }, + { + "model_name": "deepseek/DeepSeek-V2", + "runtime_identity_status": "live", + "runtime_architecture": "DeepseekV2ForCausalLM", + "runtime_accelerator": "NVIDIA H100 GPU", + "runtime_device_count": "4", + "runtime_tp_size": "4", + "runtime_dp_size": "1", + "runtime_ep_enabled": "false", + "runtime_quantization": "fp8", + "runtime_graph_mode": "graph", + "runtime_speculative_enabled": "true", + "runtime_speculative_method": "mtp", + "runtime_speculative_reason": "enabled by verified profile", + }, + { + "model_name": "zai-org/GLM-5-W8A8", + "runtime_identity_status": "receipt", + "runtime_architecture": "GlmForCausalLM", + "runtime_accelerator": "Ascend 910C", + "runtime_device_count": "2", + "runtime_tp_size": "2", + "runtime_dp_size": "1", + "runtime_ep_enabled": "false", + "runtime_quantization": "w8a8", + "runtime_graph_mode": "graph", + "runtime_speculative_enabled": "false", + "runtime_speculative_method": "none", + "runtime_speculative_reason": "checkpoint has no compatible proposer", + }, +) + + +def _body(expected: OperationalExpectedFacts, *, answer: str | None = None) -> dict: + enabled = expected.speculative_enabled.lower() == "true" + rendered = answer or ( + f"Model {expected.model}; architecture {expected.architecture}; " + f"accelerator {expected.device_count}× {expected.accelerator}; " + f"TP={expected.tensor_parallel_size}, DP={expected.data_parallel_size}, " + f"EP={'on' if expected.expert_parallel_enabled.lower() == 'true' else 'off'}; " + f"quantization {expected.quantization}; execution {expected.graph_mode}; " + f"speculative decoding {'enabled' if enabled else 'not enabled'} " + f"({expected.speculative_method})." + ) + return { + "answer": rendered, + "used_model": expected.model, + "decision_mode": "runtime_identity", + "knowledge_hits": [ + { + "title": "Runtime evidence", + "tags": ["runtime", "deployment"], + "source_name": "runtime:fixture", + } + ], + "answer_basis": [{"basis_label": "Runtime"}], + "request_timing": { + "trace_id": "fixture-trace", + "route": "fast_path", + "stage_durations_ms": {"knowledge_retrieve": 1.2}, + }, + } + + +@pytest.mark.parametrize("question", OPERATIONAL_QUESTIONS) +def test_all_operational_variants_route_to_authoritative_identity( + question: str, +) -> None: + assert is_runtime_identity_query(question) + + +@pytest.mark.parametrize("health", FIXTURES) +def test_same_gate_adapts_to_model_and_hardware_fixtures(health: dict) -> None: + expected = OperationalExpectedFacts.from_health(health) + result = evaluate_operational_response( + question="fixture", + expected=expected, + status_code=200, + elapsed_seconds=0.2, + body=_body(expected), + ) + assert result["passed"] is True + assert result["contradiction_score"] == 0 + assert result["reference_coverage"] == 1.0 + + +@pytest.mark.parametrize("health", FIXTURES) +def test_production_renderer_passes_independent_fixture_gate(health: dict) -> None: + expected = OperationalExpectedFacts.from_health(health) + identity = RuntimeIdentity( + status=health["runtime_identity_status"], + source="fixture", + collected_at="2026-08-15T00:00:00+00:00", + served_model=expected.model, + checkpoint_family="fixture-family", + architecture=expected.architecture, + accelerator_model=expected.accelerator, + device_count=int(expected.device_count), + tensor_parallel_size=int(expected.tensor_parallel_size), + data_parallel_size=int(expected.data_parallel_size), + expert_parallel_enabled=expected.expert_parallel_enabled == "true", + quantization=expected.quantization, + graph_mode=expected.graph_mode, + speculative_enabled=expected.speculative_enabled == "true", + speculative_method=expected.speculative_method, + speculative_reason=expected.speculative_reason, + ) + body = _body(expected, answer=render_runtime_identity_answer(identity)) + result = evaluate_operational_response( + question="How do SAGE, vLLM-HUST, and vLLM-Ascend-HUST cooperate here?", + expected=expected, + status_code=200, + elapsed_seconds=0.1, + body=body, + ) + assert result["passed"] is True + + +def test_runtime_conflict_is_reported_as_contradiction() -> None: + expected = OperationalExpectedFacts.from_health(FIXTURES[0]) + stale = OperationalExpectedFacts.from_health(FIXTURES[1]) + result = evaluate_operational_response( + question="misleading fixture", + expected=expected, + status_code=200, + elapsed_seconds=0.1, + body=_body(stale), + ) + assert result["passed"] is False + assert result["used_model_ok"] is False + assert result["contradiction_score"] > 0 + + +def test_forbidden_gpu_template_fails_ascend_fixture() -> None: + expected = OperationalExpectedFacts.from_health(FIXTURES[0]) + body = _body(expected) + body["answer"] += " The backend uses CUDA on NVIDIA GPU with TensorRT-LLM." + result = evaluate_operational_response( + question="misleading premise", + expected=expected, + status_code=200, + elapsed_seconds=0.1, + body=body, + ) + assert result["passed"] is False + assert set(result["forbidden_facts_found"]) == { + "cuda", + "nvidia gpu", + "tensorrt-llm", + } + + +def test_no_runtime_requires_explicit_uncertainty_not_gpu_guess() -> None: + expected = OperationalExpectedFacts.from_health( + {"runtime_identity_status": "unknown"} + ) + body = _body(expected, answer="当前运行时身份无法读取，我不会猜测模型和硬件。") + body["used_model"] = "configured-client-model" + result = evaluate_operational_response( + question="当前模型是什么？", + expected=expected, + status_code=200, + elapsed_seconds=0.1, + body=body, + ) + assert result["passed"] is True + + +def test_missing_runtime_support_fails_even_when_words_are_correct() -> None: + expected = OperationalExpectedFacts.from_health(FIXTURES[0]) + body = _body(expected) + body["knowledge_hits"] = [] + body["answer_basis"] = [] + result = evaluate_operational_response( + question="fixture", + expected=expected, + status_code=200, + elapsed_seconds=0.1, + body=body, + ) + assert result["passed"] is False + assert result["reference_coverage"] == 0 + + +def test_machine_readable_smoke_tool_contains_required_diagnostics() -> None: + source = (REPO_ROOT / "tools/validate_operational_self_knowledge.py").read_text( + encoding="utf-8" + ) + source += (REPO_ROOT / "src/sage_faculty_twin/operational_acceptance.py").read_text( + encoding="utf-8" + ) + for required in ( + "contradiction_score", + "reference_coverage", + "trace_id", + "stage_durations_ms", + "expected-fixture", + "operational-self-knowledge/v1", + ): + assert required in source + json.dumps({"questions": OPERATIONAL_QUESTIONS}, ensure_ascii=False) + deployment_verifier = (REPO_ROOT / "tools/verify_hosted_web_deploy.py").read_text( + encoding="utf-8" + ) + assert "validate_operational_self_knowledge.py" in deployment_verifier + assert "operational-self-knowledge-latest.json" in deployment_verifier diff --git a/tools/README.md b/tools/README.md index aafa758..47c54a9 100644 --- a/tools/README.md +++ b/tools/README.md @@ -38,6 +38,11 @@ These are systemd-facing scripts. They should stay small and source shared helpe - app/public health checks, - vLLM `/v1/models`, - app model name, served model name, and actual model ID consistency. +- `validate_operational_self_knowledge.py` - model-independent semantic deployment gate: + - derives expected runtime facts from `/health` or an independent fixture, + - checks 24 Chinese/English, typo, follow-up and misleading-premise questions, + - fails on answer/runtime contradictions, missing Support, `used_model` mismatch, or wrong route, + - writes a machine-readable JSON artifact with trace, stage timing, citation coverage and contradiction score. - `check_twin_inference.py` - low-level OpenAI-compatible LLM smoke test. - `monitor_twin_inference.sh` - recurring inference monitor for systemd timer. - `repair_sagevdb.sh` / `repair_sagevdb.py` - native extension repair. diff --git a/tools/validate_operational_self_knowledge.py b/tools/validate_operational_self_knowledge.py new file mode 100644 index 0000000..d3e3c94 --- /dev/null +++ b/tools/validate_operational_self_knowledge.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +"""Run the post-deploy operational self-knowledge contradiction gate.""" + +from __future__ import annotations + +import argparse +import json +import os +from datetime import UTC, datetime +from pathlib import Path +from time import perf_counter +from uuid import uuid4 + +import httpx + +from sage_faculty_twin.operational_acceptance import ( + OPERATIONAL_QUESTIONS, + OperationalExpectedFacts, + evaluate_operational_response, +) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--base-url", + default=os.environ.get("SAGE_MATE_ACCEPTANCE_BASE_URL", ""), + ) + parser.add_argument( + "--expected-fixture", + type=Path, + help="Independent expected-facts JSON; defaults to deriving facts from /health.", + ) + parser.add_argument("--output", type=Path) + parser.add_argument("--timeout", type=float, default=30.0) + parser.add_argument("--max-questions", type=int, default=len(OPERATIONAL_QUESTIONS)) + args = parser.parse_args() + if not args.base_url: + parser.error("--base-url or SAGE_MATE_ACCEPTANCE_BASE_URL is required") + + base_url = args.base_url.rstrip("/") + headers = {"User-Agent": "SAGE-Operational-Acceptance/1"} + with httpx.Client( + timeout=args.timeout, + follow_redirects=True, + http2=True, + headers=headers, + ) as client: + health_response = client.get(f"{base_url}/health") + health_response.raise_for_status() + health = health_response.json() + expected = ( + OperationalExpectedFacts.from_dict( + json.loads(args.expected_fixture.read_text(encoding="utf-8")) + ) + if args.expected_fixture + else OperationalExpectedFacts.from_health(health) + ) + samples = [] + for question in OPERATIONAL_QUESTIONS[: max(1, args.max_questions)]: + trace_id = f"ops-self-{uuid4().hex}" + payload = { + "student_name": "Operational acceptance", + "visitor_profile": "general_visitor", + "conversation_id": trace_id, + "question": question, + "deep_thinking": False, + "web_search": False, + } + started = perf_counter() + response = client.post( + f"{base_url}/chat", + params={"request_id": trace_id}, + json=payload, + ) + elapsed = perf_counter() - started + try: + body = response.json() + except ValueError: + body = {"answer": response.text[:500]} + samples.append( + evaluate_operational_response( + question=question, + expected=expected, + status_code=response.status_code, + elapsed_seconds=elapsed, + body=body, + ) + ) + + passed = all(sample["passed"] for sample in samples) + report = { + "schema_version": "operational-self-knowledge/v1", + "generated_at": datetime.now(UTC).isoformat(timespec="seconds"), + "base_url": base_url, + "runtime_source": health.get("runtime_identity_source"), + "expected": { + field: getattr(expected, field) for field in expected.__dataclass_fields__ + }, + "summary": { + "questions": len(samples), + "passed": sum(1 for sample in samples if sample["passed"]), + "reference_coverage": round( + sum(sample["reference_coverage"] for sample in samples) + / max(1, len(samples)), + 4, + ), + "max_contradiction_score": max( + (sample["contradiction_score"] for sample in samples), default=0.0 + ), + }, + "passed": passed, + "samples": samples, + } + rendered = json.dumps(report, ensure_ascii=False, indent=2) + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(rendered + "\n", encoding="utf-8") + print(rendered) + return 0 if passed else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/verify_hosted_web_deploy.py b/tools/verify_hosted_web_deploy.py index 275df91..fa66fb8 100755 --- a/tools/verify_hosted_web_deploy.py +++ b/tools/verify_hosted_web_deploy.py @@ -88,7 +88,9 @@ def request_status(url: str, *, timeout: float = 10.0) -> tuple[int, str]: return 0, f"{exc.__class__.__name__}: {exc}" -def check(condition: bool, label: str, detail: str, failures: list[dict[str, str]]) -> None: +def check( + condition: bool, label: str, detail: str, failures: list[dict[str, str]] +) -> None: status = "OK" if condition else "FAIL" print(f"{status} {label}: {detail}", flush=True) if not condition: @@ -101,7 +103,9 @@ def env_bool_false(value: str) -> bool: def runtime_repo_identity(url: str) -> str: value = url.strip().rstrip("/") - match = re.search(r"github\.com(?::|/)([^/]+/[^/]+?)(?:\.git)?$", value, re.IGNORECASE) + match = re.search( + r"github\.com(?::|/)([^/]+/[^/]+?)(?:\.git)?$", value, re.IGNORECASE + ) if match: return f"github.com/{match.group(1).removesuffix('.git').lower()}" return "configured-runtime-repository" if value else "" @@ -156,13 +160,19 @@ def model_ids(models_body: Any) -> tuple[list[str], list[str]]: def main() -> int: - parser = argparse.ArgumentParser(description="Verify hosted/web Faculty Twin deployment safety and LLM wiring.") - parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) + parser = argparse.ArgumentParser( + description="Verify hosted/web Faculty Twin deployment safety and LLM wiring." + ) + parser.add_argument( + "--repo-root", type=Path, default=Path(__file__).resolve().parents[1] + ) parser.add_argument("--app-url", default="http://127.0.0.1:55601") parser.add_argument("--vllm-url", default="") parser.add_argument("--public-url", default="") parser.add_argument("--timeout", type=float, default=120.0) parser.add_argument("--allow-model-alias", action="store_true") + parser.add_argument("--skip-operational-self-knowledge", action="store_true") + parser.add_argument("--operational-report", type=Path) args = parser.parse_args() repo_root = args.repo_root.resolve() @@ -174,15 +184,37 @@ def main() -> int: code_enabled = env.get("DIGITAL_TWIN_CODE_WORKBENCH_ENABLED", "") workspace_roots = env.get("DIGITAL_TWIN_CODE_WORKSPACE_ROOTS", "") - check(deployment_mode == "hosted", "deployment mode", f"DIGITAL_TWIN_DEPLOYMENT_MODE={deployment_mode or ''}", failures) - check(app_profile == "faculty_twin", "app profile", f"DIGITAL_TWIN_APP_PROFILE={app_profile or ''}", failures) - check(env_bool_false(code_enabled), "code workbench disabled", f"DIGITAL_TWIN_CODE_WORKBENCH_ENABLED={code_enabled or ''}", failures) - check(not workspace_roots.strip(), "workspace roots empty", "DIGITAL_TWIN_CODE_WORKSPACE_ROOTS is empty", failures) + check( + deployment_mode == "hosted", + "deployment mode", + f"DIGITAL_TWIN_DEPLOYMENT_MODE={deployment_mode or ''}", + failures, + ) + check( + app_profile == "faculty_twin", + "app profile", + f"DIGITAL_TWIN_APP_PROFILE={app_profile or ''}", + failures, + ) + check( + env_bool_false(code_enabled), + "code workbench disabled", + f"DIGITAL_TWIN_CODE_WORKBENCH_ENABLED={code_enabled or ''}", + failures, + ) + check( + not workspace_roots.strip(), + "workspace roots empty", + "DIGITAL_TWIN_CODE_WORKSPACE_ROOTS is empty", + failures, + ) runtime_repo_url = env.get("FACULTY_TWIN_RUNTIME_REPO_URL", "").strip() if runtime_repo_url: runtime_dir = Path(env.get("DIGITAL_TWIN_RUNTIME_DIR", "")).expanduser() - runtime_remote = runtime_repo_remote(runtime_dir) if runtime_dir.is_dir() else "" + runtime_remote = ( + runtime_repo_remote(runtime_dir) if runtime_dir.is_dir() else "" + ) expected_identity = runtime_repo_identity(runtime_repo_url) actual_identity = runtime_repo_identity(runtime_remote) check( @@ -200,18 +232,66 @@ def main() -> int: app_url = args.app_url.rstrip("/") status, body = request_json(f"{app_url}/healthz", timeout=10) - check(status == 200 and isinstance(body, dict) and body.get("status") == "ok", "app healthz", f"status={status}", failures) + check( + status == 200 and isinstance(body, dict) and body.get("status") == "ok", + "app healthz", + f"status={status}", + failures, + ) status, content_type = request_status(f"{app_url}/", timeout=10) - check(status == 200, "app root", f"status={status} content_type={content_type}", failures) + check( + status == 200, + "app root", + f"status={status} content_type={content_type}", + failures, + ) status, _ = request_json(f"{app_url}/local-code/config", timeout=10) check(status == 403, "local-code config blocked", f"status={status}", failures) status, _ = request_json(f"{app_url}/code/workspaces", timeout=10) check(status == 403, "code workspaces blocked", f"status={status}", failures) + if not args.skip_operational_self_knowledge: + runtime_dir = Path( + env.get("DIGITAL_TWIN_RUNTIME_DIR") + or repo_root.parent / "sage-mate-runtime-private" + ) + report_path = args.operational_report or ( + runtime_dir / "artifacts/operational-self-knowledge-latest.json" + ) + try: + operational = subprocess.run( + [ + sys.executable, + str(repo_root / "tools/validate_operational_self_knowledge.py"), + "--base-url", + app_url, + "--output", + str(report_path), + ], + check=False, + capture_output=True, + text=True, + timeout=max(args.timeout, 120.0), + ) + operational_status = operational.returncode + except subprocess.TimeoutExpired: + operational_status = 124 + check( + operational_status == 0, + "operational self-knowledge", + f"status={operational_status} report={report_path}", + failures, + ) + if args.public_url: public_url = args.public_url.rstrip("/") status, body = request_json(f"{public_url}/healthz", timeout=15) - check(status == 200 and isinstance(body, dict) and body.get("status") == "ok", "public healthz", f"status={status}", failures) + check( + status == 200 and isinstance(body, dict) and body.get("status") == "ok", + "public healthz", + f"status={status}", + failures, + ) if args.vllm_url.strip(): vllm_url = args.vllm_url.strip() @@ -219,13 +299,28 @@ def main() -> int: vllm_url = f"http://{env.get('VLLM_NVIDIA_HOST', '127.0.0.1')}:{env.get('VLLM_NVIDIA_PORT', '18000')}/v1" else: vllm_url = f"http://127.0.0.1:{env.get('VLLM_ENGINE_PORT', '8000')}/v1" - vllm_api_key = env.get("VLLM_NVIDIA_API_KEY", "") or env.get("VLLM_HUST_API_KEY", "") or env.get("VLLM_ENGINE_API_KEY", "") - status, models_body = wait_for_models(vllm_url, timeout=args.timeout, api_key=vllm_api_key) + vllm_api_key = ( + env.get("VLLM_NVIDIA_API_KEY", "") + or env.get("VLLM_HUST_API_KEY", "") + or env.get("VLLM_ENGINE_API_KEY", "") + ) + status, models_body = wait_for_models( + vllm_url, timeout=args.timeout, api_key=vllm_api_key + ) check(status == 200, "vLLM models", f"status={status} url={vllm_url}", failures) ids, roots = model_ids(models_body) - expected_model = env.get("VLLM_NVIDIA_MODEL", "").strip() or env.get("VLLM_ENGINE_MODEL_PATH", "").strip() - actual_model_id = env.get("VLLM_NVIDIA_ACTUAL_MODEL_ID", "").strip() or env.get("VLLM_ENGINE_ACTUAL_MODEL_ID", "").strip() - served_model = env.get("VLLM_NVIDIA_SERVED_MODEL_NAME", "").strip() or env.get("VLLM_ENGINE_SERVED_MODEL_NAME", "").strip() + expected_model = ( + env.get("VLLM_NVIDIA_MODEL", "").strip() + or env.get("VLLM_ENGINE_MODEL_PATH", "").strip() + ) + actual_model_id = ( + env.get("VLLM_NVIDIA_ACTUAL_MODEL_ID", "").strip() + or env.get("VLLM_ENGINE_ACTUAL_MODEL_ID", "").strip() + ) + served_model = ( + env.get("VLLM_NVIDIA_SERVED_MODEL_NAME", "").strip() + or env.get("VLLM_ENGINE_SERVED_MODEL_NAME", "").strip() + ) app_model = env.get("DIGITAL_TWIN_MODEL_NAME", "").strip() if served_model == "${DIGITAL_TWIN_MODEL_NAME}": served_model = app_model @@ -234,12 +329,29 @@ def main() -> int: if expected_model: expected_exposed = expected_model in roots or expected_model in ids if actual_model_id: - expected_exposed = expected_exposed or actual_model_id in roots or actual_model_id in ids - check(expected_exposed, "actual model exposed", f"expected={actual_model_id or expected_model}", failures) + expected_exposed = ( + expected_exposed or actual_model_id in roots or actual_model_id in ids + ) + check( + expected_exposed, + "actual model exposed", + f"expected={actual_model_id or expected_model}", + failures, + ) if served_model: - check(served_model in ids, "served model exposed", f"served={served_model}", failures) + check( + served_model in ids, + "served model exposed", + f"served={served_model}", + failures, + ) if app_model: - check(app_model in ids, "app model matches served models", f"DIGITAL_TWIN_MODEL_NAME={app_model}", failures) + check( + app_model in ids, + "app model matches served models", + f"DIGITAL_TWIN_MODEL_NAME={app_model}", + failures, + ) if actual_model_for_alias_check and served_model and not args.allow_model_alias: check( served_model == actual_model_for_alias_check, @@ -274,7 +386,10 @@ def main() -> int: print(f" {key}={masked_env_value(key, env.get(key, ''))}") if failures: - print(f"Hosted/web verification failed with {len(failures)} issue(s).", file=sys.stderr) + print( + f"Hosted/web verification failed with {len(failures)} issue(s).", + file=sys.stderr, + ) return 1 print("Hosted/web verification passed.") return 0 diff --git a/uv.lock b/uv.lock index 8a12cce..5290ce3 100644 --- a/uv.lock +++ b/uv.lock @@ -2613,7 +2613,7 @@ wheels = [ [[package]] name = "sage-mate" -version = "4.6.27" +version = "4.6.28" source = { editable = "." } dependencies = [ { name = "cloudpickle" },