Neomega

外观

持久状态、业务事务与插件服务 ​

需要跨重载保存计数，用 ctx.storage；需要把状态、事件确认和动作意图一起登记，用事务；需要给其他插件提供能力，用 ctx.services 和 @service。在线服务回执与持久业务回执有不同的生命周期。

状态、事件事务和在线服务由 Host 管理。选择性读取 keys 仅取指定键，不会退回全量读取；按下文选择状态读取、事务或跨插件服务。

Storage ​

ctx.storage 是当前安装的 JSON 状态空间，由 Host 管理。无需虚构 storage.* 操作权限，也不能用它访问其他插件的状态。可序列化值必须是有效 JSON，禁止 NaN、Infinity；读取和写入都会复制值，修改读回对象不会自动保存。

text
async get(key, default=None)
async set(key, value)
async delete(key)
async increment(key, amount=1, *, default=0)
async compare_and_set(key, expected, value)
async transaction(*, event=None, commit_id=None, keys=None)
async receipt(commit_id)
async submit(intent, *, timeout=10.0)

以上为 neomega_runtime.storage.Storage 实例方法，省略 self。

方法返回值和限制
get、set、deletekey 为 1–256 UTF-8 字节。get 返回值或 default；set/delete 成功返回 None。
incrementvalue、amount 必须为真正的 int，布尔值不算整数；返回实际保存的新值。缺失时从 default 开始。
compare_and_set当前 key 存在且 JSON 值等于 expected 时替换并返回 True，否则 False。缺失与 JSON null 不相等。
transaction返回 Transaction 状态快照，修改留在本地，必须显式 save()。
receipt / submit查询已保留的提交 / 显式提交不变的 CommitIntent。详见后文。

set/delete/increment/compare_and_set 只对明确 revision_conflict 做最多 4 次尝试，重试的是 SDK 自身的纯状态修改。未知回应和其他错误不重试，也不会执行调用者的任意业务回调。需要多键一致性、事件确认或游戏动作时，不要拆成多个便捷方法调用。

选择性读取 ​

transaction(keys=None) 读取全量快照；列表或元组最多 256 个互不重复的 key；keys=[] 只读取全局 revision。选择性快照不能读取未选择且未在本事务修改的 key，调用 tx.get 会抛 ValueError，不会将它默认为不存在。

选择性读取不会把 CAS 缩小到单 key：任何状态修改仍可能造成全局 revision_conflict。Storage.get 自身也使用选择性读取，因此旧 Host 不支持时应升级能力，不能把报错当作 key 缺失。

Transaction ​

text
get(key, default=None)
set(key, value)                 # 返回 self
delete(key)                    # 返回 self
increment(key, amount=1, *, default=0)
action(intent)                 # 返回 self
prepare()                      # 返回 CommitIntent
async save(*, timeout=10.0)     # 返回提交回执

Transaction 来自 await ctx.storage.transaction(...) 或 await event.transaction(...)。revision 是所读快照的全局版本；commit_id 默认生成一次 UUID hex，也可显式给定 1–128 字符的 ASCII 字母、数字、_、-。intent 在准备前为 None。

set/delete/increment 只修改快照；action(OperationIntent) 登记动作，要求动作 session 与状态快照 session 相同。prepare() 将修改、可选事件确认和动作封装成不可变 CommitIntent，最多 256 KiB，随后事务封闭，不得继续修改。重复 prepare() 返回原意图，不产生新 ID。

python
async def count_event(ctx, event):
    tx = await event.transaction()
    total = tx.increment("chat_count")
    tx.set("last_event_id", event.payload["event_id"])
    committed = await tx.save()
    return {"count": total, "revision": committed["revision"]}

这会把两个状态修改与当前事件 ACK 放在同一数据库提交中。需要发奖时，可以在保存前 tx.action(prepared_reward)，动作受理与状态一起提交；数据库提交成功不等于奖励动作已经完成，要逐个读取返回的 operation_ids。

Transaction 不自动重试冲突。明确版本冲突时重新读取并检查业务前提；结果不确定时保留原意图，不再运行计数或发奖逻辑。事务不是跨插件数据库事务，也不回滚外部网络请求和已发生的世界效果。

提交回执与恢复 ​

await tx.save() 和 await ctx.storage.submit(intent) 返回：

text
{commit_id: str, revision: int, duplicate: bool, operation_ids: list[str]}

await ctx.storage.receipt(commit_id) 返回保留的回执（带 found: true、commit_id、revision、operation_ids），未找到返回 None。该查询可以跨安装 generation 核对同一安装的旧提交，不重新发送动作；回执中的动作顺序对应 tx.action 登记顺序。

CommitIntent.method 为 state.commit 或 events.commit，.payload() 返回独立 JSON 字典，含 commit_id、expected_state_revision、mutations、actions，以及适用的 session 或事件投递标识。需跨进程恢复时，应在发起请求前可靠保留这些内容；单纯持有 Python 对象不具备重启持久性。

CommitUncertain.intent 保留失去回应的原请求。取消 Python 提交等待也可能携带 .intent，不能因此重新创建事务。receipt(...) is None 不证明旧请求永远不会提交；完整处理见错误页。

Business：保留业务结果 ​

text
async ctx.business.begin(business_id, request, *, keys=None)
async ctx.business.get(business_id)
async business_tx.commit(result)

begin 返回 BusinessTransaction，其 .state 是状态事务，.receipt 是已提交结果或 None。同一 business_id 使用规范化 JSON request 的摘要判定业务身份；同 ID 不同请求抛 BusinessConflict，不会重新执行业务。

python
async def credit_once(ctx, business_id, amount):
    if type(amount) is not int or amount <= 0:
        raise ValueError("amount 必须为正整数")
    tx = await ctx.business.begin(
        business_id, {"amount": amount}, keys=["balance"],
    )
    if tx.receipt is not None:
        return tx.receipt
    balance = tx.state.increment("balance", amount)
    return await tx.commit({"balance": balance})

该示例仅修改插件内部余额；没有向游戏发奖。commit(result) 原子保存业务回执与 .state 中的修改和动作，返回 business_id/request_digest/commit_id/state/result/operation_ids/revision。这里 state="committed" 只描述数据库提交。已存在 .receipt 时再次 commit 会抛 RuntimeError，应返回旧回执。

get(business_id) 找到时返回上述持久回执，否则 None。业务记录使用保留前缀 __neomega_business__/，不要自行修改。显式 keys 会追加一个业务记录 key，因此调用方最多提供 255 个不同业务数据 key。并发竞争仍遵循全局 CAS；冲突或不确定时不会自动重跑逻辑。需要跨插件持久交易时，由提供方通过服务公开 begin/get 业务协议，调用方保存自己的业务 ID，再查询同一结果。

Services：在线调用 ​

从 neomega_runtime.services 导入 Services、ServiceIntent 及异常；一般直接用 ctx.services。

text
prepare(service, arguments, *, idempotency_key, deadline, major=1)
async submit(intent, *, timeout=10)
async get(call_id, *, timeout=10)
async wait(call_id, *, timeout=5, poll_interval=0.05)
async call(service, arguments, *, major=1, timeout=5, idempotency_key=None)

service 必须是完整名 plugin.<插件ID>.<导出名>，不超过 256 字节；major 为正整数。调用方在 permissions.services 申请此完整名并获授权，提供方必须已经激活且导出对应 major。依赖提供方才能启动的插件，还要在 manifest 声明依赖。

prepare 无 IO，返回可通过 .payload() 复制原请求的 ServiceIntent；请求最多 64 KiB，key 与 deadline 格式同动作接口。submit 单次受理返回 call_id。get/wait 返回在线回执：

text
{call_id: str,
 state: "pending" | "replied" | "failed" | "timed_out" | "interrupted",
 result: JSON, error: str, rejection?: {code: str}}

wait 遇到非 pending 就返回，调用方自己检查状态。call 将 prepare、submit、wait 合并，成功返回 result；失败抛 ServiceFailed，业务拒绝抛 ServiceRejected。其 timeout 不超过 30 秒，未给 key 时生成随机键，适合一次简单请求；需要恢复证据时使用显式意图与回执，不依赖重复 call。

python
reply = await ctx.services.call(
    "plugin.example.echoprovider.echo", {"text": "hello"}, timeout=5,
)

此片段在已激活插件内执行，要求提供方公开下节 echo 服务，并给调用方授予对应完整服务权限。

在线调用边界 ​

在线回执保存在内存中，只供原调用者 generation 查询，不是持久交易日志。当前 Host 的单次 deadline 最多 30 秒，回执总保留上限 4096，提供方每代最多 16 个实际未结束调用；SDK handler 仍串行运行，不因此获得 16 路并发。

终态回执在原 deadline 之后一分钟且实际调用退出后才有资格回收；新提交触发回收。Host 重启、提供方替换或调用方退出可能中断回执；等待超时不停止已经受理的提供方执行，其效果可能稍后产生。不要重放不确定的有副作用调用；持久需求使用提供方的 Business 查询协议。

service 装饰器与服务提供方 ​

python
from neomega_runtime import Plugin, service, ServiceRejected

class EchoProvider(Plugin):
    @service("echo", major=1)
    async def echo(self, ctx, arguments):
        text = arguments["text"]
        if not text.strip():
            raise ServiceRejected("empty_text")
        return {"text": text}

if __name__ == "__main__":
    EchoProvider().run()

service(name, *, major=1, with_context=False) 只标记异步方法，不自动修改 manifest。导出名匹配 [a-z][a-z0-9_-]*，同名同 major 重复 handler 会拒绝启动。服务提供方 ID 为 example.echoprovider 时，对应 manifest 的字段片段为：

text
"exports": [{
  "name": "echo", "major": 1,
  "input_schema": {
    "type": "object", "additionalProperties": false,
    "required": ["text"], "properties": {"text": {"type": "string"}}
  },
  "output_schema": {
    "type": "object", "additionalProperties": false,
    "required": ["text"], "properties": {"text": {"type": "string"}}
  }
}]

调用方声明 "services": ["plugin.example.echoprovider.echo"]。强依赖字段可写 "dependencies": [{"id": "example.echoprovider", "major": 1, "min_version": "1.0.0"}]；完整 manifest 还需版本、入口、权限等字段，见权限。Host 校验输入与输出 schema；只加装饰器而未声明导出不会开放服务。

ServiceRejected(code="business_rejected", *, receipt=None) 表示预期业务拒绝，code 匹配 [a-z][a-z0-9_]{0,63}，不会因为这次拒绝关闭提供方。其他未处理异常可能使 Worker 失败。服务控制回调串行，慢回调可能阻塞维护或停止，不要在其中永久等待。

可信调用上下文 ​

按调用者隔离对象、余额或配额时，用 with_context=True 显式接收第三个处理器参数。以下片段只展示读取可信身份；实际服务仍需在 manifest 声明对应导出及输入、输出 schema：

python
from neomega_runtime import Plugin, ServiceCallContext, service


class OwnerProvider(Plugin):
    @service("identity", with_context=True)
    async def identity(self, ctx, arguments, call: ServiceCallContext):
        return {
            "installation_id": call.installation_id,
            "generation": call.generation,
        }

ServiceCallContext 是不可变 dataclass，字段为 installation_id: str、generation: int、call_id: str、accepted_at: datetime 和 deadline: datetime；时间字段带 UTC 时区。身份和时间来自 Host envelope，与调用方提交的 arguments 分开。accepted_at 不随排队刷新，租约应从它计算，并在后续动作前重新检查期限。

稳定 installation 用于持久业务归属，generation 用于区分当次运行；记录请求时保留二者，不能因为 caller 重载就把同一请求当成新业务。需要跨代际认领对象时，由服务明确制定规则。

原有 (ctx, arguments) 处理器继续可用。需要上下文的处理器在 Host 未提供上下文时收到 service_context_required 拒绝，不会执行；有效 envelope 的 deadline 已过期时，新旧处理器均在入口拒绝为 service_deadline_expired。格式损坏的 envelope 是协议错误，不能回退到 arguments.owner。

provider 的 ctx 仍属于 provider，动作使用它自己的权限。可信上下文不委托 caller 权限，也不保证已排队或已开始的 handler 会在 caller 重载、退出或等待超时后立即撤销。在线回执仍只保存在内存；共享能力需要自行保存请求编号、归属、目标状态和持久业务回执。持久接纳、对象 revision、TTL 与 unknown 的组合用法见共享 provider。

调用异常 ​

异常应保留的证据
ServiceUncertain.intent；受理不确定，不新建 key。
ServiceWaitTimeout.call_id；提供方可能继续执行。
ServiceFailed.receipt、.call_id；读取原状态。
ServiceRejected.code，调用端还有 .receipt、.call_id；按业务规则处理。
BusinessConflict / CommitUncertain请求身份冲突 / 原 .intent；不得把二者合并成通用重试。

保留首次业务意图 ​

SDK 提供 ctx.business.once(business_key, prepare, *, keys=None)，用于配套 Host 的原生 Bus service 回调。它在当前服务安装的存储中，以可信调用者 installation 与 business_key划分记录；调用者身份来自调用上下文，不能由请求字段冒充。

首次提交将业务状态、返回结果与至少一个完整子动作意图一起持久保存。prepare(tx) 必须是同步函数，只在事务内准备状态和动作，不执行事务外副作用。相同调用者和 key 的重复请求读取原结果及 operation IDs，不重新准备动作；member、major、schema digest 或完整参数不同则报 request_conflict，不能借同一 key 改订单。同一业务记录的并发候选可能各自执行 prepare，但提交至多保留一个胜者，此类冲突者读取胜者；若 revision 冲突来自其他事务且不存在这条业务记录，则原样报告冲突。

保留的是原完整请求、带 session 的子动作意图和提交回执；再次查询不能换成新的 deadline、动作或幂等键。accepted 仅表示意图已接纳，不能宣称物品已发放；必须继续查原 operation ID，unknown 不自动重放。完整使用见计数触发购买，签名见 business.once API。

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