diff --git a/bot/forward/manager.ts b/bot/forward/manager.ts index f45916d..a99dd4b 100644 --- a/bot/forward/manager.ts +++ b/bot/forward/manager.ts @@ -252,6 +252,9 @@ export class NodeForwardManager { case 'get_data_file_list': case 'get_file_data': case 'execute_function': + case 'world.snapshot': + case 'world.chunk': + case 'world.unwatch': case 'workflow.start': case 'workflow.stop': case 'workflow.snapshot': @@ -277,6 +280,14 @@ export class NodeForwardManager { const listing = data as {expires_at: number}; listing.expires_at = Math.min(listing.expires_at, entry.lease || entry.command.expires_at); } + } else if (request.method === 'world.snapshot') { + data = await this.until_deadline(entry.stream.world_snapshot(request.payload, request.deadline_at, check_transport), + request.deadline_at, () => {}); + } else if (request.method === 'world.chunk') { + data = await this.until_deadline(entry.stream.world_chunk(request.payload, request.deadline_at, check_transport), + request.deadline_at, () => {}); + } else if (request.method === 'world.unwatch') { + data = entry.stream.world_unwatch(request.payload, request.deadline_at, check_transport); } else if (request.method === 'workflow.start') { data = entry.stream.workflow_start(request.payload, request.deadline_at, check_transport); } else if (request.method === 'workflow.stop') { @@ -341,8 +352,15 @@ export class NodeForwardManager { let window = Date.now(); let count = 0; entry.unsubscribe = entry.stream!.subscribe(event => { - if ((event.name !== 'log' && event.name !== 'info_update') || entry.socket !== socket || !socket.connected + if ((event.name !== 'log' && event.name !== 'info_update' && event.name !== 'status_update') || entry.socket !== socket || !socket.connected || !entry.ready || entry.terminal || this.stopped) return; + if (event.name === 'status_update') { + // Lifecycle state must survive a busy log stream; reconnect rather + // than silently retaining an online snapshot when the queue is full. + if ((socket.io.engine?.writeBuffer.length ?? 0) >= 64) this.close_transport(entry); + else socket.emit('bot_event', {...event, forward_session_id: entry.command.forward_session_id}); + return; + } if (Date.now() - window >= 1000) { window = Date.now(); count = 0; } if (++count > 150 || (socket.io.engine?.writeBuffer.length ?? 0) >= 64) return; if (event.name === 'log') { diff --git a/bot/forward/stream.ts b/bot/forward/stream.ts index 973e110..cc76cc5 100644 --- a/bot/forward/stream.ts +++ b/bot/forward/stream.ts @@ -12,11 +12,13 @@ import type {RunSnapshot, StartResult, WorkflowManager, WorkflowCommand, Workflo import type {WorkflowCatalog} from '../../types/bot/workflow/metadata.ts'; import {get_workflow_catalog} from '../workflow/catalog.ts'; import {summarize_run, type RunSummary} from '../workflow/runtime/summary.ts'; -import {object, validate_workflow_payload} from '../../utils/forward_protocol.ts'; +import {object, validate_workflow_payload, validate_world_payload} from '../../utils/forward_protocol.ts'; +import {WorldPreview, type WorldBot} from './world.ts'; +import type {WorldSnapshot, WorldVoxelChunk} from '../../types/forward.ts'; export interface StreamContext { readonly bot_status: bot_status; - readonly bot_instance: { + readonly bot_instance: ({ chat(message: string): unknown; players?: Record; tabComplete?(text: string, assumeCommand?: boolean, sendBlockInSight?: boolean, timeout?: number): Promise; @@ -28,7 +30,7 @@ export interface StreamContext { on(name: string, listener: (packet: any) => void): unknown; off(name: string, listener: (packet: any) => void): unknown; }; - } | null; + } & Partial> & {game?: Partial}) | null; readonly logger?: {read_recent_log(count: number): Promise}; readonly info_manager?: Pick | null; readonly function_manager?: Pick | null; @@ -326,6 +328,7 @@ export class BotStream { private context_listeners: Array<[string, (...args: any[]) => void]> = []; private unobserve: (() => void) | undefined; private disposed = false; + private world_preview?: WorldPreview; constructor(readonly bot_id: number, options: BotStreamOptions = {}) { if (!Number.isSafeInteger(bot_id) || bot_id < 1) throw new BotStreamError('INVALID_REQUEST', 'Invalid bot ID'); @@ -402,6 +405,8 @@ export class BotStream { } private detach() { + this.world_preview?.dispose(); + this.world_preview = undefined; for (const [event, listener] of this.context_listeners) this.context?.off(event, listener); this.context_listeners = []; } @@ -414,6 +419,10 @@ export class BotStream { } private status_event() { + if (this.context?.bot_status !== 'online' || this.world_preview?.bot !== this.context?.bot_instance) { + this.world_preview?.dispose(); + this.world_preview = undefined; + } this.publish('status_update', this.state()); } @@ -463,6 +472,73 @@ export class BotStream { }; } + private world_context(deadline_at: number, check_transport: () => void, existing = false) { + this.active(); + check_transport(); + if (!Number.isSafeInteger(deadline_at) || deadline_at <= this.now()) + throw new BotStreamError('TIMEOUT', 'World request deadline expired'); + const context = this.context; + if (!context) throw new BotStreamError('NO_CONTEXT', 'Bot context does not exist'); + const instance = context.bot_instance; + if (context.bot_status !== 'online' || !instance) { + this.world_preview?.dispose(); + this.world_preview = undefined; + throw new BotStreamError('BOT_NOT_ONLINE', 'Bot is not online'); + } + if (!instance.world || !instance.game || !instance.entity || !instance.registry || !instance.version + || typeof instance.on !== 'function' || typeof instance.off !== 'function') + throw new BotStreamError('INTERNAL_ERROR', 'Bot world is unavailable'); + if (this.world_preview?.bot !== instance || this.world_preview?.is_disposed) { + this.world_preview?.dispose(); + this.world_preview = undefined; + } + if (existing && !this.world_preview) throw new BotStreamError('CANCELLED', 'World watch is unavailable'); + const service = this.world_preview ??= new WorldPreview(instance as WorldBot, generation => { + if (!this.disposed && this.source.get(this.bot_id) === context && context.bot_instance === instance + && context.bot_status === 'online') this.publish('info_update', {name: 'world_preview', value: {world_generation: generation}}); + }, this.now); + const check = () => { + check_transport(); + if (deadline_at <= this.now()) throw new BotStreamError('TIMEOUT', 'World request deadline expired'); + if (this.disposed || this.source.get(this.bot_id) !== context || context.bot_instance !== instance + || this.world_preview !== service) { + service.dispose(); + throw new BotStreamError('CANCELLED', 'Bot context changed'); + } + if (context.bot_status !== 'online') { + service.dispose(); + throw new BotStreamError('BOT_NOT_ONLINE', 'Bot is not online'); + } + }; + check(); + return {service, check}; + } + + async world_snapshot(payload: unknown, deadline_at: number, check_transport: () => void = () => {}): Promise { + const input = structuredClone(validate_world_payload('world.snapshot', payload)); + const {service, check} = this.world_context(deadline_at, check_transport); + const result = await service.snapshot(input, check); + check(); + return {...result, context_generation: this.generation!, stream_id: this.stream_id}; + } + + async world_chunk(payload: unknown, deadline_at: number, check_transport: () => void = () => {}): Promise { + const input = structuredClone(validate_world_payload('world.chunk', payload)); + const {service, check} = this.world_context(deadline_at, check_transport, true); + const result = await service.chunk(input, check); + check(); + return result; + } + + world_unwatch(payload: unknown, deadline_at: number, check_transport: () => void = () => {}): {released: true} { + const input = validate_world_payload('world.unwatch', payload); + this.active(); + check_transport(); + if (!Number.isSafeInteger(deadline_at) || deadline_at <= this.now()) + throw new BotStreamError('TIMEOUT', 'World request deadline expired'); + return this.world_preview?.unwatch(input.watch_id) ?? {released: true}; + } + get_context_info(payload: unknown, deadline_at: number): InfoSnapshot[] { this.active(); if (!exact(payload, [])) throw new BotStreamError('INVALID_REQUEST', 'Invalid info request'); diff --git a/bot/forward/world.ts b/bot/forward/world.ts new file mode 100644 index 0000000..19a8239 --- /dev/null +++ b/bot/forward/world.ts @@ -0,0 +1,616 @@ +import {randomUUID} from 'node:crypto'; +import {setImmediate as yield_turn} from 'node:timers/promises'; +import type {WorldChunk, WorldPayloads, WorldSnapshot, WorldVoxelChunk} from '../../types/forward.ts'; +import {ForwardFailure} from '../../utils/forward_protocol.ts'; + +type Position = {x: number; y: number; z: number}; +export interface WorldColumn {getBlockStateId(position: Position): number} +type LoadedWorld = {getColumn(x: number, z: number): WorldColumn | null | undefined}; +export interface WorldBot { + world: Partial & {sync?: LoadedWorld}; + game: {minY: number; height: number; dimension: string}; + entity: {position: Position; yaw: number}; + registry: {blocksByStateId: Record}>; + blocksByName?: Record}; + version: string; + on(event: any, listener: (...args: any[]) => void): unknown; + off(event: any, listener: (...args: any[]) => void): unknown; +} + +type Watch = {radius: number; expires_at: number; world_generation?: string}; +type ColumnCache = {column: WorldColumn; revision: string; chunk?: WorldChunk}; +const air = new Set(['air', 'cave_air', 'void_air']); +const water_blocks = new Set(['water', 'bubble_column', 'kelp', 'kelp_plant', 'seagrass', 'tall_seagrass']); +const colors: Record = { + air: 0, grass_block: 0x78a54a, short_grass: 0x76a644, tall_grass: 0x76a644, + fern: 0x548c3b, large_fern: 0x548c3b, moss_block: 0x637e36, dirt: 0x866043, + coarse_dirt: 0x79553a, podzol: 0x674b2a, mud: 0x49403b, mycelium: 0x766376, + water: 0x3f76e4, bubble_column: 0x3f76e4, lava: 0xf36b19, + stone: 0x858585, cobblestone: 0x787878, deepslate: 0x505054, bedrock: 0x555555, + granite: 0x956856, diorite: 0xbcbcbc, andesite: 0x888a88, tuff: 0x6c716a, + sand: 0xdbd29d, red_sand: 0xb96621, sandstone: 0xd5c58b, gravel: 0x8a817d, + clay: 0x9ba4b1, snow: 0xf3fafa, snow_block: 0xf3fafa, ice: 0x91b8ef, + packed_ice: 0x8db0e7, blue_ice: 0x6da3ed, glass: 0xb4d6d9, + netherrack: 0x713232, soul_sand: 0x514035, soul_soil: 0x4b392f, + end_stone: 0xd9dda1, obsidian: 0x30243f, bricks: 0x965f51, + terracotta: 0x9a5e44, oak_leaves: 0x4e7930, birch_leaves: 0x76a64c, + spruce_leaves: 0x3e633c, jungle_leaves: 0x4d822c, cherry_leaves: 0xe5a6bf, +}; + +function block_color(name: string): number { + if (Object.hasOwn(colors, name)) return colors[name]!; + if (name.endsWith('_leaves')) return 0x50843c; + if (name.endsWith('_log') || name.endsWith('_wood')) return 0x715334; + if (name.endsWith('_planks')) return 0xb38b56; + if (name.includes('stone') || name.endsWith('_ore')) return 0x858585; + let hash = 2166136261; + for (let index = 0; index < name.length; index++) hash = Math.imul(hash ^ name.charCodeAt(index), 16777619); + return (64 + ((hash >>> 16) & 127)) << 16 | (64 + ((hash >>> 8) & 127)) << 8 | (64 + (hash & 127)); +} + +function fluid_level(state: number, block: WorldBot['registry']['blocksByStateId'][number]): number { + if (!Number.isSafeInteger(block.minStateId) || state < block.minStateId! || !block.states) return 0; + let offset = state - block.minStateId!; + // Registry state IDs are mixed-radix indices; the last property varies fastest. + for (let index = block.states.length - 1; index >= 0; index--) { + const property = block.states[index]!; + if (!Number.isSafeInteger(property.num_values) || property.num_values < 1) return 0; + const value_index = offset % property.num_values; + offset = Math.floor(offset / property.num_values); + if (property.name === 'level') { + const level = Number(property.values?.[value_index] ?? value_index); + return Number.isInteger(level) && level >= 0 && level <= 15 ? level : 0; + } + } + return 0; +} + +type SpikeSegment = NonNullable['segments'][number]; + +function spike_properties(state: number, block: WorldBot['registry']['blocksByStateId'][number]) { + const result: Pick = {direction: 'up', thickness: 'tip'}; + if (!Number.isSafeInteger(block.minStateId) || state < block.minStateId! || !block.states) return result; + let offset = state - block.minStateId!; + for (let index = block.states.length - 1; index >= 0; index--) { + const property = block.states[index]!; + if (!Number.isSafeInteger(property.num_values) || property.num_values < 1) return {direction: 'up', thickness: 'tip'} as const; + const value = property.values?.[offset % property.num_values]; + offset = Math.floor(offset / property.num_values); + if (property.name === 'vertical_direction' && (value === 'up' || value === 'down')) result.direction = value; + if (property.name === 'thickness' && (value === 'tip' || value === 'tip_merge' + || value === 'frustum' || value === 'middle' || value === 'base')) result.thickness = value; + } + return result; +} + +const feature_properties = new Set(['facing', 'rotation', 'half', 'type', 'shape', 'open', 'attached', 'lit', 'waterlogged', + 'east', 'west', 'north', 'south', 'axis', 'up', 'vertical_direction', 'thickness']); +function surface_properties(state: number, block: WorldBot['registry']['blocksByStateId'][number]): Record { + const result: Record = {}; + if (!Number.isSafeInteger(block.minStateId) || state < block.minStateId! || !block.states) return result; + let offset = state - block.minStateId!; + for (let i = block.states.length - 1; i >= 0; i--) { + const property = block.states[i]!; + if (!Number.isSafeInteger(property.num_values) || property.num_values < 1) return {}; + const index = offset % property.num_values; + offset = Math.floor(offset / property.num_values); + if (!feature_properties.has(property.name)) continue; + const value = property.values?.[index] ?? (property.num_values === 2 ? index === 0 : index); + if (property.name === 'rotation') result.rotation = Number(value); + else if (value === 'true' || value === 'false') result[property.name] = value === 'true'; + else if (typeof value === 'string' || typeof value === 'boolean') result[property.name] = value; + } + return result; +} +function surface_feature(name: string, properties: Record): boolean { + return properties.waterlogged === true || /(?:^|_)(?:wall_)?torch$|_(?:sign|slab|stairs|fence|pane|trapdoor|wall)$/.test(name); +} + +function voxel_properties(state: number, block: WorldBot['registry']['blocksByStateId'][number]): Record { + const result: Record = {}; + if (!block.states?.length) return result; + if (block.states.length > 32 || !Number.isSafeInteger(block.minStateId) || state < block.minStateId!) + throw new ForwardFailure('INTERNAL_ERROR'); + let offset = state - block.minStateId!; + for (let i = block.states.length - 1; i >= 0; i--) { + const property = block.states[i]!; + if (typeof property.name !== 'string' || !property.name.length || property.name.length > 32 + || ['__proto__', 'prototype', 'constructor'].includes(property.name) || Object.hasOwn(result, property.name) + || !Number.isSafeInteger(property.num_values) || property.num_values < 1) + throw new ForwardFailure('INTERNAL_ERROR'); + const index = offset % property.num_values; + offset = Math.floor(offset / property.num_values); + const value = property.type === 'bool' ? index === 0 + : property.type === 'int' ? Number(property.values?.[index] ?? index) + : property.values?.[index] ?? index; + if (typeof value === 'string' ? value.length > 64 + : typeof value === 'number' ? !Number.isFinite(value) : typeof value !== 'boolean') + throw new ForwardFailure('INTERNAL_ERROR'); + result[property.name] = value as string | number | boolean; + } + if (offset) throw new ForwardFailure('INTERNAL_ERROR'); + return result; +} + +function voxel_occluding(name: string, block: WorldBot['registry']['blocksByStateId'][number]): boolean { + // Registry boundingBox describes collision presence, not whether it fills the cube. + return block.boundingBox === 'block' && block.transparent !== true && block.filterLight !== 0 && !air.has(name) + && !/(?:^|_)(?:slab|stairs|fence|gate|wall|pane|door|trapdoor|torch|leaves|glass|water|lava|snow|cake|candle|carpet|bed|button|pressure_plate|sign|banner|head|skull|coral|fan|flower|sapling|plant|roots|rail)$/.test(name) + && !['farmland', 'dirt_path', 'soul_sand', 'enchanting_table', 'brewing_stand', 'cauldron', 'hopper', + 'anvil', 'chipped_anvil', 'damaged_anvil', 'cactus', 'dragon_egg', 'end_portal_frame', 'piston_head', + 'moving_piston', 'chest', 'trapped_chest', 'ender_chest', 'bell', 'stonecutter', 'grindstone', 'lectern', + 'composter', 'daylight_detector', 'sculk_sensor', 'calibrated_sculk_sensor'].includes(name); +} + +export class WorldPreview { + private generation = randomUUID(); + private identity?: {world: WorldBot['world']; version: string; dimension: string; minY: number; height: number}; + private readonly watches = new Map(); + private readonly cache = new Map(); + private readonly voxels = new Map(); + private listeners: Array<[string, (...args: any[]) => void]> = []; + private expiry?: ReturnType; + private invalidation?: ReturnType; + private center?: {x: number; z: number}; + private disposed = false; + private pending = 0; + private pending_voxels = 0; + + constructor(readonly bot: WorldBot, private readonly hint: (generation: string) => void, + private readonly now: () => number = Date.now) {} + + private metadata() { + const {game, world, version, registry, entity} = this.bot; + // Mineflayer exposes WorldSync directly; adapters may expose the parent World's .sync. + const sync = world?.sync ?? world; + if (!sync || typeof sync.getColumn !== 'function' || !registry?.blocksByStateId + || !game || !Number.isSafeInteger(game.minY) || game.minY < -4096 || game.minY > 4096 + || !Number.isSafeInteger(game.height) || game.height < 1 || game.height > 4096 + || typeof game.dimension !== 'string' || !game.dimension.length || game.dimension.length > 256 + || typeof version !== 'string' || !version.length || version.length > 64 + || !entity?.position || ![entity.position.x, entity.position.y, entity.position.z, entity.yaw].every(Number.isFinite)) { + throw new ForwardFailure('INTERNAL_ERROR'); + } + const center = {x: Math.floor(entity.position.x / 16), z: Math.floor(entity.position.z / 16)}; + if (!Number.isSafeInteger(center.x) || !Number.isSafeInteger(center.z) + || Math.abs(center.x) > Number.MAX_SAFE_INTEGER - 3 || Math.abs(center.z) > Number.MAX_SAFE_INTEGER - 3) + throw new ForwardFailure('INTERNAL_ERROR'); + return {world, sync: sync as LoadedWorld, version, dimension: game.dimension, minY: game.minY, height: game.height, center, + position: {x: entity.position.x, y: entity.position.y, z: entity.position.z, yaw: entity.yaw}}; + } + + private refresh() { + const metadata = this.metadata(); + const previous = this.identity; + if (!previous || previous.world !== metadata.world || previous.version !== metadata.version + || previous.dimension !== metadata.dimension || previous.minY !== metadata.minY || previous.height !== metadata.height) { + this.generation = randomUUID(); + this.cache.clear(); + this.voxels.clear(); + this.identity = metadata; + if (previous) this.invalidate(); + } + if (this.center && (this.center.x !== metadata.center.x || this.center.z !== metadata.center.z)) this.invalidate(); + this.center = metadata.center; + return metadata; + } + + private invalidate() { + if (this.disposed || !this.watches.size || this.invalidation) return; + this.invalidation = setTimeout(() => { + this.invalidation = undefined; + this.prune(); + if (!this.disposed && this.watches.size) this.hint(this.generation); + }, 200); + this.invalidation.unref(); + } + + private change(position?: {x: number; z: number}) { + this.prune(); + if (!this.watches.size || this.disposed) return; + const chunk = position && Number.isFinite(position.x) && Number.isFinite(position.z) + ? {x: Math.floor(position.x / 16), z: Math.floor(position.z / 16)} : undefined; + // Invalidate immediately even if position/game metadata is transient during respawn. + if (chunk) this.forget(`${chunk.x},${chunk.z}`); + try { + const metadata = this.refresh(); + if (chunk && [...this.watches.values()].some(watch => Math.abs(chunk.x - metadata.center.x) <= watch.radius + && Math.abs(chunk.z - metadata.center.z) <= watch.radius)) this.invalidate(); + } catch { /* Metadata can be transiently unavailable during respawn; the next snapshot rechecks it. */ } + } + + private attach() { + if (this.listeners.length) return; + const listen = (name: string, callback: (...args: any[]) => void) => { + this.listeners.push([name, callback]); + this.bot.on(name, callback); + }; + listen('chunkColumnLoad', position => this.change(position)); + listen('chunkColumnUnload', position => this.change(position)); + listen('blockUpdate', (old_block, new_block) => this.change(new_block?.position ?? old_block?.position)); + listen('move', () => this.change()); + listen('respawn', () => { + if (!this.watches.size || this.disposed) return; + // Respawning in the same dimension can retain the same world object. + this.identity = undefined; + try { this.refresh(); } catch { + this.generation = randomUUID(); + this.cache.clear(); + this.voxels.clear(); + } + this.invalidate(); + }); + listen('end', () => this.dispose()); + } + + private idle() { + for (const [name, listener] of this.listeners) this.bot.off(name, listener); + this.listeners = []; + clearTimeout(this.expiry); + clearTimeout(this.invalidation); + this.expiry = this.invalidation = undefined; + this.cache.clear(); + this.voxels.clear(); + this.identity = undefined; + this.center = undefined; + } + + private prune() { + for (const [id, watch] of this.watches) if (watch.expires_at <= this.now()) this.watches.delete(id); + if (!this.watches.size) this.idle(); + } + + private arm_expiry() { + clearTimeout(this.expiry); + if (!this.watches.size) return; + const expires = Math.min(...[...this.watches.values()].map(watch => watch.expires_at)); + this.expiry = setTimeout(() => { this.prune(); this.arm_expiry(); }, Math.max(1, expires - this.now())); + this.expiry.unref(); + } + + unwatch(id: string): {released: true} { + this.watches.delete(id); + this.prune(); + this.arm_expiry(); + return {released: true}; + } + + get is_disposed() { return this.disposed; } + + private forget(key: string) { + this.cache.delete(key); + this.voxels.delete(key); + } + + private column_entry(key: string, column: WorldColumn): ColumnCache { + let entry = this.cache.get(key); + if (!entry || entry.column !== column) { + entry = {column, revision: randomUUID()}; + this.forget(key); + this.cache.set(key, entry); + while (this.cache.size > 96) this.forget(this.cache.keys().next().value!); + } + return entry; + } + + async snapshot(input: WorldPayloads['world.snapshot'], check: () => void): Promise> { + check(); + if (this.disposed) throw new ForwardFailure('CANCELLED'); + this.prune(); + if (this.pending >= 4) throw new ForwardFailure('RATE_LIMITED'); + const existing = input.watch_id && this.watches.has(input.watch_id) ? input.watch_id : undefined; + if (!existing && this.watches.size >= 64) throw new ForwardFailure('RATE_LIMITED'); + const watch_id = existing ?? randomUUID(); + const watch: Watch = {radius: input.radius, expires_at: this.now() + 45_000}; + this.watches.set(watch_id, watch); + this.attach(); + this.arm_expiry(); + this.pending++; + try { + const metadata = this.refresh(); + const generation = this.generation; + watch.world_generation = generation; + const valid = () => { + check(); + if (this.disposed || this.watches.get(watch_id) !== watch || watch.expires_at <= this.now()) + throw new ForwardFailure('CANCELLED'); + const current = this.refresh(); + if (generation !== this.generation) throw new ForwardFailure('CANCELLED'); + if (current.center.x !== metadata.center.x || current.center.z !== metadata.center.z) + throw new ForwardFailure('CANCELLED'); + }; + const known = new Map(input.world_generation === generation + ? input.known.map(item => [`${item.x},${item.z}`, item.revision]) : []); + const chunks: WorldChunk[] = [], loaded: WorldSnapshot['loaded'] = []; + const sampled: Array<{x: number; z: number; entry: ColumnCache}> = []; + let partial = false; + for (let z = metadata.center.z - input.radius; z <= metadata.center.z + input.radius; z++) { + for (let x = metadata.center.x - input.radius; x <= metadata.center.x + input.radius; x++) { + valid(); + const key = `${x},${z}`; + const column = metadata.sync.getColumn(x, z); + if (!column) { this.forget(key); partial = true; continue; } + const entry = this.column_entry(key, column); + if (known.get(key) !== entry.revision) { + if (!entry.chunk) entry.chunk = await this.scan(x, z, entry, metadata, () => { + valid(); + if (this.cache.get(key) !== entry || metadata.sync.getColumn(x, z) !== column) + throw new ForwardFailure('CANCELLED'); + }); + valid(); + if (this.cache.get(key) !== entry || metadata.sync.getColumn(x, z) !== column) { + throw new ForwardFailure('CANCELLED'); + } + chunks.push(entry.chunk); + } + sampled.push({x, z, entry}); + loaded.push({x, z, revision: entry.revision}); + if (loaded.length % 2 === 0) { await yield_turn(); valid(); } + } + } + valid(); + // A column may change during a later column's yield. Never label stale data as current. + for (const {x, z, entry} of sampled) { + if (this.cache.get(`${x},${z}`) === entry && metadata.sync.getColumn(x, z) === entry.column) continue; + throw new ForwardFailure('CANCELLED'); + } + return {world_generation: generation, watch_id, version: metadata.version, dimension: metadata.dimension, + min_y: metadata.minY, height: metadata.height, center: metadata.center, position: metadata.position, + chunks: structuredClone(chunks), loaded, partial}; + } catch (error) { + if (!existing && this.watches.get(watch_id) === watch) this.unwatch(watch_id); + throw error; + } finally { this.pending--; } + } + + async chunk(input: WorldPayloads['world.chunk'], check: () => void): Promise { + check(); + if (this.disposed) throw new ForwardFailure('CANCELLED'); + this.prune(); + const watch = this.watches.get(input.watch_id); + if (!watch) throw new ForwardFailure('CANCELLED'); + const metadata = this.refresh(), generation = this.generation; + if (input.world_generation !== generation || watch.world_generation !== generation) + throw new ForwardFailure('CANCELLED'); + if (Math.abs(input.x - metadata.center.x) > watch.radius || Math.abs(input.z - metadata.center.z) > watch.radius) + throw new ForwardFailure('FORBIDDEN'); + if (this.pending_voxels >= 2) throw new ForwardFailure('RATE_LIMITED'); + const key = `${input.x},${input.z}`, column = metadata.sync.getColumn(input.x, input.z); + if (!column) { this.forget(key); throw new ForwardFailure('CANCELLED'); } + const entry = this.column_entry(key, column); + const valid = () => { + check(); + if (this.disposed || this.watches.get(input.watch_id) !== watch || watch.expires_at <= this.now()) + throw new ForwardFailure('CANCELLED'); + const current = this.refresh(); + if (generation !== this.generation || this.cache.get(key) !== entry + || current.center.x !== metadata.center.x || current.center.z !== metadata.center.z + || current.sync.getColumn(input.x, input.z) !== column) throw new ForwardFailure('CANCELLED'); + }; + this.pending_voxels++; + try { + valid(); + const cached = this.voxels.get(key); + if (cached?.entry === entry) { + this.voxels.delete(key); + this.voxels.set(key, cached); + return structuredClone(cached.chunk); + } + const palette: WorldVoxelChunk['palette'] = [], sections: WorldVoxelChunk['sections'] = []; + const indices = new Map(); + const palette_index = (state: number) => { + let index = indices.get(state); + if (index !== undefined) return index; + const block = this.bot.registry.blocksByStateId[state]; + if (!Number.isSafeInteger(state) || state < 0 || !block || typeof block.name !== 'string' + || !block.name.length || block.name.length > 128 + || block.maxStateId !== undefined && state > block.maxStateId || palette.length >= 8192) + throw new ForwardFailure('INTERNAL_ERROR'); + const name = block.name.replace(/^minecraft:/, ''); + index = palette.length; + indices.set(state, index); + palette.push({name: block.name, color: block_color(name), properties: voxel_properties(state, block), + occluding: voxel_occluding(name, block)}); + return index; + }; + const position = {x: 0, y: 0, z: 0}, reader = column.getBlockStateId; + const read = reader.length >= 3 + ? () => (reader as unknown as (x: number, y: number, z: number) => number).call(column, position.x, position.y, position.z) + : () => reader.call(column, position); + let reads = 0; + for (let section = 0; section < Math.ceil(metadata.height / 16); section++) { + const blocks = Buffer.alloc(8192); + let non_air = false; + for (let y = 0; y < 16; y++) { + const offset_y = section * 16 + y; + if (offset_y >= metadata.height) { + // Padding is outside the world and must not duplicate the topmost solid block. + if (non_air) { + const registry = this.bot.registry; + const air_state = registry.blocksByName?.air?.defaultState ?? registry.blocksByName?.air?.minStateId + ?? Object.entries(registry.blocksByStateId).find(([, block]) => block.name.replace(/^minecraft:/, '') === 'air')?.[0]; + if (air_state === undefined) throw new ForwardFailure('INTERNAL_ERROR'); + const index = palette_index(Number(air_state)); + for (let cell = y * 256; cell < 4096; cell++) blocks.writeUInt16LE(index, cell * 2); + } + break; + } + position.y = metadata.minY + offset_y; + for (let z = 0; z < 16; z++) for (let x = 0; x < 16; x++) { + position.x = x; position.z = z; + const index = palette_index(read()); + blocks.writeUInt16LE(index, (y * 256 + z * 16 + x) * 2); + if (!air.has(palette[index]!.name.replace(/^minecraft:/, ''))) non_air = true; + if (++reads % 2048 === 0) { await yield_turn(); valid(); } + } + } + if (non_air) sections.push({y: section, blocks: blocks.toString('base64')}); + } + valid(); + const chunk: WorldVoxelChunk = {x: input.x, z: input.z, revision: entry.revision, world_generation: generation, + min_y: metadata.minY, height: metadata.height, palette, sections}; + this.voxels.delete(key); + this.voxels.set(key, {entry, chunk, bytes: Buffer.byteLength(JSON.stringify(chunk))}); + let bytes = [...this.voxels.values()].reduce((sum, value) => sum + value.bytes, 0); + while (this.voxels.size > 8 || bytes > 32 * 1024 * 1024) { + const oldest = this.voxels.keys().next().value!; + bytes -= this.voxels.get(oldest)!.bytes; + this.voxels.delete(oldest); + } + return structuredClone(chunk); + } finally { this.pending_voxels--; } + } + + private async scan(x: number, z: number, entry: ColumnCache, + metadata: ReturnType, check: () => void): Promise { + const heights = Buffer.alloc(512), blocks = Buffer.alloc(512); + const terrain_heights = Buffer.alloc(512), terrain_blocks = Buffer.alloc(512); + const fluid_heights = Buffer.alloc(512), fluid_blocks = Buffer.alloc(512), fluid_levels = Buffer.alloc(512); + const spike_heights = Buffer.alloc(512), spike_blocks = Buffer.alloc(512); + const segments: SpikeSegment[] = []; + const feature_heights = Buffer.alloc(512), feature_blocks = Buffer.alloc(512); + const features: NonNullable['entries'] = []; + const palette: WorldChunk['palette'] = []; + const indices = new Map(); + const palette_index = (name: string) => { + let index = indices.get(name); + if (index === undefined) { + index = palette.length; + if (index >= 512) throw new ForwardFailure('INTERNAL_ERROR'); + indices.set(name, index); + palette.push({name, color: block_color(name.replace(/^minecraft:/, ''))}); + } + return index; + }; + const position = {x: 0, y: 0, z: 0}; + const reader = entry.column.getBlockStateId; + const read = reader.length >= 3 + ? () => (reader as unknown as (x: number, y: number, z: number) => number).call(entry.column, position.x, position.y, position.z) + : () => reader.call(entry.column, position); + let reads = 0, has_fluid = false; + for (let local_z = 0; local_z < 16; local_z++) { + for (let local_x = 0; local_x < 16; local_x++) { + let name = 'air', height = 0; + let terrain_name = 'air', terrain_height = 0, fluid_name = '', level = 0; + let terrain_state = -1; + let spike_base_name = 'air', spike_base_height = 0; + const column_segments: SpikeSegment[] = []; + const index = local_z * 16 + local_x; + position.x = local_x; position.z = local_z; + for (let y = metadata.minY + metadata.height - 1; y >= metadata.minY; y--) { + position.y = y; + const state = read(); + const block = this.bot.registry.blocksByStateId[state]; + // Missing registry data is unknown, not evidence that this cell is air. + if (!Number.isSafeInteger(state) || !block || typeof block.name !== 'string' + || !block.name.length || block.name.length > 128) throw new ForwardFailure('INTERNAL_ERROR'); + if (++reads % 4096 === 0) { await yield_turn(); check(); } + const technical_name = block.name.replace(/^minecraft:/, ''); + const fluid = technical_name === 'lava' ? 'lava' : water_blocks.has(technical_name) ? 'water' : ''; + if (column_segments.length) { + if (technical_name !== 'pointed_dripstone') { + if (!air.has(technical_name) && !fluid) { + spike_base_name = block.name; + spike_base_height = y - metadata.minY + 1; + } + break; + } + // Commit only complete runs, so a capped column keeps its original surface. + if (column_segments.length >= 16 || segments.length + column_segments.length >= 512) { + column_segments.length = 0; + break; + } + column_segments.push({index, height: y - metadata.minY + 1, ...spike_properties(state, block)}); + continue; + } + if (!height) { + if (air.has(technical_name)) continue; + name = fluid || block.name; + height = y - metadata.minY + 1; + if (fluid) { + fluid_name = fluid; + has_fluid = true; + level = technical_name === 'water' || technical_name === 'lava' ? fluid_level(state, block) : 0; + continue; + } + } else if (fluid) { + if (fluid !== fluid_name) break; + continue; + } + // Air ends a contiguous fluid column without inventing a floor beneath the gap. + if (!air.has(technical_name)) { + terrain_name = block.name; + terrain_height = y - metadata.minY + 1; + terrain_state = state; + } + if (technical_name === 'pointed_dripstone' && segments.length < 512) { + column_segments.push({index, height: y - metadata.minY + 1, ...spike_properties(state, block)}); + continue; + } + break; + } + const offset = index * 2; + heights.writeUInt16LE(height, offset); + blocks.writeUInt16LE(palette_index(name), offset); + terrain_heights.writeUInt16LE(terrain_height, offset); + terrain_blocks.writeUInt16LE(palette_index(terrain_name), offset); + spike_heights.writeUInt16LE(column_segments.length ? spike_base_height : terrain_height, offset); + spike_blocks.writeUInt16LE(palette_index(column_segments.length ? spike_base_name : terrain_name), offset); + segments.push(...column_segments); + // Retain small surface structures separately from the solid heightfield. + let base_height = column_segments.length ? spike_base_height : terrain_height; + let base_name = column_segments.length ? spike_base_name : terrain_name; + const terrain_block = this.bot.registry.blocksByStateId[terrain_state]; + if (!column_segments.length && terrain_height && terrain_block && surface_feature(terrain_name.replace(/^minecraft:/, ''), surface_properties(terrain_state, terrain_block))) { + const entries: typeof features = []; + let remaining = 24, complete = false; + for (let y = metadata.minY + terrain_height - 1; y >= metadata.minY && remaining-- > 0; y--) { + position.y = y; + const state = read(), block = this.bot.registry.blocksByStateId[state]; + if (!Number.isSafeInteger(state) || !block || typeof block.name !== 'string' || !block.name.length || block.name.length > 128) + throw new ForwardFailure('INTERNAL_ERROR'); + if (++reads % 4096 === 0) { await yield_turn(); check(); } + const technical = block.name.replace(/^minecraft:/, ''), properties = surface_properties(state, block); + if (surface_feature(technical, properties)) { + if (entries.length >= 16 || features.length + entries.length >= 512) break; + entries.push({index, height: y - metadata.minY + 1, block: palette_index(block.name), properties}); + } else { + if (!entries.length) break; + base_height = air.has(technical) || water_blocks.has(technical) || technical === 'lava' ? 0 : y - metadata.minY + 1; + base_name = base_height ? block.name : 'air'; + complete = true; + break; + } + if (y === metadata.minY) { base_height = 0; base_name = 'air'; complete = true; } + } + if (complete) features.push(...entries); + else { base_height = terrain_height; base_name = terrain_name; } + } + feature_heights.writeUInt16LE(base_height, offset); + feature_blocks.writeUInt16LE(palette_index(base_name), offset); + if (fluid_name) { + fluid_heights.writeUInt16LE(height, offset); + fluid_blocks.writeUInt16LE(palette_index(fluid_name), offset); + fluid_levels.writeUInt16LE(level, offset); + } + } + if (local_z % 2 === 1) { await yield_turn(); check(); } + } + return {x, z, revision: entry.revision, palette, heights: heights.toString('base64'), blocks: blocks.toString('base64'), + ...(has_fluid ? { + terrain: {heights: terrain_heights.toString('base64'), blocks: terrain_blocks.toString('base64')}, + fluids: {heights: fluid_heights.toString('base64'), blocks: fluid_blocks.toString('base64'), levels: fluid_levels.toString('base64')}, + } : {}), + ...(segments.length ? {spikes: {heights: spike_heights.toString('base64'), blocks: spike_blocks.toString('base64'), segments}} : {}), + ...(features.length ? {features: {heights: feature_heights.toString('base64'), blocks: feature_blocks.toString('base64'), entries: features}} : {})}; + } + + dispose() { + if (this.disposed) return; + this.disposed = true; + this.watches.clear(); + this.idle(); + } +} diff --git a/bot/instance/bot_instance_init.ts b/bot/instance/bot_instance_init.ts index 93c7e01..8f60071 100755 --- a/bot/instance/bot_instance_init.ts +++ b/bot/instance/bot_instance_init.ts @@ -165,9 +165,9 @@ export class BotContext implements bot_instance_context { this.plugin_manager = new PluginManager(this); } - public log(log: log[]) { - this.logger.log(log); - this.emit('log', log); + public log(log: log[]) { + this.logger.log(log); + this.notify('log', log); } public create_instance(): Promise { diff --git a/tests/bot_lifecycle.test.mjs b/tests/bot_lifecycle.test.mjs index 073ee43..a0aac2c 100644 --- a/tests/bot_lifecycle.test.mjs +++ b/tests/bot_lifecycle.test.mjs @@ -678,3 +678,26 @@ test('context disposal reports offline once even when no instance was created', assert.equal(await context.connect(), 467); } }); + +test('throwing log observers cannot interrupt socketClosed cleanup or offline reporting', async () => { + const {context, bots, statuses, managers, functions, timers} = setup(); + await context.connect(); + bots[0].emit('initialLogin'); + const written = []; + context.logger.log = segments => written.push(...segments); + context.on('log', () => { throw new Error('log observer failed'); }); + let lost = 0; + context.on('connection_lost', () => lost++); + assert.doesNotThrow(() => bots[0].emit('end', 'socketClosed')); + await flush(); + assert.equal(context.bot_instance, null); + assert.equal(context.bot_status, 'offline'); + assert.deepEqual(statuses.slice(-2), ['disconnecting', 'offline']); + assert.ok(written.includes('连接已断开: socketClosed')); + assert.ok(written.includes('[NODE] 机器人实例已关闭')); + assert.equal(managers[0].disposed, true); + assert.equal(functions[0].disposed, true); + assert.equal(lost, 1); + assert.equal(timers.size, 0); + await context.clear_this_context(); +}); diff --git a/tests/forward_integration.test.mjs b/tests/forward_integration.test.mjs index b580b62..2901dbf 100644 --- a/tests/forward_integration.test.mjs +++ b/tests/forward_integration.test.mjs @@ -484,7 +484,7 @@ test('execute_function traverses the relay with boolean results, permission chec assert.equal(success_data(await ack(web.socket, 'relay_request', request('execute_function', payload))), false); }); -test('actual manager serves snapshots, logs and chat without forwarding bot events', async t => { +test('actual manager pushes lifecycle changes to snapshot readers without exposing other bot events', async t => { const f = await real_fixture(t); const s = scope(); await f.open(s); const manager = f.manager(s); const cmd = f.cmd(s); assert.deepEqual(await manager.ensure(cmd), success({...s, node_ready: true, lease_expires_at: cmd.expires_at})); @@ -497,21 +497,29 @@ test('actual manager serves snapshots, logs and chat without forwarding bot even assert.deepEqual(Object.keys(absent).sort(), ['bot_id', 'bot_status', 'captured_at', 'context_exists', 'context_generation', 'seq', 'stream_id']); assert.equal(absent.context_exists, false); assert.equal(absent.context_generation, null); assert.equal(absent.bot_status, 'offline'); code(await ack(web.socket, 'relay_request', request('bot.chat', {message: 'hello'})), 'NO_CONTEXT'); - let leaks = 0; other.socket.on('relay_request', () => leaks++); other.socket.on('bot_event', () => leaks++); + let leaks = 0; other.socket.on('relay_request', () => leaks++); + const reader_events = []; other.socket.on('bot_event', event => reader_events.push(event)); assert.equal((await ack(web.socket, 'relay_request', request('bot.subscribe', {events: ['status_update', 'log', 'message', 'bot_error']}))).status, 'success'); const events = []; web.socket.on('bot_event', value => events.push(value)); const bot = online_context(); + const initial_status = received(other.socket, 'bot_event', event => event.payload.status === 'online'); f.source(s.bot_id).set(bot); + await initial_status; const snapshot = success_data(await ack(web.socket, 'relay_request', request())); assert.equal(snapshot.context_exists, true); assert.equal(snapshot.bot_status, 'online'); assert.equal(snapshot.stream_id, absent.stream_id); assert.ok(snapshot.context_generation); for (const state of ['offline', 'ready', 'connecting', 'online', 'disconnecting', 'reconnecting']) { + const pushed = received(web.socket, 'bot_event', event => event.payload.status === state); + const reader_pushed = received(other.socket, 'bot_event', event => event.payload.status === state); bot.bot_status = state; bot.emit('status_update', state); + await Promise.all([pushed, reader_pushed]); const current = success_data(await ack(web.socket, 'relay_request', request())); assert.equal(current.context_exists, true); assert.equal(current.bot_status, state); assert.equal(current.context_generation, snapshot.context_generation); } + const back_online = received(web.socket, 'bot_event', event => event.payload.status === 'online'); bot.bot_status = 'online'; bot.emit('status_update', 'online'); + await back_online; const log_packet = received(web.socket, 'bot_log'); const metadata = {hoverEvent: {action: 'show_text', value: '悬停说明'}, clickEvent: {action: 'suggest_command', value: '/help'}, insertion: 'Alex'}; @@ -555,8 +563,9 @@ test('actual manager serves snapshots, logs and chat without forwarding bot even bot.emit('error', new Error('SECRET stack')); assert.equal(JSON.stringify({snapshot, logs}).includes('SECRET'), false); await ack(other.socket, 'relay_request', request()); assert.equal(leaks, 0); - assert.deepEqual(events, []); - assert.equal(f.packets.some(packet => packet.event === 'bot_event'), false); + assert.ok(events.length >= 7 && reader_events.length >= 7); + assert.ok([...events, ...reader_events].every(event => event.name === 'status_update')); + assert.ok(f.packets.some(packet => packet.event === 'bot_event')); manager.stop(); assert.equal(f.source(s.bot_id).count, 0); assert.equal(bot.listenerCount('log'), 0); }); @@ -626,16 +635,20 @@ test('one Node manager isolates multiple Bot sessions and observes stop/start wi await ack(web_a.socket, 'relay_request', request('bot.chat', {message: 'a'})); await ack(web_b.socket, 'relay_request', request('bot.chat', {message: 'b'})); assert.deepEqual(bot_a.sent, ['a']); assert.deepEqual(bot_b.sent, ['b']); + const stopped = received(web_a.socket, 'bot_event', event => event.payload.status === 'offline'); f.source(a.bot_id).set(undefined); + await stopped; const absent = success_data(await ack(web_a.socket, 'relay_request', request())); assert.equal(absent.bot_status, 'offline'); assert.equal(absent.context_generation, null); const replacement = online_context(); + const started = received(web_a.socket, 'bot_event', event => event.payload.status === 'online'); f.source(a.bot_id).set(replacement); + await started; const after = success_data(await ack(web_a.socket, 'relay_request', request())); assert.equal(after.stream_id, before.stream_id); assert.notEqual(after.context_generation, before.context_generation); assert.equal(web_a.socket.connected, true); assert.equal(f.node_sockets.length, 2); assert.equal(events_b.length, 0); - assert.equal(events_a.length, 0); + assert.deepEqual(events_a.map(event => event.payload.status), ['offline', 'online']); assert.equal(bot_a.listenerCount('log'), 0); }); diff --git a/tests/forward_manager.test.mjs b/tests/forward_manager.test.mjs index 20cfd11..1303600 100644 --- a/tests/forward_manager.test.mjs +++ b/tests/forward_manager.test.mjs @@ -97,6 +97,27 @@ function setup() { ensure(value = initial) { active = value; return manager.ensure(value); }}; } +test('lifecycle updates use reliable bot_event even while volatile logs are saturated', async t => { + const f = setup(); t.after(() => f.manager.stop()); + await f.ensure(); + const socket = f.sockets[0], stream = f.streams[0], received = []; + socket.on('bot_event', event => received.push(event)); + socket.volatile = {emit() {}}; + const base = {bot_id: 2, context_generation: randomUUID(), stream_id: randomUUID(), emitted_at: Date.now()}; + for (let seq = 1; seq <= 160; seq++) stream.event({...base, seq, name: 'log', payload: {records: []}}); + for (const [index, status] of ['online', 'disconnecting', 'offline'].entries()) { + stream.event({...base, seq: 161 + index, name: 'status_update', payload: {status, context_exists: true}}); + } + assert.deepEqual(received.map(event => event.payload.status), ['online', 'disconnecting', 'offline']); + assert.ok(received.every(event => event.forward_session_id === f.command.forward_session_id)); + stream.event({...base, seq: 165, name: 'message', payload: {text: 'not a lifecycle event'}}); + assert.equal(received.length, 3); + socket.io.engine.writeBuffer.length = 64; + stream.event({...base, seq: 166, name: 'status_update', payload: {status: 'offline', context_exists: true}}); + assert.equal(socket.connected, false); + assert.equal(stream.listeners, 0); +}) + const tick = () => new Promise(resolve => setImmediate(resolve)); const code = response => response.data.code; @@ -299,7 +320,7 @@ test('lease expiry retires data channel and queue capacity stays bounded under s assert.equal(stalled.sockets.length, 1); }); -test('bot events are never forwarded while connected, congested or disconnected', async t => { +test('lifecycle events are forwarded while ready and congestion closes the stale transport', async t => { const app = setup(); t.after(() => app.manager.stop()); await app.ensure(); @@ -308,23 +329,28 @@ test('bot events are never forwarded while connected, congested or disconnected' let resyncs = 0; stream.resync = () => { resyncs++; }; const event = {name: 'status_update', seq: 1, payload: {status: 'offline', context_exists: false}}; + let lifecycle = 0; + socket.on('bot_event', () => lifecycle++); stream.event(event); - assert.equal(socket.sent.filter(([name]) => name === 'bot_event').length, 0); + assert.equal(lifecycle, 1); assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 0); stream.event({name: 'log', seq: 2, payload: {records: []}}); assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 1); + socket.emit('relay_status', {forward_session_id: app.command.forward_session_id, node_ready: true, reason: 'resync_required'}); + assert.equal(resyncs, 1); socket.io.engine.writeBuffer.length = 64; stream.event({...event, seq: 2}); stream.event({name: 'log', seq: 3, payload: {records: []}}); assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 1); - assert.equal(socket.sent.filter(([name]) => name === 'bot_event').length, 0); + assert.equal(lifecycle, 1); + assert.equal(socket.connected, false); socket.emit('relay_status', {forward_session_id: app.command.forward_session_id, node_ready: true, reason: 'resync_required'}); assert.equal(resyncs, 1); socket.disconnect(); stream.event({...event, seq: 3}); stream.event({name: 'log', seq: 4, payload: {records: []}}); assert.equal(socket.sent.filter(([name]) => name === 'bot_log').length, 1); - assert.equal(socket.sent.filter(([name]) => name === 'bot_event').length, 0); + assert.equal(lifecycle, 1); assert.equal(stream.listeners, 0); }); diff --git a/tests/world_preview.test.mjs b/tests/world_preview.test.mjs new file mode 100644 index 0000000..34cc4de --- /dev/null +++ b/tests/world_preview.test.mjs @@ -0,0 +1,829 @@ +import assert from 'node:assert/strict'; +import {randomUUID} from 'node:crypto'; +import {EventEmitter} from 'node:events'; +import {createRequire, isBuiltin} from 'node:module'; +import {fileURLToPath} from 'node:url'; +import {setImmediate as tick, setTimeout as delay} from 'node:timers/promises'; +import test from 'node:test'; +import {build} from 'esbuild'; + +const root = fileURLToPath(new URL('../', import.meta.url)); +const compiled = await build({ + stdin: {contents: `export * from './bot/forward/stream.ts'; export * from './bot/forward/manager.ts'; + export * from './utils/forward_protocol.ts';`, resolveDir: root, loader: 'ts'}, + bundle: true, platform: 'node', format: 'cjs', write: false, metafile: true, + plugins: [{name: 'isolated-node', setup(build) { + build.onResolve({filter: /(?:instance\/index\.ts|^socket\.io-client$|console_print\.ts|path_utils\.ts)$/}, args => + ({path: args.path.includes('instance/') ? 'registry' : args.path === 'socket.io-client' ? 'socket' + : args.path.endsWith('path_utils.ts') ? 'path' : 'console', namespace: 'stub'})); + build.onLoad({filter: /.*/, namespace: 'stub'}, ({path}) => ({loader: 'js', contents: { + registry: 'export const get_bot_context = () => undefined; export const observe_bot_context = () => () => {};', + socket: 'export const io = () => {throw Error("Unexpected real socket")};', + console: 'export const console_print = {success: x => x, log() {}};', + path: 'export const get_project_root_path = () => {throw Error("Unexpected filesystem access")};', + }[path]})); + }}], +}); +assert.ok(Object.keys(compiled.metafile.inputs).every(path => !path.includes('forward_server'))); +const require = createRequire(import.meta.url); +const module = {exports: {}}; +new Function('require', 'module', 'exports', compiled.outputFiles[0].text)(specifier => { + assert.ok(isBuiltin(specifier), `Unexpected external import: ${specifier}`); + return require(specifier); +}, module, module.exports); +const {BotStream, NodeForwardManager, validate_rpc} = module.exports; +const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +const code = expected => error => error.code === expected; +const request = (method, payload, forward_session_id = randomUUID(), deadline_at = Date.now() + 2000) => ({ + request_id: randomUUID(), correlation_id: randomUUID(), forward_session_id, deadline_at, method, payload, +}); + +function fixture(t, {minY = -4, height = 16, direct = false} = {}) { + let now = 100; + let reads = 0; + const columns = new Map(); + const bot = Object.assign(new EventEmitter(), { + chat() {}, version: '1.21.4', game: {minY, height, dimension: 'overworld'}, + registry: {blocksByStateId: {7: {name: 'air'}, 8: {name: 'cave_air'}, 9: {name: 'void_air'}, + 101: {name: 'grass_block'}, 202: {name: 'water'}, 303: {name: 'stone'}, 404: {name: 'oak_leaves'}, + 505: {name: 'modded_marble'}}}, + entity: {position: {x: -0.25, y: 3.5, z: -16.25}, yaw: Math.PI / 3}, + }); + const sync = {getColumn(x, z) { return columns.get(`${x},${z}`); }}; + bot.world = direct ? sync : {sync}; + const context = Object.assign(new EventEmitter(), {bot_status: 'online', bot_instance: bot}); + let current = context; + const observers = new Set(); + const source = {get: () => current, observe(_id, callback) { observers.add(callback); return () => observers.delete(callback); }}; + const stream = new BotStream(1, {source, now: () => now}); + t.after(() => stream.dispose()); + const column = (surface = () => 7) => ({getBlockStateId(position) { + assert.equal(typeof position, 'object'); + assert.ok(position.x >= 0 && position.x < 16 && position.z >= 0 && position.z < 16); + assert.ok(position.y >= minY && position.y < minY + height); + reads++; + return surface(position.x, position.y, position.z); + }}); + return {bot, context, stream, columns, column, observers, + get reads() { return reads; }, + snapshot(input = {}, check = () => {}) { return stream.world_snapshot({radius: 1, known: [], ...input}, now + 2000, check); }, + time(value) { now = value; }, + set(value) { current = value; for (const observer of observers) observer(value); }, + }; +} + +function decoded(chunk, layer = chunk) { + for (const name of ['heights', 'blocks', ...('levels' in layer ? ['levels'] : [])]) { + assert.equal(layer[name].length, 684); + assert.equal(Buffer.from(layer[name], 'base64').length, 512); + assert.equal(Buffer.from(layer[name], 'base64').toString('base64'), layer[name]); + } + const heights = Buffer.from(layer.heights, 'base64'), blocks = Buffer.from(layer.blocks, 'base64'); + const levels = layer.levels && Buffer.from(layer.levels, 'base64'); + const result = []; + for (let index = 0; index < 256; index++) { + const palette_index = blocks.readUInt16LE(index * 2); + assert.ok(palette_index < chunk.palette.length); + const cell = {height: heights.readUInt16LE(index * 2), ...chunk.palette[palette_index]}; + if (levels) { + cell.level = levels.readUInt16LE(index * 2); + assert.ok(cell.level >= 0 && cell.level <= 15); + if (cell.height) assert.ok(['water', 'lava'].includes(cell.name)); + else assert.equal(cell.level, 0); + } + result.push(cell); + } + assert.ok(chunk.palette.length >= 1 && chunk.palette.length <= 512); + for (const entry of chunk.palette) { + assert.deepEqual(Object.keys(entry).sort(), ['color', 'name']); + assert.ok(entry.name.length <= 128); + assert.ok(Number.isInteger(entry.color) && entry.color >= 0 && entry.color <= 0xffffff); + } + return result; +} + +test('surface feature sampling retains torch, sign, hanging and waterlogged states above their foundation', async t => { + const h = fixture(t), mf = createRequire(require.resolve('@wp2508/mineflayer')); + const data = mf('minecraft-data')('1.21.4'); + const id = (name, properties) => { + const b = data.blocksByName[name]; let offset = 0; + for (const p of b.states) offset = offset * p.num_values + (p.type === 'bool' ? properties[p.name] === true ? 0 : 1 : p.values.indexOf(String(properties[p.name]))); + const value = b.minStateId + offset; h.bot.registry.blocksByStateId[value] = b; return value; + }; + const states = [id('wall_torch', {facing: 'east'}), id('oak_sign', {rotation: 7, waterlogged: false}), + id('oak_hanging_sign', {rotation: 3, attached: true, waterlogged: true}), + id('oak_wall_hanging_sign', {facing: 'north', waterlogged: false}), id('oak_slab', {type: 'bottom', waterlogged: true}), + id('oak_stairs', {facing: 'south', half: 'top', shape: 'inner_left', waterlogged: true})]; + h.columns.set('-1,-2', h.column((x,y,z) => z === 0 && x < states.length ? y === 3 ? states[x] : y === 2 ? 303 : 7 : y === 2 ? 303 : 7)); + const result = await h.snapshot(), chunk = result.chunks[0]; + assert.equal(chunk.features.entries.length, states.length); + const entries = chunk.features.entries; + assert.deepEqual(entries[0].properties, {facing: 'east'}); + assert.equal(entries[1].properties.rotation, 7); assert.equal(entries[1].properties.waterlogged, false); + assert.equal(entries[2].properties.attached, true); assert.equal(entries[2].properties.waterlogged, true); + assert.equal(entries[4].properties.type, 'bottom'); assert.equal(entries[5].properties.shape, 'inner_left'); + assert.ok(entries.every(e => e.height === 8)); + assert.ok(decoded(chunk, chunk.features).slice(0, states.length).every(c => c.height === 7 && c.name === 'stone')); +}); + +test('feature sampling retains short runs and falls back without holes when a column exceeds its bound', async t => { + const h = fixture(t, {minY: 0, height: 24}); + h.bot.registry.blocksByStateId[700] = {name: 'oak_sign', minStateId: 700, states: [{name: 'rotation', num_values: 16, values: Array.from({length: 16}, (_, i) => String(i))}]}; + h.columns.set('-1,-2', h.column((x,y,z) => { + if (z !== 0 || x > 1) return y === 1 ? 303 : 7; + if (x === 0) return y === 3 || y === 2 ? 700 : y === 1 ? 303 : 7; + return y >= 2 && y <= 18 ? 700 : y === 1 ? 303 : 7; + })); + const chunk = (await h.snapshot()).chunks[0]; + assert.equal(chunk.features.entries.length, 2); + assert.ok(chunk.features.entries.every(e => e.index === 0)); + const bases = decoded(chunk, chunk.features); + assert.equal(bases[0].height, 2); + assert.equal(bases[1].height, 19); + assert.equal(bases[1].name, 'oak_sign'); +}); + +test('world protocol validates exact bounded requests, revisions and descriptors', () => { + const valid = {radius: 3, known: [{x: -1, z: -2, revision: randomUUID()}], world_generation: randomUUID(), watch_id: randomUUID()}; + assert.equal(validate_rpc(request('world.snapshot', valid)).payload, valid); + assert.equal(validate_rpc(request('world.unwatch', {watch_id: valid.watch_id})).method, 'world.unwatch'); + for (const payload of [null, {}, [], {radius: 1}, {known: []}, {...valid, extra: true}, + ...[0, 4, -1, 1.5, NaN, Infinity, '1', null].map(radius => ({...valid, radius})), + {...valid, known: Array.from({length: 50}, (_, x) => ({x, z: 0, revision: randomUUID()}))}, + {...valid, known: Array(1)}, {...valid, known: [{x: 0, z: 0}]}, + {...valid, known: [{x: 1.2, z: 0, revision: randomUUID()}]}, + {...valid, known: [{x: 0, z: Infinity, revision: randomUUID()}]}, + {...valid, known: [{x: 0, z: 0, revision: 'bad'}]}, + {...valid, known: [{x: 0, z: 0, revision: randomUUID(), extra: true}]}, + {...valid, known: [valid.known[0], valid.known[0]]}, + {...valid, world_generation: undefined}, {...valid, watch_id: 'bad'}]) + assert.throws(() => validate_rpc(request('world.snapshot', payload)), code('INVALID_REQUEST')); + for (const payload of [{}, {watch_id: 'bad'}, {watch_id: valid.watch_id, extra: 1}, {watch_id: null}]) + assert.throws(() => validate_rpc(request('world.unwatch', payload)), code('INVALID_REQUEST')); + for (const name of ['radius', 'known', 'watch_id', 'world_generation']) { + let reads = 0; + const payload = Object.defineProperty({...valid}, name, {enumerable: true, get() { reads++; return valid[name]; }}); + assert.throws(() => validate_rpc(request('world.snapshot', payload)), code('INVALID_REQUEST')); + assert.equal(reads, 0); + } + let reads = 0; + const known = []; + Object.defineProperty(known, '0', {enumerable: true, get() { reads++; return valid.known[0]; }}); + assert.throws(() => validate_rpc(request('world.snapshot', {...valid, known})), code('INVALID_REQUEST')); + assert.equal(reads, 0); +}); + +test('snapshot encodes negative chunk coordinates, z-major heights and runtime registry surface names', async t => { + const app = fixture(t); + app.columns.set('-1,-2', app.column((x, y, z) => { + if (x === 1 && z === 0) return y === -1 ? 202 : 7; + if (x === 2 && z === 0) return y === 10 ? 404 : 7; + if (x === 3 && z === 0) return y === -4 ? 303 : 7; + if (x === 4 && z === 0) return y === 2 ? 505 : 7; + if (x === 0 && z === 1) return y > 4 ? (y % 2 ? 8 : 9) : y === 4 ? 101 : 7; + return 7; + })); + const result = await app.snapshot(); + assert.deepEqual(Object.keys(result).sort(), ['world_generation', 'context_generation', 'stream_id', 'watch_id', 'version', + 'dimension', 'min_y', 'height', 'center', 'position', 'chunks', 'loaded', 'partial'].sort()); + for (const key of ['world_generation', 'context_generation', 'stream_id', 'watch_id']) assert.match(result[key], uuid); + assert.deepEqual(result.center, {x: -1, z: -2}); + assert.deepEqual(result.position, {x: -0.25, y: 3.5, z: -16.25, yaw: Math.PI / 3}); + assert.equal(result.min_y, -4); assert.equal(result.height, 16); + assert.equal(result.version, app.bot.version); assert.equal(result.dimension, 'overworld'); + assert.equal(result.partial, true); assert.equal(result.chunks.length, 1); assert.equal(result.loaded.length, 1); + const chunk = result.chunks[0]; + assert.deepEqual(result.loaded, [{x: -1, z: -2, revision: chunk.revision}]); + const cells = decoded(chunk); + assert.deepEqual(cells[0], {height: 0, name: 'air', color: 0}); + assert.deepEqual(cells[1], {height: 4, name: 'water', color: 0x3f76e4}); + assert.equal(cells[2].height, 15); assert.equal(cells[2].name, 'oak_leaves'); + assert.equal(cells[3].height, 1); assert.equal(cells[3].name, 'stone'); + assert.equal(cells[4].height, 7); assert.equal(cells[4].name, 'modded_marble'); + assert.equal(cells[16].height, 9); assert.equal(cells[16].name, 'grass_block'); + assert.notEqual(cells[1].color, cells[16].color); +}); + +test('only loaded columns are returned, zero-chunk metadata remains valid and radius 3 is bounded at 49', async t => { + const app = fixture(t, {direct: true}); + const empty = await app.snapshot(); + assert.deepEqual(empty.chunks, []); assert.deepEqual(empty.loaded, []); assert.equal(empty.partial, true); + for (let z = -5; z <= 1; z++) for (let x = -4; x <= 2; x++) app.columns.set(`${x},${z}`, app.column(() => 303)); + const full = await app.snapshot({radius: 3, watch_id: empty.watch_id}); + assert.equal(full.chunks.length, 49); assert.equal(full.loaded.length, 49); assert.equal(full.partial, false); + const cached = await app.snapshot({radius: 3, watch_id: full.watch_id, world_generation: full.world_generation, known: full.loaded}); + assert.equal(cached.chunks.length, 0); assert.equal(cached.loaded.length, 49); assert.equal(cached.partial, false); +}); + +test('matching generation omits unchanged chunks without reads, block events change revisions before hint', async t => { + const app = fixture(t); + let state = 101; + app.columns.set('-1,-2', app.column((_x, y) => y === 0 ? state : 7)); + app.columns.set('0,-2', app.column(() => 303)); + const first = await app.snapshot(); + const reads = app.reads; + const input = {known: first.loaded, world_generation: first.world_generation, watch_id: first.watch_id}; + const same = await app.snapshot(input); + assert.equal(same.chunks.length, 0); assert.equal(same.watch_id, first.watch_id); assert.equal(app.reads, reads); + state = 202; + app.bot.emit('blockUpdate', null, {position: {x: -1, y: 0, z: -17}}); + const changed = await app.snapshot(input); + assert.equal(changed.world_generation, first.world_generation); + assert.equal(changed.chunks.length, 1); + assert.equal(decoded(changed.chunks[0])[0].name, 'water'); + assert.notEqual(changed.chunks[0].revision, first.chunks[0].revision); + const forced = await app.snapshot({...input, world_generation: randomUUID()}); + assert.equal(forced.chunks.length, 2); + first.chunks[0].palette[0].name = 'mutated'; + const clone = await app.snapshot({watch_id: first.watch_id}); + assert.notEqual(clone.chunks[0].palette[0].name, 'mutated'); +}); + +test('column replacement and unload cannot reuse stale cached surfaces', async t => { + const app = fixture(t); + app.columns.set('-1,-2', app.column(() => 101)); + const first = await app.snapshot(); + const input = {known: first.loaded, world_generation: first.world_generation, watch_id: first.watch_id}; + app.columns.set('-1,-2', app.column(() => 303)); + const replacement = await app.snapshot(input); + assert.notEqual(replacement.chunks[0].revision, first.loaded[0].revision); + assert.equal(decoded(replacement.chunks[0])[0].name, 'stone'); + app.columns.delete('-1,-2'); + app.bot.emit('chunkColumnUnload', {x: -16, z: -32}); + const unloaded = await app.snapshot(input); + assert.deepEqual(unloaded.loaded, []); assert.deepEqual(unloaded.chunks, []); + app.columns.set('-1,-2', app.column(() => 202)); + app.bot.emit('chunkColumnLoad', {x: -16, z: -32}); + const reloaded = await app.snapshot(input); + assert.notEqual(reloaded.loaded[0].revision, first.loaded[0].revision); +}); + +test('world object, dimension, version and explicit same-world respawn reset the generation', async t => { + const app = fixture(t); + app.columns.set('-1,-2', app.column(() => 303)); + let last = await app.snapshot(); + for (const reset of [() => { app.bot.world = {sync: app.bot.world.sync}; }, + () => { app.bot.game.dimension = 'the_nether'; }, () => { app.bot.version = '1.20.4'; }, + () => { app.bot.emit('respawn'); }]) { + reset(); + const next = await app.snapshot({watch_id: last.watch_id, world_generation: last.world_generation, known: last.loaded}); + assert.notEqual(next.world_generation, last.world_generation); + assert.equal(next.chunks.length, 1); + assert.equal(next.watch_id, last.watch_id); + last = next; + } +}); + +test('watch events coalesce at 200ms, filter by active radius and only invalidate on chunk movement', async t => { + const app = fixture(t); + const events = []; + app.stream.subscribe(event => { if (event.name === 'info_update') events.push(event); }); + assert.equal(app.bot.listenerCount('chunkColumnLoad'), 0); + const first = await app.snapshot(); + assert.equal(app.bot.listenerCount('chunkColumnLoad'), 1); + for (let index = 0; index < 20; index++) app.bot.emit('blockUpdate', null, {position: {x: -1, z: -17}}); + app.bot.emit('chunkColumnLoad', {x: 0, z: -32}); + assert.equal(events.length, 0); + await delay(230); + assert.equal(events.length, 1); + assert.deepEqual(events[0].payload, {name: 'world_preview', value: {world_generation: first.world_generation}}); + app.bot.emit('chunkColumnLoad', {x: 1600, z: 1600}); + app.bot.entity.position.x = -0.5; app.bot.emit('move'); + await delay(230); + assert.equal(events.length, 1); + app.bot.entity.position.x = 0.1; app.bot.emit('move'); + await delay(230); + assert.equal(events.length, 2); + assert.deepEqual(app.stream.world_unwatch({watch_id: first.watch_id}, 2000), {released: true}); + assert.deepEqual(app.stream.world_unwatch({watch_id: first.watch_id}, 2000), {released: true}); + for (const name of ['chunkColumnLoad', 'chunkColumnUnload', 'blockUpdate', 'move', 'respawn', 'end']) + assert.equal(app.bot.listenerCount(name), 0); +}); + +test('leases cap at 64, renew for 45 seconds and final unwatch or TTL removes caches, timers and listeners', async t => { + const app = fixture(t); + app.columns.set('-1,-2', app.column(() => 303)); + const first = await app.snapshot(); + const ids = [first.watch_id]; + for (let index = 1; index < 64; index++) ids.push((await app.snapshot()).watch_id); + await assert.rejects(app.snapshot(), code('RATE_LIMITED')); + app.time(44_000); + const renewed = await app.snapshot({watch_id: first.watch_id, known: first.loaded, world_generation: first.world_generation}); + assert.equal(renewed.watch_id, first.watch_id); + assert.equal(renewed.chunks.length, 0); + app.time(45_101); + for (const id of ids.slice(1)) app.stream.world_unwatch({watch_id: id}, 90_000); + assert.equal(app.bot.listenerCount('blockUpdate'), 1); + app.time(89_001); + app.bot.emit('move'); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + const reads = app.reads; + const fresh = await app.snapshot({watch_id: first.watch_id, known: first.loaded, world_generation: first.world_generation}); + assert.notEqual(fresh.watch_id, first.watch_id); assert.notEqual(fresh.world_generation, first.world_generation); + assert.ok(app.reads > reads); + app.bot.emit('chunkColumnLoad', {x: -16, z: -32}); + app.stream.world_unwatch({watch_id: fresh.watch_id}, 100_000); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); +}); + +test('lease expiry timer detaches without further bot traffic and cancels a queued invalidation', async t => { + t.mock.timers.enable({apis: ['setTimeout', 'Date'], now: 1000}); + const app = fixture(t); + app.time(1000); + const events = []; + app.stream.subscribe(event => { if (event.name === 'info_update') events.push(event); }); + const first = await app.snapshot(); + app.time(45_900); t.mock.timers.tick(44_900); + app.bot.emit('chunkColumnLoad', {x: -16, z: -32}); + app.time(46_000); t.mock.timers.tick(100); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + t.mock.timers.tick(1000); + assert.equal(events.length, 0); + assert.deepEqual(app.stream.world_unwatch({watch_id: first.watch_id}, 47_000), {released: true}); +}); + +test('watch radii use the union of active leases and shrink after releasing the wider watch', async t => { + t.mock.timers.enable({apis: ['setTimeout']}); + const app = fixture(t); + const events = []; + app.stream.subscribe(event => { if (event.name === 'info_update') events.push(event); }); + const narrow = await app.snapshot(); + const wide = await app.snapshot({radius: 3}); + app.bot.emit('chunkColumnLoad', {x: 32, z: -32}); + t.mock.timers.tick(200); + assert.equal(events.length, 1); + app.stream.world_unwatch({watch_id: wide.watch_id}, 2000); + app.bot.emit('chunkColumnLoad', {x: 32, z: -32}); + t.mock.timers.tick(200); + assert.equal(events.length, 1); + app.bot.emit('respawn'); + t.mock.timers.tick(200); + assert.equal(events.length, 2); + const reset = await app.snapshot({watch_id: narrow.watch_id}); + assert.equal(reset.world_generation, events[1].payload.value.world_generation); +}); + +test('caller mutation after dispatch cannot change radius or known revisions, and scans are bounded to four concurrent calls', async t => { + const app = fixture(t); + app.columns.set('-1,-2', app.column()); + const input = {radius: 1, known: []}; + const pending = app.stream.world_snapshot(input, 2100); + input.radius = 3; input.known.push({x: 0, z: 0, revision: 'bad'}); + const result = await pending; + assert.equal(result.chunks.length, 1); + app.bot.emit('blockUpdate', null, {position: {x: -1, z: -17}}); + const calls = Array.from({length: 4}, () => app.snapshot()); + await assert.rejects(app.snapshot(), code('RATE_LIMITED')); + await Promise.all(calls); +}); + +test('scalar column adapters and deterministic fallback colors remain compatible with the surface contract', async t => { + const app = fixture(t); + app.columns.set('-1,-2', {getBlockStateId(x, y, z) { + assert.equal(typeof x, 'number'); assert.equal(typeof z, 'number'); + return y === 3 ? 505 : 7; + }}); + const first = await app.snapshot(); + const cells = decoded(first.chunks[0]); + assert.equal(cells[0].name, 'modded_marble'); assert.equal(cells[0].height, 8); + assert.equal(Object.hasOwn(first.chunks[0], 'terrain'), false); + assert.equal(Object.hasOwn(first.chunks[0], 'fluids'), false); + assert.equal(Object.hasOwn(first.chunks[0], 'spikes'), false); + app.bot.emit('respawn'); + const reset = await app.snapshot({watch_id: first.watch_id}); + assert.equal(decoded(reset.chunks[0])[0].color, cells[0].color); +}); + +test('height 4096 and a full 256-entry surface palette fit UInt16, and transient metadata cannot retain stale revisions', async t => { + const app = fixture(t, {minY: -4096, height: 4096}); + for (let index = 0; index < 256; index++) app.bot.registry.blocksByStateId[1000 + index] = {name: `surface_${index}`}; + app.columns.set('-1,-2', app.column((x, _y, z) => 1000 + z * 16 + x)); + const first = await app.snapshot(); + const cells = decoded(first.chunks[0]); + assert.equal(first.chunks[0].palette.length, 256); + assert.equal(cells[255].name, 'surface_255'); assert.equal(cells[255].height, 4096); + app.bot.entity.yaw = NaN; + app.bot.emit('blockUpdate', null, {position: {x: -1, z: -17}}); + app.bot.entity.yaw = 0; + const next = await app.snapshot({known: first.loaded, world_generation: first.world_generation, watch_id: first.watch_id}); + assert.equal(next.chunks.length, 1); assert.notEqual(next.loaded[0].revision, first.loaded[0].revision); +}); + +test('surface reads yield and abort promptly on deadline or transport disconnection', async t => { + for (const cause of ['deadline', 'transport']) await t.test(cause, async t => { + const app = fixture(t, {minY: -64, height: 384}); + app.columns.set('-1,-2', app.column()); + let stale = false; + setImmediate(() => { if (cause === 'deadline') app.time(5000); else stale = true; }); + await assert.rejects(app.snapshot({}, () => { + if (stale) throw Object.assign(new Error(), {code: 'SESSION_STALE'}); + }), code(cause === 'deadline' ? 'TIMEOUT' : 'SESSION_STALE')); + assert.ok(app.reads <= 4096, `scan must stop during first read batch: ${app.reads}`); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + }); +}); + +test('in-flight reads reject context replacement, offline transition, disposal and column mutation', async t => { + for (const cause of ['context', 'instance', 'offline', 'dispose', 'block', 'replacement', 'unload', 'respawn', 'world', 'move']) + await t.test(cause, async t => { + const app = fixture(t); + app.columns.set('-1,-2', app.column()); + setImmediate(() => { + if (cause === 'context') app.set(Object.assign(new EventEmitter(), {bot_status: 'online', bot_instance: app.bot})); + if (cause === 'instance') app.context.bot_instance = {...app.bot}; + if (cause === 'offline') { app.context.bot_status = 'offline'; app.context.emit('status_update'); } + if (cause === 'dispose') app.stream.dispose(); + if (cause === 'block') app.bot.emit('blockUpdate', null, {position: {x: -1, z: -17}}); + if (cause === 'replacement') app.columns.set('-1,-2', app.column(() => 303)); + if (cause === 'unload') app.columns.delete('-1,-2'); + if (cause === 'respawn') app.bot.emit('respawn'); + if (cause === 'world') app.bot.world = {sync: app.bot.world.sync}; + if (cause === 'move') { app.bot.entity.position.x = 20; app.bot.emit('move'); } + }); + await assert.rejects(app.snapshot(), error => ['CANCELLED', 'BOT_NOT_ONLINE'].includes(error.code)); + app.stream.dispose(); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + }); +}); + +test('offline, missing context, invalid metadata and unknown registry data never fabricate air', async t => { + const app = fixture(t); + await assert.rejects(app.stream.world_snapshot({radius: 0, known: []}, 200), code('INVALID_REQUEST')); + await assert.rejects(app.stream.world_snapshot({radius: 1, known: []}, 100), code('TIMEOUT')); + app.context.bot_status = 'offline'; + await assert.rejects(app.snapshot(), code('BOT_NOT_ONLINE')); + assert.deepEqual(app.stream.world_unwatch({watch_id: randomUUID()}, 200), {released: true}); + app.context.bot_status = 'online'; + app.bot.entity.yaw = NaN; + await assert.rejects(app.snapshot(), code('INTERNAL_ERROR')); + app.bot.entity.yaw = 0; + app.columns.set('-1,-2', app.column(() => 999_999)); + await assert.rejects(app.snapshot(), code('INTERNAL_ERROR')); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + app.set(undefined); + await assert.rejects(app.snapshot(), code('NO_CONTEXT')); + assert.deepEqual(app.stream.world_unwatch({watch_id: randomUUID()}, 200), {released: true}); + app.stream.dispose(); app.stream.dispose(); + await assert.rejects(app.snapshot(), code('CANCELLED')); + assert.equal(app.observers.size, 0); +}); + +test('fluid columns retain top surfaces and sample contiguous water plants down to actual terrain', async t => { + const app = fixture(t); + const plants = ['bubble_column', 'kelp', 'kelp_plant', 'seagrass', 'tall_seagrass']; + plants.forEach((name, index) => { app.bot.registry.blocksByStateId[600 + index] = {name: `minecraft:${name}`}; }); + app.bot.registry.blocksByStateId[700] = {name: 'minecraft:lava'}; + app.columns.set('-1,-2', app.column((x, y, z) => { + if (z) return 7; + if (x === 0) return y >= -2 && y <= 2 ? 202 : y === -3 ? 303 : 7; + if (x >= 1 && x <= 5) return y === 4 ? 599 + x : y === 3 ? 602 : y === 2 ? 202 : y === 1 ? 101 : 7; + if (x === 6) return y === 6 || y === 5 ? 700 : y === 4 ? 505 : 7; + if (x === 7) return y > 8 ? 7 : y === 8 ? 404 : 999_999; // Solid tops must never inspect lower cells. + if (x === 8) return y === -4 ? 202 : 7; + if (x === 9) return y === 4 ? 202 : y === -4 ? 303 : 7; + if (x === 10 || x === 11) return y > 4 ? 7 : y === 4 ? (x === 10 ? 202 : 700) + : y === 3 ? (x === 10 ? 700 : 202) : 999_999; + return 7; + })); + const {chunks: [chunk]} = await app.snapshot(); + const top = decoded(chunk), terrain = decoded(chunk, chunk.terrain), fluids = decoded(chunk, chunk.fluids); + assert.deepEqual(top[0], {height: 7, name: 'water', color: 0x3f76e4}); + assert.deepEqual(terrain[0], {height: 2, name: 'stone', color: 0x858585}); + assert.deepEqual(fluids[0], {...top[0], level: 0}); // Missing metadata defaults to source. + for (let x = 1; x <= 5; x++) { + assert.deepEqual(top[x], {height: 9, name: 'water', color: 0x3f76e4}); + assert.deepEqual(fluids[x], {...top[x], level: 0}); + assert.equal(terrain[x].name, 'grass_block'); assert.equal(terrain[x].height, 6); + } + assert.equal(top[6].name, 'lava'); assert.equal(fluids[6].name, 'lava'); assert.equal(fluids[6].level, 0); + assert.equal(fluids[6].height, 11); assert.equal(terrain[6].name, 'modded_marble'); assert.equal(terrain[6].height, 9); + assert.deepEqual(terrain[7], top[7]); assert.equal(fluids[7].height, 0); + assert.equal(fluids[8].height, 1); assert.equal(terrain[8].height, 0); + assert.equal(fluids[9].height, 9); assert.equal(terrain[9].height, 0); // No floor across an air gap. + for (const x of [10, 11]) { + assert.equal(top[x].name, x === 10 ? 'water' : 'lava'); + assert.deepEqual(fluids[x], {...top[x], level: 0}); + assert.equal(terrain[x].height, 0); // A different fluid ends the column without becoming solid terrain. + } + assert.equal(top[12].height, 0); assert.equal(terrain[12].height, 0); assert.equal(fluids[12].height, 0); + assert.ok(!chunk.palette.some(entry => plants.some(name => entry.name.endsWith(name)))); +}); + +test('fluid level decoding respects registry property order and explicit level value order', async t => { + const app = fixture(t, {minY: -10, height: 4}); + const states = [{name: 'before', num_values: 2}, + {name: 'level', num_values: 4, values: ['8', '0', '15', '3']}, {name: 'after', num_values: 3}]; + const expected = [8, 0, 15, 3]; + expected.forEach((_level, index) => { + const state = 1000 + 12 + index * 3 + 2; + app.bot.registry.blocksByStateId[state] = {name: 'water', minStateId: 1000, states}; + }); + app.columns.set('-1,-2', app.column((x, y) => y === -7 && x < 4 ? 1000 + 12 + x * 3 + 2 : 303)); + const {chunks: [chunk]} = await app.snapshot(); + assert.deepEqual(decoded(chunk, chunk.fluids).slice(0, 4).map(cell => cell.level), expected); + assert.deepEqual(decoded(chunk, chunk.terrain).slice(0, 4).map(cell => cell.height), [3, 3, 3, 3]); +}); + +test('one shared palette accommodates 256 distinct floors plus water and lava', async t => { + const app = fixture(t, {minY: -64, height: 2}); + app.bot.registry.blocksByStateId[600] = {name: 'lava'}; + for (let index = 0; index < 256; index++) app.bot.registry.blocksByStateId[1000 + index] = {name: `floor_${index}`}; + app.columns.set('-1,-2', app.column((x, y, z) => y === -63 ? (x % 2 ? 202 : 600) : 1000 + z * 16 + x)); + const {chunks: [chunk]} = await app.snapshot(); + const terrain = decoded(chunk, chunk.terrain), fluids = decoded(chunk, chunk.fluids); + assert.equal(chunk.palette.length, 258); + for (let index = 0; index < 256; index++) { + assert.equal(terrain[index].name, `floor_${index}`); assert.equal(terrain[index].height, 1); + assert.equal(fluids[index].name, index % 2 ? 'water' : 'lava'); assert.equal(fluids[index].height, 2); + } +}); + +test('fluid terrain scans reject unknown blocks and yield before cancellation without caching partial data', async t => { + await t.test('unknown beneath fluid', async t => { + const app = fixture(t); + let unknown = true; + app.columns.set('-1,-2', app.column((_x, y) => y === 11 ? 202 : unknown ? 999_999 : 303)); + await assert.rejects(app.snapshot(), code('INTERNAL_ERROR')); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + unknown = false; + const {chunks: [chunk]} = await app.snapshot(); + assert.equal(decoded(chunk, chunk.terrain)[0].height, 15); + }); + await t.test('deep fluid cancellation', async t => { + const app = fixture(t, {minY: -64, height: 4096}); + app.columns.set('-1,-2', app.column(() => 202)); + let stale = false; + setImmediate(() => { stale = true; }); + await assert.rejects(app.snapshot({}, () => { + if (stale) throw Object.assign(new Error(), {code: 'SESSION_STALE'}); + }), code('SESSION_STALE')); + assert.equal(app.reads, 4096); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + app.columns.set('-1,-2', app.column(() => 303)); + const {chunks: [chunk]} = await app.snapshot(); + assert.equal(Object.hasOwn(chunk, 'terrain'), false); + assert.equal(Object.hasOwn(chunk, 'fluids'), false); + }); +}); + +test('spike runs decode mixed-radix direction and every thickness while preserving original surfaces', async t => { + const app = fixture(t); + const thicknesses = ['base', 'middle', 'frustum', 'tip_merge', 'tip']; + const states = [{name: 'before', num_values: 2}, + {name: 'vertical_direction', num_values: 2, values: ['down', 'up']}, + {name: 'thickness', num_values: 5, values: thicknesses}, + {name: 'waterlogged', num_values: 2, values: ['true', 'false']}, {name: 'after', num_values: 3}]; + const state_id = (direction, thickness) => 1000 + (((2 + direction) * 5 + thickness) * 2 + 1) * 3 + 2; + for (let direction = 0; direction < 2; direction++) for (let thickness = 0; thickness < 5; thickness++) + app.bot.registry.blocksByStateId[state_id(direction, thickness)] = {name: 'minecraft:pointed_dripstone', minStateId: 1000, states}; + app.columns.set('-1,-2', app.column((x, y, z) => { + if (z || x > 2) return y === 4 ? 101 : 7; + if (x === 2) return y === 9 ? 404 : y < 9 ? 999_999 : 7; + return y <= 7 && y >= 3 ? state_id(x, 7 - y) : y === 2 ? 303 : 7; + })); + const {chunks: [chunk]} = await app.snapshot(); + const top = decoded(chunk), bases = decoded(chunk, chunk.spikes); + assert.equal(Object.hasOwn(chunk, 'terrain'), false); + assert.equal(Object.hasOwn(chunk, 'fluids'), false); + for (const x of [0, 1]) { + assert.equal(top[x].name, 'minecraft:pointed_dripstone'); assert.equal(top[x].height, 12); + assert.equal(bases[x].name, 'stone'); assert.equal(bases[x].height, 7); + } + for (let index = 2; index < 256; index++) assert.deepEqual(bases[index], top[index]); + assert.deepEqual(chunk.spikes.segments, [0, 1].flatMap(index => thicknesses.map((thickness, offset) => + ({index, height: 12 - offset, direction: index ? 'up' : 'down', thickness})))); +}); + +test('spike bases stop at air gaps and the world minimum, and absent metadata defaults to up/tip', async t => { + const app = fixture(t); + app.bot.registry.blocksByStateId[600] = {name: 'pointed_dripstone'}; + app.columns.set('-1,-2', app.column((x, y, z) => { + if (z) return 7; + if (x === 0) return y === 4 || y === 3 ? 600 : y === -4 ? 303 : 7; + if (x === 1) return y === -4 ? 600 : 7; + if (x === 2) return y > 4 ? 7 : y === 4 ? 600 : y === 3 ? 8 : 999_999; + if (x === 3) return y > 4 ? 7 : y === 4 ? 600 : y === 3 ? 9 : 999_999; + return 7; + })); + const {chunks: [chunk]} = await app.snapshot(); + const bases = decoded(chunk, chunk.spikes); + for (const x of [0, 1, 2, 3]) assert.deepEqual(bases[x], {height: 0, name: 'air', color: 0}); + assert.deepEqual(chunk.spikes.segments, [ + {index: 0, height: 9, direction: 'up', thickness: 'tip'}, + {index: 0, height: 8, direction: 'up', thickness: 'tip'}, + {index: 1, height: 1, direction: 'up', thickness: 'tip'}, + {index: 2, height: 9, direction: 'up', thickness: 'tip'}, + {index: 3, height: 9, direction: 'up', thickness: 'tip'}, + ]); +}); + +test('underwater spikes override only spike bases and mirror terrain in every other cell', async t => { + const app = fixture(t); + app.bot.registry.blocksByStateId[600] = {name: 'pointed_dripstone'}; + app.bot.registry.blocksByStateId[601] = {name: 'lava'}; + app.columns.set('-1,-2', app.column((x, y, z) => { + if (z) return y === 0 ? 101 : 7; + if (x <= 1) return y === 8 || y === 7 ? 202 : y === 6 || y === 5 ? 600 : y === 4 && !x ? 303 : 7; + if (x === 2) return y === 8 ? 202 : y === 7 ? 303 : 7; + if (x === 3) return y > 8 ? 7 : y === 8 ? 202 : y === 7 ? 601 : 999_999; + if (x === 4) return y > 8 ? 7 : y === 8 ? 404 : 999_999; + return 7; + })); + const {chunks: [chunk]} = await app.snapshot(); + const top = decoded(chunk), terrain = decoded(chunk, chunk.terrain), fluids = decoded(chunk, chunk.fluids); + const bases = decoded(chunk, chunk.spikes); + for (const x of [0, 1]) { + assert.equal(top[x].name, 'water'); assert.equal(top[x].height, 13); + assert.equal(fluids[x].height, 13); assert.equal(fluids[x].level, 0); + assert.equal(terrain[x].name, 'pointed_dripstone'); assert.equal(terrain[x].height, 11); + } + assert.equal(bases[0].name, 'stone'); assert.equal(bases[0].height, 9); + assert.equal(bases[1].height, 0); assert.equal(bases[1].name, 'air'); + for (let index = 2; index < 256; index++) assert.deepEqual(bases[index], terrain[index]); + assert.deepEqual(chunk.spikes.segments.map(({index, height}) => ({index, height})), + [{index: 0, height: 11}, {index: 0, height: 10}, {index: 1, height: 11}, {index: 1, height: 10}]); +}); + +test('spike caps commit complete columns and retain original columns when limits would be exceeded', async t => { + await t.test('sixteen per column, oversized runs omitted', async t => { + const app = fixture(t, {minY: 0, height: 20}); + app.bot.registry.blocksByStateId[600] = {name: 'pointed_dripstone'}; + app.columns.set('-1,-2', app.column((x, y, z) => !z && x < 2 ? (y >= (x ? 3 : 4) ? 600 : 303) : 7)); + const {chunks: [chunk]} = await app.snapshot(); + const top = decoded(chunk), bases = decoded(chunk, chunk.spikes); + assert.equal(chunk.spikes.segments.length, 16); + assert.ok(chunk.spikes.segments.every(segment => segment.index === 0)); + assert.equal(bases[0].height, 4); assert.equal(bases[0].name, 'stone'); + assert.deepEqual(bases[1], top[1]); + }); + await t.test('chunk budget does not commit partial columns and later smaller runs can fit', async t => { + const app = fixture(t, {minY: 0, height: 20}); + app.bot.registry.blocksByStateId[600] = {name: 'pointed_dripstone'}; + app.columns.set('-1,-2', app.column((x, y, z) => { + const index = z * 16 + x; + const length = index < 31 ? 16 : index === 31 ? 15 : index === 32 ? 2 : 1; + return y >= 20 - length ? 600 : 303; + })); + const {chunks: [chunk]} = await app.snapshot(); + const top = decoded(chunk), bases = decoded(chunk, chunk.spikes); + assert.equal(chunk.spikes.segments.length, 512); + assert.equal(chunk.spikes.segments.filter(segment => segment.index === 31).length, 15); + assert.equal(chunk.spikes.segments.filter(segment => segment.index === 32).length, 0); + assert.equal(chunk.spikes.segments.filter(segment => segment.index === 33).length, 1); + assert.equal(bases[33].height, 19); assert.equal(bases[33].name, 'stone'); + for (const index of [32, ...Array.from({length: 222}, (_, offset) => offset + 34)]) + assert.deepEqual(bases[index], top[index]); + assert.ok(app.reads <= 256 + 512 + 256 + 3, `bounded spike scan: ${app.reads}`); + }); + await t.test('no successfully sampled spikes omits the feature', async t => { + const app = fixture(t, {minY: 0, height: 20}); + app.bot.registry.blocksByStateId[600] = {name: 'pointed_dripstone'}; + app.columns.set('-1,-2', app.column(() => 600)); + const {chunks: [chunk]} = await app.snapshot(); + assert.equal(Object.hasOwn(chunk, 'spikes'), false); + assert.ok(decoded(chunk).every(cell => cell.name === 'pointed_dripstone' && cell.height === 20)); + assert.equal(app.reads, 256 * 17); + }); + await t.test('underwater oversized runs retain pointed terrain and fluid surfaces', async t => { + const app = fixture(t, {minY: 0, height: 20}); + app.bot.registry.blocksByStateId[600] = {name: 'pointed_dripstone'}; + app.columns.set('-1,-2', app.column((x, y, z) => { + if (z || x > 1) return 7; + if (y === 19) return 202; + return y >= (x ? 2 : 17) ? 600 : 303; + })); + const {chunks: [chunk]} = await app.snapshot(); + const top = decoded(chunk), terrain = decoded(chunk, chunk.terrain), fluids = decoded(chunk, chunk.fluids); + const bases = decoded(chunk, chunk.spikes); + assert.equal(chunk.spikes.segments.length, 2); + assert.ok(chunk.spikes.segments.every(segment => segment.index === 0)); + assert.equal(bases[0].name, 'stone'); assert.equal(bases[0].height, 17); + assert.deepEqual(bases[1], terrain[1]); + assert.equal(terrain[1].name, 'pointed_dripstone'); assert.equal(terrain[1].height, 19); + assert.equal(top[1].name, 'water'); assert.equal(top[1].height, 20); + assert.deepEqual(fluids[1], {...top[1], level: 0}); + }); +}); + +test('spike scans reject unknown states and preserve yield/cancellation bounds without caching partial features', async t => { + await t.test('unknown segment or foundation', async t => { + const app = fixture(t); + app.bot.registry.blocksByStateId[600] = {name: 'pointed_dripstone'}; + let unknown = true; + app.columns.set('-1,-2', app.column((_x, y) => y >= 9 ? 600 : unknown ? 999_999 : 303)); + await assert.rejects(app.snapshot(), code('INTERNAL_ERROR')); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + unknown = false; + const {chunks: [chunk]} = await app.snapshot(); + assert.equal(chunk.spikes.segments.length, 510); + assert.equal(decoded(chunk, chunk.spikes)[0].height, 13); + }); + await t.test('cancel at row yield', async t => { + const app = fixture(t, {minY: 0, height: 20}); + app.bot.registry.blocksByStateId[600] = {name: 'pointed_dripstone'}; + app.columns.set('-1,-2', app.column((_x, y) => y >= 4 ? 600 : 303)); + let stale = false; + setImmediate(() => { stale = true; }); + await assert.rejects(app.snapshot({}, () => { + if (stale) throw Object.assign(new Error(), {code: 'SESSION_STALE'}); + }), code('SESSION_STALE')); + assert.equal(app.reads, 32 * 17); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + app.columns.set('-1,-2', app.column(() => 303)); + const {chunks: [chunk]} = await app.snapshot(); + assert.equal(Object.hasOwn(chunk, 'spikes'), false); + }); +}); + +test('real installed WorldSync and chunk registry produce the same surface wire format', async t => { + const installed = createRequire(require.resolve('@wp2508/mineflayer')); + const registry = installed('@wp2508/minecraft-data')('1.21.4'); + const Chunk = installed('prismarine-chunk')('1.21.4'); + const World = installed('prismarine-world')('1.21.4'); + const chunk = new Chunk({minY: -64, worldHeight: 384}); + chunk.setBlockStateId({x: 0, y: 67, z: 0}, registry.blocksByName.grass_block.minStateId); + chunk.setBlockStateId({x: 1, y: 70, z: 0}, registry.blocksByName.water.minStateId); + for (const [x, name, level] of [[3, 'water', 3], [4, 'water', 8], [5, 'water', 15], + [6, 'lava', 0], [7, 'lava', 8], [8, 'lava', 15]]) { + chunk.setBlockStateId({x, y: 70, z: 0}, registry.blocksByName[name].minStateId + level); + chunk.setBlockStateId({x, y: 69, z: 0}, registry.blocksByName.stone.minStateId); + } + const dripstone = registry.blocksByName.pointed_dripstone; + const spike_state = properties => dripstone.minStateId + dripstone.states.reduce((offset, property) => + offset * property.num_values + (property.values ? property.values.indexOf(properties[property.name]) + : properties[property.name] === 'false' ? 1 : 0), 0); + const real_spikes = []; + for (const direction of ['up', 'down']) for (const thickness of ['tip', 'tip_merge', 'frustum', 'middle', 'base']) { + const index = 16 + real_spikes.length; + const state = spike_state({vertical_direction: direction, thickness, waterlogged: 'false'}); + chunk.setBlockStateId({x: index % 16, y: 70, z: 1}, state); + chunk.setBlockStateId({x: index % 16, y: 69, z: 1}, registry.blocksByName.stone.minStateId); + real_spikes.push({index, height: 135, direction, thickness}); + } + const app = fixture(t, {minY: -64, height: 384}); + const world = new World(null, null, {savingInterval: 0}); + app.bot.world = world.sync; + app.bot.registry = registry; + app.bot.world.setColumn(-1, -2, chunk, false); + const result = await app.snapshot(); + const cells = decoded(result.chunks[0]); + assert.equal(cells[0].height, 132); assert.equal(cells[0].name, 'grass_block'); + assert.equal(cells[1].height, 135); assert.equal(cells[1].name, 'water'); + assert.equal(cells[2].height, 0); + const fluids = decoded(result.chunks[0], result.chunks[0].fluids); + const terrain = decoded(result.chunks[0], result.chunks[0].terrain); + assert.equal(fluids[1].level, 0); assert.equal(terrain[1].height, 0); + for (const [x, name, level] of [[3, 'water', 3], [4, 'water', 8], [5, 'water', 15], + [6, 'lava', 0], [7, 'lava', 8], [8, 'lava', 15]]) { + assert.equal(fluids[x].name, name); assert.equal(fluids[x].level, level); assert.equal(fluids[x].height, 135); + assert.equal(terrain[x].name, 'stone'); assert.equal(terrain[x].height, 134); + } + assert.deepEqual(result.chunks[0].spikes.segments, real_spikes); + const bases = decoded(result.chunks[0], result.chunks[0].spikes); + for (const {index} of real_spikes) { + assert.equal(terrain[index].name, 'pointed_dripstone'); assert.equal(terrain[index].height, 135); + assert.equal(bases[index].name, 'stone'); assert.equal(bases[index].height, 134); + } +}); + +function transport_fixture(t) { + const app = fixture(t); + app.columns.set('-1,-2', app.column()); + const command = {forward_id: 0, forward_session_id: randomUUID(), node_session_id: 'node-session', node_id: 1, bot_id: 1, + url: 'http://127.0.0.1:1234', path: '/api/forward_server', namespace: '/', token: randomUUID(), + expires_at: Date.now() + 30_000, deadline_at: Date.now() + 2000}; + let socket; + const manager = new NodeForwardManager(1, 'node-session', { + stream_factory: () => app.stream, + socket_factory() { + socket = new EventEmitter(); socket.auth = {}; + socket.connect = () => { socket.connected = true; queueMicrotask(() => socket.emit('connect')); }; + socket.disconnect = () => { socket.connected = false; socket.emit('disconnect'); }; + socket.io = {engine: {writeBuffer: []}}; + socket.volatile = {emit() {}}; + socket.timeout = () => ({emit(event, _payload, ack) { + const {forward_id, forward_session_id, node_session_id, node_id, bot_id, expires_at} = command; + queueMicrotask(() => ack(null, {status: 'success', data: event === 'forward_node_ready' ? {node_ready: true} + : {forward_id, forward_session_id, node_session_id, node_id, bot_id, role: 'node', node_ready: false, expires_at}})); + }}); + return socket; + }, + }); + t.after(() => manager.stop()); + return {...app, manager, command, get socket() { return socket; }, + rpc(method, payload) { return new Promise(resolve => socket.emit('relay_request', request(method, payload, command.forward_session_id), resolve)); }, + }; +} + +test('manager dispatches world reads through guarded RPC and rejects malformed requests and disconnect during yield', async t => { + const app = transport_fixture(t); + assert.equal((await app.manager.ensure(app.command)).status, 'success'); + const snapshot = await app.rpc('world.snapshot', {radius: 1, known: []}); + assert.equal(snapshot.status, 'success'); assert.match(snapshot.data.watch_id, uuid); + assert.deepEqual(await app.rpc('world.unwatch', {watch_id: snapshot.data.watch_id}), {status: 'success', data: {released: true}}); + assert.equal((await app.rpc('world.snapshot', {radius: 4, known: []})).data.code, 'INVALID_REQUEST'); + const pending = app.rpc('world.snapshot', {radius: 1, known: []}); + setImmediate(() => app.socket.disconnect()); + assert.equal((await pending).data.code, 'SESSION_STALE'); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); +}); diff --git a/tests/world_voxels.test.mjs b/tests/world_voxels.test.mjs new file mode 100644 index 0000000..a12800f --- /dev/null +++ b/tests/world_voxels.test.mjs @@ -0,0 +1,293 @@ +import assert from 'node:assert/strict'; +import {randomUUID} from 'node:crypto'; +import {EventEmitter} from 'node:events'; +import {createRequire, isBuiltin} from 'node:module'; +import {fileURLToPath} from 'node:url'; +import test from 'node:test'; +import {build} from 'esbuild'; + +const require = createRequire(import.meta.url); +const installed = createRequire(require.resolve('@wp2508/mineflayer')); +const registry = installed('@wp2508/minecraft-data')('1.21.4'); +const compiled = await build({ + stdin: {contents: `export * from './bot/forward/stream.ts'; export * from './utils/forward_protocol.ts';`, + resolveDir: fileURLToPath(new URL('../', import.meta.url)), loader: 'ts'}, + bundle: true, platform: 'node', format: 'cjs', write: false, + plugins: [{name: 'isolated-world', setup(build) { + build.onResolve({filter: /(?:instance\/index\.ts|console_print\.ts|path_utils\.ts)$/}, args => + ({path: args.path.includes('instance/') ? 'registry' : args.path.endsWith('path_utils.ts') ? 'path' : 'console', namespace: 'stub'})); + build.onLoad({filter: /.*/, namespace: 'stub'}, ({path}) => ({loader: 'js', contents: { + registry: 'export const get_bot_context = () => undefined; export const observe_bot_context = () => () => {};', + console: 'export const console_print = {success: x => x, log() {}};', + path: 'export const get_project_root_path = () => {throw Error("Unexpected filesystem access")};', + }[path]})); + }}], +}); +const module = {exports: {}}; +new Function('require', 'module', 'exports', compiled.outputFiles[0].text)(specifier => { + assert.ok(isBuiltin(specifier)); + return require(specifier); +}, module, module.exports); +const {BotStream, validate_rpc} = module.exports; +const code = expected => error => error.code === expected; +const state = (name, properties = {}) => { + const block = registry.blocksByName[name]; + return block.minStateId + block.states.reduce((offset, property) => { + const value = properties[property.name]; + const index = property.type === 'bool' ? value === true ? 0 : 1 : property.values.indexOf(String(value)); + assert.ok(index >= 0, `${name}.${property.name}`); + return offset * property.num_values + index; + }, 0); +}; + +function fixture(t, {minY = -4, height = 32} = {}) { + let now = 100, reads = 0; + const columns = new Map(); + const bot = Object.assign(new EventEmitter(), { + chat() {}, version: '1.21.4', game: {minY, height, dimension: 'overworld'}, registry, + world: {getColumn(x, z) {return columns.get(`${x},${z}`);}}, + entity: {position: {x: -0.25, y: 4, z: -16.25}, yaw: 0}, + }); + const context = Object.assign(new EventEmitter(), {bot_status: 'online', bot_instance: bot}); + let current = context; + const stream = new BotStream(1, {source: {get: () => current, observe: () => () => {}}, now: () => now}); + t.after(() => stream.dispose()); + const column = (get = () => registry.blocksByName.stone.defaultState) => ({getBlockStateId(p) { + assert.ok(p.x >= 0 && p.x < 16 && p.z >= 0 && p.z < 16); + assert.ok(p.y >= bot.game.minY && p.y < bot.game.minY + bot.game.height); + reads++; + return get(p.x, p.y, p.z); + }}); + columns.set('-1,-2', column()); + return {bot, context, stream, columns, column, + get reads() {return reads;}, + time(value) {now = value;}, set_context(value) {current = value;}, + snapshot(input = {}) {return stream.world_snapshot({radius: 1, known: [], ...input}, now + 2000);}, + chunk(snapshot, input = {}, check = () => {}) { + return stream.world_chunk({x: -1, z: -2, world_generation: snapshot.world_generation, + watch_id: snapshot.watch_id, ...input}, now + 2000, check); + }, + }; +} + +function decode(chunk) { + assert.deepEqual(Object.keys(chunk).sort(), ['x', 'z', 'revision', 'world_generation', 'min_y', 'height', 'palette', 'sections'].sort()); + assert.ok(chunk.palette.length >= 1 && chunk.palette.length <= 8192); + let previous = -1; + const sections = new Map(); + for (const section of chunk.sections) { + assert.ok(section.y > previous && section.y < Math.ceil(chunk.height / 16)); + previous = section.y; + const bytes = Buffer.from(section.blocks, 'base64'); + assert.equal(bytes.length, 8192); + assert.equal(bytes.toString('base64'), section.blocks); + for (let i = 0; i < 4096; i++) assert.ok(bytes.readUInt16LE(i * 2) < chunk.palette.length); + sections.set(section.y, bytes); + } + return (x, y, z) => { + const offset = y - chunk.min_y, bytes = sections.get(Math.floor(offset / 16)); + return bytes ? chunk.palette[bytes.readUInt16LE(((offset % 16) * 256 + z * 16 + x) * 2)] : {name: 'air'}; + }; +} + +test('world.chunk validates exact integer coordinates and UUIDs without invoking accessors', () => { + const payload = {x: -1, z: 2, world_generation: randomUUID(), watch_id: randomUUID()}; + const request = value => ({method: 'world.chunk', payload: value, request_id: randomUUID(), correlation_id: randomUUID(), + forward_session_id: randomUUID(), deadline_at: 1000}); + assert.equal(validate_rpc(request(payload)).payload, payload); + for (const invalid of [{}, {...payload, x: 1.5}, {...payload, z: Infinity}, {...payload, x: '1'}, + {...payload, watch_id: undefined}, {...payload, world_generation: 'bad'}, {...payload, radius: 3}]) + assert.throws(() => validate_rpc(request(invalid)), code('INVALID_REQUEST')); + for (const key of Object.keys(payload)) { + let reads = 0; + const invalid = Object.defineProperty({...payload}, key, {enumerable: true, get() {reads++; return payload[key];}}); + assert.throws(() => validate_rpc(request(invalid)), code('INVALID_REQUEST')); + assert.equal(reads, 0); + } +}); + +test('real installed chunks retain cave floors, interiors, roofs and light below the surface', async t => { + const app = fixture(t, {minY: -64, height: 384}); + const Chunk = installed('prismarine-chunk')('1.21.4'); + const World = installed('prismarine-world')('1.21.4'); + const column = new Chunk({minY: -64, worldHeight: 384}); + const world = new World(null, null, {savingInterval: 0}); + app.bot.world = world.sync; + app.bot.world.setColumn(-1, -2, column, false); + for (const [x, y, z, name] of [[0, -64, 0, 'bedrock'], [0, -40, 0, 'diamond_ore'], [0, 80, 0, 'stone'], + [15, 79, 15, 'glass'], [1, -39, 2, 'cave_air'], [1, -38, 2, 'void_air']]) + column.setBlockStateId({x, y, z}, registry.blocksByName[name].defaultState); + column.setBlockStateId({x: 2, y: -20, z: 3}, state('light', {level: 9, waterlogged: false})); + const snapshot = await app.snapshot(), chunk = await app.chunk(snapshot), cell = decode(chunk); + assert.equal(chunk.revision, snapshot.loaded[0].revision); + assert.deepEqual(chunk.sections.map(s => s.y), [0, 1, 2, 8, 9]); + assert.equal(cell(0, -64, 0).name, 'bedrock'); + assert.equal(cell(0, -40, 0).name, 'diamond_ore'); + assert.equal(cell(0, -39, 0).name, 'air'); + assert.equal(cell(0, 80, 0).name, 'stone'); + assert.equal(cell(15, 79, 15).name, 'glass'); + assert.equal(cell(1, -39, 2).name, 'cave_air'); + assert.equal(cell(1, -38, 2).name, 'void_air'); + assert.deepEqual(cell(2, -20, 3).properties, {waterlogged: false, level: 9}); + const unchanged = await app.snapshot({world_generation: snapshot.world_generation, known: snapshot.loaded, watch_id: snapshot.watch_id}); + assert.deepEqual(unchanged.chunks, []); +}); + +test('palette preserves generic typed registry properties, deduplicates state IDs and excludes partial occluders', async t => { + const app = fixture(t); + const samples = [ + ['wheat', {age: 7}], ['pink_petals', {facing: 'south', flower_amount: 4}], + ['snow', {layers: 6}], ['pointed_dripstone', {thickness: 'frustum', vertical_direction: 'down', waterlogged: true}], + ['lever', {face: 'ceiling', facing: 'east', powered: true}], ['cake', {bites: 5}], + ['beehive', {facing: 'west', honey_level: 4}], ['oak_slab', {type: 'bottom', waterlogged: false}], + ['oak_leaves', {distance: 3, persistent: true, waterlogged: false}], ['glass', {}], ['torch', {}], + ['water', {level: 8}], ['stone', {}], ['oak_slab', {type: 'top', waterlogged: false}], + ]; + const states = samples.map(([name, properties]) => state(name, properties)); + app.columns.set('-1,-2', app.column((x, y, z) => y === -3 && z === 3 && x < states.length ? states[x] + : y === 27 && z === 3 && x === 0 ? states[0] : registry.blocksByName.air.defaultState)); + const chunk = await app.chunk(await app.snapshot()), cell = decode(chunk); + for (let x = 0; x < samples.length; x++) { + const [name, properties] = samples[x], result = cell(x, -3, 3); + assert.equal(result.name, name); + assert.deepEqual(result.properties, properties); + assert.equal(typeof result.color, 'number'); + assert.equal(result.occluding, name === 'stone' || name === 'beehive'); + } + assert.equal(chunk.palette.filter(p => p.name === 'wheat').length, 1); + assert.equal(chunk.palette.filter(p => p.name === 'oak_slab').length, 2); + assert.equal(cell(0, 27, 3).properties.age, 7); +}); + +test('sections use offsets from nonaligned minimum, pad with air and omit only air variants', async t => { + const app = fixture(t, {minY: -3, height: 33}); + app.columns.set('-1,-2', app.column((x, y, z) => y === 29 && x === 15 && z === 14 ? state('light', {level: 0, waterlogged: true}) + : [registry.blocksByName.air.defaultState, registry.blocksByName.cave_air.defaultState, registry.blocksByName.void_air.defaultState][(x + z) % 3])); + const chunk = await app.chunk(await app.snapshot()); + assert.deepEqual(chunk.sections.map(s => s.y), [2]); + assert.equal(decode(chunk)(15, 29, 14).name, 'light'); + const last = Buffer.from(chunk.sections[0].blocks, 'base64'); + for (let i = 256; i < 4096; i++) assert.equal(chunk.palette[last.readUInt16LE(i * 2)].name, 'air'); + app.bot.emit('blockUpdate', null, {position: {x: -1, z: -17}}); + app.columns.set('-1,-2', app.column(() => registry.blocksByName.cave_air.defaultState)); + const empty = await app.chunk(await app.snapshot()); + assert.equal(empty.palette.length, 1); + assert.deepEqual(empty.sections, []); +}); + +test('chunk requires a live watch in its generation and radius, and never renews its lease', async t => { + const app = fixture(t), snapshot = await app.snapshot(); + await assert.rejects(app.chunk(snapshot, {watch_id: randomUUID()}), code('CANCELLED')); + await assert.rejects(app.chunk(snapshot, {world_generation: randomUUID()}), code('CANCELLED')); + const before = app.reads; + await assert.rejects(app.chunk(snapshot, {x: 1}), code('FORBIDDEN')); + assert.equal(app.reads, before); + app.time(44_000); + await app.chunk(snapshot); + app.time(45_101); + await assert.rejects(app.chunk(snapshot), code('CANCELLED')); + assert.equal(app.bot.listenerCount('blockUpdate'), 0); + const fresh = await app.snapshot(); + app.bot.game.dimension = 'the_nether'; + await assert.rejects(app.chunk(fresh), code('CANCELLED')); +}); + +test('voxel reads cancel after at most 2048 reads on deadline, transport and changing loaded entry', async t => { + for (const kind of ['deadline', 'transport', 'unload', 'replace', 'block', 'respawn', 'move', 'context', 'offline', 'unwatch']) { + await t.test(kind, async t => { + const app = fixture(t, {height: 64}), snapshot = await app.snapshot(); + const before = app.reads; + let disconnected = false; + setImmediate(() => { + if (kind === 'deadline') app.time(5000); + if (kind === 'transport') disconnected = true; + if (kind === 'unload') app.columns.delete('-1,-2'); + if (kind === 'replace') app.columns.set('-1,-2', app.column()); + if (kind === 'block') app.bot.emit('blockUpdate', null, {position: {x: -1, z: -17}}); + if (kind === 'respawn') app.bot.emit('respawn'); + if (kind === 'move') app.bot.entity.position.x = 32; + if (kind === 'context') app.set_context(undefined); + if (kind === 'offline') app.context.bot_status = 'offline'; + if (kind === 'unwatch') app.stream.world_unwatch({watch_id: snapshot.watch_id}, 2100); + }); + await assert.rejects(app.chunk(snapshot, {}, () => { + if (disconnected) throw Object.assign(new Error(), {code: 'SESSION_STALE'}); + }), code(kind === 'deadline' ? 'TIMEOUT' : kind === 'transport' ? 'SESSION_STALE' : kind === 'offline' ? 'BOT_NOT_ONLINE' : 'CANCELLED')); + assert.equal(app.reads - before, 2048); + }); + } +}); + +test('two voxel requests may run independently of four surface requests', async t => { + const app = fixture(t, {height: 64}), snapshot = await app.snapshot(); + const pending = [app.chunk(snapshot), app.chunk(snapshot)]; + await assert.rejects(app.chunk(snapshot), code('RATE_LIMITED')); + assert.equal((await Promise.all(pending)).length, 2); + await app.chunk(snapshot); +}); + +test('cache shares surface revisions, returns detached data and invalidates on block events, unload and idle', async t => { + const app = fixture(t), snapshot = await app.snapshot(); + const first = await app.chunk(snapshot), before = app.reads; + first.palette[0].name = 'mutated'; first.sections.length = 0; + const second = await app.chunk(snapshot); + assert.equal(app.reads, before); + assert.equal(second.palette[0].name, 'stone'); assert.equal(second.sections.length, 2); + app.bot.emit('blockUpdate', null, {position: {x: -1, z: -17}}); + const third = await app.chunk(snapshot); + assert.notEqual(third.revision, second.revision); + const updated = await app.snapshot({world_generation: snapshot.world_generation, known: snapshot.loaded, watch_id: snapshot.watch_id}); + assert.equal(updated.loaded[0].revision, third.revision); + app.columns.delete('-1,-2'); + await assert.rejects(app.chunk(updated), code('CANCELLED')); + app.columns.set('-1,-2', app.column()); + const reloaded = await app.chunk(updated); + assert.notEqual(reloaded.revision, third.revision); + app.stream.world_unwatch({watch_id: snapshot.watch_id}, 2100); + await assert.rejects(app.chunk(snapshot), code('CANCELLED')); +}); + +test('voxel LRU retains at most eight columns', async t => { + const app = fixture(t, {height: 1}); + for (let z = -3; z <= -1; z++) for (let x = -2; x <= 0; x++) app.columns.set(`${x},${z}`, app.column()); + const snapshot = await app.snapshot(); + const coordinates = snapshot.loaded; + for (const {x, z} of coordinates) await app.chunk(snapshot, {x, z}); + const before_cached = app.reads; + await app.chunk(snapshot, {x: coordinates[8].x, z: coordinates[8].z}); + assert.equal(app.reads, before_cached); + await app.chunk(snapshot, {x: coordinates[0].x, z: coordinates[0].z}); + assert.equal(app.reads - before_cached, 256); +}); + +test('unknown interior states and excessive palettes fail without caching partial data', async t => { + const app = fixture(t, {minY: 0, height: 64}); + let unknown = true; + app.columns.set('-1,-2', app.column((_x, y) => y === 0 && unknown ? 999_999 : registry.blocksByName.stone.defaultState)); + const snapshot = await app.snapshot(); + await assert.rejects(app.chunk(snapshot), code('INTERNAL_ERROR')); + unknown = false; + assert.equal((await app.chunk(snapshot)).sections.length, 4); + app.bot.registry = {blocksByStateId: Object.fromEntries(Array.from({length: 8193}, (_, i) => [i, {name: `block_${i}`}]))}; + app.columns.set('-1,-2', app.column((x, y, z) => Math.min(8192, y * 256 + z * 16 + x))); + await assert.rejects(app.chunk(snapshot), code('INTERNAL_ERROR')); + app.bot.registry.blocksByStateId[8192] = app.bot.registry.blocksByStateId[0]; + app.columns.set('-1,-2', app.column((x, y, z) => (y * 256 + z * 16 + x) % 8192)); + assert.equal((await app.chunk(snapshot)).palette.length, 8192); +}); + +test('invalid generic registry properties are rejected and maximum height reads every cell', async t => { + const app = fixture(t, {minY: -4096, height: 4096}), snapshot = await app.snapshot(); + const before = app.reads, chunk = await app.chunk(snapshot); + assert.equal(app.reads - before, 4096 * 256); + assert.equal(chunk.sections.length, 256); + assert.equal(chunk.sections[255].y, 255); + assert.equal(decode(chunk)(15, -1, 15).name, 'stone'); + for (const states of [Array.from({length: 33}, (_, i) => ({name: `p${i}`, type: 'bool', num_values: 2})), + [{name: 'age', type: 'int', num_values: 1, values: ['NaN']}], + [{name: 'shape', type: 'enum', num_values: 1, values: ['x'.repeat(65)]}]]) { + app.bot.registry = {blocksByStateId: {0: {name: 'bad', minStateId: 0, states}}}; + app.columns.set('-1,-2', app.column(() => 0)); + await assert.rejects(app.chunk(snapshot), code('INTERNAL_ERROR')); + } +}); diff --git a/types/forward.ts b/types/forward.ts index 93baa0f..107b466 100644 --- a/types/forward.ts +++ b/types/forward.ts @@ -41,7 +41,59 @@ export interface JoinReady { } export type RelayMethod = 'bot.snapshot' | 'bot.logs' | 'bot.chat' | 'get_log_history' | 'tab_complete' - | 'get_context_info' | 'execute_function' | 'get_data_file_list' | 'get_file_data' | keyof WorkflowPayloads; + | 'get_context_info' | 'execute_function' | 'get_data_file_list' | 'get_file_data' | keyof WorkflowPayloads | keyof WorldPayloads; + +export interface WorldRevision {x: number; z: number; revision: string} + +export interface WorldPayloads { + 'world.snapshot': {radius: number; known: WorldRevision[]; world_generation?: string; watch_id?: string}; + 'world.chunk': {x: number; z: number; world_generation: string; watch_id: string}; + 'world.unwatch': {watch_id: string}; +} + +export interface WorldVoxelChunk extends WorldRevision { + world_generation: string; + min_y: number; + height: number; + palette: Array<{name: string; color: number; properties: Record; occluding: boolean}>; + // y is an offset from min_y in units of 16, including when min_y is not aligned. + // Each canonical base64 section holds 4096 UInt16LE indices at localY * 256 + z * 16 + x. + sections: Array<{y: number; blocks: string}>; +} + +export interface WorldChunk extends WorldRevision { + palette: Array<{name: string; color: number}>; + // All arrays are canonical base64 of 256 UInt16LE values, indexed by z * 16 + x. + heights: string; + blocks: string; + // Emitted together when a fluid is visible; heights of zero mean no entry. + terrain?: {heights: string; blocks: string}; + fluids?: {heights: string; blocks: string; levels: string}; + // Bases mirror terrain ?? top except for fully sampled spike columns. + spikes?: {heights: string; blocks: string; segments: Array<{ + index: number; height: number; direction: 'up' | 'down'; + thickness: 'tip' | 'tip_merge' | 'frustum' | 'middle' | 'base'; + }>}; + features?: {heights: string; blocks: string; entries: Array<{ + index: number; height: number; block: number; properties: Record; + }>}; +} + +export interface WorldSnapshot { + world_generation: string; + context_generation: string; + stream_id: string; + watch_id: string; + version: string; + dimension: string; + min_y: number; + height: number; + center: {x: number; z: number}; + position: {x: number; y: number; z: number; yaw: number}; + chunks: WorldChunk[]; + loaded: WorldRevision[]; + partial: boolean; +} export interface WorkflowPayloads { 'workflow.start': {workflow_id: string; trigger?: Record}; diff --git a/utils/forward_protocol.ts b/utils/forward_protocol.ts index 81f0557..4419124 100644 --- a/utils/forward_protocol.ts +++ b/utils/forward_protocol.ts @@ -1,4 +1,4 @@ -import type {EnsureForward, ForwardAck, ForwardErrorCode, JoinReady, RelayRequest, WorkflowPayloads} from '../types/forward.ts'; +import type {EnsureForward, ForwardAck, ForwardErrorCode, JoinReady, RelayRequest, WorkflowPayloads, WorldPayloads} from '../types/forward.ts'; import {assert_json} from '../bot/workflow/runtime/value.ts'; const errors: Record = { @@ -112,6 +112,11 @@ export function validate_rpc(value: unknown): RelayRequest { || !uuid(value.request_id) || !uuid(value.correlation_id) || !uuid(value.forward_session_id) || !integer(value.deadline_at, 1)) throw new ForwardFailure('INVALID_REQUEST'); switch (value.method) { + case 'world.snapshot': + case 'world.chunk': + case 'world.unwatch': + validate_world_payload(value.method, value.payload); + break; case 'workflow.start': case 'workflow.stop': case 'workflow.snapshot': @@ -217,3 +222,31 @@ export function validate_workflow_payload(meth } return payload as WorkflowPayloads[M]; } + +export function validate_world_payload(method: M, payload: unknown): WorldPayloads[M] { + if (method === 'world.unwatch') { + if (!exact(payload, ['watch_id']) || !uuid(payload.watch_id)) throw new ForwardFailure('INVALID_REQUEST'); + } else if (method === 'world.chunk') { + if (!exact(payload, ['x', 'z', 'world_generation', 'watch_id']) + || !Number.isSafeInteger(payload.x) || !Number.isSafeInteger(payload.z) + || !uuid(payload.world_generation) || !uuid(payload.watch_id)) throw new ForwardFailure('INVALID_REQUEST'); + } else if (method === 'world.snapshot') { + if (!exact(payload, ['radius', 'known'], ['world_generation', 'watch_id']) + || !integer(payload.radius, 1) || payload.radius > 3 + || Object.hasOwn(payload, 'world_generation') && !uuid(payload.world_generation) + || Object.hasOwn(payload, 'watch_id') && !uuid(payload.watch_id) + || !Array.isArray(payload.known) || payload.known.length > 49 + || Reflect.ownKeys(payload.known).length !== payload.known.length + 1) throw new ForwardFailure('INVALID_REQUEST'); + const seen = new Set(); + for (let index = 0; index < payload.known.length; index++) { + const descriptor = Object.getOwnPropertyDescriptor(payload.known, String(index)); + const item = descriptor && 'value' in descriptor && descriptor.enumerable ? descriptor.value : undefined; + if (!exact(item, ['x', 'z', 'revision']) || !Number.isSafeInteger(item.x) + || !Number.isSafeInteger(item.z) || !uuid(item.revision)) throw new ForwardFailure('INVALID_REQUEST'); + const key = `${item.x},${item.z}`; + if (seen.has(key)) throw new ForwardFailure('INVALID_REQUEST'); + seen.add(key); + } + } else throw new ForwardFailure('INVALID_REQUEST'); + return payload as WorldPayloads[M]; +}