Something went wrong. Try again.
A bot for https://go.sky.boo that lets you play Go against TypeSafe's Jev.
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973use anyhow::Result;use async_trait::async_trait;use atproto_jetstream::{Consumer, ConsumerTaskConfig, EventHandler, JetstreamEvent};use std::collections::BTreeMap;use std::sync::atomic::{AtomicI64, Ordering};use std::sync::Arc;use tokio_util::sync::CancellationToken;use tracing::{debug, error, info, warn};
use crate::board::Board;use crate::{unknown_to_value, App, Color, GameState, LastMove, MoveRecord};
struct GameHandler { app: App, /// Latest event timestamp fully processed — mirrored into SQLite so a /// restart resumes here instead of replaying a day of backlog. cursor: Arc<AtomicI64>,}
/// Every event variant carries the sequencer timestamp — our checkpoint.fn event_time_us(event: &JetstreamEvent) -> u64 { match event { JetstreamEvent::Commit { time_us, .. } | JetstreamEvent::Delete { time_us, .. } | JetstreamEvent::Identity { time_us, .. } | JetstreamEvent::Account { time_us, .. } => *time_us, }}
#[async_trait]impl EventHandler for GameHandler { fn handler_id(&self) -> &str { "jevgo-game-handler" }
async fn handle_event(&self, event: Arc<JetstreamEvent>) -> Result<()> { let result = self.handle(&event).await; if result.is_ok() { // Everything up to this event is processed. The checkpoint // write is best-effort — the in-memory cursor still advances, // so a later success overwrites a failed write. let time_us = event_time_us(&event) as i64; self.cursor.store(time_us, Ordering::Relaxed); if let Err(e) = self.app.db.set_cursor(time_us).await { error!(error = %e, "cursor checkpoint failed"); } } result }}
impl GameHandler { async fn handle(&self, event: &JetstreamEvent) -> Result<()> { match event { JetstreamEvent::Commit { did, commit, .. } => { let collection = commit.collection.as_str(); let operation = commit.operation.as_str();
if operation != "create" && operation != "update" { return Ok(()); }
match collection { "boo.sky.go.game" => { handle_game(&self.app, did, &commit.rkey, &commit.record).await } "boo.sky.go.move" => { handle_move(&self.app, did, &commit.rkey, &commit.record).await } "boo.sky.go.pass" => { handle_pass(&self.app, did, &commit.rkey, &commit.record).await } "boo.sky.go.action" => { handle_action(&self.app, did, &commit.rkey, &commit.record).await } "boo.sky.go.resign" => { handle_resign(&self.app, did, &commit.rkey, &commit.record).await } _ => Ok(()), } } _ => Ok(()), } }}
/// Parsed game record fields — the pure data extracted from a `boo.sky.go.game` record.struct ParsedGameRecord { board_size: usize, player_one: String, player_two: String, status: String,}
/// Parse a game record JSON into its constituent fields.fn parse_game_record(record: &serde_json::Value) -> Option<ParsedGameRecord> { let _player_two = record.get("playerTwo").and_then(|v| v.as_str())?; Some(ParsedGameRecord { board_size: record.get("boardSize").and_then(|v| v.as_u64()).unwrap_or(9) as usize, player_one: record .get("playerOne") .and_then(|v| v.as_str()) .unwrap_or("") .to_string(), player_two: _player_two.to_string(), status: record .get("status") .and_then(|v| v.as_str()) .unwrap_or("") .to_string(), })}
/// Build a GameState from parsed record fields, a game URI, the bot's DID,/// and a pre-fetched list of moves (already sorted by move_number).fn build_game_state( parsed: &ParsedGameRecord, game_uri: &str, bot_did: &str, moves: &[MoveRecord],) -> GameState { let bot_color = if parsed.player_one == bot_did { Color::Black } else { Color::White };
let mut board = Board::new(parsed.board_size); let (move_number, last_move, passes_in_a_row) = crate::replay_moves(&mut board, moves);
GameState { game_uri: game_uri.to_string(), board_size: parsed.board_size, bot_color, board, move_number, status: "active".into(), player_one: parsed.player_one.clone(), player_two: parsed.player_two.clone(), last_move, passes_in_a_row, }}
/// Parsed move record fields — the pure data from a `boo.sky.go.move` record.struct ParsedMoveRecord { x: usize, y: usize, color: Color, move_number: u32, game_uri: String, player: String,}
/// Parse a move record JSON.fn parse_move_record(record: &serde_json::Value) -> Option<ParsedMoveRecord> { Some(ParsedMoveRecord { x: record.get("x").and_then(|v| v.as_u64())? as usize, y: record.get("y").and_then(|v| v.as_u64())? as usize, color: match record.get("color").and_then(|v| v.as_str()) { Some("black") => Color::Black, _ => Color::White, }, move_number: record.get("moveNumber").and_then(|v| v.as_u64())? as u32, game_uri: record.get("game").and_then(|v| v.as_str())?.to_string(), player: record .get("player") .and_then(|v| v.as_str()) .unwrap_or("") .to_string(), })}
/// Apply a parsed move to a GameState. Returns captures, or None if the/// move doesn't belong to this game or is from the bot itself.fn apply_move_to_game( game: &mut GameState, parsed: &ParsedMoveRecord, bot_did: &str,) -> Option<usize> { if parsed.game_uri != game.game_uri { return None; } if game.status != "active" { return None; } if parsed.player == bot_did { return None; } // Already applied (e.g. redelivered by a Jetstream reconnect/backfill) — skip. if parsed.move_number <= game.move_number { return None; } let captures = game .board .try_apply_move(parsed.x, parsed.y, parsed.color.as_u8())?; game.move_number = parsed.move_number; game.passes_in_a_row = 0; game.last_move = Some(LastMove { x: parsed.x, y: parsed.y, color: parsed.color, captures, }); Some(captures)}
async fn handle_game( app: &App, did: &str, rkey: &str, record: &serde_json::Value,) -> Result<()> { let parsed = match parse_game_record(record) { Some(p) => p, None => return Ok(()), };
let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str(); if parsed.player_two != bot_did_str { return Ok(()); }
let game_uri = format!("at://{}/boo.sky.go.game/{}", did, rkey);
{ let mut games = app.games.lock().await; if let Some(game) = games.get_mut(&game_uri) { // The owning repo's record update is the only end-of-game signal a // challenge-game opponent can send us directly — mirror it. if parsed.status != "active" && game.status == "active" { info!(game = %game_uri, status = %parsed.status, "game over per record update"); game.status = "finished".into(); app.db.upsert_game(game).await?; } return Ok(()); } }
// Only live games get initialized; anything else has already ended. if parsed.status != "active" { return Ok(()); }
// A challenge game can arrive already resigned (replayed backlog) — // the rebuild settles that instead of initializing play. let Some(game_state) = rebuild_game(app, &game_uri, &parsed).await? else { return Ok(()); };
info!( game = %game_uri, opponent = %parsed.player_one, board_size = parsed.board_size, bot_color = if parsed.player_one == bot_did_str { "black" } else { "white" }, move_number = game_state.move_number, "challenged, game initialized" );
if check_bot_turn(&game_state) { let mut games = app.games.lock().await; if let Some(game) = games.get_mut(&game_uri) { play_move(app, game).await?; } }
Ok(())}
/// Each player posts their own moves and passes into their own repo (same/// pattern as `boo.sky.go.action`), so a game's full ply history must be/// assembled from both players' repos, not just the game-owning repo.async fn fetch_game_plies( app: &App, game_uri: &str, player_one: &str, player_two: &str,) -> Result<Vec<MoveRecord>> { let mut repos = vec![player_one]; if player_two != player_one { repos.push(player_two); }
let mut plies = Vec::new(); for repo in repos { plies.extend(crate::fetch_plies_from_repo(app, repo, game_uri).await?); }
plies.sort_by_key(|m| m.move_number); Ok(plies)}
/// Hydrate game state from the AT Proto repos (source of truth) on/// startup. The cache remembers which games we were in — challenge games/// live in an opponent's repo and no listing of our own will ever show/// them, so their URIs are the only handle kept across a restart. Games/// that turn out to be over or gone are pruned from the cache as we go.pub async fn hydrate(app: &App) -> Result<()> { let did = app.agent.did().await.expect("authenticated");
let mut games: BTreeMap<String, Option<serde_json::Value>> = app .db .game_uris() .await? .into_iter() .map(|uri| (uri, None)) .collect(); for (uri, value) in crate::list_repo_records(app, did.as_str(), "boo.sky.go.game").await? { games.insert(uri, Some(value)); }
for (game_uri, record) in games { // One broken game must not abort hydration of the rest. if let Err(e) = hydrate_game(app, &game_uri, record).await { error!(game = %game_uri, error = %e, "failed to hydrate game"); } } Ok(())}
/// Rebuild one game from its record — fetched on demand when the cache/// only remembered its URI — and track it. Waiting and malformed games/// stay out of the way (handle_join owns their fate); foreign, over, and/// vanished ones need no further attention and leave the cache.async fn hydrate_game( app: &App, game_uri: &str, record: Option<serde_json::Value>,) -> Result<()> { let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str();
let record = match record { Some(r) => r, None => match crate::fetch_game_record(app, game_uri).await? { Some(r) => r, None => { app.db.forget_game(game_uri).await?; return Ok(()); } }, }; let Some(parsed) = parse_game_record(&record) else { return Ok(()); // waiting or malformed — nothing to rebuild yet }; if parsed.player_one != bot_did_str && parsed.player_two != bot_did_str { app.db.forget_game(game_uri).await?; return Ok(()); } if parsed.status != "active" { app.db.forget_game(game_uri).await?; return Ok(()); }
if let Some(state) = rebuild_game(app, game_uri, &parsed).await? { info!( game = %game_uri, move_number = state.move_number, passes_in_a_row = state.passes_in_a_row, "hydrated from network" ); } Ok(())}
/// Rebuild one game's live state from its record contents and the ply/// history in both players' repos, then track it. A resignation already/// sitting in either repo means the game was over before this process saw/// it — settle it (and the record, when we own the record) instead of/// tracking a zombie. Returns the state only while the game is live.async fn rebuild_game( app: &App, game_uri: &str, parsed: &ParsedGameRecord,) -> Result<Option<GameState>> { let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str();
let moves = fetch_game_plies(app, game_uri, &parsed.player_one, &parsed.player_two).await?; let mut state = build_game_state(parsed, game_uri, bot_did_str, &moves);
if let Some(winner) = fetch_resignation(app, game_uri, &parsed.player_one, &parsed.player_two).await? { info!(game = %game_uri, winner = %winner, "resignation found while rebuilding — settling"); let bot_owned = game_uri.starts_with(&format!("at://{}/boo.sky.go.game/", bot_did_str)); if bot_owned { finalize_resigned_game(app, game_uri, &winner).await; } state.status = "finished".into(); }
app.db.upsert_game(&state).await?; let live = state.status == "active"; { let mut games = app.games.lock().await; games.insert(game_uri.to_string(), state.clone()); } Ok(live.then_some(state))}
/// Find a resignation for `game_uri` in either player's repo and return/// the winner (the non-resigner). Both wire forms count — see/// `parse_resign` — and either landing ends the game.async fn fetch_resignation( app: &App, game_uri: &str, player_one: &str, player_two: &str,) -> Result<Option<String>> { let mut repos = vec![player_one]; if player_two != player_one { repos.push(player_two); }
let mut resigners = Vec::new(); for repo in repos { for collection in ["boo.sky.go.resign", "boo.sky.go.action"] { for (_, record) in crate::list_repo_records(app, repo, collection).await? { if let Some(parsed) = parse_resign(&record) { if parsed.game_uri == game_uri { resigners.push(parsed.player); } } } } }
Ok(winner_after_resign(player_one, player_two, &resigners).map(str::to_string))}
/// Settle a resigned game record: the non-resigner takes the win. Only the/// game-owning repo can be updated directly — callers guard on ownership.async fn finalize_resigned_game(app: &App, game_uri: &str, winner: &str) { let fields = serde_json::json!({ "status": "completed", "winner": winner }); if let Err(e) = update_game_record(app, game_uri, fields).await { error!(game = %game_uri, error = %e, "failed to finalize game record on resign"); }}
async fn handle_move( app: &App, _did: &str, _rkey: &str, record: &serde_json::Value,) -> Result<()> { let parsed = match parse_move_record(record) { Some(p) => p, None => return Ok(()), };
let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str();
let mut games = app.games.lock().await; let game = match games.get_mut(&parsed.game_uri) { Some(g) => g, None => return Ok(()), };
let captures = match apply_move_to_game(game, &parsed, bot_did_str) { Some(c) => c, None => return Ok(()), };
info!( game = %parsed.game_uri, color = %parsed.color.name(), x = parsed.x, y = parsed.y, move_number = parsed.move_number, captures, "move received" );
debug!("Board state after move:\n{}", format_board(&game.board)); app.db.upsert_game(game).await?;
if check_bot_turn(game) { play_move(app, game).await?; }
Ok(())}
/// Parsed pass record fields — the pure data from a `boo.sky.go.pass` record.struct ParsedPassRecord { game_uri: String, player: String, color: Color, move_number: u32,}
fn parse_pass_record(record: &serde_json::Value) -> Option<ParsedPassRecord> { Some(ParsedPassRecord { game_uri: record.get("game")?.as_str()?.to_string(), player: record .get("player") .and_then(|v| v.as_str()) .unwrap_or("") .to_string(), color: match record.get("color").and_then(|v| v.as_str()) { Some("black") => Color::Black, _ => Color::White, }, move_number: record.get("moveNumber")?.as_u64()? as u32, })}
/// Apply a parsed pass to a GameState. Returns false if it doesn't belong/// to this game, is from the bot itself, or is already reflected in state.fn apply_pass_to_game(game: &mut GameState, parsed: &ParsedPassRecord, bot_did: &str) -> bool { if parsed.game_uri != game.game_uri { return false; } if game.status != "active" { return false; } if parsed.player == bot_did { return false; } if parsed.move_number <= game.move_number { return false; } game.move_number = parsed.move_number; game.passes_in_a_row = game.passes_in_a_row.saturating_add(1); true}
async fn handle_pass( app: &App, _did: &str, _rkey: &str, record: &serde_json::Value,) -> Result<()> { let parsed = match parse_pass_record(record) { Some(p) => p, None => return Ok(()), }; let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str();
let (passes_in_a_row, is_bot_turn) = { let mut games = app.games.lock().await; let game = match games.get_mut(&parsed.game_uri) { Some(g) => g, None => return Ok(()), }; if !apply_pass_to_game(game, &parsed, bot_did_str) { return Ok(()); } app.db.upsert_game(game).await?; info!( game = %parsed.game_uri, color = %parsed.color.name(), move_number = parsed.move_number, passes_in_a_row = game.passes_in_a_row, "pass received" ); (game.passes_in_a_row, check_bot_turn(game)) };
if passes_in_a_row >= 2 { return finish_game_by_scoring(app, &parsed.game_uri).await; }
if is_bot_turn { let mut games = app.games.lock().await; if let Some(game) = games.get_mut(&parsed.game_uri) { play_move(app, game).await?; } }
Ok(())}
async fn handle_action( app: &App, did: &str, rkey: &str, record: &serde_json::Value,) -> Result<()> { match record.get("action").and_then(|v| v.as_str()) { Some("join") => handle_join(app, record).await, // The legacy record form of a resignation (see `parse_resign`). Some("resign") => handle_resign(app, did, rkey, record).await, _ => Ok(()), }}
async fn handle_join(app: &App, record: &serde_json::Value) -> Result<()> { let game_uri = match record.get("game").and_then(|v| v.as_str()) { Some(g) => g.to_string(), None => return Ok(()), }; let player = match record.get("player").and_then(|v| v.as_str()) { Some(p) => p.to_string(), None => return Ok(()), };
let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str(); if player == bot_did_str { return Ok(()); }
let prefix = format!("at://{}/boo.sky.go.game/", bot_did_str); let rkey = match game_uri.strip_prefix(&prefix) { Some(r) => r.to_string(), None => return Ok(()), };
{ let games = app.games.lock().await; if games.contains_key(&game_uri) { return Ok(()); } }
// A join can reference a game deleted since it was posted — stale noise // in replayed streams, and expected often enough that it must not fail // the event. Every other failure still takes the loud error path. // // When a join references a bot-owned game whose record no longer exists // (deleted externally), re-create it so the game can be played. let current = match app .agent .api .com .atproto .repo .get_record( atrium_api::com::atproto::repo::get_record::ParametersData { cid: None, collection: "boo.sky.go.game".parse().unwrap(), repo: bot_did_str.parse().unwrap(), rkey: rkey.as_str().parse().unwrap(), } .into(), ) .await { Ok(current) => current, Err(e) if crate::record_not_found(&e) => { // The game record was deleted externally — re-create it so // the join can proceed and the game can be played. let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); let record = serde_json::json!({ "$type": "boo.sky.go.game", "status": "waiting", "boardSize": 9u64, "createdAt": now, "playerOne": bot_did_str, }); app.agent.api.com.atproto.repo.create_record( atrium_api::com::atproto::repo::create_record::InputData { collection: "boo.sky.go.game".parse().unwrap(), record: serde_json::from_value(record).unwrap(), repo: bot_did_str.parse().unwrap(), rkey: Some(rkey.as_str().parse().unwrap()), swap_commit: None, validate: None, } .into(), ) .await?; info!(game = %game_uri, "re-created missing game record for join"); // Fetch it back so the rest of the function sees the same // shape as an existing record (Object<OutputData>). app.agent.api.com.atproto.repo.get_record( atrium_api::com::atproto::repo::get_record::ParametersData { cid: None, collection: "boo.sky.go.game".parse().unwrap(), repo: bot_did_str.parse().unwrap(), rkey: rkey.as_str().parse().unwrap(), } .into(), ) .await? } Err(e) => return Err(e.into()), };
let current_val = unknown_to_value(¤t.value); let current_obj = match current_val.as_object() { Some(o) => o, None => return Ok(()), };
// Re-created records are already in the right state; for fetched // records, verify they're still joinable. if current_obj.get("status").and_then(|v| v.as_str()) != Some("waiting") { return Ok(()); } if current_obj.get("playerTwo").is_some() { return Ok(()); }
let board_size = current_obj.get("boardSize").and_then(|v| v.as_u64()).unwrap_or(9); let created_at = current_obj .get("createdAt") .and_then(|v| v.as_str()) .unwrap_or_default() .to_string();
let mut updated = serde_json::json!({ "$type": "boo.sky.go.game", "status": "active", "boardSize": board_size, "createdAt": created_at, "playerOne": bot_did_str, "playerTwo": player, }); for key in ["winner", "blackScore", "whiteScore", "blackScorer", "whiteScorer", "deadStones", "handicap", "handicapStones"] { if let Some(v) = current_obj.get(key) { updated.as_object_mut().unwrap().insert(key.to_string(), v.clone()); } }
app.agent .api .com .atproto .repo .put_record( atrium_api::com::atproto::repo::put_record::InputData { collection: "boo.sky.go.game".parse().unwrap(), record: serde_json::from_value(updated).unwrap(), repo: bot_did_str.parse().unwrap(), rkey: rkey.as_str().parse().unwrap(), swap_commit: None, swap_record: current.cid.clone(), validate: None, } .into(), ) .await?;
info!(game = %game_uri, player = %player, board_size, "join activated");
let parsed = ParsedGameRecord { board_size: board_size as usize, player_one: bot_did_str.to_string(), player_two: player, status: "active".into(), }; let game_state = build_game_state(&parsed, &game_uri, bot_did_str, &[]); app.db.upsert_game(&game_state).await?;
let is_bot_turn = check_bot_turn(&game_state); { let mut games = app.games.lock().await; games.insert(game_uri.clone(), game_state); }
if is_bot_turn { let mut games = app.games.lock().await; if let Some(game) = games.get_mut(&game_uri) { play_move(app, game).await?; } }
crate::ensure_waiting_games(app).await?;
Ok(())}
/// Parsed resign fields — accepted from both wire forms: the dedicated/// `boo.sky.go.resign` record (upstream's form since bc4b8353) and the/// legacy `boo.sky.go.action` record with `action: "resign"`.struct ParsedResign { game_uri: String, player: String, color: Option<Color>,}
fn parse_resign(record: &serde_json::Value) -> Option<ParsedResign> { let is_resign = record.get("$type").and_then(|v| v.as_str()) == Some("boo.sky.go.resign") || record.get("action").and_then(|v| v.as_str()) == Some("resign"); if !is_resign { return None; } Some(ParsedResign { game_uri: record.get("game").and_then(|v| v.as_str())?.to_string(), player: record.get("player").and_then(|v| v.as_str())?.to_string(), color: record.get("color").and_then(|v| v.as_str()).map(|c| match c { "black" => Color::Black, _ => Color::White, }), })}
/// A resignation ends the game and the resigner loses. None when the/// resigner is not one of the two players (malformed record) or when both/// sides resigned (no meaningful winner).fn winner_after_resign<'a>( player_one: &'a str, player_two: &'a str, resigners: &[String],) -> Option<&'a str> { let one_resigned = resigners.iter().any(|r| r == player_one); let two_resigned = resigners.iter().any(|r| r == player_two); match (one_resigned, two_resigned) { (true, false) => Some(player_two), (false, true) => Some(player_one), _ => None, }}
/// A resignation ends the game no matter which record form carried it./// Resignations arrive network-wide and may reference games we no longer/// track; stale ones must stay quiet instead of failing the event.async fn handle_resign( app: &App, _did: &str, _rkey: &str, record: &serde_json::Value,) -> Result<()> { let Some(parsed) = parse_resign(record) else { return Ok(()); };
let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str();
let tracked = { let games = app.games.lock().await; games.get(&parsed.game_uri).cloned() }; if let Some(game) = &tracked { if game.status != "active" { return Ok(()); // settled already — redelivery or the second record form } } let bot_owned = parsed .game_uri .starts_with(&format!("at://{}/boo.sky.go.game/", bot_did_str)); if tracked.is_none() && !bot_owned { return Ok(()); // someone else's game }
let (player_one, player_two, record_live) = match &tracked { Some(game) => (game.player_one.clone(), game.player_two.clone(), true), None => { let Some(current) = crate::fetch_game_record(app, &parsed.game_uri).await? else { return Ok(()); // game record gone — nothing left to settle }; let Some(game) = parse_game_record(¤t) else { return Ok(()); }; (game.player_one, game.player_two, game.status == "active") } };
let Some(winner) = winner_after_resign(&player_one, &player_two, std::slice::from_ref(&parsed.player)) else { return Ok(()); // resigner is not one of the two players };
// Only the game-owning repo can be updated directly. if bot_owned && record_live { finalize_resigned_game(app, &parsed.game_uri, winner).await; }
info!( game = %parsed.game_uri, player = %parsed.player, winner = %winner, color = ?parsed.color, "resignation recorded" );
{ let mut games = app.games.lock().await; if let Some(game) = games.get_mut(&parsed.game_uri) { game.status = "finished".into(); app.db.upsert_game(game).await?; } }
crate::ensure_waiting_games(app).await?; Ok(())}
fn check_bot_turn(game: &GameState) -> bool { let next_move = game.move_number + 1; let next_color = if next_move % 2 == 1 { Color::Black } else { Color::White }; next_color == game.bot_color}
async fn play_move(app: &App, game: &GameState) -> Result<()> { let valid = game.board.valid_placements(game.bot_color.as_u8());
if valid.is_empty() { // TK: full board -> pass, not resign // TK: spawned to dodge re-locking app.games under caller's lock info!(game = %game.game_uri, "no valid placements — passing"); let app_clone = app.clone(); let game_clone = game.clone(); tokio::spawn(async move { if let Err(e) = post_pass(&app_clone, &game_clone).await { error!(game = %game_clone.game_uri, error = %e, "pass post failed"); } }); return Ok(()); }
let (state_str, sorted_valid) = crate::state::build_state( &game.board, game.bot_color.as_u8(), game.move_number + 1, game.last_move.as_ref(), );
let decision = match crate::jev::select_move(&app.config.typesafe_api_key, &state_str, &sorted_valid) .await { Ok(d) => d, Err(e) => { error!(game = %game.game_uri, error = %e, "Jev API call failed, using top candidate"); let (x, y) = sorted_valid.first().copied().unwrap_or((0, 0)); crate::jev::JevDecision { action: crate::jev::Action::Place(x, y), margin: 0.0, top_probs: vec![], source: "api_error", } } };
if let crate::jev::Action::Resign = decision.action { info!(game = %game.game_uri, margin = decision.margin, source = decision.source, "Jev chose to resign"); return resign(app, game).await; }
let delay = crate::jev::hesitation_delay(decision.margin); let game_uri = game.game_uri.clone(); let move_number = game.move_number + 1;
// Clone what the background task needs — don't block the event loop let app_clone = app.clone(); let game_clone = game.clone();
tokio::spawn(async move { if !delay.is_zero() { tokio::time::sleep(delay).await; }
let top_probs = format!("{:?}", decision.top_probs); let outcome = match decision.action { crate::jev::Action::Place(x, y) => { post_move(&app_clone, &game_clone, x, y).await.map(Some) } crate::jev::Action::Pass => post_pass(&app_clone, &game_clone).await.map(|_| None), crate::jev::Action::Resign => unreachable!("handled before spawning"), };
match outcome { Ok(captures) => info!( game = %game_uri, color = %game_clone.bot_color.name(), move_number, action = ?decision.action, margin = decision.margin, source = decision.source, hesitation_ms = delay.as_millis() as u64, top_probs, captures = captures.unwrap_or(0), "action posted" ), Err(e) => error!( game = %game_uri, action = ?decision.action, margin = decision.margin, source = decision.source, top_probs, error = %e, "action post failed" ), } });
Ok(())}
async fn post_move(app: &App, game: &GameState, x: usize, y: usize) -> Result<usize> { let move_number = game.move_number + 1; let color_name = game.bot_color.name();
let mut board = game.board.clone(); let captures = board.apply_move(x, y, game.bot_color.as_u8());
let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); let bot_did = app.agent.did().await.expect("authenticated");
let record = serde_json::json!({ "$type": "boo.sky.go.move", "x": x, "y": y, "game": game.game_uri, "color": color_name, "player": bot_did.as_str(), "createdAt": now, "moveNumber": move_number, "captureCount": captures, });
app.agent .api .com .atproto .repo .create_record( atrium_api::com::atproto::repo::create_record::InputData { collection: "boo.sky.go.move".parse().unwrap(), record: serde_json::from_value(record).unwrap(), repo: bot_did.as_str().parse().unwrap(), rkey: None, swap_commit: None, validate: None, } .into(), ) .await?;
let mut games = app.games.lock().await; if let Some(game) = games.get_mut(&game.game_uri) { game.board = board; game.move_number = move_number; game.passes_in_a_row = 0; game.last_move = Some(LastMove { x, y, color: game.bot_color, captures, }); app.db.upsert_game(game).await?; }
crate::ensure_waiting_games(app).await?;
Ok(captures)}
/// Post a pass (per the boo.sky.go.pass lexicon: same moveNumber sequence/// as moves, no coordinates). If this is the second pass in a row, the/// game is over — settle it by area score instead of replenishing the pool.async fn post_pass(app: &App, game: &GameState) -> Result<()> { let move_number = game.move_number + 1; let color_name = game.bot_color.name(); let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); let bot_did = app.agent.did().await.expect("authenticated");
let record = serde_json::json!({ "$type": "boo.sky.go.pass", "game": game.game_uri, "player": bot_did.as_str(), "moveNumber": move_number, "color": color_name, "createdAt": now, });
app.agent .api .com .atproto .repo .create_record( atrium_api::com::atproto::repo::create_record::InputData { collection: "boo.sky.go.pass".parse().unwrap(), record: serde_json::from_value(record).unwrap(), repo: bot_did.as_str().parse().unwrap(), rkey: None, swap_commit: None, validate: None, } .into(), ) .await?;
let passes_in_a_row = { let mut games = app.games.lock().await; match games.get_mut(&game.game_uri) { Some(g) => { g.move_number = move_number; g.passes_in_a_row = g.passes_in_a_row.saturating_add(1); app.db.upsert_game(g).await?; g.passes_in_a_row } None => 0, } };
if passes_in_a_row >= 2 { finish_game_by_scoring(app, &game.game_uri).await?; } else { crate::ensure_waiting_games(app).await?; }
Ok(())}
async fn resign(app: &App, game: &GameState) -> Result<()> { info!(game = %game.game_uri, "resigning");
let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str(); let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
// Record the resignation itself, in our own repo — works regardless of // whether we own the game record (same boo.sky.go.action pattern as join). // The network has used two record forms for this; post both so neither // generation of clients misses it (ingestion dedupes the pair). let records = [ ( "boo.sky.go.resign", serde_json::json!({ "$type": "boo.sky.go.resign", "game": game.game_uri, "player": bot_did_str, "color": game.bot_color.name(), "createdAt": now, }), ), ( "boo.sky.go.action", serde_json::json!({ "$type": "boo.sky.go.action", "game": game.game_uri, "player": bot_did_str, "action": "resign", "createdAt": now, }), ), ];
let mut posted = 0; for (collection, record) in records { let result = app .agent .api .com .atproto .repo .create_record( atrium_api::com::atproto::repo::create_record::InputData { collection: collection.parse().unwrap(), record: serde_json::from_value(record).unwrap(), repo: bot_did_str.parse().unwrap(), rkey: None, swap_commit: None, validate: None, } .into(), ) .await; match result { Ok(_) => posted += 1, Err(e) => error!(game = %game.game_uri, collection, error = %e, "resign record post failed"), } } // One form reaching the network is enough for the game to end. if posted == 0 { anyhow::bail!("no resign record form could be posted for {}", game.game_uri); }
// Only the game-owning repo can be updated directly; when we're // playerTwo the action record above is the only signal we can send. if game.player_one == bot_did_str { let winner = game.player_two.clone(); let fields = serde_json::json!({ "status": "completed", "winner": winner }); if let Err(e) = update_game_record(app, &game.game_uri, fields).await { error!(game = %game.game_uri, error = %e, "failed to finalize game record on resign"); } }
let mut games = app.games.lock().await; if let Some(game) = games.get_mut(&game.game_uri) { game.status = "finished".into(); app.db.upsert_game(game).await?; }
crate::ensure_waiting_games(app).await?; Ok(())}
/// End a game after two consecutive passes: settle it by Chinese area/// score (stones + surrounded territory, no dead-stone marking — every/// stone on the board counts as alive).async fn finish_game_by_scoring(app: &App, game_uri: &str) -> Result<()> { let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str();
let (black_score, white_score, player_one, player_two, bot_is_p1) = { let games = app.games.lock().await; match games.get(game_uri) { Some(g) => { let (black, white) = g.board.area_score(); (black, white, g.player_one.clone(), g.player_two.clone(), g.player_one == bot_did_str) } None => return Ok(()), } };
info!(game = %game_uri, black_score, white_score, "game finished by two passes, settling score");
// Only the game-owning repo can be updated directly; when we're // playerTwo, we can't submit the score ourselves. if bot_is_p1 { if let Err(e) = finalize_scored_game(app, game_uri, black_score, white_score, &player_one, &player_two).await { error!(game = %game_uri, error = %e, "failed to finalize scored game record"); } }
let mut games = app.games.lock().await; if let Some(g) = games.get_mut(game_uri) { g.status = "finished".into(); app.db.upsert_game(g).await?; } drop(games);
crate::ensure_waiting_games(app).await?; Ok(())}
/// Merge `fields` into the bot-owned game record and write it back,/// preserving every other field. Uses swap_record so a concurrent update/// can't be silently clobbered.async fn update_game_record(app: &App, game_uri: &str, fields: serde_json::Value) -> Result<()> { let bot_did = app.agent.did().await.expect("authenticated"); let bot_did_str = bot_did.as_str(); let rkey = game_uri.rsplit('/').next().unwrap_or_default();
let current = app .agent .api .com .atproto .repo .get_record( atrium_api::com::atproto::repo::get_record::ParametersData { cid: None, collection: "boo.sky.go.game".parse().unwrap(), repo: bot_did_str.parse().unwrap(), rkey: rkey.parse().unwrap(), } .into(), ) .await?;
let mut updated = unknown_to_value(¤t.value); let obj = match updated.as_object_mut() { Some(o) => o, None => return Ok(()), }; if let Some(new_fields) = fields.as_object() { for (k, v) in new_fields { obj.insert(k.clone(), v.clone()); } }
app.agent .api .com .atproto .repo .put_record( atrium_api::com::atproto::repo::put_record::InputData { collection: "boo.sky.go.game".parse().unwrap(), record: serde_json::from_value(updated).unwrap(), repo: bot_did_str.parse().unwrap(), rkey: rkey.parse().unwrap(), swap_commit: None, swap_record: current.cid.clone(), validate: None, } .into(), ) .await?;
Ok(())}
/// Winner is whoever has the higher area score; on a tie, no winner field/// is set (rare with area scoring, but possible on tiny/even boards).async fn finalize_scored_game( app: &App, game_uri: &str, black_score: usize, white_score: usize, player_one_black: &str, player_two_white: &str,) -> Result<()> { let mut fields = serde_json::json!({ "status": "completed", "blackScore": black_score, "whiteScore": white_score, }); // TK: player_one is black, player_two is white (see build_game_state) if black_score > white_score { fields.as_object_mut().unwrap().insert("winner".into(), serde_json::json!(player_one_black)); } else if white_score > black_score { fields.as_object_mut().unwrap().insert("winner".into(), serde_json::json!(player_two_white)); } update_game_record(app, game_uri, fields).await}
pub async fn run_event_loop(app: &App) -> Result<()> { // Resume where the last run stopped checkpointing instead of tailing // live-only — without this, any downtime (deploy, crash, network blip) // permanently drops game/move/action events that happened while // disconnected (this is exactly how two different join actions sat // unprocessed for hours). The 24h floor bounds only a first boot with // no cursor yet. Safe to replay: handle_game/handle_action guard on // already-tracked state and CAS, and apply_move_to_game/build_game_state/ // replay_moves all skip already-applied or occupied-cell moves instead // of panicking. let start = app .db .get_cursor() .await? .unwrap_or_else(|| (chrono::Utc::now() - chrono::Duration::hours(24)).timestamp_micros()); let cursor = Arc::new(AtomicI64::new(start));
let handler = Arc::new(GameHandler { app: app.clone(), cursor: cursor.clone(), });
let hostname = app .config .jetstream_url .trim_start_matches("wss://") .trim_end_matches("/subscribe") .to_string();
// The consumer runs one connection and returns when the socket closes; // a live feed must reconnect on its own or ingest dies silently until // the next deploy. Backoff doubles while reconnects keep failing and // resets once a connection survives 60 seconds. let mut backoff = std::time::Duration::from_secs(2); let max_backoff = std::time::Duration::from_secs(120); loop { let connected_at = std::time::Instant::now();
let config = ConsumerTaskConfig { user_agent: "jevgo/0.1".to_string(), compression: false, zstd_dictionary_location: String::new(), jetstream_hostname: hostname.clone(), collections: vec![ "boo.sky.go.game".to_string(), "boo.sky.go.move".to_string(), "boo.sky.go.pass".to_string(), "boo.sky.go.action".to_string(), "boo.sky.go.resign".to_string(), ], dids: vec![], max_message_size_bytes: None, cursor: Some(cursor.load(Ordering::Relaxed)), require_hello: true, };
let consumer = Consumer::new(config); consumer.register_handler(handler.clone()).await?;
let token = CancellationToken::new(); match consumer.run_background(token).await { Ok(()) => warn!("jetstream connection closed; reconnecting"), Err(e) => error!(error = %e, "jetstream connection failed; retrying"), }
let healthy = connected_at.elapsed() >= std::time::Duration::from_secs(60); if healthy { backoff = std::time::Duration::from_secs(2); } tokio::time::sleep(backoff).await; if !healthy { backoff = (backoff * 2).min(max_backoff); } }}
pub async fn reconcile_games(app: &App) -> Result<()> { let games_snapshot: Vec<(String, String, String)> = { let games = app.games.lock().await; games .values() .filter(|g| g.status == "active") .map(|g| (g.game_uri.clone(), g.player_one.clone(), g.player_two.clone())) .collect() };
for (game_uri, player_one, player_two) in games_snapshot { let plies = fetch_game_plies(app, &game_uri, &player_one, &player_two).await?;
let mut finished = false; { let mut games = app.games.lock().await; if let Some(game) = games.get_mut(&game_uri) { for m in &plies { if m.move_number <= game.move_number { continue; } if m.is_pass { game.move_number = m.move_number; game.passes_in_a_row = game.passes_in_a_row.saturating_add(1); continue; } let caps = match game.board.try_apply_move(m.x, m.y, m.color.as_u8()) { Some(c) => c, None => { error!( game = %game_uri, x = m.x, y = m.y, move_number = m.move_number, "skipping corrupt move: lands on an occupied cell" ); continue; } }; game.move_number = m.move_number; game.passes_in_a_row = 0; game.last_move = Some(LastMove { x: m.x, y: m.y, color: m.color, captures: caps, }); }
app.db.upsert_game(game).await?;
if game.passes_in_a_row >= 2 { finished = true; } else if check_bot_turn(game) { play_move(app, game).await?; } } }
if finished { finish_game_by_scoring(app, &game_uri).await?; } }
Ok(())}
fn format_board(board: &Board) -> String { let mut s = String::new(); s.push_str(" "); for x in 0..board.size { s.push_str(&format!("{} ", x)); } s.push('\n'); for y in 0..board.size { s.push_str(&format!("{} ", y)); for x in 0..board.size { s.push_str(match board.cells[y][x] { 0 => "· ", 1 => "● ", 2 => "○ ", _ => "? ", }); } s.push('\n'); } s}
#[cfg(test)]mod proptests { use super::*; use proptest::prelude::*;
const BOT_DID: &str = "did:plc:bot123";
/// Strategy for generating a game record JSON. fn game_record(bot_is_p1: bool, status: &str) -> serde_json::Value { serde_json::json!({ "$type": "boo.sky.go.game", "status": status, "boardSize": 9, "createdAt": "2026-01-01T00:00:00.000Z", "playerOne": if bot_is_p1 { BOT_DID } else { "did:plc:opponent" }, "playerTwo": if bot_is_p1 { "did:plc:opponent" } else { BOT_DID }, }) }
/// Strategy for a sequence of move indices (picks from valid_placements each turn). fn move_indices(max_moves: usize) -> impl Strategy<Value = Vec<usize>> { prop::collection::vec(0..max_moves, 0..max_moves) }
/// Build a sequence of MoveRecords by playing random moves on a board. fn generate_move_records( indices: &[usize], _game_uri: &str, _p1_did: &str, _p2_did: &str, ) -> Vec<MoveRecord> { let mut board = Board::new(9); let mut records = Vec::new(); let mut color: u8 = 1; // black let mut move_number: u32 = 0;
for &idx in indices { let valid = board.valid_placements(color); if valid.is_empty() { break; } let move_idx = idx % valid.len(); let (x, y) = valid[move_idx]; board.apply_move(x, y, color); move_number += 1;
records.push(MoveRecord { x, y, color: Color::from_u8(color), move_number, is_pass: false, });
color = if color == 1 { 2 } else { 1 }; }
records }
proptest! { #[test] fn parse_game_record_requires_player_two( bot_is_p1 in any::<bool>(), ) { // With playerTwo present → Some let with_p2 = game_record(bot_is_p1, "active"); assert!(parse_game_record(&with_p2).is_some());
// Without playerTwo → None let mut without_p2 = with_p2.clone(); without_p2.as_object_mut().unwrap().remove("playerTwo"); assert!(parse_game_record(&without_p2).is_none()); }
#[test] fn bot_color_derived_from_player_position( bot_is_p1 in any::<bool>(), ) { let record = game_record(bot_is_p1, "active"); let parsed = parse_game_record(&record).unwrap(); let state = build_game_state(&parsed, "at://test/boo.sky.go.game/abc", BOT_DID, &[]);
if bot_is_p1 { assert_eq!(state.bot_color, Color::Black); } else { assert_eq!(state.bot_color, Color::White); } }
#[test] fn turn_detection_matches_color_and_move_number( bot_is_p1 in any::<bool>(), move_count in 0u32..100, ) { let record = game_record(bot_is_p1, "active"); let parsed = parse_game_record(&record).unwrap(); let mut state = build_game_state(&parsed, "at://test/boo.sky.go.game/abc", BOT_DID, &[]); state.move_number = move_count;
let is_bot_turn = check_bot_turn(&state); let next_move = move_count + 1; let next_color = if next_move % 2 == 1 { Color::Black } else { Color::White };
// check_bot_turn should be true iff next color matches bot color assert_eq!(is_bot_turn, next_color == state.bot_color, "move_count={}, bot_color={:?}, next_color={:?}", move_count, state.bot_color, next_color); }
#[test] fn build_game_state_replays_moves_correctly( indices in move_indices(40), bot_is_p1 in any::<bool>(), ) { let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(bot_is_p1, "active"); let parsed = parse_game_record(&record).unwrap(); let moves = generate_move_records(&indices, game_uri, "did:plc:opponent", BOT_DID);
let state = build_game_state(&parsed, game_uri, BOT_DID, &moves);
// move_number should match the last move's number if let Some(last) = moves.last() { assert_eq!(state.move_number, last.move_number); } else { assert_eq!(state.move_number, 0); }
// All cells valid for y in 0..state.board.size { for x in 0..state.board.size { assert!(state.board.cells[y][x] <= 2); } }
assert_eq!(state.board_size, 9); }
#[test] fn apply_move_to_game_updates_state( indices in move_indices(30), bot_is_p1 in any::<bool>(), ) { let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(bot_is_p1, "active"); let parsed = parse_game_record(&record).unwrap();
// Build initial game state with some moves let initial_moves = generate_move_records(&indices, game_uri, "did:plc:opponent", BOT_DID); let mut state = build_game_state(&parsed, game_uri, BOT_DID, &initial_moves); let initial_move_number = state.move_number;
// Apply one more opponent move let valid = state.board.valid_placements( if initial_move_number % 2 == 0 { 1 } else { 2 } ); if let Some(&(x, y)) = valid.first() { let next_color = if initial_move_number % 2 == 0 { 1 } else { 2 }; let move_record = serde_json::json!({ "x": x, "y": y, "color": if next_color == 1 { "black" } else { "white" }, "game": game_uri, "player": "did:plc:opponent", "moveNumber": initial_move_number + 1, }); let parsed_move = parse_move_record(&move_record).unwrap(); let result = apply_move_to_game(&mut state, &parsed_move, BOT_DID);
assert!(result.is_some(), "Move should have been applied"); assert_eq!(state.move_number, initial_move_number + 1); assert!(state.last_move.is_some()); } }
#[test] fn apply_move_ignores_bot_own_moves( indices in move_indices(30), bot_is_p1 in any::<bool>(), ) { let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(bot_is_p1, "active"); let parsed = parse_game_record(&record).unwrap();
let initial_moves = generate_move_records(&indices, game_uri, "did:plc:opponent", BOT_DID); let mut state = build_game_state(&parsed, game_uri, BOT_DID, &initial_moves); let initial_move_number = state.move_number;
// Try to apply a move from the bot itself let valid = state.board.valid_placements(1); if let Some(&(x, y)) = valid.first() { let move_record = serde_json::json!({ "x": x, "y": y, "color": "black", "game": game_uri, "player": BOT_DID, "moveNumber": initial_move_number + 1, }); let parsed_move = parse_move_record(&move_record).unwrap(); let result = apply_move_to_game(&mut state, &parsed_move, BOT_DID);
assert!(result.is_none(), "Bot's own moves should be ignored"); assert_eq!(state.move_number, initial_move_number, "State should not change"); } }
#[test] fn apply_move_ignores_already_applied_move( indices in move_indices(30), bot_is_p1 in any::<bool>(), ) { // A Jetstream reconnect or backfill can redeliver an event // already reflected in state — must not reapply (and must not // panic trying to place a stone on an already-occupied cell). let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(bot_is_p1, "active"); let parsed = parse_game_record(&record).unwrap();
let initial_moves = generate_move_records(&indices, game_uri, "did:plc:opponent", BOT_DID); if let Some(last) = initial_moves.last().cloned() { let mut state = build_game_state(&parsed, game_uri, BOT_DID, &initial_moves); let move_record = serde_json::json!({ "x": last.x, "y": last.y, "color": last.color.name(), "game": game_uri, "player": "did:plc:opponent", "moveNumber": last.move_number, }); let parsed_move = parse_move_record(&move_record).unwrap(); let result = apply_move_to_game(&mut state, &parsed_move, BOT_DID);
assert!(result.is_none(), "Already-applied move must be ignored, not reapplied"); assert_eq!(state.move_number, last.move_number); } }
#[test] fn apply_move_ignores_wrong_game( indices in move_indices(20), ) { let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(true, "active"); let parsed = parse_game_record(&record).unwrap();
let initial_moves = generate_move_records(&indices, game_uri, "did:plc:opponent", BOT_DID); let mut state = build_game_state(&parsed, game_uri, BOT_DID, &initial_moves); let initial_move_number = state.move_number;
// Move for a different game let move_record = serde_json::json!({ "x": 4, "y": 4, "color": "black", "game": "at://other/boo.sky.go.game/xyz", "player": "did:plc:opponent", "moveNumber": 1, }); let parsed_move = parse_move_record(&move_record).unwrap(); let result = apply_move_to_game(&mut state, &parsed_move, BOT_DID);
assert!(result.is_none()); assert_eq!(state.move_number, initial_move_number); }
#[test] fn full_simulation_state_consistent( indices in move_indices(60), bot_is_p1 in any::<bool>(), ) { // Simulate a full game: create state, apply moves one at a time, // check invariants after each move let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(bot_is_p1, "active"); let parsed = parse_game_record(&record).unwrap();
let moves = generate_move_records(&indices, game_uri, "did:plc:opponent", BOT_DID); let mut state = build_game_state(&parsed, game_uri, BOT_DID, &[]);
for m in &moves { let move_json = serde_json::json!({ "x": m.x, "y": m.y, "color": m.color.name(), "game": game_uri, "player": "did:plc:opponent", "moveNumber": m.move_number, }); let parsed_move = parse_move_record(&move_json).unwrap(); let old_move_number = state.move_number; apply_move_to_game(&mut state, &parsed_move, BOT_DID);
// Invariants after each move assert_eq!(state.move_number, m.move_number); assert!(state.move_number > old_move_number || m.move_number == 0);
// All cells valid for y in 0..state.board.size { for x in 0..state.board.size { assert!(state.board.cells[y][x] <= 2, "Invalid cell {} at ({},{}) after move {}", state.board.cells[y][x], x, y, m.move_number); } }
// Last move is always set after applying a move assert!(state.last_move.is_some());
assert_eq!(state.board_size, 9); } } }
#[test] fn parse_pass_record_reads_fields() { let record = serde_json::json!({ "$type": "boo.sky.go.pass", "game": "at://test/boo.sky.go.game/abc", "player": "did:plc:opponent", "moveNumber": 5, "color": "white", "createdAt": "2026-01-01T00:00:00.000Z", }); let parsed = parse_pass_record(&record).expect("valid pass record"); assert_eq!(parsed.game_uri, "at://test/boo.sky.go.game/abc"); assert_eq!(parsed.player, "did:plc:opponent"); assert_eq!(parsed.move_number, 5); assert_eq!(parsed.color, Color::White); }
#[test] fn parse_pass_record_requires_game_and_move_number() { let missing_game = serde_json::json!({ "player": "did:plc:opponent", "moveNumber": 1, }); assert!(parse_pass_record(&missing_game).is_none());
let missing_move_number = serde_json::json!({ "game": "at://test/boo.sky.go.game/abc", "player": "did:plc:opponent", }); assert!(parse_pass_record(&missing_move_number).is_none()); }
#[test] fn apply_pass_to_game_advances_turn_and_streak() { let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(true, "active"); let parsed = parse_game_record(&record).unwrap(); let mut state = build_game_state(&parsed, game_uri, BOT_DID, &[]);
let pass_record = serde_json::json!({ "game": game_uri, "player": "did:plc:opponent", "moveNumber": 1, "color": "black", }); let parsed_pass = parse_pass_record(&pass_record).unwrap(); assert!(apply_pass_to_game(&mut state, &parsed_pass, BOT_DID)); assert_eq!(state.move_number, 1); assert_eq!(state.passes_in_a_row, 1); }
#[test] fn apply_pass_to_game_ignores_bot_own_and_stale_passes() { let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(true, "active"); let parsed = parse_game_record(&record).unwrap(); let mut state = build_game_state(&parsed, game_uri, BOT_DID, &[]);
let own_pass = serde_json::json!({ "game": game_uri, "player": BOT_DID, "moveNumber": 1, "color": "black", }); assert!(!apply_pass_to_game(&mut state, &parse_pass_record(&own_pass).unwrap(), BOT_DID)); assert_eq!(state.passes_in_a_row, 0, "bot's own pass must not be applied");
state.move_number = 5; let stale_pass = serde_json::json!({ "game": game_uri, "player": "did:plc:opponent", "moveNumber": 5, "color": "white", }); assert!(!apply_pass_to_game(&mut state, &parse_pass_record(&stale_pass).unwrap(), BOT_DID)); assert_eq!(state.passes_in_a_row, 0, "already-applied pass must be ignored"); }
#[test] fn parse_resign_reads_dedicated_record_form() { // Shape mirrors a real boo.sky.go.resign record from the wild. let record = serde_json::json!({ "$type": "boo.sky.go.resign", "game": "at://test/boo.sky.go.game/abc", "color": "white", "player": "did:plc:opponent", "createdAt": "2026-09-23T19:37:34.912Z", }); let parsed = parse_resign(&record).expect("valid resign record"); assert_eq!(parsed.game_uri, "at://test/boo.sky.go.game/abc"); assert_eq!(parsed.player, "did:plc:opponent"); assert_eq!(parsed.color, Some(Color::White)); }
#[test] fn parse_resign_reads_legacy_action_form() { let record = serde_json::json!({ "$type": "boo.sky.go.action", "game": "at://test/boo.sky.go.game/abc", "player": "did:plc:opponent", "action": "resign", "createdAt": "2026-09-23T19:37:34.912Z", }); let parsed = parse_resign(&record).expect("legacy resign action"); assert_eq!(parsed.game_uri, "at://test/boo.sky.go.game/abc"); assert_eq!(parsed.player, "did:plc:opponent"); assert_eq!(parsed.color, None); }
#[test] fn parse_resign_rejects_joins_and_malformed_records() { let join = serde_json::json!({ "$type": "boo.sky.go.action", "game": "at://test/boo.sky.go.game/abc", "player": "did:plc:opponent", "action": "join", }); assert!(parse_resign(&join).is_none(), "a join is not a resignation");
let no_game = serde_json::json!({ "$type": "boo.sky.go.resign", "player": "did:plc:opponent" }); assert!(parse_resign(&no_game).is_none(), "a resign without a game is malformed"); }
#[test] fn winner_after_resign_is_the_other_player() { let p1 = "did:plc:black".to_string(); let p2 = "did:plc:white".to_string();
assert_eq!(winner_after_resign(&p1, &p2, &[p1.clone()]), Some(p2.as_str())); assert_eq!(winner_after_resign(&p1, &p2, &[p2.clone()]), Some(p1.as_str())); assert_eq!( winner_after_resign(&p1, &p2, &[p1.clone(), p2.clone()]), None, "both resigned — no meaningful winner" ); assert_eq!( winner_after_resign(&p1, &p2, &["did:plc:stranger".to_string()]), None, "a resigner outside the game is malformed" ); }
#[test] fn apply_move_ignores_finished_games() { let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(true, "active"); let parsed = parse_game_record(&record).unwrap(); let mut state = build_game_state(&parsed, game_uri, BOT_DID, &[]); state.status = "finished".into();
let move_record = serde_json::json!({ "x": 4, "y": 4, "color": "black", "game": game_uri, "player": "did:plc:opponent", "moveNumber": 1, }); let parsed_move = parse_move_record(&move_record).unwrap(); assert!( apply_move_to_game(&mut state, &parsed_move, BOT_DID).is_none(), "a finished game must not accept moves" ); assert_eq!(state.move_number, 0); }
#[test] fn apply_pass_ignores_finished_games() { let game_uri = "at://test/boo.sky.go.game/abc"; let record = game_record(true, "active"); let parsed = parse_game_record(&record).unwrap(); let mut state = build_game_state(&parsed, game_uri, BOT_DID, &[]); state.status = "finished".into();
let pass_record = serde_json::json!({ "game": game_uri, "player": "did:plc:opponent", "moveNumber": 1, "color": "white", }); assert!( !apply_pass_to_game(&mut state, &parse_pass_record(&pass_record).unwrap(), BOT_DID), "a finished game must not accept passes" ); assert_eq!(state.passes_in_a_row, 0); }}