sofiel_document_service.py

"""Per-session document operations: copy outside lock, merge/validate/save under lock."""
from __future__ import annotations

import copy
from dataclasses import asdict, dataclass, field
import time
from uuid import uuid4

from sofiel_documents import (DocumentError, Limits, digest, empty_store, extract_isolated,
    memory_index, now, packed, reconcile_memory_refs, retrieve, validate_store)
from sofiel_evidence import structured_payload_budget


@dataclass
class OperationResult:
    status: str
    operation_id: str
    error_code: str | None = None
    extraction_status: str = "pending"
    interpretation_status: str = "pending"
    incorporated: int = 0
    pending_segments: int = 0
    coverage: dict = field(default_factory=dict)
    warnings: list = field(default_factory=list)


class LegacyDocumentService:
    def __init__(self, owner, limits=None):
        self.owner = owner
        self.limits = limits or Limits.from_env()

    def evidence(self, query):
        # No stale indexes or old list references after restore; live portable state only.
        return retrieve(self.owner.memory.data.get("documents"), query,
            atoms=getattr(getattr(self.owner, "synaptic_manager", None), "kernel", []))

    def _context(self, query):
        persisted = memory_index(self.owner.memory.data.get("chats", []))
        terms = set(str(query).casefold().split())
        ranked = sorted(persisted.items(), key=lambda row: -sum(t in packed(row[1]).casefold() for t in terms))
        chosen = dict(ranked[:3])
        # The reference ID always hashes full saved fields; payload text alone is bounded.
        return {"human_message": str(query)[:700],
            "memories": {k: {f: str(v)[:350] for f, v in row.items()} for k, row in chosen.items()},
            "symbolic_traits": copy.deepcopy(self.owner.memory.data.get("self", {}).get("traits", {})),
            "purposes": str(self.owner.memory.data.get("purpose_memory", {}))[:500]}

    def ingest(self, path=None, *, query="", source_id=None, operation_id=None, encoding=None, delimiter=None,

               extractor=None):
        operation_id = operation_id or "op_" + uuid4().hex
        owner = self.owner
        if owner is None or not hasattr(owner, "state_lock") or not getattr(owner, "memory", None):
            return OperationResult("failed", operation_id, "session_missing")
        if not isinstance(operation_id, str) or len(operation_id) > 100:
            return OperationResult("failed", "", "invalid_operation_id")
        with owner.state_lock:
            try:
                store = validate_store(owner.memory.data["documents"], self.limits) if "documents" in owner.memory.data else empty_store()
            except DocumentError as exc:
                return OperationResult("failed", operation_id, exc.code)
            existing = store["operations"].get(operation_id)
            if existing:
                return OperationResult(**existing["result"])
            epoch = getattr(owner, "_ui_epoch", 0)
            memory_object = owner.memory
            memory_data = owner.memory.data
            revision = store["revision"]
            context = self._context(query)
        previous = store["sources"].get(source_id) if source_id else None
        if source_id and previous is None:
            return OperationResult("failed", operation_id, "unknown_source")
        result = OperationResult("pending", operation_id)
        extraction = None
        # First finish interpreting retained segments. Subsequent continuation can extend extraction.
        analyzed_before = {s for op in store["operations"].values() for s in op["analyzed_segment_ids"]}
        pending_retained = [s for s in store["segments"].values() if s["source_id"] == source_id and s["segment_id"] not in analyzed_before]
        if path and (not previous or not pending_retained):
            extractor = extractor or extract_isolated
            try:
                extraction = extractor(path, limits=self.limits, operation_id=operation_id, instance_id=store["instance_id"],
                    cursor=previous["coverage"]["cursor"] if previous else None,
                    expected_hash=previous["sha256"] if previous else None, encoding=encoding, delimiter=delimiter)
            except Exception:
                # No commit or generation has happened yet. Preserve the session and
                # return a public code rather than leaving a UI operation unresolved.
                return OperationResult("failed", operation_id, "document_extraction_failed", extraction_status="failed")
            if extraction.error_code:
                return OperationResult("failed", operation_id, extraction.error_code, extraction_status="failed", warnings=extraction.warnings)
            source_id = extraction.source["source_id"]
            previous = store["sources"].get(source_id)
            if previous:
                # Repeated upload is idempotent by bytes. Continue only at the confirmed cursor.
                if not pending_retained and previous["coverage"]["cursor"] is None:
                    extraction.segments = {}
                    extraction.source = copy.deepcopy(previous)
                else:
                    for bucket in ("visited", "empty", "failed"):
                        extraction.source["coverage"][bucket] = sorted(set(previous["coverage"][bucket] + extraction.source["coverage"][bucket]))
                    for key in ("created_at", "active", "participant", "declared_authors"):
                        extraction.source[key] = previous[key]
            store["sources"][source_id] = extraction.source
            for key, segment in extraction.segments.items():
                if key in store["segments"] and store["segments"][key] != segment:
                    return OperationResult("failed", operation_id, "conflicting_segment")
                store["segments"][key] = segment
            result.extraction_status = extraction.status
            result.warnings.extend(extraction.warnings)
        elif not previous:
            return OperationResult("failed", operation_id, "file_required")
        source = store["sources"][source_id]
        result.coverage = copy.deepcopy(source["coverage"])
        if result.extraction_status == "pending":
            result.extraction_status = "partial" if source["coverage"]["cursor"] or source["coverage"]["failed"] else "complete"
        if source["coverage"]["cursor"] and not path:
            result.warnings.append("Recargá el mismo original para extraer lo aún no retenido; se comprobará su hash.")
        op = {"operation_id": operation_id, "instance_id": store["instance_id"], "created_at": now(),
              "submitted_segment_ids": [], "analyzed_segment_ids": [], "attempts": [], "result": {}}
        store["operations"][operation_id] = op
        # Reject quota violations before inference, too.
        try:
            validate_store(store, self.limits)
        except DocumentError as exc:
            return OperationResult("failed", operation_id, exc.code)
        pending = [s for s in store["segments"].values() if s["source_id"] == source_id and s["segment_id"] not in analyzed_before]
        pending.sort(key=lambda s: (s["locator"]["unit"], s["char_start"]))
        deadline = time.monotonic() + self.limits.interpretation_seconds
        calls = 0
        while pending and calls < self.limits.calls and time.monotonic() < deadline:
            # One full segment at a time, so output/citations cannot erase a long block's tail.
            segment = pending[0]
            payload = {"segments": [segment], "source": {k: source[k] for k in ("source_id", "name", "declared_authors")},
                       "context": context, "existing_knowledge": [
                           {"knowledge_id": k["knowledge_id"], "statement": k["statement"][:250], "attribution": k["attribution"]}
                           for k in list(store["knowledge"].values())[-3:]]}
            # Context must fit a conservative character/token bound without cutting a segment.
            while len(packed(payload)) > min(4900, structured_payload_budget(max_tokens=1200)) and payload["context"]["memories"]:
                payload["context"]["memories"].pop(next(iter(payload["context"]["memories"])))
            bridge = getattr(owner, "llm_bridge", None)
            if bridge is None or not hasattr(bridge, "generate_structured"):
                result.error_code = "structured_provider_unavailable"
                break
            calls += 1
            op["submitted_segment_ids"] = list(dict.fromkeys(op["submitted_segment_ids"] + [segment["segment_id"]]))
            try:
                inferred = bridge.generate_structured(payload, operation_id=operation_id,
                    timeout=min(30, max(0, deadline-time.monotonic())), max_tokens=1200)
            except Exception:
                result.error_code = "structured_provider_failed"
                break
            op["attempts"].append({"status": inferred.status, "error_code": inferred.error_code,
                                   "finish_reason": inferred.finish_reason, "truncated": inferred.truncated})
            if time.monotonic() > deadline:
                result.error_code = "interpretation_timeout"
                break
            if inferred.status != "complete":
                result.error_code = inferred.error_code
                if inferred.error_code in {"structured_provider_unavailable", "context_budget_exceeded"}:
                    break
                continue
            try:
                records, analyzed = self._admit(inferred.value, payload, store, operation_id)
            except (DocumentError, TypeError, ValueError, KeyError, AttributeError) as exc:
                result.error_code = exc.code if isinstance(exc, DocumentError) else "invalid_interpretation_structure"
                continue
            result.error_code = None
            op["analyzed_segment_ids"].extend(analyzed)
            for record in records:
                if record["knowledge_id"] not in store["knowledge"]:
                    store["knowledge"][record["knowledge_id"]] = record
                    result.incorporated += record["state"] == "incorporated"
            pending.pop(0)
        result.pending_segments = len(pending)
        result.interpretation_status = "partial" if pending and op["analyzed_segment_ids"] else "failed" if pending else "complete"
        if pending and result.error_code is None:
            result.error_code = "interpretation_budget_exhausted"
        result.status = "partial" if pending or result.extraction_status == "partial" else "complete"
        result.warnings.append("Soporte textual y esquema verificados; ni verdad factual ni implicación semántica están demostradas.")
        op["result"] = asdict(result)
        with owner.state_lock:
            if owner.memory is not memory_object or owner.memory.data is not memory_data or getattr(owner, "_ui_epoch", 0) != epoch:
                return OperationResult("failed", operation_id, "session_changed")
            current = owner.memory.data.get("documents")
            if current and (current["revision"] != revision or current["instance_id"] != store["instance_id"]):
                return OperationResult("failed", operation_id, "document_revision_changed")
            reconcile_memory_refs(store, owner.memory.data.get("chats", []))
            store["revision"] += 1
            try:
                validated = validate_store(store, self.limits)
            except DocumentError as exc:
                return OperationResult("failed", operation_id, exc.code)
            prior_metadata = copy.deepcopy(owner.memory.data.get("metadata", {}))
            owner.memory.data["documents"] = validated
            try:
                if not owner.memory.save():
                    raise DocumentError("document_save_failed")
            except Exception:
                if current is None:
                    owner.memory.data.pop("documents", None)
                else:
                    owner.memory.data["documents"] = current
                owner.memory.data["metadata"] = prior_metadata
                return OperationResult("failed", operation_id, "document_save_failed")
            realtime = getattr(owner, "realtime_memory", None)
            if realtime:
                realtime.cached_data = copy.deepcopy(owner.memory.data)
            return result

    def _admit(self, value, payload, store, operation_id):
        if not isinstance(value, dict) or not isinstance(value.get("claims"), list) or len(value["claims"]) > 8:
            raise DocumentError("invalid_claims")
        allowed = {s["segment_id"]: s for s in payload["segments"]}
        analyzed = value.get("analyzed_segment_ids")
        if not isinstance(analyzed, list) or set(analyzed) != set(allowed) or len(analyzed) != len(set(analyzed)):
            raise DocumentError("invalid_analyzed_ids")
        records = []
        for claim in value["claims"]:
            if not isinstance(claim, dict):
                raise DocumentError("invalid_claim")
            refs = claim.get("support_refs")
            if not isinstance(refs, list) or not refs:
                raise DocumentError("missing_support")
            for ref in refs:
                if not isinstance(ref, dict) or ref.get("segment_id") not in allowed:
                    raise DocumentError("invented_segment")
                quote = ref.get("quote")
                if not isinstance(quote, str) or not quote.strip() or quote not in allowed[ref["segment_id"]]["text"]:
                    raise DocumentError("unsupported_quote")
            attribution = claim.get("attribution")
            if not isinstance(attribution, dict):
                raise DocumentError("invalid_attribution")
            label = attribution.get("label", "")
            if not isinstance(label, str):
                raise DocumentError("invalid_attribution_label")
            if label and not any(label in r["quote"] for r in refs) and label not in payload["source"]["declared_authors"]:
                raise DocumentError("unsupported_attribution_label")
            memory_refs = claim.get("memory_refs", [])
            if not isinstance(memory_refs, list) or any(not isinstance(r, str) or r not in payload["context"]["memories"] for r in memory_refs):
                raise DocumentError("invented_memory")
            relations = claim.get("relations", [])
            existing = {k["knowledge_id"] for k in payload["existing_knowledge"]}
            if not isinstance(relations, list) or any(not isinstance(r, dict) or r.get("target_id") not in existing for r in relations):
                raise DocumentError("invented_relation")
            content = {key: copy.deepcopy(claim.get(key)) for key in ("statement", "claim_kind", "attribution", "support_refs", "relevance_reasons")}
            key = "kn_" + digest(packed({"source": payload["source"]["source_id"],
                **{k: v for k, v in content.items() if k != "relevance_reasons"}}))
            reasons = content.get("relevance_reasons")
            state = "incorporated" if isinstance(reasons, list) and reasons else "candidate"
            record = {**content, "knowledge_id": key, "source_id": payload["source"]["source_id"],
                "instance_id": store["instance_id"], "operation_id": operation_id, "state": state,
                "created_at": now(), "incorporated_at": now() if state == "incorporated" else None,
                "memory_refs": [{"memory_id": r, "status": "resolved"} for r in memory_refs], "relations": relations}
            check = copy.deepcopy(store)
            check["knowledge"][key] = record
            validate_store(check, self.limits)
            records.append(record)
        return records, analyzed

    def set_active(self, source_id, active):
        if type(active) is not bool:
            raise DocumentError("invalid_active")
        owner = self.owner
        with owner.state_lock:
            previous = owner.memory.data.get("documents")
            if not previous or source_id not in previous["sources"]:
                raise DocumentError("unknown_source")
            updated = copy.deepcopy(previous)
            updated["sources"][source_id]["active"] = active
            updated["revision"] += 1
            prior_metadata = copy.deepcopy(owner.memory.data.get("metadata", {}))
            owner.memory.data["documents"] = validate_store(updated, self.limits)
            try:
                if not owner.memory.save():
                    raise DocumentError("document_save_failed")
            except Exception:
                owner.memory.data["documents"] = previous
                owner.memory.data["metadata"] = prior_metadata
                raise DocumentError("document_save_failed")
            realtime = getattr(owner, "realtime_memory", None)
            if realtime:
                realtime.cached_data = copy.deepcopy(owner.memory.data)


class DocumentService(LegacyDocumentService):
    """v2 writer; legacy validation remains available for migration and regression fixtures."""
    def __init__(self, owner, limits=None):
        super().__init__(owner, limits)
        from sofiel_selective import DocumentSessionStore
        self.session = DocumentSessionStore()

    def clear_session(self):
        self.session.clear()

    def evidence(self, query, *, focus=None, max_chars=12000):
        from sofiel_selective import retrieve_v2, search_originals
        with self.owner.state_lock:
            store = self.owner.memory.data.get('documents')
            if not store: return {}
            if store.get('schema_version') == 1: return super().evidence(query)
            session_token=self.session.read(self.owner)
            session=copy.deepcopy(session_token)
            originals=dict(self.session.originals)
            memory=self.owner.memory; data=memory.data; epoch=getattr(self.owner,'_ui_epoch',0)
            store=copy.deepcopy(store)
            atoms=copy.deepcopy(getattr(getattr(self.owner,'synaptic_manager',None),'kernel',[]))
        # Parsing is outside the state lock and never modifies the consolidation cursor.
        extra,diagnostics=search_originals(store,session,originals,query,self.limits,extract_isolated,focus=focus)
        transient={'segments':dict((session or {}).get('segments',{}))}
        transient['segments'].update(extra)
        reserve=len(packed(diagnostics))+30 if diagnostics else 0
        out=retrieve_v2(store,query,atoms=atoms,session=transient,focus=focus,
                        max_chars=max(0,max_chars-reserve),original_sources=originals)
        if diagnostics:
            candidate={**out,'original_search':diagnostics}
            if len(packed(candidate))<=max_chars: out=candidate
        with self.owner.state_lock:
            current=self.owner.memory.data.get('documents',{})
            if (self.owner.memory is not memory or self.owner.memory.data is not data
                or getattr(self.owner,'_ui_epoch',0)!=epoch or current.get('revision')!=store['revision']
                or current.get('instance_id')!=store['instance_id']
                or self.session.data is not session_token or self.session.originals!=originals): return {}
        return out

    def view_store(self):
        """Compatibility presentation only; never serialized or installed in memory."""
        from sofiel_selective import passage
        store=self.owner.memory.data.get('documents',{})
        if store.get('schema_version') != 2: return store
        view=copy.deepcopy(store); view['knowledge']={}
        for key,k in store['insights'].items():
            row=copy.deepcopy(k)
            row['support_refs']=[{'segment_id':r['excerpt_id'],'quote':store['excerpts'][r['excerpt_id']]['text']} for r in k['support_refs']]
            view['knowledge'][key]=row
        return view

    def _durable(self, store):
        from sofiel_selective import empty_v2, migrate_v1
        if not store: return empty_v2()
        return migrate_v1(store,self.owner.memory.data,self.limits) if store.get('schema_version') == 1 else validate_store(store,self.limits)

    def ingest(self, path=None, *, query='', source_id=None, operation_id=None, encoding=None, delimiter=None, extractor=None):
        from sofiel_selective import make_event, add_excerpt, validate_v2, record_document_authors, purpose_index, reconcile_purpose_refs, fit_session, event_context
        from pathlib import Path
        operation_id=operation_id or 'op_'+uuid4().hex
        owner=self.owner
        if owner is None or not hasattr(owner,'state_lock') or not getattr(owner,'memory',None):
            return OperationResult('failed',operation_id,'session_missing')
        if not isinstance(operation_id,str) or not operation_id or len(operation_id)>100:
            return OperationResult('failed','','invalid_operation_id')
        with owner.state_lock:
            raw=owner.memory.data.get('documents')
            try: durable=self._durable(raw)
            except DocumentError as exc: return OperationResult('failed',operation_id,exc.code)
            if operation_id in durable['operations']:
                prior=durable['operations'][operation_id]
                if source_id and prior['source_id']!=source_id:
                    return OperationResult('failed',operation_id,'operation_id_conflict')
                if path:
                    try:
                        with Path(path).open('rb') as stream:
                            submitted=stream.read(self.limits.file_bytes+1)
                        if len(submitted)>self.limits.file_bytes or 'src_'+digest(submitted)!=prior['source_id']:
                            return OperationResult('failed',operation_id,'operation_id_conflict')
                    except OSError: return OperationResult('failed',operation_id,'file_required')
                return OperationResult(**prior['result'])
            epoch=getattr(owner,'_ui_epoch',0); memory=owner.memory; data=memory.data
            original_revision=raw.get('revision',0) if raw else 0
            cached=self.session.read(owner)
            work=copy.deepcopy(cached) if cached else empty_store(durable['instance_id'])
            work['sources']=copy.deepcopy(durable['sources'])
            work.setdefault('analyzed',[])
            context=self._context(query)
            context['purpose_refs']=dict(list(purpose_index(data).items())[-5:])
            participant=data.get('user_profile',{}).get('nombre_primario') or data.get('user_profile',{}).get('name')
        previous=work['sources'].get(source_id) if source_id else None
        if source_id and previous is None: return OperationResult('failed',operation_id,'unknown_source')
        shared_upload=bool(path)
        if not path and cached and source_id in self.session.originals:
            path=self.session.originals[source_id]
        result=OperationResult('pending',operation_id)
        pending_before=[s for s in work['segments'].values() if s['source_id']==source_id and s['segment_id'] not in work['analyzed']]
        if path:
            # A reload after restore must start extraction at the beginning: unselected text is absent.
            cursor=previous['coverage']['cursor'] if previous and cached and not pending_before else None
            if previous and pending_before: cursor=None
            try:
                extraction=(extractor or extract_isolated)(path,limits=self.limits,operation_id=operation_id,
                    instance_id=durable['instance_id'],cursor=cursor,expected_hash=previous['sha256'] if previous else None,
                    encoding=encoding,delimiter=delimiter)
            except Exception: return OperationResult('failed',operation_id,'document_extraction_failed',extraction_status='failed')
            if extraction.error_code: return OperationResult('failed',operation_id,extraction.error_code,extraction_status='failed',warnings=extraction.warnings)
            source_id=extraction.source['source_id']; previous=work['sources'].get(source_id)
            source=copy.deepcopy(extraction.source)
            if previous:
                for key in ('created_at','active','participant','declared_authors'): source[key]=previous[key]
                if not shared_upload: source['name']=previous['name']
                if cached:
                    for bucket in ('visited','empty','failed'):
                        source['coverage'][bucket]=sorted(set(source['coverage'][bucket]+previous['coverage'][bucket]))
                    if pending_before: source['coverage']=copy.deepcopy(previous['coverage'])
            source['availability']='reload_original'
            work['sources'][source_id]=source
            work['segments'].update(copy.deepcopy(extraction.segments))
            result.extraction_status=extraction.status
            result.warnings=extraction.warnings[:4]
        elif not previous: return OperationResult('failed',operation_id,'file_required')
        elif not cached or not any(s['source_id']==source_id for s in work['segments'].values()):
            return OperationResult('failed',operation_id,'original_required')
        source=work['sources'][source_id]; durable['sources']=copy.deepcopy(work['sources'])
        result.coverage=copy.deepcopy(source['coverage'])
        if result.extraction_status=='pending':
            result.extraction_status='partial' if source['coverage']['cursor'] or source['coverage']['failed'] else 'complete'
        # Every new send has its own event; continuation retains the latest declared purpose.
        prior_events=sorted((e for e in durable['sharing_events'].values() if e['source_id']==source_id), key=lambda e:e['created_at'])
        try:
            event=make_event(source_id,query,participant,event_id='share_'+digest(operation_id)) if shared_upload or not prior_events else copy.deepcopy(prior_events[-1])
            record_document_authors(event,(s for s in work['segments'].values() if s['source_id']==source_id))
            durable['sharing_events'][event['event_id']]=event
            if shared_upload and prior_events and prior_events[-1]['stated_purpose'] != event['stated_purpose']:
                # A new purpose may reconsider cached windows, within the same call budget.
                source_segments={s['segment_id'] for s in work['segments'].values() if s['source_id']==source_id}
                work['analyzed']=[sid for sid in work['analyzed'] if sid not in source_segments]
            context['sharing_event']=event_context(event)
            # Bound both temporary and durable data before any provider call. No silent eviction.
            fit_session(work,self.limits)
            if len(durable['operations'])>=self.limits.operations: raise DocumentError('document_history_limit')
            if path and source_id not in self.session.originals:
                total=sum(Path(p).stat().st_size for p in self.session.originals.values())+Path(path).stat().st_size
                if total>self.limits.session_file_bytes: raise DocumentError('document_session_file_limit')
            validate_v2(durable,self.limits)
        except (OSError,DocumentError) as exc:
            return OperationResult('failed',operation_id,exc.code if isinstance(exc,DocumentError) else 'document_original_cache_failed')
        pending=sorted((s for s in work['segments'].values() if s['source_id']==source_id and s['segment_id'] not in work['analyzed']),
                       key=lambda s:(s['locator']['unit'],s['char_start']))
        attempts=[]; admitted=0; deadline=time.monotonic()+self.limits.interpretation_seconds
        # One bounded response per segment. A truncated response is never retried identically in this operation.
        while pending and len(attempts)<self.limits.calls and time.monotonic()<deadline:
            segment=pending[0]
            payload={'segments':[segment], 'source':{k:source[k] for k in ('source_id','name','declared_authors')},
                     'context':copy.deepcopy(context), 'existing_knowledge':[
                        {'knowledge_id':k['knowledge_id'],'statement':k['statement'][:250],'attribution':k['attribution']}
                        for k in list(durable['insights'].values())[-3:]]}
            # Full declarations stay durable. The provider receives identified bounded excerpts.
            payload['context']['sharing_event']['message']=event['message'][:350]
            payload['context']['sharing_event']['stated_purpose']=event['stated_purpose'][:350]
            payload['context']['sharing_event']['corrections']=event['corrections'][-1:]
            while len(packed(payload))>min(4900,structured_payload_budget(max_tokens=1200)) and payload['context']['memories']:
                payload['context']['memories'].pop(next(iter(payload['context']['memories'])))
            bridge=getattr(owner,'llm_bridge',None)
            if bridge is None or not hasattr(bridge,'generate_structured'):
                result.error_code='structured_provider_unavailable'; break
            try:
                inferred=bridge.generate_structured(payload,operation_id=operation_id,
                    timeout=min(30,max(0,deadline-time.monotonic())),max_tokens=1200)
            except Exception:
                result.error_code='structured_provider_failed'; break
            attempts.append({'status':inferred.status,'error_code':inferred.error_code,
                             'finish_reason':inferred.finish_reason,'truncated':inferred.truncated})
            if time.monotonic()>deadline: result.error_code='interpretation_timeout'; break
            if inferred.status!='complete': result.error_code=inferred.error_code; break
            try:
                self._select(inferred.value,payload,durable,event,self.limits)
            except (DocumentError,TypeError,ValueError,KeyError,AttributeError) as exc:
                if isinstance(exc,DocumentError) and exc.code in {'selection_limit','source_store_limit','document_store_limit','document_history_limit'}:
                    return OperationResult('failed',operation_id,exc.code)
                result.error_code=exc.code if isinstance(exc,DocumentError) else 'invalid_interpretation_structure'; break
            work.setdefault('candidates',{})[segment['segment_id']]=copy.deepcopy(inferred.value)
            work['analyzed'].append(segment['segment_id']); pending.pop(0); admitted+=1
        result.pending_segments=len(pending)
        result.interpretation_status='partial' if pending and admitted else 'failed' if pending else 'complete'
        if pending and not result.error_code: result.error_code='interpretation_budget_exhausted'
        result.incorporated=len(durable['insights'])-len(raw.get('insights',raw.get('knowledge',{})) if raw else {})
        result.incorporated=max(0,result.incorporated)
        result.status='partial' if pending or result.extraction_status=='partial' else 'complete'
        result.warnings.append('Selección provisional de lo leído; las citas no prueban la verdad ni la interpretación.')
        durable['operations'][operation_id]={'operation_id':operation_id,'created_at':now(),'event_id':event['event_id'],
            'source_id':source_id,'result':asdict(result),'diagnostics':attempts}
        durable['revision']+=1
        with owner.state_lock:
            if owner.memory is not memory or owner.memory.data is not data or getattr(owner,'_ui_epoch',0)!=epoch:
                return OperationResult('failed',operation_id,'session_changed')
            current=owner.memory.data.get('documents')
            if (current.get('revision',0) if current else 0)!=original_revision or (current and current.get('instance_id')!=durable['instance_id']):
                return OperationResult('failed',operation_id,'document_revision_changed')
            reconcile_memory_refs(durable,owner.memory.data.get('chats',[]))
            reconcile_purpose_refs(durable,owner.memory.data)
            try:
                validated=validate_v2(durable,self.limits)
                fit_session(work,self.limits)
                if work.get('evicted_windows'):
                    result.warnings.append('Caché acotada: algunas ventanas ya interpretadas requieren releer el original; las selecciones guardadas se conservan.')
                    validated['operations'][operation_id]['result']=asdict(result)
                    validated=validate_v2(validated,self.limits)
                staged=self.session.stage_original(path,source_id,self.limits) if path else None
            except (OSError,DocumentError) as exc:
                return OperationResult('failed',operation_id,exc.code if isinstance(exc,DocumentError) else 'document_original_cache_failed')
            try:
                self._commit(validated,current)
            except DocumentError as exc:
                if staged:
                    from pathlib import Path
                    Path(staged).unlink(missing_ok=True)
                return OperationResult('failed',operation_id,exc.code)
            self.session.install(owner,work,self.limits)
            if staged: self.session.originals[source_id]=staged
        return result

    def _select(self,value,payload,durable,event,limits):
        from sofiel_selective import add_excerpt,validate_v2,effective_authorship
        import re
        if not isinstance(value,dict) or not isinstance(value.get('claims'),list) or len(value['claims'])>8:
            raise DocumentError('invalid_claims')
        allowed={s['segment_id']:s for s in payload['segments']}
        ids=value.get('analyzed_segment_ids')
        if not isinstance(ids,list) or set(ids)!=set(allowed) or len(ids)!=len(set(ids)): raise DocumentError('invalid_analyzed_ids')
        candidate=copy.deepcopy(durable)
        for claim in value['claims']:
            if not isinstance(claim,dict): raise DocumentError('invalid_claim')
            refs=claim.get('support_refs'); attribution=claim.get('attribution')
            if not isinstance(refs,list) or not refs or len(refs)>8: raise DocumentError('missing_support')
            if not isinstance(attribution,dict): raise DocumentError('invalid_attribution')
            for ref in refs:
                if not isinstance(ref,dict) or ref.get('segment_id') not in allowed: raise DocumentError('invented_segment')
                quote=ref.get('quote')
                if not isinstance(quote,str) or not quote.strip() or quote not in allowed[ref['segment_id']]['text']: raise DocumentError('unsupported_quote')
            label=attribution.get('label','')
            source_authors=[] if any(c['author'] for c in event['corrections']) else payload['source']['declared_authors']
            declared=(source_authors+[a['label'] for a in effective_authorship(event)
                      if a['status'] in {'user_declared','document_reported'}]) if attribution.get('role')=='author' else []
            if not isinstance(label,str) or (label and not any(label in r['quote'] for r in refs) and label not in declared): raise DocumentError('unsupported_attribution_label')
            memories=claim.get('memory_refs',[]); relations=claim.get('relations',[])
            if not isinstance(memories,list) or any(m not in payload['context']['memories'] for m in memories): raise DocumentError('invented_memory')
            purposes=claim.get('purpose_refs',[])
            if not isinstance(purposes,list) or any(p not in payload['context'].get('purpose_refs',{}) for p in purposes): raise DocumentError('invented_purpose')
            known={k['knowledge_id'] for k in payload['existing_knowledge']}
            if not isinstance(relations,list) or any(not isinstance(r,dict) or r.get('target_id') not in known for r in relations): raise DocumentError('invented_relation')
            reasons=claim.get('relevance_reasons',[])
            if not isinstance(reasons,list) or any(not isinstance(r,str) for r in reasons): raise DocumentError('invalid_relevance')
            # Structural grounding is necessary, not proof of semantic entailment.
            basis=payload['context']['human_message']+' '+payload['context'].get('sharing_event',{}).get('stated_purpose','')+' '+packed(payload['context']['memories'])+' '+packed(payload['context']['symbolic_traits'])+' '+payload['context']['purposes']+' '+' '.join(r['quote'] for r in refs)
            terms=set(re.findall(r'\w{4,}',basis.casefold()))-{'este','esta','texto','documento','para','como','with','that','source','user','sofiel'}
            grounded=[r for r in reasons if len(r.strip())>=24 and len(set(re.findall(r'\w{4,}',r.casefold())) & terms)>=1
                      and not re.fullmatch(r'(es |muy )?(profundo|interesante|relevante)[.! ]*',r.casefold())]
            if not grounded: continue
            support=[{'excerpt_id':add_excerpt(candidate,allowed[r['segment_id']],r['quote'],attribution,limits)} for r in refs]
            sid=payload['source']['source_id']
            key='kn_'+digest(packed({'source':sid,'statement':claim.get('statement'),'kind':claim.get('claim_kind'),'voice':attribution,'support':support}))
            if key in candidate['insights']:
                row=candidate['insights'][key]
                row['event_ids']=list(dict.fromkeys(row['event_ids']+[event['event_id']]))
                continue
            row={'knowledge_id':key,'source_id':sid,'statement':claim.get('statement'),'claim_kind':claim.get('claim_kind'),
                 'attribution':copy.deepcopy(attribution),'support_refs':support,'relevance_reasons':grounded,'state':'incorporated',
                 'created_at':now(),'incorporated_at':now(),'memory_refs':[{'memory_id':m,'status':'resolved'} for m in memories],
                 'purpose_refs':[{'purpose_id':p,'status':'resolved'} for p in purposes],
                 'relations':copy.deepcopy(relations),'event_ids':[event['event_id']],'review_state':'selection_provisional'}
            candidate['insights'][key]=row
        validate_v2(candidate,limits)
        durable.clear();durable.update(candidate)

    def _commit(self, updated, previous):
        owner=self.owner
        metadata=copy.deepcopy(owner.memory.data.get('metadata',{}))
        if previous and previous.get('schema_version') == 1:
            from sofiel_selective import backup_migration
            try:
                backup_migration(owner.memory.data, owner.memory.memory_file)
            except (AttributeError, OSError, ValueError, TypeError):
                raise DocumentError('migration_backup_failed') from None
        owner.memory.data['documents']=updated
        try:
            if not owner.memory.save(): raise DocumentError('document_save_failed')
        except Exception:
            if previous is None: owner.memory.data.pop('documents',None)
            else: owner.memory.data['documents']=previous
            owner.memory.data['metadata']=metadata
            raise DocumentError('document_save_failed') from None
        realtime=getattr(owner,'realtime_memory',None)
        if realtime: realtime.cached_data=copy.deepcopy(owner.memory.data)

    def set_active(self, source_id, active):
        if type(active) is not bool: raise DocumentError('invalid_active')
        with self.owner.state_lock:
            old=self.owner.memory.data.get('documents')
            updated=self._durable(old)
            if source_id not in updated['sources']: raise DocumentError('unknown_source')
            updated['sources'][source_id]['active']=active; updated['revision']+=1
            self._commit(validate_store(updated,self.limits),old)

    def edit_provenance(self, source_id, author='', purpose=''):
        """An explicit optional UI correction; never rewrites literal excerpts or fictional voices."""
        from sofiel_selective import validate_v2, update_authorship_review
        with self.owner.state_lock:
            old=self.owner.memory.data.get('documents'); updated=self._durable(old)
            events=sorted((e for e in updated['sharing_events'].values() if e['source_id']==source_id), key=lambda e:e['created_at'])
            if not events: raise DocumentError('unknown_source')
            event=events[-1]
            event['corrections'].append({'created_at':now(),'author':author,'purpose':purpose})
            if author:
                event['authorship'].append({'label':author,'status':'user_declared','basis':'Corrección explícita en procedencia'})
            if purpose:
                event['stated_purpose']=purpose;event['purpose_status']='explicit'
            update_authorship_review(event)
            updated['revision']+=1
            self._commit(validate_v2(updated,self.limits),old)