yj.V @dZddlZddlZddlZddlmZddlmZddlm Z m Z m Z m Z m Z ddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddl m!Z!ddl"m#Z#ddl$m%Z%m&Z&m'Z'ej(e)Z*dZ+dZ,dZ-ej.dZ/deejdzdee0fdZ1eGddZ2e+fdede3de4ej5fdZ6e+fdede3de4ej7fdZ8ded e4ej5ddfd!Z9ded"e4ej7ddfd#Z:ded e4ej5d$e&de;e3e3ffd%Zd$e&d*e2de0fd+Z?d$e&d*e2de0fd,Z@d$e&d*e2de0fd-ZAd*e2de0fd.ZBde2fd/ZCdS)0z Vector store reconciliation job. This module provides a periodic reconciliation job that syncs documents and message embeddings to the vector store on a rolling basis, healing any missed writes. N) dataclass)cast)and_deleteor_selectupdate) AsyncSession)InstrumentedAttribute) ColumnElement)func)models)settings) tracked_db)embedding_client)VectorStoreError)EmbeddingCallPurpose)embedding_call_purpose) VectorRecord VectorStoreget_external_vector_store2 minutes last_sync_atreturnct|d|tjtz kS)zIRows are eligible for sync if never attempted or past the backoff window.N)ris_r now SYNC_BACKOFF)rs C/DATA/AppData/hermes/projects/honcho/src/reconciler/sync_vectors.py_backoff_eligibler%(s: txzzL00  ceZdZUdZdZeed<dZeed<dZeed<dZ eed<dZ eed<e defd Z e defd Z e defd Zd S) ReconciliationMetricsz#Metrics for a reconciliation cycle.rdocuments_synceddocuments_faileddocuments_cleanedmessage_embeddings_syncedmessage_embeddings_failedrc |j|jzSN)r)r,selfs r$ total_syncedz"ReconciliationMetrics.total_synced<$t'EEEr&c |j|jzSr/)r*r-r0s r$ total_failedz"ReconciliationMetrics.total_failed@r3r&c|jSr/)r+r0s r$ total_cleanedz#ReconciliationMetrics.total_cleanedDs %%r&N)__name__ __module__ __qualname____doc__r)int__annotations__r*r+r,r-propertyr2r5r7r&r$r(r(2s--ccs%&s&&&%&s&&& FcFFFXFFcFFFXF&s&&&X&&&r&r(db batch_sizec Kttjt tjjdtjjdkttjj  tjj  |d}||d{V}t!|S)ac Get documents that need to be synced to the vector store. Finds documents where: - not soft-deleted (deleted_at is NULL) - sync_state is "pending" (never synced or retry needed) - Note: "synced" = done forever, "failed" = permanent failure (manual intervention) Uses FOR UPDATE SKIP LOCKED to prevent concurrent processing. NpendingT skip_locked)rrDocumentwherer deleted_atr! sync_stater%rorder_byasc nullsfirstlimitwith_for_updateexecutelistscalarsallr@rAstmtresults r$_get_documents_needing_syncrVIs v  *..t44*i7!&/">??     &/.2244??AA B B z   T * * ::d## # # # # # #F   $$&& ' ''r&cNKttjt tjjdkt tjjtjj  | d}| |d{V}t|S)a Get pending message embeddings that need to be synced to the vector store. Returns only pending embeddings (with full data including embedding vectors). The batch_size limits the number of embeddings returned. Uses FOR UPDATE SKIP LOCKED to prevent concurrent processing and orders by last_sync_at (nulls first) to prioritize never-synced records. Note: "synced" = done forever, "failed" = permanent failure (manual intervention) rCTrDN)rrMessageEmbeddingrGrrIr%rrJrKrLrMrNrOrPrQrRrSs r$$_get_message_embeddings_needing_syncrYis v&''  '2i?!&"9"FGG    &)6::<<GGII J J z   T * * ::d## # # # # # #F   $$&& ' ''r& documentsc VK|sdS|D]}|jdz}|tkrdnd}|ttjtjj|jk||tj d{VdSNfailedrC)rI sync_attemptsr) r_MAX_SYNC_ATTEMPTSrOr rrFrGidvaluesr r")r@rZdoc new_attempts new_states r$_bump_document_sync_attemptsrfs    (1, ,0A A AHHy jj 6? # # U6?%/ 0 0 V$*!XZZ             r& embeddingsc VK|sdS|D]}|jdz}|tkrdnd}|ttjtjj|jk||tj d{VdSr\) r_r`rOr rrXrGrarbr r")r@rgembrdres r$%_bump_message_embedding_sync_attemptsrjs    (1, ,0A A AHHy jj 6* + + U6*-7 8 8 V$*!XZZ            r&external_vector_storecfK|sdSd}d}tjjdkptjj }d|D}i|r d|D}t t jjd5tj |d{V}dddn #1swxYwYt|t|kr6t d t|t|t||d D]\} } | | j<|r| | _n8#t $r+td t|YnwxYwfd |D} | r(t%|| d{V|t| z }i} |D]R} |d| j| j| j} | | g| S| D]-\}}g}g}|D]} t5t6t8dz| j}||n| j}|K|t=| jd|D| j| j| j| j| j d|| |s |!||d{V|"tGtHj%&tHj%j'd|D(dtSj*dd{V|t|z }#tV$rGt d|t%||d{V|t|z }Yt $rGtd|t%||d{V|t|z }Y+wxYw||fS)u Sync a batch of pending documents to the external vector store. Handles three cases for each document: 1. Embedding exists in postgres → use it for external upsert 2. Embedding missing + need postgres storage → re-embed, write to both stores 3. Embedding missing + external-only mode → re-embed, write to external only Returns (synced_count, failed_count). rrrpgvectorcbg|],}tttdz|j*|-Sr/)rrPfloat embedding.0rcs r$ z#_sync_documents..s9Det);S]$K$K$S$S$S$Sr&cg|] }|j Sr?contentrrs r$rtz#_sync_documents..BBB BBBr&reconciliationparent_categoryNz6Re-embedded %s/%s documents; remaining will be retriedFstrictzFailed to re-embed %s documentsc&g|] }|jv |Sr?ra)rsrcfreshly_embeddeds r$rtz#_sync_documents..s-SV;K-K-K-K-K-Kr&documentc,g|]}t|Sr?rprsxs r$rtz#_sync_documents.. ;;;AuQxx;;;r&)workspace_nameobserverobserved session_namelevelrarqmetadatacg|] }|j Sr?r)rsds r$rtz#_sync_documents..s.J.J.Jqt.J.J.Jr&syncedrIrr_z3Vector store unavailable while syncing namespace %sz2Unexpected error syncing documents to namespace %s),r VECTOR_STORETYPEMIGRATEDrr VECTOR_SYNCvaluersimple_batch_embedlenloggerwarningziprarq Exception exceptionrfget_vector_namespacerrr setdefaultappenditemsrrPrpgetrrr upsert_manyrOr rrFrGin_rbr r"r)r@rZrk synced_count failed_countstore_in_postgresdocs_needing_embedcontentsnew_embeddingsrcrifailed_to_embed by_namespacens namespacedocs docs_to_syncvector_recordsexistingrqrs @r$_sync_documentsrs tLL "j0V8M8V4V  02Y YBB/ABBBH'$06 0 U U(8'J8'T'T!T!T!T!T!T!T  U U U U U U U U U U U U U U U >""c*<&=&===L''*++  2N5QQQ ( (S+. ($($'CM ( Y Y Y   >DV@W@W X X X X X Y)O-*2?????????O,,, 68L44 " 7 7 *CL#,   B''..s3333'--//2.2. 4.0 -/ % %CDK$. >>H$06F6J6J366R6R    ! !v;;;;;*-*<$'L$'L(+(8!$           $ $ $ $   .'33I~NN N N N N N N N**v''v)--.J.J\.J.J.JKKLL8$(**TUVV        C -- -LL . . . NNEy   /r<@@ @ @ @ @ @ @ @ C -- -LLL . . .   Di   /r<@@ @ @ @ @ @ @ @ C -- -LLL  .  %%sR,D%1B D%BD%B BD%%2EECN  A P,A P,+P,c K|sdSd}d}tjjdkptjj }d|D}i|r d|D}t t jjd5tj |d{V}dddn #1swxYwYt|t|kr6t d t|t|t||d D]\} } | | j<|r| | _n8#t $r+td t|YnwxYwfd |D} | r(t%|| d{V|t| z }t'd|D} t)t*jjt*jjt*jj| t*jjt*jj} || d{V}i}|D].\}}||g|/i}|D]}tA|D] \}}|||< i}|D]F} |!d| j"}||g| G|#D]\}}g}g}|D]} | j}||n$| j}|*|tK| jd|| jd|D| j| j&| j'd|| |s |(||d{V|tSt*jt*jjd|DdtUj+dd{V|t|z }o#tX$rGt d|t%||d{V|t|z }Yt $rGtd|t%||d{V|t|z }YwxYw||fS)u Sync a batch of pending message embeddings to the external vector store. Handles three cases for each embedding: 1. Embedding exists in postgres → use it for external upsert 2. Embedding missing + need postgres storage → re-embed, write to both stores 3. Embedding missing + external-only mode → re-embed, write to external only Returns (synced_count, failed_count). rmrrnc g|] }|j | Sr/)rqrsris r$rtz,_sync_message_embeddings..Ms&999S]%:%:%:%:r&cg|] }|j Sr?rvrs r$rtz,_sync_message_embeddings..Trxr&ryrzNz?Re-embedded %s/%s message embeddings; remaining will be retriedFr|z(Failed to re-embed %s message embeddingsc&g|] }|jv |Sr?r)rsrirs r$rtz,_sync_message_embeddings..ls-666SV;K-K-K-K-K-Kr&ch|] }|j Sr?) message_idrs r$ z+_sync_message_embeddings..~s===3===r&message_c,g|]}t|Sr?rrs r$rtz,_sync_message_embeddings..rr&)rr peer_namercg|] }|j Sr?r)rses r$rtz,_sync_message_embeddings..s6R6R6Rqt6R6R6Rr&rrzIVector store unavailable while syncing message embeddings to namespace %sz;Unexpected error syncing message embeddings to namespace %s)-rrrrrrrrrrrrrrrarqrrrjrPrrrXrrGrrJrOrRrrrb enumeraterrrrrrrrr r r"r)r@rgrkrrrembs_needing_embedrrrinew_embr message_ids sibling_stmt sibling_rowsembs_by_messageemb_idmsg_idchunk_positionemb_idsposrrrembs embs_to_syncrrrqrs @r$_sync_message_embeddingsr2s tLL "j0V8M8V4V 99!99902 BB/ABBBH'$06 0 U U(8'J8'T'T!T!T!T!T!T!T  U U U U U U U U U U U U U U U >""c*<&=&===U''*++ !$$6u U U U , , W+2 ($,$+CM ,      :C@R>Kv&)6+B+MNN v&155kBB C C &)4f6M6P Q Q **\222222227799L,.O&>>""62..55f====%'N"))++))$W-- ) )KC%(N6 " " )>@L44 " 7 7 3CU V VB''..s3333'--//2.2. 468 -/ % %C}H$06F6J6J366R6R    ! !.CC>#&+ACC;;;;;&)n(+(8%(]       $ $ $ $   .'33I~NN N N N N N N N**v.//v.1556R6R\6R6R6RSSTT8$(**TUVV        C -- -LL . . . NN[   8LII I I I I I I I C -- -LLL . . .   M   8LII I I I I I I I C -- -LLL  .  %%sR,D%1B D%BD%B BD%%2EECR$$A U4A UUolder_than_minutescKtjtjjtj|z }t t jj t jj d t jj |k | d}||d{V}d|D}|sdS|t!t j t jj|d{Vt$dt)|dt)|S) z( Cleanup soft-deleted documents rNTrDcg|] }|d S)rr?)rsrows r$rtz<_cleanup_soft_deleted_documents_pgvector..s...#s1v...r&rz Cleaned up z' soft-deleted documents (pgvector mode))datetimer"timezoneutc timedeltarrrFrarGrHis_notrMrNrOrRrrrdebugr)r@rArcutoffrTrUdoc_idss r$(_cleanup_soft_deleted_documents_pgvectorrsx  " "8#4#8 9 9HKddlm}td4d{V}|||td{V}|s dddd{VdS|xj|z c_|d{V dddd{VdS#1d{VswxYwYdS)zq Clean up a single batch of soft-deleted documents. Returns True if work was done, False otherwise. r)cleanup_soft_deleted_documentsreconciliation_cleanupNrAFT)src.crud.documentrrRECONCILIATION_BATCH_SIZEr+r)rkrrr@cleaneds r$_cleanup_documents_batchrsA@@@@@233       r66  !0                          !!W,!!iikk                              sB +B  BBc8Ktd4d{V}t|td{V}|s dddd{VdS|xj|z c_|d{V dddd{VdS#1d{VswxYwYdS)z Clean up a single batch of soft-deleted documents in pgvector-only mode. Returns True if work was done, False otherwise. reconciliation_pgvector_cleanupNrFT)rrrr+r)rr@rs r$_cleanup_pgvector_batchr0s;<<       @ 4                         !!W,!!iikk                              s B  +B  BBcKt}t}tjtz}|btj|kr/t |d{V}|sntj|k/t d|Stj|krt||d{V}tj|krn|t||d{V}tj|krnNt||d{V}|s|s|st dntj|kt d|S)at Run a complete reconciliation cycle. Runs a rolling sweep to reconcile missing vectors and clean up soft deletes. Uses batching and FOR UPDATE SKIP LOCKED for safe concurrent operation. Each batch operation uses its own database session to avoid holding connections open for the entire cycle duration. Returns metrics about what was synced. Nz5Vector reconciliation cycle completed (pgvector mode)z*No work done, breaking reconciliation loopz%Vector reconciliation cycle completed) r(rtime monotonic"RECONCILIATION_TIME_BUDGET_SECONDSrrinforrrr)rrkdeadlinedid_work docs_work embs_work cleanup_works r$run_vector_reconciliation_cyclerDs$%%G577~"DDH$n))4W========H n))  KLLL .  X % %45JGTTTTTTTT >  x ' ' > !7         >  x ' ' 66KWUUUUUUUU  Y ,  LLE F F F + .  X % %. KK7888 Nr&)Dr;rloggingr dataclassesrtypingr sqlalchemyrrrrr sqlalchemy.ext.asyncior sqlalchemy.orm.attributesr sqlalchemy.sqlr sqlalchemy.sql.functionsr srcr src.configrsrc.dependenciesrsrc.embedding_clientrsrc.exceptionsrsrc.telemetry.eventsrsrc.utils.typesrsrc.vector_storerrr getLoggerr8rrrr`rr#boolr%r(r<rPrFrVrXrYrfrjtuplerrrrrrrrr?r&r$rs !!!!!!88888888888888//////;;;;;;(((((())))))''''''111111++++++555555222222QQQQQQQQQQ  8 $ $%(""x!"--- '(9D(@A4 &&&&&&& &00(((( &/((((D0(((( & !"((((@  FO$      *  V,-      ,{&{&FO${&'{& 38_ {&{&{&{&|W&W&V,-W&'W& 38_ W&W&W&W&x0 B& " *& " *& " 2 " (1/D111111r&