外观
背包、装备观察与物品证据
ctx.inventory 操作连接机器人的库存。先观察真实槽位,再准备动作;未知槽位不等于空槽。这里的装备支持原生槽位观察,快捷栏选槽与手持动作见物品选择与使用。
本页对应工具包中的 inventory、inventory_layout、inventory_context、inventory_evidence 和 item_reconciliation 模块。也可从 neomega_runtime.inventory 导入 Inventory,用 Inventory(operations) 绑定现有 Operations;不要另建绕开 Host 的库存连接。
调用与权限
除标为 async 或纯函数的方法外,下列方法都只返回 OperationIntent,不会立即读取或修改游戏。保存 intent.payload() 后,通过 ctx.operations.submit() 提交并保留操作 ID,再 wait() / get() 查询。回执的操作结果位于 receipt["result"]["result"],只在状态及结构确认后读取。
所有意图方法都要求关键字参数 idempotency_key: str、deadline: str,无默认值。前者由业务为每个独立动作分配并持久保留;后者是带时区的绝对截止时间。下文列出各方法的必需参数与默认值。
| 入口 | 必需的 operation 权限 | 效果 |
|---|---|---|
observe | framework.self.inventory.observe | 被动读取原生入站观察,不请求新包。 |
read | framework.inventory.read | 旧式单槽缓存读取。 |
refresh | framework.self.inventory.refresh | 主动开关背包窗口,需要世界写入授权。 |
move / split | framework.inventory.move | 向明确空槽移动指定数量。 |
merge / swap | framework.inventory.merge / framework.inventory.swap | 精确引用下合堆或整栈交换。 |
权限须在插件清单中声明并获管理员授权。写入还要求机器人进服、原生连接可用及世界写入门允许;清单授权不等于所有运行条件已满足。observe_context 额外需要 framework.self.death.read。
原生观察与装备槽
text
ctx.inventory.observe(window_id: int, *, idempotency_key: str, deadline: str) -> OperationIntent
ctx.inventory.read(window_id: int, slot: int, *, idempotency_key: str, deadline: str) -> OperationIntent
ctx.inventory.refresh(window_id: int, *, idempotency_key: str, deadline: str) -> OperationIntent
ctx.inventory.layout(snapshot: dict, *, max_age_ms: int,
at: str | None = None) -> dict| 原生窗口 | 观察形状 | 槽位 |
|---|---|---|
0 | 主背包,36 槽 | 0..8 快捷栏;9..35 主存储。 |
119 | 副手,1 槽 | 0 副手。 |
120 | 护甲,4 或 5 槽 | 0..3 头、胸、腿、脚;仅观察到五槽形状时出现 4,标为 extra,不推断其装备语义。 |
observe() 返回 InventoryObservation 形状的字典。InventoryObservation、InventorySlot 可从 neomega_runtime 导入,它们是 TypedDict 类型提示,不会替你验证任意输入。
| 字段组 | 含义 |
|---|---|
schema_version, subject, source | 模式版本、self 主体与 native_inventory_packets 来源。 |
connection_id, connected_at | 原生游戏连接及起点;不是 Host 会话 ID,也不是插件传输连接。 |
revision, sampled_at | 该次采样的修订号与时间;采样缓存不代表刚收到新包。 |
window_id, slot_count, shape_known, complete | 窗口、形状与槽位覆盖情况。 |
slots, coverage, freshness, packet_diagnostics | 槽记录、覆盖、新鲜度及包诊断;需要同时查看,不能只看 complete。 |
每个 InventorySlot 必有 slot: int、known: bool、state: "unknown" | "empty" | "occupied"。可选字段如下:
| 字段 | 用途 |
|---|---|
empty, stack_id, item | 空/非空证据、原生栈 ID 和可见物品记录;已知空槽必须有明确空证据。 |
fingerprint, attributes_fingerprint | 完整物品指纹和忽略数量的属性指纹;直接保留,不自行重算。 |
nbt_types, reconciliation_supported | NBT 类型信息与保守严格对账支持标记。 |
observed_revision, observed_at, packet_id, age_ms | 该槽原生观察的修订号、时间、包来源与年龄。 |
read() 的 window_id 为 0..4294967295,slot 为 0..255;这是参数范围,不代表任意窗口已被观察。其结果为 {window_id, slot, cached: true, item},成功证据为 dispatched,缓存不可用时原因是 inventory_cache_unavailable。返回的旧缓存不是原生证据,缓存缺失也不是空槽。不要用它构造 HeldItemRef、合成引用或原生新鲜度结论。
observe() 成功证据为 verified,只验证此次返回的观察内容,不会提升未知槽或陈旧槽。self_inventory_unavailable 表示不能取得观察,self_inventory_invalid_result / self_inventory_result_too_large 表示结果不可返回。
refresh() 接受与 observe() 相同的 0/119/120 窗口,是有界的开窗/关窗尝试,成功表示生命周期得到确认。仍须检查 fresh_complete、fresh_slots 和返回的原生槽记录;succeeded 且原因 self_inventory_refresh_partial 表示仍未获得完整新鲜槽证据。unknown/self_inventory_refresh_lifecycle_uncertain 会保留恢复隔离,不能换 key 再开窗。
按槽展示与计算年龄
layout() 是纯函数;独立入口为 neomega_runtime.inventory_layout.describe_inventory(snapshot, *, max_age_ms, at=None),签名与返回相同。它复制槽记录,加上 section、freshness: fresh|stale|unknown、age_ms,返回 window_id、slot_count、connection_id、sampled_at、evaluated_at、slots。
max_age_ms 必须是非负整数。省略 at 时以 Host 的 sampled_at 为年龄终点;需要判断调用时的年龄,应显式传入当前 UTC 时间。它只认可同一原生证据中的有效修订号、观察时间及包来源,不会补空气。未知形状、重复/遗漏槽位、非法时间边界会抛出 ValueError。
python
from datetime import datetime, timezone
# snapshot 是成功 observe 回执中的原生结果。
view = ctx.inventory.layout(
snapshot, max_age_ms=1000,
at=datetime.now(timezone.utc).isoformat(),
)
usable = [row for row in view["slots"]
if row["known"] and row["freshness"] == "fresh"]组合原生身份与背包观察
text
await ctx.inventory.observe_context(
*, idempotency_key: str, deadline: str, journal,
timeout: float = 6
) -> dict
await neomega_runtime.inventory_context.observe_context(
inventory, *, idempotency_key, deadline, journal, timeout=6
) -> dict该方法依次读取原生身份、主背包、原生身份,核验三次观察属于同一原生连接、身份未变、时间顺序有效。返回 native_epoch(connection_id、connected_at)、self_runtime_id、inventory、保留意图/操作 ID/回执的 operations,并明确给出 atomic_snapshot: false、current_state_guaranteed: false。
journal(record) 必须是异步持久化函数,返回前存好独立记录。调用方先保存业务请求并完成持久的一次性准入;重复请求或重启应读取原记录,不能再次调用来“恢复”。SDK 会在提交前、取得操作 ID 后、取得回执后保存进度,不提供跨调用去重或自动重试。
python
# admit_once 和 save_context 是业务提供的持久化函数,不是 SDK 接口。
# 仅在业务已确认本次为首次准入后执行:
context = await ctx.inventory.observe_context(
idempotency_key=business_key, deadline=deadline,
journal=save_context, timeout=6,
)
snapshot = context["inventory"]等待预算取 timeout 与 deadline 剩余时间的较小值,包含 journal 时间;为保留持久化所有权,不会为了预算取消正在执行的 journal,因此该函数本身也必须有界。非法输入抛 ValueError,预算用尽抛 TimeoutError,身份缺失、连接变化或不确认的回执抛 ProtocolError。已完成的部分读取保留在 journal。它不主动刷新,也不证明新包已到达。
移动、拆分、合堆和交换
text
ctx.inventory.move(source_slot: int, destination_slot: int, count: int, *,
expected_source_stack_id: int, expected_destination_stack_id: int = 0,
idempotency_key: str, deadline: str) -> OperationIntent
ctx.inventory.split(source_slot: int, destination_slot: int, count: int, *,
expected_source_stack_id: int, idempotency_key: str, deadline: str) -> OperationIntent
ctx.inventory.merge(source_slot: int, destination_slot: int, count: int, *,
expected_source_stack_id: int, expected_destination_stack_id: int,
expected_source_fingerprint: str, expected_destination_fingerprint: str,
expected_source_count: int, expected_destination_count: int,
destination_count_limit: int, idempotency_key: str, deadline: str) -> OperationIntent
ctx.inventory.swap(source_slot: int, destination_slot: int, *,
expected_source_stack_id: int, expected_destination_stack_id: int,
expected_source_fingerprint: str, expected_destination_fingerprint: str,
idempotency_key: str, deadline: str) -> OperationIntent这些动作只操作窗口 0 的 0..35 槽,源和目的不能相同。count 为 1..64 且不能超过实际源数量。move / split 的目标必须已经观察为空,目的栈 ID 为 0;移动部分数量即拆堆。默认 0 只是参数默认值,不能替代空槽证据。
merge 要求两端非空、栈 ID 与完整指纹精确匹配、属性兼容、数量精确;count <= expected_source_count,并且 count + expected_destination_count <= destination_count_limit。数量参数均为 1..64。上界来自已知原数量或明确玩法约束,不代表所有物品都能堆叠 64。swap 交换两端整栈,不接收数量。
python
# source、destination 来自保留的原生观察;还须核验连接与新鲜度。
if not (source["known"] and source["state"] == "occupied"
and destination["known"] and destination["state"] == "empty"
and destination.get("stack_id") == 0):
raise ValueError("需要已知物品源与已知空目的槽")
intent = ctx.inventory.split(
source["slot"], destination["slot"], 3,
expected_source_stack_id=source["stack_id"],
idempotency_key=business_key, deadline=deadline,
)
await save_intent(intent.payload()) # 业务持久化,且保留 source/destination
operation_id = await ctx.operations.submit(intent)
await save_operation_id(operation_id)
receipt = await ctx.operations.wait(operation_id)成功回执为 succeeded/acknowledged,结果包含 server_acknowledged: true、independent_readback: false。后续必须重新观察,不能从 ACK 数量拼出下一步物品引用。Host 在发送前再次检查实际槽位;引用变化可能拒绝动作。
| 结果/原因 | 处理 |
|---|---|
failed/inventory_move_not_sent | 未进入发送;检查槽位、引用及运行条件。 |
failed/inventory_move_rejected | 服务端明确拒绝,证据为 ACK;保留原回执。 |
unknown/inventory_move_uncertain | 发送后完成情况不明,先核对原操作及原生库存。 |
unknown/inventory_move_lifecycle_uncertain | 窗口生命周期不明,保留恢复隔离,交管理员处理。 |
非法参数、权限拒绝和未进服也会阻止执行;不要只处理本表的业务原因。
typed NBT 与严格数量对账
物品 JSON 只适合读取可见字段。NBT 的 byte/short/int/long、浮点和各类数组在 JSON 中可能失去类型;nbt_types 保留类型树,指纹在类型丢失前生成。严格对账需要同时保留 item、nbt_types、完整指纹、属性指纹及 reconciliation_supported。
名称、附魔、耐久等可在支持范围内严格比较;嵌套容器 Items、BlockEntityTag、ChargedItem 等复杂内容可能标为不支持严格对账。这个标记不等于原生传输必然不支持,不能把一次搬运成功扩展成所有复杂 NBT 都已验证。物品生成见世界与建造,不要用生成物品替代原物品转移。
text
from neomega_runtime.item_reconciliation import compare_transfer
compare_transfer(*, source_before, destination_before,
source_after, destination_after, count) -> bool这是无 IO 纯函数,count 为 1..64。它检查源减少、目的增加及完整属性保持;可用于移入空槽或兼容非空栈。调用前由业务确认观察来源、同槽身份与已知状态。比较器本身不检查 known/state,空槽仅按 empty=True、stack_id=0 且没有物品判断;非空记录要求数量、正栈 ID、合法指纹、类型树及支持标记。证据不适用抛 ValueError;证据格式有效但数量/属性不匹配返回 False。
python
from neomega_runtime.item_reconciliation import compare_transfer
matched = compare_transfer(
source_before=source_before, destination_before=destination_before,
source_after=source_after, destination_after=destination_after, count=3,
)该比较器不认证 JSON 来源,不证明时间、因果或当前仍保持这个状态,也不会把历史 unknown 回执改写为成功。容器其余槽位的保全比较见容器。
动作后的新鲜槽证据
text
from neomega_runtime.inventory_evidence import verify_slots_after
verify_slots_after(before: Mapping[str, Any], after: Mapping[str, Any], *,
slots: Sequence[int], after_time: str, max_age_ms: int,
require_empty: bool = False) -> dict该纯函数要求同一原生连接、连接起点和窗口,快照及指定槽位的修订号更新,原生包来源有效,并且观察时间晚于 after_time、截至采样时间年龄不超过 max_age_ms。require_empty=True 还要求指定槽均有明确空证据。
通过时返回 verified: true、scope: "requested_slots"、slots、connection_id、revision;通常的不匹配返回 verified: false 和 reason。这只验证所选槽位,不验证整个背包。after_time 是显式 UTC 动作边界;使用完成时间更保守,但可能拒绝在完成回执前已经到达的真实更新。
python
from neomega_runtime.inventory_evidence import verify_slots_after
evidence = verify_slots_after(
before_snapshot, after_snapshot, slots=[source_slot, destination_slot],
after_time=action_boundary_utc, max_age_ms=1000,
)
# 再对这些槽运行 compare_transfer,分别验证新鲜度与物品变化。检查不通过应保留“未证实”,重新取得有来源的观察或人工核对。不要自动重放未知写入,详细恢复规则见错误与结果核对。
原生库存 View
本节接口自配套 1.0.0 Host 与 SDK 提供。通过 Bus 目录与 Watch读取 major 1 的 inventory.current;托管插件须声明并获准对应 permissions.services。它由原生库存 owner 的原子初始快照及后续更新派生,不主动开背包窗口,也不能把旧 ctx.inventory.read 单槽缓存当成同等完整的库存快照。
先取得原子基线,再在同一 epoch 内应用更新。检查 View 的 complete,以及每个窗口的 available、complete、shape_known 和槽位 known;未知槽位不等于空槽,complete=false 时暂停依赖完整库存的判断。来源从不完整恢复为完整时开启新 epoch,旧流结束,消费者重新打开取得新基线;不要跨 epoch 拼接槽位或补猜漏掉的更新。
主动刷新使用 inventory.refresh op,另需该成员的 permissions.services 授权和世界写入条件,接纳后仍查原操作回执。原有观察、移动与装备 helper 保留各自权限和回执合同;它们与容器操作共享原生窗口会话,发送可能受窗口租约限制。取消等待不代表窗口已经释放,unknown 沿用原 ID 对账。
多个受管动作需要连续占用同一组资源时,可使用 1.1.0 的资源会话 Scope。关闭会话与动作结果分开处理,资源占用不会增加本页成员权限,也不会回滚游戏效果。