from __future__ import annotations import re import time import uuid from datetime import datetime, timezone from difflib import SequenceMatcher from typing import Any, Iterable TOKEN_RE = re.compile(r"[A-Za-z0-9]+(?:['’-][A-Za-z0-9]+)*") STOPWORDS = { "a", "an", "the", "and", "or", "but", "if", "then", "than", "to", "of", "in", "on", "at", "for", "from", "with", "by", "as", "is", "am", "are", "was", "were", "be", "been", "being", "do", "does", "did", "have", "has", "had", "can", "could", "will", "would", "should", "may", "might", "must", "i", "me", "my", "we", "us", "our", "you", "your", "he", "him", "his", "she", "her", "they", "them", "their", "it", "this", "that", "these", "those", "there", "here", "not", "no", "yes", "very", "just", "also", "so", "too", "about", "into", "over", "under", "after", "before", "during", "up", "down", "out", "off", "again", "more", "most", "some", "any", "all", "both", "each", "every", "which", "what", "when", "where", "who", "why", "how", } TEMPORAL_TERMS = { "today", "tomorrow", "yesterday", "now", "later", "soon", "morning", "afternoon", "evening", "night", "week", "month", "year", "monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday", "january", "february", "march", "april", "may", "june", "july", "august", "september", "october", "november", "december", "next", "last", "this", } NONLEXICAL_EDIT_TERMS = { "please", "kindly", "just", "really", "very", "maybe", "perhaps", "actually", "basically", "simply", "also", "then", "now", "well", } COMMON_STYLE_WORDS = { "good", "great", "nice", "bad", "better", "best", "love", "like", "want", "need", "think", "feel", "really", "very", "small", "large", "big", "new", "old", "fast", "slow", "important", "useful", "easy", "hard", "clear", "simple", "quick", "maybe", "perhaps", "probably", "definitely", "soon", } CONTEXT_TAGS = { "software": { "software", "code", "coding", "python", "javascript", "typescript", "api", "sdk", "github", "git", "repository", "repo", "server", "client", "backend", "frontend", "database", "sql", "docker", "deploy", "deployment", "function", "class", "library", "framework", "mcp", "langchain", "llm", "model", "ml", "ai", "embedding", "rag", "agent", "service", "pytorch", "torch", "grpc", "firestore", "bazel", "pub", "sub", "pipeline", "platform", "technical", "tool", "tools", "cto", "laptop", "laptops", "debugging", "workshop", "prototype", "terminal", }, "engineering": { "engineering", "civil", "structural", "column", "columns", "beam", "beams", "concrete", "steel", "seismic", "earthquake", "foundation", "construction", "load", "loads", "design", "joint", "slab", "soil", "bridge", "fan", "fans", "rpm", "motor", "airflow", "air", "delivery", "rjb", "vs30", "opensees", "p-delta", }, "work": { "meeting", "meeting", "client", "customer", "team", "manager", "review", "deadline", "project", "task", "sprint", "standup", "email", "report", "presentation", "launch", "roadmap", "action", "followup", "follow", "call", }, "food": { "food", "fruit", "fruits", "grocery", "groceries", "eat", "eating", "lunch", "dinner", "breakfast", "banana", "bananas", "apple", "apples", "kiwi", "kiwis", "recipe", "restaurant", "meal", "snack", "cook", "cooking", }, "finance": { "finance", "money", "budget", "salary", "tax", "invoice", "payment", "bank", "stock", "stocks", "investment", "invest", "revenue", "profit", "expense", "investor", "investors", "fund", "funding", "portfolio", "partner", "diligence", "market", "round", "seed", "series", }, "personal": { "personal", "journal", "diary", "home", "family", "friend", "friends", "travel", "trip", "vacation", "fitness", "gym", "health", "shopping", "buy", "bought", }, "education": { "class", "lecture", "exam", "assignment", "course", "professor", "college", "university", "study", "jee", "math", "chemistry", "physics", }, "physical": { "wall", "door", "floor", "ceiling", "smudge", "stain", "paint", "mark", "scratch", "dent", "damage", "damaged", "surface", "object", "paper", }, "nature": { "mountain", "landscape", "forest", "river", "tree", "trees", "weather", "autumn", "spring", "summer", "winter", "sky", "view", }, "shopping": { "shop", "shopping", "buy", "bought", "purchase", "store", "order", }, } KIND_VALUES = { "person_name", "organization", "product_service", "project", "place", "technical_term", "acronym", "identifier", "domain_term", "spelling", "casing", "other", } MAX_EVIDENCE = 16 MAX_CONTEXTS = 10 MAX_EXCEPTIONS = 16 MAX_HISTORY = 50 ACTIVE_CONFIDENCE = 0.68 MIN_ACTIVE_POSITIVE = 2 EXPLICIT_ACTIVE_CONFIDENCE = 0.93 NEGATIVE_RATIO_DEMOTE = 0.85 CONTEXT_APPLY_THRESHOLD = 0.60 CONTEXT_STRONG_THRESHOLD = 0.82 CONTEXT_NEGATIVE_BLOCK_THRESHOLD = 0.72 CONTEXT_CONFLICT_THRESHOLD = 0.68 SPECIFIC_KINDS = { "organization", "product_service", "project", "technical_term", "acronym", "identifier", "domain_term", } ORDINARY_CONFLICT_TAGS = { "food", "physical", "nature", "shopping", } def now_iso() -> str: return datetime.now(timezone.utc).isoformat() def clamp(value: float, low: float = 0.0, high: float = 0.99) -> float: try: value = float(value) except (TypeError, ValueError): value = low return max(low, min(high, value)) def normalize_text(text: str) -> str: return re.sub(r"\s+", " ", (text or "").strip()).lower() def tokenize(text: str) -> list[str]: return [match.group(0) for match in TOKEN_RE.finditer(text or "")] def tokenize_with_spans(text: str) -> list[dict[str, Any]]: return [ {"text": match.group(0), "start": match.start(), "end": match.end()} for match in TOKEN_RE.finditer(text or "") ] def similarity(a: str, b: str) -> float: return SequenceMatcher(None, a.lower(), b.lower(), autojunk=False).ratio() def is_stopword(term: str) -> bool: return normalize_text(term) in STOPWORDS def is_capitalized(term: str) -> bool: letters = [char for char in term if char.isalpha()] return bool(letters) and letters[0].isupper() def has_internal_casing(term: str) -> bool: letters = [char for char in term if char.isalpha()] return len(letters) >= 2 and any(char.isupper() for char in letters[1:]) def term_specificity(term: str) -> float: clean = re.sub(r"[^A-Za-z0-9]", "", term) if not clean: return 0.0 if len(clean) >= 12: return 0.90 if len(clean) >= 9: return 0.78 if len(clean) >= 6: return 0.64 if len(clean) >= 4: return 0.40 return 0.15 def extract_context_tags(text: str, note_title: str = "") -> list[str]: source = normalize_text(f"{note_title} {text}") tokens = set(tokenize(source)) tags = [] for tag, vocabulary in CONTEXT_TAGS.items(): hits = len(tokens.intersection(vocabulary)) if hits: tags.append(tag) return tags def build_scene( note_title: str, current_text: str, input_text: str, window: int = 600, ) -> dict[str, Any]: current_tail = (current_text or "")[-window:] current_tail = re.split(r"\n\s*\n", current_tail)[-1].strip() scene_text = re.sub( r"\s+", " ", f"Note: {note_title}. Local context: {current_tail}. Current input: {input_text}", ).strip() input_tags = extract_context_tags(input_text, note_title) local_tags = extract_context_tags(current_tail, note_title) tags = list(dict.fromkeys([*input_tags, *local_tags]))[:16] lowered = normalize_text(input_text) quoted = bool(re.search(r'["“].*?["”]', input_text or "")) code_like = ( "```" in (input_text or "") or ( bool(re.search(r"\b[a-z_][a-z0-9_]*(?:\([^)]*\))?\b", lowered)) and "software" in input_tags ) ) return { "text": scene_text, "input_text": input_text, "tags": tags, "input_tags": input_tags, "local_tags": local_tags, "quoted": quoted, "code_like": code_like, "note_title": note_title, } def context_overlap( current_scene: str, note_title: str, stored_context: str, stored_tags: list[str] | None = None, ) -> float: current_tokens = { token for token in tokenize(current_scene) if len(token) >= 3 and token not in STOPWORDS } stored_tokens = { token for token in tokenize(stored_context) if len(token) >= 3 and token not in STOPWORDS } if not current_tokens or not stored_tokens: lexical = 0.0 else: intersection = len(current_tokens.intersection(stored_tokens)) lexical = intersection / max(1, min(len(current_tokens), len(stored_tokens))) title_bonus = 0.0 if note_title and stored_context: title_tokens = set(tokenize(normalize_text(note_title))) if title_tokens: title_bonus = len(title_tokens.intersection(stored_tokens)) / max(1, len(title_tokens)) current_tags = set(extract_context_tags(current_scene, note_title)) stored_tag_set = set(stored_tags or extract_context_tags(stored_context, "")) tag_score = len(current_tags.intersection(stored_tag_set)) / max( 1, len(current_tags.union(stored_tag_set)) ) if current_tags or stored_tag_set else 0.0 return clamp( 0.52 * lexical + 0.18 * title_bonus + 0.30 * tag_score, 0.0, 1.0, ) def normalize_context_item(item: dict[str, Any]) -> dict[str, Any]: value = dict(item or {}) value.setdefault("note_title", "") value.setdefault("context", "") value.setdefault("tags", []) value.setdefault("timestamp", now_iso()) value.setdefault("strength", 1.0) value["tags"] = list(dict.fromkeys(str(tag) for tag in value.get("tags", []) if str(tag).strip()))[:12] value["context"] = str(value.get("context", ""))[:1800] value["note_title"] = str(value.get("note_title", ""))[:120] value["strength"] = clamp(value.get("strength", 1.0), 0.1, 1.0) return value def normalize_memory(memory: dict[str, Any]) -> dict[str, Any]: item = dict(memory or {}) item.setdefault("id", f"mem_{uuid.uuid4().hex[:10]}") item.setdefault("source", "") item.setdefault("target", "") item.setdefault("kind", "other") item.setdefault("confidence", 0.0) item.setdefault("evidence_count", 0) item.setdefault("positive_evidence_count", 0) item.setdefault("negative_evidence_count", 0) item.setdefault("explicit_evidence_count", 0) item.setdefault("status", "candidate") item.setdefault("created_at", now_iso()) item.setdefault("updated_at", now_iso()) item.setdefault("last_used_at", None) item.setdefault("last_reason", "") item.setdefault("aliases", []) item.setdefault("positive_contexts", []) item.setdefault("negative_contexts", []) item.setdefault("exceptions", []) item.setdefault("evidence", []) item.setdefault("scope", "global") if item["kind"] not in KIND_VALUES: item["kind"] = "other" item["source"] = str(item.get("source", "")).strip() item["target"] = str(item.get("target", "")).strip() item["confidence"] = round(clamp(item.get("confidence", 0.0)), 4) item["evidence_count"] = int(item.get("evidence_count", 0) or 0) item["positive_evidence_count"] = int(item.get("positive_evidence_count", 0) or 0) item["negative_evidence_count"] = int(item.get("negative_evidence_count", 0) or 0) item["explicit_evidence_count"] = int(item.get("explicit_evidence_count", 0) or 0) if not item["positive_evidence_count"] and item["evidence_count"]: item["positive_evidence_count"] = item["evidence_count"] item["aliases"] = list(dict.fromkeys( str(alias).strip() for alias in item.get("aliases", []) if str(alias).strip() ))[:12] item["positive_contexts"] = [ normalize_context_item(x) for x in item.get("positive_contexts", []) if isinstance(x, dict) ][-MAX_CONTEXTS:] item["negative_contexts"] = [ normalize_context_item(x) for x in item.get("negative_contexts", []) if isinstance(x, dict) ][-MAX_CONTEXTS:] item["exceptions"] = [ normalize_context_item(x) for x in item.get("exceptions", []) if isinstance(x, dict) ][-MAX_EXCEPTIONS:] item["evidence"] = [ dict(x) for x in item.get("evidence", []) if isinstance(x, dict) ][-MAX_EVIDENCE:] return item def _contexts_similarity(scene_text: str, note_title: str, contexts: list[dict[str, Any]]) -> float: if not contexts: return 0.0 return max( context_overlap( scene_text, note_title, item.get("context", ""), item.get("tags", []), ) * float(item.get("strength", 1.0)) for item in contexts ) def memory_context_score( memory: dict[str, Any], scene: dict[str, Any], ) -> dict[str, float]: scene_text = scene.get("text", "") note_title = scene.get("note_title", "") positive = _contexts_similarity( scene_text, note_title, memory.get("positive_contexts", []), ) negative = _contexts_similarity( scene_text, note_title, memory.get("negative_contexts", []), ) if not memory.get("positive_contexts") and memory.get("scope", "global") == "global": positive = 0.62 exception_score = _contexts_similarity( scene_text, note_title, memory.get("exceptions", []), ) comparison_text = scene.get("input_text", "") or scene_text for item in memory.get("exceptions", []): if item.get("note_title") != note_title: continue exception_score = max( exception_score, similarity(comparison_text, item.get("context", "")), similarity(normalize_text(comparison_text), normalize_text(item.get("context", ""))), ) return { "positive": round(positive, 4), "negative": round(negative, 4), "exception": round(exception_score, 4), } def recompute_statuses(memories: list[dict[str, Any]]) -> list[dict[str, Any]]: normalized = [normalize_memory(memory) for memory in (memories or [])] for memory in normalized: if memory["status"] == "rejected": continue positive = memory["positive_evidence_count"] negative = memory["negative_evidence_count"] explicit = memory["explicit_evidence_count"] if explicit > 0 and memory["confidence"] >= EXPLICIT_ACTIVE_CONFIDENCE: memory["status"] = "active" elif ( positive >= MIN_ACTIVE_POSITIVE and memory["confidence"] >= ACTIVE_CONFIDENCE ): memory["status"] = "active" else: memory["status"] = "candidate" by_source: dict[str, list[dict[str, Any]]] = {} for memory in normalized: if memory["status"] == "rejected": continue by_source.setdefault(normalize_text(memory["source"]), []).append(memory) for group in by_source.values(): targets = {normalize_text(memory["target"]): memory for memory in group} if len(targets) <= 1: continue active = [memory for memory in group if memory["status"] == "active"] if len(active) <= 1: continue for index, left in enumerate(active): for right in active[index + 1:]: conflict = 0.0 for left_ctx in left.get("positive_contexts", []): for right_ctx in right.get("positive_contexts", []): conflict = max( conflict, context_overlap( left_ctx.get("context", ""), left_ctx.get("note_title", ""), right_ctx.get("context", ""), right_ctx.get("tags", []), ), ) if conflict >= CONTEXT_CONFLICT_THRESHOLD or ( not left.get("positive_contexts") and not right.get("positive_contexts") ): left["status"] = "ambiguous" right["status"] = "ambiguous" return normalized def same_memory(memory: dict[str, Any], source: str, target: str) -> bool: return ( normalize_text(memory.get("source", "")) == normalize_text(source) and normalize_text(memory.get("target", "")) == normalize_text(target) ) def find_memory(memories: Iterable[dict[str, Any]], source: str, target: str) -> dict[str, Any] | None: for memory in memories: if same_memory(memory, source, target): return memory return None def _contains_term(text: str, term: str) -> bool: if not term: return False return bool( re.search( rf"(? tuple[float, str]: source_norm = normalize_text(source) target_norm = normalize_text(target) if source_norm == target_norm: if source == target: return 0.0, "ignored" if is_capitalized(target) or has_internal_casing(target): return 0.78, "casing" return 0.55, "casing" if is_stopword(source) or is_stopword(target): return 0.0, "grammar" if ( source_norm in NONLEXICAL_EDIT_TERMS or target_norm in NONLEXICAL_EDIT_TERMS ): return 0.0, "style" source_tokens = tokenize(source) target_tokens = tokenize(target) if source_tokens and target_tokens: if all(normalize_text(x) in TEMPORAL_TERMS for x in source_tokens + target_tokens): return 0.0, "temporal" if all(normalize_text(x) in NONLEXICAL_EDIT_TERMS for x in source_tokens + target_tokens): return 0.0, "style" if len(source_tokens) == 1 and len(target_tokens) == 1: if source_norm in COMMON_STYLE_WORDS or target_norm in COMMON_STYLE_WORDS: return 0.0, "generic_style" ratio = similarity(source, target) score = 0.42 if ratio >= 0.94: score += 0.22 elif ratio >= 0.84: score += 0.18 elif ratio >= 0.72: score += 0.12 elif ratio >= 0.60: score += 0.05 if is_capitalized(target) and not is_capitalized(source): score += 0.10 if has_internal_casing(target): score += 0.08 if any(char.isdigit() for char in target): score += 0.08 specificity = term_specificity(target) if specificity >= 0.75: score += 0.12 elif specificity >= 0.60: score += 0.08 if ratio < 0.58 and not is_capitalized(target) and not has_internal_casing(target): return 0.0, "generic_or_broad" return round(min(score, 0.96), 4), "lexical_term" def _compact_alnum(text: str) -> str: return re.sub(r"[^a-z0-9]", "", (text or "").lower()) NUMBER_WORDS = { "zero", "one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten", "eleven", "twelve", "thirteen", "fourteen", "fifteen", "sixteen", "seventeen", "eighteen", "nineteen", "twenty", "thirty", "forty", "fifty", "sixty", "seventy", "eighty", "ninety", } def _phrase_candidate(old_raw: str, new_raw: str) -> dict[str, Any] | None: source = re.sub(r"\s+", " ", (old_raw or "").strip()) target = re.sub(r"\s+", " ", (new_raw or "").strip()) if not source or not target or source == target: return None source_tokens = tokenize(source) target_tokens = tokenize(target) if not source_tokens or not target_tokens: return None meaningful_source_tokens = [token for token in source_tokens if normalize_text(token) not in {"a", "an", "the"}] meaningful_target_tokens = [token for token in target_tokens if normalize_text(token) not in {"a", "an", "the"}] if not meaningful_source_tokens or not meaningful_target_tokens: return None if all(token.lower() in COMMON_STYLE_WORDS for token in meaningful_source_tokens + meaningful_target_tokens): return None compact_source = _compact_alnum(source) compact_target = _compact_alnum(target) if compact_source == compact_target: punctuation_changed = bool( re.sub(r"[A-Za-z0-9]", "", source) != re.sub(r"[A-Za-z0-9]", "", target) ) kind = "spelling" if punctuation_changed else "casing" confidence = 0.82 else: compact_ratio = similarity(compact_source, compact_target) source_token_norms = {normalize_text(token) for token in source_tokens} acronym_like = any(char.isdigit() for char in target) or len(target.replace(" ", "")) <= 6 and target.replace(" ", "").isalnum() and any(token in NUMBER_WORDS for token in source_token_norms) if compact_ratio < 0.64 and not acronym_like: return None kind = "acronym" if acronym_like else "lexical_term" confidence = 0.70 if acronym_like and compact_ratio < 0.64 else 0.72 + max(0.0, compact_ratio - 0.72) * 0.45 return { "source": source, "target": target, "kind": kind, "confidence": round(min(confidence, 0.93), 4), } def candidate_from_groups(old_group: list[str], new_group: list[str]) -> dict[str, Any] | None: if not old_group or not new_group or len(old_group) > 5 or len(new_group) > 5: return None source = " ".join(old_group).strip() target = " ".join(new_group).strip() if not source or not target: return None old_norm = [normalize_text(x) for x in old_group] new_norm = [normalize_text(x) for x in new_group] if old_norm == new_norm: if source != target: confidence = 0.80 if is_capitalized(target) or has_internal_casing(target) else 0.74 return { "source": source, "target": target, "kind": "casing", "confidence": confidence, } return None if len(old_norm) < len(new_norm): for start in range(len(new_norm) - len(old_norm) + 1): if new_norm[start:start + len(old_norm)] == old_norm: extras = new_norm[:start] + new_norm[start + len(old_norm):] if extras and all(token in NONLEXICAL_EDIT_TERMS for token in extras): return None if len(new_norm) < len(old_norm): for start in range(len(old_norm) - len(new_norm) + 1): if old_norm[start:start + len(new_norm)] == new_norm: extras = old_norm[:start] + old_norm[start + len(new_norm):] if extras and all(token in NONLEXICAL_EDIT_TERMS for token in extras): return None if len(old_group) > 1 or len(new_group) > 1: phrase = _phrase_candidate(source, target) if phrase is not None: return phrase confidence, kind = score_word_change(source, target) if confidence <= 0: return None compact = similarity(source.replace(" ", ""), target.replace(" ", "")) if len(old_group) > 1 or len(new_group) > 1: confidence = max(confidence, 0.62) if compact >= 0.66: confidence += 0.07 if is_capitalized(new_group[0]): confidence += 0.05 if compact < 0.46 and not is_capitalized(new_group[0]) and not has_internal_casing(target): return None confidence = min(confidence, 0.95) elif confidence < 0.55: return None return { "source": source, "target": target, "kind": kind, "confidence": round(confidence, 4), } def extract_edit_candidates(before: str, after: str) -> list[dict[str, Any]]: old_tokens = tokenize_with_spans(before) new_tokens = tokenize_with_spans(after) old_values = [token["text"] for token in old_tokens] new_values = [token["text"] for token in new_tokens] matcher = SequenceMatcher(None, old_values, new_values, autojunk=False) replace_ops = [opcode for opcode in matcher.get_opcodes() if opcode[0] == "replace"] if not replace_ops: return [] total_old = sum(opcode[2] - opcode[1] for opcode in replace_ops) total_new = sum(opcode[4] - opcode[3] for opcode in replace_ops) if total_old > 10 or total_new > 10 or len(replace_ops) > 4: return [] candidates: list[dict[str, Any]] = [] for _, old_start, old_end, new_start, new_end in replace_ops: old_group = old_values[old_start:old_end] new_group = new_values[new_start:new_end] if old_start < len(old_tokens) and old_end > old_start: old_raw = before[old_tokens[old_start]["start"]:old_tokens[old_end - 1]["end"]] else: old_raw = " ".join(old_group) if new_start < len(new_tokens) and new_end > new_start: new_raw = after[new_tokens[new_start]["start"]:new_tokens[new_end - 1]["end"]] else: new_raw = " ".join(new_group) if len(old_group) > 1 or len(new_group) > 1: phrase = _phrase_candidate(old_raw, new_raw) if phrase is not None: token_start = new_tokens[new_start]["start"] if new_start < len(new_tokens) else 0 token_end = new_tokens[new_end - 1]["end"] if new_end > new_start else token_start phrase["context"] = context_for_change(after, token_start, token_end) candidates.append(phrase) continue expanded = [] if len(old_group) == len(new_group): for index, (old_token, new_token) in enumerate(zip(old_group, new_group)): if normalize_text(old_token) == normalize_text(new_token) and old_token == new_token: continue candidate = candidate_from_groups([old_token], [new_token]) if candidate is not None: token_start = new_tokens[new_start + index]["start"] token_end = new_tokens[new_start + index]["end"] candidate["context"] = context_for_change(after, token_start, token_end) expanded.append(candidate) if expanded and len(expanded) <= 3: candidates.extend(expanded) continue candidate = candidate_from_groups(old_group, new_group) if candidate is None: continue if new_start < len(new_tokens) and new_end > new_start: context_start = new_tokens[new_start]["start"] context_end = new_tokens[min(new_end - 1, len(new_tokens) - 1)]["end"] candidate["context"] = context_for_change(after, context_start, context_end) candidates.append(candidate) if len(candidates) > 1: candidates = [candidate for candidate in candidates if candidate["confidence"] >= 0.58] return candidates[:6] def context_for_change(text: str, start: int, end: int, radius: int = 140) -> str: left = max(0, start - radius) right = min(len(text), end + radius) return re.sub(r"\s+", " ", text[left:right].strip()) def _add_positive_context( memory: dict[str, Any], note_title: str, context: str, strength: float = 1.0, ) -> None: memory["positive_contexts"].append( normalize_context_item( { "note_title": note_title, "context": context, "tags": extract_context_tags(context, note_title), "strength": strength, "timestamp": now_iso(), } ) ) memory["positive_contexts"] = memory["positive_contexts"][-MAX_CONTEXTS:] def _add_negative_context( memory: dict[str, Any], note_title: str, context: str, kind: str, ) -> None: memory["negative_contexts"].append( normalize_context_item( { "note_title": note_title, "context": context, "tags": extract_context_tags(context, note_title), "strength": 1.0, "timestamp": now_iso(), "type": kind, } ) ) memory["negative_contexts"] = memory["negative_contexts"][-MAX_CONTEXTS:] def _exception_exists(memory: dict[str, Any], note_title: str, context: str) -> bool: return any( item.get("note_title") == note_title and context_overlap( context, note_title, item.get("context", ""), item.get("tags", []), ) >= 0.76 for item in memory.get("exceptions", []) ) def _record_exception( memory: dict[str, Any], note_title: str, context: str, reason: str, ) -> None: if not _exception_exists(memory, note_title, context): memory["exceptions"].append( normalize_context_item( { "note_title": note_title, "context": context, "tags": extract_context_tags(context, note_title), "strength": 1.0, "timestamp": now_iso(), "type": "one_off_override", "reason": reason, } ) ) memory["exceptions"] = memory["exceptions"][-MAX_EXCEPTIONS:] def detect_generation_overrides( before: str, after: str, memories: list[dict[str, Any]], last_generation: dict[str, Any] | None, note_title: str = "", ) -> tuple[list[dict[str, Any]], dict[str, Any]]: memories = [normalize_memory(memory) for memory in (memories or [])] if not last_generation or not isinstance(last_generation, dict): return memories, {"type": "no_generation_override", "candidates": []} if last_generation.get("note_title") != note_title: return memories, {"type": "no_generation_override", "candidates": []} generation_changes = last_generation.get("changes") or [] if not generation_changes: return memories, {"type": "no_generation_override", "candidates": []} reversal_candidates = extract_edit_candidates(before, after) if not reversal_candidates: return memories, {"type": "no_generation_override", "candidates": []} applied = [] for change in generation_changes: memory_id = change.get("memory_id") original = str(change.get("original", "")) replacement = str(change.get("replacement", "")) memory = next( (item for item in memories if item["id"] == memory_id), None, ) if memory is None: continue reversed_here = False for candidate in reversal_candidates: candidate_source = str(candidate.get("source", "")) candidate_target = str(candidate.get("target", "")) if memory.get("kind") == "casing": if candidate_source == replacement and candidate_target == original: reversed_here = True break elif ( normalize_text(candidate_source) == normalize_text(replacement) and normalize_text(candidate_target) == normalize_text(original) ): reversed_here = True break if not reversed_here: continue context = after _record_exception( memory, note_title, context, "The user reversed a Lexi-generated substitution for this scene.", ) _add_negative_context( memory, note_title, context, "generated_override", ) memory["negative_evidence_count"] += 1 memory["evidence_count"] += 1 memory["updated_at"] = now_iso() memory["last_reason"] = ( "The user explicitly overrode this generated substitution for the current scene." ) memory.setdefault("evidence", []).append( { "type": "generated_override", "before": before, "after": after, "context": context, "note_title": note_title, "polarity": "negative", "timestamp": now_iso(), } ) memory["evidence"] = memory["evidence"][-MAX_EVIDENCE:] applied.append( { "memory_id": memory["id"], "source": memory["source"], "target": memory["target"], } ) if not applied: return memories, {"type": "no_generation_override", "candidates": []} recompute_statuses(memories) return memories, { "type": "memory_override", "message": ( "Lexi recorded a contextual override of its own substitution and did not create a reverse memory." ), "candidates": applied, } def _find_conflicting_memory( memories: list[dict[str, Any]], source: str, target: str, ) -> dict[str, Any] | None: source_norm = normalize_text(source) target_norm = normalize_text(target) for memory in memories: memory_source = str(memory.get("source", "")) memory_target = str(memory.get("target", "")) if memory.get("kind") in {"casing", "spelling"}: if ( memory_source != memory_target and source == memory_target and _compact_alnum(target) == _compact_alnum(memory_source) and target != memory_target ): return memory continue if ( source_norm != target_norm and normalize_text(memory_source) == target_norm and normalize_text(memory_target) == source_norm ): return memory return None def detect_memory_reversals( before: str, after: str, memories: list[dict[str, Any]], note_title: str = "", ) -> tuple[list[dict[str, Any]], dict[str, Any]]: """Detect user edits that reverse an existing memory mapping. This is deterministic and runs before LLM learning. Reversing a known mapping is negative evidence for that memory in the current context, not a request to create a reverse memory. """ memories = [normalize_memory(memory) for memory in (memories or [])] candidates = extract_edit_candidates(before, after) if not candidates: return memories, { "type": "no_memory_reversal", "candidates": [], } negative_candidates = [] seen = set() for candidate in candidates: observed_source = str(candidate.get("source", "")).strip() observed_target = str(candidate.get("target", "")).strip() if not observed_source or not observed_target: continue memory = _find_conflicting_memory( memories, observed_source, observed_target, ) if memory is None: continue if memory.get("status") == "rejected": continue context = str(candidate.get("context", "") or after) key = ( memory["id"], normalize_text(context), normalize_text(observed_source), normalize_text(observed_target), ) if key in seen: continue seen.add(key) _record_exception( memory, note_title, context, "The user manually reversed a learned memory in this context.", ) _add_negative_context( memory, note_title, context, "manual_reverse", ) memory["negative_evidence_count"] += 1 memory["evidence_count"] += 1 memory["updated_at"] = now_iso() memory["last_reason"] = ( "The user manually reversed this learned substitution in this context." ) memory.setdefault("evidence", []).append( { "type": "manual_reverse", "before": before, "after": after, "context": context, "note_title": note_title, "polarity": "negative", "timestamp": now_iso(), } ) memory["evidence"] = memory["evidence"][-MAX_EVIDENCE:] negative_candidates.append( { "memory_id": memory["id"], "source": memory["source"], "target": memory["target"], "observed_source": observed_source, "observed_target": observed_target, "context": context, "negative_evidence_count": memory["negative_evidence_count"], "confidence": round(memory["confidence"], 3), "action": "negative_evidence", } ) if not negative_candidates: return memories, { "type": "no_memory_reversal", "candidates": [], } recompute_statuses(memories) return memories, { "type": "memory_negative_evidence", "message": ( "Lexi recorded negative evidence for the existing memory " "in this context. It did not create a reverse memory." ), "candidates": negative_candidates, } def incorporate_llm_proposals( memories: list[dict[str, Any]], proposals: list[dict[str, Any]], before: str, after: str, note_title: str = "", decision: str = "learn", ) -> tuple[list[dict[str, Any]], dict[str, Any]]: memories = [normalize_memory(memory) for memory in (memories or [])] learned = [] conflicts = [] for raw in proposals or []: source = str(raw.get("source", "")).strip() target = str(raw.get("target", "")).strip() kind = str(raw.get("kind", "other")) if not source or not target or source == target: continue if normalize_text(source) == normalize_text(target) and kind != "casing": continue if not _contains_term(before, source) or not _contains_term(after, target): continue if kind not in KIND_VALUES: kind = "other" confidence = clamp(raw.get("confidence", 0.0)) explicit = bool(raw.get("explicit", False)) rationale = str(raw.get("rationale", "")).strip() aliases = [source, *raw.get("aliases", [])] compact_source = _compact_alnum(source) if compact_source and compact_source != normalize_text(source): aliases.append(compact_source) scope = str(raw.get("scope", "global")) if scope not in {"global", "note"}: scope = "global" reverse = _find_conflicting_memory(memories, source, target) if reverse is not None: _add_negative_context(reverse, note_title, after, "reverse_evidence") reverse["negative_evidence_count"] += 1 reverse["evidence_count"] += 1 reverse["updated_at"] = now_iso() reverse["last_reason"] = rationale or "The user supplied evidence that conflicts with an existing mapping." conflicts.append( { "existing_memory_id": reverse["id"], "existing_mapping": f'{reverse["source"]} → {reverse["target"]}', "observed_mapping": f"{source} → {target}", } ) continue existing = find_memory(memories, source, target) evidence_context = str(raw.get("context", "") or after) evidence = { "type": raw.get("evidence_type", "direct_correction"), "before": before, "after": after, "context": evidence_context, "note_title": note_title, "confidence": confidence, "explicit": explicit, "timestamp": now_iso(), } if existing is None: existing = normalize_memory( { "id": f"mem_{uuid.uuid4().hex[:10]}", "source": source, "target": target, "kind": kind, "confidence": confidence, "evidence_count": 1, "positive_evidence_count": 1, "negative_evidence_count": 0, "explicit_evidence_count": 1 if explicit else 0, "status": "candidate", "created_at": now_iso(), "updated_at": now_iso(), "evidence": [evidence], "aliases": list(dict.fromkeys(aliases))[:12], "scope": scope, "last_reason": rationale, } ) memories.append(existing) _add_positive_context(existing, note_title, evidence_context, 1.0) action = "created" else: existing["positive_evidence_count"] += 1 existing["evidence_count"] += 1 existing["confidence"] = clamp( max(existing["confidence"], confidence) + (0.06 if explicit else 0.06) ) if explicit: existing["explicit_evidence_count"] += 1 existing["updated_at"] = now_iso() existing["last_reason"] = rationale or existing.get("last_reason", "") existing["aliases"] = list(dict.fromkeys( existing.get("aliases", []) + aliases ))[:12] existing["scope"] = scope if scope == "note" else existing.get("scope", "global") existing.setdefault("evidence", []).append(evidence) existing["evidence"] = existing["evidence"][-MAX_EVIDENCE:] _add_positive_context(existing, note_title, evidence_context, 1.0) action = "reinforced" learned.append( { "memory_id": existing["id"], "source": existing["source"], "target": existing["target"], "kind": existing["kind"], "confidence": round(existing["confidence"], 3), "positive_evidence_count": existing["positive_evidence_count"], "action": action, } ) memories = recompute_statuses(memories) if conflicts and not learned: return memories, { "type": "memory_conflict", "message": "Lexi recorded conflicting contextual evidence without creating a reverse mapping.", "candidates": [], "conflicts": conflicts, } if not learned and decision != "learn": return memories, { "type": "ignored_edit", "message": "Lexi decided this edit was not durable personal lexical evidence.", "candidates": [], "conflicts": conflicts, } active_learned = [ item for item in learned if next( (memory for memory in memories if memory["id"] == item["memory_id"]), {}, ).get("status") == "active" ] event_type = "memory_learned" if active_learned else "memory_candidate" if conflicts: event_type = "memory_conflict" if active_learned: message = "Lexi learned or reinforced: " + ", ".join( f'{item["source"]} → {item["target"]}' for item in active_learned ) + "." elif conflicts: message = "Lexi kept the existing memory and recorded conflicting contextual evidence." else: message = "Lexi stored the correction as candidate personal vocabulary." return memories, { "type": event_type, "message": message, "candidates": learned, "conflicts": conflicts, } def learn_from_edit( before: str, after: str, memories: list[dict[str, Any]], previous_edit: dict[str, Any] | None = None, last_generation: dict[str, Any] | None = None, note_title: str = "", ) -> tuple[list[dict[str, Any]], dict[str, Any]]: memories = [normalize_memory(memory) for memory in (memories or [])] if before == after: return memories, {"type": "no_change", "message": "No manual edit detected.", "candidates": []} if previous_edit and previous_edit.get("before") == after and previous_edit.get("after") == before: return memories, { "type": "undo", "message": "Lexi ignored an exact undo of the previous manual edit.", "candidates": [], } override_memories, override_event = detect_generation_overrides( before, after, memories, last_generation, note_title, ) if override_event["type"] == "memory_override": return override_memories, override_event negative_memories, negative_event = detect_memory_reversals( before, after, memories, note_title, ) if negative_event["type"] == "memory_negative_evidence": return negative_memories, negative_event candidates = extract_edit_candidates(before, after) if not candidates: return memories, { "type": "ignored_edit", "message": "The edit did not contain a localized lexical correction worth learning.", "candidates": [], } scene = build_scene(note_title, before, after) proposals = [] for candidate in candidates: source = candidate["source"] target = candidate["target"] confidence = candidate["confidence"] if candidate.get("kind") in {"generic_style", "temporal", "grammar", "style"}: continue proposals.append( { "source": source, "target": target, "kind": candidate.get("kind", "other"), "confidence": confidence, "rationale": "A localized edit provided evidence of a preferred lexical form.", "aliases": [source], "evidence_type": "direct_correction", "explicit": False, "scope": "global", "context": candidate.get("context", scene["text"]), } ) if not proposals: return memories, { "type": "ignored_edit", "message": "Lexi treated this change as ordinary writing rather than personal vocabulary.", "candidates": candidates, } return incorporate_llm_proposals( memories, proposals, before, after, note_title=note_title, ) def active_memories(memories: list[dict[str, Any]]) -> list[dict[str, Any]]: return [ memory for memory in recompute_statuses( [normalize_memory(memory) for memory in (memories or [])] ) if ( memory["status"] == "active" and ( normalize_text(memory["source"]) != normalize_text(memory["target"]) or memory.get("kind") == "casing" ) ) ] def memory_pattern(source: str) -> str: parts = source.split() if not parts: return "" escaped = [re.escape(part) for part in parts] body = escaped[0] if len(escaped) == 1 else r"\s+".join(escaped) return rf"(? bool: if negative >= CONTEXT_NEGATIVE_BLOCK_THRESHOLD and negative > positive: return False current_tags = set(scene.get("tags", [])) stored_tags = { tag for item in memory.get("positive_contexts", []) for tag in item.get("tags", []) } kind = memory.get("kind", "other") if current_tags.intersection(stored_tags): return True if not current_tags or not stored_tags: return True if current_tags.intersection(ORDINARY_CONFLICT_TAGS) and not current_tags.intersection(stored_tags): return False if kind == "person_name" and "physical" in current_tags and not current_tags.intersection({"work", "software", "education"}): return False if kind in SPECIFIC_KINDS and current_tags.intersection(ORDINARY_CONFLICT_TAGS) and not current_tags.intersection(stored_tags): return False if not stored_tags: return positive >= CONTEXT_APPLY_THRESHOLD or positive >= 0.60 return positive >= CONTEXT_APPLY_THRESHOLD def _span_protection_reason(text: str, start: int, end: int) -> str: value = text or "" prefix = value[:start] suffix = value[end:] quote_pairs = [ (prefix.rfind('"'), suffix.find('"')), (prefix.rfind("'"), suffix.find("'")), (prefix.rfind("“"), suffix.find("”")), ] for opening, closing in quote_pairs: if opening >= 0 and closing >= 0: return "inside quoted text" local = context_for_change(value, start, end, radius=90) lowered = normalize_text(local) historical_patterns = [ r"\bold transcript\b", r"\bprevious transcript\b", r"\bprevious draft\b", r"\bhistorical quote\b", r"\blegacy spelling\b", r"\bformerly\b", r"\bused to (?:call|be|use)\b", r"\bbefore (?:it|we) (?:renamed|changed)\b", r"\b(?:old|previous|earlier) (?:name|version|term|spelling|draft|transcript|quote)\b", r"\bthe old .*?(?:said|says|wrote|read|used|called|spelled)\b", r"\b(?:old|previous|earlier) (?:design|architecture|technical|product) (?:doc|document|draft)\b.*?\b(?:said|says|wrote|read|used|called|spelled)\b", ] if any(re.search(pattern, lowered) for pattern in historical_patterns): return "historical or legacy reference" if re.search(r"\b(?:do not|don't|never|avoid)\b", lowered): return "negated or contrastive reference" if re.search(r"\bnot\s+(?:the|this|that|a|an)?\s*\w+", lowered): return "negated or contrastive reference" if re.search(r"\brather than\b", lowered): return "negated or contrastive reference" meta_patterns = [ r"\bthe word\b", r"\bspelling\b", r"\bquoted\b", r"\bquote\b", r"\bwhat .*? is called\b", ] if any(re.search(pattern, lowered) for pattern in meta_patterns): return "meta-language reference" return "" def apply_memories( text: str, memories: list[dict[str, Any]], selected_ids: set[str] | None = None, note_title: str = "", scene_text: str = "", ) -> tuple[str, list[dict[str, Any]]]: if not text: return text, [] scene = build_scene(note_title, scene_text or "", text) normalized_memories = recompute_statuses([normalize_memory(memory) for memory in (memories or [])]) candidates = [ memory for memory in normalized_memories if memory.get("status") in {"active", "candidate"} ] if selected_ids is not None: candidates = [memory for memory in candidates if memory["id"] in selected_ids] changes: list[dict[str, Any]] = [] replacement_pairs = [] for memory in candidates: if memory.get("scope") == "note": allowed_notes = { item.get("note_title") for item in memory.get("positive_contexts", []) if item.get("note_title") } if allowed_notes and note_title not in allowed_notes: continue score = memory_context_score(memory, scene) if score["exception"] >= 0.76: continue if score["negative"] > score["positive"] and score["negative"] >= CONTEXT_NEGATIVE_BLOCK_THRESHOLD: continue current_tags = set(scene.get("tags", [])) stored_tags = { tag for item in memory.get("positive_contexts", []) for tag in item.get("tags", []) } if current_tags.intersection(ORDINARY_CONFLICT_TAGS) and not current_tags.intersection(stored_tags): continue if memory.get("kind") in SPECIFIC_KINDS and current_tags.intersection(ORDINARY_CONFLICT_TAGS) and not current_tags.intersection(stored_tags): continue effective = max(score["positive"], 0.60 if not memory.get("positive_contexts") else 0.0) if effective < CONTEXT_APPLY_THRESHOLD and not _context_allows_memory(memory, scene, score["positive"], score["negative"]): continue for term in list(dict.fromkeys([memory["source"], *memory.get("aliases", [])])): if ( term and ( normalize_text(term) != normalize_text(memory["target"]) or memory.get("kind") == "casing" ) ): replacement_pairs.append((memory, term, score)) replacement_pairs.sort(key=lambda item: len(item[1]), reverse=True) result = text for memory, source, score in replacement_pairs: pattern = memory_pattern(source) if not pattern: continue regex = re.compile(pattern, re.IGNORECASE) def replace(match: re.Match[str]) -> str: original = match.group(0) if ( normalize_text(original) == normalize_text(memory["target"]) and memory.get("kind") != "casing" ): return original if memory.get("kind") == "casing" and original == memory["target"]: return original protection = _span_protection_reason(result, match.start(), match.end()) if protection: return original local_context = context_for_change(result, match.start(), match.end()) local_scene = build_scene(note_title, scene_text or "", local_context) local_score = memory_context_score(memory, local_scene) if local_score["exception"] >= 0.76: return original if local_score["negative"] > local_score["positive"] and local_score["negative"] >= CONTEXT_NEGATIVE_BLOCK_THRESHOLD: return original effective = max(local_score["positive"], 0.60 if not memory.get("positive_contexts") else 0.0) if effective < CONTEXT_APPLY_THRESHOLD and not _context_allows_memory(memory, local_scene, local_score["positive"], local_score["negative"]): return original memory["last_used_at"] = now_iso() changes.append( { "original": original, "replacement": memory["target"], "memory_id": memory["id"], "reason": memory.get("last_reason") or "Contextually relevant personal lexical memory.", "confidence": round(memory["confidence"], 3), "context_score": round(effective, 3), "positive_context_score": round(local_score["positive"], 3), "negative_context_score": round(local_score["negative"], 3), "note_title": note_title, } ) return memory["target"] result = regex.sub(replace, result) deduped = [] seen = set() for change in changes: key = (change["memory_id"], change["original"], change["replacement"]) if key not in seen: seen.add(key) deduped.append(change) return result, deduped def memory_context( memories: list[dict[str, Any]], selected_ids: set[str] | None = None, limit: int = 8, ) -> list[dict[str, Any]]: normalized = recompute_statuses([normalize_memory(memory) for memory in (memories or [])]) active = [ memory for memory in normalized if memory.get("status") in {"active", "candidate"} and ( selected_ids is None or memory["id"] in selected_ids ) ] active.sort( key=lambda memory: ( memory["confidence"], memory["positive_evidence_count"], -memory["negative_evidence_count"], ), reverse=True, ) return [ { "id": memory["id"], "source": memory["source"], "target": memory["target"], "kind": memory["kind"], "confidence": round(memory["confidence"], 3), "positive_evidence_count": memory["positive_evidence_count"], "negative_evidence_count": memory["negative_evidence_count"], "aliases": memory.get("aliases", []), "scope": memory.get("scope", "global"), "positive_contexts": memory.get("positive_contexts", [])[-4:], "negative_contexts": memory.get("negative_contexts", [])[-4:], "exceptions": memory.get("exceptions", [])[-4:], } for memory in active[:limit] ] def memory_summary(memories: list[dict[str, Any]]) -> dict[str, Any]: normalized = recompute_statuses([normalize_memory(memory) for memory in (memories or [])]) normalized.sort( key=lambda memory: ( 0 if memory["status"] == "active" else 1 if memory["status"] == "candidate" else 2, -memory["confidence"], ) ) return { "memories": normalized, "active_count": sum(memory["status"] == "active" for memory in normalized), "candidate_count": sum(memory["status"] == "candidate" for memory in normalized), "ambiguous_count": sum(memory["status"] == "ambiguous" for memory in normalized), } def trim_history(history: list[dict[str, Any]]) -> list[dict[str, Any]]: return list(history or [])[-MAX_HISTORY:]