"""记录的编解码与 schema 版本校验。 读者是写存储实现的人，以及**在库之外**读那批记录的下游代码：一个编排层要跨进程判断 「这个阶段上次跑完没有」，一份分析代码要脱离运行时读几个月前的轨迹 （`research-wiki/design/0003-public-api-shape.md` 决策二）。这个模块公开，就是为了这两处不必 各自照着字段手写一份解析器然后各自漂移。 **载荷是普通的字典与列表**，不是字节串。落成什么格式（一行 JSON、一列数据库、一个键值对） 由存储实现自己定，这里只管把记录变成可序列化的形状、以及把那个形状变回记录。 ## 编码的形状：就是那个记录类的字段，一个不多一个不少 键名逐字取自数据类的字段名，枚举取它的字符串值，元组变列表，嵌套的结构变嵌套的字典。 **没有任何元信息键**——没有类型标签、没有时间戳、没有写入者标识。哪一行是哪种记录由存储 实现自己解决（它写的时候知道），库不替它定文件布局；这样一条步记录的载荷和迁移前那份 逐行 jsonl 轨迹是同一个形状，`research-wiki/migrations/dissect.md` 那条「轨迹逐字段可比」的 验收标准才成立。 ## 解码的三条规矩 **一、带版本的记录，版本必须在载荷里，而且必须是认得的那几个之一。** 缺版本直接失败，不走 数据类上那个默认值——那个默认值是给**构造**用的（库写一条新记录时不必每处手填当前版本）， 而读取时「这条记录是哪个版本写的」只有载荷知道。靠默认值补齐会把「这是旧版本」和「这条 没写版本」压成同一个答案（`CLAUDE.md` §1.4）。 带版本的是三个：运行开始记录、步记录、运行结果。判据是「它会不会被下游单独拿出来读」； 其余几种只在恢复时被库自己读，而恢复读的是整份日志，版本由运行开始那一条统一交代 （`research-wiki/design/0006-public-names-and-signatures.md` 决策八）。 **二、记录类上的每一个字段都必须在载荷里，有默认值的也一样。** 少一个就是损坏，直接失败。 默认值补齐会把「这件事没发生过」改写成「发生了但值为空」，而这种损坏要到统计阶段才暴露， 那时已经分不清哪些行是真的。 直接后果：**往持久化结构里加字段，必须同时抬 schema 版本。** `CLAUDE.md` §1.3 那条「新增 字段必带默认值」管的是 Python 构造器——它保证已经在跑的下游代码不会因为多了个字段就崩； 持久化这一侧由 §1.4 管，走显式版本。两条不冲突，但都得照做。 **三、载荷里多出来的键一律忽略。** 存储实现常常要在同一个字典里塞自己的东西（哪一行是哪种 记录、写入时刻、分片键），把它整个递回来解码是最自然的写法。多余键报错的话，每个存储都得 先手动摘干净，而摘的过程正是最容易出错的地方。 ## 一个 `encode` 对七个 `decode_*` 编码不需要知道目标类型，对象自己知道自己是什么；解码需要，因为一个字典什么都不知道。 不对称是这个原因，不是疏忽。 """ from collections.abc import Mapping, Sequence from typing import TypeVar from polyloop.types import ( CURRENT_SCHEMA_VERSION, ActionOutcome, ActionStatus, Intent, IntentKind, ModelCallResult, ModelReply, ReplayPolicy, RunFinished, RunResult, RunStarted, StepCompleted, StepRecord, StopReason, ) #: 这个库认得的持久化 schema 版本。读到不在这里面的版本直接失败，不猜、不降级读。 #: #: 一个被猜着修好的日志会让后面每一个基于它的判断都建立在编造的数据上，而那些判断不会 #: 报错——它们只是错。 KNOWN_SCHEMA_VERSIONS = frozenset({CURRENT_SCHEMA_VERSION}) class DecodeError(ValueError): """一份载荷解不成记录：缺字段、类型不对、枚举取值不认得。""" class SchemaVersionError(DecodeError): """载荷的 schema 版本缺失，或者不是这个库认得的版本。 **和普通的解码失败分开**，因为两者要人做的事完全不同：这一个说「你的库比数据旧，去升级」 或者「这份数据不是本库写的」，那一个说「这条数据坏了」。压成一个异常的话，一次例行升级 漏做会被读成数据损坏，然后有人去修数据。 """ _T = TypeVar("_T") # --------------------------------------------------------------------------- # 取值：每一个都在缺失或类型不对时抛 DecodeError # --------------------------------------------------------------------------- def _raw(payload: Mapping[str, object], key: str, owner: str) -> object: if key not in payload: raise DecodeError(f"{owner} 的载荷缺字段 {key!r}") return payload[key] def _as_str(payload: Mapping[str, object], key: str, owner: str) -> str: value = _raw(payload, key, owner) if not isinstance(value, str): raise DecodeError(f"{owner}.{key} 应当是字符串，收到 {type(value).__name__}") return value def _as_optional_str(payload: Mapping[str, object], key: str, owner: str) -> str | None: value = _raw(payload, key, owner) if value is None or isinstance(value, str): return value raise DecodeError(f"{owner}.{key} 应当是字符串或空，收到 {type(value).__name__}") def _as_int(payload: Mapping[str, object], key: str, owner: str) -> int: value = _raw(payload, key, owner) # 布尔在 Python 里是整数的子类。不排掉的话，一个被写坏成 true 的计数会静默读成 1。 if not isinstance(value, int) or isinstance(value, bool): raise DecodeError(f"{owner}.{key} 应当是整数，收到 {type(value).__name__}") return value def _as_bool(payload: Mapping[str, object], key: str, owner: str) -> bool: value = _raw(payload, key, owner) if not isinstance(value, bool): raise DecodeError(f"{owner}.{key} 应当是布尔，收到 {type(value).__name__}") return value def _as_enum(payload: Mapping[str, object], key: str, owner: str, enum_cls: type[_T]) -> _T: value = _raw(payload, key, owner) if not isinstance(value, str): raise DecodeError(f"{owner}.{key} 应当是字符串，收到 {type(value).__name__}") try: return enum_cls(value) # type: ignore[call-arg] except ValueError as exc: raise DecodeError( f"{owner}.{key} 的取值 {value!r} 不是 {enum_cls.__name__} 认得的取值" ) from exc def _as_optional_enum( payload: Mapping[str, object], key: str, owner: str, enum_cls: type[_T] ) -> _T | None: if _raw(payload, key, owner) is None: return None return _as_enum(payload, key, owner, enum_cls) def _as_str_mapping(payload: Mapping[str, object], key: str, owner: str) -> dict[str, str]: value = _raw(payload, key, owner) if not isinstance(value, Mapping): raise DecodeError(f"{owner}.{key} 应当是映射，收到 {type(value).__name__}") result: dict[str, str] = {} for entry_key, entry_value in value.items(): if not isinstance(entry_key, str) or not isinstance(entry_value, str): raise DecodeError(f"{owner}.{key} 的键与值都必须是字符串，撞到 {entry_key!r}") result[entry_key] = entry_value return result def _as_nested(payload: Mapping[str, object], key: str, owner: str) -> Mapping[str, object]: value = _raw(payload, key, owner) if not isinstance(value, Mapping): raise DecodeError(f"{owner}.{key} 应当是一份嵌套载荷，收到 {type(value).__name__}") return value def _as_optional_nested( payload: Mapping[str, object], key: str, owner: str ) -> Mapping[str, object] | None: if _raw(payload, key, owner) is None: return None return _as_nested(payload, key, owner) def _as_sequence(payload: Mapping[str, object], key: str, owner: str) -> Sequence[object]: value = _raw(payload, key, owner) # 字符串也是序列，但它显然不是我们要的那种。 if isinstance(value, str) or not isinstance(value, Sequence): raise DecodeError(f"{owner}.{key} 应当是列表，收到 {type(value).__name__}") return value def _checked_version(payload: Mapping[str, object], owner: str) -> int: """取出并校验 schema 版本。缺它或者认不得都在这里失败。""" if "schema_version" not in payload: raise SchemaVersionError( f"{owner} 的载荷没有 schema_version。缺版本的载荷不按当前版本读——" "那会把「这是旧版本」和「这条没写版本」压成同一个答案" ) version = payload["schema_version"] if not isinstance(version, int) or isinstance(version, bool): raise SchemaVersionError( f"{owner}.schema_version 应当是整数，收到 {type(version).__name__}" ) if version not in KNOWN_SCHEMA_VERSIONS: raise SchemaVersionError( f"{owner} 的 schema 版本 {version} 不是本库认得的版本" f"（认得的是 {sorted(KNOWN_SCHEMA_VERSIONS)}）" ) return version # --------------------------------------------------------------------------- # 编码 # --------------------------------------------------------------------------- def _encode_model_reply(reply: ModelReply) -> dict[str, object]: return {"call_id": reply.call_id, "content": reply.content, "thinking": reply.thinking} def _encode_action_outcome(outcome: ActionOutcome) -> dict[str, object]: return { "status": outcome.status.value, "observation": outcome.observation, "observation_is_synthetic": outcome.observation_is_synthetic, "env_reported_completion": outcome.env_reported_completion, "observation_truncated_chars": outcome.observation_truncated_chars, } def _encode_step_record(step: StepRecord) -> dict[str, object]: return { "step_idx": step.step_idx, "raw_output": step.raw_output, "content_chars": step.content_chars, "thinking_chars": step.thinking_chars, "action": step.action, "parse_ok": step.parse_ok, "parse_error": step.parse_error, "observation": step.observation, "observation_is_synthetic": step.observation_is_synthetic, "observation_truncated_chars": step.observation_truncated_chars, "prompt_chars": step.prompt_chars, "call_id": step.call_id, "step_wall_ms": step.step_wall_ms, "tool_name": step.tool_name, "tool_arguments": step.tool_arguments, "action_status": None if step.action_status is None else step.action_status.value, "env_reported_completion": step.env_reported_completion, "schema_version": step.schema_version, } def _encode_run_result(result: RunResult) -> dict[str, object]: return { "run_id": result.run_id, "stop_reason": result.stop_reason.value, "final_answer": result.final_answer, "steps": [_encode_step_record(step) for step in result.steps], "schema_version": result.schema_version, "event_delivery_failures": result.event_delivery_failures, } def encode( record: RunStarted | Intent | ModelCallResult | StepCompleted | RunFinished | StepRecord | RunResult, ) -> dict[str, object]: """把一条记录变成可序列化的字典。 收得下七种：五种日志记录，加上会被下游单独拿出来读的步记录与运行结果。别的类型直接 报错，不做「尽力而为地把 dataclass 的字段抄一遍」——那种写法会在字段类型不可序列化时 把问题推到 `json.dumps` 那一步，而那时错误信息里已经看不出是哪条记录了。 """ if isinstance(record, RunStarted): return { "run_id": record.run_id, "parameter_snapshot": dict(record.parameter_snapshot), "schema_version": record.schema_version, } if isinstance(record, Intent): return { "run_id": record.run_id, "kind": record.kind.value, "call_index": record.call_index, "result_id": record.result_id, "replay_policy": record.replay_policy.value, } if isinstance(record, ModelCallResult): return { "run_id": record.run_id, "result_id": record.result_id, "reply": None if record.reply is None else _encode_model_reply(record.reply), "failure": record.failure, } if isinstance(record, StepCompleted): return { "run_id": record.run_id, "result_id": record.result_id, "action_outcome": ( None if record.action_outcome is None else _encode_action_outcome(record.action_outcome) ), "step": _encode_step_record(record.step), } if isinstance(record, RunFinished): return {"run_id": record.run_id, "result": _encode_run_result(record.result)} if isinstance(record, StepRecord): return _encode_step_record(record) if isinstance(record, RunResult): return _encode_run_result(record) raise TypeError(f"这个模块不认得的记录类型：{type(record).__name__}") # --------------------------------------------------------------------------- # 解码 # --------------------------------------------------------------------------- def _decode_model_reply(payload: Mapping[str, object]) -> ModelReply: owner = "ModelReply" return ModelReply( call_id=_as_optional_str(payload, "call_id", owner), content=_as_str(payload, "content", owner), thinking=_as_str(payload, "thinking", owner), ) def _decode_action_outcome(payload: Mapping[str, object]) -> ActionOutcome: owner = "ActionOutcome" return ActionOutcome( status=_as_enum(payload, "status", owner, ActionStatus), observation=_as_str(payload, "observation", owner), observation_is_synthetic=_as_bool(payload, "observation_is_synthetic", owner), env_reported_completion=_as_bool(payload, "env_reported_completion", owner), observation_truncated_chars=_as_int(payload, "observation_truncated_chars", owner), ) def decode_step_record(payload: Mapping[str, object]) -> StepRecord: """解一条步记录。 这是下游读几个月前那份轨迹时用的入口：一行一条。**它要求 `schema_version` 在场**，所以 迁移前那份轨迹（本库存在之前写的，根本没有这个字段）走不通这条路——那份文件由下游自己 构造 `StepRecord`，五个新增字段取数据类上的默认值 （`research-wiki/design/0006-public-names-and-signatures.md` 决策九）。 """ owner = "StepRecord" version = _checked_version(payload, owner) return StepRecord( step_idx=_as_int(payload, "step_idx", owner), raw_output=_as_str(payload, "raw_output", owner), content_chars=_as_int(payload, "content_chars", owner), thinking_chars=_as_int(payload, "thinking_chars", owner), action=_as_optional_str(payload, "action", owner), parse_ok=_as_bool(payload, "parse_ok", owner), parse_error=_as_optional_str(payload, "parse_error", owner), observation=_as_str(payload, "observation", owner), observation_is_synthetic=_as_bool(payload, "observation_is_synthetic", owner), observation_truncated_chars=_as_int(payload, "observation_truncated_chars", owner), prompt_chars=_as_int(payload, "prompt_chars", owner), call_id=_as_optional_str(payload, "call_id", owner), step_wall_ms=_as_int(payload, "step_wall_ms", owner), tool_name=_as_optional_str(payload, "tool_name", owner), tool_arguments=_as_optional_str(payload, "tool_arguments", owner), action_status=_as_optional_enum(payload, "action_status", owner, ActionStatus), env_reported_completion=_as_bool(payload, "env_reported_completion", owner), schema_version=version, ) def decode_run_result(payload: Mapping[str, object]) -> RunResult: """解一份运行结果。跨进程判断「上一次跑完没有」读的就是它。""" owner = "RunResult" version = _checked_version(payload, owner) steps = _as_sequence(payload, "steps", owner) decoded_steps = [] for index, entry in enumerate(steps): if not isinstance(entry, Mapping): raise DecodeError(f"{owner}.steps[{index}] 应当是一份嵌套载荷") decoded_steps.append(decode_step_record(entry)) return RunResult( run_id=_as_str(payload, "run_id", owner), stop_reason=_as_enum(payload, "stop_reason", owner, StopReason), final_answer=_as_optional_str(payload, "final_answer", owner), steps=tuple(decoded_steps), schema_version=version, event_delivery_failures=_as_int(payload, "event_delivery_failures", owner), ) def decode_run_started(payload: Mapping[str, object]) -> RunStarted: """解一条运行开始记录。一份日志的头，读它才知道整份日志怎么解。""" owner = "RunStarted" version = _checked_version(payload, owner) return RunStarted( run_id=_as_str(payload, "run_id", owner), parameter_snapshot=_as_str_mapping(payload, "parameter_snapshot", owner), schema_version=version, ) def decode_intent(payload: Mapping[str, object]) -> Intent: """解一条意图。 **不带 schema 版本**，它只在恢复时被库自己读，而恢复读的是整份日志，版本由运行开始 那一条统一交代。给每一条都带一个版本会让同一份日志里出现好几个可以各自演进的版本号， 「这份日志是哪个版本」就没有答案了。 """ owner = "Intent" return Intent( run_id=_as_str(payload, "run_id", owner), kind=_as_enum(payload, "kind", owner, IntentKind), call_index=_as_int(payload, "call_index", owner), result_id=_as_str(payload, "result_id", owner), replay_policy=_as_enum(payload, "replay_policy", owner, ReplayPolicy), ) def decode_model_call_result(payload: Mapping[str, object]) -> ModelCallResult: """解一条模型调用结果。""" owner = "ModelCallResult" reply_payload = _as_optional_nested(payload, "reply", owner) return ModelCallResult( run_id=_as_str(payload, "run_id", owner), result_id=_as_str(payload, "result_id", owner), reply=None if reply_payload is None else _decode_model_reply(reply_payload), failure=_as_optional_str(payload, "failure", owner), ) def decode_step_completed(payload: Mapping[str, object]) -> StepCompleted: """解一条逐步结果。 `result_id` 与 `action_outcome` 那条「同时为空或同时有值」的不变量由 `StepCompleted` 自己在构造时校验，这里不重复一遍——重复的校验迟早有一处被改。 """ owner = "StepCompleted" outcome_payload = _as_optional_nested(payload, "action_outcome", owner) return StepCompleted( run_id=_as_str(payload, "run_id", owner), result_id=_as_optional_str(payload, "result_id", owner), action_outcome=( None if outcome_payload is None else _decode_action_outcome(outcome_payload) ), step=decode_step_record(_as_nested(payload, "step", owner)), ) def decode_run_finished(payload: Mapping[str, object]) -> RunFinished: """解一条运行结束记录。""" owner = "RunFinished" return RunFinished( run_id=_as_str(payload, "run_id", owner), result=decode_run_result(_as_nested(payload, "result", owner)), ) __all__ = [ "KNOWN_SCHEMA_VERSIONS", "DecodeError", "SchemaVersionError", "decode_intent", "decode_model_call_result", "decode_run_finished", "decode_run_result", "decode_run_started", "decode_step_completed", "decode_step_record", "encode", ]