Batch authoritative snapshots and bound rejected retreat fallback

ale/azerothai_survival.lua

@@ -102,7 +102,13 @@ local function sample()
  local now=os.time();refresh(now);restoreExpiredHolds(now)
  for _,p in pairs(GetPlayersInWorld() or {}) do
   local previous=observed[p:GetGUIDLow()]
-  if previous then outcome(p,previous,now) end
+  if previous then
+   if previous.nativeFleeUntil and now>=previous.nativeFleeUntil then
+    if p:GetMovementType()==9 or p:GetMovementType()==13 then p:MoveExpire() end
+    previous.nativeFleeUntil=nil
+   end
+   outcome(p,previous,now)
+  end
   if managed(p) then
    local id=p:GetGUIDLow();local ai=p:GetBotAI();local r=observed[id] or {};observed[id]=r
    local previousAction,previousAt=r.action,r.actionAt or 0
@@ -174,6 +180,17 @@ local function sample()
       r.dangerAt=now;r.retreatAt=now;r.action='retreat requested';r.actionAt=now
       if p:IsCasting() then p:StopSpellCast() end
       r.accepted=ai:DoSpecificAction('flee',true)
+      -- The bot action may reject fleeing when its target/context is missing.
+      -- Bound a native pathfinding attempt; never teleport or clear combat/threat.
+      if not r.accepted and not p:IsMoving() and now-(r.nativeFleeAt or 0)>=30 then
+       local threat=enemy
+       if not threat then for _,attacker in pairs(p:GetAttackers()) do threat=attacker;break end end
+       if threat then
+        p:AttackStop();p:MoveFleeing(threat,8000)
+        r.nativeFleeAt=now;r.nativeFleeUntil=now+8
+        audit(p,'retreat-path-request','Playerbot flee rejected; bounded native flee path requested')
+       end
+      end
       audit(p,'retreat-request', 'accepted='..tostring(r.accepted)..'; hp='..math.floor(hp)..'; dangerous='..tostring(overmatched)..'; attackers='..attackers..'; nearbySupport='..support)
       if not r.retreat then r.retreat={at=now,map=p:GetMapId(),x=p:GetX(),y=p:GetY()} end
      elseif avoid and now-(r.avoidAt or 0)>=8 then

app/api/bridge/ale/snapshots/route.ts

@@ -0,0 +1,26 @@
+import {env} from 'cloudflare:workers';
+import {z} from 'zod';
+import {requireAleBridge} from '@/core/ale-auth';
+import {recordAleEvent} from '@/core/ale-store';
+import {queueSocialEncounter} from '@/core/conversation-store';
+import {D1Store} from '@/core/store';
+import {applyAuthoritativeAleSnapshots} from '@/core/service';
+import type {RuntimeConfig} from '@/core/types';
+const schema=z.object({payloads:z.array(z.record(z.unknown())).min(1).max(64)}).strict();
+export async function POST(request:Request){
+ try{
+  requireAleBridge(request,env as unknown as RuntimeConfig);
+  const text=await request.text();
+  if(text.length>512000)return Response.json({error:'Snapshot batch too large'},{status:413});
+  const {payloads}=schema.parse(JSON.parse(text));
+  const result=await applyAuthoritativeAleSnapshots(new D1Store(),payloads);
+  for(const payload of payloads){
+   await recordAleEvent(env.DB!,'player.snapshot',payload);
+   if(payload.roster===true)await queueSocialEncounter(env.DB!,payload);
+  }
+  return Response.json(result);
+ }catch(e){
+  const error=e instanceof Error?e.message:'Snapshot batch failure';
+  return Response.json({error},{status:error.startsWith('Unauthorized')?401:400});
+ }
+}

core/service.ts

@@ -70,6 +70,16 @@ function observedHardcoreDeath(s:State,b:Bot,observation:Observation){
  return true;
 }
 
+// Apply one transport batch atomically; one large state write instead of one per bot.
+export async function applyAuthoritativeAleSnapshots(store:StateStore,payloads:Record<string,unknown>[]){
+ const initial=await store.load(),revision=initial.revision;
+ let staged=structuredClone(initial),dirty=false;
+ const batchStore:StateStore={load:async()=>staged,save:async(next)=>{staged=next;dirty=true;}};
+ for(const payload of payloads)await applyAuthoritativeAleEvent(batchStore,'player.snapshot',payload);
+ if(dirty)await store.save(staged,revision);
+ return {ok:true,count:payloads.length};
+}
+
 export async function applyAuthoritativeAleEvent(store:StateStore,kind:string,payload:Record<string,unknown>){
  let s=await store.load();
  const revision=s.revision;

scripts/ale-db-relay.mjs

@@ -99,14 +99,22 @@ async function deliverEvents(){
  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))}`);
+ ORDER BY id ASC LIMIT 64`);
+ if(fresh.length){
+  const payloads=fresh.map(([, ,hex])=>JSON.parse(Buffer.from(hex,'hex').toString('utf8')));
+  const response=await backend('/api/bridge/ale/snapshots',{
+   method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({payloads})
+  });
+  if(!response.ok)throw new Error(`snapshot batch HTTP ${response.status}: ${await response.text()}`);
+  mysql(`UPDATE acore_auth.azerothai_event_outbox SET delivered_at=NOW(6) WHERE id IN (${fresh.map(row=>Number(row[0])).join(',')})`);
+ }
  const batch=rows(`SELECT id,HEX(kind),HEX(payload)
  FROM acore_auth.azerothai_event_outbox
  WHERE delivered_at IS NULL
  ${fresh.length?`AND id NOT IN (${fresh.map(row=>Number(row[0])).join(',')})`:''}
  ORDER BY CASE WHEN kind IN ('player.permanent-death','bridge.ready') THEN 0 WHEN kind IN ('player.whisper','player.chat') AND created_at >= NOW() - INTERVAL 90 SECOND THEN 1 ELSE 2 END, id
- LIMIT ${Math.max(1,EVENT_BATCH-fresh.length)}`);
+ LIMIT ${EVENT_BATCH}`);
- for(const [id,kindHex,payloadHex] of [...fresh,...batch]){
+ for(const [id,kindHex,payloadHex] of batch){
   const kind=Buffer.from(kindHex||'','hex').toString('utf8');
   const payloadText=Buffer.from(payloadHex||'','hex').toString('utf8');
   let payload;

tests/ale.test.cjs

@@ -34,3 +34,14 @@ test('surface inventory includes every documented ALE class family',()=>{
  assert.equal(ALE_SURFACE.playerbotClasses.length,7);
  assert.equal(ALE_SURFACE.modelAccess,'none');
 });
+
+test('snapshot batch saves once and does not commit a partial invalid roster',async()=>{
+ const {applyAuthoritativeAleSnapshots}=require('../.test-build/service.js');
+ const {seed}=require('../.test-build/seed.js');
+ let state=seed(),writes=0;
+ const store={load:async()=>structuredClone(state),save:async(s,r)=>{assert.equal(r,state.revision);writes++;state=s;}};
+ await applyAuthoritativeAleSnapshots(store,[{roster:true,observedAt:100,characterGuids:[]},{roster:true,observedAt:101,characterGuids:[]}]);
+ assert.equal(writes,1);assert.equal(state.aleRosterAt,101);
+ await assert.rejects(()=>applyAuthoritativeAleSnapshots(store,[{roster:true,observedAt:102,characterGuids:[]},{roster:true,observedAt:103,characterGuids:[-1]}]));
+ assert.equal(writes,1);assert.equal(state.aleRosterAt,101);
+});

tests/survival-lua.test.py

@@ -15,7 +15,7 @@ function make(id,name)
  function p:GetGUIDLow()return self.id end;function p:GetName()return self.name end;function p:GetAccountId()return 10+self.id end
  function p:IsPlayerbot()return not self.human end;function p:IsDead()return self.dead end;function p:GetBotAI()return self.ai end
  function p:GetPowerType()return self.powerType or 1 end;function p:GetPowerPct()return self.mana or 100 end;function p:GetAttackers()return self.attackers or {} end;function p:GetHealthPct()return self.hp end;function p:IsInCombat()return self.combat end
- function p:HasUnitState(state)assert(state==0x00000100);return self.taxi end;function p:IsCasting()return self.cast end;function p:IsRooted()return false end
+ function p:HasUnitState(state)assert(state==0x00000100);return self.taxi end;function p:IsMoving()return self.moving end;function p:GetMovementType()return self.motion or 0 end;function p:MoveExpire()self.expired=true;self.motion=0 end;function p:MoveFleeing(t,ms)self.fleeTarget=t;self.motion=9;self.fleeDuration=ms end;function p:IsCasting()return self.cast end;function p:IsRooted()return false end
  function p:GetX()return self.x or 0 end;function p:GetY()return 0 end
  function p:GetUnfriendlyUnitsInRange()return self.friendly and {} or self.selected and {self.selected} or {} end;function p:GetSelection()return self.selected end;function p:StopSpellCast()self.cast=false;self.interrupted=true end
  function p:AttackStop()self.attackStopped=true end;function p:MoveStop()self.waited=true end
@@ -69,3 +69,7 @@ print('Cohesion suppresses competing work, keeps catch-up hysteresis, yields to
 l=new();l.execute("A.strategies={stay=true};step(0);assert(not A.strategies.stay,'Fully recovered autonomous resident must not remain in stay')")
 l=new();l.execute("group();B.distance=50;step(0);step(40);assert(B.master==nil,'No-progress regroup must release its owned master');assert(not AzerothAISurvivalOverride(B));step(30);assert(B.master==nil,'Backoff must prevent immediate retry')")
 print('Recovered stay cleared and non-progressing regroup bounded')
+
+l=new();l.execute("A.combat=true;A.hp=20;A.attackers={B};function A.ai:DoSpecificAction(s)table.insert(calls,s);return false end;step(0);assert(A.fleeTarget==B and A.fleeDuration==8000);step(8);assert(A.expired);A.fleeTarget=nil;step(8);assert(A.fleeTarget==nil,'Native fallback must honor cooldown')")
+l=new();l.execute("A.combat=true;A.hp=20;A.controlled=true;A.attackers={B};function A.ai:DoSpecificAction()return false end;step(0);assert(A.fleeTarget==nil)")
+print('Rejected flee fallback is bounded, cooled down, and preserves human control')