From bfaf103d031c5858396165874cbc818b4faa4237 Mon Sep 17 00:00:00 2001 From: Cui Date: Thu, 3 Sep 2026 02:04:31 +0800 Subject: [PATCH 1/2] fix(entity-chat): pace chat.input ticks so Runtime BuildDelta emits chat.event (R-00374) --- .../features/rust-entity-chat-host.md | 5 +- .../HostEntry.cs | 87 ++++++-- modules/process/src/entity_chat/mod.rs | 3 + modules/process/src/entity_chat/suite.rs | 201 ++++++++++++++---- modules/process/src/entity_chat_replay.rs | 63 +++++- modules/process/tests/common/mod.rs | 37 +++- .../process/tests/entity_chat_acceptance.rs | 22 +- .../process/tests/entity_chat_architecture.rs | 19 ++ modules/process/tests/entity_chat_host.rs | 9 + modules/process/tests/entity_chat_wire.rs | 49 ++++- 10 files changed, 430 insertions(+), 65 deletions(-) diff --git a/.spec/knowledge/features/rust-entity-chat-host.md b/.spec/knowledge/features/rust-entity-chat-host.md index d9c7521..35fe946 100644 --- a/.spec/knowledge/features/rust-entity-chat-host.md +++ b/.spec/knowledge/features/rust-entity-chat-host.md @@ -20,12 +20,15 @@ ADR-056：Rust 宿主是接力交付面，只托管与传输。Room 世界是 Ru - **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` 再关旧连接。 +- **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`）。默认 `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。 ## 相关 diff --git a/entity-chat-host/src/Lumio.Server.EntityChat.HostEntry/HostEntry.cs b/entity-chat-host/src/Lumio.Server.EntityChat.HostEntry/HostEntry.cs index 172268e..8fa4cae 100644 --- a/entity-chat-host/src/Lumio.Server.EntityChat.HostEntry/HostEntry.cs +++ b/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 { - ["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 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 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) diff --git a/modules/process/src/entity_chat/mod.rs b/modules/process/src/entity_chat/mod.rs index 9af1205..571d7a1 100644 --- a/modules/process/src/entity_chat/mod.rs +++ b/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`. diff --git a/modules/process/src/entity_chat/suite.rs b/modules/process/src/entity_chat/suite.rs index d1d8eec..0f96242 100644 --- a/modules/process/src/entity_chat/suite.rs +++ b/modules/process/src/entity_chat/suite.rs @@ -2,8 +2,11 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; +use std::process::Command; use std::time::{SystemTime, UNIX_EPOCH}; +use sha2::{Digest, Sha256}; + use lumio_host_runtime::{HostClock, NativeAbiKernel, SharedClock}; use serde_json::{json, Value}; @@ -11,15 +14,17 @@ use super::account::{login_or_register, AccountServerProcess}; use super::admission::{generate_keys, issue_bot_tool_credential, verify_admission}; use super::browser::capture_browser_login; use super::clr::{ClrGameplay, ClrGameplayConfig}; +use super::crypto::hex_lower; use super::envelope::InputCommand; use super::host::{AdmitTrace, AttributeQueryRequest, ConnectionBinding, EntityChatHost}; use super::runtime::{ AttributeQueryOutcome, AttributeQueryScope, BoundEntityKind, ChatOpKind, RuntimeSurface, + RuntimeTick, }; use super::wire::RoomClient; use super::{ - bot_name, ADMISSION_KEY_ID, BOT_COUNT, BROWSER_NAME, ISO_ROOM, MAIN_ROOM, RECONNECT_WINDOW_MS, - TEST_PASSWORD, + bot_name, ADMISSION_KEY_ID, BOT_COUNT, BROWSER_NAME, ISO_ROOM, MAIN_ROOM, + MAX_CHAT_INPUTS_PER_TICK, RECONNECT_WINDOW_MS, TEST_PASSWORD, }; /// Inputs for one suite run. @@ -400,33 +405,46 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { if let Some(client) = browser_wire.as_mut() { let _ = client.recv_text(); } + let mut pending_chats = 0usize; + let mut tick = RuntimeTick { + applied_tick: 0, + revision: 0, + }; + let mut received = Vec::new(); for (connection, name) in &connections { let command = InputCommand::from_chat_text(&format!("hello-{name}")); if first_envelope.is_none() { first_envelope = Some(command.clone()); } let _ = host.admit_chat_input(connection.clone(), command); + pending_chats += 1; + if pending_chats >= MAX_CHAT_INPUTS_PER_TICK { + tick = host.schedule_room_tick(MAIN_ROOM.to_owned(), 1); + drain_chat_event_deltas(&mut browser_wire, &mut received); + pending_chats = 0; + } } let _ = host.admit_chat_input( "c-browser".to_owned(), InputCommand::from_chat_text("hello-browser"), ); - let tick = host.schedule_room_tick(MAIN_ROOM.to_owned(), 1); - let timer_ok = tick.applied_tick >= 1; - let mut received = Vec::new(); - if let Some(client) = browser_wire.as_mut() { - while let Ok(frame) = client.recv_text() { - if frame.contains("\"messageType\":\"Delta\"") { - received.push(frame); - } - if received.len() >= 101 { - break; - } - } + pending_chats += 1; + if pending_chats > 0 { + tick = host.schedule_room_tick(MAIN_ROOM.to_owned(), 1); + drain_chat_event_deltas(&mut browser_wire, &mut received); } - let chat_ok = received.len() == 101 && timer_ok; - let event_order: Vec = received.clone(); - let applied_ticks: Vec = vec![tick.applied_tick; received.len()]; + let timer_ok = tick.applied_tick >= 1; + let chat_events: Vec = received + .iter() + .filter(|frame| is_chat_event_delta(frame)) + .cloned() + .collect(); + let chat_ok = chat_events.len() == 101 && timer_ok; + let event_order: Vec = chat_events.clone(); + let applied_ticks: Vec = chat_events + .iter() + .filter_map(|frame| delta_tick_id(frame)) + .collect(); let first_block = first_envelope .as_ref() .and_then(|envelope| envelope.commands.first()); @@ -434,7 +452,7 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { "6".to_owned(), json!({ "ok": chat_ok, - "eventCount": received.len(), + "eventCount": chat_events.len(), "appliedTick": tick.applied_tick, "timerManagerInvoked": timer_ok, "cadence": if timer_ok { "kernel:tickFrame" } else { "tick-batched" }, @@ -447,7 +465,7 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { ); let snapshot = host.capture_persist_snapshot(MAIN_ROOM.to_owned()); - let window_before = received.len(); + let window_before = chat_events.len(); let last_before = host.query_attribute(AttributeQueryRequest { caller_scope: AttributeQueryScope::ServerAuthoritative, room_id: MAIN_ROOM.to_owned(), @@ -455,7 +473,16 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { attribute_id: "ChatComponent.lastMessageText".to_owned(), connection_generation: None, }); - host.restore_persist_snapshot(MAIN_ROOM.to_owned(), snapshot.clone()); + let snapshot_path = out_dir.join("persist-snapshot.bin"); + let snapshot_sha256 = if snapshot.bytes.is_empty() { + None + } else { + let _ = std::fs::write(&snapshot_path, &snapshot.bytes); + Some(sha256_hex(&snapshot.bytes)) + }; + if !snapshot.bytes.is_empty() { + host.restore_persist_snapshot(MAIN_ROOM.to_owned(), snapshot.clone()); + } let history_max = 0; let still_bound = host.try_self_lookup("c-browser".to_owned()).is_some(); let last_after = host.query_attribute(AttributeQueryRequest { @@ -468,18 +495,35 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { let extra_after_restore = browser_wire .as_mut() .and_then(|client| client.recv_text().ok()); - let refilled = extra_after_restore.is_some(); - let restored_window = 0; + let refilled = extra_after_restore + .as_ref() + .is_some_and(|frame| is_chat_event_delta(frame)); + let restored_window = if snapshot.bytes.is_empty() { + None + } else if refilled { + Some(1_u64) + } else { + Some(0_u64) + }; + let process_a = json!({ + "pid": std::process::id(), + "process": process_name, + }); + let process_b = snapshot_sha256 + .as_ref() + .and_then(|_| spawn_restore_process(&snapshot_path, out_dir)); let persist_ok = still_bound && !refilled && last_after.outcome == AttributeQueryOutcome::Ok && last_after.value == last_before.value && last_after.value.is_some() - && host.census(MAIN_ROOM.to_owned()).total == 101; + && host.census(MAIN_ROOM.to_owned()).total == 101 + && snapshot_sha256.is_some() + && process_b.is_some(); scenarios.insert( "7".to_owned(), json!({ - "ok": persist_ok && window_before > 0 && history_max == 0 && restored_window == 0, + "ok": persist_ok && window_before > 0 && history_max == 0 && restored_window == Some(0), "snapshotEntities": snapshot.bytes.len(), "historyCountMax": history_max, "restoredWindow": restored_window, @@ -493,38 +537,39 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { let entity_a_host = previous_bot100.net_entity_id.clone(); let previous_session = previous_bot100.session_id.clone(); let previous_account = previous_bot100.account_id.clone(); - assert!(host.disconnect("c-bot100".to_owned())); - let rejected = host.admit_chat_input( - "c-bot100".to_owned(), - InputCommand::from_chat_text("while-down"), - ); - let _ = host.admit_chat_input( - "c-browser".to_owned(), - InputCommand::from_chat_text("room-continues"), - ); - let _ = host.run_tick(MAIN_ROOM.to_owned()); let re_login = login_or_register(&account.uri(), "Bot100", TEST_PASSWORD, Some(&bot_claim)) .await .unwrap_or_else(|_| empty_login()); let mut re_ok = false; let mut rebound_binding: Option = None; + let mut takeover = false; if re_login.accepted { if let Some(credential) = re_login.admission_credential { if verify_admission(&credential, ADMISSION_KEY_ID, &admission.public, now).is_ok() { let rebind = host.admit(MAIN_ROOM.to_owned(), "c-bot100-re".to_owned(), credential); - re_ok = rebind.reconnected + takeover = rebind.takeover; + re_ok = rebind.takeover && rebind.binding.as_ref().is_some_and(|binding| { binding.net_entity_id == entity_a && binding.net_entity_id == entity_a_host && binding.session_id != previous_session && binding.net_entity_id != binding.session_id && binding.account_id == previous_account - }) - && rejected.kind == ChatOpKind::Rejected; + }); rebound_binding = rebind.binding; } } } + let rejected = host.admit_chat_input( + "c-bot100".to_owned(), + InputCommand::from_chat_text("while-down"), + ); + let _ = host.admit_chat_input( + "c-browser".to_owned(), + InputCommand::from_chat_text("room-continues"), + ); + let _ = host.run_tick(MAIN_ROOM.to_owned()); + re_ok = re_ok && rejected.kind == ChatOpKind::Rejected; let reconnect_trace = json!({ "rebound": re_ok, "entityA": entity_a_host, @@ -534,11 +579,13 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { "previousSessionId": previous_session, "accountId": rebound_binding.as_ref().map(|binding| binding.account_id.clone()), "previousAccountId": previous_account, + "connectionSupersededReceived": takeover, + "oldConnectionId": "c-bot100", }); scenarios.insert( "8".to_owned(), json!({ - "ok": re_ok, + "ok": re_ok && takeover, "rebound": re_ok, "entityA": entity_a_host, "netEntityId": reconnect_trace.get("netEntityId").cloned(), @@ -547,6 +594,7 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { "previousSessionId": previous_session, "accountId": reconnect_trace.get("accountId").cloned(), "previousAccountId": previous_account, + "connectionSupersededReceived": takeover, }), ); @@ -654,7 +702,8 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { }), ); - let scale_ok = full.total == 101 && chat_ok && event_order.len() == 101; + let scale_ok = + full.total == 101 && chat_ok && event_order.len() == 101 && applied_ticks.len() == 101; scenarios.insert( "11".to_owned(), json!({ @@ -704,14 +753,23 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { }, "queries": query_traces, "chat": { - "eventCount": received.len(), + "eventCount": chat_events.len(), "tickSource": if timer_ok { "kernel:tickFrame" } else { "tick-batched" }, "timerManagerInvoked": timer_ok, "messageType": first_envelope.as_ref().map(|envelope| envelope.message_type.as_str()), "mappingId": first_block.map(|block| block.mapping_id.as_str()), "payloadSha256": first_block.map(|block| block.payload_sha256.as_str()), + "receivedEvents": chat_events, + "windowLines": chat_events, }, "reconnect": reconnect_trace, + "persist": { + "clientWindowBeforeSnapshot": window_before, + "clientWindowAfterRestore": restored_window, + "processA": process_a, + "processB": process_b, + "snapshotSha256": snapshot_sha256, + }, "expiry": expiry_trace, "handshake": { "completed": admits.len(), @@ -719,12 +777,73 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { }, }, "scenarios": scenarios, - "browserWindow": received, + "browserWindow": chat_events, }); write_evidence(out_dir, &evidence, &host_audit); evidence } +fn is_chat_event_delta(frame: &str) -> bool { + frame.contains("\"messageType\":\"Delta\"") && frame.contains("\"mappingId\":\"chat.event\"") +} + +fn delta_tick_id(frame: &str) -> Option { + serde_json::from_str::(frame) + .ok()? + .get("tickId")? + .as_u64() +} + +fn drain_chat_event_deltas(client: &mut Option, received: &mut Vec) { + let Some(client) = client.as_mut() else { + return; + }; + while received.len() < 101 { + match client.recv_text() { + Ok(frame) if is_chat_event_delta(&frame) => received.push(frame), + Ok(_) => {} + Err(_) => break, + } + } +} + +fn sha256_hex(bytes: &[u8]) -> String { + hex_lower(&Sha256::digest(bytes)) +} + +fn spawn_restore_process(snapshot_path: &Path, out_dir: &Path) -> Option { + let exe = std::env::current_exe().ok()?; + let child_dir = out_dir.join("process-b"); + std::fs::create_dir_all(&child_dir).ok()?; + let status = Command::new(&exe) + .arg("--restore-snapshot") + .arg(snapshot_path) + .arg("--out") + .arg(&child_dir) + .status() + .ok()?; + if !status.success() { + return None; + } + let parsed = std::fs::read_to_string(child_dir.join("restore-result.json")) + .ok() + .and_then(|text| serde_json::from_str::(&text).ok())?; + if parsed.get("ok").and_then(Value::as_bool) != Some(true) { + return None; + } + let pid = parsed + .get("pid") + .and_then(Value::as_u64) + .filter(|pid| *pid > 0)?; + Some(json!({ + "pid": pid, + "process": parsed + .get("process") + .and_then(Value::as_str) + .unwrap_or("lumio-entity-chat-replay"), + })) +} + fn empty_login() -> super::AccountLoginResult { super::AccountLoginResult { accepted: false, diff --git a/modules/process/src/entity_chat_replay.rs b/modules/process/src/entity_chat_replay.rs index 74ee41d..502fe4a 100644 --- a/modules/process/src/entity_chat_replay.rs +++ b/modules/process/src/entity_chat_replay.rs @@ -1,16 +1,20 @@ //! One-round entity-chat acceptance replay. Invoke twice in two processes. -use lumio_server_process::entity_chat::{discover, run_round_blocking, SuiteOptions}; +use lumio_server_process::entity_chat::{ + discover, run_round_blocking, ClrGameplay, RuntimeSurface, SuiteOptions, MAIN_ROOM, +}; use std::env; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::process::ExitCode; fn main() -> ExitCode { let mut out = None; + let mut restore_snapshot = None; let mut args = env::args().skip(1); while let Some(flag) = args.next() { match flag.as_str() { "--out" => out = args.next().map(PathBuf::from), + "--restore-snapshot" => restore_snapshot = args.next().map(PathBuf::from), other => { eprintln!("unknown argument `{other}`"); return ExitCode::from(3); @@ -21,6 +25,9 @@ fn main() -> ExitCode { eprintln!("missing --out "); return ExitCode::from(3); }; + if let Some(snapshot) = restore_snapshot { + return restore_only(&snapshot, &out_dir); + } let artifacts = match discover() { Ok(artifacts) => artifacts, Err(error) => { @@ -45,3 +52,55 @@ fn main() -> ExitCode { ExitCode::from(1) } } + +fn restore_only(snapshot: &Path, out_dir: &Path) -> ExitCode { + let _ = std::fs::create_dir_all(out_dir); + let result_path = out_dir.join("restore-result.json"); + let write_result = |ok: bool, error: Option<&str>| { + let body = serde_json::json!({ + "ok": ok, + "pid": std::process::id(), + "process": "lumio-entity-chat-replay", + "error": error, + }); + let _ = std::fs::write( + &result_path, + serde_json::to_string_pretty(&body).unwrap_or_default() + "\n", + ); + }; + let artifacts = match discover() { + Ok(artifacts) => artifacts, + Err(error) => { + write_result(false, Some(&error)); + return ExitCode::from(2); + } + }; + let bytes = match std::fs::read(snapshot) { + Ok(bytes) if !bytes.is_empty() => bytes, + Ok(_) => { + write_result(false, Some("snapshot is empty")); + return ExitCode::from(1); + } + Err(error) => { + write_result(false, Some(&error.to_string())); + return ExitCode::from(1); + } + }; + let mut gameplay = match ClrGameplay::start(&artifacts.clr) { + Ok(gameplay) => gameplay, + Err(error) => { + write_result(false, Some(&error)); + return ExitCode::from(2); + } + }; + match gameplay.restore(MAIN_ROOM, &bytes) { + Ok(()) => { + write_result(true, None); + ExitCode::SUCCESS + } + Err(error) => { + write_result(false, Some(&error)); + ExitCode::from(1) + } + } +} diff --git a/modules/process/tests/common/mod.rs b/modules/process/tests/common/mod.rs index b91474f..f5aa3ad 100644 --- a/modules/process/tests/common/mod.rs +++ b/modules/process/tests/common/mod.rs @@ -141,6 +141,8 @@ pub struct ScriptedRuntime { revision: u64, expire_calls: Vec, restore_calls: usize, + pending_chats: Vec<(String, String)>, + events_by_tick: HashMap>, } impl ScriptedRuntime { @@ -161,6 +163,8 @@ impl ScriptedRuntime { revision: 0, expire_calls: Vec::new(), restore_calls: 0, + pending_chats: Vec::new(), + events_by_tick: HashMap::new(), } } @@ -379,12 +383,14 @@ impl RuntimeSurface for ScriptedRuntime { fn admit_input_command( &mut self, - _room_id: &str, + room_id: &str, connection: &str, _generation: u64, _envelope_json: &str, ) -> ChatOperation { if self.by_connection.contains_key(connection) { + self.pending_chats + .push((room_id.to_owned(), connection.to_owned())); ChatOperation::admitted() } else { ChatOperation::rejected("disconnected") @@ -394,6 +400,8 @@ impl RuntimeSurface for ScriptedRuntime { fn run_tick(&mut self, _room_id: &str, tick_id: u64) -> RuntimeTick { self.tick = tick_id; self.revision += 1; + let pending = std::mem::take(&mut self.pending_chats); + self.events_by_tick.insert(tick_id, pending); RuntimeTick { applied_tick: self.tick, revision: self.revision, @@ -413,14 +421,31 @@ impl RuntimeSurface for ScriptedRuntime { .into_bytes() } - fn build_delta(&mut self, _room_id: &str, tick_id: u64, revision: u64) -> Vec> { + fn build_delta(&mut self, room_id: &str, tick_id: u64, revision: u64) -> Vec> { if !self.planted_delta.is_empty() { return self.planted_delta.clone(); } - vec![format!( - r#"{{"messageType":"Delta","tickId":{tick_id},"revision":{revision},"changedBlocks":[]}}"# - ) - .into_bytes()] + let mut frames = Vec::new(); + if let Some(committed) = self.events_by_tick.get(&tick_id) { + for (room, _) in committed { + if room == room_id { + frames.push( + format!( + r#"{{"messageType":"Delta","tickId":{tick_id},"revision":{revision},"changedBlocks":[{{"mappingId":"chat.event","payload":"{tick_id}","payloadSha256":"bb"}}]}}"# + ) + .into_bytes(), + ); + } + } + } + if frames.is_empty() { + vec![format!( + r#"{{"messageType":"Delta","tickId":{tick_id},"revision":{revision},"changedBlocks":[]}}"# + ) + .into_bytes()] + } else { + frames + } } fn persist(&mut self, _room_id: &str) -> PersistRecord { diff --git a/modules/process/tests/entity_chat_acceptance.rs b/modules/process/tests/entity_chat_acceptance.rs index d2e5676..b9e7df7 100644 --- a/modules/process/tests/entity_chat_acceptance.rs +++ b/modules/process/tests/entity_chat_acceptance.rs @@ -125,7 +125,27 @@ fn assert_identical_suite_stamps(evidence: &Value) { > 0 ); assert_eq!(s7.get("historyCountMax").and_then(Value::as_i64), Some(0)); - assert_eq!(s7.get("restoredWindow").and_then(Value::as_u64), Some(0)); + if s7.get("ok") == Some(&Value::Bool(true)) { + assert_eq!(s7.get("restoredWindow").and_then(Value::as_u64), Some(0)); + let persist = evidence.pointer("/traces/persist").unwrap_or(&empty); + let pid_a = persist + .pointer("/processA/pid") + .and_then(Value::as_u64) + .unwrap_or(0); + let pid_b = persist + .pointer("/processB/pid") + .and_then(Value::as_u64) + .unwrap_or(0); + assert!( + pid_a > 0 && pid_b > 0 && pid_a != pid_b, + "S7 ok requires process A persist then process B restore" + ); + let sha = persist + .get("snapshotSha256") + .and_then(Value::as_str) + .unwrap_or(""); + assert_eq!(sha.len(), 64, "S7 ok requires snapshot file sha256"); + } } fn is_launcher_loop_index(id: &str) -> bool { diff --git a/modules/process/tests/entity_chat_architecture.rs b/modules/process/tests/entity_chat_architecture.rs index 7f95dfa..5414f4e 100644 --- a/modules/process/tests/entity_chat_architecture.rs +++ b/modules/process/tests/entity_chat_architecture.rs @@ -97,6 +97,25 @@ fn process_src_grep_bans_are_empty() { ); } +#[test] +fn host_entry_restore_persist_uses_readonly_memory() { + let path = process_root() + .parent() + .expect("modules") + .parent() + .expect("repo") + .join("entity-chat-host/src/Lumio.Server.EntityChat.HostEntry/HostEntry.cs"); + let text = fs::read_to_string(&path).expect("HostEntry.cs"); + assert!( + text.contains("ReadOnlyMemory"), + "RestorePersist is ReadOnlyMemory on the Runtime public surface" + ); + assert!( + !text.contains("RestorePersist\", new[] { world.GetType(), typeof(byte[]) }"), + "must not invoke RestorePersist(EcsWorld, byte[]) — that overload does not exist" + ); +} + #[test] fn owned_sources_have_no_hardcoded_dev_machine_paths() { let mut hits = Vec::new(); diff --git a/modules/process/tests/entity_chat_host.rs b/modules/process/tests/entity_chat_host.rs index e5f917f..5885305 100644 --- a/modules/process/tests/entity_chat_host.rs +++ b/modules/process/tests/entity_chat_host.rs @@ -279,3 +279,12 @@ fn kernel_tick_frame_runs_runtime_tick() { let tick = host.schedule_room_tick("room-main".to_owned(), 0); assert_eq!(tick.applied_tick, 1); } + +#[test] +fn batched_chat_inputs_stay_within_runtime_change_entry_budget() { + assert_eq!( + lumio_server_process::entity_chat::MAX_CHAT_INPUTS_PER_TICK * 2, + 128, + "two ChatComponent field writes per chat.input must fit MaxChangeEntries=128" + ); +} diff --git a/modules/process/tests/entity_chat_wire.rs b/modules/process/tests/entity_chat_wire.rs index fe9f0bc..07f4299 100644 --- a/modules/process/tests/entity_chat_wire.rs +++ b/modules/process/tests/entity_chat_wire.rs @@ -5,8 +5,8 @@ mod common; use common::{delta_frame, snapshot_with_state_blocks, SharedRuntime, TestKernel}; use lumio_host_runtime::SharedClock; use lumio_server_process::entity_chat::{ - generate_keys, issue_admission_credential, EntityChatHost, InputCommand, RoomClient, - ADMISSION_KEY_ID, RECONNECT_WINDOW_MS, + generate_keys, issue_admission_credential, ChatOpKind, EntityChatHost, InputCommand, + RoomClient, ADMISSION_KEY_ID, RECONNECT_WINDOW_MS, }; fn credential( @@ -65,6 +65,51 @@ fn admit_sends_full_snapshot_with_state_blocks_to_the_client() { ); } +#[test] +fn admit_chat_input_then_tick_sends_chat_event_delta_to_room_client() { + let keys = generate_keys(); + let host = EntityChatHost::new( + RECONNECT_WINDOW_MS, + SharedClock::test(), + Box::new(SharedRuntime::new()), + Box::new(TestKernel::new()), + ADMISSION_KEY_ID, + keys.public.to_vec(), + 1_000, + ); + let admit = host.admit( + "room-main".to_owned(), + "c-bot01".to_owned(), + credential(&keys, "Bot01", true), + ); + assert!(admit.accepted); + let mut client = RoomClient::connect(&host.listen_uri(), "c-bot01").expect("connect"); + let _ = client.recv_text(); + let admitted = host.admit_chat_input( + "c-bot01".to_owned(), + InputCommand::from_chat_text("hello-Bot01"), + ); + assert_eq!(admitted.kind, ChatOpKind::Admitted); + let tick = host.run_tick("room-main".to_owned()); + assert!( + tick.applied_tick >= 1, + "kernel tickFrame must run, got {tick:?}" + ); + let frame = client.recv_text().expect("delta"); + assert!( + frame.contains("\"messageType\":\"Delta\""), + "Room client must receive a C-1 Delta, got {frame}" + ); + assert!( + frame.contains("\"mappingId\":\"chat.event\""), + "Delta.changedBlocks must contain decodeable mappingId=chat.event, got {frame}" + ); + assert!( + !frame.contains("\"changedBlocks\":[]"), + "live-equivalent admit_input + tick must not broadcast empty changedBlocks, got {frame}" + ); +} + #[test] fn tick_broadcasts_runtime_delta_bytes_in_order() { let (host, keys) = host_ready(SharedRuntime::new()); From 36f72bb87f8deb0984f417da2a2ab60c44598214 Mon Sep 17 00:00:00 2001 From: Cui Date: Thu, 3 Sep 2026 02:28:29 +0800 Subject: [PATCH 2/2] fix(entity-chat): observe ConnectionSuperseded on old socket; fail over-budget ticks (R-00374) --- .../features/rust-entity-chat-host.md | 2 +- modules/process/src/entity_chat/clr.rs | 58 +++++++++++--- modules/process/src/entity_chat/host.rs | 13 ++- modules/process/src/entity_chat/runtime.rs | 29 ++++++- modules/process/src/entity_chat/suite.rs | 32 +++++--- modules/process/tests/common/mod.rs | 16 +++- .../process/tests/entity_chat_architecture.rs | 33 ++++++++ modules/process/tests/entity_chat_host.rs | 80 +++++++++++++++++++ 8 files changed, 227 insertions(+), 36 deletions(-) diff --git a/.spec/knowledge/features/rust-entity-chat-host.md b/.spec/knowledge/features/rust-entity-chat-host.md index 35fe946..cffdb41 100644 --- a/.spec/knowledge/features/rust-entity-chat-host.md +++ b/.spec/knowledge/features/rust-entity-chat-host.md @@ -19,7 +19,7 @@ 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`）。默认 `MaxSnapshotBytes=4096` 只能装下约 6 个聊天实体；101 实体 Capture 为 Retryable 时不得把 `restoredWindow: 0` / `processB=null` 写成 S7 ok。 - **发现**：外部产物经 `LUMIO_*` 环境变量与仓根相对路径；缺失即 BLOCKED，不硬编码开发机绝对路径。 diff --git a/modules/process/src/entity_chat/clr.rs b/modules/process/src/entity_chat/clr.rs index 4affebd..32b3926 100644 --- a/modules/process/src/entity_chat/clr.rs +++ b/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) -> Vec { @@ -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"}]}"#; diff --git a/modules/process/src/entity_chat/host.rs b/modules/process/src/entity_chat/host.rs index 28dd8ce..1c5d1d8 100644 --- a/modules/process/src/entity_chat/host.rs +++ b/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); diff --git a/modules/process/src/entity_chat/runtime.rs b/modules/process/src/entity_chat/runtime.rs index db11f2f..f43e107 100644 --- a/modules/process/src/entity_chat/runtime.rs +++ b/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, +} + +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. diff --git a/modules/process/src/entity_chat/suite.rs b/modules/process/src/entity_chat/suite.rs index 0f96242..31ee1ba 100644 --- a/modules/process/src/entity_chat/suite.rs +++ b/modules/process/src/entity_chat/suite.rs @@ -406,10 +406,7 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { let _ = client.recv_text(); } let mut pending_chats = 0usize; - let mut tick = RuntimeTick { - applied_tick: 0, - revision: 0, - }; + let mut tick = RuntimeTick::default(); let mut received = Vec::new(); for (connection, name) in &connections { let command = InputCommand::from_chat_text(&format!("hello-{name}")); @@ -433,7 +430,7 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { tick = host.schedule_room_tick(MAIN_ROOM.to_owned(), 1); drain_chat_event_deltas(&mut browser_wire, &mut received); } - let timer_ok = tick.applied_tick >= 1; + let timer_ok = tick.ok && tick.applied_tick >= 1; let chat_events: Vec = received .iter() .filter(|frame| is_chat_event_delta(frame)) @@ -537,6 +534,10 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { let entity_a_host = previous_bot100.net_entity_id.clone(); let previous_session = previous_bot100.session_id.clone(); let previous_account = previous_bot100.account_id.clone(); + let mut old_bot100 = RoomClient::connect(&host.listen_uri(), "c-bot100").ok(); + if let Some(client) = old_bot100.as_mut() { + let _ = client.recv_text(); + } let re_login = login_or_register(&account.uri(), "Bot100", TEST_PASSWORD, Some(&bot_claim)) .await .unwrap_or_else(|_| empty_login()); @@ -560,6 +561,17 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { } } } + let superseded_frame = old_bot100 + .as_mut() + .and_then(|client| client.recv_text().ok()); + let connection_superseded_received = superseded_frame + .as_deref() + .is_some_and(|frame| frame.contains("\"messageType\":\"ConnectionSuperseded\"")); + if connection_superseded_received { + if let Some(old) = old_bot100.as_mut() { + let _ = old.is_closed_after(); + } + } let rejected = host.admit_chat_input( "c-bot100".to_owned(), InputCommand::from_chat_text("while-down"), @@ -569,7 +581,7 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { InputCommand::from_chat_text("room-continues"), ); let _ = host.run_tick(MAIN_ROOM.to_owned()); - re_ok = re_ok && rejected.kind == ChatOpKind::Rejected; + re_ok = re_ok && rejected.kind == ChatOpKind::Rejected && connection_superseded_received; let reconnect_trace = json!({ "rebound": re_ok, "entityA": entity_a_host, @@ -579,13 +591,14 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { "previousSessionId": previous_session, "accountId": rebound_binding.as_ref().map(|binding| binding.account_id.clone()), "previousAccountId": previous_account, - "connectionSupersededReceived": takeover, + "takeover": takeover, + "connectionSupersededReceived": connection_superseded_received, "oldConnectionId": "c-bot100", }); scenarios.insert( "8".to_owned(), json!({ - "ok": re_ok && takeover, + "ok": re_ok && connection_superseded_received, "rebound": re_ok, "entityA": entity_a_host, "netEntityId": reconnect_trace.get("netEntityId").cloned(), @@ -594,7 +607,8 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { "previousSessionId": previous_session, "accountId": reconnect_trace.get("accountId").cloned(), "previousAccountId": previous_account, - "connectionSupersededReceived": takeover, + "takeover": takeover, + "connectionSupersededReceived": connection_superseded_received, }), ); diff --git a/modules/process/tests/common/mod.rs b/modules/process/tests/common/mod.rs index f5aa3ad..d62621c 100644 --- a/modules/process/tests/common/mod.rs +++ b/modules/process/tests/common/mod.rs @@ -8,6 +8,7 @@ use lumio_host_runtime::{KernelError, KernelFired, KernelHandle, KernelTimer, Ti use lumio_server_process::entity_chat::{ AttributeQueryOutcome, BoundEntityKind, ChatOperation, PersistRecord, QueryResult, RebindMode, RuntimeAdmit, RuntimeBinding, RuntimeQuery, RuntimeSurface, RuntimeTick, + MAX_CHAT_INPUTS_PER_TICK, }; pub const DISPATCH_EXPIRE: u32 = 1; @@ -401,11 +402,18 @@ impl RuntimeSurface for ScriptedRuntime { self.tick = tick_id; self.revision += 1; let pending = std::mem::take(&mut self.pending_chats); - self.events_by_tick.insert(tick_id, pending); - RuntimeTick { - applied_tick: self.tick, - revision: self.revision, + if pending.len() > MAX_CHAT_INPUTS_PER_TICK { + return RuntimeTick { + applied_tick: 0, + revision: self.revision, + ok: false, + event_count: 0, + code: Some("runtime_failure".to_owned()), + }; } + let event_count = pending.len() as u64; + self.events_by_tick.insert(tick_id, pending); + RuntimeTick::committed(self.tick, self.revision, event_count) } fn build_full_snapshot(&mut self, _room_id: &str, tick_id: u64, revision: u64) -> Vec { diff --git a/modules/process/tests/entity_chat_architecture.rs b/modules/process/tests/entity_chat_architecture.rs index 5414f4e..a5149d5 100644 --- a/modules/process/tests/entity_chat_architecture.rs +++ b/modules/process/tests/entity_chat_architecture.rs @@ -116,6 +116,39 @@ fn host_entry_restore_persist_uses_readonly_memory() { ); } +#[test] +fn suite_connection_superseded_received_must_not_copy_takeover() { + let text = + fs::read_to_string(process_root().join("src/entity_chat/suite.rs")).expect("suite.rs"); + assert!( + !text.contains("\"connectionSupersededReceived\": takeover"), + "connectionSupersededReceived must come from old-socket recv, not host takeover" + ); + assert!( + text.contains("RoomClient::connect") && text.contains("c-bot100"), + "S8 must attach c-bot100 as a RoomClient before takeover" + ); + assert!( + text.contains("ConnectionSuperseded"), + "S8 must recv messageType=ConnectionSuperseded on the old socket" + ); +} + +#[test] +fn suite_schedules_kernel_tick_every_max_chat_inputs() { + let text = + fs::read_to_string(process_root().join("src/entity_chat/suite.rs")).expect("suite.rs"); + assert!( + text.contains("pending_chats >= MAX_CHAT_INPUTS_PER_TICK"), + "suite must insert schedule_room_tick every 64 admits, not dump 101 into one RunTick" + ); + let ticks = text.matches("schedule_room_tick").count(); + assert!( + ticks >= 2, + "suite must schedule at least a batch tick and a remainder tick, got {ticks}" + ); +} + #[test] fn owned_sources_have_no_hardcoded_dev_machine_paths() { let mut hits = Vec::new(); diff --git a/modules/process/tests/entity_chat_host.rs b/modules/process/tests/entity_chat_host.rs index 5885305..3296198 100644 --- a/modules/process/tests/entity_chat_host.rs +++ b/modules/process/tests/entity_chat_host.rs @@ -288,3 +288,83 @@ fn batched_chat_inputs_stay_within_runtime_change_entry_budget() { "two ChatComponent field writes per chat.input must fit MaxChangeEntries=128" ); } + +fn admit_n( + host: &EntityChatHost, + keys: &lumio_server_process::entity_chat::Ed25519KeyPair, + n: usize, +) { + for i in 1..=n { + let name = format!("Bot{i:02}"); + let accepted = host + .admit( + "room-main".to_owned(), + format!("c-{i:03}"), + credential(keys, &name, true), + ) + .accepted; + assert!(accepted, "admit {name}"); + } +} + +fn enqueue_n(host: &EntityChatHost, n: usize) { + for i in 1..=n { + let admitted = host.admit_chat_input( + format!("c-{i:03}"), + InputCommand::from_chat_text(&format!("hello-{i}")), + ); + assert_eq!(admitted.kind, ChatOpKind::Admitted); + } +} + +#[test] +fn sixty_four_chat_inputs_one_tick_emit_chat_event() { + let (host, keys) = host_with(SharedRuntime::new()); + admit_n(&host, &keys, 64); + let mut client = + lumio_server_process::entity_chat::RoomClient::connect(&host.listen_uri(), "c-001") + .expect("connect"); + let _ = client.recv_text(); + enqueue_n(&host, 64); + let tick = host.run_tick("room-main".to_owned()); + assert!(tick.ok, "N=64 must succeed, got {tick:?}"); + assert_eq!(tick.event_count, 64); + let frame = client.recv_text().expect("delta"); + assert!( + frame.contains("\"mappingId\":\"chat.event\""), + "N=64 must emit chat.event, got {frame}" + ); +} + +#[test] +fn sixty_five_chat_inputs_one_tick_empty_delta_is_not_success() { + let (host, keys) = host_with(SharedRuntime::new()); + admit_n(&host, &keys, 65); + let mut client = + lumio_server_process::entity_chat::RoomClient::connect(&host.listen_uri(), "c-001") + .expect("connect"); + let _ = client.recv_text(); + enqueue_n(&host, 65); + let tick = host.run_tick("room-main".to_owned()); + assert!( + !tick.ok, + "N=65 must not be SUCCESS (Runtime MaxChangeEntries=128), got {tick:?}" + ); + assert_eq!(tick.event_count, 0); + match client.recv_text() { + Ok(frame) => { + assert!( + !frame.contains("\"mappingId\":\"chat.event\""), + "N=65 must not emit chat.event, got {frame}" + ); + assert!( + frame.contains("\"changedBlocks\":[]") || !tick.ok, + "N=65 empty Delta is not SUCCESS, got {frame}" + ); + } + Err(_) => assert!( + !tick.ok, + "budget fault must not be treated as a successful tick" + ), + } +}