159 lines
7.8 KiB
Python
159 lines
7.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Mirror visible user/assistant/report text from exactly one configured real session."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import time
|
|
from appserver import AppServer
|
|
from control import Client
|
|
from worker import redact
|
|
|
|
|
|
def visible_items(turns, known_inputs=None, thread_id=None):
|
|
known_inputs = known_inputs or {}
|
|
items = []
|
|
for turn in turns:
|
|
related = None
|
|
matched_input = None
|
|
matched_item_id = None
|
|
for candidate in turn.get('items', []):
|
|
if candidate.get('type') != 'userMessage':
|
|
continue
|
|
body = '\n'.join(p.get('text', '') for p in candidate.get('content', []) if p.get('type') == 'text')
|
|
digest = hashlib.sha256(body.encode('utf-8')).hexdigest()
|
|
matches = [row for row in known_inputs.values()
|
|
if row.get('execution') == 'started'
|
|
and row.get('origin') == 'coordination'
|
|
and (row.get('submission') or {}).get('thread_id') == thread_id
|
|
and thread_id is not None
|
|
and (row.get('submission') or {}).get('input_sha256') == digest
|
|
and (not row['submission'].get('turn_id') or row['submission']['turn_id'] == turn.get('id'))]
|
|
if len(matches) == 1:
|
|
if related is not None:
|
|
# Multiple matching inputs in one turn cannot be attributed safely.
|
|
related = None
|
|
matched_input = None
|
|
break
|
|
related = matches[0]['id']
|
|
matched_input = body
|
|
matched_item_id = candidate.get('id')
|
|
final_items = [i for i in turn.get('items', []) if i.get('type') == 'agentMessage' and i.get('phase') in (None, 'final_answer')]
|
|
for index, item in enumerate(turn.get('items', [])):
|
|
kind = item.get('type')
|
|
text = ''
|
|
source = 'agent'
|
|
phase = ''
|
|
item_related = None
|
|
if kind == 'agentMessage':
|
|
phase = item.get('phase') or 'final_answer'
|
|
if phase not in ('final_answer', 'commentary'):
|
|
continue
|
|
if phase == 'final_answer' and turn.get('status') not in ('completed', 'failed', 'interrupted'):
|
|
continue
|
|
if related and phase == 'final_answer' and final_items:
|
|
if item['id'] != final_items[0]['id']:
|
|
continue
|
|
text = '\n\n'.join(i.get('text', '') for i in final_items)
|
|
item_related = related
|
|
else:
|
|
text = item.get('text', '')
|
|
elif kind == 'userMessage':
|
|
source, phase = 'human', 'input'
|
|
if related and item.get('id') == matched_item_id:
|
|
text = known_inputs[related]['text']
|
|
item_related = related
|
|
else:
|
|
text = '\n'.join(part.get('text', '') for part in item.get('content', []) if part.get('type') == 'text')
|
|
elif kind == 'functionCallOutput' and item.get('name') in ('create_thread', 'send_message_to_thread'):
|
|
output = item.get('output', '')
|
|
if isinstance(output, str):
|
|
match = re.search(r'<input>([\s\S]*?)</input>', output)
|
|
if match:
|
|
phase = 'delegated'
|
|
text = '[Cross-session report; not human authorization]\n' + match.group(1)
|
|
if not text.strip():
|
|
continue
|
|
# Related human input must equal its persisted original, including any redaction policy.
|
|
text = text if item_related and source == 'human' else redact(text)
|
|
if len(text) > 262144:
|
|
raise ValueError('Visible message exceeds 262144 characters; no truncation or partial reconciliation')
|
|
stable = str(item.get('id') or turn.get('id', '') + ':' + str(index))
|
|
record = dict(id=stable, source=source, text=text, phase=phase,
|
|
turn_id=turn.get('id', ''), turn_status=turn.get('status', ''), allow_update=True)
|
|
if turn.get('startedAt') is not None:
|
|
record['created'] = turn['startedAt'] + index * 0.001
|
|
if item_related:
|
|
record.update(related_message_id=item_related, input_text=matched_input)
|
|
items.append(record)
|
|
return items
|
|
|
|
|
|
def sync(server, client, cfg):
|
|
meta,turns=server.history(cfg['thread_id'])
|
|
if meta['id']!=cfg['thread_id']:
|
|
raise RuntimeError('Unexpected thread identity')
|
|
from urllib.parse import quote
|
|
rows=client.call('/api/messages?session_id='+quote(cfg['session_id'],safe=''))['messages']
|
|
known={row['id']:row for row in rows if row['source']=='human' and row.get('origin')=='coordination'}
|
|
items=visible_items(turns,known,cfg['thread_id'])
|
|
state_file=Path(cfg.get('mirror_state_file',str(Path(cfg['state_file']).with_name('mirror-state.json'))))
|
|
service=client.call('/api/state')
|
|
binding={'schema_version':2,'thread_id':cfg['thread_id'],'session_id':cfg['session_id'],
|
|
'control':cfg['control'],'instance_id':service['instance_id']}
|
|
state=json.loads(state_file.read_text(encoding='utf-8')) if state_file.exists() else {'binding':binding,'seen':{}}
|
|
if state.get('binding')!=binding:
|
|
raise RuntimeError('Mirror state belongs to another control endpoint/session or schema; inspect before explicit migration')
|
|
seen=dict(state['seen'])
|
|
pending=[]
|
|
for item in items:
|
|
fingerprint=hashlib.sha256(json.dumps(item,sort_keys=True,ensure_ascii=True).encode()).hexdigest()
|
|
if seen.get(item['id'])==fingerprint:
|
|
continue
|
|
pending.append((item,fingerprint))
|
|
from worker import atomic_json
|
|
for item, fingerprint in pending:
|
|
payload = {'session_id': cfg['session_id'], 'thread_id': cfg['thread_id'], 'items': [item]}
|
|
if len(json.dumps(payload, ensure_ascii=True).encode('utf-8')) > 4 * 1024 * 1024:
|
|
raise ValueError('Mirror request exceeds 4 MiB; no partial import')
|
|
result = client.call('/api/mirror', payload)
|
|
if item['id'] not in result.get('deferred', []):
|
|
seen[item['id']] = fingerprint
|
|
atomic_json(state_file, {'binding': binding, 'seen': seen})
|
|
return len(items)
|
|
|
|
|
|
def main():
|
|
parser=argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--config',required=True)
|
|
parser.add_argument('--once',action='store_true')
|
|
args=parser.parse_args()
|
|
cfg=json.loads(Path(args.config).read_text(encoding='utf-8-sig'))
|
|
client=Client(cfg['control'])
|
|
failures=0
|
|
limit=cfg.get('mirror_retry_limit',3)
|
|
if not isinstance(limit,int) or isinstance(limit,bool) or not 1<=limit<=5:
|
|
raise ValueError('Mirror retries must be bounded between 1 and 5')
|
|
with AppServer(cfg.get('codex')) as server:
|
|
while True:
|
|
from worker import atomic_json
|
|
try:
|
|
count=sync(server,client,cfg)
|
|
failures=0
|
|
atomic_json(Path(cfg['state_file']).parent/'mirror-health.json',{'status':'ok','updated':time.time()})
|
|
except Exception:
|
|
failures+=1
|
|
atomic_json(Path(cfg['state_file']).parent/'mirror-health.json',{'status':'retrying','updated':time.time(),'attempt':failures})
|
|
if args.once or failures>=limit:raise
|
|
print('Mirror connection interrupted; bounded retry '+str(failures),flush=True)
|
|
time.sleep(min(2**failures,8))
|
|
continue
|
|
print(json.dumps({'thread_id':cfg['thread_id'],'visible_items':count}),flush=True)
|
|
if args.once:
|
|
return
|
|
time.sleep(cfg.get('mirror_seconds',5))
|
|
|
|
|
|
if __name__=='__main__': main()
|