yj#@ddlZddlmZddlmZddlmZmZddlm Z m Z m Z ddl m Z ddlmZddlmZdd lmZdd lmZdd lmZmZdd lmZdd lmZddlmZddlm Z ddl!m"Z"ddl#m$Z$ddl%m&Z&m'Z'm(Z(ddl)m*Z*ddl+m,Z,m-Z-m.Z.ee/Z0ddddde1de1de1de2e1efdzde3de4dzdee5ej6fdZ7dddde1de2e1efdzde3dee5ej6fd Z8d!dd"d#ede1de1de1de4d$e1dzdeej6fd%Z9d!d&d#ede1de1de1de4deej6f d'Z:de3fd(Z; dJde1de1de1d*e dKd#ede1de1de1d.eZHd#ed?ejBde1de1de1de3f d@ZI dLd#edBe-dCe4dDe4de4f dEZJd#ede1d.e>}}V_7<<>>?? zz%   K)rrcttjtjj|ktjjd}t|tj|}|r7|tjj }n6|tjj }|S)ax Get all documents using custom filters. Returns a Select query for pagination support via apaginate(). Results are ordered by created_at timestamp. Args: workspace_name: Name of the workspace filters: Optional filters to apply reverse: Whether to reverse the order (oldest first) Returns: Select query for documents N) rrr&r'r!r(r)rr*r+r,r-)r!rrr.s r/get_documents_with_filtersr3Ss* v v-? @ @ v)--d33 4 4  fow 7 7D@}}V_7;;==>>}}V_7<<>>?? Kr1 )r session_namedbr5cKttjtjj|ktjj|ktjj|ktjjd}|(|tjj |k}| tjj  |}||d{V}|S)a Query most recent documents. Args: db: Database session workspace_name: Name of the workspace observer: Name of the observing peer observed: Name of the observed peer limit: Maximum number of documents to return session_name: Optional session name to filter by Returns: Sequence of documents ordered by created_at descending N)rrr&r'r!r"r#r(r)r5r*r+r-r executescalarsall)r6r!r"r#r r5r.results r/query_documents_recentr<ys. &/ " " ( (&.8 H, H,"&&t,,   Dzz&/6,FGG ==388:: ; ; A A% H HD::d## # # # # # #F >>     ! !!r1)r c*Kttjtjj|ktjj|ktjj|ktjjd tjj  |}| |d{V}|S)a~ Query documents sorted by times_derived (most reinforced first). Args: db: Database session workspace_name: Name of the workspace observer: Name of the observing peer observed: Name of the observed peer limit: Maximum number of documents to return Returns: Sequence of documents ordered by times_derived descending N)rrr&r'r!r"r#r(r)r* times_derivedr-r r8r9r:)r6r!r"r#r r.r;s r/query_documents_most_derivedr?s, v  O *n < O $ 0 O $ 0 O & * *4 0 0   &//4466 7 7 u ::d## # # # # # #F >>     ! !!r1cPtjjdkptjj S)z@Check whether queries should go through pgvector (DB-only) path.pgvector)r VECTOR_STORETYPEMIGRATEDr1r/_uses_pgvectorrFs% "j0V8M8V4Vr1 embeddingtop_k max_distancecKtrdSt}|gS|d|||}i} |rdD]} | |vr || | | <|||||| r| nddd{V} | sgSd| DS)uaQuery external vector store for document IDs sorted by similarity. No DB session needed — safe to call outside a tracked_db scope. Returns: Ordered list of document IDs on the external-store path, empty list when the external store has no results, or None when the pgvector (DB-only) path should be used instead. Ndocument)levelr5F)rIrJrinclude_attributescg|] }|j SrEid).0r;s r/ z6query_external_vector_document_ids..s 3 3 3&FI 3 3 3r1)rFrget_vector_namespacequery) r!r"r#rHrIrJrexternal_vector_store namespacevector_filterskeyvector_resultss r/"query_external_vector_document_idsr[s$t577$ %::NHhI&(N3, 3 3Cg~~&-cls#066!"0:d 7N  3 3N 3 3 33r1 document_idscK|sgSttjtjj|ktjj|ktjj|ktjjdtjj |}t|tj|}| |d{V}d| Dfd|DS)zBFetch documents by IDs, preserving input order. DB-only operation.Nci|] }|j| SrErPrRdocs r/ z*fetch_documents_by_ids..s??????r1c(g|]}|v|SrErE)rRdoc_id documentss r/rSz*fetch_documents_by_ids..s( P P P&Fi(>???I P P P PL P P PPr1cKttjtjj|ktjj|ktjj|ktjjdtjj d}|;|tjj ||k}t|tj|}| tjj ||}||d{V} t!| S)u1pgvector similarity search — pure DB operation.N)rrr&r'r!r"r#rHisnotr(r)cosine_distancerr*r r8listr9r:) r6r!r"r#rHrrJrIr.r;s r/_query_documents_pgvectorrksj v v-? @ @ v'83 4 4 v'83 4 4 v(..t44 5 5 v)--d33 4 4  zz O % 5 5i @ @L P   fow 7 7D ==2BB9MM N N T T   D::d## # # # # # #F   $$&& ' ''r1)rrJrIrHrUc HK|S tj|d{V}n7#t$r*} tdtjjdz| d} ~ wwxYwtr|t||||||||d{VStd4d{V} t| |||||||d{V} | D]} | | | cdddd{VS#1d{VswxYwYt|||||||d{V} | sgS|t||||| |d{VStd4d{V} t| |||| |d{V} | D]} | | | cdddd{VS#1d{VswxYwYdS)aq Query documents using semantic similarity. When *db* is provided the caller owns the session lifetime. When *db* is ``None`` the function opens (and closes) its own short-lived session so that no DB connection is held during external vector-store calls. Args: db: Database session, or None to let the function manage its own workspace_name: Name of the workspace query: Search query text observer: Name of the observing peer observed: Name of the observed peer filters: Optional filters to apply at vector store level (supports: level, session_name) max_distance: Maximum cosine distance for results top_k: Number of results to return embedding: Optional pre-computed embedding for the query (avoids extra API call if possible) Returns: Sequence of matching documents Nz%Query exceeds maximum token limit of .zquery_documents.pgvector)r!r"r#rHrIrJr)r6r!r"r#r\rzquery_documents.fetch) rembed ValueErrorrr EMBEDDINGMAX_INPUT_TOKENSrFrkrexpunger[rf)r6r!rUr"r#rrJrIrHe managed_dbdocsr`r\s r/query_documentsrv<sD .4U;;;;;;;;II   %7'8;;;<    >2         899       Z2        D ( (""3''''                             <%!L   ~+)%           122       j+)%           $ $C   s # # # #                              s8! A%AA8C$$ C.1C.7F FF) deduplicaterdrwcV Kg}g}g}|D]} |rt|| |||d{V} | r#| jd} tjjdkptjj } | rD| jr=tj |||| j | j | j | | j | j| j } n6tj |||| j | j | j | | j | j } | jrd| _|| || | jr|| | jf>#t"$r9}t$d |d | j d |d |d | Yd}~|d}~wwxYw |||d{V|rd |D}t-}||t1tj tj j|d t;jdd{V|d{Vn|d|||}g}|D]?\} }|tA| j||||| j | j d@ |!||d{V|t1tj tj j|d t;jdd{V|d{Vn#tD$rt$#d|t1tj tj j|tj j$dzt;jd{V|d{VYnt"$rt$%d|t1tj tj j|tj j$dzt;jd{V|d{VYnwxYwn<#tL$r/}|'d{VtQd|d}~wwxYw|S)a Create multiple documents with optional duplicate detection. Args: db: Database session documents: List of document creation schemas workspace_name: Name of the workspace observer: Name of the observing peer observed: Name of the observed peer Returns: List of DocumentCreate schemas that were actually inserted (excludes duplicates and failures). r"r#NT) exclude_nonerA) r!r"r#contentrMr>internal_metadatar5rH source_ids) r!r"r#r{rMr>r|r5r}pendingzError adding new document to /: c"g|] \}}|j SrErP)rRr`_s r/rSz$create_documents..sAAA&#qsvAAAr1syncedr sync_state last_sync_at sync_attemptsrLr!r"r#r5rMrQrHmetadataz/Vector store unavailable; leaving docs unsyncedrrz"Unexpected error upserting vectorsz@Failed to create documents due to integrity constraint violation))is_rejected_duplicater model_dumprrBrCrDrHrr&r{rMr>r5r}rappend Exceptionloggererroradd_allcommitrr8r r'rQrevaluesrnowrTr upsert_manyrwarningr exceptionr rollbackr)r6rdr!r"r#rwhoncho_documentsaccepted_documentsdocs_with_embeddingsr` is_duplicate metadata_dictstore_embeddings_in_postgresnew_docrsdoc_idsrVrWvector_recordsrHs r/create_documentsrs./179FH>>=  %:^h&&&        L333FFM %*j86,55 ) ,    /#1%%K)"%"3&3!$!1!m"~   !/#1%%K)"%"3&3!$!1"~   } /%."  # #G , , ,  % %c * * *} F$++Wcm,DEEE    LLnnnAQnnT\nn_gnnklnn    HHHH   _ #$$$ iikk P &AA,@AAAG$=$?$? !%,jj6?++U6?-11'::;;V#+%)XZZ&'iikk!!!!!!!!2FF"  68&:  NC"))$"v&/2@,4,4030@), &&       %&/;;I~VVVVVVVVV**v//v155g>>??'/)-*+ ))++%%%%%%%%' & & &NN#TUUU**v//v155g>>??*0/*G!*K)- ))++%%%%%%%%% & & &$$%IJJJ**v//v155g>>??*0/*G!*K)- ))++%%%%%%%%% & kkmm! N    siE DE  F .FF EU-%CN+)U-+CU)U- CU)&U-(U))U-- V&7*V!!V&)r5 document_idcKtjj|ktjj|ktjj|ktjj|ktjjdg}|(|tjj |kttjj | tj}tt t"||d{V}|jdkrt)d|d|d{VdS)a Soft-delete a document by ID. Sets deleted_at timestamp to mark the document as deleted. The reconciliation job handles vector store cleanup and hard deletion from the database. Args: db: Database session workspace_name: Name of the workspace document_id: ID of the document to delete observer: Name of the observing peer (for authorization) observed: Name of the observed peer (for authorization) session_name: Optional session name to verify document belongs to session Raises: ResourceNotFoundException: If document not found or doesn't match criteria Nr(r Document zA not found or does not belong to the specified collection/session)rr&rQr!r"r#r(r)rr5r r'rrrrr rr8rowcountrr) r6r!rr"r#r5 conditions update_stmtr;s r/delete_documentrhs=6 k)&.8 H, H,"&&t,, J&/6,FGGG &v%z299TXZZ9PP,s#2::k+B+B%B%B%B%B%B%B C CF !' f f f f    ))++r1cK|sgStjj|tjj|ktjj|ktjj|ktjjdg}|(| tjj |kttjj | tjtjjtjj}||d{V}|} |d{Vd| DS)uz Soft-delete multiple documents in a single UPDATE ... RETURNING statement. Returns (id, level) tuples for rows that actually got deleted — i.e. rows that matched the workspace/observer/observed filter and were not already soft-deleted. IDs that didn't match are silently skipped; callers can diff the returned ids against the input to detect misses. Nrc*g|]}|j|jfSrE)rQrM)rRrows r/rSz$delete_documents..s! 0 0 0CSVSY  0 0 0r1)rr&rQrer!r"r#r(r)rr5r r'rrr returningrMr8r:r) r6r!r\r"r#r5rr.r;rowss r/delete_documentsrsK"   |,,&.8 H, H,"&&t,, J&/6,FGGG v    48:: & & 6?%v'< = =  ::d## # # # # # #F ::<.s444 444r1rAT)strictexplicitr) r!r"r#r{rMr>r|r5rH)r!r"r#r{rMr>r|r5r~cg|] }|j SrErPr_s r/rSz'create_observations..[s:::#sv:::r1rrrrLrrzHVector store unavailable for namespace %s; leaving observations unsyncedrz)Unexpected error upserting vectors for %szCFailed to create observations due to integrity constraint violationz'Created %d observations in workspace %s)5set session_idadd observer_id observed_idrrr PeerCreaterrsimple_batch_embedrorstrrrBrCrDziprr&r{rrrrrefreshrr8r r'rQrerrritemsrTrr5rMrrrrrrrr rdebuglen)r6rr!sessions_to_validatepeers_to_validatecollection_pairsrr5 peer_namer"r#contents embeddingsrsrcollection_embeddingsrrHr`collection_keyrV all_doc_idsrrWrrs r/create_observationsrs.  &)UU"%%%-0UUAA > % $ $S^ 4 4 4co...co...cos?@@@@-<< "lN;;;;;;;;;;'OO r>7+=9+M+M+MNNNNNNNNNN/  (& H           54|444H1+>xHHHHHHHH 111!#a&&))q01/1  "j0V8M8V4V!lJtDDDGGY ' /-  "$ ^#   CC/-  "$ ^   C#$$$/3?; !6 6 646 !. 1n-44c95EFFFFg #$$$iikk# " "C**S// ! ! ! ! ! ! ! !!: ; ;::)9:::  ! (**v''v)--k::;;'!%"#       ))++         (='B'B'D'DH &H &$#1FF"  68%'&:NCNN36***"))$"v&/2@,4,4030@), &&       )&/;;I~VVVVVVVVV**v//v155g>>??'/)-*+ ))++%%%%%%%%' & & &NNb!**v//v155g>>??*0/*G!*K)- ))++%%%%%%%%% & & &$$CY**v//v155g>>??*0/*G!*K)- ))++%%%%%%%%% & kkmm! Q     LL1  sd%E E, E''E,'F*ZCSZCZ6Z9CZZZZ [(*[[r`c &Kt|||j||dd|jd{V}|sdS|d}ttj|j}ttj|j}t||z } t||z } t|| dzz} t|| dzz} | | krzt d|jd |jd tj tj j |_|d{VdSt d |jd |jd d S) a Check if a document is a duplicate of an existing document. Uses: 1) Cosine similarity (>=0.95), 2) Token diff for retention. Returns True if both: - the document is deemed a duplicate of an existing document - the existing document is deemed a superior duplicate If the document is not a duplicate, returns False. If the document is a duplicate AND the new document is superior, deletes the existing document and returns False. g?r)r6r!rUr"r#rJrIrHNFrr4z>[DUPLICATE DETECTION] Deleting existing in favor of new. new='z ', existing='z'.z?[DUPLICATE DETECTION] Rejecting new in favor of existing. new='T)rvr{rHrrencodingencoderrrdatetimertimezoneutcr(flush) r6r`r!r"r# similar_docs existing_doc tokens_newtokens_existing unique_newunique_existing score_newscore_existings r/rrs.) %k-         L u?L%.55ckBBCCJ*3::<;OPPQQOZ/122J/J677OJ:?3I))_r-ABNN"" S[  gsg{      #+"3"7"78I8M"N"N hhjju NN|#+||dpdx||| 4r1drV batch_sizeolder_than_minutescKtjtjjtj|z }t t jt jj dt jj |k | d}| |d{V}t|}|sdSi}|D]W} |d| j| j| j} || g| jXt1} |D]n\} } || | d{V| | 8#t8$r*} t:d| d| Yd} ~ gd} ~ wwxYw| r| t?t jt jj | d{V|!d{Vt:"d tG| d tG| S|$d{VdS) a Cleanup soft-deleted documents by removing their vectors and database records. This function implements a two-phase cleanup process for documents that have been soft-deleted (deleted_at is not NULL) Args: db: Database session for executing queries external_vector_store: External vector store instance for deleting vectors batch_size: Maximum number of documents to process per call (default 100) older_than_minutes: Only process documents soft-deleted more than this many minutes ago (default 5). Returns: Count of documents cleaned up (only those where vector deletion succeeded). )minutesNT) skip_lockedrrLzFailed to delete vectors from rz Cleaned up z soft-deleted documents)%rrrr timedeltarrr&r'r(is_notr with_for_updater8rjr9r:rTr!r"r# setdefaultrrQrr delete_manyr rrrrrerrrr)r6rVrrcutoffr.r;rd by_namespacer`rWsuccessfully_deleted_idsidsrss r/cleanup_soft_deleted_documentsrs%,  " "8#4#8 9 9H>)>>    L L    2..55cf====*-&,,..NN 3 N'33IsCC C C C C C C C $ + +C 0 0 0 0 N N N NNLILLLL M M M M M M M M N  -jj 6? # # ) )"&&'?@@           iikk P#677 P P P   +,,, ++-- 1s81G** H4 HHcK|sgSttjtjj|ktjj|tjjd}| |d{V}| S)a" Get multiple documents by their IDs. Args: db: Database session workspace_name: Workspace identifier document_ids: List of document IDs to retrieve Returns: Sequence of documents found (may be fewer than requested if some IDs don't exist) N) rrr&r'r!rQrer(r)r8r9r:)r6r!r\r.r;s r/get_documents_by_idsr]s  &/ " " ( (&.8|,,"&&t,,  D ::d## # # # # # #F >>     ! !!r1ry parent_idc<Kttjtjj|ktjj|gtjjd}|r(|tjj |k}|r(|tjj |k}| |d{V}| S)a Get all observations that have this document as a source/premise. Useful for traversing the reasoning tree upward (source -> derived observations). Uses GIN index on source_ids for efficient lookups. Args: db: Database session workspace_name: Workspace identifier parent_id: Document ID to find children of observer: Optional filter by observer observed: Optional filter by observed Returns: Sequence of documents that reference this document as a source N)rrr&r'r!r}containsr(r)r"r#r8r9r:)r6r!rr"r#r.r;s r/get_child_observationsrxs2 &/ " " ( (&.8"++YK88"&&t,,  D @zz&/2h>??@zz&/2h>??::d## # # # # # #F >>     ! !!r1)rGNN)N)rrG)Mrcollections.abcrloggingrtypingrr sqlalchemyrrr sqlalchemy.enginer sqlalchemy.excr sqlalchemy.ext.asyncior sqlalchemy.sqlr sqlalchemy.sql.functionsrsrcrr src.configrsrc.crud.collectionr src.crud.peerrsrc.crud.sessionrsrc.dependenciesrsrc.embedding_clientrsrc.exceptionsrrrsrc.utils.filterrsrc.vector_storerrr__name__rrdictboolinttupler&r0r3r<r?rFrjfloatr[rfrkrvDocumentCreaterrrrConclusionCreaterrrrrrEr1r/rse $$$$$$----------******))))))//////!!!!!!))))))888888""""""((((((''''''111111 *))))) 8  &*----- - #s(^d " -  - :- E&/ "#----f&* #####s(^d "# #  E&/ "# ####X#$"$"$"$"$" $"  $"  $"*$"fo$"$"$"$"Z """""""""" ""  ""  ""fo""""""""J!%%)/4/4/4/4/4E{ /4  /4 $, /4#s(^d "/4 #Y/4/4/4/4p&* QQQQQ Q s) Q #s(^d " Q &/QQQQ8(((( ( E{ ( #s(^d " ($,( ( &/((((R&*!%$(iiitii i  i  i#s(^d "i$,i iE{T!ifoiiiif}}}}G*+}}  }  }} ' !}}}}N $......  .  .*. ....p $'1'1'1'1'13-'1  '1  '1*'1 %S/'1'1'1'1T#### ####LWW734WW &/ WWWWt??  ??  ?  ? ????J P P P &P P  P  P P P P p"""s)"fo """"@  $"$"$"$"$"$" Dj $" Dj $"fo$"$"$"$"$"$"r1