File size: 11,296 Bytes
f592f80 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | """Private JSON-lines bridge for the native app. No listening network port."""
import sys,os,json,threading,copy,time,hashlib,traceback
from pathlib import Path
from geometry import Library,new,validate,import_atlas,atomic,point,now
import documents,providers,generation
from rewrite_policy import SYSTEM,VERSION as REWRITE_POLICY,prompt as rewrite_prompt
ROOT=Path(__file__).resolve().parent
SUPPORT=Path(os.environ.get('AFFECTOR_SUPPORT',str(Path.home()/'Library/Application Support/AFFECTOR')))
SUPPORT.mkdir(parents=True,exist_ok=True)
LIB=Library(SUPPORT/'geometries');LOCK=threading.RLock();OUTPUT=threading.Lock();cancel=threading.Event();engine=None;run=None;busy=False
def emit(event,**data):
with OUTPUT:print(json.dumps({'event':event,**data},ensure_ascii=False,allow_nan=False),flush=True)
def progress(text):emit('progress',message=text)
def local():
global engine
if engine is None:
from steering import Engine
engine=Engine(progress)
return engine
def state():return {'geometries':LIB.list(),'support':str(SUPPORT),'has_checkpoint':(SUPPORT/'last-run.json').exists()}
def summarize(data):
counts={'rewritten':0,'unchanged':0,'failed':0,'recovered':0,'warnings':0,'pending':len(data['spans'])-len(data['records'])}
for k,r in data['records'].items():
a,b=data['spans'][int(k)]
category='failed' if r.get('original_retained') or r.get('error') else 'unchanged' if r['text'].strip()==data['input'][a:b].strip() else 'rewritten'
counts[category]+=1
counts['recovered']+=bool(r.get('recovery'))
counts['warnings']+=bool(r.get('warning'))
counts['message']=f"{counts['rewritten']} rewritten 路 {counts['unchanged']} unchanged 路 {counts['failed']} failed (original kept) 路 {counts['pending']} pending 路 {counts['recovered']} recovered 路 {counts['warnings']} need review"
return counts
def save_run():
with LOCK:
if run:
run['summary']=summarize(run);atomic(SUPPORT/'last-run.json',run)
def update():
with LOCK:
if run:
run['summary']=summarize(run)
emit('run',data=copy.deepcopy(run),output=documents.assemble(run['input'],run['spans'],run['records']))
def execute(q):
global run
op=q['op']
if op=='reset_text':
with LOCK:
(SUPPORT/'last-run.json').unlink(missing_ok=True)
run=None
return {'reset_text':True,'message':'Target and transformed text cleared.'}
if op=='list':return state()
if op=='new':g=LIB.save(new(q.get('name','Shape '+now().replace('T',' ')[:19])));return {'geometry':g,**state()}
if op=='save':g=LIB.save(q['geometry']);return {'geometry':g,**state()}
if op=='import':
data=json.loads(Path(q['path']).read_text(encoding='utf-8'));g=import_atlas(data,Path(q['path']).stem)
# Import is a new library entry, never a silent replacement.
import uuid
g['id']=str(uuid.uuid4());g=LIB.save(g);return {'geometry':g,**state()}
if op=='export_geometry':atomic(q['path'],validate(q['geometry']));return {}
if op=='delete':LIB.delete(q['id']);return state()
if op=='read_doc':return {'text':documents.read(q['path'])}
if op=='calibrate':
g=validate(q['geometry']);g['basis']=local().compile(g,cancel,progress);g=LIB.save(g);return {'geometry':g,**state()}
if op=='load_run':
data=json.loads(Path(q['path']).read_text())
if data.get('format')!='affector-run-v1':raise ValueError('Not an AFFECTOR run checkpoint.')
validate(data['geometry']);spans=documents.validate_spans(data['input'],data['spans'])
for k,r in data['records'].items():
if not k.isdigit() or not 0<=int(k)<len(spans) or not isinstance(r.get('text'),str):raise ValueError('Invalid checkpoint records.')
with LOCK:run=data
update();return {'geometry':run['geometry'],'input':run['input'],'backend':run['backend'],'model':run['model'],'sampling':run['sampling'],'rewrite_style':run.get('rewrite_style','Creative')}
if op=='export_doc':
if run is None:raise ValueError('Transform a document first.')
documents.write(q['path'],documents.assemble(run['input'],run['spans'],run['records']));return {}
if op in ('export_report','save_run'):
if run is None:raise ValueError('No transfer to save.')
with LOCK:atomic(q['path'],run)
return {}
if op=='transform':
g=validate(q['geometry']);text=q['text'];backend=q['backend'];model='Qwen/Qwen3.5-0.8B' if backend=='Local Qwen' else q['model'];sampling=q['sampling']
style=q.get('rewrite_style','Creative');system=rewrite_prompt(style)
if not text.strip():raise ValueError('Open a document or enter some text.')
if not (0<=float(sampling['temperature'])<=2 and 0<float(sampling['top_p'])<=1 and 1<=int(sampling['top_k'])<=1000 and 32<=int(sampling['max_tokens'])<=2048):raise ValueError('Invalid sampling controls.')
if backend=='Local Qwen':
e=local()
if g['basis'] is None or any(k not in g['basis']['vectors'] for k in g['layers']):
g['basis']=e.compile(g,cancel,progress);g=LIB.save(g);emit('geometry',data=g)
signature=hashlib.sha256(json.dumps([text,g,backend,model,sampling,REWRITE_POLICY,style],sort_keys=True).encode()).hexdigest()
if run is None or run.get('signature')!=signature:
run={'format':'affector-run-v1','created_at':now(),'input':text,'geometry':g,'backend':backend,'model':model,'sampling':sampling,'rewrite_policy':REWRITE_POLICY,'rewrite_style':style,'signature':signature,'spans':[list(p) for p in documents.spans(text,limit=700)],'records':{},'complete':False,'verification':{}}
save_run()
for index,(a,b) in enumerate(run['spans']):
if cancel.is_set():raise InterruptedError('Stopped; completed passages saved.')
if str(index) in run['records']:continue
max_chars=documents.rewrite_allowance(text,run['spans'],run['records'],index,multiplier=1.25)
passage_system=system+f'\nAim for at most {b-a} characters; finish before the hard limit of {max_chars} characters, including spaces and punctuation.'
desired=point(g,(a+b)/max(1,2*len(text)));progress(f'Transforming passage {index+1}/{len(run["spans"])} 路 {backend}')
def generate_attempt(adjusted,attempt):
instruction=passage_system
if backend=='Local Qwen':
options=dict(sampling)
# Token allowance is a safety ceiling, not the output length target.
options['max_tokens']=min(2048,max(sampling['max_tokens'],max_chars))
return e.generate(text[a:b],g=adjusted,coordinates=desired,cancel=cancel,max_chars=max_chars,soft_chars=max(1,int((b-a)*.85)),system=instruction,time_limit=120,**options)
instruction+='\nAim for valence %.3f, inward/outward direction %.3f, and intensity %.3f.'%(desired[0],desired[1],(desired[2]+1)/2)
return providers.request(backend,model,q.get('key',''),instruction,text[a:b],max_tokens=min(4096,max(sampling['max_tokens'],max_chars)),allow_partial=True)
result=generation.rewrite(text[a:b],generate_attempt,g,backend=='Local Qwen',cancel,progress,max_chars=max_chars)
result['target']=desired;result['character_limit']=max_chars
with LOCK:run['records'][str(index)]=result
save_run();update()
if cancel.is_set():raise InterruptedError('Stopped; generated text saved.')
run['complete']=True;save_run();update();return {'message':"Transformation complete 路 "+run['summary']['message']+". Verify output to measure emotional fit."}
if op=='verify':
if run is None or not run['records']:raise ValueError('Transform at least one passage first.')
provider=q['provider'];model='Qwen/Qwen3.5-0.8B' if provider=='Local Qwen' else q['model'];key=q.get('key','');e=local() if provider=='Local Qwen' else None
scores={};paired=[]
run['verification']={};save_run()
for index,(a,b) in enumerate(run['spans']):
record=run['records'].get(str(index))
if not record:continue
if cancel.is_set():raise InterruptedError('Verification stopped; completed scores retained.')
progress(f'Verifying passage {index+1}/{len(run["spans"])} 路 {provider}')
item={}
for label,content in [('before',run['input'][a:b]),('after',record['text'])]:
if cancel.is_set():raise InterruptedError('Verification stopped; completed scores retained.')
progress(f'Verifying {label} text 路 passage {index+1}/{len(run["spans"])} 路 {provider}')
try:
result=providers.evaluate(content,provider,model,key,cancel,e);item[label]=result;item[label+'_fit']=providers.metric(result['row'],record['target'])
except Exception as error:
if cancel.is_set():raise InterruptedError('Verification stopped; completed scores retained.') from error
item[label+'_error']=str(error);item[label+'_error_type']=type(error).__name__
if item.get('before_fit') and item.get('after_fit'):paired.append(item)
scores[str(index)]=item
run['verification']={'provider':provider,'model':model,'scores':scores,'paired_count':len(paired),'processed_count':len(scores),'total_count':len(run['spans']),'failed_assessments':sum(sum(k.endswith('_error') for k in item) for item in scores.values()),'complete':False,'metric':'similarity = 1 - Euclidean distance / (2 sqrt(3)); not a probability'}
if paired:
before=sum(i['before_fit']['similarity'] for i in paired)/len(paired);after=sum(i['after_fit']['similarity'] for i in paired)/len(paired)
run['verification'].update(before_similarity=before,after_similarity=after,improvement=after-before)
save_run();update()
run['verification']['complete']=True;save_run();update();return {'message':f"Verification complete 路 {run['verification']['failed_assessments']} failed assessments flagged. Scores exclude failures and unknown directions."}
raise ValueError('Unknown operation.')
def worker(q):
global busy
try:
result=execute(q);emit('result',data=result)
except Exception as error:
save_run();emit('error',message=str(error))
finally:busy=False;emit('idle')
def main():
global busy
if not (SUPPORT/'initialized').exists():
if not LIB.list():LIB.save(new('From unease to possibility'))
(SUPPORT/'initialized').touch()
emit('ready',data=state())
for line in sys.stdin:
try:
q=json.loads(line)
if q['op']=='cancel':cancel.set();continue
if q['op']=='save_run':
with LOCK:result=execute(q)
emit('saved',data=result);continue
if busy:emit('error',message='Wait for the current operation or stop it first.');continue
busy=True;cancel.clear();threading.Thread(target=worker,args=(q,),daemon=True).start()
except Exception as error:emit('error',message=str(error))
if __name__=='__main__':main()
|