"""Isolated differential evidence for the original current generic change tracker.
No live process is opened. Manager, registry, entity, storage and queues are
private writable Unicorn data. The original PE remains read-only.
"""
import importlib.util
import json
import struct
import sys
from pathlib import Path
sys.path.insert(0, 'src/loader/runtime/profile-tools/dev/tools/enshrouded')
from discovery.pe import PE
from unicorn.x86_const import (UC_X86_REG_RCX,UC_X86_REG_RDX,UC_X86_REG_R8,
 UC_X86_REG_R9,UC_X86_REG_RSP,UC_X86_REG_RIP)
spec=importlib.util.spec_from_file_location('attribute_oracle', 'src/loader/runtime/profile-tools/dev/tools/enshrouded/verify-attribute-vm.py')
vm=importlib.util.module_from_spec(spec)
spec.loader.exec_module(vm)
pe=PE(r'C:\Program Files (x86)\Steam\steamapps\common\Enshrouded\enshrouded.exe')
assert pe.sha256=='4d6d43ea4a91e71c82d60b045dde932179c99e32e40657efe374688b2933589a'
o=vm.Oracle(pe)
u=o.uc
manager=0x70000000
registry=0x72000000
storage=0x73000000
entity=0x75000000
queue_a=0x76000000
queue_b=0x76001000
context=manager+0xebc030
index=37
for base,size in ((manager,0xf00000),(registry,0x2000),(storage,0x1000),(entity,0x1000),(queue_a,0x2000)):
 u.mem_map(base,size)
def write(address,fmt,*values): u.mem_write(address,struct.pack(fmt,*values))
def read(address,fmt): return struct.unpack(fmt,u.mem_read(address,struct.calcsize(fmt)))[0]
def setup(a=True,b=True,capacity_a=4,capacity_b=4,metadata_direct=True):
 for base,size in ((manager,0x1000),(context,0x300),(registry,0x2000),(storage,0x1000),(entity,0x1000),(queue_a,0x2000),(o.STACK,0x10000)):
  u.mem_write(base,bytes(size))
 write(manager,'<Q',registry)
 write(manager+0x4dc,'<I',123)
 write(manager+0xebc2d0,'<H',1)
 write(registry+0x190,'<Q',(1<<index) if a else 0)
 write(registry+0x230,'<Q',(1<<index) if b else 0)
 write(entity+0x10,'<I',0x1234)
 write(entity+0x20,'<Q',storage)
 write(storage+0x14,'<H',0x80)
 write(storage+0x1c,'<H',0x90)
 write(storage+0x1e,'<H',0x94)
 write(storage+0x80,'<I',0x1234)
 write(storage+0x8c,'<B',int(metadata_direct))
 write(context,'<QQQQQ',queue_a,0,capacity_a,0,0)
 write(context+0x140,'<QQQQQ',queue_b,0,capacity_b,0,0)
 write(context+0x290,'<Q',manager)
def execute(queue_index=0):
 rsp=o.STACK+0xff08
 write(rsp,'<Q',o.STOP)
 write(rsp+0x28,'<Q',queue_index)
 for reg,value in ((UC_X86_REG_RCX,manager),(UC_X86_REG_RDX,context),
  (UC_X86_REG_R8,entity),(UC_X86_REG_R9,index),(UC_X86_REG_RSP,rsp)):
  u.reg_write(reg,value)
 u.emu_start(pe.base+0x9c5960,o.STOP,timeout=100000,count=100000)
 assert u.reg_read(UC_X86_REG_RIP)==o.STOP
 return dict(stampA=read(storage+0x90,'<I'),stampB=read(storage+0x94,'<I'),
  countA=read(context+8,'<Q'),countB=read(context+0x148,'<Q'),
  firstA=read(queue_a,'<I'),firstB=read(queue_b,'<I'),
  pendingA=read(context+0x280,'<I'),pendingB=read(context+0x284,'<I'))
cases=[]
setup()
first=execute()
assert first==dict(stampA=123,stampB=123,countA=1,countB=1,firstA=0x1234,firstB=0x1234,pendingA=0,pendingB=0),first
cases.append(dict(name='both-categories',result=first))
second=execute()
assert second==first,second
cases.append(dict(name='same-sequence-is-deduplicated',result=second))
setup(a=True,b=False)
a=execute()
assert a['countA']==1 and a['countB']==0 and a['stampA']==123 and a['stampB']==0,a
cases.append(dict(name='registration-selects-only-A',result=a))
setup(a=False,b=True)
b=execute()
assert b['countA']==0 and b['countB']==1 and b['stampA']==0 and b['stampB']==123,b
cases.append(dict(name='registration-selects-only-B',result=b))
setup(a=False,b=False)
none=execute()
assert none['countA']==none['countB']==none['stampA']==none['stampB']==0,none
cases.append(dict(name='untracked-registration',result=none))
setup(capacity_a=0,capacity_b=0)
full=execute()
assert full['countA']==full['countB']==0 and full['stampA']==full['stampB']==0x8000007b and full['pendingA']==full['pendingB']==1,full
cases.append(dict(name='full-queues-without-allocator-record-pending',result=full))
full_again=execute()
assert full_again==full,full_again
cases.append(dict(name='pending-marker-is-not-counted-again',result=full_again))
setup(metadata_direct=False)
indirect_disabled=execute()
assert indirect_disabled['countA']==1 and indirect_disabled['countB']==0,indirect_disabled
cases.append(dict(name='B-metadata-neither-direct-nor-alias',result=indirect_disabled))
u.mem_map(0x74000000,0x10000)
def setup_alias(cycle=False):
 setup(metadata_direct=False)
 u.mem_write(0x74000000,bytes(0x10000))
 layout=0x74000000
 bits=0x74001000
 keys=0x74002000
 table=0x74003000
 write(manager+0x150,'<Q',bits)
 write(manager+0x158,'<Q',16)
 write(manager+0x170,'<Q',keys)
 write(manager+0x188,'<Q',table)
 write(manager+0x19c,'<I',3)
 write(registry+0x138,'<H',5)
 write(layout,'<Q',1<<5)
 write(layout+0x84+5*2,'<H',0x100)
 write(layout+0xa84+5*2,'<H',4)
 rows=[(0x1234,entity,storage,0x2345,False),
       (0x2345,entity+0x100,0x74004000,0x1234 if cycle else 0x3456,False)]
 if not cycle: rows.append((0x3456,entity+0x200,0x74005000,0,True))
 occupied=set()
 for entity_id,entry,store,next_id,direct in rows:
  write(entry+0x10,'<I',entity_id)
  write(entry+0x18,'<Q',layout)
  write(entry+0x20,'<Q',store)
  write(store+0x14,'<H',0x80)
  write(store+0x1c,'<H',0x90)
  write(store+0x1e,'<H',0x94)
  write(store+0x80,'<I',entity_id)
  write(store+0x8c,'<B',int(direct))
  write(store+0x8d,'<B',0 if direct else 8)
  write(store+0x100,'<I',next_id)
  h=(((entity_id>>16)^entity_id)*0x45d9f3b)&0xffffffff
  h=(((h>>16)^h)*0x45d9f3b)&0xffffffff
  slot=((h>>16)^h)&15
  while slot in occupied: slot=(slot+1)&15
  occupied.add(slot)
  write(keys+slot*4,'<I',entity_id)
  write(table+slot*8,'<Q',entry)
 write(bits,'<Q',sum(1<<slot for slot in occupied))
 return rows
rows=setup_alias()
alias=execute()
assert alias['countA']==alias['countB']==1 and alias['firstB']==0x3456,alias
assert all(read(store+0x94,'<I')==123 for _,_,store,_,_ in rows)
cases.append(dict(name='B-alias-chain-appends-one-final-target',result=alias,
                 aliasStamps=[read(store+0x94,'<I') for _,_,store,_,_ in rows]))
rows=setup_alias(cycle=True)
cycle=execute()
assert cycle['countA']==1 and cycle['countB']==0,cycle
assert all(read(store+0x94,'<I')==123 for _,_,store,_,_ in rows)
cases.append(dict(name='B-alias-cycle-stops-on-visited-sequence',result=cycle))
result=dict(executable=pe.identity(),entryRva=0x9c5960,
 environment='Unicorn x86_64; original read-only PE; isolated private synthetic state',
 processOpened=False,processWrites=False,nativeGameCalls=False,allocatorCalls=False,
 cases=cases,limitations=['Synthetic metadata, not a live gameplay test','No concurrent thread test or allocator safety claim'])
out=Path('src/loader/runtime/profile-tools/devdata/creative-mode-146/current-change-tracker-oracle-20261016.json')
out.write_text(json.dumps(result,indent=2))
print(json.dumps(dict(cases=len(cases),passed=True,path=str(out))))
