taolib.harness.devtools.inspector

taolib.harness.devtools.inspector#

实时状态检查器 - 订阅、对比与展示统一状态管理器中的状态。

依托 UnifiedStateManager 暴露的快照 / 订阅能力，本模块提供：

Attributes#

WatchCallback

订阅回调签名：接收状态变更事件，可同步或异步处理。

Classes#

InspectorConfig

状态检查器配置。

StateDiff

两个状态快照的差异报告。

StateInspector

状态检查器 - 提供同步/异步两种交互方式查看与监听状态。

Functions#

format_state_tree(→ str)

将嵌套状态格式化为树形文本。

Module Contents#

class taolib.harness.devtools.inspector.InspectorConfig#

Bases: pydantic.BaseModel

状态检查器配置。

exclude_keys: list[str]#
include_sources: list[str]#
max_history: int#
model_config#
poll_interval_seconds: float#
verbosity: int#
class taolib.harness.devtools.inspector.StateDiff#

Bases: pydantic.BaseModel

两个状态快照的差异报告。

added: dict[str, Any]#
changed: dict[str, dict[str, Any]]#
property is_empty: bool#

差异是否为空。

model_config#
removed: dict[str, Any]#
thread_id: str#
version_from: int = 0#
version_to: int = 0#
class taolib.harness.devtools.inspector.StateInspector(manager: taolib.harness.core.state.UnifiedStateManager, *, config: InspectorConfig | None = None)#

状态检查器 - 提供同步/异步两种交互方式查看与监听状态。

构造检查器。

参数:

  • manager -- 关联的统一状态管理器。

  • config -- 检查器配置，缺省采用默认值。

async awatch(thread_id: str, callback: WatchCallback, *, stop: asyncio.Event | None = None) → None#

异步监听：阻塞直到 stop 被设置后取消订阅。

close() → None#

注销监听。

static diff(snapshot_a: taolib.harness.core.state.StateSnapshot, snapshot_b: taolib.harness.core.state.StateSnapshot) → StateDiff#

对比两个快照，返回结构化差异。

format(snapshot: taolib.harness.core.state.StateSnapshot) → str#

以树形结构格式化快照。

history(thread_id: str, *, limit: int | None = None) → list[taolib.harness.core.state.StateChangeEvent]#

获取指定线程的状态变更历史。

async inspect(thread_id: str) → taolib.harness.core.state.StateSnapshot#

获取指定线程的当前聚合快照。

watch(thread_id: str, callback: WatchCallback) → collections.abc.Callable[[], None]#

同步订阅状态变更事件，返回取消订阅函数。

property known_threads: collections.abc.Iterable[str]#

已观测到的线程 ID 集合。

taolib.harness.devtools.inspector.format_state_tree(payload: Any, *, indent: int = 0, prefix: str = '') → str#

将嵌套状态格式化为树形文本。

type taolib.harness.devtools.inspector.WatchCallback = Callable[[StateChangeEvent], Awaitable[None] | None]#

订阅回调签名：接收状态变更事件，可同步或异步处理。