From b81e74d3ef2ff5987df92791a607bfe619a1c760 Mon Sep 17 00:00:00 2001 From: Cui Date: Thu, 3 Sep 2026 06:40:50 +0800 Subject: [PATCH] fix(entity-chat): hold Bot.Host Room sockets until chat.event observed (R-00374) Keep Lumio.Client.Bot.Host connected after timer-trace until the suite observes Room chat.event (or writes release.flag). Tick from host wire pending, not sent.txt. Do not treat Sleep(400) or early Environment.Exit as SUCCESS. --- .../features/rust-entity-chat-host.md | 2 +- .../bot_startup_hook/StartupHook.cs | 16 ++++ modules/process/src/entity_chat/bots.rs | 90 +++++++++++++++--- modules/process/src/entity_chat/host.rs | 18 +++- modules/process/src/entity_chat/mod.rs | 2 +- modules/process/src/entity_chat/suite.rs | 95 ++++++++++++++----- .../process/tests/entity_chat_architecture.rs | 57 +++++++++++ modules/process/tests/entity_chat_wire.rs | 37 ++++++++ 8 files changed, 275 insertions(+), 42 deletions(-) diff --git a/.spec/knowledge/features/rust-entity-chat-host.md b/.spec/knowledge/features/rust-entity-chat-host.md index 96ee369..ba8af0d 100644 --- a/.spec/knowledge/features/rust-entity-chat-host.md +++ b/.spec/knowledge/features/rust-entity-chat-host.md @@ -22,7 +22,7 @@ ADR-056：Rust 宿主是接力交付面，只托管与传输。Room 世界是 Ru - **Room 网线**：loopback WebSocket。准入/重连发送 Runtime `BuildFullSnapshot`（含 `stateBlocks`）；每 Tick 把 `BuildDelta` 字节广播给本 Room 连接；同一 `connectionId` 可有多个观察者（Playwright + harness），后连者不得顶掉先连者的 egress。顶号先发 `ConnectionSuperseded` 再关旧连接。S8 证据 `connectionSupersededReceived` 只来自旧 `RoomClient` 收帧，不得用宿主 `takeover` 布尔冒充。S3 在 101 条 `chat.input` 之前挂上 `c-browser` Room WS；`playwrightRan` 只在浏览器真正从网线收到 Room 帧时为 true。 - **解析 / 查询**：`ResolveByNetEntityId` 接受 Runtime 32-hex 与 C-1 u64；HostEntry 把 Runtime `OkEntity`（无 Binding）补成列出的五元组。S5 unauthorized 走声明过的 claim-scoped `EntityIdentity.claimedMark`（`restrictedFlag` 未声明 → `RequestError`，不得冒充 Unauthorized）。 - **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`，不自建第二份事件队列。 -- **Client Bot**：S6 发言由 suite spawn `Lumio.Client.Bot.Host`（`LUMIO_BOT_HOST` / `LUMIO_CLIENT_ROOT` 或仓根相对 `LumioClient` 兄弟，缺失 BLOCKED）。Bot 进程经 Client Timer Manager drain NativeCore `tickFrame`（`native-kernel/tickFrame`，utteranceTicks 含 5/10/15），再把 `chat.input` 发上 Room WS。禁止 `host.admit_chat_input` 冒充 101 条 Bot 发言，禁止把常量 `[5,10,15]` 写成证据。生成的 hook 工程写在证据目录时必须自带 `Directory.Build.props`（`TreatWarningsAsErrors=false`），不得继承 Game 仓 analyzer；编译失败要把 **stdout+stderr** 写入 BLOCKED。 +- **Client Bot**：S6 发言由 suite spawn `Lumio.Client.Bot.Host`（`LUMIO_BOT_HOST` / `LUMIO_CLIENT_ROOT` 或仓根相对 `LumioClient` 兄弟，缺失 BLOCKED）。Bot 进程经 Client Timer Manager drain NativeCore `tickFrame`（`native-kernel/tickFrame`，utteranceTicks 含 5/10/15），再把 `chat.input` 发上 Room WS。写出 `timer-trace.json` 后仍保持 Room 套接字，直到 suite 在 Room 上观察到 submitted `chat.event`（或显式写 `release.flag`）才 Dispose / `Environment.Exit`。suite 按宿主 `pending_wire_chat_inputs`（Room 网线入站，不是 `sent.txt`）穿插 `tickFrame`，余量另 tick，不得把 `sent.txt` / `Sleep(400)` / 提前 Exit 当成 SUCCESS。禁止 `host.admit_chat_input` 冒充 101 条 Bot 发言，禁止把常量 `[5,10,15]` 写成证据。生成的 hook 工程写在证据目录时必须自带 `Directory.Build.props`（`TreatWarningsAsErrors=false`），不得继承 Game 仓 analyzer；编译失败要把 **stdout+stderr** 写入 BLOCKED。 - **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 通过之后写。`--restore-snapshot` 供 S7 进程 B 单独启 CLR 恢复。 diff --git a/modules/process/src/entity_chat/bot_startup_hook/StartupHook.cs b/modules/process/src/entity_chat/bot_startup_hook/StartupHook.cs index da93ed5..27f2441 100644 --- a/modules/process/src/entity_chat/bot_startup_hook/StartupHook.cs +++ b/modules/process/src/entity_chat/bot_startup_hook/StartupHook.cs @@ -95,6 +95,20 @@ private static int Run(string specPath) } WriteTrace(spec, invoked && sent == n, source, ticks, sent, null); + if (!string.IsNullOrWhiteSpace(spec.ReleasePath)) + { + DateTime deadline = DateTime.UtcNow.AddMinutes(2); + while (!File.Exists(spec.ReleasePath)) + { + if (DateTime.UtcNow >= deadline) + { + break; + } + + Thread.Sleep(50); + } + } + return invoked && sent == n ? 0 : 2; } finally @@ -198,6 +212,8 @@ private sealed class FleetSpec public string SentPath { get; set; } = ""; + public string ReleasePath { get; set; } = ""; + public ulong AdvanceToTick { get; set; } public List Bots { get; set; } = new List(); diff --git a/modules/process/src/entity_chat/bots.rs b/modules/process/src/entity_chat/bots.rs index 1e0888c..89495db 100644 --- a/modules/process/src/entity_chat/bots.rs +++ b/modules/process/src/entity_chat/bots.rs @@ -2,7 +2,7 @@ use std::fs::File; use std::path::{Path, PathBuf}; -use std::process::{Command, Stdio}; +use std::process::{Child, Command, Stdio}; use std::thread; use std::time::{Duration, Instant}; @@ -21,6 +21,45 @@ pub struct ClientBotTrace { pub blocked: Option, } +/// Live Bot.Host process whose Room sockets stay open until [`ClientBotFleet::release`]. +pub struct ClientBotFleet { + pub trace: ClientBotTrace, + child: Option, + release_path: PathBuf, +} + +impl ClientBotFleet { + /// Lets the hook dispose Room sockets and exit after Room observed chat.event. + pub fn release(mut self) { + self.release_mut(); + } + + fn release_mut(&mut self) { + let _ = std::fs::write(&self.release_path, "release\n"); + let Some(mut child) = self.child.take() else { + return; + }; + let deadline = Instant::now() + Duration::from_secs(15); + loop { + match child.try_wait() { + Ok(None) if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + break; + } + Ok(None) => thread::sleep(Duration::from_millis(50)), + Ok(Some(_)) | Err(_) => break, + } + } + } +} + +impl Drop for ClientBotFleet { + fn drop(&mut self) { + self.release_mut(); + } +} + /// Env lookup used by discovery. Process env in production; map in unit tests. pub trait BotHostEnv { /// Reads one environment variable. @@ -126,9 +165,9 @@ pub fn run_client_bot_fleet( out_dir: &Path, dotnet: &str, mut on_progress: F, -) -> Result +) -> Result where - F: FnMut(u32), + F: FnMut(), { std::fs::create_dir_all(out_dir).map_err(|error| error.to_string())?; let host = ensure_bot_host_executable(bot_host, dotnet)?; @@ -137,11 +176,13 @@ where let spec_path = out_dir.join("fleet-spec.json"); let trace_path = out_dir.join("timer-trace.json"); let sent_path = out_dir.join("sent.txt"); + let release_path = out_dir.join("release.flag"); let spec = json!({ "roomUri": room_uri, "engineNative": engine_native.display().to_string(), "tracePath": trace_path.display().to_string(), "sentPath": sent_path.display().to_string(), + "releasePath": release_path.display().to_string(), "advanceToTick": 15, "bots": envelopes.iter().map(|(connection, envelope)| { json!({ @@ -174,7 +215,7 @@ where .map_err(|error| format!("BLOCKED: spawn Lumio.Client.Bot.Host: {error}"))?; let deadline = Instant::now() + Duration::from_secs(60); loop { - on_progress(read_sent(&sent_path)); + on_progress(); if trace_path.is_file() { break; } @@ -202,9 +243,20 @@ where } thread::sleep(Duration::from_millis(50)); } - let _ = child.wait(); - on_progress(read_sent(&sent_path)); - parse_trace(&trace_path) + on_progress(); + let trace = match parse_trace(&trace_path) { + Ok(trace) => trace, + Err(reason) => { + let _ = child.kill(); + let _ = child.wait(); + return Err(reason); + } + }; + Ok(ClientBotFleet { + trace, + child: Some(child), + release_path, + }) } fn process_repo_root() -> PathBuf { @@ -384,13 +436,6 @@ fn bot_host_command(dotnet: &str, host: &Path) -> Command { } } -fn read_sent(path: &Path) -> u32 { - std::fs::read_to_string(path) - .ok() - .and_then(|text| text.trim().parse().ok()) - .unwrap_or(0) -} - fn tail_logs(stdout_path: &Path, stderr_path: &Path) -> String { let stdout = std::fs::read_to_string(stdout_path).unwrap_or_default(); let stderr = std::fs::read_to_string(stderr_path).unwrap_or_default(); @@ -449,6 +494,7 @@ fn parse_trace(path: &Path) -> Result { mod tests { use super::{ discover_bot_host_in, hook_compile_failure_text, write_hook_isolation_files, BotHostEnv, + ClientBotFleet, ClientBotTrace, }; use std::collections::HashMap; use std::fs; @@ -608,6 +654,22 @@ internal static class Warn ); } + #[test] + fn fleet_release_writes_release_path() { + let tmp = tempfile::tempdir().expect("tmp"); + let release_path = tmp.path().join("release.flag"); + let fleet = ClientBotFleet { + trace: ClientBotTrace::default(), + child: None, + release_path: release_path.clone(), + }; + fleet.release(); + assert!( + release_path.is_file(), + "suite release must create the hook wait file" + ); + } + #[test] fn hook_compile_blocked_text_includes_stdout_when_stderr_empty() { let text = hook_compile_failure_text(b"error CA1869: cache JsonSerializerOptions\n", b""); diff --git a/modules/process/src/entity_chat/host.rs b/modules/process/src/entity_chat/host.rs index f287a48..f53070f 100644 --- a/modules/process/src/entity_chat/host.rs +++ b/modules/process/src/entity_chat/host.rs @@ -15,8 +15,8 @@ use super::envelope::{ }; use super::runtime::BoundEntityKind; use super::runtime::{ - AttributeQueryScope, ChatOperation, PersistRecord, QueryResult, RebindMode, RuntimeAdmit, - RuntimeBinding, RuntimeQuery, RuntimeSurface, RuntimeTick, + AttributeQueryScope, ChatOpKind, ChatOperation, PersistRecord, QueryResult, RebindMode, + RuntimeAdmit, RuntimeBinding, RuntimeQuery, RuntimeSurface, RuntimeTick, }; use super::wire::{RoomListener, WireEvent, WireSender}; @@ -154,6 +154,7 @@ struct Inner { expire_watch: HashMap, pending_egress: HashMap>, tick_id: u64, + wire_chat_pending: u64, } enum OwnerWork { @@ -213,6 +214,7 @@ impl EntityChatHost { expire_watch: HashMap::new(), pending_egress: HashMap::new(), tick_id: 0, + wire_chat_pending: 0, }; if inner .kernel @@ -358,6 +360,12 @@ impl EntityChatHost { self.on_owner(move |inner| inner.try_resolve_by_net_entity_id(&room_id, &net_entity_id)) } + /// Chat.input frames admitted from Room WS and not yet applied by a tick. + #[must_use] + pub fn pending_wire_chat_inputs(&self) -> usize { + self.on_owner(move |inner| usize::try_from(inner.wire_chat_pending).unwrap_or(usize::MAX)) + } + /// Count live Room WS observers for a connection (harness wait). #[must_use] pub fn wire_observer_count(&self, connection_id: String) -> usize { @@ -625,6 +633,7 @@ impl Inner { if !tick.ok { return tick; } + self.wire_chat_pending = 0; let frames = self .runtime .build_delta(room_id, tick.applied_tick, tick.revision); @@ -699,7 +708,10 @@ impl Inner { text, } => { if let Ok(envelope) = parse_input_command_json(&text) { - let _ = self.admit_chat_input(&connection_id, &envelope); + let admitted = self.admit_chat_input(&connection_id, &envelope); + if admitted.kind == ChatOpKind::Admitted { + self.wire_chat_pending = self.wire_chat_pending.saturating_add(1); + } } } WireEvent::Closed { .. } => { diff --git a/modules/process/src/entity_chat/mod.rs b/modules/process/src/entity_chat/mod.rs index d1eebd6..760372c 100644 --- a/modules/process/src/entity_chat/mod.rs +++ b/modules/process/src/entity_chat/mod.rs @@ -36,7 +36,7 @@ pub use admission::{ generate_keys, issue_admission_credential, issue_bot_tool_credential, verify_admission, AdmissionPayload, Ed25519KeyPair, }; -pub use bots::{discover_bot_host, run_client_bot_fleet, ClientBotTrace}; +pub use bots::{discover_bot_host, run_client_bot_fleet, ClientBotFleet, ClientBotTrace}; pub use clr::{ClrGameplay, ClrGameplayConfig}; pub use discover::{discover, ReplayArtifacts}; pub use envelope::{ diff --git a/modules/process/src/entity_chat/suite.rs b/modules/process/src/entity_chat/suite.rs index 5d65544..a66ece0 100644 --- a/modules/process/src/entity_chat/suite.rs +++ b/modules/process/src/entity_chat/suite.rs @@ -13,7 +13,7 @@ use serde_json::{json, Value}; use super::account::{login_or_register, AccountServerProcess}; use super::admission::{generate_keys, issue_bot_tool_credential, verify_admission}; -use super::bots::{discover_bot_host, run_client_bot_fleet, ClientBotTrace}; +use super::bots::{discover_bot_host, run_client_bot_fleet, ClientBotFleet, ClientBotTrace}; use super::browser::capture_browser_login; use super::clr::{ClrGameplay, ClrGameplayConfig}; use super::crypto::hex_lower; @@ -469,11 +469,10 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { }) .collect(); let first_envelope = envelopes.first().map(|(_, envelope)| envelope.clone()); - let mut pending_chats = 0usize; - let mut last_ticked = 0u32; let mut tick = RuntimeTick::default(); let mut received = Vec::new(); let fleet_dir = out_dir.join("client-bots"); + let mut bot_fleet: Option = None; let bot_trace = match run_client_bot_fleet( &bot_host, &engine_native, @@ -481,36 +480,41 @@ async fn run_round_async(options: &SuiteOptions, out_dir: &Path) -> Value { &envelopes, &fleet_dir, &options.dotnet, - |sent| { - pending_chats = sent.saturating_sub(last_ticked) as usize; - 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); - last_ticked = sent; - pending_chats = 0; - } + || { + apply_pending_chat_ticks(&host, &mut tick, &mut browser_wire, &mut received); }, ) { - Ok(trace) => trace, + Ok(fleet) => { + let trace = fleet.trace.clone(); + bot_fleet = Some(fleet); + trace + } Err(reason) => { blocked = blocked.or(Some(reason)); ClientBotTrace::default() } }; + wait_for_observed_chat_events( + &host, + &mut tick, + &mut browser_wire, + &mut received, + BOT_COUNT as usize, + Duration::from_secs(30), + ); if let Some(client) = browser_wire.as_mut() { let _ = client.send_text(&InputCommand::from_chat_text("hello-browser").to_json()); } - pending_chats = pending_chats.saturating_add(1); - if pending_chats > 0 { - tick = host.schedule_room_tick(MAIN_ROOM.to_owned(), 1); - drain_chat_event_deltas(&mut browser_wire, &mut received); - } - for _ in 0..6 { - if received.len() >= 101 { - break; - } - tick = host.schedule_room_tick(MAIN_ROOM.to_owned(), 1); - drain_chat_event_deltas(&mut browser_wire, &mut received); + wait_for_observed_chat_events( + &host, + &mut tick, + &mut browser_wire, + &mut received, + 101, + Duration::from_secs(10), + ); + if let Some(fleet) = bot_fleet.take() { + fleet.release(); } let timer_ok = bot_trace.timer_manager_invoked && bot_trace.tick_source == "native-kernel/tickFrame" @@ -952,6 +956,51 @@ fn drain_chat_event_deltas(client: &mut Option, received: &mut Vec, + received: &mut Vec, +) { + loop { + let pending_chats = host.pending_wire_chat_inputs(); + if pending_chats >= MAX_CHAT_INPUTS_PER_TICK { + *tick = host.schedule_room_tick(MAIN_ROOM.to_owned(), 1); + drain_chat_event_deltas(browser_wire, received); + continue; + } + if pending_chats > 0 { + *tick = host.schedule_room_tick(MAIN_ROOM.to_owned(), 1); + drain_chat_event_deltas(browser_wire, received); + } + break; + } +} + +fn wait_for_observed_chat_events( + host: &EntityChatHost, + tick: &mut RuntimeTick, + browser_wire: &mut Option, + received: &mut Vec, + want: usize, + budget: Duration, +) { + let deadline = Instant::now() + budget; + loop { + apply_pending_chat_ticks(host, tick, browser_wire, received); + drain_chat_event_deltas(browser_wire, received); + if received.len() >= want { + return; + } + if Instant::now() >= deadline { + apply_pending_chat_ticks(host, tick, browser_wire, received); + drain_chat_event_deltas(browser_wire, received); + return; + } + thread::sleep(Duration::from_millis(50)); + } +} + fn sha256_hex(bytes: &[u8]) -> String { hex_lower(&Sha256::digest(bytes)) } diff --git a/modules/process/tests/entity_chat_architecture.rs b/modules/process/tests/entity_chat_architecture.rs index 90a8fc4..24e7e12 100644 --- a/modules/process/tests/entity_chat_architecture.rs +++ b/modules/process/tests/entity_chat_architecture.rs @@ -300,6 +300,63 @@ fn suite_chat_burst_does_not_host_admit_bot_utterances() { ); } +#[test] +fn bot_host_must_not_exit_or_dispose_sockets_before_room_observes_chat_events() { + let hook = + fs::read_to_string(process_root().join("src/entity_chat/bot_startup_hook/StartupHook.cs")) + .expect("StartupHook.cs"); + let bots = fs::read_to_string(process_root().join("src/entity_chat/bots.rs")).expect("bots.rs"); + let suite = + fs::read_to_string(process_root().join("src/entity_chat/suite.rs")).expect("suite.rs"); + let host = fs::read_to_string(process_root().join("src/entity_chat/host.rs")).expect("host.rs"); + let production_bots = bots + .split("#[cfg(test)]") + .next() + .expect("production bots.rs"); + + let run_start = hook + .find("private static int Run(") + .expect("StartupHook.Run"); + let run_end = hook + .find("private static void SendText") + .expect("StartupHook.Run end"); + let run = &hook[run_start..run_end]; + let write_trace = run + .find("WriteTrace(spec, invoked && sent == n") + .expect("success path writes timer-trace"); + assert!( + run[write_trace..].contains("ReleasePath") && run[write_trace..].contains("File.Exists"), + "after timer-trace, Bot.Host must wait for a suite ReleasePath before returning to Environment.Exit / finally dispose" + ); + assert!( + !run[write_trace..].contains("Thread.Sleep(400)"), + "must not treat Sleep(400) after timer-trace as SUCCESS; hold is suite/Room observation" + ); + assert!( + hook.contains("public string ReleasePath"), + "fleet spec must carry ReleasePath so the suite can release the held sockets" + ); + + assert!( + production_bots.contains("releasePath") + && (production_bots.contains("fn release") + || production_bots.contains("fn release_mut")), + "bots.rs must keep Lumio.Client.Bot.Host alive and expose an explicit suite release" + ); + assert!( + !suite.contains("sent.saturating_sub"), + "suite must not schedule ticks from sent.txt; that races host receive" + ); + assert!( + host.contains("pending_wire_chat_inputs") && suite.contains("pending_wire_chat_inputs"), + "tick budget must follow Room wire observation of chat.input, not Bot.Host sent.txt" + ); + assert!( + suite.contains("release(") && suite.contains("drain_chat_event_deltas"), + "suite must hold the fleet until Room observed chat.event, then release" + ); +} + #[test] fn generated_hook_build_isolates_from_parent_directory_build_props() { let bots = fs::read_to_string(process_root().join("src/entity_chat/bots.rs")).expect("bots.rs"); diff --git a/modules/process/tests/entity_chat_wire.rs b/modules/process/tests/entity_chat_wire.rs index e7d44ac..ea8d27f 100644 --- a/modules/process/tests/entity_chat_wire.rs +++ b/modules/process/tests/entity_chat_wire.rs @@ -98,6 +98,43 @@ fn room_client_chat_input_over_wire_then_tick_sends_chat_event_delta() { ); } +#[test] +fn pending_wire_chat_inputs_counts_room_ingress_until_tick() { + 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(); + client + .send_text(&InputCommand::from_chat_text("hello-Bot01").to_json()) + .expect("wire chat.input"); + let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500); + while host.pending_wire_chat_inputs() == 0 && std::time::Instant::now() < deadline { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + assert_eq!( + host.pending_wire_chat_inputs(), + 1, + "Room WS chat.input must be observed as pending before tick" + ); + let tick = host.run_tick("room-main".to_owned()); + assert!(tick.ok, "kernel tickFrame must run, got {tick:?}"); + assert_eq!(host.pending_wire_chat_inputs(), 0); +} + #[test] fn admit_chat_input_then_tick_sends_chat_event_delta_to_room_client() { let keys = generate_keys();