Spaces:
Running
Running
| """Selective documentary contract v2. Pure conversion; no provider or global session state.""" | |
| from __future__ import annotations | |
| import copy | |
| import json | |
| import re | |
| from pathlib import Path | |
| import tempfile | |
| from uuid import uuid4 | |
| from sofiel_documents import (DocumentError, Limits, _require, _text, _date, | |
| digest, packed, now, empty_store, _validate_store) | |
| def empty_v2(instance_id=None): | |
| old = empty_store(instance_id) | |
| return {k: v for k, v in old.items() if k not in ('segments', 'knowledge')} | { | |
| 'schema_version': 2, 'sharing_events': {}, 'excerpts': {}, 'insights': {}, 'legacy_refs': {}} | |
| def size(value): | |
| return len(json.dumps({'documents': value}, ensure_ascii=False, allow_nan=False, indent=2).encode('utf-8')) | |
| def backup_migration(memory, memory_file): | |
| """Retain an immutable valid v1 input, independently of the rotating save backup.""" | |
| directory = Path(memory_file).parent / 'document-migrations' | |
| directory.mkdir(exist_ok=True) | |
| target = directory / ('v1-' + uuid4().hex + '.json') | |
| with target.open('x', encoding='utf-8') as stream: | |
| json.dump(memory, stream, ensure_ascii=False, indent=2, allow_nan=False) | |
| stream.flush() | |
| return target | |
| def _fields(row, allowed, required=None): | |
| _require(isinstance(row, dict) and not set(row) - set(allowed), 'unknown_document_field') | |
| _require(set(required or allowed).issubset(row), 'missing_document_field') | |
| def validate_v2(value, limits=None): | |
| lim = limits or Limits.from_env() | |
| _fields(value, ('schema_version', 'instance_id', 'revision', 'sources', 'sharing_events', | |
| 'excerpts', 'insights', 'operations', 'legacy_refs')) | |
| _require(type(value['schema_version']) is int and value['schema_version'] == 2, 'unsupported_documents_version') | |
| _require(_text(value['instance_id'], 100), 'invalid_instance_id') | |
| _require(type(value['revision']) is int and value['revision'] >= 0, 'invalid_revision') | |
| for name in ('sources', 'sharing_events', 'excerpts', 'insights', 'operations', 'legacy_refs'): | |
| _require(isinstance(value[name], dict), 'invalid_' + name) | |
| _require(len(value['sources']) <= lim.sources, 'source_limit') | |
| _require(len(value['sharing_events']) <= lim.events and len(value['operations']) <= lim.operations, 'document_history_limit') | |
| # Reuse all original source/coverage checks, without accepting extra unbounded fields. | |
| shell = empty_store(value['instance_id']) | |
| shell['sources'] = copy.deepcopy(value['sources']) | |
| for source in shell['sources'].values(): | |
| _fields(source, ('source_id','instance_id','sha256','name','format','created_at','participant', | |
| 'declared_authors','active','availability','normalization_version','encoding','coverage')) | |
| _require(source['availability'] == 'reload_original', 'invalid_availability') | |
| _fields(source['coverage'], ('total_units','visited','empty','failed','pending_units','cursor')) | |
| _require(source['coverage']['cursor'] is None or set(source['coverage']['cursor']) == {'unit','offset'}, 'invalid_cursor') | |
| _require(source['source_id'] == 'src_' + source['sha256'], 'invalid_source_hash_id') | |
| _require(source['encoding'] is None or _text(source['encoding'],80), 'invalid_encoding') | |
| _validate_store(shell, lim) | |
| for key, event in value['sharing_events'].items(): | |
| _fields(event, ('event_id','source_id','participant','message','message_ref','created_at', | |
| 'authorship','attribution_review','stated_purpose','interpreted_purpose','purpose_status','corrections')) | |
| _require(key == event['event_id'] and _text(key, 150) and event['source_id'] in value['sources'], 'invalid_event') | |
| _require(event['participant'] is None or _text(event['participant'],200), 'invalid_participant') | |
| _require(_text(event['message'], 2000, True) and event['message_ref'] == 'msg_' + digest(event['message']), 'invalid_message_reference') | |
| _require(_date(event['created_at']), 'invalid_event_date') | |
| _require(_text(event['stated_purpose'], 2000, True) and _text(event['interpreted_purpose'], 600, True), 'invalid_purpose') | |
| _require(event['purpose_status'] in {'explicit','inferred','unknown'}, 'invalid_purpose_status') | |
| _require(event['attribution_review'] in {'different_declarations','consistent_or_unknown'}, 'invalid_authorship_review') | |
| _require(isinstance(event['authorship'], list) and len(event['authorship']) <= 16, 'invalid_authorship') | |
| for author in event['authorship']: | |
| _fields(author, ('label','status','basis')) | |
| _require(_text(author['label'],200,True) and author['status'] in {'user_declared','document_reported','hypothesis','unknown'} | |
| and _text(author['basis'],600,True), 'invalid_authorship') | |
| _require(isinstance(event['corrections'], list) and len(event['corrections']) <= 16, 'correction_limit') | |
| for change in event['corrections']: | |
| _fields(change, ('created_at','author','purpose')) | |
| _require(_date(change['created_at']) and _text(change['author'],200,True) and _text(change['purpose'],600,True), 'invalid_correction') | |
| for key, row in value['excerpts'].items(): | |
| _fields(row, ('excerpt_id','source_id','text','context','sha256','locator','char_start','char_end','voice','review_state')) | |
| _require(row['source_id'] in value['sources'] and key == row['excerpt_id'], 'dangling_excerpt') | |
| _require(_text(row['text'], lim.quote_chars) and _text(row['context'], lim.context_chars, True), 'invalid_excerpt_text') | |
| _require(row['sha256'] == digest(row['text']), 'invalid_excerpt_hash') | |
| _require(key == excerpt_id(row), 'invalid_excerpt_id') | |
| _require(isinstance(row['locator'],dict) and type(row['locator'].get('unit')) is int, 'invalid_locator') | |
| _require(not set(row['locator']) - {'kind','unit','page','paragraph','row','column','columns','table','cell','block','delimiter'}, 'invalid_locator') | |
| _require(row['locator'].get('kind') in {'text','csv_row','docx_cell','docx_paragraph','pdf_page'}, 'invalid_locator') | |
| _require(all((k=='kind' and _text(x,80)) or (k=='delimiter' and x in {',',';','\t','|'}) or | |
| (k not in {'kind','delimiter'} and type(x) is int and x>=(0 if k=='unit' else 1)) | |
| for k,x in row['locator'].items()), 'invalid_locator') | |
| a,b = row['char_start'],row['char_end'] | |
| _require((a is None and b is None) or (type(a) is int and type(b) is int and 0 <= a < b and b-a == len(row['text'])), 'invalid_offsets') | |
| _fields(row['voice'], ('role','label')) | |
| _require(row['voice']['role'] in {'author','narrator','character','quotation','unknown'} and _text(row['voice']['label'],200,True), 'invalid_attribution') | |
| _require(row['review_state'] in {'literal_verified','legacy_location_ambiguous'}, 'invalid_review_state') | |
| for key, row in value['insights'].items(): | |
| _fields(row, ('knowledge_id','source_id','statement','claim_kind','attribution','support_refs','relevance_reasons', | |
| 'state','created_at','incorporated_at','memory_refs','purpose_refs','relations','event_ids','review_state')) | |
| _require(key == row['knowledge_id'] and _text(key,150) and row['source_id'] in value['sources'], 'invalid_knowledge') | |
| _require(_text(row['statement'],2000 if row.get('review_state') == 'legacy_preserved' else 600) | |
| and row['claim_kind'] in {'source_claim','sofiel_interpretation','question'}, 'invalid_statement') | |
| allowed_states = {'incorporated','candidate','rejected'} if row.get('review_state') == 'legacy_preserved' else {'incorporated'} | |
| _require(row['state'] in allowed_states and _date(row['created_at']) | |
| and (_date(row['incorporated_at']) if row['state'] == 'incorporated' else row['incorporated_at'] is None), 'invalid_knowledge_state') | |
| _require(row['review_state'] in {'selection_provisional','legacy_preserved'}, 'invalid_review_state') | |
| _fields(row['attribution'], ('role','label')) | |
| _require(row['attribution']['role'] in {'author','narrator','character','quotation','unknown'} and _text(row['attribution']['label'],200,True), 'invalid_attribution') | |
| _require(isinstance(row['support_refs'],list) and 0 < len(row['support_refs']) <= 32, 'missing_support') | |
| for ref in row['support_refs']: | |
| _fields(ref, ('excerpt_id',)) | |
| _require(value['excerpts'].get(ref['excerpt_id'],{}).get('source_id') == row['source_id'], 'foreign_support') | |
| _require(isinstance(row['event_ids'],list) and row['event_ids'] and all(value['sharing_events'].get(e,{}).get('source_id') == row['source_id'] for e in row['event_ids']), 'invalid_event_reference') | |
| _require(isinstance(row['relevance_reasons'],list) and 0 < len(row['relevance_reasons']) <= 8 and all(_text(r,1000) for r in row['relevance_reasons']), 'invalid_relevance') | |
| _require(isinstance(row['memory_refs'],list) and len(row['memory_refs']) <= 16, 'invalid_memory_refs') | |
| for ref in row['memory_refs']: | |
| _fields(ref, ('memory_id','status')) | |
| _require(_text(ref['memory_id'],100) and ref['status'] in {'resolved','orphaned'}, 'invalid_memory_reference') | |
| _require(isinstance(row['purpose_refs'],list) and len(row['purpose_refs'])<=16, 'invalid_purpose_refs') | |
| for ref in row['purpose_refs']: | |
| _fields(ref, ('purpose_id','status')) | |
| _require(_text(ref['purpose_id'],100) and ref['status'] in {'resolved','orphaned'}, 'invalid_purpose_reference') | |
| _require(isinstance(row['relations'],list) and len(row['relations']) <= 16, 'invalid_relations') | |
| for ref in row['relations']: | |
| _fields(ref, ('kind','target_id')) | |
| _require(ref['kind'] in {'contradiction','duplication','derivation'} and ref['target_id'] in value['insights'], 'invalid_relation') | |
| for key, ids in value['legacy_refs'].items(): | |
| _require(_text(key,150) and isinstance(ids,list) and ids and all(i in value['excerpts'] for i in ids), 'invalid_legacy_reference') | |
| for key, op in value['operations'].items(): | |
| _fields(op, ('operation_id','created_at','event_id','source_id','result','diagnostics')) | |
| _require(op['operation_id'] == key and _text(key,100) and _date(op['created_at']), 'invalid_operation') | |
| _require((op['source_id'] is None and op['event_id'] is None) or | |
| (op['source_id'] in value['sources'] and value['sharing_events'].get(op['event_id'],{}).get('source_id') == op['source_id']), 'invalid_operation_reference') | |
| _fields(op['result'], ('status','operation_id','error_code','extraction_status','interpretation_status','incorporated','pending_segments','coverage','warnings')) | |
| r = op['result'] | |
| _require(r['status'] in {'complete','partial','failed'} and r['operation_id'] == key, 'invalid_result') | |
| _require(r['error_code'] is None or _text(r['error_code'],100), 'invalid_result') | |
| _require(all(r[f] in {'pending','complete','partial','failed'} for f in ('extraction_status','interpretation_status')), 'invalid_result') | |
| _require(all(type(r[f]) is int and r[f] >= 0 for f in ('incorporated','pending_segments')), 'invalid_result') | |
| _fields(r['coverage'], ('total_units','visited','empty','failed','pending_units','cursor')) | |
| coverage_check = empty_store(value['instance_id']) | |
| if op['source_id'] is not None: | |
| coverage_check['sources'][op['source_id']] = copy.deepcopy(value['sources'][op['source_id']]) | |
| coverage_check['sources'][op['source_id']]['coverage'] = copy.deepcopy(r['coverage']) | |
| _validate_store(coverage_check, lim) | |
| else: | |
| _require(all(type(r['coverage'][f]) is int and r['coverage'][f]>=0 for f in ('total_units','pending_units')) | |
| and all(isinstance(r['coverage'][f],list) and all(type(n) is int and n>=0 for n in r['coverage'][f]) for f in ('visited','empty','failed')) | |
| and r['coverage']['cursor'] is None, 'invalid_coverage') | |
| _require(isinstance(r['warnings'],list) and len(r['warnings']) <= 8 and all(_text(w,300) for w in r['warnings']), 'invalid_result') | |
| _require(isinstance(op['diagnostics'],list) and len(op['diagnostics']) <= lim.calls, 'invalid_diagnostics') | |
| for d in op['diagnostics']: | |
| _fields(d, ('status','error_code','finish_reason','truncated')) | |
| _require(d['status'] in {'complete','failed'} and type(d['truncated']) is bool and all(d[f] is None or _text(d[f],100) for f in ('error_code','finish_reason')), 'invalid_diagnostics') | |
| for sid in value['sources']: | |
| subset = {name: {k:r for k,r in value[name].items() if r.get('source_id') == sid} | |
| for name in ('sources','sharing_events','excerpts','insights','operations')} | |
| _require(len(subset['insights']) <= lim.selections, 'selection_limit') | |
| _require(size(subset) <= lim.source_bytes, 'source_store_limit') | |
| _require(size(value) <= min(lim.durable_bytes,lim.store_bytes), 'document_store_limit') | |
| result=json.loads(packed(value)) | |
| for event in result['sharing_events'].values(): | |
| update_authorship_review(event) | |
| return result | |
| def excerpt_id(row): | |
| return 'ex_' + digest(packed({k:row[k] for k in ('source_id','text','locator','char_start','char_end')})) | |
| def add_excerpt(store, segment, quote, voice=None, limits=None): | |
| lim = limits or Limits.from_env() | |
| _require(quote and quote in segment['text'], 'unsupported_quote') | |
| _require(len(quote) <= lim.quote_chars, 'quote_limit') | |
| offset = segment['text'].find(quote) | |
| ambiguous = segment['text'].find(quote,offset+1) >= 0 | |
| a = None if ambiguous else segment['char_start']+offset | |
| # Keep a bounded window around the exact quote so negation and nearby attribution | |
| # are not discarded. It is context, distinct from the literal support itself. | |
| separator=' [鈥 ' | |
| flank=max(0,(lim.context_chars-len(separator))//2) | |
| before=segment['text'][max(0,offset-flank):offset] | |
| after=segment['text'][offset+len(quote):offset+len(quote)+flank] | |
| context=before+separator+after if before or after else '' | |
| row = {'source_id':segment['source_id'], 'text':quote, 'context':context, 'sha256':digest(quote), | |
| 'locator':copy.deepcopy(segment['locator']), 'char_start':a, 'char_end':None if a is None else a+len(quote), | |
| 'voice':copy.deepcopy(voice or {'role':'unknown','label':''}), | |
| 'review_state':'legacy_location_ambiguous' if ambiguous else 'literal_verified'} | |
| row['excerpt_id'] = excerpt_id(row) | |
| store['excerpts'].setdefault(row['excerpt_id'], row) | |
| return row['excerpt_id'] | |
| def make_event(sid, query='', participant=None, *, event_id=None): | |
| query = str(query) | |
| _require(len(query) <= 2000, 'sharing_message_limit') | |
| authors = [] | |
| if re.search(r'(?:^|[.!?]\s+)lo escrib[i铆] yo\b',query,re.I): | |
| authors.append({'label':participant or '', 'status':'user_declared', 'basis':'Lo escrib铆 yo'}) | |
| match = re.search(r'(?:^|[.!?]\s+)este texto es de ([^\n.!?]{1,200})',query,re.I) | |
| if match: | |
| authors.append({'label':match.group(1).strip(),'status':'user_declared','basis':match.group(0)}) | |
| explicit = bool(re.search(r'\?|\b(quiero|para que|le[e茅]|leas|analiz[a谩]|analices|revis[a谩]|revises|compar[a谩]|explor[a谩]|ayud[a谩]|explic[a谩]|interpret[a谩]|record[a谩]|conserv[a谩])\b', query, re.I)) | |
| return {'event_id':event_id or 'share_'+digest(sid+now()), 'source_id':sid, 'participant':participant, | |
| 'message':query,'message_ref':'msg_'+digest(query),'created_at':now(), 'authorship':authors,'attribution_review':'consistent_or_unknown', | |
| 'stated_purpose':query if explicit else '', 'interpreted_purpose':'', 'purpose_status':'explicit' if explicit else 'unknown', 'corrections':[]} | |
| def dependency_ids(memory): | |
| segments,knowledge = set(),set() | |
| def walk(item): | |
| if isinstance(item,dict): | |
| for k,v in item.items(): | |
| if k == 'documents': continue | |
| if k == 'segment_ids' and isinstance(v,list): segments.update(x for x in v if isinstance(x,str)) | |
| elif k == 'knowledge_ids' and isinstance(v,list): knowledge.update(x for x in v if isinstance(x,str)) | |
| else: walk(v) | |
| elif isinstance(item,list): | |
| for v in item: walk(v) | |
| walk(memory) | |
| return segments,knowledge | |
| def migrate_v1(store, memory=None, limits=None): | |
| """No mutation. Any protected support that cannot fit aborts the entire migration.""" | |
| lim = limits or Limits.from_env() | |
| old = _validate_store(store, lim) | |
| out = empty_v2(old['instance_id']); out['revision'] = old['revision'] | |
| out['sources'] = copy.deepcopy(old['sources']) | |
| for sid,s in out['sources'].items(): | |
| s['availability'] = 'reload_original' | |
| e = make_event(sid,event_id='share_legacy_'+sid[4:]) | |
| out['sharing_events'][e['event_id']] = e | |
| protected,knowledge_ids = dependency_ids(memory or {}) | |
| _require(protected.issubset(old['segments']), 'migration_missing_support') | |
| _require(knowledge_ids.issubset(old['knowledge']), 'migration_missing_knowledge') | |
| keep = {k for k,r in old['knowledge'].items() if r['state'] == 'incorporated'} | knowledge_ids | |
| while True: | |
| expanded = keep | {x['target_id'] for k in keep for x in old['knowledge'][k]['relations']} | |
| if expanded == keep: break | |
| keep = expanded | |
| for key in keep: | |
| row = copy.deepcopy(old['knowledge'][key]); refs = [] | |
| for ref in row['support_refs']: | |
| seg = old['segments'][ref['segment_id']] | |
| for pos in range(0,len(ref['quote']),lim.quote_chars): | |
| refs.append({'excerpt_id':add_excerpt(out,seg,ref['quote'][pos:pos+lim.quote_chars],row['attribution'],lim)}) | |
| for f in ('instance_id','operation_id'): row.pop(f,None) | |
| row.update(support_refs=refs,event_ids=['share_legacy_'+row['source_id'][4:]], review_state='legacy_preserved',purpose_refs=[]) | |
| if not row['relevance_reasons']: row['relevance_reasons'] = ['Soporte de una referencia hist贸rica protegida; pertinencia pendiente de revisi贸n.'] | |
| out['insights'][key] = row | |
| for sid in protected: | |
| seg = old['segments'][sid]; ids=[] | |
| for pos in range(0,len(seg['text']),lim.quote_chars): | |
| ids.append(add_excerpt(out,seg,seg['text'][pos:pos+lim.quote_chars],limits=lim)) | |
| out['legacy_refs'][sid] = ids | |
| # Compact operation history: only bounded summaries, no retired temporary IDs or model payloads. | |
| for key,op in old['operations'].items(): | |
| source_ids = {old['segments'][s]['source_id'] for s in op['submitted_segment_ids']} | |
| sid = next(iter(source_ids)) if len(source_ids)==1 else None | |
| result=copy.deepcopy(op.get('result',{})) | |
| if not result or sid is None: | |
| result={'status':'failed','operation_id':key,'error_code':'legacy_operation_unattributed', | |
| 'extraction_status':'pending','interpretation_status':'pending','incorporated':0,'pending_segments':0, | |
| 'coverage':{'total_units':0,'visited':[],'empty':[],'failed':[],'pending_units':0,'cursor':None}, | |
| 'warnings':['Operaci贸n hist贸rica sin una fuente inequ铆voca; se conserva su ID sin inventar atribuci贸n.']} | |
| out['operations'][key] = {'operation_id':key,'created_at':op['created_at'], 'source_id':sid, | |
| 'event_id':'share_legacy_'+sid[4:] if sid else None, 'result':result, 'diagnostics':copy.deepcopy(op.get('attempts',[]))} | |
| return validate_v2(out,lim) | |
| def record_document_authors(event, segments): | |
| """Only explicit signature lines; uploading and first-person prose infer no author.""" | |
| for segment in segments: | |
| for match in re.finditer(r'^(?:Autor(?:a)?\s*:|Escrito por\s*:?)\s*([^\r\n]{1,200})$',segment['text'],re.I|re.M): | |
| item={'label':match.group(1).strip(),'status':'document_reported','basis':match.group(0)} | |
| if item not in event['authorship']: event['authorship'].append(item) | |
| update_authorship_review(event) | |
| def effective_authorship(event): | |
| """Last explicit correction replaces human declarations; signatures remain reports. | |
| Derived from the existing v2 history, so older exports get the same rule. Multiple | |
| uncorrected human declarations remain an unresolved conflict, never inferred coauthorship. | |
| """ | |
| correction=next((c['author'] for c in reversed(event['corrections']) if c['author']),None) | |
| authors=copy.deepcopy(event['authorship']) | |
| if correction: | |
| authors=[a for a in authors if a['status']!='user_declared'] | |
| authors.append({'label':correction,'status':'user_declared','basis':'Correcci贸n expl铆cita en procedencia'}) | |
| return authors | |
| def event_context(event): | |
| result=copy.deepcopy(event) | |
| result['effective_authorship']=effective_authorship(event) | |
| result['superseded_authorship']=[a for a in event['authorship'] if a not in result['effective_authorship']] | |
| update_authorship_review(result) | |
| return result | |
| def update_authorship_review(event): | |
| declared={a['label'].casefold().strip() for a in effective_authorship(event) if a['status']=='user_declared' and a['label'].strip()} | |
| reported={a['label'].casefold().strip() for a in event['authorship'] if a['status']=='document_reported' and a['label'].strip()} | |
| event['attribution_review']='different_declarations' if len(declared)>1 or (declared and reported and declared!=reported) else 'consistent_or_unknown' | |
| def purpose_index(memory): | |
| state=memory.get('purpose_memory',{}) | |
| if not isinstance(state,dict): return {} | |
| rows=[r for field in ('purposes','fulfilled_purposes') for r in state.get(field,[]) if isinstance(r,dict) and _text(r.get('id'),100)] | |
| counts={r['id']:sum(s['id']==r['id'] for s in rows) for r in rows} | |
| return {r['id']:str(r.get('description',r.get('content','')))[:250] for r in rows if counts[r['id']]==1} | |
| def reconcile_purpose_refs(store, memory): | |
| known=purpose_index(memory) | |
| for row in store.get('insights',{}).values(): | |
| for ref in row['purpose_refs']: | |
| ref['status']='resolved' if ref['purpose_id'] in known else 'orphaned' | |
| def passage(row, source): | |
| return {'segment_id':row['excerpt_id'], 'excerpt_id':row['excerpt_id'], 'source_id':row['source_id'], | |
| 'text':row['text'], 'sha256':row['sha256'], 'locator':copy.deepcopy(row['locator']), | |
| 'char_start':row['char_start'],'char_end':row['char_end'], 'voice':copy.deepcopy(row['voice']), | |
| 'surrounding_context':row['context'], | |
| 'source_name':source['name'], 'declared_authors':source['declared_authors'], 'availability':'selected_excerpt'} | |
| def retrieve_v2(store, query, *, max_chars=12000, limit=6, atoms=None, session=None, focus=None, original_sources=()): | |
| terms=set(re.findall(r'\w{3,}',str(query).casefold())) | |
| out={'revision':store['revision'],'instance_id':store['instance_id'],'retrieval':'selective', | |
| 'passages':[],'knowledge':[], 'availability':[], 'sharing_events':[]} | |
| ranked=[]; selected=set() | |
| for atom in atoms or []: | |
| if not terms.intersection(re.findall(r'\w{3,}',getattr(atom,'concepto','').casefold())): continue | |
| for event in getattr(atom,'evidence_provenance',[]): | |
| for sid in event.get('segment_ids',[]): | |
| selected.update(store['legacy_refs'].get(sid,[sid])) | |
| for sid,source in store['sources'].items(): | |
| if not source['active'] or focus and sid != focus: continue | |
| events=sorted((e for e in store['sharing_events'].values() if e['source_id']==sid), key=lambda e:e['created_at']) | |
| purpose=' '.join(e['stated_purpose']+' '+e['interpreted_purpose'] for e in events) | |
| rows=[passage(e,source) for e in store['excerpts'].values() if e['source_id']==sid] | |
| if session: | |
| rows += [{**copy.deepcopy(s),'source_name':source['name'],'availability':'session_text', | |
| 'declared_authors':source['declared_authors']} for s in session.get('segments',{}).values() if s['source_id']==sid] | |
| for p in rows: | |
| insights=[k for k in store['insights'].values() if k['state']=='incorporated' and any(r['excerpt_id']==p['segment_id'] for r in k['support_refs'])] | |
| corpus=p['text']+' '+purpose+' '+' '.join(k['statement']+' '+' '.join(k['relevance_reasons']) for k in insights) | |
| score=(4*len(terms.intersection(re.findall(r'\w{3,}',p['text'].casefold()))) | |
| +len(terms.intersection(re.findall(r'\w{3,}',corpus.casefold()))) + (3 if p['segment_id'] in selected else 0)) | |
| if score or focus: ranked.append((score,p,insights,events)) | |
| seen=set() | |
| for score,p,insights,events in sorted(ranked,key=lambda x:(-x[0], x[1].get('availability')!='selected_excerpt',x[1]['segment_id'])): | |
| content_key=(p['source_id'],p['text']) | |
| if content_key in seen: continue | |
| candidate=copy.deepcopy(out); candidate['passages'].append(p) | |
| # Knowledge is bounded and clearly attributed; support excerpts may be retrieved separately. | |
| known={x['knowledge_id'] for x in candidate['knowledge']} | |
| candidate['knowledge'].extend(copy.deepcopy(k) for k in insights if k['knowledge_id'] not in known) | |
| eids={x['event_id'] for x in candidate['sharing_events']} | |
| candidate['sharing_events'].extend(event_context(e) for e in events[-2:] if e['event_id'] not in eids) | |
| if len(packed(candidate)) <= max_chars: | |
| out=candidate;seen.add(content_key) | |
| if len(out['passages']) >= limit: break | |
| metadata_sources={p['source_id'] for p in out['passages']} | |
| for sid,source in store['sources'].items(): | |
| if not source['active'] or focus and sid!=focus: continue | |
| name_terms=set(re.findall(r'\w{3,}',source['name'].casefold())) | |
| if focus or terms & (name_terms | {'documento','documentos','archivo','archivos','fuente','texto','original'}): | |
| metadata_sources.add(sid) | |
| for sid in sorted(metadata_sources): | |
| available=bool(session and any(s['source_id']==sid for s in session.get('segments',{}).values())) | |
| item={'source_id':sid,'source_name':store['sources'][sid]['name'],'text_in_session':available, | |
| 'original_in_session':sid in original_sources,'original_required_for_unretained':True} | |
| candidate=copy.deepcopy(out);candidate['availability'].append(item) | |
| if len(packed(candidate)) <= max_chars: out=candidate | |
| return out if out['passages'] or out['availability'] else {} | |
| def search_originals(store, session, originals, query, limits, extractor, *, focus=None): | |
| """Bounded read-only scan, independent of ingestion. Returned text is request-local. | |
| Every parser invocation checks the original hash and uses the existing isolation | |
| limits. A budget-limited scan reports its stopping point instead of claiming absence. | |
| """ | |
| import time | |
| from dataclasses import replace | |
| terms=set(re.findall(r'\w{3,}',str(query).casefold()))-{ | |
| 'que','qu茅','del','las','los','una','por','con','para','dice','sobre','este','esta', | |
| 'documento','archivo','texto','original','fuente','pasaje','the','and','what'} | |
| if not terms or not session or not session.get('evicted_windows'): return {},[] | |
| deadline=time.monotonic()+limits.retrieval_seconds | |
| calls=0; matches={}; diagnostics=[] | |
| for sid,source in store['sources'].items(): | |
| if not source['active'] or (focus and sid!=focus) or not session['evicted_windows'].get(sid): continue | |
| if sid not in originals: continue | |
| cached=[s for s in session.get('segments',{}).values() if s['source_id']==sid] | |
| covered=set(re.findall(r'\w{3,}',' '.join(s['text'] for s in cached).casefold())) | |
| if terms <= covered: continue | |
| cursor=None; visited=0; incomplete=False; status='budget_exhausted'; error=None | |
| while calls<limits.retrieval_windows and time.monotonic()<deadline: | |
| remaining=deadline-time.monotonic() | |
| if remaining<=0: break | |
| scan_limits=replace(limits,parser_seconds=min(limits.parser_seconds,remaining), | |
| extract_chars=min(limits.extract_chars,max(1,limits.session_bytes//4))) | |
| delimiter=next((s['locator'].get('delimiter') for s in cached if s['locator'].get('delimiter')),None) | |
| calls+=1 | |
| try: | |
| result=extractor(originals[sid],limits=scan_limits,instance_id=store['instance_id'], | |
| cursor=cursor,expected_hash=source['sha256'],encoding=source['encoding'],delimiter=delimiter) | |
| except Exception: | |
| status='failed'; error='document_extraction_failed'; break | |
| if result.error_code: | |
| status='failed'; error=result.error_code; break | |
| if time.monotonic()>deadline: break | |
| visited+=1 | |
| incomplete=incomplete or bool(result.source['coverage']['failed']) | |
| for key,row in result.segments.items(): | |
| score=len(terms & set(re.findall(r'\w{3,}',row['text'].casefold()))) | |
| if score: matches[key]=copy.deepcopy(row) | |
| # Keep only bounded best matches, not a second full-document cache. | |
| ranked=sorted(matches.items(),key=lambda item:-len(terms & set(re.findall(r'\w{3,}',item[1]['text'].casefold())))) | |
| matches={} | |
| for key,row in ranked[:6]: | |
| trial={**matches,key:row} | |
| if len(packed(trial).encode('utf-8'))<=limits.session_bytes: matches=trial | |
| next_cursor=result.source['coverage']['cursor'] | |
| if any(terms <= set(re.findall(r'\w{3,}',r['text'].casefold())) for r in matches.values() if r['source_id']==sid): | |
| status='matched'; break | |
| if next_cursor is None: | |
| status='partial' if incomplete else 'searched'; break | |
| if next_cursor==cursor: | |
| status='failed';error='nonadvancing_cursor';break | |
| cursor=next_cursor | |
| diagnostics.append({'source_id':sid,'status':status,'windows':visited,'error_code':error}) | |
| return matches,diagnostics | |
| class DocumentSessionStore: | |
| """Owned by one service, deliberately outside every serialized memory object.""" | |
| def __init__(self): | |
| self.data=None; self.memory=None; self.epoch=None | |
| self.originals={}; self.directory=None | |
| def clear(self): | |
| self.data=None; self.memory=None; self.epoch=None | |
| self.originals={} | |
| if self.directory: | |
| self.directory.cleanup(); self.directory=None | |
| def read(self, owner): | |
| if self.memory is not owner.memory.data or self.epoch != getattr(owner,'_ui_epoch',0): | |
| self.clear() | |
| return self.data | |
| def install(self, owner, work, limits): | |
| # Count serialized cache too: pending candidates, metadata and all UTF-8 text. | |
| _require(len(packed(work).encode('utf-8')) <= limits.session_bytes,'document_session_limit') | |
| self.data=copy.deepcopy(work); self.memory=owner.memory.data; self.epoch=getattr(owner,'_ui_epoch',0) | |
| def stage_original(self, path, source_id, limits): | |
| """Stage a verified original; caller discards it if the durable transaction fails.""" | |
| if source_id in self.originals: return None | |
| with Path(path).open('rb') as stream: | |
| content=stream.read(limits.file_bytes+1) | |
| _require(len(content)<=limits.file_bytes,'file_size_limit') | |
| _require('src_'+digest(content)==source_id,'source_changed_during_extraction') | |
| _require(sum(Path(p).stat().st_size for p in self.originals.values())+len(content)<=limits.session_file_bytes, | |
| 'document_session_file_limit') | |
| if self.directory is None: self.directory=tempfile.TemporaryDirectory(prefix='sofiel-document-session-') | |
| target=Path(self.directory.name)/(source_id+Path(path).suffix.lower()) | |
| with target.open('xb') as f: f.write(content) | |
| return str(target) | |
| def fit_session(work, limits): | |
| """Evict only completely interpreted windows; never durable or pending support. | |
| This operates on the transaction's private copy. Its result becomes visible only | |
| after the corresponding durable result is saved, so rollback retains the old cache. | |
| """ | |
| analyzed=set(work.get('analyzed',[])) | |
| while len(packed(work).encode('utf-8'))>limits.session_bytes: | |
| victim=next((key for key in work['segments'] if key in analyzed),None) | |
| if victim is None: raise DocumentError('document_session_limit') | |
| row=work['segments'].pop(victim) | |
| work.get('candidates',{}).pop(victim,None) | |
| counts=work.setdefault('evicted_windows',{}) | |
| counts[row['source_id']]=counts.get(row['source_id'],0)+1 | |
| return work | |