feat(chat): HTTP chat endpoints on gateway; chat CLI bounded budgets, no more silent hangs

The chat CLI opened a fresh observer WebSocket per invocation. Over a TLS
tunnel (split-topology benchmark runs) that handshake is flaky, and a send
against a dead game-client relay burned the SDK's 60s actionTimeout per
80-char chunk - multi-chunk sends stalled 120s+ with zero output.

- gateway: GET /chat/:bot serves recent chat from a per-bot accumulated
  ChatHistory (fed from state snapshots); POST /say/:bot relays through the
  bot's game client with a hard per-chunk deadline (default 10s) and aborts
  remaining chunks on first failure. Same authenticateSDK gate as sdk_connect.
- sdk/chat.ts: HTTP-first with per-attempt timeouts + one retry on network
  failure, a 90s force-exit backstop on every non-watch path, and clear
  non-zero exits. Falls back to the observer WebSocket (now with bounded
  connectTimeout/actionTimeout) for gateways without the endpoints; --watch
  keeps the long-lived observer connection.
- chunkMessage moved to sdk/chunking.ts so the gateway can share it without
  importing the full SDK; re-exported from sdk/index.
- tests: sdk/test/gateway-http-chat.test.ts runs the real gateway subprocess
  with a scripted bot client, covering read/send/chunking/409/timeout paths
  and the CLI end-to-end.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
max bittkerandClaude Fable 5 committed 2026-08-16 22:29:53 -04:00
1 parent 2ae032c998
commit 98b28b1ca3
6 files changed
+698 -137

No files matched your search

+4
View File
@@ -530,6 +530,10 @@ interface BotWorldState {
combatStyle?: CombatStyleState;
combatEvents: CombatEvent[];
prayers: PrayerState;
/** Absent when the connected client predates authoritative PK-skull telemetry. */
isSkulled?: boolean;
/** Absent when the connected client predates authoritative auto-retaliate telemetry. */
autoRetaliateEnabled?: boolean;
/** Absent when the connected client predates the signal. */
opFeedback?: OpFeedbackState;
}
+238 -81
View File
@@ -1,9 +1,12 @@
#!/usr/bin/env bun
// SDK Chat CLI - send and read chat without taking control of the bot
//
// Connects in 'observe' mode: the gateway lets observers dispatch 'say' (and
// nothing else), and observers neither pre-empt nor get pre-empted by whatever
// controller currently owns the bot. Safe to use while a script is running.
// Talks to the gateway's HTTP chat endpoints (POST /say/:bot, GET /chat/:bot),
// so every invocation is one bounded request/response - there is no
// per-invocation WebSocket handshake to stall on a flaky link, and a dead
// game-client relay fails fast with a clear message instead of hanging.
// Gateways that predate the HTTP endpoints get the legacy observer-WebSocket
// path automatically; --watch always uses it (it needs the live feed).
//
// Usage:
// bun sdk/chat.ts <botname> # Show recent chat
@@ -15,6 +18,11 @@ import type { GameMessage } from './types';
import { existsSync, readFileSync } from 'fs';
import { join } from 'path';
// No invocation (except --watch) may outlive this, whatever goes wrong - a
// chat check that silently eats a shell tool's whole timeout is worse than a
// failed one.
const HARD_DEADLINE_MS = 90_000;
function printUsage() {
console.log(`
SDK Chat CLI - send and read chat without stealing bot control
@@ -28,7 +36,7 @@ Options:
--limit <n> Messages of history to show (default: 20)
--system Include system/game messages, not just player chat
--server <host> Server hostname (default: from bot.env or rs-sdk-demo.fly.dev)
--timeout <ms> Budget for connecting / first state (default: 5000)
--timeout <ms> Budget per network request (default: 5000)
--help Show this help
Examples:
@@ -77,6 +85,212 @@ function formatMessage(m: GameMessage): string {
const PLAYER_AND_SYSTEM_TYPES = [0, 1, 2, 3, 6, 7] as const;
/** ws://host -> http://host, wss://host/path -> https://host/path (the gateway serves both on one port). */
function gatewayHttpBase(gatewayUrl: string): string {
return gatewayUrl.replace(/^ws(s?):\/\//, 'http$1://').replace(/\/$/, '');
}
/**
* fetch with a hard per-attempt timeout and one retry on network-level
* failure (refused/reset/stalled connect - the flaky-tunnel cases). HTTP
* error statuses are answers, not failures, and are returned as-is.
*/
async function boundedFetch(url: string, init: RequestInit, timeoutMs: number): Promise<Response> {
let lastErr: unknown;
for (let attempt = 0; attempt < 2; attempt++) {
try {
return await fetch(url, { ...init, signal: AbortSignal.timeout(timeoutMs) });
} catch (err) {
lastErr = err;
if (attempt === 0) await new Promise(resolve => setTimeout(resolve, 500));
}
}
throw lastErr;
}
/** True if this response came from a gateway that has the HTTP chat endpoints. */
function isChatEndpointResponse(res: Response): boolean {
// Old gateways answer unknown paths with a 200 text/plain banner; the chat
// endpoints always answer JSON (including their error statuses).
return (res.headers.get('content-type') ?? '').includes('json');
}
interface CliContext {
username: string;
password: string;
gatewayUrl: string;
timeout: number;
limit: number;
system: boolean;
}
// ============ HTTP paths (primary) ============
/** Returns exit code, or null if the gateway lacks the HTTP endpoints. */
async function sendViaHttp(ctx: CliContext, text: string): Promise<number | null> {
const url = `${gatewayHttpBase(ctx.gatewayUrl)}/say/${encodeURIComponent(ctx.username)}`;
let res: Response;
try {
// The gateway bounds each chunk at 10s and aborts on first failure;
// this attempt timeout only has to outlast a legitimate slow send.
res = await boundedFetch(url, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ text, password: ctx.password, timeoutMs: 10_000 })
}, 60_000);
} catch (err: any) {
console.error(`Error: Failed to reach gateway at ${url}`);
console.error(` ${err?.message ?? err}`);
return 1;
}
if (!isChatEndpointResponse(res)) return null;
const body: any = await res.json().catch(() => null);
if (res.status === 409) {
console.error(`Error: Failed to send: ${body?.error ?? 'Bot not connected'}`);
console.error(` No game client is logged in as '${ctx.username}' - chat is relayed through it.`);
console.error(` Log the bot in (browser or lite client), then retry.`);
return 1;
}
if (!body || (!res.ok && !Array.isArray(body.results))) {
console.error(`Error: Failed to send: ${body?.error ?? `gateway answered ${res.status}`}`);
return 1;
}
let ok = true;
for (const result of body.results ?? []) {
if (result.success) {
console.log(`Sent: ${result.data?.finalText ?? result.chunk}`);
if (result.data?.truncated) console.log(' (truncated by server cap)');
if (result.data?.filtered) console.log(' (altered by word filter)');
} else {
ok = false;
console.error(`Error: Failed to send: ${result.message}`);
}
}
return ok ? 0 : 1;
}
/** Returns exit code, or null if the gateway lacks the HTTP endpoints. */
async function readViaHttp(ctx: CliContext): Promise<number | null> {
const params = new URLSearchParams({ limit: String(ctx.limit) });
if (ctx.system) params.set('system', '1');
if (ctx.password) params.set('password', ctx.password);
const url = `${gatewayHttpBase(ctx.gatewayUrl)}/chat/${encodeURIComponent(ctx.username)}?${params}`;
let res: Response;
try {
res = await boundedFetch(url, {}, ctx.timeout);
} catch (err: any) {
console.error(`Error: Failed to reach gateway at ${url}`);
console.error(` ${err?.message ?? err}`);
return 1;
}
if (!isChatEndpointResponse(res)) return null;
const body: any = await res.json().catch(() => null);
if (!res.ok || !body || !Array.isArray(body.messages)) {
console.error(`Error: Failed to read chat: ${body?.error ?? `gateway answered ${res.status}`}`);
return 1;
}
if (body.messages.length === 0) {
console.log('(no chat in recent history)');
if (body.botStatus && body.botStatus !== 'active') {
console.log(`(bot client is ${body.botStatus} - history only accumulates while it is connected)`);
}
}
for (const m of body.messages as GameMessage[]) {
console.log(formatMessage(m));
}
return 0;
}
// ============ Observer-WebSocket path (fallback + --watch) ============
async function runOverWebSocket(ctx: CliContext, messageToSend: string, watch: boolean): Promise<number> {
const sdk = new BotSDK({
botUsername: ctx.username,
password: ctx.password,
gatewayUrl: ctx.gatewayUrl,
connectionMode: 'observe',
autoReconnect: watch, // one-shots fail fast; watch survives blips
autoLaunchBrowser: false,
readyTimeout: 0,
connectTimeout: ctx.timeout, // SDK default (30s) is far past our budget
actionTimeout: 15_000, // a dead relay reports failure, not a 60s stall
showChat: true
});
try {
await sdk.connect();
} catch (err: any) {
console.error(`Error: Failed to connect to ${ctx.gatewayUrl}`);
console.error(` ${err.message}`);
return 1;
}
if (messageToSend) {
const results = await sdk.say(messageToSend);
let ok = true;
for (const result of results) {
if (result.success) {
const data = result.data as { finalText?: string; truncated?: boolean; filtered?: boolean } | undefined;
console.log(`Sent: ${data?.finalText ?? messageToSend}`);
if (data?.truncated) console.log(' (truncated by server cap)');
if (data?.filtered) console.log(' (altered by word filter)');
} else {
ok = false;
console.error(`Error: Failed to send: ${result.message}`);
if (/bot not connected/i.test(result.message)) {
console.error(` No game client is logged in as '${ctx.username}' - chat is relayed through it.`);
console.error(` Log the bot in (browser or lite client), then retry.`);
}
}
}
sdk.disconnect();
return ok ? 0 : 1;
}
// Reading (history or watch) needs the state feed.
const types = ctx.system ? PLAYER_AND_SYSTEM_TYPES : undefined;
try {
await sdk.waitForCondition(s => s !== null, ctx.timeout);
} catch {
if (sdk.isAuthenticated()) {
console.error(`Error: No game state for '${ctx.username}' - nothing is logged in as this bot, so there's no chat feed.`);
} else {
console.error(`Error: No state received for '${ctx.username}'`);
}
sdk.disconnect();
return 1;
}
const history = sdk.getChat({ limit: ctx.limit, types, includeSelf: true });
if (history.length === 0) {
console.log('(no chat in recent history)');
}
for (const m of history) {
console.log(formatMessage(m));
}
// Drain the new-message cursor so watch mode only prints what arrives next.
sdk.getNewChat({ types, includeSelf: true });
if (!watch) {
sdk.disconnect();
return 0;
}
console.log(`--- watching chat for '${ctx.username}' (Ctrl+C to stop) ---`);
setInterval(() => {
for (const m of sdk.getNewChat({ types, includeSelf: true })) {
const time = new Date().toTimeString().slice(0, 8);
console.log(`[${time}] ${formatMessage(m)}`);
}
}, 500);
return await new Promise<never>(() => {}); // runs until Ctrl+C
}
async function main() {
const args = process.argv.slice(2);
@@ -135,88 +349,31 @@ async function main() {
process.exit(1);
}
const gatewayUrl = deriveGatewayUrl(server);
const sdk = new BotSDK({
botUsername: username,
const ctx: CliContext = {
username,
password,
gatewayUrl,
connectionMode: 'observe',
autoReconnect: watch, // one-shots fail fast; watch survives blips
autoLaunchBrowser: false,
readyTimeout: 0,
showChat: true
});
try {
const connectTimeout = new Promise<never>((_, reject) => {
setTimeout(() => reject(new Error('Connection timeout')), timeout);
});
await Promise.race([sdk.connect(), connectTimeout]);
} catch (err: any) {
console.error(`Error: Failed to connect to ${gatewayUrl}`);
console.error(` ${err.message}`);
process.exit(1);
}
if (messageToSend) {
const results = await sdk.say(messageToSend);
let ok = true;
for (const result of results) {
if (result.success) {
const data = result.data as { finalText?: string; truncated?: boolean; filtered?: boolean } | undefined;
console.log(`Sent: ${data?.finalText ?? messageToSend}`);
if (data?.truncated) console.log(' (truncated by server cap)');
if (data?.filtered) console.log(' (altered by word filter)');
} else {
ok = false;
console.error(`Error: Failed to send: ${result.message}`);
if (/bot not connected/i.test(result.message)) {
console.error(` No game client is logged in as '${username}' - chat is relayed through it.`);
console.error(` Log the bot in (browser or lite client), then retry.`);
}
}
}
sdk.disconnect();
process.exit(ok ? 0 : 1);
}
// Reading (history or watch) needs the state feed.
const types = system ? PLAYER_AND_SYSTEM_TYPES : undefined;
try {
await sdk.waitForCondition(s => s !== null, timeout);
} catch {
if (sdk.isAuthenticated()) {
console.error(`Error: No game state for '${username}' - nothing is logged in as this bot, so there's no chat feed.`);
} else {
console.error(`Error: No state received for '${username}'`);
}
sdk.disconnect();
process.exit(1);
}
const history = sdk.getChat({ limit, types, includeSelf: true });
if (history.length === 0) {
console.log('(no chat in recent history)');
}
for (const m of history) {
console.log(formatMessage(m));
}
// Drain the new-message cursor so watch mode only prints what arrives next.
sdk.getNewChat({ types, includeSelf: true });
gatewayUrl: deriveGatewayUrl(server),
timeout,
limit,
system
};
if (!watch) {
sdk.disconnect();
process.exit(0);
// Absolute backstop: no code path (including the WS fallback) may park
// this process past the deadline.
const deadline = setTimeout(() => {
console.error(`Error: chat CLI exceeded its ${HARD_DEADLINE_MS / 1000}s deadline - giving up`);
process.exit(1);
}, HARD_DEADLINE_MS);
if (typeof (deadline as any).unref === 'function') (deadline as any).unref();
const httpExit = messageToSend ? await sendViaHttp(ctx, messageToSend) : await readViaHttp(ctx);
if (httpExit !== null) process.exit(httpExit);
// Gateway predates the HTTP chat endpoints - use the observer socket.
process.exit(await runOverWebSocket(ctx, messageToSend, false));
}
console.log(`--- watching chat for '${username}' (Ctrl+C to stop) ---`);
setInterval(() => {
for (const m of sdk.getNewChat({ types, includeSelf: true })) {
const time = new Date().toTimeString().slice(0, 8);
console.log(`[${time}] ${formatMessage(m)}`);
}
}, 500);
await runOverWebSocket(ctx, '', true);
}
main();
+58
View File
@@ -0,0 +1,58 @@
/**
* Chat message chunking, shared by the SDK's say() and the gateway's HTTP say
* endpoint. Lives in its own module so the gateway can import it without
* pulling in the full SDK (pathfinding, collision data, puppeteer glue).
*/
/**
* Nibble-cost mirror of the client's WordPack encoding: the 13 most common
* chars pack to 1 nibble, everything else (digits, punctuation, rarer letters)
* to 2. Chat frames carry a 1-byte length, so packed text is limited to ~240
* bytes (480 nibbles) regardless of the char cap - digit-heavy text hits it
* well before 400 chars.
*/
const WORDPACK_COMMON_CHARS = ' etaoihnsrdlu';
const MAX_PACKED_NIBBLES = 240 * 2;
function packedNibbles(text: string): number {
let nibbles = 0;
for (const ch of text.toLowerCase()) {
nibbles += WORDPACK_COMMON_CHARS.includes(ch) ? 1 : 2;
}
return nibbles;
}
/**
* Split text into chunks on word boundaries such that each chunk fits both the
* char cap and the wire's packed-byte budget (so the client never silently
* truncates a chunk). Words too big for one chunk are hard-split so nothing is
* dropped. Exported for tests.
*/
export function chunkMessage(text: string, maxLen: number): string[] {
const fits = (s: string) => s.length <= maxLen && packedNibbles(s) <= MAX_PACKED_NIBBLES;
const chunks: string[] = [];
let current = '';
for (const word of text.trim().split(/\s+/)) {
// A single oversized word: flush, then hard-split it greedily.
if (!fits(word)) {
if (current) { chunks.push(current); current = ''; }
let rest = word;
while (rest.length > 0) {
let take = Math.min(rest.length, maxLen);
while (take > 1 && !fits(rest.slice(0, take))) take--;
chunks.push(rest.slice(0, take));
rest = rest.slice(take);
}
continue;
}
const candidate = current ? `${current} ${word}` : word;
if (fits(candidate)) {
current = candidate;
} else {
if (current) chunks.push(current);
current = word;
}
}
if (current) chunks.push(current);
return chunks;
}
+4 -52
View File
@@ -114,58 +114,10 @@ interface PendingScreenshot {
timeout: ReturnType<typeof setTimeout>;
}
/**
* Nibble-cost mirror of the client's WordPack encoding: the 13 most common
* chars pack to 1 nibble, everything else (digits, punctuation, rarer letters)
* to 2. Chat frames carry a 1-byte length, so packed text is limited to ~240
* bytes (480 nibbles) regardless of the char cap - digit-heavy text hits it
* well before 400 chars.
*/
const WORDPACK_COMMON_CHARS = ' etaoihnsrdlu';
const MAX_PACKED_NIBBLES = 240 * 2;
function packedNibbles(text: string): number {
let nibbles = 0;
for (const ch of text.toLowerCase()) {
nibbles += WORDPACK_COMMON_CHARS.includes(ch) ? 1 : 2;
}
return nibbles;
}
/**
* Split text into chunks on word boundaries such that each chunk fits both the
* char cap and the wire's packed-byte budget (so the client never silently
* truncates a chunk). Words too big for one chunk are hard-split so nothing is
* dropped. Exported for tests.
*/
export function chunkMessage(text: string, maxLen: number): string[] {
const fits = (s: string) => s.length <= maxLen && packedNibbles(s) <= MAX_PACKED_NIBBLES;
const chunks: string[] = [];
let current = '';
for (const word of text.trim().split(/\s+/)) {
// A single oversized word: flush, then hard-split it greedily.
if (!fits(word)) {
if (current) { chunks.push(current); current = ''; }
let rest = word;
while (rest.length > 0) {
let take = Math.min(rest.length, maxLen);
while (take > 1 && !fits(rest.slice(0, take))) take--;
chunks.push(rest.slice(0, take));
rest = rest.slice(take);
}
continue;
}
const candidate = current ? `${current} ${word}` : word;
if (fits(candidate)) {
current = candidate;
} else {
if (current) chunks.push(current);
current = word;
}
}
if (current) chunks.push(current);
return chunks;
}
// Chat chunking lives in ./chunking so the gateway can share it without
// importing the full SDK; re-exported here for existing consumers and tests.
export { chunkMessage } from './chunking';
import { chunkMessage } from './chunking';
export class BotSDK {
readonly config: Required<SDKConfig>;
+219
View File
@@ -0,0 +1,219 @@
// The gateway's HTTP chat endpoints (GET /chat/:bot, POST /say/:bot) exist so
// chat tools can work in one bounded request/response instead of a fresh
// observer WebSocket per invocation - over TLS tunnels that per-invocation
// handshake stalls, and the old CLI could eat a shell tool's whole 120s
// timeout in silence. These tests run the real gateway as a subprocess with a
// scripted bot client on the other side.
import { afterAll, beforeAll, describe, expect, test } from 'bun:test';
import { join } from 'path';
const PORT = 20000 + Math.floor(Math.random() * 10000);
const BASE = `http://localhost:${PORT}`;
const REPO_ROOT = join(import.meta.dir, '..', '..');
let gateway: ReturnType<typeof Bun.spawn> | null = null;
let bot: WebSocket | null = null;
/** Say actions the fake bot client has received, in order. */
const receivedSays: string[] = [];
async function waitFor(check: () => boolean | Promise<boolean>, timeoutMs: number, what: string) {
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
if (await check()) return;
await new Promise(resolve => setTimeout(resolve, 100));
}
throw new Error(`Timed out waiting for ${what}`);
}
beforeAll(async () => {
gateway = Bun.spawn({
cmd: ['bun', join(REPO_ROOT, 'server', 'gateway', 'gateway.ts')],
env: { ...process.env, AGENT_PORT: String(PORT), LOGIN_SERVER: 'false' },
stdout: 'ignore',
stderr: 'ignore'
});
await waitFor(async () => {
try {
return (await fetch(`${BASE}/status`, { signal: AbortSignal.timeout(500) })).ok;
} catch {
return false;
}
}, 15000, 'gateway to boot');
// Fake bot client: registers as 'tbot', answers say actions unless the
// text contains 'noreply' (which simulates a dead game client).
bot = new WebSocket(`ws://localhost:${PORT}`);
bot.onopen = () => {
bot!.send(JSON.stringify({ type: 'connected', clientId: 'bot-test', username: 'tbot', maxMessageLength: 400 }));
bot!.send(JSON.stringify({
type: 'state',
state: {
tick: 1,
inGame: true,
player: { name: 'tbot', worldX: 3222, worldZ: 3218 },
gameMessages: [
{ type: 0, text: 'Welcome to RuneScape', sender: '', tick: 1, observationId: 1, fromSelf: false },
{ type: 2, text: 'meet me at the bank', sender: 'Partner', tick: 1, observationId: 1, fromSelf: false }
]
}
}));
};
bot.onmessage = (event) => {
const msg = JSON.parse(String(event.data));
if (msg.type === 'action' && msg.action?.type === 'say') {
receivedSays.push(msg.action.message);
if (msg.action.message.includes('noreply')) return;
bot!.send(JSON.stringify({
type: 'actionResult',
actionId: msg.actionId,
result: { success: true, message: 'Said', data: { finalText: msg.action.message } }
}));
}
};
await waitFor(async () => {
try {
const status: any = await (await fetch(`${BASE}/status/tbot`)).json();
return status.status === 'active';
} catch {
return false;
}
}, 10000, 'fake bot to register');
});
afterAll(() => {
try { bot?.close(); } catch {}
gateway?.kill();
});
describe('gateway HTTP chat endpoints', () => {
test('/status/:bot reports the relay cap alongside liveness', async () => {
const status: any = await (await fetch(`${BASE}/status/tbot`)).json();
expect(status.status).toBe('active');
expect(status.inGame).toBe(true);
expect(status.maxMessageLength).toBe(400);
});
test('GET /chat/:bot returns player chat and hides system lines by default', async () => {
const body: any = await (await fetch(`${BASE}/chat/tbot`)).json();
expect(body.messages.map((m: any) => m.text)).toEqual(['meet me at the bank']);
expect(body.botStatus).toBe('active');
const withSystem: any = await (await fetch(`${BASE}/chat/tbot?system=1`)).json();
expect(withSystem.messages.map((m: any) => m.text)).toEqual(['Welcome to RuneScape', 'meet me at the bank']);
});
test('GET /chat for an unknown bot answers empty, not an error', async () => {
const res = await fetch(`${BASE}/chat/ghost`);
expect(res.status).toBe(200);
const body: any = await res.json();
expect(body.messages).toEqual([]);
expect(body.botStatus).toBe('dead');
});
test('POST /say/:bot relays through the client and echoes the final text', async () => {
const res = await fetch(`${BASE}/say/tbot`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ text: 'heading to the anvil now' })
});
expect(res.status).toBe(200);
const body: any = await res.json();
expect(body.ok).toBe(true);
expect(body.results).toHaveLength(1);
expect(body.results[0].data.finalText).toBe('heading to the anvil now');
expect(receivedSays).toContain('heading to the anvil now');
});
test('POST /say splits long messages using the relay cap', async () => {
const before = receivedSays.length;
const text = Array.from({ length: 60 }, (_, i) => `word${i}`).join(' '); // > 400 chars
const res = await fetch(`${BASE}/say/tbot`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ text })
});
const body: any = await res.json();
expect(body.ok).toBe(true);
expect(body.results.length).toBeGreaterThan(1);
const chunks = receivedSays.slice(before);
expect(chunks.length).toBe(body.results.length);
for (const chunk of chunks) expect(chunk.length).toBeLessThanOrEqual(400);
expect(chunks.join(' ')).toBe(text);
}, 15000);
test('POST /say for an unconnected bot fails fast with 409', async () => {
const start = Date.now();
const res = await fetch(`${BASE}/say/ghost`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ text: 'anyone home?' })
});
expect(res.status).toBe(409);
const body: any = await res.json();
expect(body.error).toMatch(/not connected/i);
expect(Date.now() - start).toBeLessThan(2000);
});
test('POST /say answers a failed result within its budget when the client never replies', async () => {
const start = Date.now();
const res = await fetch(`${BASE}/say/tbot`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ text: 'noreply please', timeoutMs: 2000 })
});
expect(res.status).toBe(502);
const body: any = await res.json();
expect(body.ok).toBe(false);
expect(body.results[0].reason).toBe('timeout');
expect(Date.now() - start).toBeLessThan(5000);
}, 10000);
test('POST /say rejects a missing text body without dispatching', async () => {
const res = await fetch(`${BASE}/say/tbot`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({})
});
expect(res.status).toBe(400);
});
});
describe('chat CLI over HTTP', () => {
async function runCli(args: string[]): Promise<{ exitCode: number; stdout: string; stderr: string }> {
const proc = Bun.spawn({
cmd: ['bun', join(REPO_ROOT, 'sdk', 'chat.ts'), ...args, '--server', `localhost:${PORT}`],
cwd: REPO_ROOT,
stdout: 'pipe',
stderr: 'pipe'
});
const [stdout, stderr, exitCode] = await Promise.all([
new Response(proc.stdout).text(),
new Response(proc.stderr).text(),
proc.exited
]);
return { exitCode, stdout, stderr };
}
test('sends a message end-to-end', async () => {
const { exitCode, stdout } = await runCli(['tbot', 'cli says hi']);
expect(exitCode).toBe(0);
expect(stdout).toContain('Sent: cli says hi');
expect(receivedSays).toContain('cli says hi');
}, 20000);
test('reads chat end-to-end', async () => {
const { exitCode, stdout } = await runCli(['tbot', '--limit', '50']);
expect(exitCode).toBe(0);
expect(stdout).toContain('Partner: meet me at the bank');
}, 20000);
test('reports a dead relay within the chunk budget instead of stalling', async () => {
const start = Date.now();
const { exitCode, stderr } = await runCli(['tbot', 'noreply again']);
expect(exitCode).toBe(1);
expect(stderr).toMatch(/timed out/i);
expect(Date.now() - start).toBeLessThan(20000);
}, 30000);
});
+175 -4
View File
@@ -10,6 +10,9 @@ import type {
SyncToSDKMessage,
SDKConnectionMode
} from './types';
import { PLAYER_CHAT_TYPES } from './types';
import { ChatHistory } from '../../sdk/chat-history';
import { chunkMessage } from '../../sdk/chunking';
const GATEWAY_PORT = parseInt(process.env.AGENT_PORT || '7780');
@@ -129,7 +132,7 @@ interface BotSession {
// Session status for diagnostics
type SessionStatus = 'active' | 'stale' | 'dead';
const STALE_THRESHOLD_MS = 8000; // 30 seconds without state = stale
const STALE_THRESHOLD_MS = 8000; // 8 seconds without state = stale (bots publish ~every tick)
function getSessionStatus(session: BotSession): SessionStatus {
if (!session.ws) return 'dead';
@@ -174,6 +177,26 @@ const botSessions = new Map<string, BotSession>(); // username -> BotSessio
const sdkSessions = new Map<string, SDKSession>(); // sessionId -> SDKSession
const wsToType = new Map<any, { type: 'bot' | 'sdk'; id: string }>();
// Per-bot accumulated chat, fed from state snapshots (same merge logic the SDK
// uses). Keyed by username so history survives bot reconnects/takeovers.
// Backs the HTTP GET /chat/:username endpoint, which lets chat tools read
// recent messages with a single bounded request instead of opening an observer
// WebSocket and waiting for a state frame.
const chatHistories = new Map<string, ChatHistory>();
function chatHistoryFor(username: string): ChatHistory {
let history = chatHistories.get(username);
if (!history) {
history = new ChatHistory();
chatHistories.set(username, history);
}
return history;
}
// Actions dispatched from HTTP requests (POST /say/:username) awaiting their
// actionResult from the bot client. Keyed by actionId; entries self-expire.
const pendingHttpActions = new Map<string, { resolve: (result: any) => void; timeout: ReturnType<typeof setTimeout> }>();
let httpActionSeq = 0;
// Monotonic counter so server-generated session ids can never collide, even
// within the same millisecond.
let sdkSessionSeq = 0;
@@ -351,6 +374,18 @@ const SyncModule = {
const actionId = message.actionId || session.currentActionId || undefined;
console.log(`[Gateway] [${session.username}] Action result: ${message.result.success ? 'success' : 'failed'} - ${message.result.message}`);
// Actions dispatched over HTTP resolve their waiting request here.
if (actionId) {
const pendingHttp = pendingHttpActions.get(actionId);
if (pendingHttp) {
pendingHttpActions.delete(actionId);
clearTimeout(pendingHttp.timeout);
pendingHttp.resolve(message.result);
session.currentActionId = null;
return;
}
}
for (const sdkSession of this.getSDKSessionsForBot(session.username)) {
this.sendToSDK(sdkSession, {
type: 'sdk_action_result',
@@ -365,6 +400,9 @@ const SyncModule = {
if (message.type === 'state' && message.state) {
session.lastState = message.state;
session.lastStateReceivedAt = Date.now();
if (message.state.gameMessages?.length) {
chatHistoryFor(session.username).record(message.state.gameMessages);
}
for (const sdkSession of this.getSDKSessionsForBot(session.username)) {
this.sendToSDK(sdkSession, {
type: 'sdk_state',
@@ -602,6 +640,47 @@ const SyncModule = {
}
};
// ============ HTTP Chat Helpers ============
/**
* Auth for the HTTP chat endpoints - same per-bot credential check as
* sdk_connect. Password comes from the JSON body, an Authorization: Bearer
* header, or a ?password= query param (a no-op unless LOGIN_SERVER is on,
* matching the WebSocket paths).
*/
async function authenticateHttpRequest(req: Request, url: URL, username: string, bodyPassword?: unknown): Promise<{ success: boolean; error?: string }> {
const header = req.headers.get('authorization');
const bearer = header?.startsWith('Bearer ') ? header.slice(7) : undefined;
const password = typeof bodyPassword === 'string' && bodyPassword !== ''
? bodyPassword
: (bearer ?? url.searchParams.get('password') ?? '');
return authenticateSDK(username, password);
}
/**
* Dispatch one say chunk to the bot client and wait (bounded) for its
* actionResult. Never rejects: a client that doesn't answer inside the budget
* yields a failed result, so the HTTP handler always responds.
*/
function dispatchHttpSay(botSession: BotSession, text: string, timeoutMs: number): Promise<{ success: boolean; message: string; data?: any; reason?: string }> {
const actionId = `http-${++httpActionSeq}-${Math.random().toString(36).slice(2, 8)}`;
return new Promise(resolve => {
const timeout = setTimeout(() => {
pendingHttpActions.delete(actionId);
resolve({ success: false, message: `Say timed out after ${timeoutMs}ms (game client did not answer)`, reason: 'timeout' });
}, timeoutMs);
pendingHttpActions.set(actionId, { resolve, timeout });
// Also track on the session so clients that don't echo actionId still route back.
botSession.currentActionId = actionId;
SyncModule.sendToBot(botSession, {
type: 'action',
action: { type: 'say', message: text, reason: 'HTTP chat' },
actionId,
actionTimeoutMs: timeoutMs
});
});
}
// ============ Message Router ============
/**
@@ -679,7 +758,7 @@ console.log(`[Gateway] Starting Gateway Service on port ${GATEWAY_PORT}...`);
const server = Bun.serve({
port: GATEWAY_PORT,
fetch(req, server) {
async fetch(req, server) {
const url = new URL(req.url);
// WebSocket upgrade
@@ -692,8 +771,8 @@ const server = Bun.serve({
// CORS headers
const corsHeaders = {
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': 'GET, OPTIONS',
'Access-Control-Allow-Headers': 'Content-Type'
'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
'Access-Control-Allow-Headers': 'Content-Type, Authorization'
};
if (req.method === 'OPTIONS') {
@@ -718,6 +797,7 @@ const server = Bun.serve({
status: botSession ? getSessionStatus(botSession) : 'dead',
inGame: isConnected ? (botSession?.lastState?.inGame || false) : false,
stateAge,
maxMessageLength: botSession?.maxMessageLength ?? null,
controllers,
observers,
player: isConnected && botSession?.lastState?.player ? {
@@ -732,6 +812,95 @@ const server = Bun.serve({
});
}
const jsonResponse = (body: unknown, status = 200) => new Response(JSON.stringify(body, null, 2), {
status,
headers: { 'Content-Type': 'application/json', ...corsHeaders }
});
// Read recent chat: GET /chat/:username?limit=20&system=1
//
// Serves from the gateway's accumulated per-bot history, so one bounded
// HTTP request replaces the observer-WebSocket dance (connect, wait for
// a state frame, read, disconnect). Over TLS tunnels that dance is a
// real handshake per invocation and the dominant source of chat-tool
// flakiness.
const chatMatch = url.pathname.match(/^\/chat\/(.+)$/);
if (chatMatch && chatMatch[1] && req.method === 'GET') {
const username = decodeURIComponent(chatMatch[1]);
const auth = await authenticateHttpRequest(req, url, username);
if (!auth.success) {
return jsonResponse({ error: `Authentication failed: ${auth.error}` }, 401);
}
const limitRaw = parseInt(url.searchParams.get('limit') || '20', 10);
const limit = Number.isFinite(limitRaw) ? limitRaw : 20;
const systemParam = url.searchParams.get('system');
const includeSystem = systemParam === '1' || systemParam === 'true';
const types = new Set<number>(includeSystem ? [0, ...PLAYER_CHAT_TYPES] : PLAYER_CHAT_TYPES);
const all = (chatHistories.get(username)?.all() ?? []).filter(m => types.has(m.type));
const botSession = botSessions.get(username);
return jsonResponse({
messages: limit > 0 ? all.slice(-limit) : all,
botStatus: botSession ? getSessionStatus(botSession) : 'dead'
});
}
// Send chat: POST /say/:username body: {"text": "...", "password"?: "...", "timeoutMs"?: n}
//
// Relays through the bot's game client like observer 'say' does, but as
// a single request/response with a hard per-chunk deadline - a caller
// can trust that this either answers or fails within its budget.
const sayMatch = url.pathname.match(/^\/say\/(.+)$/);
if (sayMatch && sayMatch[1] && req.method === 'POST') {
const username = decodeURIComponent(sayMatch[1]);
let body: any;
try {
body = await req.json();
} catch {
return jsonResponse({ error: 'Body must be JSON: {"text": "..."}' }, 400);
}
const text = typeof body?.text === 'string' ? body.text.trim() : '';
if (!text) {
return jsonResponse({ error: 'Missing "text"' }, 400);
}
const auth = await authenticateHttpRequest(req, url, username, body?.password);
if (!auth.success) {
return jsonResponse({ error: `Authentication failed: ${auth.error}` }, 401);
}
const botSession = botSessions.get(username);
if (!botSession || !botSession.ws) {
return jsonResponse({
error: `Bot not connected - no game client is logged in as '${username}', so there is nothing to relay chat through`
}, 409);
}
const timeoutRaw = Number(body?.timeoutMs);
const perChunkTimeout = Math.min(Math.max(Number.isFinite(timeoutRaw) && timeoutRaw > 0 ? timeoutRaw : 10000, 2000), 30000);
const chunks = chunkMessage(text, botSession.maxMessageLength ?? 80);
const results: any[] = [];
let ok = true;
for (let i = 0; i < chunks.length; i++) {
const result = await dispatchHttpSay(botSession, chunks[i]!, perChunkTimeout);
results.push({ chunk: chunks[i], ...result });
if (!result.success) {
// The relay is broken; burning the timeout again per
// remaining chunk helps nobody.
ok = false;
for (let j = i + 1; j < chunks.length; j++) {
results.push({ chunk: chunks[j], success: false, message: 'Skipped: previous chunk failed' });
}
break;
}
if (i < chunks.length - 1) {
await new Promise(resolve => setTimeout(resolve, 600)); // ~1 tick so chunks don't collide
}
}
return jsonResponse({ ok, results }, ok ? 200 : 502);
}
// Status endpoint (admin/debugging - more detailed)
if (url.pathname === '/' || url.pathname === '/status') {
const bots: Record<string, any> = {};
@@ -769,6 +938,8 @@ const server = Bun.serve({
Endpoints:
- GET /status All connections status
- GET /status/:username Per-bot status (controllers, observers)
- GET /chat/:username Recent chat (?limit=20&system=1)
- POST /say/:username Send chat ({"text": "..."}); relayed through the bot's game client
WebSocket:
- ws://localhost:${GATEWAY_PORT} Bot/SDK connections