Spaces:
Running
Running
| """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 | |
| 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) | |