274 lines
12 KiB
Python
274 lines
12 KiB
Python
# ============================================================
|
||
# 会话引用 v1（moduleId: core-session-ref, 可重生 ✅）
|
||
# plan.md §4.7 / 矩阵 63 行：@会话/@方案/@版本/@报告/@知识 结构化引用。
|
||
# 硬规则：默认按时点快照取值（被引对象后续变化不影响已得结论），
|
||
# 来源删除/无权访问时显式失败，每次解析结果进入证据链（evidenceRefs）。
|
||
# @知识：经注入的 knowledge_lookup 回调从知识库强制命中（带出处），
|
||
# 未命中/未接线时显式 RefResolutionError（中文说明），绝不编造。
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
import copy # 时点快照：深拷贝隔离后续变更
|
||
import re # 引用语法解析
|
||
from dataclasses import dataclass # 结构化解析结果
|
||
from datetime import UTC, datetime # 快照时点
|
||
from typing import Any, Callable # 类型标注（knowledge_lookup 回调）
|
||
|
||
from server.timeutil import fmt_dt # 统一时间戳格式
|
||
|
||
# 引用种类：中文标签 -> 稳定 kind（与 plan.md §4.7 表格一致）
|
||
REF_KIND_LABELS = {
|
||
"会话": "session",
|
||
"方案": "scenario",
|
||
"版本": "version",
|
||
"报告": "report",
|
||
"知识": "knowledge",
|
||
}
|
||
REF_KINDS = frozenset(REF_KIND_LABELS.values())
|
||
|
||
# 引用语法：@会话:<id> / @方案:<id> / @版本:<id> / @报告:<id> / @知识:<assetId 或标题关键词>
|
||
# id 取到空白/中英文标点/括号/引号/下一个 @ 为止（支持中文 ID 与 -_. 版本号）
|
||
_REF_RE = re.compile(
|
||
r"@(会话|方案|版本|报告|知识)\s*[:：]\s*"
|
||
r"([^\s，。！？、；;：,:“”\"'\u2018\u2019\u201c\u201d「」『』()（）\[\]{}@#]+)"
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ParsedRef:
|
||
"""一次解析出的结构化引用（稳定：相同文本 -> 相同引用）。"""
|
||
|
||
kind: str # session / scenario / version / report
|
||
target_id: str # 目标对象稳定 ID
|
||
raw: str # 原始匹配串（含 @ 前缀）
|
||
|
||
@property
|
||
def ref_id(self) -> str:
|
||
"""稳定引用 ID：<kind>:<target_id>（与证据引用语法同构）。"""
|
||
return f"{self.kind}:{self.target_id}"
|
||
|
||
def to_evidence_ref(self) -> str:
|
||
"""证据链引用串（供 write_audit evidence_refs）。"""
|
||
return self.ref_id
|
||
|
||
|
||
@dataclass
|
||
class ResolvedRef:
|
||
"""按时点解析出的引用值：快照 + 出处 + 证据引用。"""
|
||
|
||
kind: str # session / scenario / version / report
|
||
target_id: str # 目标对象稳定 ID
|
||
snapshot_at: str # 取值时点（引用快照时间）
|
||
value: Any # 时点快照（深拷贝，后续变更不影响）
|
||
source: dict[str, Any] # 出处：来源表/消息数/来源会话等
|
||
|
||
@property
|
||
def ref(self) -> str:
|
||
"""证据链引用串：<kind>:<target_id>。"""
|
||
return f"{self.kind}:{self.target_id}"
|
||
|
||
|
||
class RefResolutionError(Exception):
|
||
"""引用解析显式失败：来源删除、无权访问或目标不存在。
|
||
|
||
携带 kind/target_id/reason，调用方可显式报告而非泛化掩盖。
|
||
"""
|
||
|
||
def __init__(self, kind: str, target_id: str, reason: str) -> None:
|
||
super().__init__(f"ref {kind}:{target_id} unresolved: {reason}")
|
||
self.kind = kind
|
||
self.target_id = target_id
|
||
self.reason = reason
|
||
|
||
|
||
def parse_refs(text: str) -> list[ParsedRef]:
|
||
"""解析文本中的 @会话/@方案/@版本/@报告 结构化引用为稳定 ID 列表。
|
||
|
||
- 相同输入 -> 相同输出（确定性、顺序稳定、按 (kind, target_id) 去重）
|
||
- 返回空列表当无引用；非法/缺 ID 的 @ 不产生引用
|
||
"""
|
||
out: list[ParsedRef] = []
|
||
seen: set[tuple[str, str]] = set()
|
||
for match in _REF_RE.finditer(text or ""):
|
||
label, target_id = match.group(1), match.group(2).strip()
|
||
kind = REF_KIND_LABELS[label]
|
||
key = (kind, target_id)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(ParsedRef(kind=kind, target_id=target_id, raw=match.group(0)))
|
||
return out
|
||
|
||
|
||
def _session_messages(store_data: dict[str, Any], session_id: str) -> list[dict[str, Any]]:
|
||
"""取会话消息（快照：messages[session_id] -> list[dict]）。"""
|
||
return list((store_data.get("messages") or {}).get(session_id) or [])
|
||
|
||
|
||
def _lookup_list(store_data: dict[str, Any], *keys: str) -> list[Any]:
|
||
"""按候选键取首个非空表（兼容世界/工作区不同命名）。"""
|
||
for key in keys:
|
||
value = store_data.get(key)
|
||
if value:
|
||
return list(value)
|
||
return []
|
||
|
||
|
||
def _now_local() -> datetime:
|
||
"""本地无时区时间（与仓库 fmt_dt 语义一致）。"""
|
||
return datetime.now(UTC).astimezone().replace(tzinfo=None)
|
||
|
||
|
||
def _now_snapshot(now: datetime | None) -> str:
|
||
return fmt_dt(now or _now_local())
|
||
|
||
|
||
def _check_project_access(store_data: dict[str, Any], session: dict[str, Any], reason: str) -> None:
|
||
"""会话归属项目校验：项目不在可见列表或已归档 -> 无权访问，显式失败。"""
|
||
project_id = session.get("projectId") or "__personal__"
|
||
if project_id == "__personal__":
|
||
return
|
||
projects = _lookup_list(store_data, "projects")
|
||
if not projects:
|
||
return
|
||
for project in projects:
|
||
if project.get("id") == project_id and not project.get("archived"):
|
||
return
|
||
raise RefResolutionError("session", session.get("id") or "", reason)
|
||
|
||
|
||
def _find_session(store_data: dict[str, Any], ref_id: str) -> dict[str, Any]:
|
||
"""在可见会话列表中定位目标；来源删除（不在可见列表）显式失败。"""
|
||
sessions = _lookup_list(store_data, "sessions")
|
||
for session in sessions:
|
||
if session.get("id") == ref_id:
|
||
_check_project_access(store_data, session, "no-access: project archived or not in scope")
|
||
return session
|
||
raise RefResolutionError("session", ref_id, "deleted-or-inaccessible")
|
||
|
||
|
||
def _resolve_session(store_data: dict[str, Any], ref_id: str, snapshot_at: str) -> ResolvedRef:
|
||
session = _find_session(store_data, ref_id)
|
||
messages = _session_messages(store_data, ref_id)
|
||
value = {"session": copy.deepcopy(session), "messages": copy.deepcopy(messages)}
|
||
source = {
|
||
"table": "sessions",
|
||
"sessionId": ref_id,
|
||
"messageCount": len(messages),
|
||
"projectId": session.get("projectId"),
|
||
}
|
||
return ResolvedRef(kind="session", target_id=ref_id, snapshot_at=snapshot_at,
|
||
value=value, source=source)
|
||
|
||
|
||
def _resolve_scenario(store_data: dict[str, Any], ref_id: str, snapshot_at: str) -> ResolvedRef:
|
||
cards = _lookup_list(store_data, "scenarios", "scenarioCards")
|
||
for card in cards:
|
||
if card.get("scenarioId") == ref_id:
|
||
return ResolvedRef(kind="scenario", target_id=ref_id, snapshot_at=snapshot_at,
|
||
value=copy.deepcopy(card), source={"table": "scenarios"})
|
||
raise RefResolutionError("scenario", ref_id, "deleted-or-inaccessible")
|
||
|
||
|
||
def _resolve_version(store_data: dict[str, Any], ref_id: str, snapshot_at: str) -> ResolvedRef:
|
||
versions = _lookup_list(store_data, "versions", "scheduleVersions", "flexScheduleVersions")
|
||
for version in versions:
|
||
if version.get("id") == ref_id or version.get("versionId") == ref_id \
|
||
or version.get("version") == ref_id:
|
||
return ResolvedRef(kind="version", target_id=ref_id, snapshot_at=snapshot_at,
|
||
value=copy.deepcopy(version), source={"table": "scheduleVersions"})
|
||
raise RefResolutionError("version", ref_id, "deleted-or-inaccessible")
|
||
|
||
|
||
def _resolve_report(store_data: dict[str, Any], ref_id: str, snapshot_at: str) -> ResolvedRef:
|
||
reports = _lookup_list(store_data, "reports", "reportAssets")
|
||
for report in reports:
|
||
if report.get("reportId") == ref_id or report.get("id") == ref_id:
|
||
return ResolvedRef(kind="report", target_id=ref_id, snapshot_at=snapshot_at,
|
||
value=copy.deepcopy(report), source={"table": "reports"})
|
||
raise RefResolutionError("report", ref_id, "deleted-or-inaccessible")
|
||
|
||
|
||
def _resolve_knowledge(ref_id: str, snapshot_at: str,
|
||
knowledge_lookup: Callable[[str], dict[str, Any] | None] | None) -> ResolvedRef:
|
||
"""从知识库强制命中 @知识 引用（RAG 命中轮，矩阵 63 行剩余项）。
|
||
|
||
- 命中：返回资产时点快照（正文 + 出处：assetId/title/kind/version）
|
||
- 未命中：显式 RefResolutionError（中文说明），绝不编造
|
||
- knowledge_lookup 未注入（默认 None）：显式报错说明 @知识 未接线，不静默
|
||
"""
|
||
if knowledge_lookup is None:
|
||
raise RefResolutionError(
|
||
"knowledge", ref_id,
|
||
"not-wired: 知识库检索未接线（resolve 未注入 knowledge_lookup 回调）")
|
||
asset = knowledge_lookup(ref_id)
|
||
if not asset:
|
||
raise RefResolutionError(
|
||
"knowledge", ref_id,
|
||
f"knowledge-miss: 知识库未命中《{ref_id}》（未编造）")
|
||
asset_id = str(asset.get("assetId") or ref_id) # 稳定目标 ID = 资产 ID
|
||
value = copy.deepcopy(asset) # 时点快照：深拷贝隔离后续变更
|
||
source = {
|
||
"table": "knowledge",
|
||
"assetId": asset.get("assetId"),
|
||
"title": asset.get("title"),
|
||
"kind": asset.get("kind"),
|
||
"version": asset.get("version"),
|
||
"query": ref_id, # 原始引用串（可回查请求）
|
||
}
|
||
return ResolvedRef(kind="knowledge", target_id=asset_id, snapshot_at=snapshot_at,
|
||
value=value, source=source)
|
||
|
||
|
||
_RESOLVERS = {
|
||
"session": _resolve_session,
|
||
"scenario": _resolve_scenario,
|
||
"version": _resolve_version,
|
||
"report": _resolve_report,
|
||
}
|
||
|
||
|
||
def resolve_ref(store_data: dict[str, Any], ref_kind: str, ref_id: str,
|
||
*, now: datetime | None = None,
|
||
knowledge_lookup: Callable[[str], dict[str, Any] | None] | None = None) -> ResolvedRef:
|
||
"""按时点解析一条引用：取时点快照；来源删除/无权访问显式失败。
|
||
|
||
Args:
|
||
store_data: 工作区快照（projects/sessions/messages/scenarios/versions/reports）
|
||
ref_kind: session / scenario / version / report / knowledge
|
||
ref_id: 目标对象稳定 ID（@知识 时取 assetId 或标题关键词）
|
||
now: 取值时点（测试可注入；默认当前时间）
|
||
knowledge_lookup: @知识 专用：知识库查找回调（入参目标串，返回资产 dict 或 None）。
|
||
默认 None 时解析 @知识 显式抛 RefResolutionError 说明未接线（不静默）。
|
||
Returns:
|
||
ResolvedRef：value 为深拷贝快照（被引对象后续变化不影响已得结论）
|
||
Raises:
|
||
RefResolutionError: 目标删除/不可见/无权访问/@知识 未命中或未接线
|
||
ValueError: 未知 ref_kind
|
||
"""
|
||
if ref_kind not in REF_KINDS:
|
||
raise ValueError(f"unknown ref kind: {ref_kind!r}")
|
||
snapshot_at = _now_snapshot(now)
|
||
if ref_kind == "knowledge":
|
||
return _resolve_knowledge(ref_id, snapshot_at, knowledge_lookup)
|
||
return _RESOLVERS[ref_kind](store_data, ref_id, snapshot_at)
|
||
|
||
|
||
def resolve_all(store_data: dict[str, Any], refs: list[ParsedRef],
|
||
*, now: datetime | None = None,
|
||
knowledge_lookup: Callable[[str], dict[str, Any] | None] | None = None,
|
||
) -> tuple[list[ResolvedRef], list[str]]:
|
||
"""批量解析引用，返回 (解析结果, 证据链引用串列表)。
|
||
|
||
证据链引用串形如 session:<id> / scenario:<id> / version:<id> / report:<id> /
|
||
knowledge:<assetId>，可直接传入 write_audit(..., evidence_refs=...) 进入审计证据链。
|
||
knowledge_lookup 透传给 @知识 解析（见 resolve_ref）；不含 @知识 引用时保持
|
||
既有行为完全兼容（不传亦可）。
|
||
"""
|
||
resolved: list[ResolvedRef] = []
|
||
for ref in refs:
|
||
resolved.append(resolve_ref(store_data, ref.kind, ref.target_id, now=now,
|
||
knowledge_lookup=knowledge_lookup))
|
||
evidence_refs = [item.ref for item in resolved]
|
||
return resolved, evidence_refs
|