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