11"""
22SynapseMemory — Zero-Knowledge Memory Core with Cognitive Security™
33
4- Orchestrates the full memory pipeline:
5- raw_text → sanitize → validate_intent → encrypt → embed → DP noise → upsert
4+ Orchestrates the secure memory pipeline with built-in sanitization,
5+ intent validation, differential privacy, and self-healing recall.
66
7- Every call to store() returns an audit-ready payload with flags:
8- {"sanitized": True, "privacy_applied": True, "intent": {...}}
9-
10- Recall integrates self-healing: semantically proximate memories with
11- conflicting categories are automatically reclassified via keyword
12- consensus before being returned to the caller.
7+ Every call to store() returns an audit-ready payload.
8+ Enterprise tier extends with pluggable pipeline stages.
139
1410Author: Security & Architecture Team @ Synapse Layer
1511License: Apache 2.0
@@ -214,17 +210,9 @@ async def store(
214210 ).hexdigest ()
215211 memory_id = content_hash [:32 ]
216212
217- # Trust Quotient (TQ) = merged_confidence × validation_score.
218- # TQ is the primary ranking signal for recall operations.
219- trust_quotient = round (
220- validation .confidence * validation .validation_score , 4
221- )
222- # Apply additive confidence boost for CRITICAL memories so they
223- # consistently surface at the top of recall results.
224- if validation .confidence_boost > 0 :
225- trust_quotient = min (
226- trust_quotient + validation .confidence_boost * 0.1 , 1.0
227- )
213+ # Trust Quotient — proprietary ranking signal (Enterprise extends
214+ # with multi-factor TQ including recency decay and agent reputation).
215+ trust_quotient = self ._compute_tq (validation )
228216
229217 record : Dict [str , Any ] = {
230218 'memory_id' : memory_id ,
@@ -545,6 +533,17 @@ def _generate_pseudo_embedding(
545533
546534 return values
547535
536+ def _compute_tq (self , validation : ValidationResult ) -> float :
537+ """Compute Trust Quotient for a validated memory.
538+
539+ OSS uses a baseline formula. Enterprise tier adds recency decay,
540+ agent-reputation weighting, and cross-session normalization.
541+ """
542+ tq = round (validation .confidence * validation .validation_score , 4 )
543+ if validation .confidence_boost > 0 :
544+ tq = min (tq + validation .confidence_boost * 0.1 , 1.0 )
545+ return tq
546+
548547 @staticmethod
549548 def _cosine_similarity (a : List [float ], b : List [float ]) -> float :
550549 """Compute cosine similarity between two vectors.
@@ -563,136 +562,3 @@ def _cosine_similarity(a: List[float], b: List[float]) -> float:
563562
564563 return dot / (norm_a * norm_b )
565564
566-
567- # ══ Inline Tests (run with: python -m synapse_memory.core) ═════════════
568- if __name__ == "__main__" :
569- import asyncio
570-
571- print ("=" * 60 )
572- print ("SynapseMemory — Inline Test Suite (v1.0.5)" )
573- print ("=" * 60 )
574-
575- async def run_tests ():
576- # Test 1: Basic store with full pipeline
577- mem = SynapseMemory (agent_id = "test-agent" )
578- r = await mem .store (
579- content = "User prefers concise answers in Brazilian Portuguese" ,
580- confidence = 0.95 ,
581- )
582- assert r .sanitized is True
583- assert r .privacy_applied is True
584- assert r .trust_quotient > 0
585- assert r .memory_id
586- assert r .validation_details ['final_intent' ] in [
587- 'preference' , 'fact' , 'procedural' , 'bio' ,
588- 'ephemeral' , 'critical' , 'unknown' ,
589- ]
590- assert r .validation_details ['source_type' ] in [
591- 'validated' , 'inference' , 'critical_override' ,
592- ]
593- print (f"[PASS] Basic store: TQ={ r .trust_quotient :.4f}
594- f"intent={ r .validation_details ['final_intent' ]}
595- f"source={ r .validation_details ['source_type' ]} )
596-
597- # Test 2: CRITICAL auto-promotion via keyword
598- r2 = await mem .store (
599- content = "Security breach detected in the authentication system" ,
600- confidence = 0.60 ,
601- )
602- assert r2 .validation_details ['final_intent' ] == 'critical'
603- assert r2 .validation_details ['source_type' ] == 'critical_override'
604- assert r2 .validation_details ['confidence_boost' ] == 1.0
605- print (f"[PASS] Critical override: "
606- f"intent={ r2 .validation_details ['final_intent' ]}
607- f"boost={ r2 .validation_details ['confidence_boost' ]} )
608-
609- # Test 3: Low confidence → inference + warning
610- r3 = await mem .store (
611- content = "Something happened yesterday with something" ,
612- confidence = 0.30 ,
613- )
614- assert r3 .validation_details ['source_type' ] == 'inference'
615- assert r3 .validation_details ['warning' ] is not None
616- print (f"[PASS] Low confidence: source={ r3 .validation_details ['source_type' ]}
617- f"warning present" )
618-
619- # Test 4: Store with PII — must be sanitized
620- r4 = await mem .store (
621- content = "Call john@acme.com at +55 11 99999-8888" ,
622- confidence = 0.85 ,
623- )
624- assert r4 .sanitized is True
625- assert r4 .sanitization_details ['pii_count' ] >= 2
626- print (f"[PASS] PII sanitized: { r4 .sanitization_details ['pii_count' ]} )
627-
628- # Test 5: Recall with results
629- r5 = await mem .recall ("concise answers Portuguese" )
630- assert len (r5 ) > 0
631- assert r5 [0 ].trust_quotient > 0
632- assert r5 [0 ].intent # Has intent field
633- print (f"[PASS] Recall: { len (r5 )}
634- f"TQ={ r5 [0 ].trust_quotient :.4f}
635- f"intent={ r5 [0 ].intent } )
636-
637- # Test 6: Recall self-healing integration
638- # Store two memories with different contents but similar query match
639- await mem .store (
640- content = "User prefers dark mode in all applications" ,
641- confidence = 0.92 ,
642- )
643- await mem .store (
644- content = "User likes dark theme and prefers minimal UI" ,
645- confidence = 0.90 ,
646- )
647- r6 = await mem .recall ("dark mode preference" )
648- assert len (r6 ) > 0
649- print (f"[PASS] Recall self-healing: { len (r6 )} )
650-
651- # Test 7: Disabled sanitization
652- mem_raw = SynapseMemory (
653- agent_id = "raw-agent" ,
654- sanitize_enabled = False ,
655- )
656- r7 = await mem_raw .store (
657- content = "john@test.com is the contact" ,
658- confidence = 0.80 ,
659- )
660- assert r7 .sanitized is False
661- assert r7 .privacy_applied is True
662- print (f"[PASS] Sanitize disabled: sanitized={ r7 .sanitized } )
663-
664- # Test 8: Disabled privacy
665- mem_nodp = SynapseMemory (
666- agent_id = "no-dp-agent" ,
667- privacy_enabled = False ,
668- )
669- r8 = await mem_nodp .store (
670- content = "Important architecture decision for the project" ,
671- confidence = 0.90 ,
672- )
673- assert r8 .sanitized is True
674- assert r8 .privacy_applied is False
675- print (f"[PASS] Privacy disabled: dp={ r8 .privacy_applied } )
676-
677- # Test 9: Invalid agent_id
678- try :
679- SynapseMemory (agent_id = "" )
680- assert False , "Should have raised ValueError"
681- except ValueError :
682- print ("[PASS] Invalid agent_id rejected" )
683-
684- # Test 10: Custom epsilon
685- mem_eps = SynapseMemory (
686- agent_id = "eps-agent" ,
687- privacy_epsilon = 0.1 ,
688- )
689- r10 = await mem_eps .store (
690- content = "Highly sensitive configuration data" ,
691- confidence = 0.99 ,
692- )
693- assert r10 .privacy_details ['epsilon' ] == 0.1
694- print (f"[PASS] Custom ε=0.1: σ={ r10 .privacy_details ['sigma' ]:.4f} )
695-
696- print ("\n ✅ All inline tests passed." )
697-
698- asyncio .run (run_tests ())
0 commit comments