UiwoUdZddlmZddlZddlZddlZddlmZmZmZddl m Z m Z m Z m Z ddlmZddlmZmZmZmZdd lmZmZmZmZej4d ZGd d Zdd Zej<j>dej@j>dejBj>dejDj>diZ#de$d<GddZ%y)aRLM Neo-Cortex -- Ebbinghaus Decay Scheduler. Implements the forgetting curve: R = e^(-t/S) where R = retention, t = time since last access, S = memory strength. Manages: - Scheduled decay cycles (daily at 3 AM UTC) - Tier-specific decay policies (aggressive/moderate/conservative/infinite) - REM sleep consolidation (weekly on Sunday at 2 AM UTC) - Memory access tracking for spaced-repetition reinforcement - Dry-run mode for previewing decay operations All storage uses Elestio PostgreSQL + Redis. NO SQLite. NO C: drive. ) annotationsN)datetime timedeltatimezone)AnyDictListOptional)UUID) CustomerTierDecaySchedulerProtocolEntitlementManifest MemoryTier)DECAY_POLICIEScalculate_retentioncalculate_strength should_decayzcore.rlm.decaycFeZdZdZdZ d ddZy) _MemoryRowzIn-memory representation of a memory row during decay processing. This is an internal class -- not part of the public API. It mirrors the columns we read from PostgreSQL for decay calculations. id tenant_idcontent content_hashsurprise_score memory_tier access_count last_accessed created_at vector_idNc ||_||_||_||_||_||_||_||_| |_| |_ yNr) selfrrrrrrrrr r!s '/mnt/e/genesis-system/core/rlm/decay.py__init__z_MemoryRow.__init__9sO" (,&(*$"r#)rintrr rstrrr)rfloatrr)rr(rrr rr! Optional[str])__name__ __module__ __qualname____doc__ __slots__r&r'r%rr,sz I"$(# ## #  #  ### ##!#r'rcftj|jdjS)z;SHA-256 hash of content for dedup and similarity detection.zutf-8)hashlibsha256encode hexdigest)rs r% _content_hashr7Rs# >>'..1 2 < < >>r' aggressivemoderate conservativeinfiniteDict[str, str]TIER_POLICY_MAPcVeZdZdZeZ d ddZddZddZeddZ d ddZ ddZ dd Z d dd Z d d Zd!d Z d"d Zd#dZ d$ d%dZ d$ d&dZ d$ d&dZ d'dZ d(dZ d$ d)dZed*dZy)+DecaySchedulerahManages memory decay cycles per tenant. Implements DecaySchedulerProtocol from contracts.py. Usage: scheduler = DecayScheduler(pg_dsn="postgresql://...", redis_url="redis://...") await scheduler.initialize() result = await scheduler.run_decay_cycle() print(result) # {"deleted": 5, "demoted": 12, "retained": 83} Nc|xs tjjdd|_|xs tjjdd|_|xs tjjdd|_|xs tjjdd|_d|_d|_d|_ d|_ i|_ i|_ y)aInitialize DecayScheduler. Args: pg_dsn: PostgreSQL connection DSN. Falls back to DATABASE_URL env. redis_url: Redis connection URL. Falls back to REDIS_URL env. qdrant_url: Qdrant URL for vector cleanup. Falls back to QDRANT_URL env. qdrant_api_key: Qdrant API key. Falls back to QDRANT_API_KEY env. Raises: ValueError: If pg_dsn cannot be resolved. DATABASE_URL REDIS_URL QDRANT_URLQDRANT_API_KEYNF) osenvironget_pg_dsn _redis_url _qdrant_url_qdrant_api_key_pg_pool_redis_qdrant _initialized_access_buffer_access_timestamps)r$pg_dsn redis_url qdrant_urlqdrant_api_keys r%r&zDecayScheduler.__init__vs$C!C #Frzz~~k2'F%I b)I-U@PRT1U"    "'/179r'cK|js td ddlm}||jddd|_|jr) dd l m }|j|jd |_ |jr. dd lm}||j|j"xsd |_d|_t j)d|jd u|jd u|j$d uy #t $rt jdd |_YwxYw#t tf$r'}t jd |d |_ Yd }~d }~wwxYw#t tf$r'}t jd|d |_Yd }~d }~wwxYww)z|Lazy-init backend connections. Raises: ValueError: If required connection strings are missing. zkDecayScheduler requires a PostgreSQL DSN. Provide pg_dsn argument or set DATABASE_URL environment variable.r)create_async_engineT) pool_size max_overflow pool_pre_pingz,sqlalchemy not available; using mock PG poolN)decode_responseszRedis not available: %s) QdrantClient)urlapi_keyzQdrant not available: %sz7DecayScheduler initialized (PG=%s, Redis=%s, Qdrant=%s))rI ValueErrorsqlalchemy.ext.asynciorXrM ImportErrorloggerwarningrJ redis.asyncioasynciofrom_urlrN ExceptionrK qdrant_clientr_rLrOrPinfo)r$rXaioredisexcr_s r% initializezDecayScheduler.initializesj ||T  ! B/ " DM ?? #0&//OOd0     $6+(( 008D  ! M]]$.[[,\\- /5 ! NNI J DM ! + #8#>"  # + $93?#  $s|F !C/ F  (D1 F >-E+AF /%DF DF E)E F  EF F "F?F F  F cDK|j* |jjd{d|_|j* |jj d{d|_d|_d|_y7P#t$rYYwxYw7+#t$rY4wxYww)z)Gracefully close all backend connections.NF)rMdisposerjrNcloserOrPr$s r%rrzDecayScheduler.closes == $ mm++---!DM ;; " kk'')))DK !.   *  sm B BA>BB B#B$B(B >B B  B  B  B B BB BB c|jS)z%Whether initialize() has been called.)rPrss r%is_initializedzDecayScheduler.is_initializeds   r'c4K|j|d{}|sddddStjtj}d}d}d}g} g} |D]} |xs|j | j } | j} | j | jtj} || z jdz }t|| j| j| }|dk(r!|dz }| j| j|d k(r>|dz }|j!| j"}| j| j|f|dz }|s|j%| | d{|||d}t&j)d |||S7|7&w) aRun decay for one tenant or all tenants if tenant_id is None. Algorithm: 1. Fetch all memories where last_accessed < threshold 2. Calculate retention: R = e^(-t/S) where S = f(access_count, surprise_score) 3. If R < retention_floor and access_count < 3 -> DELETE (forgotten) 4. If R < 0.5 -> demote tier (semantic -> episodic -> working) 5. If R >= 0.5 -> retain as-is Args: tenant_id: Process specific tenant only, or None for all. dry_run: If True, calculate but do not apply changes. policy_override: Override the tenant's decay policy. Useful for testing. Returns: {"deleted": N, "demoted": N, "retained": N} Nr)deleteddemotedretainedtzinfo @)hours_since_accessrrpolicydeleter demotez%Decay cycle complete (dry_run=%s): %s)_fetch_decay_candidatesrnowrutc_resolve_policyrrr{replace total_secondsrrrappendr _demote_tierr_apply_decay_actionsrerl)r$rdry_runpolicy_overridememoriesrrwrxry delete_idsdemote_entriesmemr~ last_access hours_elapsedactionnew_tierresults r%run_decay_cyclezDecayScheduler.run_decay_cycles.55i@@ QA> >ll8<<( " &( C$K(<(z6DecayScheduler.run_rem_consolidation..s a6F6Fr')keyc34K|]}|jywr#)r).0rs r% z7DecayScheduler.run_rem_consolidation..s"Ca1>>"Cs) survivor_idtotal_access_countrr rrrz+REM consolidation complete (dry_run=%s): %s)_fetch_all_memoriesr setdefaultritemslenr)rmaxrsum _apply_mergererl)r$rrrbucketsrprefixrr merge_opsgroup tenant_groupsrt_idt_groupsurvivorr to_delete total_accessoprs r%run_rem_consolidationz$DecayScheduler.run_rem_consolidationOs(11)<<1- -02 7C%%bq)F   vr * 1 1# 6 7*, $]]_ )MFE5zA~:#--(((b188= >"/!4!4!6 ) gw'?!,/( #f5 A7FS k=<H#<sAGGC&G>GG!/GG#AG+G,%G Gc dddS)zReturn cron expressions for decay and REM jobs. Returns: Dict mapping job name to cron expression string. z 0 3 * * *z 0 2 * * 0 decay_cyclerem_consolidationr1rss r%get_cron_schedulez DecayScheduler.get_cron_schedules'!,  r'c K|j|jd}|j|}|(td|dt |j |d{S7w)aEntry point for cron runner. Args: job_name: One of "decay_cycle", "rem_consolidation". Returns: Result dict from the corresponding method. Raises: ValueError: If job_name is not recognized. rNzUnknown scheduled job: r)rrrHrbrr)r$job_namedispatchhandlers r% run_scheduledzDecayScheduler.run_scheduledss //!%!;!; ,,x( ?)(6x}}/02 YsAA(!A&"A(c K|d|}tjtj}|j_ d|}|jj |d{|jj d||jd{y|jj|ddz|j|<||j|<y7v7C#t$r }tjd|Yd}~dd}~wwxYww)aRecord that a memory was accessed (read/search hit). Updates: last_accessed timestamp, access_count += 1. This strengthens the memory against decay. Uses Redis INCR for fast counting, flushed to PG periodically. If Redis is not available, falls back to direct PG update. Non-existent memory_id is silently ignored. Args: tenant_id: Tenant UUID. memory_id: Memory ID (string representation of PG id). :N decay:access:decay:last_access:z Redis access tracking failed: %srr )rrrrrNincrset isoformatrjrerfrQrHrR)r$r memory_id buffer_keyr redis_keyrns r% record_accesszDecayScheduler.record_accesss""{!I;/ ll8<<( ;; " H+J<8 kk&&y111kkoo( 5MMO    # #J 2Q 6 J'/2 +2  HA3GG HsR7D#CC4CCC=DCC D!C<7D<DDcjKd}|j9 g}|jjd23d{}|j|t|jjD]y\}}|j dd}t |d k7r'|\} } |j j|} |j| || r| j#nd d{|dz }{|jj%|j j%|S76|D]}|jj|d{7}|*|j dd}d|}|jj|d{7}|j dd}t |d k7r|\} } |j| t|| d{7|jj||d{7|dz }#t$r!} tjd | Yd} ~ d} ~ wwxYw7Sw) znFlush buffered access counts to PostgreSQL. Returns: Number of memories updated. rNzdecay:access:*rrBrrr rZ)rcount_incrementrzRedis flush failed: %s)rN scan_iterrrHrsplitr_update_access_in_pgr(rrjrerfrrQrrRrclear) r$updatesrrcountrts_key timestampparts tenant_id_strrrntss r%flush_access_bufferz"DecayScheduler.flush_access_buffers/  ;; " >!%!6!67G!H%%#KK$8"&d&9&9&?&?&A!B  J$$S!,E5zQ', $M9((,,Z8B++# %02blln,   qLG  !!# %%'Y%!H !C"&++//#"666E} !$_b!AJ1*>F&*kkoof&= = =I&,,S!4E5zQ /4,M933"+(+E &/4 ++,,S&999qLG+!, >7== > sH3HD!DD!H BH3H0AH3D!!$HE=HFA HG$H4G75 HH3 H- H("H3(H--H3cK|j|d{}|sdddidStjtj}d}i}|D]}|j }|j |jtj}||z jdz }t||j|j} || z }|j} |j| ddz|| <t|} | d| dkDr|| z |dSd|dS7w) zGet decay statistics. Args: tenant_id: Filter to specific tenant, or None for global stats. Returns: Dict with keys: total_memories, total_decayed, avg_retention, etc. Nrg)total_memories total_decayed avg_retentionby_tierrzr|)r}rrr )rrrrrrr{rrrrrrrHr) r$rrrtotal_retentionrrrhoursrtiertotals r%get_decay_statszDecayScheduler.get_decay_stats)s911)<<"#!"!$  ll8<<("$ 5C++K!!))111F ;&557&@E##( --"11A q O??D#KKa014GDM 5 H #8= _u4   HK   ==sD D C4D cKt|dr/|j}||Dcgc]}|j|k(s|}}|S|jgS ddlm}|jj 4d{}|d}|j|d|r t|ndid{}|j}|D cgc]Y} t| dtt| d| d| d t| d | d t| d | d | d| d [c} cdddd{Scc}w77cc} w7#1d{7swYyxYw#t$r"} tj!d| gcYd} ~ Sd} ~ wwxYww)zFetch memories eligible for decay evaluation. In production, this queries PostgreSQL. For testability, this method can be overridden or the internal _memories list can be pre-populated. _test_memoriesNrtextaj SELECT id, tenant_id, content, content_hash, surprise_score, memory_tier, access_count, last_accessed, created_at, vector_id FROM rlm_memories WHERE (:tid IS NULL OR tenant_id = :tid) ORDER BY last_accessed ASC tidr rZrYr rz$Failed to fetch decay candidates: %s)rrrrM sqlalchemyrconnectexecuter)fetchallrr r*r(rjreerror) r$rrrrconnsqlrrowsrowrns r%rz&DecayScheduler._fetch_decay_candidates]s 4) ***H$'/L!1;;)3KALLO == I  '}},,.  $ $||%9Y$G ( $q6"&s3q6{"3 #A%(V',SV}$'F%(Q[&)!f#&q6"%a&    M     8  LL? EI sFD-D-F $E1D22E5-D="D4#D=;AD6D= E'D;(E,F2E4D=6D=;E=EE E EFE E=E82E=3F8E==Fc@K|j|d{S7w)z2Fetch all memories for stats or REM consolidation.N)rrs r%rz"DecayScheduler._fetch_all_memoriess11)<<<D?D&E D& D DD&D#D D#D&"E#D&& E/E E EEc K|jy ddlm}|xs1tjt j j}|jj4d{}|j|d||t|dd{dddd{y7B77 #1d{7swYyxYw#ttf$r!}tjd||Yd}~yd}~wwxYww)zoUpdate access_count and last_accessed in PostgreSQL. Silently ignores non-existent memory_id. Nrrz UPDATE rlm_memories SET access_count = access_count + :inc, last_accessed = :ts WHERE id = :mid )incrmidz Access update skipped for %s: %s)rMrrrrrrrrrr(rbrjredebug)r$rrrrrrrns r%rz#DecayScheduler._update_access_in_pgs ==  M 'H(,,x||"<"F"F"HB}}**,  ll ,2c)nM       I& M LL;Y L L MsC;AC*B-+C.)B3B/B3 C'B1(C,C;-C/B31C3C9B< :CCC;CC8C3.C;3C88C;ctjjtjjtjjtjjtjjtjjtj jtj ji}|j |tjjS)ziDemote a memory tier one level down. semantic -> episodic -> working -> working (floor) )rSEMANTICvalueEPISODICWORKINGDISCARDrH) current_tier demotion_maps r%rzDecayScheduler._demote_tiers    % %z':':'@'@    % %z'9'9'?'?    $ $j&8&8&>&>    $ $j&8&8&>&>   j.@.@.F.FGGr')NNNN)rSr+rTr+rUr+rVr+)returnNone)rbool)NFN)rOptional[UUID]rrrr+rDict[str, int])rr rr))rr r~r)rr)NF)rrrrrr)rr<)rr)rDict[str, Any])rr rr)rr)rr(r#)rrrr)rrrzList[_MemoryRow])r List[int]rz List[tuple]rr)rr(rr(rrrr)rr)rr(rr+rr)rr)rr))r,r-r.r/rr&rorrpropertyrurrrrrrrrrrrrrr staticmethodrr1r'r%r?r?gs $N!%#'$((, ::!:" : & :B1/f"&!!%))- C!CC' C  CRE72%)I!II  I^  8$2$2*-$2 $2L8~+/. '. . j+/4'4 4n+/='= = (C(C$(C  (CT-;-; -; -;  -;f(, MMM% M  M< H Hr'r?)rr)rr))&r/ __future__rr3loggingrFrrrtypingrrr r uuidr contractsr rrr decay_curvesrrrr getLoggerrerr7STARTERr PROFESSIONAL ENTERPRISEQUEENr=__annotations__r?r1r'r%r*s # 22,,    + ,####L? ##Z!!>j #u Hu Hr'