"""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