qaiji.core.semantics.handle 源代码

# Copyright 2026 Qaiji-IR core developer: Ye Jun <yjmaxpayne@hotmail.com>
# SPDX-License-Identifier: Apache-2.0
"""承载电路摘要及其内容哈希的冻结、JSON 安全的句柄。"""

from __future__ import annotations

from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from enum import StrEnum
from types import MappingProxyType
from typing import Any

__all__ = ["HandleStatus", "JSONValue", "SemanticIRHandle", "freeze_summary"]

_SCHEMA_VERSION = "qaiji.semantic_ir_handle.v0"


[文档] class HandleStatus(StrEnum): """句柄摘要载荷的可用性状态。 只有四个取值,而非上游的五个:qaiji 不存在"不适用"的句柄场景,因此 ``NOT_APPLICABLE`` 是被有意舍弃的。 """ AVAILABLE = "available" PLANNED = "planned" OMITTED = "omitted" FAILED = "failed"
type JSONValue = None | bool | int | float | str | Sequence["JSONValue"] | Mapping[str, "JSONValue"] """无需 ``default=`` 钩子即可被 ``json.dumps`` 编码的值树。""" def _freeze(value: Any) -> Any: if isinstance(value, Mapping): for key in value: if not isinstance(key, str): raise TypeError(f"summary keys must be str, got {key!r} ({type(key).__name__})") return MappingProxyType({key: _freeze(item) for key, item in value.items()}) if isinstance(value, list | tuple): return tuple(_freeze(item) for item in value) if value is None or isinstance(value, bool | int | float | str): return value raise TypeError(f"value is not JSON-compatible: {type(value).__name__}")
[文档] def freeze_summary(value: Mapping[str, Any]) -> Mapping[str, JSONValue]: """递归冻结一个 JSON 形状的映射,得到不可变树。 每个 ``Mapping`` 变成 ``MappingProxyType``;每个 ``list``/``tuple`` 变成 ``tuple``。与本实现所借鉴的上游模型不同(上游把嵌套容器重建为新的、仍然可变的 list),这里的冻结是深度不可变的:真正做到这一点的是 tuple 转换,而不是某种 共享引用的取巧手法。 Args: value: JSON 形状的映射,原始的或已冻结的均可(幂等)。 Returns: 每一层都已冻结的同一棵树。 Raises: TypeError: 映射的键不是 ``str``,或叶子值不属于 ``None``/``bool``/``int``/``float``/``str``/list/tuple/Mapping。 """ return _freeze(value) # type: ignore[no-any-return]
[文档] @dataclass(frozen=True, slots=True) class SemanticIRHandle: """下游坍缩阶段消费的唯一对象:摘要、哈希与状态。 Example: >>> from qaiji.core.semantics.handle import SemanticIRHandle >>> SemanticIRHandle().status <HandleStatus.PLANNED: 'planned'> """ schema_version: str = _SCHEMA_VERSION handle_id: str | None = None producer: str | None = None summary: Mapping[str, JSONValue] = field(default_factory=dict) content_hash: str | None = None status: HandleStatus = HandleStatus.PLANNED def __post_init__(self) -> None: """强制校验 schema 版本字面量,并对摘要做深度冻结。""" if self.schema_version != _SCHEMA_VERSION: raise ValueError( f"schema_version must be the frozen literal {_SCHEMA_VERSION!r}, " f"got {self.schema_version!r}" ) object.__setattr__(self, "summary", freeze_summary(self.summary))