Neomega

外观

物品选择、使用与动作组合 ​

ctx.item_actions 将快捷栏选择、攻击、对实体交互、对空使用、点击方块和释放物品准备成可追踪的一次性动作。它不会自动选槽、等待蓄力、补发释放或推断命中。需要持续瞄准、近战组合和弩状态观察时，继续阅读瞄准与战斗。

下面的插件 SDK 接口使用 ctx.operations 受理动作。先从当前连接观察物品与目标，按各方法声明操作和包权限，再提交意图。

一次动作的输入与权限 ​

text
held_item_ref(snapshot, slot) -> dict
ActionAttempt(intent, operation_id=None, receipt=None, started=False)
ItemActions(operations)

从 neomega_runtime.item_actions 导入。snapshot 必须是 source='native_inventory_packets' 的原生背包窗口 0 快照。快捷栏 slot 为 0..8；槽位必须明确 known=True，有正 observed_revision 和合法 stack_id。未知槽位不能假定为空气。

held_item_ref() 返回 ConnectionID、Slot、ObservedRevision、StackNetworkID。Host 写出前会重新校验这个引用；发生背包变化时应停止当前计划，再观察，而不是用一个任意 stack ID 修补请求。

动作typed packet权限
selectmob_equipment.hotbar，包 31framework.packet.send_typed，packet_send_ids 包含 31
attack、interact_entityinventory_transaction.use_item_on_entity_attack / inventory_transaction.use_item_on_entity_interact，包 30同一操作权限，包号 30
use_air、use_block、releaseinventory_transaction.use_item_air / inventory_transaction.use_item_block / inventory_transaction.release_item，包 30同上
观察选槽原始包流中的入站 MobEquipment另申请所用观察流的权限；不是发送包授权
使用后中性输入player_auth_input.movement，包 144另授权 144

世界写入还要求已进游戏、原生连接可用且 Host 允许写入。库存观察、目标观察和包观察各有自己的权限，见权限与 manifest。

完整动作方法 ​

以下方法都只准备 ActionAttempt，不执行 IO；idempotency_key 和带时区的固定 deadline 由调用者提供。

text
ItemActions.select(snapshot, slot, *, entity_runtime_id,
                   idempotency_key, deadline)
ItemActions.attack(snapshot, slot, *, target_runtime_id, position,
                   clicked_position, idempotency_key, deadline)
ItemActions.interact_entity(snapshot, slot, *, target_runtime_id, position,
                            clicked_position, idempotency_key, deadline)
ItemActions.use_air(snapshot, slot, *, trigger_type, block_position, block_face,
                    position, clicked_position, block_runtime_id,
                    client_prediction, idempotency_key, deadline)
ItemActions.use_block(snapshot, slot, *, trigger_type, block_position, block_face,
                      position, clicked_position, block_runtime_id,
                      client_prediction, idempotency_key, deadline)
ItemActions.release(snapshot, slot, *, head_position, idempotency_key, deadline)
async ItemActions.dispatch(attempt, *, timeout=10, on_admitted=None,
                           poll_interval=.05)

entity_runtime_id 是当前机器人运行时 ID；target_runtime_id 必须来自当前目标观察。position、clicked_position、head_position、block_position、block_face 与触发参数都由调用者根据具体协议动作和实际场景提供。use_block 的 block_runtime_id 必须来自实际方块，不能猜；use_air 也不提供通用协议默认值。SDK 没有隐式朝向、命中点或物品专用交互策略。

dispatch() 最多提交一次，受理前就将 attempt.started=True，受理后写入 operation_id。on_admitted(attempt) 是异步回调，适合立即保存 ID；回调失败不会撤销已受理动作。成功返回完整 receipt，必须同时为 state='succeeded' 和 result.evidence='dispatched'，否则抛 ProtocolError。取消或网络异常后，原 attempt 仍是恢复入口，不能再次 dispatch。

可复制的单次选择封装 ​

此函数需要真实库存快照、机器人 runtime ID、业务固定 key/deadline 和持久化回调。它只报告发送回执；若产品需要“选槽已被观察到”，使用下一节的 HotbarSelection。control 是该 worker 所有控制动作共用的 ActionControl。

python
async def select_once(ctx, inventory, slot, runtime_id, *,
                      control, journal, key, deadline):
    async with control:
        attempt = ctx.item_actions.select(
            inventory, slot, entity_runtime_id=runtime_id,
            idempotency_key=key, deadline=deadline,
        )

        def record():
            return dict(intent=attempt.intent.payload(), started=attempt.started,
                        operation_id=attempt.operation_id, receipt=attempt.receipt)

        async def admitted(_):
            await journal(record())

        await journal(record())
        try:
            return await ctx.item_actions.dispatch(attempt, on_admitted=admitted)
        finally:
            if attempt.started and (attempt.receipt is None or
                                    attempt.receipt.get('state') == 'unknown'):
                control.block(record())
            await journal(record())

journal 的返回代表保存完成；不要用 print() 替代。如果提交响应丢失，可能没有 operation ID，但已经保存的 intent 含原始 key；按原 key 核对受理状态，而不是生成新 key。控制恢复条件见移动页。

观察选槽与归档检查 ​

text
equipment_matches(event, *, entity_runtime_id, slot, stack_network_id,
                  after_sequence) -> bool
async wait_equipment(stream, *, entity_runtime_id, slot, stack_network_id,
                     after_sequence, timeout=5) -> event
HotbarSelection.prepare(actions, inventory, slot, *, entity_runtime_id,
                        idempotency_key, deadline) -> HotbarSelection
selection.record() -> dict
async selection.dispatch(*, observer_packets, journal, output,
                         expected_epoch=None, observer_session=None,
                         timeout=5, max_events=1000, max_bytes=8388608) -> dict
inspect_selection_recording(source) -> dict

前两个函数在 neomega_runtime.item_actions，其余在 neomega_runtime.selection_recipe。equipment_matches 检查新的 sequence、包 31、neomega-json-v1 和匹配的 runtime ID、槽位、stack ID；它本身不替你维护完整流的连接与缺口边界。wait_equipment 消费调用方拥有的有序流，超时不触发发送；调用方在 finally 关闭流，禁止多个协程争读同一流。

HotbarSelection 将独立观察与一次发送组合：先打开 observer_packets.open([31], text_profile='selected_only')，清空旧事件，持久化 intent，再设定 sequence 边界并发送，直到同时拿到 dispatched 回执和匹配入站观察。observer_packets 使用 SDK 的 Packets 兼容接口；观察者可以是另一机器人，写入者不保证收到自己的 MobEquipment 回显。

output 是调用方打开的私有文本文件对象，写入有界 JSONL 录制；文件关闭由调用方负责，观察流由 recipe 关闭。expected_epoch 可固定观察连接；observer_session 未填时使用观察上下文的 operations session。流缺口、溢出、重连、日志失败或截断都会阻止 observed=True。

python
from neomega_runtime.selection_recipe import HotbarSelection

async def select_and_observe(ctx, inventory, slot, runtime_id, *, control,
                             observer_packets, journal, output, key, deadline):
    async with control:
        selection = HotbarSelection.prepare(
            ctx.item_actions, inventory, slot,
            entity_runtime_id=runtime_id, idempotency_key=key, deadline=deadline,
        )
        try:
            result = await selection.dispatch(
                observer_packets=observer_packets, journal=journal, output=output,
            )
            await journal(selection.record())  # 保存关闭流后确认的 observed。
            return result
        finally:
            state = selection.record()
            action = selection.action
            if action.started and (action.receipt is None or
                                   action.receipt.get('state') == 'unknown'):
                control.block(state)
            await journal(state)

record() 包含 intent、operation_id、receipt、started、observed、dispatched；成功返回还带 recording。超时或异常时保留同一 selection/action，重建 recipe 不属于恢复。

已有录制可以完全离线检查：

sh
python -m neomega_runtime.selection_recipe selection.jsonl

或 inspect_selection_recording(source)，其中 source 是录制路径/可读文本流。返回 complete、observed、dispatched、confirmed、issues；confirmed 需要完整观察和成功回执同时成立。CLI 确认成功退出 0，未确认 1，输入文件无效 2。录制证明历史观察，不是当前选槽状态，也不是重放指令。

对空使用后发送中性输入 ​

text
UseAirWithNeutralInput.prepare(actions, use, *, neutral_input,
                               idempotency_key, deadline)
async pair.dispatch(*, journal, timeout=10, poll_interval=.05,
                    before_dispatch=None) -> UseAirWithNeutralInput

从 neomega_runtime.item_use_input 导入。use 必须是未启动的 ItemActions.use_air()；中性输入是另一个独立 key 的动作。journal(payload) 在任何世界 IO 前保存两份完整 intent；before_dispatch(stage, intent_payload) 可在 'use' 和 'neutral' 各阶段重新检查观察。调用者持有共享锁并保存 pair.use、pair.neutral 的 ID 和回执。

neutral_input 必须完整含 Position、Pitch、Yaw、HeadYaw、MoveVector、InputMode、PlayMode、InteractionModel、Tick、Delta、InputFlags；可再同时提供 PitchRepeat、YawRepeat、InteractPitch、InteractYaw。MoveVector=[0,0]、Delta=[0,0,0]、InputFlags=[]；InputMode 为 1/2/3，PlayMode 为 0，InteractionModel 为 0/1/2。tick 必须来自真实原生时钟证据，不能以等待时间推算。

这是环境特定的两步组合，不是通用弩 API。它不选槽、不自动蓄力、不释放、不重试，也不在失败后补发剩余步骤。release() 本身是独立协议动作，不是所有弩发射流程必需的步骤。

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