Restore live autonomy with direct runtime and fresh snapshot capacity
This commit is contained in:
+19
-7
@@ -103,8 +103,8 @@ export async function runLiveAutonomyTick(
|
||||
options:{forceBotId?:string;now?:number}={}
|
||||
){
|
||||
const now=options.now??Date.now();
|
||||
const s=await store.load();
|
||||
const revision=s.revision;
|
||||
let s=await store.load();
|
||||
let revision=s.revision;
|
||||
const events=await recentAleEvents(db,400);
|
||||
|
||||
const candidates=s.bots
|
||||
@@ -139,29 +139,41 @@ export async function runLiveAutonomyTick(
|
||||
return {ok:true,ran:false,reason:options.forceBotId?'No recent authoritative observation for that managed Playerbot':'No managed Playerbot is due with a recent authoritative observation'};
|
||||
}
|
||||
|
||||
const bot=pick.bot;
|
||||
let bot=pick.bot;
|
||||
const observation=observationFrom(bot,pick.obs,events);
|
||||
const brain=providerForBot(bot,s.settings,config);
|
||||
let decision:Decision;
|
||||
let usedFallback=false;
|
||||
try{
|
||||
decision=await brain.provider.decide(bot,observation);
|
||||
}catch(error){
|
||||
if(brain.route.provider==='deterministic')throw error;
|
||||
decision=await new DeterministicProvider().decide(bot,observation);
|
||||
realmEvent(s,'system',`${bot.name} LLM route unavailable; used deterministic fallback for this heartbeat.`,bot.id);
|
||||
usedFallback=true;
|
||||
}
|
||||
// Model inference can take seconds while authoritative events update state.
|
||||
// Commit the decision against current state, not the pre-inference revision.
|
||||
const botId=bot.id;
|
||||
s=await store.load();
|
||||
revision=s.revision;
|
||||
const currentBot=s.bots.find(b=>b.id===botId);
|
||||
if(!currentBot||currentBot.status==='dead'||currentBot.status==='retired'||!currentBot.autonomy?.managed){
|
||||
return {ok:true,ran:false,reason:'Resident is no longer eligible for an autonomous intention'};
|
||||
}
|
||||
bot=currentBot;
|
||||
if(usedFallback)realmEvent(s,'system',`${bot.name} LLM route unavailable; used deterministic fallback for this heartbeat.`,bot.id);
|
||||
const source=compileLivePlayerbotIntent(bot.name,decision);
|
||||
const queued=await queueAleCommand(db,source);
|
||||
|
||||
bot.lastDecision=decision;
|
||||
bot.decisions++;
|
||||
bot.goal=decision.goal;
|
||||
bot.level=observation.level;
|
||||
bot.zone=observation.zone;
|
||||
const newerObservation=Date.parse(bot.autonomy?.observedAt??'')>Date.parse(pick.obs.at);
|
||||
if(!newerObservation){bot.level=observation.level;bot.zone=observation.zone;}
|
||||
bot.autonomy={
|
||||
...(bot.autonomy??{}),
|
||||
managed:true,
|
||||
observedAt:pick.obs.at,
|
||||
observedAt:newerObservation?bot.autonomy?.observedAt:pick.obs.at,
|
||||
lastIntentAt:new Date(now).toISOString(),
|
||||
nextIntentAt:new Date(now+nextIntervalMs(bot)).toISOString(),
|
||||
lastAleCommandId:queued.id
|
||||
|
||||
+2
-2
@@ -9,11 +9,11 @@
|
||||
"install:ci": "bash scripts/install-pnpm.sh",
|
||||
"dev": "node scripts/run-framework.mjs dev",
|
||||
"build": "node scripts/run-framework.mjs build",
|
||||
"start": "node --import ./scripts/sites-env.mjs ./node_modules/wrangler/bin/wrangler.js dev --config dist/server/wrangler.json --local --persist-to .wrangler/state --ip 127.0.0.1 --inspector-port 0 --env-file .dev.vars",
|
||||
"start": "node scripts/start-local-runtime.mjs",
|
||||
"lint": "eslint . --ignore-pattern dist --ignore-pattern .next",
|
||||
"db:generate": "drizzle-kit generate",
|
||||
"bench:brains": "node scripts/benchmark-brains.mjs",
|
||||
"start:lan": "node --import ./scripts/sites-env.mjs ./node_modules/wrangler/bin/wrangler.js dev --config dist/server/wrangler.json --local --persist-to .wrangler/state --ip 0.0.0.0 --port 8787 --inspector-port 0 --env-file .dev.vars"
|
||||
"start:lan": "node scripts/start-local-runtime.mjs --lan"
|
||||
},
|
||||
"dependencies": {
|
||||
"@base-ui/react": "^1.7.0",
|
||||
|
||||
@@ -93,12 +93,20 @@ function rows(sql){
|
||||
}
|
||||
|
||||
async function deliverEvents(){
|
||||
// Reserve capacity for current snapshots so historical events cannot starve
|
||||
// live decisions. The remaining slots continue draining history in order.
|
||||
const fresh=rows(`SELECT id,HEX(kind),HEX(payload)
|
||||
FROM acore_auth.azerothai_event_outbox
|
||||
WHERE delivered_at IS NULL AND kind='player.snapshot'
|
||||
AND created_at >= NOW() - INTERVAL 90 SECOND
|
||||
ORDER BY id ASC LIMIT ${Math.max(1,Math.floor(EVENT_BATCH/2))}`);
|
||||
const batch=rows(`SELECT id,HEX(kind),HEX(payload)
|
||||
FROM acore_auth.azerothai_event_outbox
|
||||
WHERE delivered_at IS NULL
|
||||
ORDER BY CASE WHEN kind IN ('player.whisper','player.chat') THEN 0 ELSE 1 END, id
|
||||
LIMIT ${EVENT_BATCH}`);
|
||||
for(const [id,kindHex,payloadHex] of batch){
|
||||
${fresh.length?`AND id NOT IN (${fresh.map(row=>Number(row[0])).join(',')})`:''}
|
||||
ORDER BY CASE WHEN kind IN ('player.whisper','player.chat') AND created_at >= NOW() - INTERVAL 90 SECOND THEN 0 ELSE 1 END, id
|
||||
LIMIT ${Math.max(1,EVENT_BATCH-fresh.length)}`);
|
||||
for(const [id,kindHex,payloadHex] of [...fresh,...batch]){
|
||||
const kind=Buffer.from(kindHex||'','hex').toString('utf8');
|
||||
const payloadText=Buffer.from(payloadHex||'','hex').toString('utf8');
|
||||
let payload;
|
||||
@@ -116,7 +124,7 @@ async function deliverEvents(){
|
||||
mysql(`UPDATE acore_auth.azerothai_event_outbox SET delivered_at=NOW(6) WHERE id=${Number(id)}`);
|
||||
if(EVENT_GAP_MS)await sleep(EVENT_GAP_MS);
|
||||
}
|
||||
return batch.length;
|
||||
return fresh.length+batch.length;
|
||||
}
|
||||
|
||||
async function deliverResults(){
|
||||
@@ -136,6 +144,14 @@ async function deliverResults(){
|
||||
|
||||
function cleanup(){
|
||||
mysql(`
|
||||
CREATE TABLE IF NOT EXISTS acore_auth.azerothai_snapshot_archive LIKE acore_auth.azerothai_event_outbox;
|
||||
INSERT IGNORE INTO acore_auth.azerothai_snapshot_archive
|
||||
SELECT * FROM acore_auth.azerothai_event_outbox
|
||||
WHERE delivered_at IS NULL AND kind='player.snapshot'
|
||||
AND created_at < NOW() - INTERVAL 90 SECOND;
|
||||
UPDATE acore_auth.azerothai_event_outbox e
|
||||
JOIN acore_auth.azerothai_snapshot_archive a ON a.id=e.id
|
||||
SET e.delivered_at=NOW(6) WHERE e.delivered_at IS NULL;
|
||||
UPDATE acore_auth.azerothai_event_outbox
|
||||
SET delivered_at=NOW(6)
|
||||
WHERE delivered_at IS NULL
|
||||
@@ -151,10 +167,10 @@ console.log(`AzerothAI ALE DB relay started: ${BASE}, poll ${POLL_MS}ms`);
|
||||
|
||||
while(!stopping){
|
||||
try{
|
||||
if(Date.now()-lastCleanup>60_000){cleanup();lastCleanup=Date.now();}
|
||||
await deliverEvents();
|
||||
await deliverResults();
|
||||
await pullCommand();
|
||||
if(Date.now()-lastCleanup>60_000){cleanup();lastCleanup=Date.now();}
|
||||
await sleep(POLL_MS);
|
||||
}catch(error){
|
||||
console.error(new Date().toISOString(),error instanceof Error?error.message:String(error));
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
// Run the built worker without Wrangler's development proxy or file watcher.
|
||||
import './sites-env.mjs';
|
||||
import {createRequire} from 'node:module';
|
||||
import path from 'node:path';
|
||||
import {readdirSync} from 'node:fs';
|
||||
import {unstable_getMiniflareWorkerOptions} from 'wrangler';
|
||||
|
||||
const require = createRequire(import.meta.url);
|
||||
const wranglerRequire = createRequire(require.resolve('wrangler'));
|
||||
const {Miniflare} = wranglerRequire('miniflare');
|
||||
const config = path.resolve('dist/server/wrangler.json');
|
||||
const {workerOptions, externalWorkers, main} = unstable_getMiniflareWorkerOptions(config);
|
||||
if (!main) throw new Error('Build the application before starting the local runtime');
|
||||
const root = path.dirname(main);
|
||||
const modules = readdirSync(root, {recursive: true})
|
||||
.filter(file => /\.(m?js)$/.test(file))
|
||||
.map(file => ({type: 'ESModule', path: path.join(root, file)}));
|
||||
modules.sort((a, b) => a.path === main ? -1 : b.path === main ? 1 : 0);
|
||||
const runtime = new Miniflare({
|
||||
host: process.env.RUNTIME_HOST || (process.argv.includes('--lan') ? '0.0.0.0' : '127.0.0.1'),
|
||||
port: Number(process.env.RUNTIME_PORT || 8787),
|
||||
d1Persist: path.resolve('.wrangler/state/v3/d1'),
|
||||
workers: [{...workerOptions, modules, modulesRoot: root}, ...externalWorkers],
|
||||
});
|
||||
console.log(`AzerothAI local runtime ready: ${await runtime.ready}`);
|
||||
for (const signal of ['SIGINT', 'SIGTERM']) {
|
||||
process.once(signal, async () => { await runtime.dispose(); process.exit(0); });
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
import {spawnSync} from 'node:child_process';
|
||||
import {writeFileSync,mkdirSync} from 'node:fs';
|
||||
const result=spawnSync(process.execPath,['node_modules/typescript/bin/tsc','core/service.ts','core/seed.ts','core/ale.ts','core/conversation.ts','--types','@cloudflare/workers-types,node','--outDir','.test-build','--module','commonjs','--target','es2022','--moduleResolution','node','--esModuleInterop','--skipLibCheck','--strict'],{stdio:'inherit'});
|
||||
const result=spawnSync(process.execPath,['node_modules/typescript/bin/tsc','core/service.ts','core/seed.ts','core/ale.ts','core/conversation.ts','core/autonomy.ts','--types','@cloudflare/workers-types,node','--outDir','.test-build','--module','commonjs','--target','es2022','--moduleResolution','node','--esModuleInterop','--skipLibCheck','--strict'],{stdio:'inherit'});
|
||||
if(result.status)process.exit(result.status);
|
||||
mkdirSync('.test-build',{recursive:true});writeFileSync('.test-build/package.json','{"type":"commonjs"}');
|
||||
process.exit(spawnSync(process.execPath,['--test','tests/core.test.cjs','tests/hardcore.test.cjs','tests/living-world.test.cjs','tests/ale.test.cjs','tests/brains.test.cjs','tests/conversation.test.cjs'],{stdio:'inherit'}).status??1);
|
||||
process.exit(spawnSync(process.execPath,['--test','tests/core.test.cjs','tests/hardcore.test.cjs','tests/living-world.test.cjs','tests/ale.test.cjs','tests/brains.test.cjs','tests/conversation.test.cjs','tests/autonomy.test.cjs'],{stdio:'inherit'}).status??1);
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
const {test}=require('node:test');
|
||||
const assert=require('node:assert/strict');
|
||||
const ale=require('../.test-build/ale-store.js');
|
||||
const llm=require('../.test-build/llm/index.js');
|
||||
const {runLiveAutonomyTick}=require('../.test-build/autonomy.js');
|
||||
|
||||
test('inference preserves intervening observations and cancels after retirement',async()=>{
|
||||
const now=Date.now();
|
||||
const bot={id:'hero',name:'Zaku',status:'online',tier:'hero',decisions:0,
|
||||
personality:{activities:['explore']},autonomy:{managed:true},level:5,zone:'Old zone'};
|
||||
let state={revision:1,bots:[bot],events:[],sequence:0,settings:{}};
|
||||
const store={load:async()=>structuredClone(state),save:async(s,revision)=>{
|
||||
assert.equal(revision,state.revision);state=s;
|
||||
}};
|
||||
ale.recentAleEvents=async()=>[{at:new Date(now).toISOString(),kind:'player.snapshot',payload:{name:'Zaku',isBot:true,level:5,zoneName:'Old zone'}}];
|
||||
let queued=0,retire=false;
|
||||
ale.queueAleCommand=async()=>{queued++;return {id:'intent'};};
|
||||
llm.providerForBot=()=>({route:{provider:'ollama',model:'test'},provider:{decide:async()=>{
|
||||
state.revision++;
|
||||
state.bots[0].level=6;state.bots[0].zone='New zone';
|
||||
state.bots[0].autonomy.observedAt=new Date(now+1000).toISOString();
|
||||
if(retire)state.bots[0].status='retired';
|
||||
return {action:'explore',goal:'Explore nearby',reason:'Curious'};
|
||||
}}});
|
||||
assert.equal((await runLiveAutonomyTick(store,{}, {},{now})).ran,true);
|
||||
assert.equal(state.bots[0].level,6);assert.equal(state.bots[0].zone,'New zone');
|
||||
assert.equal(queued,1);
|
||||
retire=true;
|
||||
assert.equal((await runLiveAutonomyTick(store,{}, {},{now,forceBotId:'hero'})).ran,false);
|
||||
assert.equal(queued,1);
|
||||
});
|
||||
Reference in New Issue
Block a user