Skip to main content

self_hosted_node/engine_backend/
java_backend.rs

1#![allow(dead_code)]
2
3#[cfg(forge_backend)]
4use std::collections::HashMap;
5use std::env;
6#[cfg(feature = "java-forge")]
7use std::io::{BufRead, BufReader, BufWriter, Write};
8use std::path::{Path, PathBuf};
9#[cfg(feature = "java-forge")]
10use std::process::{Child, ChildStdin, Command, Stdio};
11use std::sync::atomic::AtomicBool;
12use std::sync::mpsc as std_mpsc;
13#[cfg(feature = "java-forge")]
14use std::sync::mpsc::RecvTimeoutError;
15#[cfg(forge_backend)]
16use std::sync::mpsc::TryRecvError;
17use std::sync::Arc;
18#[cfg(feature = "java-forge")]
19use std::sync::Mutex;
20#[cfg(forge_backend)]
21use std::time::Duration;
22#[cfg(forge_backend)]
23use std::time::Instant;
24
25use manabrew_protocol::deck_dto::{Deck, DeckCardIdentity};
26
27use crate::config::DeckSelection;
28#[cfg(feature = "java-forge")]
29use manabot::{BotAgent, SimpleAi};
30#[cfg(forge_backend)]
31use manabrew_agent_interface::game_view_dto::GameViewDto;
32use manabrew_agent_interface::prompt::{AgentMessage, ClientToServerMessage};
33#[cfg(forge_backend)]
34use manabrew_agent_interface::prompt::{
35    AgentPrompt, ChooseActionOutput, DiceRolledOutput, DirectiveInput, GameOverInput, PromptInput,
36    PromptOutput, ProtocolError, ProtocolErrorCode, ResponseViolation, StateUpdate,
37};
38#[cfg(feature = "java-forge")]
39use manabrew_agent_interface::prompt::{MulliganOutput, MulliganPutBackOutput};
40use serde::Serialize;
41#[cfg(feature = "java-forge")]
42use serde_json::json;
43#[cfg(feature = "java-forge")]
44use serde_json::Value;
45#[cfg(forge_backend)]
46use tracing::warn;
47#[cfg(forge_backend)]
48use tracing::{debug, info};
49
50use super::HostedGameOver;
51use crate::config::workspace_root;
52
53pub fn unsupported_message() -> &'static str {
54    "hosted java-forge backend is unavailable; rebuild self-hosted-node with --features java-forge"
55}
56
57#[cfg(feature = "java-forge")]
58pub fn run_smoke_game(max_prompts: usize) -> Result<(), String> {
59    let config = JavaRuntimeConfig::from_env();
60    let assets_dir = config.assets_dir.to_string_lossy().to_string();
61    let bridge = SubprocessBridge::spawn(&config)?;
62    let mut session = JavaForgeSession::new(bridge);
63    session.initialize(&assets_dir)?;
64
65    let deck_a = smoke_deck("Mountain", "Lightning Bolt");
66    let deck_b = smoke_deck("Forest", "Grizzly Bears");
67    let request = StartGameRequest::new(
68        "self-hosted-java-smoke".to_string(),
69        String::new(),
70        20,
71        42,
72        vec![
73            PlayerConfig::new("Smoke A".to_string(), &deck_a, Vec::new()),
74            PlayerConfig::new("Smoke B".to_string(), &deck_b, Vec::new()),
75        ],
76    );
77    let session_id = session.start_game(&request)?;
78    info!(session_id, "java-forge smoke session started");
79
80    let mut prompts_seen = 0usize;
81    while prompts_seen < max_prompts {
82        let Some(prompt_json) = wait_for_prompt(&mut session, 600)? else {
83            session.end_game()?;
84            return Err("timed out waiting for java-forge smoke prompt".to_string());
85        };
86        let prompt: AgentPrompt = serde_json::from_str(&prompt_json)
87            .map_err(|err| format!("failed to parse java-forge smoke prompt: {err}"))?;
88        let player = player_index(&prompt.deciding_player_id);
89        info!(prompts_seen, player, "java-forge smoke prompt");
90        let pass = PromptOutput::ChooseAction(ChooseActionOutput::Pass {
91            until: None,
92            exhaust_stack: false,
93        });
94        session.submit_action(&serde_json::to_string(&pass).map_err(|err| err.to_string())?)?;
95        prompts_seen += 1;
96    }
97
98    let snapshot_json = session.get_snapshot(Some(0))?;
99    let snapshot: Value = serde_json::from_str(&snapshot_json)
100        .map_err(|err| format!("failed to parse java-forge smoke snapshot: {err}"))?;
101    info!(
102        turn = snapshot
103            .get("turn")
104            .and_then(|value| value.as_i64())
105            .unwrap_or_default(),
106        phase = snapshot
107            .get("phase")
108            .and_then(|value| value.as_str())
109            .unwrap_or("<missing>"),
110        "java-forge smoke snapshot"
111    );
112    session.end_game()?;
113    Ok(())
114}
115
116#[cfg(not(feature = "java-forge"))]
117pub fn run_smoke_game(_max_prompts: usize) -> Result<(), String> {
118    Err(
119        "java-forge smoke requires building self-hosted-node with --features java-forge"
120            .to_string(),
121    )
122}
123
124// Unlike run_smoke_game, which spawns the JAR in a subprocess, this drives the
125// GraalVM native library in-process: isolate creation, forge_initialize (which
126// loads the card database) and a real game start. CI runs it after restoring
127// forge-harness/native/build from cache, where nothing else would notice a
128// stale or broken libforgeharness.
129#[cfg(feature = "graal-forge")]
130pub fn run_graal_smoke() -> Result<(), String> {
131    let config = JavaRuntimeConfig::from_env();
132    let engine = GraalEngineHandle::create(&config.assets_dir)?;
133
134    let deck_a = smoke_deck("Mountain", "Lightning Bolt");
135    let deck_b = smoke_deck("Forest", "Grizzly Bears");
136    let request = StartGameRequest::new(
137        "self-hosted-graal-smoke".to_string(),
138        String::new(),
139        20,
140        42,
141        vec![
142            PlayerConfig::new("Smoke A".to_string(), &deck_a, Vec::new()),
143            PlayerConfig::new("Smoke B".to_string(), &deck_b, Vec::new()),
144        ],
145    );
146
147    let session_id = engine.start_game(&request.to_json().map_err(|err| err.to_string())?)?;
148    info!(session_id, "graal-forge smoke session started");
149    engine.end_game(&session_id)?;
150    Ok(())
151}
152
153#[cfg(not(feature = "graal-forge"))]
154pub fn run_graal_smoke() -> Result<(), String> {
155    Err(
156        "graal-forge smoke requires building self-hosted-node with --features graal-forge"
157            .to_string(),
158    )
159}
160
161#[cfg(feature = "java-forge")]
162pub fn run_scenario(name: &str, max_prompts: usize) -> Result<(), String> {
163    let scenario = JavaScenario::from_name(name)?;
164    let config = JavaRuntimeConfig::from_env();
165    let assets_dir = config.assets_dir.to_string_lossy().to_string();
166    let bridge = SubprocessBridge::spawn(&config)?;
167    let mut session = JavaForgeSession::new(bridge);
168    session.initialize(&assets_dir)?;
169
170    let request = StartGameRequest::new(
171        format!("self-hosted-java-scenario-{}", scenario.name()),
172        String::new(),
173        20,
174        42,
175        vec![
176            PlayerConfig::new(
177                "Scenario A".to_string(),
178                &scenario_deck("Swamp"),
179                Vec::new(),
180            ),
181            PlayerConfig::new(
182                "Scenario B".to_string(),
183                &scenario_deck("Forest"),
184                Vec::new(),
185            ),
186        ],
187    );
188    let session_id = session.start_game(&request)?;
189    info!(
190        session_id,
191        scenario = scenario.name(),
192        "java-forge scenario started"
193    );
194
195    let result = run_scenario_loop(&mut session, scenario, max_prompts);
196    let end_result = session.end_game();
197    result.and(end_result)
198}
199
200#[cfg(not(feature = "java-forge"))]
201pub fn run_scenario(_name: &str, _max_prompts: usize) -> Result<(), String> {
202    Err(
203        "java-forge scenarios require building self-hosted-node with --features java-forge"
204            .to_string(),
205    )
206}
207
208#[cfg(feature = "java-forge")]
209pub fn run_self_play(
210    seats: &[DeckSelection],
211    starting_life: i32,
212    seed: u64,
213    max_prompts: usize,
214    games: usize,
215) -> Result<(), String> {
216    let config = JavaRuntimeConfig::from_env();
217    let assets_dir = config.assets_dir.to_string_lossy().to_string();
218    let bridge = SubprocessBridge::spawn(&config)?;
219    let mut session = JavaForgeSession::new(bridge);
220    session.initialize(&assets_dir)?;
221
222    let mut players = Vec::with_capacity(seats.len());
223    for (i, seat) in seats.iter().enumerate() {
224        let identities = deck_card_identities(&seat.deck);
225        players.push(PlayerConfig::new(
226            format!("Self-Play {}", i + 1),
227            &identities,
228            commander_names_for_java(&seat.deck, seat.commander_name.as_deref()),
229        ));
230    }
231
232    for game_index in 0..games.max(1) {
233        let request = StartGameRequest::new(
234            format!("self-hosted-java-self-play-{game_index}"),
235            String::new(),
236            starting_life,
237            seed.wrapping_add(game_index as u64),
238            players.clone(),
239        );
240        let session_id = session.start_game(&request)?;
241        info!(
242            session_id,
243            game_index,
244            games,
245            players = seats.len(),
246            starting_life,
247            max_prompts,
248            "java-forge self-play game started"
249        );
250        let result = run_self_play_loop(&mut session, max_prompts);
251        let end_result = session.end_game();
252        result.and(end_result)?;
253    }
254    Ok(())
255}
256
257#[cfg(not(feature = "java-forge"))]
258pub fn run_self_play(
259    _seats: &[DeckSelection],
260    _starting_life: i32,
261    _seed: u64,
262    _max_prompts: usize,
263    _games: usize,
264) -> Result<(), String> {
265    Err(
266        "java-forge self-play requires building self-hosted-node with --features java-forge"
267            .to_string(),
268    )
269}
270
271#[cfg(feature = "java-forge")]
272type SharedBridge = Arc<Mutex<SubprocessBridge>>;
273
274#[cfg(feature = "java-forge")]
275struct PoolSlot {
276    bridge: SharedBridge,
277    active: usize,
278}
279
280#[cfg(feature = "java-forge")]
281pub struct JavaEnginePool {
282    config: JavaRuntimeConfig,
283    max_sessions: usize,
284    sessions_per_process: usize,
285    slots: Mutex<Vec<PoolSlot>>,
286    in_use: Mutex<HashMap<String, SharedBridge>>,
287}
288
289#[cfg(feature = "java-forge")]
290#[derive(Clone)]
291pub struct JavaEngineHandle {
292    pool: Arc<JavaEnginePool>,
293}
294
295#[cfg(feature = "java-forge")]
296impl JavaEnginePool {
297    pub fn start(
298        config: &JavaRuntimeConfig,
299        max_sessions: usize,
300        sessions_per_process: usize,
301    ) -> Result<Arc<Self>, String> {
302        let max_sessions = max_sessions.max(1);
303        let sessions_per_process = sessions_per_process.max(1);
304        let processes = max_sessions.div_ceil(sessions_per_process);
305        let mut slots = Vec::with_capacity(processes);
306        for slot in 0..processes {
307            info!(
308                slot,
309                processes, sessions_per_process, "pre-warming java subprocess"
310            );
311            let bridge = SubprocessBridge::spawn(config)?;
312            slots.push(PoolSlot {
313                bridge: Arc::new(Mutex::new(bridge)),
314                active: 0,
315            });
316        }
317        Ok(Arc::new(Self {
318            config: config.clone(),
319            max_sessions,
320            sessions_per_process,
321            slots: Mutex::new(slots),
322            in_use: Mutex::new(HashMap::new()),
323        }))
324    }
325
326    pub fn handle(self: &Arc<Self>) -> JavaEngineHandle {
327        JavaEngineHandle {
328            pool: Arc::clone(self),
329        }
330    }
331}
332
333#[cfg(feature = "java-forge")]
334impl Drop for JavaEnginePool {
335    fn drop(&mut self) {
336        let slots = self.slots.get_mut().map(std::mem::take).unwrap_or_default();
337        for slot in slots {
338            if let Ok(mutex) = Arc::try_unwrap(slot.bridge) {
339                if let Ok(inner) = mutex.into_inner() {
340                    inner.shutdown();
341                }
342            }
343        }
344    }
345}
346
347#[cfg(feature = "java-forge")]
348impl JavaEnginePool {
349    fn acquire(&self) -> Result<SharedBridge, String> {
350        let deadline = Instant::now() + Duration::from_secs(60);
351        loop {
352            let claimed = {
353                let mut slots = self
354                    .slots
355                    .lock()
356                    .map_err(|_| "java engine slots poisoned".to_string())?;
357                let mut claimed = None;
358                for slot in slots.iter_mut() {
359                    if slot.active < self.sessions_per_process {
360                        slot.active += 1;
361                        claimed = Some(Arc::clone(&slot.bridge));
362                        break;
363                    }
364                }
365                claimed
366            };
367            if let Some(bridge) = claimed {
368                let alive = bridge
369                    .lock()
370                    .ok()
371                    .map(|mut guard| guard.is_alive())
372                    .unwrap_or(false);
373                if alive {
374                    return Ok(bridge);
375                }
376                warn!("discarding dead java subprocess from pool");
377                self.replace_slot(&bridge);
378                continue;
379            }
380            if Instant::now() >= deadline {
381                return Err(format!(
382                    "java engine pool exhausted (max_sessions={}); no free session slot after 60s",
383                    self.max_sessions
384                ));
385            }
386            std::thread::sleep(Duration::from_millis(50));
387        }
388    }
389
390    fn release(&self, bridge: SharedBridge) {
391        let now_idle = {
392            let mut slots = match self.slots.lock() {
393                Ok(slots) => slots,
394                Err(_) => return,
395            };
396            match slots.iter_mut().find(|s| Arc::ptr_eq(&s.bridge, &bridge)) {
397                Some(slot) => {
398                    slot.active = slot.active.saturating_sub(1);
399                    slot.active == 0
400                }
401                None => return,
402            }
403        };
404        if !now_idle {
405            return;
406        }
407        let healthy = {
408            let mut guard = match bridge.lock() {
409                Ok(guard) => guard,
410                Err(_) => return,
411            };
412            guard.is_alive() && guard.reset().is_ok()
413        };
414        if !healthy {
415            warn!("java subprocess unhealthy at idle; respawning");
416            self.replace_slot(&bridge);
417        }
418    }
419
420    fn replace_slot(&self, dead: &SharedBridge) {
421        let Ok(mut slots) = self.slots.lock() else {
422            return;
423        };
424        let Some(index) = slots.iter().position(|s| Arc::ptr_eq(&s.bridge, dead)) else {
425            return;
426        };
427        match SubprocessBridge::spawn(&self.config) {
428            Ok(replacement) => {
429                slots[index] = PoolSlot {
430                    bridge: Arc::new(Mutex::new(replacement)),
431                    active: 0,
432                };
433            }
434            Err(error) => {
435                warn!(%error, "failed to respawn java subprocess; retiring pool slot");
436                slots.remove(index);
437            }
438        }
439    }
440}
441
442#[cfg(feature = "java-forge")]
443impl JavaEngineHandle {
444    fn bridge_for(&self, session_id: &str) -> Result<SharedBridge, String> {
445        let in_use = self
446            .pool
447            .in_use
448            .lock()
449            .map_err(|_| "java engine in_use map poisoned".to_string())?;
450        in_use
451            .get(session_id)
452            .cloned()
453            .ok_or_else(|| format!("unknown java session: {session_id}"))
454    }
455
456    pub fn start_game(&self, request_json: &str) -> Result<String, String> {
457        let bridge = self.pool.acquire()?;
458        let response = {
459            let mut guard = bridge
460                .lock()
461                .map_err(|_| "java subprocess mutex poisoned".to_string())?;
462            guard.start_game_json(request_json)
463        };
464        let response = match response {
465            Ok(response) => response,
466            Err(error) => {
467                self.pool.release(bridge);
468                return Err(error);
469            }
470        };
471        let parsed: StartGameResponse = match serde_json::from_str(&response) {
472            Ok(parsed) => parsed,
473            Err(error) => {
474                self.pool.release(bridge);
475                return Err(format!("malformed startGame response: {error}"));
476            }
477        };
478        let session_id = parsed.session_id.clone();
479        let displaced = {
480            let mut in_use = self
481                .pool
482                .in_use
483                .lock()
484                .map_err(|_| "java engine in_use map poisoned".to_string())?;
485            in_use.insert(session_id.clone(), bridge)
486        };
487        if let Some(displaced) = displaced {
488            warn!(
489                session_id,
490                "session_id collision; releasing displaced java subprocess"
491            );
492            self.pool.release(displaced);
493        }
494        Ok(session_id)
495    }
496
497    pub fn submit_action(&self, session_id: &str, action_json: &str) -> Result<String, String> {
498        let bridge = self.bridge_for(session_id)?;
499        let mutex_started = Instant::now();
500        let mut guard = bridge
501            .lock()
502            .map_err(|_| "java subprocess mutex poisoned".to_string())?;
503        crate::metrics::record_forge_decision_stage("bridge_mutex", mutex_started.elapsed());
504        let call_started = Instant::now();
505        let result = guard.submit_action(session_id, action_json);
506        crate::metrics::record_forge_decision_stage("submit_action", call_started.elapsed());
507        result
508    }
509
510    pub fn get_prompt(
511        &self,
512        session_id: &str,
513        player_index: usize,
514    ) -> Result<Option<String>, String> {
515        let bridge = self.bridge_for(session_id)?;
516        let mut guard = bridge
517            .lock()
518            .map_err(|_| "java subprocess mutex poisoned".to_string())?;
519        guard.get_prompt(session_id, player_index)
520    }
521
522    pub fn is_game_over(&self, session_id: &str) -> Result<bool, String> {
523        let bridge = self.bridge_for(session_id)?;
524        let mut guard = bridge
525            .lock()
526            .map_err(|_| "java subprocess mutex poisoned".to_string())?;
527        guard.is_game_over(session_id)
528    }
529
530    pub fn get_snapshot(&self, session_id: &str, viewer: Option<usize>) -> Result<String, String> {
531        let bridge = self.bridge_for(session_id)?;
532        let mut guard = bridge
533            .lock()
534            .map_err(|_| "java subprocess mutex poisoned".to_string())?;
535        guard.get_snapshot(session_id, viewer)
536    }
537
538    pub fn end_game(&self, session_id: &str) -> Result<(), String> {
539        let bridge = {
540            let mut in_use = self
541                .pool
542                .in_use
543                .lock()
544                .map_err(|_| "java engine in_use map poisoned".to_string())?;
545            in_use.remove(session_id)
546        };
547        let Some(bridge) = bridge else {
548            return Ok(());
549        };
550        let result = {
551            let mut guard = bridge
552                .lock()
553                .map_err(|_| "java subprocess mutex poisoned".to_string())?;
554            guard.end_game(session_id)
555        };
556        self.pool.release(bridge);
557        result
558    }
559
560    pub fn abort_game(&self, session_id: &str) -> Result<(), String> {
561        let bridge = {
562            let mut in_use = self
563                .pool
564                .in_use
565                .lock()
566                .map_err(|_| "java engine in_use map poisoned".to_string())?;
567            in_use.remove(session_id)
568        };
569        let Some(bridge) = bridge else {
570            return Ok(());
571        };
572        let result = {
573            let mut guard = bridge
574                .lock()
575                .map_err(|_| "java subprocess mutex poisoned".to_string())?;
576            guard.abort_game(session_id)
577        };
578        self.pool.release(bridge);
579        result
580    }
581}
582
583#[cfg(feature = "java-forge")]
584static JAVA_ENGINE: std::sync::OnceLock<Arc<JavaEnginePool>> = std::sync::OnceLock::new();
585
586#[cfg(feature = "java-forge")]
587pub fn init_engine() -> Result<(), String> {
588    if JAVA_ENGINE.get().is_some() {
589        return Ok(());
590    }
591    let config = JavaRuntimeConfig::from_env();
592    // SELF_HOSTED_NODE_MAX_GAMES is the concurrent-session ceiling for this node.
593    // SELF_HOSTED_NODE_GAMES_PER_JVM multiplexes that many sessions into each
594    // subprocess (the engine is concurrency-safe since the endstep patches), so
595    // the pool spawns ceil(max_games / games_per_jvm) processes. Default 1 keeps
596    // the historical one-subprocess-per-game shape.
597    let max_sessions = env::var("SELF_HOSTED_NODE_MAX_GAMES")
598        .ok()
599        .and_then(|value| value.parse::<usize>().ok())
600        .filter(|n| *n >= 1)
601        .unwrap_or(1);
602    let sessions_per_process = env::var("SELF_HOSTED_NODE_GAMES_PER_JVM")
603        .ok()
604        .and_then(|value| value.parse::<usize>().ok())
605        .filter(|n| *n >= 1)
606        .unwrap_or(1);
607    let pool = JavaEnginePool::start(&config, max_sessions, sessions_per_process)?;
608    JAVA_ENGINE
609        .set(pool)
610        .map_err(|_| "java engine already initialized".to_string())
611}
612
613#[cfg(all(feature = "graal-forge", not(feature = "java-forge")))]
614pub fn init_engine() -> Result<(), String> {
615    // Loading the card database into the isolate takes tens of seconds on a
616    // small box. Pay it here, before any room is advertised, rather than
617    // leaving it for whoever starts the first game.
618    if shared_isolate_enabled() {
619        drop(GraalEngineHandle::create(
620            &JavaRuntimeConfig::from_env().assets_dir,
621        )?);
622    }
623    Ok(())
624}
625
626#[cfg(not(forge_backend))]
627pub fn init_engine() -> Result<(), String> {
628    Err(
629        "forge engine requires building self-hosted-node with --features java-forge or graal-forge"
630            .to_string(),
631    )
632}
633
634#[cfg(feature = "java-forge")]
635fn engine_handle() -> Result<JavaEngineHandle, String> {
636    JAVA_ENGINE
637        .get()
638        .map(JavaEnginePool::handle)
639        .ok_or_else(|| "java engine is not initialized".to_string())
640}
641
642#[cfg(feature = "java-forge")]
643type ForgeEngine = JavaEngineHandle;
644
645#[cfg(feature = "java-forge")]
646fn obtain_engine() -> Result<ForgeEngine, String> {
647    engine_handle()
648}
649
650#[cfg(all(feature = "graal-forge", not(feature = "java-forge")))]
651type ForgeEngine = GraalEngineHandle;
652
653#[cfg(all(feature = "graal-forge", not(feature = "java-forge")))]
654fn obtain_engine() -> Result<ForgeEngine, String> {
655    GraalEngineHandle::create(&JavaRuntimeConfig::from_env().assets_dir)
656}
657
658#[cfg(feature = "graal-forge")]
659mod graal_ffi {
660    use std::os::raw::{c_char, c_int};
661
662    #[allow(non_camel_case_types)]
663    pub type graal_isolate_t = std::ffi::c_void;
664    #[allow(non_camel_case_types)]
665    pub type graal_isolatethread_t = std::ffi::c_void;
666
667    extern "C" {
668        pub fn graal_create_isolate(
669            params: *mut std::ffi::c_void,
670            isolate: *mut *mut graal_isolate_t,
671            thread: *mut *mut graal_isolatethread_t,
672        ) -> c_int;
673        pub fn graal_tear_down_isolate(thread: *mut graal_isolatethread_t) -> c_int;
674        pub fn graal_attach_thread(
675            isolate: *mut graal_isolate_t,
676            thread: *mut *mut graal_isolatethread_t,
677        ) -> c_int;
678        pub fn graal_detach_thread(thread: *mut graal_isolatethread_t) -> c_int;
679        pub fn forge_initialize(
680            thread: *mut graal_isolatethread_t,
681            assets_dir: *const c_char,
682        ) -> *mut c_char;
683        pub fn forge_start_game(
684            thread: *mut graal_isolatethread_t,
685            request_json: *const c_char,
686        ) -> *mut c_char;
687        pub fn forge_submit_action(
688            thread: *mut graal_isolatethread_t,
689            session_id: *const c_char,
690            action_json: *const c_char,
691        ) -> *mut c_char;
692        pub fn forge_get_prompt(
693            thread: *mut graal_isolatethread_t,
694            session_id: *const c_char,
695            player_index: c_int,
696        ) -> *mut c_char;
697        pub fn forge_get_snapshot(
698            thread: *mut graal_isolatethread_t,
699            session_id: *const c_char,
700            viewer: c_int,
701        ) -> *mut c_char;
702        pub fn forge_get_game_over(
703            thread: *mut graal_isolatethread_t,
704            session_id: *const c_char,
705        ) -> *mut c_char;
706        pub fn forge_end_game(
707            thread: *mut graal_isolatethread_t,
708            session_id: *const c_char,
709        ) -> *mut c_char;
710        pub fn forge_abort_game(
711            thread: *mut graal_isolatethread_t,
712            session_id: *const c_char,
713        ) -> *mut c_char;
714        pub fn forge_free_string(thread: *mut graal_isolatethread_t, ptr: *mut c_char);
715        pub fn forge_dump_heap(
716            thread: *mut graal_isolatethread_t,
717            path: *const c_char,
718            live_objects_only: c_int,
719        ) -> *mut c_char;
720        pub fn forge_gc_stats(thread: *mut graal_isolatethread_t) -> *mut c_char;
721    }
722}
723
724/// Isolate-wide GC and heap counters, as reported by the engine itself.
725/// `collections` and `pause_millis` are -1 when the native image exposes no
726/// collector beans; the heap figures are always present.
727#[cfg(feature = "graal-forge")]
728#[derive(serde::Deserialize)]
729#[serde(rename_all = "camelCase")]
730pub struct GcStats {
731    pub heap_used_bytes: u64,
732    pub heap_total_bytes: u64,
733    pub heap_max_bytes: u64,
734    pub collections: i64,
735    pub pause_millis: i64,
736    /// Keyed by collector name. An incremental collection is a few milliseconds;
737    /// a complete one traces the whole live set, which here is a permanently
738    /// reachable card database, so it costs the same whether it reclaims
739    /// anything or not. The totals above hide that difference.
740    #[serde(default)]
741    pub by_collector: std::collections::HashMap<String, GcCollector>,
742    /// Wall clock the engine was stopped, measured inside the isolate by sleep
743    /// overshoot rather than by the collector, because a G1 image registers no
744    /// collector beans and leaves every field above at zero. Counts scheduler
745    /// preemption as well as collection pauses.
746    #[serde(default)]
747    pub stalled_millis: u64,
748    #[serde(default)]
749    pub max_stall_millis: u64,
750    #[serde(default)]
751    pub long_stalls: u64,
752}
753
754#[cfg(feature = "graal-forge")]
755#[derive(serde::Deserialize)]
756#[serde(rename_all = "camelCase")]
757pub struct GcCollector {
758    pub collections: u64,
759    pub pause_millis: u64,
760}
761
762#[cfg(feature = "graal-forge")]
763#[derive(serde::Deserialize)]
764struct ForgeReply {
765    ok: bool,
766    #[serde(default)]
767    result: String,
768    #[serde(default)]
769    error: Option<String>,
770}
771
772// A GraalVM isolate hosts the in-process Forge engine. The `thread` handle is
773// bound to the hosted-engine thread that created or attached it (isolate
774// threads are not portable), so the handle is Rc/!Send by design. In shared
775// mode one isolate serves all rooms: the first creator initializes Forge, later
776// rooms attach their own thread to it and sessions coexist in the adapter.
777#[cfg(feature = "graal-forge")]
778struct GraalBridge {
779    thread: *mut graal_ffi::graal_isolatethread_t,
780    attached: bool,
781}
782
783#[cfg(feature = "graal-forge")]
784struct SharedIsolate(*mut graal_ffi::graal_isolate_t);
785#[cfg(feature = "graal-forge")]
786unsafe impl Send for SharedIsolate {}
787
788#[cfg(feature = "graal-forge")]
789static SHARED_GRAAL_ISOLATE: std::sync::Mutex<Option<SharedIsolate>> = std::sync::Mutex::new(None);
790
791// One isolate hosts every room's game (safe since the endstep concurrency
792// patches), so the card database loads once per node. Off keeps the historical
793// isolate-per-game shape.
794#[cfg(feature = "graal-forge")]
795fn shared_isolate_enabled() -> bool {
796    let asked = env::var("SELF_HOSTED_NODE_SHARED_ISOLATE")
797        .map(|value| value == "1" || value.eq_ignore_ascii_case("true"))
798        .unwrap_or(false);
799    // G1 permits one isolate per process and aborts the process on the second:
800    //   guarantee(SVMIsolateData::_heap_base == nullptr) failed:
801    //   G1 doesn't support multiple isolates at the moment.
802    // The published image is built with G1, and the collector is fixed at build
803    // time and not visible from here, so isolate-per-game cannot be honoured
804    // without risking that abort on the second room. One isolate is correct for
805    // a Serial build too — it is what production has run since the card
806    // database became shared — so this is forced rather than merely defaulted.
807    if !asked {
808        warn!(
809            "isolate-per-game is not available: the engine image is built with G1, \
810             which aborts on a second isolate. Hosting every room in one isolate."
811        );
812    }
813    true
814}
815
816#[cfg(feature = "graal-forge")]
817impl GraalBridge {
818    fn create() -> Result<Self, String> {
819        let mut isolate: *mut graal_ffi::graal_isolate_t = std::ptr::null_mut();
820        let mut thread: *mut graal_ffi::graal_isolatethread_t = std::ptr::null_mut();
821        let rc = unsafe {
822            graal_ffi::graal_create_isolate(std::ptr::null_mut(), &mut isolate, &mut thread)
823        };
824        if rc != 0 {
825            return Err(format!("graal_create_isolate failed with code {rc}"));
826        }
827        Ok(Self {
828            thread,
829            attached: false,
830        })
831    }
832
833    fn create_in_shared_isolate(assets_dir: &Path) -> Result<Self, String> {
834        let mut guard = SHARED_GRAAL_ISOLATE
835            .lock()
836            .map_err(|_| "shared graal isolate poisoned".to_string())?;
837        if let Some(shared) = guard.as_ref() {
838            let mut thread: *mut graal_ffi::graal_isolatethread_t = std::ptr::null_mut();
839            let rc = unsafe { graal_ffi::graal_attach_thread(shared.0, &mut thread) };
840            if rc != 0 {
841                return Err(format!("graal_attach_thread failed with code {rc}"));
842            }
843            return Ok(Self {
844                thread,
845                attached: true,
846            });
847        }
848        let mut isolate: *mut graal_ffi::graal_isolate_t = std::ptr::null_mut();
849        let mut thread: *mut graal_ffi::graal_isolatethread_t = std::ptr::null_mut();
850        let rc = unsafe {
851            graal_ffi::graal_create_isolate(std::ptr::null_mut(), &mut isolate, &mut thread)
852        };
853        if rc != 0 {
854            return Err(format!("graal_create_isolate failed with code {rc}"));
855        }
856        let mut bridge = Self {
857            thread,
858            attached: false,
859        };
860        let assets = cstring(&assets_dir.to_string_lossy())?;
861        bridge.decode(unsafe { graal_ffi::forge_initialize(bridge.thread, assets.as_ptr()) })?;
862        bridge.attached = true;
863        *guard = Some(SharedIsolate(isolate));
864        info!("shared graal isolate initialized");
865        Ok(bridge)
866    }
867
868    fn decode(&self, raw: *mut std::os::raw::c_char) -> Result<String, String> {
869        if raw.is_null() {
870            return Err("forge native lib returned null".to_string());
871        }
872        let envelope = unsafe { std::ffi::CStr::from_ptr(raw) }
873            .to_string_lossy()
874            .into_owned();
875        unsafe { graal_ffi::forge_free_string(self.thread, raw) };
876        let reply: ForgeReply = serde_json::from_str(&envelope)
877            .map_err(|err| format!("malformed forge envelope: {err}"))?;
878        if reply.ok {
879            Ok(reply.result)
880        } else {
881            Err(reply
882                .error
883                .unwrap_or_else(|| "unknown forge error".to_string()))
884        }
885    }
886}
887
888#[cfg(feature = "graal-forge")]
889impl Drop for GraalBridge {
890    fn drop(&mut self) {
891        if self.attached {
892            unsafe { graal_ffi::graal_detach_thread(self.thread) };
893        } else {
894            // Reachable only from the isolate-per-game path, which
895            // `shared_isolate_enabled` forces off. Teardown refuses while any
896            // thread is still attached, and the engine now keeps a permanent
897            // stall-probe thread inside the isolate, so restoring that path
898            // means giving the probe a way to be stopped first.
899            unsafe { graal_ffi::graal_tear_down_isolate(self.thread) };
900        }
901    }
902}
903
904#[cfg(feature = "graal-forge")]
905#[derive(Clone)]
906struct GraalEngineHandle {
907    bridge: std::rc::Rc<GraalBridge>,
908}
909
910#[cfg(feature = "graal-forge")]
911impl GraalEngineHandle {
912    fn create(assets_dir: &Path) -> Result<Self, String> {
913        let shared = shared_isolate_enabled();
914        let bridge = if shared {
915            GraalBridge::create_in_shared_isolate(assets_dir)?
916        } else {
917            let bridge = GraalBridge::create()?;
918            let assets = cstring(&assets_dir.to_string_lossy())?;
919            bridge
920                .decode(unsafe { graal_ffi::forge_initialize(bridge.thread, assets.as_ptr()) })?;
921            bridge
922        };
923        if shared {
924            start_engine_gc_sampler();
925        }
926        Ok(Self {
927            bridge: std::rc::Rc::new(bridge),
928        })
929    }
930
931    fn start_game(&self, request_json: &str) -> Result<String, String> {
932        let request = cstring(request_json)?;
933        let response = self
934            .bridge
935            .decode(unsafe { graal_ffi::forge_start_game(self.bridge.thread, request.as_ptr()) })?;
936        let parsed: StartGameResponse = serde_json::from_str(&response)
937            .map_err(|err| format!("malformed startGame response: {err}"))?;
938        Ok(parsed.session_id)
939    }
940
941    fn submit_action(&self, session_id: &str, action_json: &str) -> Result<String, String> {
942        let session = cstring(session_id)?;
943        let action = cstring(action_json)?;
944        let started = Instant::now();
945        let result = self.bridge.decode(unsafe {
946            graal_ffi::forge_submit_action(self.bridge.thread, session.as_ptr(), action.as_ptr())
947        });
948        crate::metrics::record_forge_decision_stage("submit_action", started.elapsed());
949        result
950    }
951
952    /// Cumulative collections, pause time and heap for the whole isolate. Every
953    /// room on this node shares it, so a collection here stops all of them.
954    fn gc_stats(&self) -> Result<GcStats, String> {
955        gc_stats_via(&self.bridge)
956    }
957
958    fn get_prompt(&self, session_id: &str, player_index: usize) -> Result<Option<String>, String> {
959        let session = cstring(session_id)?;
960        let prompt = self.bridge.decode(unsafe {
961            graal_ffi::forge_get_prompt(
962                self.bridge.thread,
963                session.as_ptr(),
964                player_index as std::os::raw::c_int,
965            )
966        })?;
967        Ok((!prompt.is_empty()).then_some(prompt))
968    }
969
970    fn is_game_over(&self, session_id: &str) -> Result<bool, String> {
971        let session = cstring(session_id)?;
972        let value = self.bridge.decode(unsafe {
973            graal_ffi::forge_get_game_over(self.bridge.thread, session.as_ptr())
974        })?;
975        Ok(value.trim() == "true")
976    }
977
978    fn get_snapshot(&self, session_id: &str, viewer: Option<usize>) -> Result<String, String> {
979        let session = cstring(session_id)?;
980        let viewer = viewer.map_or(-1, |v| v as std::os::raw::c_int);
981        self.bridge.decode(unsafe {
982            graal_ffi::forge_get_snapshot(self.bridge.thread, session.as_ptr(), viewer)
983        })
984    }
985
986    fn end_game(&self, session_id: &str) -> Result<(), String> {
987        let session = cstring(session_id)?;
988        self.bridge
989            .decode(unsafe { graal_ffi::forge_end_game(self.bridge.thread, session.as_ptr()) })
990            .map(|_| ())
991    }
992
993    fn abort_game(&self, session_id: &str) -> Result<(), String> {
994        let session = cstring(session_id)?;
995        self.bridge
996            .decode(unsafe { graal_ffi::forge_abort_game(self.bridge.thread, session.as_ptr()) })
997            .map(|_| ())
998    }
999}
1000
1001/// Write an hprof dump of the shared isolate's heap. Attaches its own isolate
1002/// thread, so it is callable from a signal handler task rather than the engine
1003/// thread that owns the room's bridge.
1004#[cfg(feature = "graal-forge")]
1005pub fn dump_shared_heap(path: &str) -> Result<(), String> {
1006    let isolate = SHARED_GRAAL_ISOLATE
1007        .lock()
1008        .map_err(|_| "shared graal isolate poisoned".to_string())?
1009        .as_ref()
1010        .map(|shared| shared.0)
1011        .ok_or_else(|| "no shared isolate to dump".to_string())?;
1012    let mut thread: *mut graal_ffi::graal_isolatethread_t = std::ptr::null_mut();
1013    let rc = unsafe { graal_ffi::graal_attach_thread(isolate, &mut thread) };
1014    if rc != 0 {
1015        return Err(format!("graal_attach_thread failed with code {rc}"));
1016    }
1017    let bridge = GraalBridge {
1018        thread,
1019        attached: true,
1020    };
1021    let target = cstring(path)?;
1022    bridge
1023        .decode(unsafe { graal_ffi::forge_dump_heap(bridge.thread, target.as_ptr(), 1) })
1024        .map(|_| ())
1025}
1026
1027#[cfg(feature = "graal-forge")]
1028fn cstring(value: &str) -> Result<std::ffi::CString, String> {
1029    std::ffi::CString::new(value).map_err(|_| "string contained interior NUL".to_string())
1030}
1031
1032#[cfg(feature = "java-forge")]
1033pub fn run_concurrent_self_play(
1034    seats: &[DeckSelection],
1035    starting_life: i32,
1036    seed: u64,
1037    max_prompts: usize,
1038    concurrency: usize,
1039) -> Result<(), String> {
1040    let config = JavaRuntimeConfig::from_env();
1041    let games_per_process = env::var("SELF_HOSTED_NODE_GAMES_PER_JVM")
1042        .ok()
1043        .and_then(|value| value.parse::<usize>().ok())
1044        .filter(|n| *n >= 1)
1045        .unwrap_or(1);
1046    let pool = JavaEnginePool::start(&config, concurrency.max(1), games_per_process)?;
1047    info!(
1048        concurrency,
1049        "java-engine started; launching concurrent games"
1050    );
1051
1052    let mut players = Vec::with_capacity(seats.len());
1053    for (i, seat) in seats.iter().enumerate() {
1054        let identities = deck_card_identities(&seat.deck);
1055        players.push(PlayerConfig::new(
1056            format!("Self-Play {}", i + 1),
1057            &identities,
1058            commander_names_for_java(&seat.deck, seat.commander_name.as_deref()),
1059        ));
1060    }
1061
1062    let mut joins = Vec::with_capacity(concurrency.max(1));
1063    for game_index in 0..concurrency.max(1) {
1064        let handle = pool.handle();
1065        let request = StartGameRequest::new(
1066            format!("self-hosted-java-concurrent-{game_index}"),
1067            String::new(),
1068            starting_life,
1069            seed.wrapping_add(game_index as u64),
1070            players.clone(),
1071        );
1072        joins.push(std::thread::spawn(move || -> Result<(), String> {
1073            let request_json = request.to_json().map_err(|error| error.to_string())?;
1074            let session_id = handle.start_game(&request_json)?;
1075            info!(session_id, game_index, "concurrent java game started");
1076            let result = drive_game_via_handle(&handle, &session_id, max_prompts);
1077            let _ = handle.end_game(&session_id);
1078            result
1079        }));
1080    }
1081
1082    let mut outcome = Ok(());
1083    for join in joins {
1084        match join.join() {
1085            Ok(Ok(())) => {}
1086            Ok(Err(error)) => outcome = Err(error),
1087            Err(_) => outcome = Err("concurrent game thread panicked".to_string()),
1088        }
1089    }
1090    outcome
1091}
1092
1093#[cfg(not(feature = "java-forge"))]
1094pub fn run_concurrent_self_play(
1095    _seats: &[DeckSelection],
1096    _starting_life: i32,
1097    _seed: u64,
1098    _max_prompts: usize,
1099    _concurrency: usize,
1100) -> Result<(), String> {
1101    Err(
1102        "java-forge concurrent self-play requires building self-hosted-node with --features java-forge"
1103            .to_string(),
1104    )
1105}
1106
1107#[cfg(all(feature = "graal-forge", not(feature = "java-forge")))]
1108fn report_engine_gc(engine: &ForgeEngine) {
1109    match engine.gc_stats() {
1110        Ok(stats) => publish_engine_gc(&stats),
1111        Err(err) => debug!(%err, "engine gc stats unavailable"),
1112    }
1113}
1114
1115#[cfg(feature = "graal-forge")]
1116fn publish_engine_gc(stats: &GcStats) {
1117    crate::metrics::record_engine_gc(
1118        stats.collections,
1119        stats.pause_millis,
1120        stats.heap_used_bytes,
1121        stats.heap_max_bytes,
1122    );
1123    crate::metrics::record_engine_stall(
1124        stats.stalled_millis,
1125        stats.max_stall_millis,
1126        stats.long_stalls,
1127    );
1128    for (name, per) in &stats.by_collector {
1129        crate::metrics::record_engine_gc_collector(name, per.collections, per.pause_millis);
1130    }
1131}
1132
1133#[cfg(feature = "graal-forge")]
1134fn gc_stats_via(bridge: &GraalBridge) -> Result<GcStats, String> {
1135    let raw = bridge.decode(unsafe { graal_ffi::forge_gc_stats(bridge.thread) })?;
1136    serde_json::from_str(&raw).map_err(|err| format!("malformed gcStats response: {err}"))
1137}
1138
1139#[cfg(feature = "graal-forge")]
1140const ENGINE_GC_SAMPLE_INTERVAL: Duration = Duration::from_secs(15);
1141
1142/// Sampling on the decision path made an idle node and a wedged node look the
1143/// same: both stop reporting. A node that is stalling is exactly the node that
1144/// stops completing decisions, so the sample has to come from a thread that is
1145/// not on that path. It attaches its own isolate thread because the handles the
1146/// rooms hold are bound to their creating thread.
1147#[cfg(feature = "graal-forge")]
1148fn start_engine_gc_sampler() {
1149    static STARTED: std::sync::Once = std::sync::Once::new();
1150    STARTED.call_once(|| {
1151        let spawned = std::thread::Builder::new()
1152            .name("engine-gc-sampler".to_string())
1153            .spawn(|| {
1154                let isolate = match SHARED_GRAAL_ISOLATE.lock() {
1155                    Ok(guard) => guard.as_ref().map(|shared| shared.0),
1156                    Err(_) => None,
1157                };
1158                let Some(isolate) = isolate else {
1159                    return;
1160                };
1161                let mut thread: *mut graal_ffi::graal_isolatethread_t = std::ptr::null_mut();
1162                let rc = unsafe { graal_ffi::graal_attach_thread(isolate, &mut thread) };
1163                if rc != 0 {
1164                    warn!(rc, "gc sampler could not attach to the engine isolate");
1165                    return;
1166                }
1167                let bridge = GraalBridge {
1168                    thread,
1169                    attached: true,
1170                };
1171                loop {
1172                    std::thread::sleep(ENGINE_GC_SAMPLE_INTERVAL);
1173                    match gc_stats_via(&bridge) {
1174                        Ok(stats) => publish_engine_gc(&stats),
1175                        Err(err) => debug!(%err, "engine gc stats unavailable"),
1176                    }
1177                }
1178            });
1179        if let Err(err) = spawned {
1180            warn!(%err, "could not start the engine gc sampler");
1181        }
1182    });
1183}
1184
1185/// The subprocess backend already reports this by parsing its own GC log.
1186#[cfg(feature = "java-forge")]
1187fn report_engine_gc(_engine: &ForgeEngine) {}
1188
1189#[cfg(feature = "java-forge")]
1190fn drive_game_via_handle(
1191    handle: &JavaEngineHandle,
1192    session_id: &str,
1193    max_prompts: usize,
1194) -> Result<(), String> {
1195    let mut bots: HashMap<usize, SimpleAi> = HashMap::new();
1196    let mut last_prompt: Option<String> = None;
1197    let mut acted = 0usize;
1198    let mut seen_prompt = false;
1199    let max_iterations = max_prompts.saturating_mul(200).max(2_000);
1200
1201    for _ in 0..max_iterations {
1202        if let Some(prompt_json) = handle.get_prompt(session_id, 0)? {
1203            seen_prompt = true;
1204            if last_prompt.as_deref() == Some(prompt_json.as_str()) {
1205                if handle.is_game_over(session_id)? {
1206                    return Ok(());
1207                }
1208                std::thread::sleep(Duration::from_millis(20));
1209                continue;
1210            }
1211            let prompt: AgentPrompt = serde_json::from_str(&prompt_json)
1212                .map_err(|error| format!("failed to parse concurrent prompt: {error}"))?;
1213            let player = player_index(&prompt.deciding_player_id);
1214            if let Some(action) = bots.entry(player).or_default().decide(prompt) {
1215                let action_json = serde_json::to_string(&action).map_err(|err| err.to_string())?;
1216                handle.submit_action(session_id, &action_json)?;
1217                acted += 1;
1218                if acted >= max_prompts {
1219                    return Err(format!(
1220                        "concurrent game {session_id} did not finish within {max_prompts} decisions"
1221                    ));
1222                }
1223            }
1224            last_prompt = Some(prompt_json);
1225            continue;
1226        }
1227        if seen_prompt && handle.is_game_over(session_id)? {
1228            return Ok(());
1229        }
1230        std::thread::sleep(Duration::from_millis(20));
1231    }
1232    Err(format!(
1233        "concurrent game {session_id} exceeded its iteration cap"
1234    ))
1235}
1236
1237/// G1 keeps pauses roughly bounded as the live set grows; Serial does not.
1238const DEFAULT_JAVA_COLLECTOR: &str = "G1";
1239const DEFAULT_JAVA_GC_LOG: &str = "stderr";
1240const DEFAULT_JAVA_HEAP_MB: u64 = 1024;
1241const DEFAULT_JAVA_ACTIVE_PROCESSORS: u64 = 2;
1242
1243#[derive(Debug, Clone)]
1244pub struct JavaRuntimeConfig {
1245    pub assets_dir: PathBuf,
1246    pub harness_jar: PathBuf,
1247    pub java_home: Option<PathBuf>,
1248    pub extra_classpath: Vec<PathBuf>,
1249    pub heap_mb: Option<u64>,
1250    pub active_processor_count: Option<u64>,
1251    pub gc_log: Option<String>,
1252    pub collector: Option<String>,
1253    pub extra_jvm_args: Vec<String>,
1254}
1255
1256impl JavaRuntimeConfig {
1257    pub fn from_env() -> Self {
1258        let root = workspace_root();
1259        Self {
1260            assets_dir: env_path("SELF_HOSTED_NODE_FORGE_ASSETS_DIR")
1261                .or_else(|| env_path("MANA_BREW_FORGE_ASSETS_DIR"))
1262                .unwrap_or_else(|| root.join("forge/forge-gui")),
1263            harness_jar: env_path("SELF_HOSTED_NODE_FORGE_HARNESS_JAR")
1264                .or_else(|| env_path("MANA_BREW_FORGE_HARNESS_JAR"))
1265                .unwrap_or_else(|| {
1266                    root.join("forge-harness/target/forge-harness-jar-with-dependencies.jar")
1267                }),
1268            java_home: env_path("SELF_HOSTED_NODE_JAVA_HOME")
1269                .or_else(|| env_path("MANA_BREW_JAVA_HOME"))
1270                .or_else(|| env_path("JAVA_HOME")),
1271            extra_classpath: env_classpath("SELF_HOSTED_NODE_FORGE_EXTRA_CLASSPATH")
1272                .into_iter()
1273                .chain(env_classpath("MANA_BREW_FORGE_EXTRA_CLASSPATH"))
1274                .collect(),
1275            heap_mb: env_sizing("SELF_HOSTED_NODE_JAVA_HEAP_MB", DEFAULT_JAVA_HEAP_MB),
1276            active_processor_count: env_sizing(
1277                "SELF_HOSTED_NODE_JAVA_ACTIVE_PROCESSORS",
1278                DEFAULT_JAVA_ACTIVE_PROCESSORS,
1279            ),
1280            collector: env::var("SELF_HOSTED_NODE_JAVA_COLLECTOR")
1281                .ok()
1282                .map(|value| value.trim().to_string())
1283                .filter(|value| !value.is_empty())
1284                .or_else(|| Some(DEFAULT_JAVA_COLLECTOR.to_string())),
1285            // Defaults on: the fleet ships no logs, so this log exists to be
1286            // turned into metrics by the stderr pump (see metrics.rs).
1287            gc_log: env::var("SELF_HOSTED_NODE_JAVA_GC_LOG")
1288                .ok()
1289                .map(|value| value.trim().to_string())
1290                .filter(|value| !value.is_empty())
1291                .or_else(|| Some(DEFAULT_JAVA_GC_LOG.to_string())),
1292            extra_jvm_args: env::var("SELF_HOSTED_NODE_JAVA_OPTS")
1293                .unwrap_or_default()
1294                .split_whitespace()
1295                .map(str::to_string)
1296                .collect(),
1297        }
1298    }
1299
1300    #[cfg(feature = "java-forge")]
1301    fn jvm_args(&self) -> Vec<String> {
1302        let mut args = vec![
1303            "-Dfile.encoding=UTF-8".to_string(),
1304            "-Dsun.stdout.encoding=UTF-8".to_string(),
1305            "-Dsun.stderr.encoding=UTF-8".to_string(),
1306            "-Djava.awt.headless=true".to_string(),
1307            // Forge's endstep concurrency patches assume a heap exhaustion kills the process
1308            // instead of thrashing; a supervised node is restarted, a thrashing one is not.
1309            "-XX:+ExitOnOutOfMemoryError".to_string(),
1310            // Forge calls System.gc() on match boundaries (Match.java,
1311            // HostedMatch.java) and so does the harness. Each one is a full
1312            // collection, and a full collection costs about 1.3ms per MB of
1313            // live set, so on a large board they are seconds of stop-the-world
1314            // for no benefit the collector would not have reached anyway.
1315            "-XX:+DisableExplicitGC".to_string(),
1316        ];
1317        // Pause time is linear in the live set and Serial collects on one
1318        // thread, so a bigger heap under Serial means longer freezes, not
1319        // shorter ones. Name the collector rather than inheriting whatever the
1320        // host's environment happens to set.
1321        if let Some(collector) = &self.collector {
1322            args.push(format!("-XX:+Use{collector}GC"));
1323        }
1324        // A fleet runs many of these side by side, and every JVM sizes its heap and its GC
1325        // thread count from the whole machine unless told otherwise.
1326        if let Some(heap_mb) = self.heap_mb {
1327            args.push(format!("-Xmx{heap_mb}m"));
1328        }
1329        if let Some(processors) = self.active_processor_count {
1330            args.push(format!("-XX:ActiveProcessorCount={processors}"));
1331        }
1332        // In a container the only log that reaches Loki is the node's own, and the
1333        // subprocess owns stdout for the protocol, so "stderr" routes GC through the
1334        // stderr pump instead of a file nothing ships.
1335        match self.gc_log.as_deref() {
1336            None => {}
1337            Some("stderr") => args.push("-Xlog:gc*:stderr:time,uptime,level,tags".to_string()),
1338            Some(dir) => args.push(format!(
1339                "-Xlog:gc*:file={}:time,uptime,level,tags:filecount=5,filesize=20M",
1340                Path::new(dir).join("engine-gc-%p.log").display()
1341            )),
1342        }
1343        args.extend(self.extra_jvm_args.iter().cloned());
1344        args
1345    }
1346
1347    pub fn validate(&self) -> Result<(), String> {
1348        require_dir(&self.assets_dir, "Forge assets directory")?;
1349        require_file(&self.harness_jar, "Forge harness jar")?;
1350        if let Some(java_home) = &self.java_home {
1351            require_dir(java_home, "Java home")?;
1352        }
1353        for entry in &self.extra_classpath {
1354            if !entry.exists() {
1355                return Err(format!(
1356                    "Classpath entry does not exist: {}",
1357                    entry.display()
1358                ));
1359            }
1360        }
1361        Ok(())
1362    }
1363
1364    pub fn classpath_entries(&self) -> Vec<PathBuf> {
1365        let mut entries = Vec::with_capacity(1 + self.extra_classpath.len());
1366        entries.push(self.harness_jar.clone());
1367        entries.extend(self.extra_classpath.iter().cloned());
1368        entries
1369    }
1370}
1371
1372#[cfg(forge_backend)]
1373#[allow(clippy::too_many_arguments)]
1374pub fn run_hosted_engine_game(
1375    game_id: String,
1376    player_names: Vec<String>,
1377    decks: Vec<Deck>,
1378    commander_names: Vec<Option<String>>,
1379    commander_variant: bool,
1380    game_variant: String,
1381    local_player_index: Option<usize>,
1382    ai_player_indices: Vec<usize>,
1383    starting_life: i32,
1384    remote_prompt_tx: std_mpsc::Sender<(usize, AgentMessage)>,
1385    remote_response_rxs: Vec<(usize, std_mpsc::Receiver<ClientToServerMessage>)>,
1386    game_over_tx: std_mpsc::Sender<HostedGameOver>,
1387    cancel: Arc<AtomicBool>,
1388) -> Result<(), String> {
1389    run_hosted_engine_game_inner(
1390        game_id,
1391        player_names,
1392        decks,
1393        commander_names,
1394        commander_variant,
1395        game_variant,
1396        local_player_index,
1397        ai_player_indices,
1398        starting_life,
1399        remote_prompt_tx,
1400        remote_response_rxs,
1401        game_over_tx,
1402        cancel,
1403    )
1404}
1405
1406#[cfg(not(forge_backend))]
1407#[allow(clippy::too_many_arguments)]
1408pub fn run_hosted_engine_game(
1409    _game_id: String,
1410    _player_names: Vec<String>,
1411    _decks: Vec<Deck>,
1412    _commander_names: Vec<Option<String>>,
1413    _commander_variant: bool,
1414    _game_variant: String,
1415    _local_player_index: Option<usize>,
1416    _ai_player_indices: Vec<usize>,
1417    _starting_life: i32,
1418    _remote_prompt_tx: std_mpsc::Sender<(usize, AgentMessage)>,
1419    _remote_response_rxs: Vec<(usize, std_mpsc::Receiver<ClientToServerMessage>)>,
1420    _game_over_tx: std_mpsc::Sender<HostedGameOver>,
1421    _cancel: Arc<AtomicBool>,
1422) -> Result<(), String> {
1423    Err(unsupported_message().to_string())
1424}
1425
1426#[cfg(forge_backend)]
1427const SLOW_DECISION: Duration = Duration::from_secs(10);
1428
1429/// Serde tag of the prompt variant, for the slow-decision log. `PromptInput` is
1430/// internally tagged, so this is the same string the captures carry. Only called
1431/// on the rare slow path, so the trip through a `Value` is not worth avoiding.
1432#[cfg(forge_backend)]
1433fn prompt_input_name(input: &PromptInput) -> String {
1434    serde_json::to_value(input)
1435        .ok()
1436        .and_then(|value| {
1437            value
1438                .get("type")
1439                .and_then(serde_json::Value::as_str)
1440                .map(str::to_owned)
1441        })
1442        .unwrap_or_else(|| "unknown".to_string())
1443}
1444
1445#[cfg(forge_backend)]
1446#[allow(clippy::too_many_arguments)]
1447fn run_hosted_engine_game_inner(
1448    game_id: String,
1449    player_names: Vec<String>,
1450    decks: Vec<Deck>,
1451    commander_names: Vec<Option<String>>,
1452    commander_variant: bool,
1453    game_variant: String,
1454    local_player_index: Option<usize>,
1455    ai_player_indices: Vec<usize>,
1456    starting_life: i32,
1457    remote_prompt_tx: std_mpsc::Sender<(usize, AgentMessage)>,
1458    remote_response_rxs: Vec<(usize, std_mpsc::Receiver<ClientToServerMessage>)>,
1459    game_over_tx: std_mpsc::Sender<HostedGameOver>,
1460    cancel: Arc<AtomicBool>,
1461) -> Result<(), String> {
1462    let engine = obtain_engine()?;
1463
1464    let mut players = Vec::with_capacity(player_names.len());
1465    for (index, name) in player_names.iter().enumerate() {
1466        let identities = deck_card_identities(&decks[index]);
1467        let seat_commander_names = if commander_variant {
1468            commander_names_for_java(&decks[index], commander_names[index].as_deref())
1469        } else {
1470            Vec::new()
1471        };
1472        players.push(PlayerConfig::new(
1473            name.clone(),
1474            &identities,
1475            seat_commander_names,
1476        ));
1477    }
1478    for &idx in &ai_player_indices {
1479        if let Some(player) = players.get_mut(idx) {
1480            player.ai = true;
1481        }
1482    }
1483    let request = StartGameRequest::new(
1484        game_id.clone(),
1485        game_variant,
1486        starting_life,
1487        rand::random(),
1488        players,
1489    );
1490    let session_id = engine.start_game(&request.to_json().map_err(|err| err.to_string())?)?;
1491    info!(game_id, session_id, "hosted java-forge session started");
1492
1493    struct SessionGuard {
1494        engine: ForgeEngine,
1495        session_id: String,
1496        armed: std::cell::Cell<bool>,
1497    }
1498    impl Drop for SessionGuard {
1499        fn drop(&mut self) {
1500            if self.armed.get() {
1501                if let Err(error) = self.engine.abort_game(&self.session_id) {
1502                    warn!(session_id = %self.session_id, %error, "failed to abort java session; context may leak");
1503                }
1504            }
1505        }
1506    }
1507    let guard = SessionGuard {
1508        engine: engine.clone(),
1509        session_id: session_id.clone(),
1510        armed: std::cell::Cell::new(true),
1511    };
1512
1513    let mut remote_response_rxs: HashMap<usize, std_mpsc::Receiver<ClientToServerMessage>> =
1514        remote_response_rxs.into_iter().collect();
1515    let mut last_prompt: Option<AgentPrompt> = None;
1516    let mut pending_roll_acks: usize = 0;
1517    let mut decision_received: Option<Instant> = None;
1518    let mut decision_submitted: Option<Instant> = None;
1519
1520    loop {
1521        if cancel.load(std::sync::atomic::Ordering::Relaxed) {
1522            info!(
1523                session_id,
1524                "hosted java-forge session cancelled; player left the game"
1525            );
1526            return Ok(());
1527        }
1528        for (player_index, rx) in &mut remote_response_rxs {
1529            loop {
1530                match rx.try_recv() {
1531                    Ok(ClientToServerMessage::Response {
1532                        action: PromptOutput::DiceRolled(DiceRolledOutput::DiceRolledAcknowledged),
1533                        ..
1534                    }) => {
1535                        if pending_roll_acks > 0 {
1536                            pending_roll_acks -= 1;
1537                            if pending_roll_acks == 0 {
1538                                let ack = serde_json::to_string(&PromptOutput::DiceRolled(
1539                                    DiceRolledOutput::DiceRolledAcknowledged,
1540                                ))
1541                                .map_err(|err| format!("failed to serialize roll ack: {err}"))?;
1542                                engine.submit_action(&session_id, &ack)?;
1543                            }
1544                        }
1545                    }
1546                    Ok(ClientToServerMessage::Response { prompt_id, action }) => {
1547                        // prompt_id 0 is transport-synthesized: the
1548                        // absent-player default, exempt from validation.
1549                        if prompt_id != 0 {
1550                            let Some(prompt) =
1551                                last_prompt.as_ref().filter(|p| p.prompt_id == prompt_id)
1552                            else {
1553                                reject_response(
1554                                    &remote_prompt_tx,
1555                                    *player_index,
1556                                    last_prompt.as_ref().filter(|p| {
1557                                        self::player_index(&p.deciding_player_id) == *player_index
1558                                    }),
1559                                    ProtocolErrorCode::StalePrompt,
1560                                    format!("response for prompt {prompt_id} is not open"),
1561                                );
1562                                continue;
1563                            };
1564                            if self::player_index(&prompt.deciding_player_id) != *player_index {
1565                                reject_response(
1566                                    &remote_prompt_tx,
1567                                    *player_index,
1568                                    None,
1569                                    ProtocolErrorCode::WrongPlayer,
1570                                    format!(
1571                                        "prompt {prompt_id} is for {}",
1572                                        prompt.deciding_player_id
1573                                    ),
1574                                );
1575                                continue;
1576                            }
1577                            match prompt.input.validate_response(&action) {
1578                                Ok(()) => {}
1579                                Err(ResponseViolation::WrongPromptType) => {
1580                                    reject_response(
1581                                        &remote_prompt_tx,
1582                                        *player_index,
1583                                        Some(prompt),
1584                                        ProtocolErrorCode::WrongPromptType,
1585                                        "response output does not match the prompt type"
1586                                            .to_string(),
1587                                    );
1588                                    continue;
1589                                }
1590                                Err(ResponseViolation::UnknownActionId(id)) => {
1591                                    reject_response(
1592                                        &remote_prompt_tx,
1593                                        *player_index,
1594                                        Some(prompt),
1595                                        ProtocolErrorCode::UnknownActionId,
1596                                        format!(
1597                                            "action id {id:?} was not advertised by the prompt"
1598                                        ),
1599                                    );
1600                                    continue;
1601                                }
1602                                Err(ResponseViolation::CancelNotAllowed) => {
1603                                    reject_response(
1604                                        &remote_prompt_tx,
1605                                        *player_index,
1606                                        Some(prompt),
1607                                        ProtocolErrorCode::CancelNotAllowed,
1608                                        "this prompt is not cancellable".to_string(),
1609                                    );
1610                                    continue;
1611                                }
1612                            }
1613                        }
1614                        decision_received = Some(Instant::now());
1615                        let action_json = serde_json::to_string(&action).map_err(|err| {
1616                            format!(
1617                                "failed to serialize prompt output for player {player_index}: {err}"
1618                            )
1619                        })?;
1620                        debug!(player_index, %action_json, "submitting remote response to java");
1621                        engine.submit_action(&session_id, &action_json)?;
1622                        decision_submitted = Some(Instant::now());
1623                    }
1624                    Ok(ClientToServerMessage::Directive {
1625                        directive: DirectiveInput::Concede,
1626                    }) => {
1627                        // A directive can arrive while another player's prompt
1628                        // is open (this loop drains every seat), so it names
1629                        // its seat.
1630                        let directive_json = directive_concede_json(*player_index);
1631                        debug!(player_index, %directive_json, "submitting concede directive to java");
1632                        engine.submit_action(&session_id, &directive_json)?;
1633                    }
1634                    Err(TryRecvError::Empty) => break,
1635                    Err(TryRecvError::Disconnected) => {
1636                        debug!(player_index, "java-forge response channel disconnected");
1637                        break;
1638                    }
1639                }
1640            }
1641        }
1642
1643        if let Some(prompt_json) = engine.get_prompt(&session_id, 0)? {
1644            let prompt: AgentPrompt = serde_json::from_str(&prompt_json)
1645                .map_err(|err| format!("failed to parse java prompt: {err}"))?;
1646            if last_prompt.as_ref().map(|p| p.prompt_id) != Some(prompt.prompt_id) {
1647                if let Some(started) = decision_submitted.take() {
1648                    crate::metrics::record_forge_decision_stage("next_prompt", started.elapsed());
1649                }
1650                if let Some(started) = decision_received.take() {
1651                    let elapsed = started.elapsed();
1652                    crate::metrics::record_forge_decision_stage("decision_total", elapsed);
1653                    crate::metrics::record_forge_decision(player_names.len(), elapsed);
1654                    // The metric alone cannot say which game was slow, and these
1655                    // are rare enough that finding one afterwards meant replaying
1656                    // captures by hand. Named so a log query can list them.
1657                    if elapsed >= SLOW_DECISION {
1658                        warn!(
1659                            game_id,
1660                            session_id,
1661                            seats = player_names.len(),
1662                            bots = ai_player_indices.len(),
1663                            elapsed_ms = elapsed.as_millis() as u64,
1664                            prompt = prompt_input_name(&prompt.input),
1665                            "slow engine decision"
1666                        );
1667                    }
1668                    report_engine_gc(&engine);
1669                }
1670                last_prompt = Some(prompt.clone());
1671                let player = player_index(&prompt.deciding_player_id);
1672                debug!(player, "forwarding java prompt to remote");
1673                if matches!(prompt.input, PromptInput::DiceRolled(_)) {
1674                    let snapshots_started = Instant::now();
1675                    let prompt_msg = AgentMessage::Prompt(prompt);
1676                    for &agent_index in remote_response_rxs.keys() {
1677                        let state = AgentMessage::State(state_via_handle(
1678                            &engine,
1679                            &session_id,
1680                            Some(agent_index),
1681                        )?);
1682                        let _ = remote_prompt_tx.send((agent_index, state));
1683                        let _ = remote_prompt_tx.send((agent_index, prompt_msg.clone()));
1684                    }
1685                    send_observer_state(&engine, &session_id, &remote_prompt_tx);
1686                    pending_roll_acks = remote_response_rxs.len();
1687                    if pending_roll_acks == 0 {
1688                        let ack = serde_json::to_string(&PromptOutput::DiceRolled(
1689                            DiceRolledOutput::DiceRolledAcknowledged,
1690                        ))
1691                        .map_err(|err| format!("failed to serialize roll ack: {err}"))?;
1692                        engine.submit_action(&session_id, &ack)?;
1693                    }
1694                    crate::metrics::record_forge_decision_stage(
1695                        "snapshots",
1696                        snapshots_started.elapsed(),
1697                    );
1698                } else if Some(player) == local_player_index {
1699                    if let Some(output) = auto_action(&prompt) {
1700                        let action_json = serde_json::to_string(&output)
1701                            .map_err(|err| format!("failed to serialize auto action: {err}"))?;
1702                        engine.submit_action(&session_id, &action_json)?;
1703                    }
1704                } else {
1705                    let snapshots_started = Instant::now();
1706                    for &agent_index in remote_response_rxs.keys() {
1707                        let state = AgentMessage::State(state_via_handle(
1708                            &engine,
1709                            &session_id,
1710                            Some(agent_index),
1711                        )?);
1712                        if remote_prompt_tx.send((agent_index, state)).is_err() {
1713                            return Ok(());
1714                        }
1715                    }
1716                    send_observer_state(&engine, &session_id, &remote_prompt_tx);
1717                    let prompt_msg = AgentMessage::Prompt(prompt);
1718                    if remote_prompt_tx.send((player, prompt_msg)).is_err() {
1719                        return Ok(());
1720                    }
1721                    crate::metrics::record_forge_decision_stage(
1722                        "snapshots",
1723                        snapshots_started.elapsed(),
1724                    );
1725                }
1726            }
1727        }
1728
1729        if engine.is_game_over(&session_id)? {
1730            info!("hosted java-forge session reached game over");
1731            let mut final_messages = Vec::new();
1732            for &agent_index in remote_response_rxs.keys() {
1733                match state_via_handle(&engine, &session_id, Some(agent_index)) {
1734                    Ok(state_update) => {
1735                        final_messages.push((agent_index, AgentMessage::State(state_update)));
1736                    }
1737                    Err(error) => {
1738                        warn!(%error, agent_index, "game over: final snapshot unavailable; sending game-over prompt only");
1739                    }
1740                }
1741            }
1742            if let Ok(state_update) = state_via_handle(&engine, &session_id, None) {
1743                final_messages.push((
1744                    crate::host::OBSERVER_SEAT,
1745                    AgentMessage::State(state_update),
1746                ));
1747            }
1748            let game_over = AgentMessage::Prompt(game_over_prompt());
1749            for &agent_index in remote_response_rxs.keys() {
1750                final_messages.push((agent_index, game_over.clone()));
1751            }
1752            let _ = game_over_tx.send(HostedGameOver {
1753                game_id: game_id.clone(),
1754                messages: final_messages,
1755            });
1756            engine.end_game(&session_id)?;
1757            guard.armed.set(false);
1758            return Ok(());
1759        }
1760
1761        std::thread::sleep(Duration::from_millis(50));
1762    }
1763}
1764
1765#[cfg(feature = "java-forge")]
1766fn wait_for_prompt<B: JavaBridge>(
1767    session: &mut JavaForgeSession<B>,
1768    max_polls: usize,
1769) -> Result<Option<String>, String> {
1770    for _ in 0..max_polls {
1771        if let Some(prompt) = session.get_prompt(0)? {
1772            return Ok(Some(prompt));
1773        }
1774        std::thread::sleep(Duration::from_millis(50));
1775    }
1776    Ok(None)
1777}
1778
1779#[cfg(forge_backend)]
1780fn send_observer_state(
1781    engine: &ForgeEngine,
1782    session_id: &str,
1783    remote_prompt_tx: &std_mpsc::Sender<(usize, AgentMessage)>,
1784) {
1785    match state_via_handle(engine, session_id, None) {
1786        Ok(state_update) => {
1787            let _ = remote_prompt_tx.send((
1788                crate::host::OBSERVER_SEAT,
1789                AgentMessage::State(state_update),
1790            ));
1791        }
1792        Err(error) => warn!(%error, "observer snapshot unavailable"),
1793    }
1794}
1795
1796#[cfg(forge_backend)]
1797fn player_index(deciding_player_id: &str) -> usize {
1798    deciding_player_id
1799        .strip_prefix("player-")
1800        .and_then(|n| n.parse().ok())
1801        .unwrap_or(0)
1802}
1803
1804#[cfg(forge_backend)]
1805fn reject_response(
1806    remote_prompt_tx: &std_mpsc::Sender<(usize, AgentMessage)>,
1807    seat: usize,
1808    reopen_prompt: Option<&AgentPrompt>,
1809    code: ProtocolErrorCode,
1810    message: String,
1811) {
1812    let _ = remote_prompt_tx.send((
1813        seat,
1814        AgentMessage::Error(ProtocolError {
1815            code,
1816            message,
1817            prompt_id: reopen_prompt.map(|p| p.prompt_id),
1818        }),
1819    ));
1820    if let Some(prompt) = reopen_prompt {
1821        let _ = remote_prompt_tx.send((seat, AgentMessage::Prompt(prompt.clone())));
1822    }
1823}
1824
1825#[cfg(forge_backend)]
1826fn auto_action(prompt: &AgentPrompt) -> Option<PromptOutput> {
1827    match prompt.input {
1828        PromptInput::ChooseAction(_) => {
1829            Some(PromptOutput::ChooseAction(ChooseActionOutput::Pass {
1830                until: None,
1831                exhaust_stack: false,
1832            }))
1833        }
1834        _ => None,
1835    }
1836}
1837
1838#[cfg(forge_backend)]
1839fn game_over_prompt() -> AgentPrompt {
1840    AgentPrompt {
1841        prompt_id: u32::MAX,
1842        deciding_player_id: "player-0".to_string(),
1843        source_card: None,
1844        input: PromptInput::GameOver(GameOverInput {}),
1845    }
1846}
1847
1848#[cfg(forge_backend)]
1849fn state_via_handle(
1850    engine: &ForgeEngine,
1851    session_id: &str,
1852    viewer: Option<usize>,
1853) -> Result<StateUpdate, String> {
1854    let game_view: GameViewDto = serde_json::from_str(&engine.get_snapshot(session_id, viewer)?)
1855        .map_err(|err| format!("failed to parse java snapshot: {err}"))?;
1856    Ok(StateUpdate { game_view })
1857}
1858
1859#[cfg(forge_backend)]
1860fn deck_card_identities(deck: &Deck) -> Vec<DeckCardIdentity> {
1861    deck.cards
1862        .iter()
1863        .chain(deck.commanders.iter().flatten())
1864        .map(|card| card.identity.clone())
1865        .collect()
1866}
1867
1868#[cfg(forge_backend)]
1869fn commander_names_for_java(deck: &Deck, fallback: Option<&str>) -> Vec<String> {
1870    let names: Vec<String> = deck
1871        .commanders
1872        .iter()
1873        .flatten()
1874        .map(|card| java_card_name(&card.identity.name))
1875        .collect();
1876    if !names.is_empty() {
1877        return names;
1878    }
1879    fallback
1880        .filter(|name| !name.is_empty())
1881        .map(|name| vec![java_card_name(name)])
1882        .unwrap_or_default()
1883}
1884
1885#[cfg(forge_backend)]
1886fn smoke_deck(land_name: &str, spell_name: &str) -> Vec<DeckCardIdentity> {
1887    (0..24)
1888        .map(|_| DeckCardIdentity {
1889            name: land_name.to_string(),
1890            ..Default::default()
1891        })
1892        .chain((0..36).map(|_| DeckCardIdentity {
1893            name: spell_name.to_string(),
1894            ..Default::default()
1895        }))
1896        .collect()
1897}
1898
1899#[cfg(feature = "java-forge")]
1900fn scenario_deck(land_name: &str) -> Vec<DeckCardIdentity> {
1901    (0..60)
1902        .map(|_| DeckCardIdentity {
1903            name: land_name.to_string(),
1904            ..Default::default()
1905        })
1906        .collect()
1907}
1908
1909#[cfg(feature = "java-forge")]
1910enum JavaScenario {
1911    KeepAndPlayLand {
1912        played_land: bool,
1913    },
1914    MulliganOncePlayLand {
1915        mulliganed: bool,
1916        kept_second_hand: bool,
1917        put_back_done: bool,
1918        played_land: bool,
1919    },
1920}
1921
1922#[cfg(feature = "java-forge")]
1923impl JavaScenario {
1924    fn from_name(name: &str) -> Result<Self, String> {
1925        match name {
1926            "keep-and-play-land" => Ok(Self::KeepAndPlayLand { played_land: false }),
1927            "mulligan-once-play-land" => Ok(Self::MulliganOncePlayLand {
1928                mulliganed: false,
1929                kept_second_hand: false,
1930                put_back_done: false,
1931                played_land: false,
1932            }),
1933            _ => Err(format!(
1934                "unknown java-forge scenario '{name}'. Supported scenarios: keep-and-play-land, mulligan-once-play-land"
1935            )),
1936        }
1937    }
1938
1939    fn name(&self) -> &'static str {
1940        match self {
1941            Self::KeepAndPlayLand { .. } => "keep-and-play-land",
1942            Self::MulliganOncePlayLand { .. } => "mulligan-once-play-land",
1943        }
1944    }
1945
1946    fn next_action(
1947        &mut self,
1948        prompt: &Value,
1949        game_view: &GameViewDto,
1950    ) -> Result<Option<PromptOutput>, String> {
1951        match self {
1952            Self::KeepAndPlayLand { played_land } => {
1953                if *played_land && battlefield_contains(game_view, "Swamp") {
1954                    return Ok(None);
1955                }
1956                match prompt_type(prompt) {
1957                    Some("mulligan") => Ok(Some(PromptOutput::Mulligan(
1958                        MulliganOutput::MulliganDecision { keep: true },
1959                    ))),
1960                    Some("chooseAction") => {
1961                        if let Some(action) = play_first_card_action(prompt, "Swamp")? {
1962                            *played_land = true;
1963                            Ok(Some(action))
1964                        } else {
1965                            Ok(Some(PromptOutput::ChooseAction(ChooseActionOutput::Pass {
1966                                until: None,
1967                                exhaust_stack: false,
1968                            })))
1969                        }
1970                    }
1971                    other => Err(format!(
1972                        "scenario '{}' expected mulligan or chooseAction, got {:?}",
1973                        self.name(),
1974                        other
1975                    )),
1976                }
1977            }
1978            Self::MulliganOncePlayLand {
1979                mulliganed,
1980                kept_second_hand,
1981                put_back_done,
1982                played_land,
1983            } => {
1984                if *played_land && battlefield_contains(game_view, "Swamp") {
1985                    return Ok(None);
1986                }
1987                match prompt_type(prompt) {
1988                    Some("mulligan") if !*mulliganed => {
1989                        *mulliganed = true;
1990                        Ok(Some(PromptOutput::Mulligan(MulliganOutput::MulliganDecision { keep: false })))
1991                    }
1992                    Some("mulligan") if !*kept_second_hand => {
1993                        *kept_second_hand = true;
1994                        Ok(Some(PromptOutput::Mulligan(MulliganOutput::MulliganDecision { keep: true })))
1995                    }
1996                    Some("mulliganPutBack") if !*put_back_done => {
1997                        let count = prompt
1998                            .get("input")
1999                            .and_then(|input| input.get("count"))
2000                            .and_then(Value::as_u64)
2001                            .unwrap_or(1) as usize;
2002                        let card_ids = prompt_card_ids(prompt, "handCardIds", count)?;
2003                        *put_back_done = true;
2004                        Ok(Some(PromptOutput::MulliganPutBack(MulliganPutBackOutput::MulliganPutBackDecision { card_ids })))
2005                    }
2006                    Some("chooseAction") => {
2007                        if let Some(action) = play_first_card_action(prompt, "Swamp")? {
2008                            *played_land = true;
2009                            Ok(Some(action))
2010                        } else {
2011                            Ok(Some(PromptOutput::ChooseAction(ChooseActionOutput::Pass { until: None, exhaust_stack: false })))
2012                        }
2013                    }
2014                    other => Err(format!(
2015                        "scenario '{}' expected mulligan, mulliganPutBack, or chooseAction, got {:?}",
2016                        self.name(),
2017                        other
2018                    )),
2019                }
2020            }
2021        }
2022    }
2023}
2024
2025#[cfg(feature = "java-forge")]
2026fn run_scenario_loop<B: JavaBridge>(
2027    session: &mut JavaForgeSession<B>,
2028    mut scenario: JavaScenario,
2029    max_prompts: usize,
2030) -> Result<(), String> {
2031    let mut prompts_seen = 0usize;
2032    let mut last_prompt_json: Option<String> = None;
2033    while prompts_seen < max_prompts {
2034        let Some(prompt_json) = wait_for_prompt(session, 600)? else {
2035            return Err(format!(
2036                "timed out waiting for java-forge scenario '{}' prompt",
2037                scenario.name()
2038            ));
2039        };
2040        if last_prompt_json.as_deref() == Some(prompt_json.as_str()) {
2041            std::thread::sleep(Duration::from_millis(50));
2042            continue;
2043        }
2044        last_prompt_json = Some(prompt_json.clone());
2045        prompts_seen += 1;
2046
2047        let prompt: AgentPrompt = serde_json::from_str(&prompt_json)
2048            .map_err(|err| format!("failed to parse java scenario prompt: {err}"))?;
2049        let player = player_index(&prompt.deciding_player_id);
2050        if player != 0 {
2051            if let Some(output) = auto_action(&prompt) {
2052                session.submit_action(
2053                    &serde_json::to_string(&output).map_err(|err| err.to_string())?,
2054                )?;
2055            }
2056            continue;
2057        }
2058
2059        let game_view: GameViewDto = serde_json::from_str(&session.get_snapshot(Some(0))?)
2060            .map_err(|err| format!("failed to parse java scenario snapshot: {err}"))?;
2061        let normalized_prompt = serde_json::to_value(&prompt).map_err(|err| err.to_string())?;
2062        info!(
2063            scenario = scenario.name(),
2064            prompts_seen,
2065            prompt_type = prompt_type(&normalized_prompt).unwrap_or("<missing>"),
2066            "java-forge scenario prompt"
2067        );
2068        let Some(action) = scenario.next_action(&normalized_prompt, &game_view)? else {
2069            info!(
2070                scenario = scenario.name(),
2071                prompts_seen, "java-forge scenario assertions satisfied"
2072            );
2073            return Ok(());
2074        };
2075        submit_player_action(session, &action)?;
2076    }
2077    Err(format!(
2078        "java-forge scenario '{}' did not complete within {max_prompts} prompts",
2079        scenario.name()
2080    ))
2081}
2082
2083#[cfg(feature = "java-forge")]
2084pub fn run_concede_smoke() -> Result<(), String> {
2085    let config = JavaRuntimeConfig::from_env();
2086    let assets_dir = config.assets_dir.to_string_lossy().to_string();
2087    let bridge = SubprocessBridge::spawn(&config)?;
2088    let mut session = JavaForgeSession::new(bridge);
2089    session.initialize(&assets_dir)?;
2090    run_concede_game(&mut session, 3, 2, 12)?;
2091    run_concede_game(&mut session, 2, 1, 8)?;
2092    Ok(())
2093}
2094
2095#[cfg(not(feature = "java-forge"))]
2096pub fn run_concede_smoke() -> Result<(), String> {
2097    Err(unsupported_message().to_string())
2098}
2099
2100#[cfg(forge_backend)]
2101fn directive_concede_json(player: usize) -> String {
2102    format!(r#"{{"type":"directive","directive":{{"type":"concede"}},"player":{player}}}"#)
2103}
2104
2105#[cfg(feature = "java-forge")]
2106fn run_concede_game<B: JavaBridge>(
2107    session: &mut JavaForgeSession<B>,
2108    seats: usize,
2109    conceder: usize,
2110    concede_after: usize,
2111) -> Result<(), String> {
2112    const POST_CONCEDE_DECISIONS: usize = 12;
2113    const STALL_REPEATS: usize = 300;
2114
2115    let cards: Vec<manabrew_protocol::deck_dto::DeckCard> = (0..60)
2116        .map(|_| manabrew_protocol::deck_dto::DeckCard {
2117            identity: DeckCardIdentity {
2118                name: "Mountain".to_string(),
2119                set_code: "M20".to_string(),
2120                ..Default::default()
2121            },
2122            ..Default::default()
2123        })
2124        .collect();
2125    let deck = Deck {
2126        name: "concede-smoke".to_string(),
2127        cards,
2128        ..Default::default()
2129    };
2130    let identities = deck_card_identities(&deck);
2131    let players: Vec<PlayerConfig> = (0..seats)
2132        .map(|i| PlayerConfig::new(format!("Concede {}", i + 1), &identities, Vec::new()))
2133        .collect();
2134    let request = StartGameRequest::new(
2135        format!("concede-smoke-{seats}p"),
2136        String::new(),
2137        20,
2138        7,
2139        players,
2140    );
2141    let session_id = session.start_game(&request)?;
2142    info!(
2143        session_id,
2144        seats, conceder, concede_after, "concede smoke game started"
2145    );
2146
2147    let mut bots: HashMap<usize, SimpleAi> = HashMap::new();
2148    let mut last_prompt_json: Option<String> = None;
2149    let mut acted = 0usize;
2150    let mut acted_after_concede = 0usize;
2151    let mut conceded = false;
2152    let mut repeat_count = 0usize;
2153
2154    for _ in 0..40_000 {
2155        if session.is_game_over()? {
2156            if !conceded {
2157                return Err("concede smoke: game ended before the concede fired".to_string());
2158            }
2159            break;
2160        }
2161        let Some(prompt_json) = session.get_prompt(0)? else {
2162            std::thread::sleep(Duration::from_millis(10));
2163            continue;
2164        };
2165        if last_prompt_json.as_deref() == Some(prompt_json.as_str()) {
2166            repeat_count += 1;
2167            if repeat_count > STALL_REPEATS {
2168                return Err(format!(
2169                    "concede smoke stalled on the same prompt (seats={seats} acted={acted} conceded={conceded} after={acted_after_concede}): {prompt_json}"
2170                ));
2171            }
2172            std::thread::sleep(Duration::from_millis(10));
2173            continue;
2174        }
2175        repeat_count = 0;
2176
2177        let prompt: AgentPrompt = serde_json::from_str(&prompt_json)
2178            .map_err(|err| format!("concede smoke: bad prompt: {err}"))?;
2179        let player = player_index(&prompt.deciding_player_id);
2180
2181        if !conceded && acted >= concede_after {
2182            info!(conceder, acted, "concede smoke: injecting concede");
2183            session.submit_action(&directive_concede_json(conceder))?;
2184            conceded = true;
2185            last_prompt_json = None;
2186            continue;
2187        }
2188        if conceded && player == conceder {
2189            session.submit_action(&directive_concede_json(conceder))?;
2190            last_prompt_json = Some(prompt_json);
2191            continue;
2192        }
2193
2194        if let Some(action) = bots.entry(player).or_default().decide(prompt) {
2195            submit_player_action(session, &action)?;
2196            acted += 1;
2197            if conceded {
2198                acted_after_concede += 1;
2199            }
2200        }
2201        last_prompt_json = Some(prompt_json);
2202        if conceded && seats > 2 && acted_after_concede >= POST_CONCEDE_DECISIONS {
2203            break;
2204        }
2205    }
2206
2207    if !conceded {
2208        return Err("concede smoke: never reached the injection point".to_string());
2209    }
2210    if seats == 2 {
2211        if !session.is_game_over()? {
2212            return Err("concede smoke: 2p concession did not end the game".to_string());
2213        }
2214        info!(acted, "concede smoke: 2p concession ended the game");
2215    } else {
2216        if acted_after_concede < POST_CONCEDE_DECISIONS && !session.is_game_over()? {
2217            return Err(format!(
2218                "concede smoke: game did not progress after the concession (only {acted_after_concede} decisions)"
2219            ));
2220        }
2221        let snapshot = parse_snapshot(session)?;
2222        let status = snapshot
2223            .pointer(&format!("/players/{conceder}/status"))
2224            .and_then(Value::as_str)
2225            .unwrap_or("");
2226        if status != "conceded" {
2227            return Err(format!(
2228                "concede smoke: seat {conceder} has status '{status}', expected 'conceded'"
2229            ));
2230        }
2231        info!(
2232            acted,
2233            acted_after_concede, "concede smoke: game continued past the concession"
2234        );
2235    }
2236    session.end_game()?;
2237    Ok(())
2238}
2239
2240#[cfg(feature = "java-forge")]
2241fn run_self_play_loop<B: JavaBridge>(
2242    session: &mut JavaForgeSession<B>,
2243    max_prompts: usize,
2244) -> Result<(), String> {
2245    const STALL_REPEATS: usize = 100;
2246
2247    let mut bots: HashMap<usize, SimpleAi> = HashMap::new();
2248    let mut last_prompt_json: Option<String> = None;
2249    let mut acted = 0usize;
2250    let mut repeat_count = 0usize;
2251    let mut seen_prompt = false;
2252    let max_iterations = max_prompts.saturating_mul(200).max(2_000);
2253
2254    for _ in 0..max_iterations {
2255        if let Some(prompt_json) = session.get_prompt(0)? {
2256            seen_prompt = true;
2257            if last_prompt_json.as_deref() == Some(prompt_json.as_str()) {
2258                if session.is_game_over()? {
2259                    info!(acted, "java-forge self-play reached game over");
2260                    return Ok(());
2261                }
2262                repeat_count += 1;
2263                if repeat_count > STALL_REPEATS {
2264                    let raw_value: Value =
2265                        serde_json::from_str(&prompt_json).unwrap_or(Value::Null);
2266                    dump_stuck(
2267                        "java re-emitted the same prompt after the bot acted (stall)",
2268                        &raw_value,
2269                        None,
2270                        session,
2271                    );
2272                    return Err(
2273                        "self-play stalled: java re-emitted the same prompt after the bot's action"
2274                            .to_string(),
2275                    );
2276                }
2277                std::thread::sleep(Duration::from_millis(20));
2278                continue;
2279            }
2280            repeat_count = 0;
2281
2282            let prompt: AgentPrompt = serde_json::from_str(&prompt_json)
2283                .map_err(|err| format!("failed to parse java self-play prompt: {err}"))?;
2284            let player = player_index(&prompt.deciding_player_id);
2285            let raw_value: Value = serde_json::from_str(&prompt_json).unwrap_or(Value::Null);
2286            let normalized = raw_value.clone();
2287
2288            match bots.entry(player).or_default().decide(prompt) {
2289                Some(action) => {
2290                    if let Err(err) = submit_player_action(session, &action) {
2291                        dump_stuck(
2292                            "java rejected the bot action",
2293                            &raw_value,
2294                            Some(&normalized),
2295                            session,
2296                        );
2297                        return Err(format!(
2298                            "self-play: java rejected action for player {player}: {err}"
2299                        ));
2300                    }
2301                    acted += 1;
2302                    if acted >= max_prompts {
2303                        dump_stuck(
2304                            "did not reach game over within max prompts",
2305                            &raw_value,
2306                            Some(&normalized),
2307                            session,
2308                        );
2309                        return Err(format!(
2310                            "self-play did not reach game over within {max_prompts} decisions"
2311                        ));
2312                    }
2313                }
2314                None => debug!(
2315                    player,
2316                    prompt_type = prompt_type(&normalized).unwrap_or("<missing>"),
2317                    "self-play: no action for prompt (display-only)"
2318                ),
2319            }
2320            last_prompt_json = Some(prompt_json);
2321            continue;
2322        }
2323
2324        if seen_prompt && session.is_game_over()? {
2325            info!(acted, "java-forge self-play reached game over");
2326            return Ok(());
2327        }
2328        std::thread::sleep(Duration::from_millis(20));
2329    }
2330
2331    dump_stuck(
2332        "self-play exceeded its iteration cap without game over",
2333        &Value::Null,
2334        None,
2335        session,
2336    );
2337    Err("self-play exceeded its iteration cap without reaching game over".to_string())
2338}
2339
2340#[cfg(feature = "java-forge")]
2341fn parse_snapshot<B: JavaBridge>(session: &mut JavaForgeSession<B>) -> Result<Value, String> {
2342    let snapshot_json = session.get_snapshot(Some(0))?;
2343    serde_json::from_str(&snapshot_json)
2344        .map_err(|err| format!("failed to parse java self-play snapshot: {err}"))
2345}
2346
2347#[cfg(feature = "java-forge")]
2348fn dump_stuck<B: JavaBridge>(
2349    reason: &str,
2350    prompt: &Value,
2351    normalized: Option<&Value>,
2352    session: &mut JavaForgeSession<B>,
2353) {
2354    let snapshot = parse_snapshot(session).unwrap_or(Value::Null);
2355    let artifact = json!({
2356        "reason": reason,
2357        "rawPrompt": prompt,
2358        "normalizedPrompt": normalized,
2359        "snapshot": snapshot,
2360    });
2361    let ts = std::time::SystemTime::now()
2362        .duration_since(std::time::UNIX_EPOCH)
2363        .map(|d| d.as_millis())
2364        .unwrap_or(0);
2365    let path = workspace_root().join(format!("target/self-play-stuck-{ts}.json"));
2366    match serde_json::to_string_pretty(&artifact) {
2367        Ok(body) => {
2368            if let Err(error) = std::fs::write(&path, body) {
2369                warn!(%error, reason, "self-play stuck; failed to write artifact");
2370            } else {
2371                warn!(path = %path.display(), reason, "self-play stuck; wrote artifact");
2372            }
2373        }
2374        Err(error) => warn!(%error, reason, "self-play stuck; failed to serialize artifact"),
2375    }
2376}
2377
2378#[cfg(feature = "java-forge")]
2379fn submit_player_action<B: JavaBridge>(
2380    session: &mut JavaForgeSession<B>,
2381    action: &PromptOutput,
2382) -> Result<(), String> {
2383    let action_json = serde_json::to_string(action)
2384        .map_err(|err| format!("failed to serialize scenario action: {err}"))?;
2385    session.submit_action(&action_json)?;
2386    Ok(())
2387}
2388
2389#[cfg(feature = "java-forge")]
2390fn prompt_type(prompt: &Value) -> Option<&str> {
2391    prompt
2392        .get("input")
2393        .and_then(|input| input.get("type"))
2394        .and_then(Value::as_str)
2395}
2396
2397#[cfg(feature = "java-forge")]
2398fn play_first_card_action(prompt: &Value, card_name: &str) -> Result<Option<PromptOutput>, String> {
2399    let Some(action) = prompt
2400        .get("input")
2401        .and_then(|input| input.get("actions"))
2402        .and_then(Value::as_array)
2403        .and_then(|actions| {
2404            actions.iter().find(|action| {
2405                action
2406                    .get("modeLabel")
2407                    .and_then(Value::as_str)
2408                    .is_some_and(|label| label.contains(card_name))
2409            })
2410        })
2411    else {
2412        return Ok(None);
2413    };
2414    let action_id = action
2415        .get("id")
2416        .and_then(Value::as_str)
2417        .ok_or_else(|| format!("playable action for '{card_name}' is missing id"))?;
2418    Ok(Some(PromptOutput::ChooseAction(ChooseActionOutput::Act {
2419        action_id: action_id.to_string(),
2420    })))
2421}
2422
2423#[cfg(feature = "java-forge")]
2424fn prompt_card_ids(prompt: &Value, field: &str, count: usize) -> Result<Vec<String>, String> {
2425    let card_ids = prompt
2426        .get("input")
2427        .and_then(|input| input.get(field))
2428        .and_then(Value::as_array)
2429        .ok_or_else(|| format!("prompt is missing {field}"))?;
2430    if card_ids.len() < count {
2431        return Err(format!(
2432            "prompt field {field} has {} cards, need {count}",
2433            card_ids.len()
2434        ));
2435    }
2436    Ok(card_ids
2437        .iter()
2438        .take(count)
2439        .filter_map(Value::as_str)
2440        .map(str::to_string)
2441        .collect())
2442}
2443
2444#[cfg(feature = "java-forge")]
2445fn battlefield_contains(game_view: &GameViewDto, card_name: &str) -> bool {
2446    use manabrew_agent_interface::game_view_dto::{CardView, ZoneKind};
2447    game_view.zones.iter().any(|zone| {
2448        zone.zone == ZoneKind::Battlefield
2449            && zone.owner_id == "player-0"
2450            && zone.cards.iter().any(|card| match card {
2451                CardView::Visible(dto) => dto.identity.name == card_name,
2452                CardView::Hidden { .. } => false,
2453            })
2454    })
2455}
2456
2457pub trait JavaBridge {
2458    fn initialize(&mut self, assets_dir: &str) -> Result<(), String>;
2459    fn start_game_json(&mut self, request_json: &str) -> Result<String, String>;
2460    fn submit_action(&mut self, session_id: &str, action_json: &str) -> Result<String, String>;
2461    fn get_prompt(
2462        &mut self,
2463        session_id: &str,
2464        player_index: usize,
2465    ) -> Result<Option<String>, String>;
2466    fn get_snapshot(&mut self, session_id: &str, viewer: Option<usize>) -> Result<String, String>;
2467    fn is_game_over(&mut self, session_id: &str) -> Result<bool, String>;
2468    fn end_game(&mut self, session_id: &str) -> Result<(), String>;
2469    fn abort_game(&mut self, session_id: &str) -> Result<(), String>;
2470}
2471
2472pub struct JavaForgeSession<B> {
2473    bridge: B,
2474    session_id: Option<String>,
2475}
2476
2477impl<B: JavaBridge> JavaForgeSession<B> {
2478    pub fn new(bridge: B) -> Self {
2479        Self {
2480            bridge,
2481            session_id: None,
2482        }
2483    }
2484
2485    pub fn initialize(&mut self, assets_dir: &str) -> Result<(), String> {
2486        self.bridge.initialize(assets_dir)
2487    }
2488
2489    pub fn start_game(&mut self, request: &StartGameRequest) -> Result<String, String> {
2490        let request_json = request.to_json().map_err(|err| err.to_string())?;
2491        let response_json = self.bridge.start_game_json(&request_json)?;
2492        let response: StartGameResponse =
2493            serde_json::from_str(&response_json).map_err(|err| err.to_string())?;
2494        self.session_id = Some(response.session_id.clone());
2495        Ok(response.session_id)
2496    }
2497
2498    pub fn submit_action(&mut self, action_json: &str) -> Result<String, String> {
2499        let session_id = self.require_session_id()?.to_string();
2500        self.bridge.submit_action(&session_id, action_json)
2501    }
2502
2503    pub fn get_prompt(&mut self, player_index: usize) -> Result<Option<String>, String> {
2504        let session_id = self.require_session_id()?.to_string();
2505        self.bridge.get_prompt(&session_id, player_index)
2506    }
2507
2508    pub fn get_snapshot(&mut self, viewer: Option<usize>) -> Result<String, String> {
2509        let session_id = self.require_session_id()?.to_string();
2510        self.bridge.get_snapshot(&session_id, viewer)
2511    }
2512
2513    pub fn is_game_over(&mut self) -> Result<bool, String> {
2514        let session_id = self.require_session_id()?.to_string();
2515        self.bridge.is_game_over(&session_id)
2516    }
2517
2518    pub fn end_game(&mut self) -> Result<(), String> {
2519        let Some(session_id) = self.session_id.take() else {
2520            return Ok(());
2521        };
2522        self.bridge.end_game(&session_id)
2523    }
2524
2525    fn require_session_id(&self) -> Result<&str, String> {
2526        self.session_id
2527            .as_deref()
2528            .ok_or_else(|| "java-forge session has not started".to_string())
2529    }
2530}
2531
2532pub struct UnavailableJavaBridge;
2533
2534impl JavaBridge for UnavailableJavaBridge {
2535    fn initialize(&mut self, _assets_dir: &str) -> Result<(), String> {
2536        Err(unsupported_message().to_string())
2537    }
2538
2539    fn start_game_json(&mut self, _request_json: &str) -> Result<String, String> {
2540        Err(unsupported_message().to_string())
2541    }
2542
2543    fn submit_action(&mut self, _session_id: &str, _action_json: &str) -> Result<String, String> {
2544        Err(unsupported_message().to_string())
2545    }
2546
2547    fn get_prompt(
2548        &mut self,
2549        _session_id: &str,
2550        _player_index: usize,
2551    ) -> Result<Option<String>, String> {
2552        Err(unsupported_message().to_string())
2553    }
2554
2555    fn get_snapshot(
2556        &mut self,
2557        _session_id: &str,
2558        _viewer: Option<usize>,
2559    ) -> Result<String, String> {
2560        Err(unsupported_message().to_string())
2561    }
2562
2563    fn is_game_over(&mut self, _session_id: &str) -> Result<bool, String> {
2564        Err(unsupported_message().to_string())
2565    }
2566
2567    fn end_game(&mut self, _session_id: &str) -> Result<(), String> {
2568        Err(unsupported_message().to_string())
2569    }
2570
2571    fn abort_game(&mut self, _session_id: &str) -> Result<(), String> {
2572        Err(unsupported_message().to_string())
2573    }
2574}
2575
2576#[cfg(feature = "java-forge")]
2577#[derive(serde::Deserialize)]
2578struct SubprocessReply {
2579    ok: bool,
2580    #[serde(default)]
2581    result: String,
2582    #[serde(default)]
2583    error: Option<String>,
2584}
2585
2586#[cfg(feature = "java-forge")]
2587const CALL_TIMEOUT: Duration = Duration::from_secs(60);
2588#[cfg(feature = "java-forge")]
2589const SHUTDOWN_GRACE: Duration = Duration::from_secs(5);
2590
2591#[cfg(feature = "java-forge")]
2592pub struct SubprocessBridge {
2593    child: Child,
2594    stdin: BufWriter<ChildStdin>,
2595    stdout_rx: std_mpsc::Receiver<String>,
2596    stdout_handle: Option<std::thread::JoinHandle<()>>,
2597    stderr_handle: Option<std::thread::JoinHandle<()>>,
2598}
2599
2600#[cfg(feature = "java-forge")]
2601impl SubprocessBridge {
2602    fn spawn(config: &JavaRuntimeConfig) -> Result<Self, String> {
2603        config.validate()?;
2604
2605        let java_bin = resolve_java_bin(config);
2606        let jvm_args = config.jvm_args();
2607        info!(target: "self_hosted_node::java", args = %jvm_args.join(" "), "spawning java engine");
2608        let mut cmd = Command::new(&java_bin);
2609        // The JVM applies JAVA_TOOL_OPTIONS before anything on the command line,
2610        // so whatever is set on the host silently governs every flag we do not
2611        // name. Production ran -Xmx512m -XX:+UseSerialGC that way for months
2612        // with nothing in the repo to show it. Use SELF_HOSTED_NODE_JAVA_OPTS
2613        // instead: it is passed here, logged above, and lives in config.
2614        cmd.env_remove("JAVA_TOOL_OPTIONS");
2615        cmd.args(&jvm_args);
2616        cmd.arg("-jar").arg(&config.harness_jar);
2617        cmd.arg("--interactive-server");
2618        cmd.arg("--forge-home")
2619            .arg(format!("{}/", config.assets_dir.display()));
2620
2621        let mut child = cmd
2622            .stdin(Stdio::piped())
2623            .stdout(Stdio::piped())
2624            .stderr(Stdio::piped())
2625            .spawn()
2626            .map_err(|err| format!("failed to spawn java subprocess: {err}"))?;
2627
2628        let stdin = child
2629            .stdin
2630            .take()
2631            .ok_or_else(|| "java subprocess has no stdin".to_string())?;
2632        let stdout = child
2633            .stdout
2634            .take()
2635            .ok_or_else(|| "java subprocess has no stdout".to_string())?;
2636        let stderr = child.stderr.take();
2637
2638        // Bounded so a chatty Java side can't grow the queue without bound.
2639        // Cap is generous — protocol replies are one line per request.
2640        let (stdout_tx, stdout_rx) = std_mpsc::sync_channel::<String>(1024);
2641        let stdout_handle = std::thread::spawn(move || {
2642            let reader = BufReader::new(stdout);
2643            for line in reader.lines().map_while(Result::ok) {
2644                if stdout_tx.send(line).is_err() {
2645                    break;
2646                }
2647            }
2648        });
2649
2650        let stderr_handle = std::thread::spawn(move || {
2651            if let Some(stderr) = stderr {
2652                let reader = BufReader::new(stderr);
2653                for line in reader.lines().map_while(Result::ok) {
2654                    if let Some(pause) = parse_gc_pause(&line) {
2655                        crate::metrics::record_jvm_gc(
2656                            pause.kind,
2657                            Duration::from_secs_f64(pause.millis / 1000.0),
2658                            pause.heap_after_mb,
2659                        );
2660                    }
2661                    if line.contains("Exception") || line.contains("ERROR") {
2662                        warn!(target: "self_hosted_node::java", "[java] {line}");
2663                    } else if line.contains("][gc") {
2664                        // -Xlog:gc*:stderr is opt-in, and debug would drop it before Loki.
2665                        info!(target: "self_hosted_node::java", "[java] {line}");
2666                    } else {
2667                        debug!(target: "self_hosted_node::java", "[java] {line}");
2668                    }
2669                }
2670            }
2671        });
2672
2673        Ok(Self {
2674            child,
2675            stdin: BufWriter::new(stdin),
2676            stdout_rx,
2677            stdout_handle: Some(stdout_handle),
2678            stderr_handle: Some(stderr_handle),
2679        })
2680    }
2681
2682    fn call(&mut self, request_json: &str) -> Result<String, String> {
2683        // Drain anything still queued from a prior request — a previous call()
2684        // that timed out may have left its reply in the channel, and consuming
2685        // it now would shift every subsequent call off-by-one.
2686        loop {
2687            match self.stdout_rx.try_recv() {
2688                Ok(stale) => {
2689                    debug!(target: "self_hosted_node::java", line = %stale, "discarding stale stdout line");
2690                }
2691                Err(TryRecvError::Empty) => break,
2692                Err(TryRecvError::Disconnected) => {
2693                    return Err("java subprocess closed stdout (crashed?)".to_string());
2694                }
2695            }
2696        }
2697
2698        self.stdin
2699            .write_all(request_json.as_bytes())
2700            .map_err(|err| format!("failed to write subprocess stdin: {err}"))?;
2701        self.stdin
2702            .write_all(b"\n")
2703            .map_err(|err| format!("failed to write subprocess newline: {err}"))?;
2704        self.stdin
2705            .flush()
2706            .map_err(|err| format!("failed to flush subprocess stdin: {err}"))?;
2707
2708        let deadline = Instant::now() + CALL_TIMEOUT;
2709        loop {
2710            let remaining = deadline.saturating_duration_since(Instant::now());
2711            if remaining.is_zero() {
2712                return Err(format!(
2713                    "java subprocess timed out after {}s",
2714                    CALL_TIMEOUT.as_secs()
2715                ));
2716            }
2717            match self.stdout_rx.recv_timeout(remaining) {
2718                Ok(line) => {
2719                    let trimmed = line.trim();
2720                    if trimmed.is_empty() {
2721                        continue;
2722                    }
2723                    match serde_json::from_str::<SubprocessReply>(trimmed) {
2724                        Ok(reply) if reply.ok => return Ok(reply.result),
2725                        Ok(reply) => {
2726                            return Err(reply.error.unwrap_or_else(|| "unknown java error".into()));
2727                        }
2728                        Err(_) => {
2729                            debug!(target: "self_hosted_node::java", line = trimmed, "non-protocol stdout line");
2730                        }
2731                    }
2732                }
2733                Err(RecvTimeoutError::Timeout) => {
2734                    return Err(format!(
2735                        "java subprocess timed out after {}s",
2736                        CALL_TIMEOUT.as_secs()
2737                    ));
2738                }
2739                Err(RecvTimeoutError::Disconnected) => {
2740                    return Err("java subprocess closed stdout (crashed?)".to_string());
2741                }
2742            }
2743        }
2744    }
2745
2746    fn reset(&mut self) -> Result<(), String> {
2747        self.call("{\"command\":\"reset\"}").map(|_| ())
2748    }
2749
2750    fn is_alive(&mut self) -> bool {
2751        matches!(self.child.try_wait(), Ok(None))
2752    }
2753
2754    fn shutdown(mut self) {
2755        let _ = self.stdin.write_all(b"{\"command\":\"quit\"}\n");
2756        let _ = self.stdin.flush();
2757        let deadline = Instant::now() + SHUTDOWN_GRACE;
2758        loop {
2759            match self.child.try_wait() {
2760                Ok(Some(_)) => break,
2761                Ok(None) if Instant::now() >= deadline => {
2762                    let _ = self.child.kill();
2763                    let _ = self.child.wait();
2764                    break;
2765                }
2766                Ok(None) => std::thread::sleep(Duration::from_millis(100)),
2767                Err(_) => break,
2768            }
2769        }
2770        if let Some(handle) = self.stdout_handle.take() {
2771            let _ = handle.join();
2772        }
2773        if let Some(handle) = self.stderr_handle.take() {
2774            let _ = handle.join();
2775        }
2776    }
2777}
2778
2779#[cfg(feature = "java-forge")]
2780impl JavaBridge for SubprocessBridge {
2781    fn initialize(&mut self, _assets_dir: &str) -> Result<(), String> {
2782        Ok(())
2783    }
2784
2785    fn start_game_json(&mut self, request_json: &str) -> Result<String, String> {
2786        let body = json!({ "command": "startGame", "payload": request_json });
2787        self.call(&body.to_string())
2788    }
2789
2790    fn submit_action(&mut self, session_id: &str, action_json: &str) -> Result<String, String> {
2791        let body = json!({
2792            "command": "submitAction",
2793            "sessionId": session_id,
2794            "payload": action_json,
2795        });
2796        self.call(&body.to_string())
2797    }
2798
2799    fn get_prompt(
2800        &mut self,
2801        session_id: &str,
2802        player_index: usize,
2803    ) -> Result<Option<String>, String> {
2804        let body = json!({
2805            "command": "getPrompt",
2806            "sessionId": session_id,
2807            "playerIndex": player_index,
2808        });
2809        let prompt = self.call(&body.to_string())?;
2810        Ok((!prompt.is_empty()).then_some(prompt))
2811    }
2812
2813    fn get_snapshot(&mut self, session_id: &str, viewer: Option<usize>) -> Result<String, String> {
2814        let mut body = json!({ "command": "getSnapshot", "sessionId": session_id });
2815        if let Some(viewer) = viewer {
2816            body["viewer"] = json!(viewer);
2817        }
2818        self.call(&body.to_string())
2819    }
2820
2821    fn is_game_over(&mut self, session_id: &str) -> Result<bool, String> {
2822        let body = json!({ "command": "getGameOver", "sessionId": session_id });
2823        let value = self.call(&body.to_string())?;
2824        Ok(value.trim() == "true")
2825    }
2826
2827    fn end_game(&mut self, session_id: &str) -> Result<(), String> {
2828        let body = json!({ "command": "endGame", "sessionId": session_id });
2829        self.call(&body.to_string()).map(|_| ())
2830    }
2831
2832    fn abort_game(&mut self, session_id: &str) -> Result<(), String> {
2833        let body = json!({ "command": "abortGame", "sessionId": session_id });
2834        self.call(&body.to_string()).map(|_| ())
2835    }
2836}
2837
2838#[cfg(feature = "java-forge")]
2839impl Drop for SubprocessBridge {
2840    fn drop(&mut self) {
2841        let _ = self.child.kill();
2842        let _ = self.child.wait();
2843        if let Some(handle) = self.stdout_handle.take() {
2844            let _ = handle.join();
2845        }
2846        if let Some(handle) = self.stderr_handle.take() {
2847            let _ = handle.join();
2848        }
2849    }
2850}
2851
2852#[cfg(feature = "java-forge")]
2853fn resolve_java_bin(config: &JavaRuntimeConfig) -> String {
2854    if let Some(home) = &config.java_home {
2855        let bin = home.join("bin").join("java");
2856        if bin.is_file() {
2857            return bin.to_string_lossy().to_string();
2858        }
2859    }
2860    if let Ok(home) = env::var("JAVA_HOME") {
2861        let bin = PathBuf::from(home).join("bin").join("java");
2862        if bin.is_file() {
2863            return bin.to_string_lossy().to_string();
2864        }
2865    }
2866    "java".to_string()
2867}
2868
2869#[derive(Debug, Clone, Serialize)]
2870#[serde(rename_all = "camelCase")]
2871pub struct StartGameRequest {
2872    game_id: String,
2873    variant: String,
2874    starting_life: i32,
2875    seed: u64,
2876    players: Vec<PlayerConfig>,
2877}
2878
2879#[derive(Debug, Clone, Serialize)]
2880#[serde(rename_all = "camelCase")]
2881pub struct PlayerConfig {
2882    name: String,
2883    deck: Vec<CardIdentityForJava>,
2884    commander_names: Vec<String>,
2885    ai: bool,
2886}
2887
2888#[derive(Debug, Clone, Serialize)]
2889#[serde(rename_all = "camelCase")]
2890pub struct CardIdentityForJava {
2891    name: String,
2892    set_code: Option<String>,
2893    collector_number: Option<String>,
2894    foil: bool,
2895}
2896
2897#[derive(Debug, serde::Deserialize)]
2898#[serde(rename_all = "camelCase")]
2899struct StartGameResponse {
2900    session_id: String,
2901    #[allow(dead_code)]
2902    player_indexes: Vec<usize>,
2903}
2904
2905impl StartGameRequest {
2906    pub fn new(
2907        game_id: String,
2908        variant: String,
2909        starting_life: i32,
2910        seed: u64,
2911        players: Vec<PlayerConfig>,
2912    ) -> Self {
2913        Self {
2914            game_id,
2915            variant,
2916            starting_life,
2917            seed,
2918            players,
2919        }
2920    }
2921
2922    pub fn to_json(&self) -> Result<String, serde_json::Error> {
2923        serde_json::to_string(self)
2924    }
2925}
2926
2927impl PlayerConfig {
2928    pub fn new(name: String, deck: &[DeckCardIdentity], commander_names: Vec<String>) -> Self {
2929        Self {
2930            name,
2931            deck: deck.iter().map(CardIdentityForJava::from).collect(),
2932            commander_names,
2933            ai: false,
2934        }
2935    }
2936}
2937
2938impl From<&DeckCardIdentity> for CardIdentityForJava {
2939    fn from(identity: &DeckCardIdentity) -> Self {
2940        Self {
2941            name: java_card_name(&identity.name),
2942            set_code: (!identity.set_code.is_empty()).then(|| identity.set_code.clone()),
2943            collector_number: (!identity.card_number.is_empty())
2944                .then(|| identity.card_number.clone()),
2945            foil: identity.foil.unwrap_or(false),
2946        }
2947    }
2948}
2949
2950fn java_card_name(name: &str) -> String {
2951    name.split_once(" // ")
2952        .map(|(front, _)| front.to_string())
2953        .unwrap_or_else(|| name.to_string())
2954}
2955
2956fn env_path(key: &str) -> Option<PathBuf> {
2957    env::var_os(key)
2958        .filter(|value| !value.is_empty())
2959        .map(PathBuf::from)
2960}
2961
2962fn env_sizing(key: &str, default: u64) -> Option<u64> {
2963    let value = match env::var(key) {
2964        Ok(raw) if !raw.trim().is_empty() => match raw.trim().parse::<u64>() {
2965            Ok(parsed) => parsed,
2966            Err(_) => {
2967                tracing::warn!(target: "self_hosted_node::java", key, raw, "ignoring unparseable jvm sizing");
2968                default
2969            }
2970        },
2971        _ => default,
2972    };
2973    (value > 0).then_some(value)
2974}
2975
2976fn env_classpath(key: &str) -> Vec<PathBuf> {
2977    let Some(value) = env::var_os(key) else {
2978        return Vec::new();
2979    };
2980    env::split_paths(&value).collect()
2981}
2982
2983fn require_dir(path: &Path, label: &str) -> Result<(), String> {
2984    if path.is_dir() {
2985        Ok(())
2986    } else {
2987        Err(format!("{label} does not exist: {}", path.display()))
2988    }
2989}
2990
2991fn require_file(path: &Path, label: &str) -> Result<(), String> {
2992    if path.is_file() {
2993        Ok(())
2994    } else {
2995        Err(format!("{label} does not exist: {}", path.display()))
2996    }
2997}
2998
2999/// One `Pause` line from the JVM's unified GC log.
3000struct GcPause {
3001    kind: &'static str,
3002    millis: f64,
3003    heap_after_mb: Option<u64>,
3004}
3005
3006/// Parse a unified-log GC pause, e.g.
3007/// `[12.3s][info][gc] GC(7) Pause Full (Allocation Failure) 240M->171M(247M) 226.856ms`
3008fn parse_gc_pause(line: &str) -> Option<GcPause> {
3009    let kind = if line.contains("Pause Full") {
3010        "full"
3011    } else if line.contains("Pause Young") {
3012        "young"
3013    } else {
3014        return None;
3015    };
3016    let millis = line
3017        .rsplit(' ')
3018        .find_map(|token| token.strip_suffix("ms")?.parse::<f64>().ok())?;
3019    // "240M->171M(247M)": the figure after the arrow is what survived.
3020    let heap_after_mb = line.split_once("->").and_then(|(_, rest)| {
3021        let digits: String = rest.chars().take_while(char::is_ascii_digit).collect();
3022        digits.parse::<u64>().ok()
3023    });
3024    Some(GcPause {
3025        kind,
3026        millis,
3027        heap_after_mb,
3028    })
3029}
3030
3031#[cfg(test)]
3032mod gc_log_tests {
3033    use super::{parse_gc_pause, PlayerConfig, StartGameRequest};
3034    use manabrew_protocol::deck_dto::DeckCardIdentity;
3035
3036    #[test]
3037    fn reads_a_full_collection() {
3038        let pause = parse_gc_pause(
3039            "[117.6s][info][gc] GC(21) Pause Full (Allocation Failure) 240M->171M(247M) 226.856ms",
3040        )
3041        .expect("parsed");
3042        assert_eq!(pause.kind, "full");
3043        assert_eq!(pause.heap_after_mb, Some(171));
3044        assert!((pause.millis - 226.856).abs() < f64::EPSILON);
3045    }
3046
3047    #[test]
3048    fn reads_a_young_collection() {
3049        let pause = parse_gc_pause(
3050            "[3.1s][info][gc] GC(2) Pause Young (Allocation Failure) 216M->91M(494M) 17.254ms",
3051        )
3052        .expect("parsed");
3053        assert_eq!(pause.kind, "young");
3054        assert_eq!(pause.heap_after_mb, Some(91));
3055    }
3056
3057    /// The node asks for `-Xlog:gc*:stderr:time,uptime,level,tags`, which is a
3058    /// richer line than plain `-Xlog:gc`. Real production output.
3059    #[test]
3060    fn reads_the_format_the_node_actually_requests() {
3061        let pause = parse_gc_pause(
3062            "[2026-08-18T20:41:42.907+0000][120025.819s][info][gc] GC(1202) Pause Full (Allocation Failure) 494M->487M(494M) 545.344ms",
3063        )
3064        .expect("parsed");
3065        assert_eq!(pause.kind, "full");
3066        assert_eq!(pause.heap_after_mb, Some(487));
3067        assert!((pause.millis - 545.344).abs() < f64::EPSILON);
3068    }
3069
3070    /// `gc*` also emits region and metaspace lines that carry `->`; they must
3071    /// not be mistaken for pauses.
3072    #[test]
3073    fn ignores_the_extra_lines_gc_star_adds() {
3074        assert!(
3075            parse_gc_pause("[120025.819s][info][gc,heap] GC(1202) Eden regions: 10->0(12)")
3076                .is_none()
3077        );
3078        assert!(
3079            parse_gc_pause("[120025.819s][info][gc,metaspace] Metaspace: 48M->48M(1088M)")
3080                .is_none()
3081        );
3082    }
3083
3084    #[test]
3085    fn ignores_other_output() {
3086        assert!(parse_gc_pause("[java] LOGGER ERROR: something").is_none());
3087        assert!(parse_gc_pause("[1.0s][info][gc,init] CardTable entry size: 512").is_none());
3088    }
3089
3090    #[test]
3091    fn serializes_exact_printing_identity_for_java() {
3092        let card = DeckCardIdentity {
3093            name: "Forest".to_string(),
3094            set_code: "EOE".to_string(),
3095            card_number: "266".to_string(),
3096            foil: Some(true),
3097            ..Default::default()
3098        };
3099        let request = StartGameRequest::new(
3100            "printing-test".to_string(),
3101            "Constructed".to_string(),
3102            20,
3103            42,
3104            vec![PlayerConfig::new("Player".to_string(), &[card], Vec::new())],
3105        );
3106        let json = request.to_json().expect("serialized");
3107
3108        assert!(json.contains(
3109            r#""deck":[{"name":"Forest","setCode":"EOE","collectorNumber":"266","foil":true}]"#
3110        ));
3111    }
3112}