R-00374 r3: pace chat ticks so Room Delta emits chat.event

.spec/knowledge/features/rust-entity-chat-host.md

@@ -19,13 +19,16 @@ ADR-056：Rust 宿主是接力交付面，只托管与传输。Room 世界是 Ru
 - **会话表**：只保存 `connection ↔ Runtime 绑定句柄` 与 `sess-*` 会话号。`NetEntityId` 由 Runtime 身份表发号（32 位小写 hex）。
 - **Runtime 消费**：CoreCLR `entity-chat-host` 转发 `Admit` / `Disconnect` / `Rebind` / `Expire` / `QueryAttribute` / `BuildFullSnapshot` / `BuildDelta` / `CapturePersist` / `RestorePersist`。恢复路径不 `Admit`、不新建 Active 绑定。
 - **定时**：host-runtime 是 NativeCore ABI 适配层。五分钟断线保留走 `wallClock` one-shot；Tick 走 `tickFrame` repeating。删除 `expire_due` 轮询。
-- **Room 网线**：loopback WebSocket。准入/重连发送 Runtime `BuildFullSnapshot`（含 `stateBlocks`）；每 Tick 把 `BuildDelta` 字节广播给本 Room 连接；顶号先发 `ConnectionSuperseded` 再关旧连接。
+- **Room 网线**：loopback WebSocket。准入/重连发送 Runtime `BuildFullSnapshot`（含 `stateBlocks`）；每 Tick 把 `BuildDelta` 字节广播给本 Room 连接；顶号先发 `ConnectionSuperseded` 再关旧连接。S8 证据 `connectionSupersededReceived` 只来自旧 `RoomClient` 收帧，不得用宿主 `takeover` 布尔冒充。
+- **Tick 分批**：Runtime `ChatCommandRuntime.RunTick` 经 `ChatIngressWorld` 默认 `EcsBudget.MaxChangeEntries=128`。每条 `chat.input` 写两个 ChatComponent 字段，单 Tick 最多 64 条；超过则 `Command reservation budget exceeded`、Runtime `_faulted`、`BuildDelta` 为 `changedBlocks:[]`。宿主按 `MAX_CHAT_INPUTS_PER_TICK` 穿插 `tickFrame`，不自建第二份事件队列。
+- **Persist**：`CapturePersist` / `RestorePersist` 走 Runtime 公开面（`RestorePersist` 的第二参是 `ReadOnlyMemory<byte>`）。默认 `MaxSnapshotBytes=4096` 只能装下约 6 个聊天实体；101 实体 Capture 为 Retryable 时不得把 `restoredWindow: 0` / `processB=null` 写成 S7 ok。
 - **发现**：外部产物经 `LUMIO_*` 环境变量与仓根相对路径；缺失即 BLOCKED，不硬编码开发机绝对路径。
-- **复跑**：`lumio-entity-chat-replay` 两轮；`manifest.conclusion=SUCCESS` 只在 Game `verify-evidence.mjs` oracle 通过之后写。
+- **复跑**：`lumio-entity-chat-replay` 两轮；`manifest.conclusion=SUCCESS` 只在 Game `verify-evidence.mjs` oracle 通过之后写。`--restore-snapshot` 供 S7 进程 B 单独启 CLR 恢复。
 
 ## 待解决
 
 - 完整 101 实体 acceptance 依赖 Runtime / NativeCore / Game 产物路径；缺失时测试以 BLOCKED 失败而非跳过。
+- Runtime `ChatIngressWorld.Create` 默认预算装不下 101 实体 Persist；S7 跨进程恢复待 Runtime 放大 `MaxSnapshotBytes`。
 - `mvp-host/` 仍冻结，归 N-13。
 
 ## 相关

entity-chat-host/src/Lumio.Server.EntityChat.HostEntry/HostEntry.cs

@@ -348,13 +348,36 @@ private static (int, byte[]) Tick(JsonElement root)
         }
 
         object result = ChatType!.GetMethod("RunTick")!.Invoke(Chat, new object[] { tickId })!;
-        ulong applied = Convert.ToUInt64(result.GetType().GetProperty("AppliedTick")!.GetValue(result)!, CultureInfo.InvariantCulture);
-        ulong revision = Convert.ToUInt64(result.GetType().GetProperty("Revision")!.GetValue(result)!, CultureInfo.InvariantCulture);
+        Type tickType = result.GetType();
+        ulong applied = Convert.ToUInt64(tickType.GetProperty("AppliedTick")!.GetValue(result)!, CultureInfo.InvariantCulture);
+        ulong revision = Convert.ToUInt64(tickType.GetProperty("Revision")!.GetValue(result)!, CultureInfo.InvariantCulture);
+        int eventCount = 0;
+        if (tickType.GetProperty("Events")!.GetValue(result) is System.Collections.ICollection events)
+        {
+            eventCount = events.Count;
+        }
+
+        string? failed = null;
+        if (tickType.GetProperty("Results")!.GetValue(result) is System.Collections.IEnumerable rows)
+        {
+            foreach (object row in rows)
+            {
+                bool succeeded = Convert.ToBoolean(row.GetType().GetProperty("Succeeded")!.GetValue(row)!, CultureInfo.InvariantCulture);
+                if (!succeeded)
+                {
+                    failed = row.GetType().GetProperty("Code")!.GetValue(row) as string;
+                    break;
+                }
+            }
+        }
+
         return (EntrySuccess, Json(new Dictionary<string, object?>
         {
-            ["ok"] = true,
+            ["ok"] = failed is null,
             ["appliedTick"] = applied,
             ["revision"] = revision,
+            ["eventCount"] = eventCount,
+            ["code"] = failed,
         }));
     }
 
@@ -398,23 +421,32 @@ private static (int, byte[]) BuildDelta(JsonElement root)
 
     private static (int, byte[]) Persist(JsonElement root)
     {
-        if (Chat is null || PersistType is null)
+        if (Chat is null)
         {
             return (EntrySuccess, Fail("invalid_request"));
         }
 
         object? world = GetChatWorld();
-        if (world is null)
+        Type? persistType = PersistPipelineOf(world);
+        if (world is null || persistType is null)
+        {
+            return (EntrySuccess, Fail("runtime_failure"));
+        }
+
+        MethodInfo? capture = FindStatic(
+            persistType,
+            "CapturePersist",
+            static parameters => parameters.Length == 2 && parameters[1].ParameterType.IsByRef);
+        if (capture is null)
         {
             return (EntrySuccess, Fail("runtime_failure"));
         }
 
         object[] args = { world, null! };
-        object result = PersistType.GetMethod("CapturePersist", new[] { world.GetType(), typeof(byte[]).MakeByRefType() })!
-            .Invoke(null, args)!;
+        object result = capture.Invoke(null, args)!;
         bool accepted = Convert.ToInt32(result.GetType().GetProperty("Status")!.GetValue(result)!, CultureInfo.InvariantCulture) == 0;
         byte[]? bytes = args[1] as byte[];
-        if (!accepted || bytes is null)
+        if (!accepted || bytes is null || bytes.Length == 0)
         {
             return (EntrySuccess, Fail("runtime_failure"));
         }
@@ -428,24 +460,55 @@ private static (int, byte[]) Persist(JsonElement root)
 
     private static (int, byte[]) Restore(JsonElement root)
     {
-        if (Chat is null || PersistType is null || !TryReadString(root, "bytesHex", out string? hex) || hex is null)
+        if (Chat is null || !TryReadString(root, "bytesHex", out string? hex) || hex is null)
         {
             return (EntrySuccess, Fail("invalid_request"));
         }
 
         object? world = GetChatWorld();
-        if (world is null)
+        Type? persistType = PersistPipelineOf(world);
+        if (world is null || persistType is null)
+        {
+            return (EntrySuccess, Fail("runtime_failure"));
+        }
+
+        MethodInfo? restore = FindStatic(
+            persistType,
+            "RestorePersist",
+            static parameters =>
+                parameters.Length == 2
+                && parameters[1].ParameterType.IsGenericType
+                && parameters[1].ParameterType.GetGenericTypeDefinition() == typeof(ReadOnlyMemory<>));
+        if (restore is null)
         {
             return (EntrySuccess, Fail("runtime_failure"));
         }
 
         byte[] bytes = Convert.FromHexString(hex);
-        object result = PersistType.GetMethod("RestorePersist", new[] { world.GetType(), typeof(byte[]) })!
-            .Invoke(null, new object[] { world, bytes })!;
+        ReadOnlyMemory<byte> memory = bytes;
+        object result = restore.Invoke(null, new object[] { world, memory })!;
         bool accepted = Convert.ToInt32(result.GetType().GetProperty("Status")!.GetValue(result)!, CultureInfo.InvariantCulture) == 0;
         return (EntrySuccess, accepted ? Ok() : Fail("runtime_failure"));
     }
 
+    private static Type? PersistPipelineOf(object? world)
+    {
+        return world?.GetType().Assembly.GetType("Lumio.GameRuntime.Ecs.EcsPersistSnapshotPipeline") ?? PersistType;
+    }
+
+    private static MethodInfo? FindStatic(Type type, string name, Func<ParameterInfo[], bool> match)
+    {
+        foreach (MethodInfo method in type.GetMethods(BindingFlags.Public | BindingFlags.Static))
+        {
+            if (method.Name == name && match(method.GetParameters()))
+            {
+                return method;
+            }
+        }
+
+        return null;
+    }
+
     private static object? GetChatWorld()
     {
         if (Chat is null || ChatType is null)

modules/process/src/entity_chat/clr.rs

@@ -285,19 +285,9 @@ impl RuntimeSurface for ClrGameplay {
     }
 
     fn run_tick(&mut self, room_id: &str, tick_id: u64) -> RuntimeTick {
-        let Ok(value) = self.call(json!({ "op": "tick", "roomId": room_id, "tickId": tick_id }))
-        else {
-            return RuntimeTick {
-                applied_tick: 0,
-                revision: 0,
-            };
-        };
-        RuntimeTick {
-            applied_tick: value
-                .get("appliedTick")
-                .and_then(Value::as_u64)
-                .unwrap_or(0),
-            revision: value.get("revision").and_then(Value::as_u64).unwrap_or(0),
+        match self.call(json!({ "op": "tick", "roomId": room_id, "tickId": tick_id })) {
+            Ok(value) => tick_from_hostentry_json(value),
+            Err(_) => RuntimeTick::failed("runtime_failure"),
         }
     }
 
@@ -392,6 +382,33 @@ fn hex_lower(bytes: &[u8]) -> String {
     out
 }
 
+/// HostEntry `tick` JSON: `ok:false` is a failed tick even if appliedTick >= 1.
+pub(crate) fn tick_from_hostentry_json(value: Value) -> RuntimeTick {
+    let event_count = value.get("eventCount").and_then(Value::as_u64).unwrap_or(0);
+    let code = value
+        .get("code")
+        .and_then(Value::as_str)
+        .filter(|code| !code.is_empty())
+        .map(str::to_owned);
+    if value.get("ok").and_then(Value::as_bool) != Some(true) {
+        return RuntimeTick {
+            applied_tick: 0,
+            revision: value.get("revision").and_then(Value::as_u64).unwrap_or(0),
+            ok: false,
+            event_count: 0,
+            code: code.or_else(|| Some("runtime_failure".to_owned())),
+        };
+    }
+    RuntimeTick::committed(
+        value
+            .get("appliedTick")
+            .and_then(Value::as_u64)
+            .unwrap_or(0),
+        value.get("revision").and_then(Value::as_u64).unwrap_or(0),
+        event_count,
+    )
+}
+
 /// Maps a Runtime `build_full_snapshot` JSON envelope to wire bytes.
 /// Missing/failed Runtime responses must not become a host-minted FullSnapshot.
 pub(crate) fn full_snapshot_bytes_from_runtime(response: Option<Value>) -> Vec<u8> {
@@ -432,6 +449,21 @@ mod tests {
         );
     }
 
+    #[test]
+    fn budget_fault_tick_is_not_success_even_when_applied_tick_is_one() {
+        let tick = tick_from_hostentry_json(json!({
+            "ok": false,
+            "appliedTick": 1,
+            "revision": 1,
+            "eventCount": 0,
+            "code": "runtime_failure"
+        }));
+        assert!(!tick.ok);
+        assert_eq!(tick.applied_tick, 0);
+        assert_eq!(tick.event_count, 0);
+        assert_eq!(tick.code.as_deref(), Some("runtime_failure"));
+    }
+
     #[test]
     fn runtime_json_is_forwarded_unchanged() {
         let runtime = r#"{"messageType":"FullSnapshot","tickId":1,"revision":1,"stateBlocks":[{"mappingId":"entity.identity","payload":"aa","payloadSha256":"bb"}]}"#;

modules/process/src/entity_chat/host.rs

@@ -593,18 +593,15 @@ impl Inner {
     fn run_tick(&mut self, room_id: &str) -> RuntimeTick {
         self.tick_id = self.tick_id.saturating_add(1);
         let Ok(fired) = self.kernel.advance_tick_frame(self.tick_id) else {
-            return RuntimeTick {
-                applied_tick: 0,
-                revision: 0,
-            };
+            return RuntimeTick::failed("runtime_failure");
         };
         if !fired.iter().any(|row| row.dispatch_id == DISPATCH_TICK) {
-            return RuntimeTick {
-                applied_tick: 0,
-                revision: 0,
-            };
+            return RuntimeTick::failed("runtime_failure");
         }
         let tick = self.runtime.run_tick(room_id, self.tick_id);
+        if !tick.ok {
+            return tick;
+        }
         let frames = self
             .runtime
             .build_delta(room_id, tick.applied_tick, tick.revision);

modules/process/src/entity_chat/mod.rs

@@ -57,6 +57,9 @@ pub const TEST_PASSWORD: &str = "123456";
 pub const ADMISSION_KEY_ID: u8 = 1;
 pub const RECONNECT_WINDOW_MS: u64 = 300_000;
 pub const INGRESS_QUEUE_PER_CONNECTION: usize = 64;
+/// Runtime `ChatIngressWorld` default `MaxChangeEntries` is 128; each chat.input
+/// commits two ChatComponent fields, so one `RunTick` can take at most 64 chats.
+pub const MAX_CHAT_INPUTS_PER_TICK: usize = 64;
 pub const BOT_COUNT: u32 = 100;
 
 /// Formats `Bot01`…`Bot100`.

modules/process/src/entity_chat/runtime.rs

@@ -163,10 +163,37 @@ impl QueryResult {
 }
 
 /// Tick result used only to know which tick/revision to request on the wire.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
 pub struct RuntimeTick {
     pub applied_tick: u64,
     pub revision: u64,
+    pub ok: bool,
+    pub event_count: u64,
+    pub code: Option<String>,
+}
+
+impl RuntimeTick {
+    #[must_use]
+    pub fn failed(code: &str) -> Self {
+        Self {
+            applied_tick: 0,
+            revision: 0,
+            ok: false,
+            event_count: 0,
+            code: Some(code.to_owned()),
+        }
+    }
+
+    #[must_use]
+    pub fn committed(applied_tick: u64, revision: u64, event_count: u64) -> Self {
+        Self {
+            applied_tick,
+            revision,
+            ok: true,
+            event_count,
+            code: None,
+        }
+    }
 }
 
 /// Chat admit/apply outcome.