跳转到正文

背包、装备观察与物品证据 ​

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 权限效果
observeframework.self.inventory.observe被动读取原生入站观察,不请求新包。
readframework.inventory.read旧式单槽缓存读取。
refreshframework.self.inventory.refresh主动开关背包窗口,需要世界写入授权。
move / splitframework.inventory.move向明确空槽移动指定数量。
merge / swapframework.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_supportedNBT 类型信息与保守严格对账支持标记。
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。关闭会话与动作结果分开处理,资源占用不会增加本页成员权限,也不会回滚游戏效果。

Neomega · 使用、开发与参考文档