diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 966ff07..ea87a23 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -152,6 +152,9 @@ jobs: bridge_intelligence.py \ tests/contract/test_phase5_invariants.py \ supervisor/dispatch.py supervisor/collector.py supervisor/router.py \ + engines/synthesis.py intelligence/memory/hybrid_retriever.py \ + intelligence/memory/maintenance.py scripts/memory_maintenance.py \ + tests/test_correction_acceptance.py \ store_registry.py tests/test_dispatch_integrity.py \ tests/test_orchestration_pipeline.py tests/test_model_router.py \ tests/test_doctor_json_reset.py tests/test_mcp_research_tools.py tests/network_guard.py @@ -247,6 +250,31 @@ jobs: /tmp/wheelvenv/bin/python -c "import cli, mcp_server, store_registry, recall_confidence; print('wheel imports OK')" /tmp/wheelvenv/bin/cortex --help + - name: Installed correction acceptance + run: | + cd /tmp + /tmp/wheelvenv/bin/python - <<'PYTHON' + import json, os, subprocess, sys, tempfile + with tempfile.TemporaryDirectory() as state: + os.environ["CORTEX_STATE_DIR"] = state + os.environ["CORTEX_HOME"] = state + "/wrong" + from mcp_handlers import record_learning_decision + from engines.synthesis import ContextGraph + from intelligence.memory.hybrid_retriever import HybridRetriever + old = record_learning_decision("obsolete canary advice") + graph = ContextGraph() + graph.import_decisions() + recall = HybridRetriever([], include_conversation_digests=False) + new = record_learning_decision("corrected canary advice", supersedes=old["decision_id"]) + expected = ["decision:" + new["decision_id"]] + assert [p.id for p, _ in recall.search("canary", alpha=0)] == expected + assert [n.id for n in graph.query("canary")] == expected + output = subprocess.check_output([sys.executable, "-m", "intelligence.memory.maintenance", "--graph-only"], text=True) + assert json.loads(output)["ok"] is True + assert [n.id for n in ContextGraph().query("canary")] == expected + print("Installed correction acceptance passed") + PYTHON + integration: # Separate job so integration tests don't bitrot. They're skipped by # default in `pytest` (see addopts in pyproject.toml); this job runs diff --git a/api/routes/decisions.py b/api/routes/decisions.py index aeecc4b..f5199fe 100644 --- a/api/routes/decisions.py +++ b/api/routes/decisions.py @@ -17,17 +17,15 @@ from __future__ import annotations -import json -import time +import uuid from datetime import datetime -from pathlib import Path from typing import Any, Dict, Optional from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field -DECISIONS_FILE = Path.home() / ".cortex" / "decisions.jsonl" +DECISIONS_FILE = None # Optional override; resolve configured state at call time. class DecisionRecordRequest(BaseModel): @@ -57,8 +55,10 @@ async def record_decision(req: DecisionRecordRequest) -> Dict[str, Any]: persisted record. """ try: - DECISIONS_FILE.parent.mkdir(parents=True, exist_ok=True) - decision_id = f"dec_{int(time.time())}_{req.prediction_id[:8]}" + from intelligence.durable_jsonl import append_decision + from state_paths import get_cortex_dir + + decision_id = f"dec_{uuid.uuid4().hex[:12]}" entry = { "decision_id": decision_id, "prediction_id": req.prediction_id, @@ -68,8 +68,7 @@ async def record_decision(req: DecisionRecordRequest) -> Dict[str, Any]: "override_reason": req.override_reason, "timestamp": datetime.now().isoformat(), } - with open(DECISIONS_FILE, "a", encoding="utf-8") as f: - f.write(json.dumps(entry) + "\n") + append_decision(DECISIONS_FILE or get_cortex_dir() / "decisions.jsonl", entry) return { "recorded": True, "decision_id": decision_id, @@ -93,7 +92,10 @@ class LearningDecisionRequest(BaseModel): alternatives: str = Field(default="", description="Other options considered") rationale: str = Field(default="", description="Why this option was chosen over alternatives") project: str = Field(default="", description="Project this decision belongs to (optional)") - supersedes: str = Field(default="", description="decision_id of a prior decision this one replaces (tombstoned out of recall)") + supersedes: str = Field( + default="", + description="decision_id of a prior decision this one replaces (tombstoned out of recall)", + ) @router.post("/decisions/learning") diff --git a/cli/commands/demo.py b/cli/commands/demo.py index d2021a0..618b7d8 100644 --- a/cli/commands/demo.py +++ b/cli/commands/demo.py @@ -7,8 +7,8 @@ No API key required. No network. Read-only against the system Cortex install (uses a tempdir for its data). -This is the falsifiable demonstration of the "compounding intelligence" claim: -if it prints links, the FK contract is live in this build. +This demonstrates synthetic advisory associations, not causal learning or +verified user outcomes. """ from __future__ import annotations @@ -87,7 +87,7 @@ def cmd_demo(args) -> None: print("⚠ No links produced — FK contract is BROKEN in this build.") sys.exit(1) - print("FK trail:") + print("Advisory FK trail (unverified; not learning eligible):") print("─" * 56) for entry in linked: score = entry["outcome_score"] @@ -99,7 +99,7 @@ def cmd_demo(args) -> None: print(f" {flag} [score {score:>4.2f}] {prompt[:48]:<48}{commit_str}") print("─" * 56) print() - print(f"Score components (per intelligence/outcome_linker.py):") + print("Score components (per intelligence/outcome_linker.py):") print(" 0.4 * test_pass_ratio + 0.4 * commit_landed + 0.2 * activity") print() print("This output was generated with NO API key and NO network call.") diff --git a/docs/design/durable-evidence-memory.md b/docs/design/durable-evidence-memory.md new file mode 100644 index 0000000..fa485c7 --- /dev/null +++ b/docs/design/durable-evidence-memory.md @@ -0,0 +1,123 @@ +# Durable evidence and memory corrections + +This tranche builds on truthful execution (PR #21). It repairs demonstrated +linkage and memory failures without certifying task outcomes or enabling adaptive +learning. No state migration or runtime cutover is performed by the code change. + +## Evidence contract + +- Only git_commit and test_result events are associated. Session must be known; + session and project must match exactly (including a missing project). +- Producer prompt_id/event_id identities are scoped by session and project. + Legacy prompt identities use a content hash, not queue position. Indistinguishable + legacy duplicates cannot be separated; producer IDs are the migration target. +- An explicit prompt_id on evidence may link after 90 seconds and after a later + prompt. Unreferenced events remain temporal hints within 90 seconds of a unique + latest prompt. Explicit reference does not prove causality or verification. +- The persisted outcome_score is retained for demo/reader compatibility as an + activity heuristic. All records remain verification_status=unverified and + learning_eligible=false. This does not switch off every other Cortex heuristic. +- Writers lock a separate file, read the current snapshot, merge evidence by + stable identity, and atomically replace the JSONL file. Concurrent/stale workers + cannot erase newer evidence. Conflicting IDs fail the transaction. Revisions + retain evidence digests, counts and heuristic scores; evidence itself is retained. +- Old position-based rows remain audit records and are not silently matched to + new identities. Historical duplicate IDs require explicit repair; malformed + lines are preserved. Old and new identities may coexist and inflate raw counts + until a reviewed migration; historical metrics are not proof of learning. +- New/revised evidence must have a new event ID. Reusing an ID with changed content + is rejected. Retraction and evaluator revision schemas remain future work. + +## Memory contract + +The primary journal update commits the replacement decision and its compatibility +tombstone in one snapshot. Readers also honor the replacement's own supersedes +field, repairing recall for historical partial writes and spooled corrections. +Original journal lines are retained verbatim; the physical file is replaced. + +MCP decision writes, the HTTP decision route, spool replay and importance backfill +use the same lock. A temporary snapshot is fsynced before replacement and the +parent directory is fsynced before acknowledgement. A failed primary write falls +back to an atomically written spool; if both fail, the call raises. A spooled +record is pending replay, not immediately recallable. Conflicting replay content +is retained for inspection rather than silently discarded. + +Decision indexing and hybrid retrieval honor CORTEX_STATE_DIR before CORTEX_HOME. +The hybrid retriever replaces cached decision patterns with the live journal at +construction, so a stale pattern index cannot resurrect corrected advice. Cache +identity includes the state path, inode, nanosecond timestamp and size. Long-lived +retrievers refresh journal decisions before search; a changed decision set +invalidates that instance's vectors and uses current BM25 until reconstruction. +Graph query/traversal APIs reconcile decision status against the journal on reads. +Retired graph nodes and edges remain audit records, excluded from active context. +An absent/unreadable journal hides imported decision nodes. This is not +system-wide memory erasure or a guarantee for callers reading raw graph storage. + +## Deployment and repository alignment + +These are POSIX local-file guarantees for cooperating writers on Linux/macOS. +They are not distributed transactions, network-filesystem guarantees or machine +memory synchronization. Updates rewrite the file and cost O(file size). Before +cutover, stop old writers, back up the state, validate the journal, install one +reviewed version, and verify its actual imported modules and state directory. +For growing multi-writer workloads, prefer a measured SQLite migration with one +local owner, then a service protocol if multiple machines need concurrent access. + +The standalone main remains 90ca32d8 at review. PR #21 is open; this tranche is +stacked on its tested head 8d75a115. Repository reconciliation remains a separate +reviewed process: preserve unique integrations, port reusable behavior through +PRs, and avoid publishing an entire divergent tree over the standalone source. +No observed repository change establishes what either live machine imports. + +## Verification and remaining acceptance gates + +Failure tests cover cross-session/project attribution, stable IDs, late evidence, +stale replay, concurrent writers, conflict rollback, interrupted replacement, +spool failure and correction replay. A fresh-process test records a correction +through the public handler, reindexes in another process, and searches through +HybridRetriever using real local BM25 with no state-path monkeypatches. + +Still required: producer IDs wired end to end; evaluator evidence tied to a task, +attempt and acceptance criterion; live +machine path checks; one bounded pilot task that resumes after interruption and +produces independently verified user-visible output. Green synthetic tests alone +do not establish trusted compounding or autonomous project progress. + +## Reassessment: finish a user outcome before expanding architecture + +The first acceptance outcome is: record advice, correct it, then retrieve only +the correction through both recall and graph context in a new process. It must +also work in a long-lived reader, and after graph saving is interrupted. Old +nodes remain auditable; no graph path may use them as active context. Maintenance +must report failure when persistence fails. The new correction acceptance suite +exercises these conditions, including real local embedding-cache invalidation. + +The decision journal is authoritative. The graph is a derived view; its two +legacy snapshot files are not a multi-file transaction. Reconciliation on reads +protects the correction outcome even if graph persistence is interrupted. This +does not repair every graph concurrency or historical corruption problem. + +Do not equate this completed acceptance path with useful autonomous learning. +Next, install one reviewed artifact on one development machine and exercise one +real project task with explicit user acceptance. Resolve the runtime dependency +closure needed for that pilot before cutover; keep the broader repository +inventory as a staged reconciliation queue. A blanket review of every historical +file is not the release gate for one isolated pilot. + +Defer general-purpose self-improvement, cross-machine memory synchronization, +and broad unattended orchestration until the pilot produces accepted work, +recovers after interruption, and uses corrected context. Model-generated +success scores and a larger test count cannot substitute for that evidence. + +The installed pilot command is: + +```sh +python -m intelligence.memory.maintenance --graph-only +``` + +Run it with CORTEX_STATE_DIR pointing at the pilot state. A persistence failure +returns a nonzero exit status. The existing scripts/memory_maintenance.py remains +a source-checkout compatibility entrypoint. The package command was added because +the installed-wheel acceptance test exposed that scripts/ is excluded from the +wheel. The default full maintenance job also emits operational failure signals; +that separate learning path is not part of this correction acceptance proof. diff --git a/engines/synthesis.py b/engines/synthesis.py index 3b00f81..122905c 100644 --- a/engines/synthesis.py +++ b/engines/synthesis.py @@ -150,7 +150,12 @@ class ContextGraph: """ def __init__(self, storage_path: Optional[Path] = None): - self.storage_path = storage_path or Path.home() / ".cortex" / "graph" + from state_paths import get_cortex_dir + + self.storage_path = storage_path or get_cortex_dir() / "graph" + self._decisions_path = None + self._active_decision_ids: Set[str] = set() + self._decisions_dirty = False self.nodes: Dict[str, Node] = {} self.edges: List[Edge] = [] self._adjacency: Dict[str, Set[str]] = {} # node_id -> connected node_ids @@ -392,8 +397,9 @@ def _bucket_key(node: Node, field: str) -> Optional[str]: by_project_decisions: Dict[str, List[str]] = {} by_pattern_key: Dict[str, List[str]] = {} - for nid in sorted(self.nodes): - node = self.nodes[nid] + active = self.active_nodes() + for nid in sorted(active): + node = active[nid] proj = _bucket_key(node, "project") pkey = _bucket_key(node, "pattern_key") if node.type == NodeType.LESSON: @@ -449,17 +455,17 @@ def _add(src: str, dst: str, etype: EdgeType, reason: str) -> None: # 3. decision -> project lessons/patterns (capped per decision) so # recorded decisions are reachable via graph traversal. for proj, decisions in by_project_decisions.items(): - neighbors = ( - by_project_lessons.get(proj, []) + by_project_patterns.get(proj, []) - )[:per_node_cap] + neighbors = (by_project_lessons.get(proj, []) + by_project_patterns.get(proj, []))[ + :per_node_cap + ] for did in decisions: for nid in neighbors: _add(did, nid, EdgeType.RELATES_TO, "decision_project") self.edges.extend(new_edges) self._rebuild_adjacency() - if save: - self._save() + if save and not self._save(): + raise OSError("Graph edge regeneration could not be persisted") logger.info( "regenerate_edges: added %d edges (total now %d)", len(new_edges), len(self.edges) ) @@ -478,27 +484,46 @@ def import_decisions(self, decisions_path: Optional[Path] = None, save: bool = T is safe to re-run from the maintenance loop. Node ids use a ``decision:`` scheme matching the retriever's prefix. - Returns the number of new decision nodes added. + Returns the number of added, refreshed or retired decision nodes. """ - path = decisions_path or (Path.home() / ".cortex" / "decisions.jsonl") + from intelligence.durable_jsonl import active_decisions, read_records + from state_paths import get_cortex_dir + + if decisions_path is not None: + self._decisions_path = decisions_path + path = self._decisions_path or get_cortex_dir() / "decisions.jsonl" + self._active_decision_ids = set() if not path.exists(): return 0 - + # Do not report successful synchronization if the authoritative source + # cannot be read. Public retrieval handles this by hiding decision nodes. + records = list(read_records(path.read_text(encoding="utf-8"))) + active = active_decisions(records) + self._active_decision_ids = { + "decision:" + d["decision_id"] for d in active if d.get("decision_id") + } + superseded = { + d["decision_id"]: d["superseded_by"] + for d in records + if d.get("decision_id") and d.get("superseded_by") + } + superseded.update( + { + d["supersedes"]: d["decision_id"] + for d in records + if d.get("decision") and d.get("supersedes") and d.get("decision_id") + } + ) added = 0 - try: - lines = path.read_text().splitlines() - except OSError as exc: - logger.warning("cannot read decisions file %s: %s", path, exc) - return 0 - - for line in lines: - line = line.strip() - if not line: - continue - try: - d = json.loads(line) - except json.JSONDecodeError: - continue + for old_id, new_id in superseded.items(): + old = self.nodes.get("decision:" + old_id) + if old is not None and old.data.get("superseded_by") != new_id: + # Preserve the original node and its edges for audit. Retrieval + # excludes retired nodes and traversals through them. + old.data["superseded_by"] = new_id + old.updated_at = datetime.now() + added += 1 + for d in active: decision_text = d.get("decision") if not decision_text: continue @@ -543,23 +568,52 @@ def import_decisions(self, decisions_path: Optional[Path] = None, save: bool = T ) added += 1 - if added and save: - self._save() + self._decisions_dirty = self._decisions_dirty or added > 0 + if save and self._decisions_dirty: + if not self._save(): + raise OSError("Decision graph synchronization could not be persisted") + self._decisions_dirty = False logger.info("import_decisions: added %d decision nodes (total %d)", added, len(self.nodes)) return added + def active_nodes(self) -> Dict[str, Node]: + """Live decision-journal view over a derived graph snapshot. + + Reconcile on reads so a failed/stale graph save cannot resurrect advice. + Raw nodes/edges remain available for audit, not recommendation retrieval. + If the journal is absent/unreadable, imported decisions fail closed. + """ + try: + self.import_decisions(save=False) + except (OSError, ValueError, TypeError): + self._active_decision_ids = set() + logger.warning("Decision journal unavailable; hiding imported decision nodes") + return { + key: node + for key, node in self.nodes.items() + if not node.data.get("superseded_by") + and ( + node.type != NodeType.DECISION + or not key.startswith("decision:") + or key in self._active_decision_ids + ) + } + def get_node(self, node_id: str) -> Optional[Node]: - """Get a node by ID.""" - return self.nodes.get(node_id) + """Get an active node; retired decision nodes remain in raw audit storage.""" + return self.active_nodes().get(node_id) def get_nodes_by_type(self, node_type: NodeType) -> List[Node]: - """Get all nodes of a specific type.""" - return [n for n in self.nodes.values() if n.type == node_type] + """Get active nodes of a specific type.""" + return [n for n in self.active_nodes().values() if n.type == node_type] def get_related(self, node_id: str, edge_type: Optional[EdgeType] = None) -> List[Node]: """Get nodes related to a given node (outgoing edges).""" + active = self.active_nodes() + if node_id not in active: + return [] related_ids = self._adjacency.get(node_id, set()) - related_nodes = [self.nodes[nid] for nid in related_ids if nid in self.nodes] + related_nodes = [active[nid] for nid in related_ids if nid in active] if edge_type: # Filter by edge type @@ -572,8 +626,11 @@ def get_related(self, node_id: str, edge_type: Optional[EdgeType] = None) -> Lis def get_pointing_to(self, node_id: str, edge_type: Optional[EdgeType] = None) -> List[Node]: """Get nodes pointing to a given node (incoming edges).""" + active = self.active_nodes() + if node_id not in active: + return [] source_ids = self._reverse_adjacency.get(node_id, set()) - source_nodes = [self.nodes[nid] for nid in source_ids if nid in self.nodes] + source_nodes = [active[nid] for nid in source_ids if nid in active] if edge_type: valid_sources = { @@ -585,7 +642,8 @@ def get_pointing_to(self, node_id: str, edge_type: Optional[EdgeType] = None) -> def find_path(self, source_id: str, target_id: str, max_depth: int = 10) -> Optional[List[str]]: """Find shortest path between two nodes (BFS).""" - if source_id not in self.nodes or target_id not in self.nodes: + active = self.active_nodes() + if source_id not in active or target_id not in active: return None visited = {source_id} @@ -598,7 +656,7 @@ def find_path(self, source_id: str, target_id: str, max_depth: int = 10) -> Opti return path for neighbor in self._adjacency.get(current, set()): - if neighbor not in visited: + if neighbor in active and neighbor not in visited: visited.add(neighbor) queue.append((neighbor, path + [neighbor])) @@ -606,7 +664,8 @@ def find_path(self, source_id: str, target_id: str, max_depth: int = 10) -> Opti def get_subgraph(self, center_id: str, depth: int = 2) -> Dict[str, Any]: """Get subgraph around a node.""" - if center_id not in self.nodes: + active = self.active_nodes() + if center_id not in active: return {"nodes": [], "edges": []} visited = {center_id} @@ -622,11 +681,11 @@ def get_subgraph(self, center_id: str, depth: int = 2) -> Dict[str, Any]: node_id, set() ) for neighbor in neighbors: - if neighbor not in visited: + if neighbor in active and neighbor not in visited: visited.add(neighbor) to_visit.append((neighbor, current_depth + 1)) - nodes = [self.nodes[nid].to_dict() for nid in visited if nid in self.nodes] + nodes = [active[nid].to_dict() for nid in visited if nid in active] edges = [ e.to_dict() for e in self.edges if e.source_id in visited and e.target_id in visited ] @@ -647,7 +706,7 @@ def query( query_lower = query_text.lower() results = [] - for node in self.nodes.values(): + for node in self.active_nodes().values(): if node_types and node.type not in node_types: continue diff --git a/intelligence/durable_jsonl.py b/intelligence/durable_jsonl.py new file mode 100644 index 0000000..b432873 --- /dev/null +++ b/intelligence/durable_jsonl.py @@ -0,0 +1,106 @@ +"""Local POSIX JSONL transactions for cooperating Cortex writers. + +A separate lock survives replacement of the data inode. Readers see the complete +old or new snapshot. This is not a distributed store; older unlocked writers must +be stopped before cutover. Each update rewrites the file (O(file size)). +""" + +import fcntl +import json +import os +import tempfile +from contextlib import contextmanager +from pathlib import Path + + +@contextmanager +def locked(path: Path): + path.parent.mkdir(parents=True, exist_ok=True) + with open(path.with_name(path.name + ".lock"), "a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(lock, fcntl.LOCK_UN) + + +def atomic_write(path: Path, content: str): + """Replace using a same-directory, fsynced temporary file; fsync the directory.""" + path.parent.mkdir(parents=True, exist_ok=True) + fd, name = tempfile.mkstemp(prefix="." + path.name + "-", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as stream: + stream.write(content) + stream.flush() + os.fsync(stream.fileno()) + os.replace(name, path) + directory = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + finally: + if os.path.exists(name): + os.unlink(name) + + +def read_records(text: str): + for line in text.splitlines(): + try: + entry = json.loads(line) + except (ValueError, TypeError): + continue + if isinstance(entry, dict): + yield entry + + +def append_decision(path: Path, entry: dict) -> bool: + """Commit a decision and compatibility tombstone together; replay by ID.""" + with locked(path): + old = path.read_text(encoding="utf-8") if path.exists() else "" + existing = [ + d + for d in read_records(old) + if d.get("decision_id") == entry["decision_id"] and not d.get("superseded_by") + ] + if existing: + if existing[0] != entry: + raise ValueError("Conflicting decision replay: " + entry["decision_id"]) + # A previous attempt may have failed after replace but before the + # directory fsync. Re-establish durability before deleting a spool. + with path.open("rb") as stream: + os.fsync(stream.fileno()) + directory = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + return False + additions = [entry] + if entry.get("supersedes"): + additions.append( + { + "decision_id": entry["supersedes"], + "superseded_by": entry["decision_id"], + "timestamp": entry.get("timestamp"), + "source": "supersede", + } + ) + content = old + ("\n" if old and not old.endswith("\n") else "") + content += "".join(json.dumps(d) + "\n" for d in additions) + atomic_write(path, content) + return True + + +def active_decisions(records): + """New self-contained corrections and legacy tombstones both suppress old IDs.""" + records = list(records) + superseded = {d.get("decision_id") for d in records if d.get("superseded_by")} + superseded.update(d["supersedes"] for d in records if d.get("decision") and d.get("supersedes")) + return [ + d + for d in records + if d.get("decision") + and not d.get("superseded_by") + and d.get("decision_id") not in superseded + ] diff --git a/intelligence/memory/hybrid_retriever.py b/intelligence/memory/hybrid_retriever.py index ce2a15b..823a49c 100644 --- a/intelligence/memory/hybrid_retriever.py +++ b/intelligence/memory/hybrid_retriever.py @@ -20,21 +20,22 @@ from typing import Dict, List, Optional, Tuple import numpy as np +from state_paths import get_cortex_dir from intelligence.embeddings_client import EmbeddingsClient from intelligence.memory.pattern_indexer import Pattern, PatternSearcher logger = logging.getLogger(__name__) # Outcome data path -_OUTCOMES_PATH = Path.home() / ".cortex" / "outcomes.jsonl" +_OUTCOMES_PATH = None # Conversation digests path -_DIGESTS_PATH = Path.home() / ".cortex" / "conversation_digests.jsonl" +_DIGESTS_PATH = None # Recorded-decisions path (written by mcp_handlers.record_learning_decision / # POST /decisions/learning). These are the "your past decisions come back to # you" memories; without loading them here they were write-only. -_DECISIONS_PATH = Path.home() / ".cortex" / "decisions.jsonl" +_DECISIONS_PATH = None # Optional explicit override; default resolves at call time. # Module-level cache for decision patterns, keyed by decisions.jsonl mtime, so @@ -61,11 +62,15 @@ def _load_decision_patterns() -> List[Pattern]: """ global _decision_cache, _decision_cache_mtime, _decision_weights - if not _DECISIONS_PATH.exists(): + from state_paths import get_cortex_dir + + path = _DECISIONS_PATH or (get_cortex_dir() / "decisions.jsonl") + if not path.exists(): return [] try: - mtime = _DECISIONS_PATH.stat().st_mtime + stat = path.stat() + mtime = (str(path.resolve()), stat.st_ino, stat.st_mtime_ns, stat.st_size) except OSError: mtime = None if _decision_cache is not None and _decision_cache_mtime == mtime: @@ -74,7 +79,7 @@ def _load_decision_patterns() -> List[Pattern]: patterns: List[Pattern] = [] weights: Dict[str, float] = {} try: - raw_lines = _DECISIONS_PATH.read_text().splitlines() + raw_lines = path.read_text().splitlines() # First pass: collect ids marked superseded by a tombstone (P1 curation). # A superseded decision is dropped from recall so stale/reversed calls @@ -90,6 +95,8 @@ def _load_decision_patterns() -> List[Pattern]: continue if d.get("superseded_by"): superseded.add(d.get("decision_id")) + if d.get("decision") and d.get("supersedes"): + superseded.add(d["supersedes"]) for line in raw_lines: line = line.strip() @@ -120,9 +127,7 @@ def _load_decision_patterns() -> List[Pattern]: desc_parts.append(f"alternatives: {d['alternatives']}") try: - commit_date = datetime.fromisoformat( - d.get("timestamp", "").replace("Z", "+00:00") - ) + commit_date = datetime.fromisoformat(d.get("timestamp", "").replace("Z", "+00:00")) except (ValueError, TypeError): commit_date = datetime.now() @@ -153,7 +158,9 @@ def _load_decision_patterns() -> List[Pattern]: age_days = max(0.0, (datetime.now() - commit_date.replace(tzinfo=None)).days) except Exception: age_days = 0.0 - decay = math.exp(-age_days / _DECAY_HALF_LIFE_DAYS) if _DECAY_HALF_LIFE_DAYS > 0 else 1.0 + decay = ( + math.exp(-age_days / _DECAY_HALF_LIFE_DAYS) if _DECAY_HALF_LIFE_DAYS > 0 else 1.0 + ) # Map importance 1..10 to ~0.5..1.1 so low-signal is penalised but # never zeroed (still findable), high-signal mildly boosted. imp_factor = 0.5 + (importance / 10.0) * 0.6 @@ -175,12 +182,13 @@ def _load_digest_patterns() -> List[Pattern]: search pipeline as git-derived patterns. Digest patterns use a "conversation:" prefix in their ID to distinguish them from commit-based patterns. """ - if not _DIGESTS_PATH.exists(): + path = _DIGESTS_PATH or get_cortex_dir() / "conversation_digests.jsonl" + if not path.exists(): return [] patterns = [] try: - for line in _DIGESTS_PATH.read_text().splitlines(): + for line in path.read_text().splitlines(): line = line.strip() if not line: continue @@ -274,6 +282,9 @@ def __init__( # Merge recorded decisions so past decisions can be recalled. Always # loaded (not gated on include_conversation_digests): a decision store # is the core "memory comes back" signal, independent of chat history. + # Cached indexes cannot override the current decision journal, including + # after a correction or a change of configured state directory. + patterns = [p for p in patterns if not p.id.startswith("decision:")] decision_patterns = _load_decision_patterns() if decision_patterns: logger.info(f"Loaded {len(decision_patterns)} recorded-decision patterns") @@ -283,7 +294,7 @@ def __init__( self.bm25_searcher = PatternSearcher(patterns) if cache_dir is None: - cache_dir = Path.home() / ".cortex" / "patterns" + cache_dir = get_cortex_dir() / "patterns" self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) @@ -314,12 +325,13 @@ def _load_outcome_boosts(self) -> None: These boosts are applied during RRF merge to close the feedback loop: patterns from projects where outcomes were successful rank higher. """ - if not _OUTCOMES_PATH.exists(): + path = _OUTCOMES_PATH or get_cortex_dir() / "outcomes.jsonl" + if not path.exists(): return project_outcomes: Dict[str, list] = defaultdict(list) try: - with open(_OUTCOMES_PATH) as f: + with open(path) as f: for line in f: line = line.strip() if not line: @@ -403,8 +415,7 @@ def _load_or_generate_embeddings(self): return backend_ok = ( - not isinstance(current_backend, str) - or cached_backend == current_backend + not isinstance(current_backend, str) or cached_backend == current_backend ) if meta.get("pattern_count") == len(self.patterns) and backend_ok: # Check if pattern IDs match @@ -495,9 +506,7 @@ def _scoped_index( scoped_patterns = [self.patterns[i] for i in indices] scoped_embeddings = ( - self.pattern_embeddings[indices] - if self.pattern_embeddings is not None - else None + self.pattern_embeddings[indices] if self.pattern_embeddings is not None else None ) return scoped_patterns, scoped_embeddings, PatternSearcher(scoped_patterns) @@ -524,6 +533,17 @@ def search( Returns: List of (Pattern, score) tuples sorted by score """ + # A long-lived bridge must not keep recommending a corrected decision. + # Invalidate vectors rather than reusing rows for a changed pattern set + # or making an unrequested provider call during this refresh. + current = _load_decision_patterns() + previous = [p for p in self.patterns if p.id.startswith("decision:")] + if current != previous: + self.patterns = [p for p in self.patterns if not p.id.startswith("decision:")] + current + self.bm25_searcher = PatternSearcher(self.patterns) + self.pattern_embeddings = None + self.embeddings_available = False + # Validate alpha alpha = max(0.0, min(1.0, alpha)) @@ -582,9 +602,7 @@ def _semantic_search( """ patterns = patterns if patterns is not None else self.patterns pattern_embeddings = ( - pattern_embeddings - if pattern_embeddings is not None - else self.pattern_embeddings + pattern_embeddings if pattern_embeddings is not None else self.pattern_embeddings ) if not self.embeddings_available or pattern_embeddings is None: diff --git a/intelligence/memory/maintenance.py b/intelligence/memory/maintenance.py new file mode 100644 index 0000000..c2fbe1d --- /dev/null +++ b/intelligence/memory/maintenance.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +"""Memory-loop maintenance — keeps the compounding layer from silently rotting. + +Runs three idempotent jobs that the 2026-07 memory-restoration work identified +as needing a periodic driver (nothing was scheduling them, so they decayed): + + 1. failure emission — fold operational failures (restarts, alerts, scheduler + errors, pytest) into the outcome stream as weighted `failed` outcomes, so + the learning loop has a real failure signal to calibrate against. + 2. edge regeneration — rebuild the knowledge-graph edges from current nodes + when the edge set has collapsed (the schema-mismatch regression wiped + 1247 edges to 0 between April and June 2026). + 3. decision index — recorded decisions are auto-loaded by HybridRetriever + on construction, so a rebuilt retriever is enough; we just report the + count here for the maintenance log. + +Idempotent and safe to run on a short interval (LaunchAgent / cron). Each job +guards its own exceptions so one failing job never blocks the others. Writes a +JSON run-report to ~/.cortex/maintenance/ for auditability. + +Run installed: python -m intelligence.memory.maintenance +Graph acceptance: python -m intelligence.memory.maintenance --graph-only +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone + + +def _emit_failures(lookback_minutes: int) -> dict: + """Job 1: emit collapsed operational failures into the outcome stream.""" + try: + from intelligence import failure_emitter + + result = failure_emitter.run_once(lookback_minutes=lookback_minutes) + return {"job": "failure_emission", "ok": True, **result} + except Exception as exc: # never let one job kill the run + return {"job": "failure_emission", "ok": False, "error": str(exc)} + + +def _regenerate_edges() -> dict: + """Job 2: sync decisions into the graph, then rebuild edges if collapsed. + + First imports any new recorded decisions as DECISION nodes (idempotent), + then regenerates edges when the edge set is empty (the failure mode) OR + when new decision nodes were just added (so they get linked). A healthy + graph with no new decisions is left untouched to avoid churn. + """ + try: + from engines.synthesis import ContextGraph + + graph = ContextGraph() + before = len(graph.edges) + # Keep the graph in sync with newly recorded decisions. + new_decisions = graph.import_decisions(save=True) + # Regenerate when edges collapsed, or to link freshly imported decisions. + if (before == 0 or new_decisions > 0) and graph.nodes: + total = graph.regenerate_edges(save=True) + return { + "job": "edge_regeneration", + "ok": True, + "regenerated": True, + "new_decision_nodes": new_decisions, + "edges_before": before, + "edges_after": total, + "nodes": len(graph.nodes), + } + return { + "job": "edge_regeneration", + "ok": True, + "regenerated": False, + "new_decision_nodes": new_decisions, + "edges": before, + "nodes": len(graph.nodes), + } + except Exception as exc: + return {"job": "edge_regeneration", "ok": False, "error": str(exc)} + + +def _decision_index() -> dict: + """Job 3: report how many recorded decisions are recall-indexable.""" + try: + from intelligence.memory.hybrid_retriever import _load_decision_patterns + + return { + "job": "decision_index", + "ok": True, + "decisions_indexable": len(_load_decision_patterns()), + } + except Exception as exc: + return {"job": "decision_index", "ok": False, "error": str(exc)} + + +def run_once(lookback_minutes: int = 90) -> dict: + """Run all maintenance jobs once and persist a report.""" + report = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "jobs": [ + _emit_failures(lookback_minutes), + _regenerate_edges(), + _decision_index(), + ], + } + try: + from state_paths import get_cortex_dir + + report_dir = get_cortex_dir() / "maintenance" + report_dir.mkdir(parents=True, exist_ok=True) + with (report_dir / "maintenance_history.jsonl").open("a") as f: + f.write(json.dumps(report) + "\n") + except OSError: + pass # reporting is best-effort + return report + + +def backfill_importance() -> dict: + """One-time (idempotent) P1 backfill: score decisions that predate the + importance heuristic. Entries already carrying `importance` are left as-is, + tombstones are skipped. Writes a `.bak` and rewrites atomically (temp + + rename). Safe to run repeatedly — a second run is a no-op. + """ + from state_paths import get_cortex_dir + from intelligence.memory.importance import _importance_score, IMPORTANCE_FLOOR + + path = get_cortex_dir() / "decisions.jsonl" + if not path.exists(): + return {"job": "backfill_importance", "status": "no_file", "scored": 0} + + from intelligence.durable_jsonl import locked, atomic_write + + with locked(path): + lines = path.read_text().splitlines() + scored = skipped = flagged = 0 + out = [] + for line in lines: + s = line.strip() + if not s: + continue + try: + d = json.loads(s) + except json.JSONDecodeError: + out.append(line) # preserve unparseable lines verbatim + skipped += 1 + continue + # Skip tombstones and already-scored entries (idempotent). + if d.get("superseded_by") or "importance" in d or not d.get("decision"): + out.append(json.dumps(d)) + skipped += 1 + continue + score = _importance_score( + d.get("decision", ""), + d.get("context", ""), + d.get("alternatives", ""), + d.get("rationale", ""), + ) + d["importance"] = score + if score < IMPORTANCE_FLOOR: + d["low_signal"] = True + flagged += 1 + out.append(json.dumps(d)) + scored += 1 + + if scored: + path.with_suffix(".jsonl.bak").write_text("\n".join(lines) + "\n") + atomic_write(path, "\n".join(out) + "\n") + + return { + "job": "backfill_importance", + "status": "ok", + "scored": scored, + "skipped": skipped, + "flagged_low_signal": flagged, + } + + +def main() -> int: + import argparse + + p = argparse.ArgumentParser(description="Cortex memory-loop maintenance") + p.add_argument("--lookback-minutes", type=int, default=90) + p.add_argument( + "--backfill-importance", + action="store_true", + help="One-time P1 backfill: score decisions lacking an importance key (idempotent)", + ) + p.add_argument( + "--graph-only", + action="store_true", + help="Synchronize decision graph only; no operational failure emission", + ) + args = p.parse_args() + if args.graph_only: + report = _regenerate_edges() + print(json.dumps(report, indent=2)) + return 0 if report["ok"] else 1 + if args.backfill_importance: + print(json.dumps(backfill_importance(), indent=2)) + return 0 + print(json.dumps(run_once(lookback_minutes=args.lookback_minutes), indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/intelligence/memory/pattern_indexer.py b/intelligence/memory/pattern_indexer.py index 2a51048..eede20d 100644 --- a/intelligence/memory/pattern_indexer.py +++ b/intelligence/memory/pattern_indexer.py @@ -12,7 +12,6 @@ """ import json -import os import re import subprocess from dataclasses import dataclass, field @@ -403,7 +402,9 @@ def save_patterns(self, patterns: Dict[str, List[Pattern]]): metadata_file = self.cache_dir / "metadata.json" metadata_file.write_text(json.dumps(metadata, indent=2)) - def load_patterns(self, include_seeds: bool = True, include_decisions: bool = True) -> List[Pattern]: + def load_patterns( + self, include_seeds: bool = True, include_decisions: bool = True + ) -> List[Pattern]: """Load patterns from cache + recorded decisions, with seed fallback. Git-derived patterns come from the cache (patterns.json) or, on cold @@ -464,35 +465,32 @@ def _load_decisions() -> List[Pattern]: Durable source: independent of patterns.json, so decisions persist across a git re-index and new decisions are picked up on the next index build. """ - from pathlib import Path + from intelligence.durable_jsonl import active_decisions, read_records + from state_paths import get_cortex_dir - # Resolve the data home the same way the rest of cortex does - # (CORTEX_HOME, default ~/.cortex) so this is overridable/isolatable. - cortex_home = Path(os.environ.get("CORTEX_HOME", str(Path.home() / ".cortex"))) - decisions_file = cortex_home / "decisions.jsonl" + decisions_file = get_cortex_dir() / "decisions.jsonl" if not decisions_file.exists(): return [] out: List[Pattern] = [] - for line in decisions_file.read_text().splitlines(): - line = line.strip() - if not line: - continue - try: - d = json.loads(line) - except ValueError: - continue + for d in active_decisions(read_records(decisions_file.read_text())): did = d.get("decision_id") or d.get("id") if not did: continue alts = d.get("alternatives") - desc = " | ".join(x for x in [ - d.get("context"), - f"Rationale: {d['rationale']}" if d.get("rationale") else None, - f"Alternatives: {alts}" if isinstance(alts, str) and alts else None, - ] if x) + desc = " | ".join( + x + for x in [ + d.get("context"), + f"Rationale: {d['rationale']}" if d.get("rationale") else None, + f"Alternatives: {alts}" if isinstance(alts, str) and alts else None, + ] + if x + ) ts = d.get("timestamp") or d.get("created_at") try: - cd = datetime.fromisoformat(str(ts).replace("Z", "+00:00")) if ts else datetime.now() + cd = ( + datetime.fromisoformat(str(ts).replace("Z", "+00:00")) if ts else datetime.now() + ) if cd.tzinfo: cd = cd.replace(tzinfo=None) except (ValueError, TypeError): diff --git a/intelligence/outcome_linker.py b/intelligence/outcome_linker.py index 6100280..ef8cbd6 100644 --- a/intelligence/outcome_linker.py +++ b/intelligence/outcome_linker.py @@ -1,81 +1,108 @@ -"""Links prompts to downstream outcomes via temporal proximity. +"""Scoped prompt/evidence associations, persisted as local atomic snapshots. -A git_commit or test_result within 90 seconds of a prompt_received -is attributed to that prompt. Writes FK linkage to ~/.cortex/prompt_outcomes.jsonl. +Temporal associations and activity scores are advisory, never verified causal +credit. Explicit prompt IDs permit late evidence outside the legacy 90s window. """ +import hashlib import json -from datetime import datetime +from datetime import datetime, timezone from pathlib import Path from typing import Optional -QUEUE = Path.home() / ".cortex" / "interaction_queue.jsonl" -OUTCOMES = Path.home() / ".cortex" / "prompt_outcomes.jsonl" +from intelligence.durable_jsonl import atomic_write, locked, read_records +from state_paths import get_cortex_dir + +QUEUE = None # Optional overrides; default state is resolved at call time. +OUTCOMES = None WINDOW_SECONDS = 90 +EVIDENCE_TYPES = {"git_commit", "test_result"} + + +def _digest(value) -> str: + return hashlib.sha256( + json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() def _parse_ts(ts_str: str) -> datetime: - return datetime.fromisoformat(ts_str.replace("Z", "+00:00")) + ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) + return ts.replace(tzinfo=timezone.utc) if ts.tzinfo is None else ts -def link_outcomes(queue_path: Optional[Path] = None) -> list[dict]: - """Read interaction queue, link outcomes to prompts within WINDOW_SECONDS. +def _prompt_id(entry): + explicit = entry.get("prompt_id") or entry.get("event_id") + identity = explicit or {k: entry.get(k) for k in ("queued_at", "prompt")} + return "prompt:" + _digest([entry.get("session_id"), entry.get("project"), identity]) - Args: - queue_path: Optional path to read the interaction queue from. - Defaults to the module-level QUEUE. - """ - queue = queue_path if queue_path is not None else QUEUE + +def link_outcomes(queue_path: Optional[Path] = None) -> list[dict]: + queue = queue_path or QUEUE or get_cortex_dir() / "interaction_queue.jsonl" if not queue.exists(): return [] - entries = [] - for line in queue.read_text().strip().splitlines(): - if not line: - continue + for entry in read_records(queue.read_text(encoding="utf-8")): try: - entries.append(json.loads(line)) - except json.JSONDecodeError: + ts = _parse_ts(entry["queued_at"]) + except (ValueError, TypeError, KeyError, AttributeError): continue - - # Sort by timestamp - entries.sort(key=lambda e: e.get("queued_at", "")) - + entries.append((ts, entry)) + entries.sort(key=lambda pair: pair[0]) + prompts = [(ts, e) for ts, e in entries if e.get("type") == "prompt_received"] linked = [] - for i, entry in enumerate(entries): - if entry.get("type") != "prompt_received": - continue - - prompt_ts = _parse_ts(entry["queued_at"]) - prompt_id = entry.get("session_id", "?") + "_" + str(i) - + for prompt_ts, entry in prompts: + scope = (entry.get("session_id"), entry.get("project")) + if not scope[0]: + continue # Unknown session cannot receive inferred credit. + explicit = entry.get("prompt_id") or entry.get("event_id") outcomes = [] - for j in range(i + 1, len(entries)): - other = entries[j] - if other.get("type") == "prompt_received" and other.get("session_id") == entry.get( - "session_id" - ): - break # Next prompt in same session — stop looking - other_ts = _parse_ts(other["queued_at"]) + methods = set() + for other_ts, other in entries: + if other.get("type") not in EVIDENCE_TYPES: + continue + if (other.get("session_id"), other.get("project")) != scope: + continue delta = (other_ts - prompt_ts).total_seconds() - if 0 <= delta <= WINDOW_SECONDS: - outcomes.append(other) - elif delta > WINDOW_SECONDS: - break - + if delta < 0: + continue + reference = other.get("prompt_id") + if reference: + if not explicit or reference != explicit: + continue + method = "explicit_reference" + else: + if delta > WINDOW_SECONDS: + continue + # Attribute only to the unique most recent prompt in this scope. + preceding = [ + (ts, p) + for ts, p in prompts + if ts <= other_ts and (p.get("session_id"), p.get("project")) == scope + ] + latest = max(ts for ts, _ in preceding) + candidates = {_prompt_id(p) for ts, p in preceding if ts == latest} + if candidates != {_prompt_id(entry)}: + continue + method = "temporal_hint" + outcomes.append(other) + methods.add(method) if outcomes: - score = _compute_outcome_score(outcomes) + outcomes = list({_digest(e): e for e in outcomes}.values()) linked.append( { - "prompt_id": prompt_id, + "prompt_id": _prompt_id(entry), + "identity_source": "producer" if explicit else "content_fallback", "prompt_text": entry.get("prompt", "")[:120], - "session_id": entry.get("session_id"), + "session_id": scope[0], + "project": scope[1], "prompt_ts": entry["queued_at"], - "outcome_score": score, + "association_methods": sorted(methods), + "outcome_score": _compute_outcome_score(outcomes), + "verification_status": "unverified", + "learning_eligible": False, "outcomes": outcomes, } ) - return linked @@ -98,46 +125,85 @@ def _compute_outcome_score(outcomes: list[dict]) -> float: def _existing_prompt_ids(outcomes_path: Optional[Path] = None) -> set[str]: - path = outcomes_path if outcomes_path is not None else OUTCOMES + path = outcomes_path or OUTCOMES or get_cortex_dir() / "prompt_outcomes.jsonl" if not path.exists(): return set() - seen = set() - for line in path.read_text().splitlines(): - if not line: - continue - try: - seen.add(json.loads(line).get("prompt_id")) - except json.JSONDecodeError: - continue - return seen + return {e["prompt_id"] for e in read_records(path.read_text()) if e.get("prompt_id")} -def write_linked_outcomes( - linked: list[dict], outcomes_path: Optional[Path] = None -) -> None: - """Append new linked entries; skips prompt_ids already on disk (idempotent). +def write_linked_outcomes(linked: list[dict], outcomes_path: Optional[Path] = None) -> None: + """Locked read/merge/replace, with revision history and monotonic evidence union. - Args: - linked: List of linked entries from link_outcomes(). - outcomes_path: Optional override for the destination jsonl file. + Replaying an older queue snapshot cannot erase newer evidence. Legacy rows + retain their identity and audit data but cannot become learning eligible. """ - path = outcomes_path if outcomes_path is not None else OUTCOMES - existing = _existing_prompt_ids(path) - with open(path, "a") as f: - for entry in linked: - if entry.get("prompt_id") in existing: + path = outcomes_path or OUTCOMES or get_cortex_dir() / "prompt_outcomes.jsonl" + with locked(path): + old = path.read_text(encoding="utf-8") if path.exists() else "" + rows = {} + preserved = [] + for line in old.splitlines(): + parsed = list(read_records(line)) + if not parsed or not parsed[0].get("prompt_id"): + preserved.append(line) continue - f.write(json.dumps(entry) + "\n") + row = parsed[0] + if row["prompt_id"] in rows: + raise ValueError("Duplicate persisted prompt identity; repair required") + row.update(verification_status="unverified", learning_eligible=False) + rows[row["prompt_id"]] = row + for candidate in linked: + row = dict(candidate) + key = row["prompt_id"] + previous = rows.get(key) + if previous: + for field in ("session_id", "project", "prompt_ts", "prompt_text"): + if previous.get(field) != row.get(field): + raise ValueError("Conflicting prompt identity: " + key) + evidence = previous.get("outcomes", []) + row.get("outcomes", []) + row["association_methods"] = sorted( + set(previous.get("association_methods", [])) + | set(row.get("association_methods", [])) + ) + else: + evidence = row.get("outcomes", []) + by_id = {} + for event in evidence: + if event.get("type") not in EVIDENCE_TYPES: + raise ValueError("Unsupported evidence type") + if (event.get("session_id"), event.get("project")) != ( + row.get("session_id"), + row.get("project"), + ): + raise ValueError("Evidence scope mismatch") + event_id = event.get("event_id") or _digest(event) + if event_id in by_id and by_id[event_id] != event: + raise ValueError("Conflicting evidence identity: " + event_id) + by_id[event_id] = event + row["outcomes"] = sorted( + by_id.values(), key=lambda e: (_parse_ts(e["queued_at"]), _digest(e)) + ) + row["outcome_score"] = _compute_outcome_score(row["outcomes"]) + row.update(verification_status="unverified", learning_eligible=False) + revision = _digest(row["outcomes"]) + history = list(previous.get("evidence_revisions", [])) if previous else [] + if not history or history[-1]["digest"] != revision: + history.append( + { + "digest": revision, + "evidence_count": len(row["outcomes"]), + "outcome_score": row["outcome_score"], + } + ) + row["evidence_revisions"] = history + rows[key] = row + content = "".join(line + "\n" for line in preserved) + content += "".join(json.dumps(row) + "\n" for row in rows.values()) + if content != old: + atomic_write(path, content) if __name__ == "__main__": linked = link_outcomes() - existing_before = len(_existing_prompt_ids()) write_linked_outcomes(linked) - new_count = len(_existing_prompt_ids()) - existing_before - print(f"Linked {len(linked)} candidates, {new_count} new (idempotent skip of duplicates)") - if linked: - scores = [l["outcome_score"] for l in linked] - print( - f"Score range: {min(scores):.2f} – {max(scores):.2f}, mean: {sum(scores) / len(scores):.2f}" - ) + print(f"Linked {len(linked)} advisory candidates; all remain unverified") diff --git a/mcp_handlers.py b/mcp_handlers.py index 6e1884d..9a7c1f3 100644 --- a/mcp_handlers.py +++ b/mcp_handlers.py @@ -13,11 +13,11 @@ - Exceptions propagate; callers decide whether to raise HTTPException (route) or wrap in an error envelope (MCP tool). -Crash-proof decision writes: - record_learning_decision() appends directly to ~/.cortex/decisions.jsonl. +Durable local decision writes: + record_learning_decision() commits a locked snapshot of decisions.jsonl. If the primary append fails (permissions, transient FS error), the entry is spooled to ~/.cortex/spool/decision-.json — one file per entry, so - concurrent sessions never contend — and flushed opportunistically on the + concurrent sessions use independent spool files — and flushed opportunistically on the next successful record call or explicitly via `cortex doctor --fix`. Paths resolve through state_paths.get_cortex_dir() at call time (honors @@ -56,15 +56,14 @@ def _spool_dir() -> Path: return get_cortex_dir() / "spool" -# ─── Decision recording (crash-proof) ────────────────────────────────── +# ─── Decision recording (durable local snapshots) ────────────────────────────────── def _append_line(path: Path, entry: Dict[str, Any]) -> None: - """Append one JSON line. Single-line O_APPEND writes are atomic in - practice for the handful of local processes that share this file.""" - path.parent.mkdir(parents=True, exist_ok=True) - with open(path, "a", encoding="utf-8") as f: - f.write(json.dumps(entry) + "\n") + """Commit a decision and its correction marker in one local transaction.""" + from intelligence.durable_jsonl import append_decision + + append_decision(path, entry) def record_learning_decision( @@ -88,8 +87,8 @@ def record_learning_decision( mutated (append-only, audit-safe) — this is the RDF-triple supersession pattern the P1 curation research called for. - Never loses a decision: on primary-append failure the entry lands in - the spool instead and the response carries `"spooled": true`. + Primary failure falls back to a durable spool and reports "spooled": true. + If both stores fail, the error propagates; no success is acknowledged. """ # Opportunistic flush of anything a previous failure left behind. try: @@ -128,18 +127,6 @@ def record_learning_decision( try: _append_line(_decisions_file(), entry) - if supersedes: - # Append-only tombstone: mark the old id superseded without mutating - # its original line. A tombstone failure must not fail the record. - try: - _append_line(_decisions_file(), { - "decision_id": supersedes, - "superseded_by": decision_id, - "timestamp": entry["timestamp"], - "source": "supersede", - }) - except Exception: - pass return { "recorded": True, "decision_id": decision_id, @@ -149,7 +136,9 @@ def record_learning_decision( spool_dir = _spool_dir() spool_dir.mkdir(parents=True, exist_ok=True) spool_path = spool_dir / f"decision-{decision_id}.json" - spool_path.write_text(json.dumps(entry), encoding="utf-8") + from intelligence.durable_jsonl import atomic_write + + atomic_write(spool_path, json.dumps(entry)) return { "recorded": True, "spooled": True, @@ -177,33 +166,23 @@ def flush_spool() -> Dict[str, Any]: if not spool.exists(): return {"flushed": 0, "skipped": 0, "remaining": 0} - decisions_path = _decisions_file() - existing_ids = set() - if decisions_path.exists(): - for line in decisions_path.read_text(encoding="utf-8").splitlines(): - try: - existing_ids.add(json.loads(line).get("decision_id")) - except (json.JSONDecodeError, AttributeError): - continue + from intelligence.durable_jsonl import append_decision, locked flushed = skipped = 0 - for spool_path in sorted(spool.glob("decision-*.json")): - try: - entry = json.loads(spool_path.read_text(encoding="utf-8")) - except (json.JSONDecodeError, OSError): - continue # unreadable spool entry: leave for inspection - if entry.get("decision_id") in existing_ids: - spool_path.unlink() # already durable in decisions.jsonl - skipped += 1 - continue - decisions_path.parent.mkdir(parents=True, exist_ok=True) - with open(decisions_path, "a", encoding="utf-8") as f: - f.write(json.dumps(entry) + "\n") - f.flush() - os.fsync(f.fileno()) - existing_ids.add(entry.get("decision_id")) - spool_path.unlink() - flushed += 1 + # Serialize spool readers as well as journal writers. No writer takes these + # locks in reverse order; primary writes release the journal before spooling. + with locked(spool / "replay"): + for spool_path in sorted(spool.glob("decision-*.json")): + try: + entry = json.loads(spool_path.read_text(encoding="utf-8")) + if not isinstance(entry, dict) or not entry.get("decision_id"): + continue + added = append_decision(_decisions_file(), entry) + except (ValueError, OSError): + continue # preserve corrupt, conflicting or unavailable entries + spool_path.unlink() + flushed += int(added) + skipped += int(not added) return {"flushed": flushed, "skipped": skipped, "remaining": spool_depth()} @@ -274,9 +253,7 @@ def normalize_recommendations( Mutates and returns the passed dict (parity with the route). """ if project and "recommendations" in recommendations: - filtered = [ - r for r in recommendations["recommendations"] if r.get("project") == project - ] + filtered = [r for r in recommendations["recommendations"] if r.get("project") == project] recommendations["recommendations"] = filtered[:limit] elif "recommendations" in recommendations: recommendations["recommendations"] = recommendations["recommendations"][:limit] @@ -402,9 +379,7 @@ def plans_progress(project: str = "", limit: int = 25) -> Dict[str, Any]: ) if project: proj_lower = project.lower() - summaries = [ - s for s in summaries if str(s.get("project") or "").lower() == proj_lower - ] + summaries = [s for s in summaries if str(s.get("project") or "").lower() == proj_lower] # Newest first: created_at is ISO-8601, so it sorts lexically. Missing # timestamps sort last rather than crashing the comparison. @@ -432,9 +407,7 @@ def create_plan(project: str, title: Optional[str] = None) -> Dict[str, Any]: all_goals = parser.parse() project_lower = project.lower() - project_goals = [ - g for g in all_goals if not g.project or g.project.lower() == project_lower - ] + project_goals = [g for g in all_goals if not g.project or g.project.lower() == project_lower] ts = int(time.time()) plan_id = f"plan_{project}_{ts}" @@ -467,9 +440,7 @@ def create_plan(project: str, title: Optional[str] = None) -> Dict[str, Any]: # ─── /v2/outcomes ────────────────────────────────────────────────────── -def read_outcomes( - project: str = "", limit: int = 20, exclude_types: str = "" -) -> Dict[str, Any]: +def read_outcomes(project: str = "", limit: int = 20, exclude_types: str = "") -> Dict[str, Any]: """Read recorded outcomes from ~/.cortex/outcomes.jsonl. This is the real outcome store written by feedback.FeedbackLogger @@ -522,9 +493,7 @@ def read_outcomes( prefixes = tuple(p.strip() for p in exclude_types.split(",") if p.strip()) if prefixes: entries = [ - e - for e in entries - if not str(e.get("recommendation_type") or "").startswith(prefixes) + e for e in entries if not str(e.get("recommendation_type") or "").startswith(prefixes) ] # Newest first by timestamp string (ISO-8601 sorts lexically). diff --git a/scripts/memory_maintenance.py b/scripts/memory_maintenance.py index 022f371..f7ae1dd 100644 --- a/scripts/memory_maintenance.py +++ b/scripts/memory_maintenance.py @@ -1,192 +1,21 @@ #!/usr/bin/env python3 -"""Memory-loop maintenance — keeps the compounding layer from silently rotting. +"""Compatibility entrypoint; implementation ships in the installed package.""" -Runs three idempotent jobs that the 2026-07 memory-restoration work identified -as needing a periodic driver (nothing was scheduling them, so they decayed): - - 1. failure emission — fold operational failures (restarts, alerts, scheduler - errors, pytest) into the outcome stream as weighted `failed` outcomes, so - the learning loop has a real failure signal to calibrate against. - 2. edge regeneration — rebuild the knowledge-graph edges from current nodes - when the edge set has collapsed (the schema-mismatch regression wiped - 1247 edges to 0 between April and June 2026). - 3. decision index — recorded decisions are auto-loaded by HybridRetriever - on construction, so a rebuilt retriever is enough; we just report the - count here for the maintenance log. - -Idempotent and safe to run on a short interval (LaunchAgent / cron). Each job -guards its own exceptions so one failing job never blocks the others. Writes a -JSON run-report to ~/.cortex/maintenance/ for auditability. - -Run standalone: python -m scripts.memory_maintenance -""" - -from __future__ import annotations - -import json -import os import sys -from datetime import datetime, timezone from pathlib import Path -# Make repo-root modules importable when invoked as a script. _REPO_ROOT = Path(__file__).resolve().parent.parent if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) -CORTEX_DIR = Path(os.environ.get("CORTEX_HOME", str(Path.home() / ".cortex"))) -REPORT_DIR = CORTEX_DIR / "maintenance" - - -def _emit_failures(lookback_minutes: int) -> dict: - """Job 1: emit collapsed operational failures into the outcome stream.""" - try: - from intelligence import failure_emitter - - result = failure_emitter.run_once(lookback_minutes=lookback_minutes) - return {"job": "failure_emission", "ok": True, **result} - except Exception as exc: # never let one job kill the run - return {"job": "failure_emission", "ok": False, "error": str(exc)} - - -def _regenerate_edges() -> dict: - """Job 2: sync decisions into the graph, then rebuild edges if collapsed. - - First imports any new recorded decisions as DECISION nodes (idempotent), - then regenerates edges when the edge set is empty (the failure mode) OR - when new decision nodes were just added (so they get linked). A healthy - graph with no new decisions is left untouched to avoid churn. - """ - try: - from engines.synthesis import ContextGraph - - graph = ContextGraph() - before = len(graph.edges) - # Keep the graph in sync with newly recorded decisions. - new_decisions = graph.import_decisions(save=False) - # Regenerate when edges collapsed, or to link freshly imported decisions. - if (before == 0 or new_decisions > 0) and graph.nodes: - total = graph.regenerate_edges(save=True) - return { - "job": "edge_regeneration", - "ok": True, - "regenerated": True, - "new_decision_nodes": new_decisions, - "edges_before": before, - "edges_after": total, - "nodes": len(graph.nodes), - } - return { - "job": "edge_regeneration", - "ok": True, - "regenerated": False, - "new_decision_nodes": new_decisions, - "edges": before, - "nodes": len(graph.nodes), - } - except Exception as exc: - return {"job": "edge_regeneration", "ok": False, "error": str(exc)} - - -def _decision_index() -> dict: - """Job 3: report how many recorded decisions are recall-indexable.""" - try: - from intelligence.memory.hybrid_retriever import _load_decision_patterns - - return { - "job": "decision_index", - "ok": True, - "decisions_indexable": len(_load_decision_patterns()), - } - except Exception as exc: - return {"job": "decision_index", "ok": False, "error": str(exc)} - - -def run_once(lookback_minutes: int = 90) -> dict: - """Run all maintenance jobs once and persist a report.""" - report = { - "timestamp": datetime.now(timezone.utc).isoformat(), - "jobs": [ - _emit_failures(lookback_minutes), - _regenerate_edges(), - _decision_index(), - ], - } - try: - REPORT_DIR.mkdir(parents=True, exist_ok=True) - with (REPORT_DIR / "maintenance_history.jsonl").open("a") as f: - f.write(json.dumps(report) + "\n") - except OSError: - pass # reporting is best-effort - return report - - -def backfill_importance() -> dict: - """One-time (idempotent) P1 backfill: score decisions that predate the - importance heuristic. Entries already carrying `importance` are left as-is, - tombstones are skipped. Writes a `.bak` and rewrites atomically (temp + - rename). Safe to run repeatedly — a second run is a no-op. - """ - from state_paths import get_cortex_dir - from intelligence.memory.importance import _importance_score, IMPORTANCE_FLOOR - - path = get_cortex_dir() / "decisions.jsonl" - if not path.exists(): - return {"job": "backfill_importance", "status": "no_file", "scored": 0} - - lines = path.read_text().splitlines() - scored = skipped = flagged = 0 - out = [] - for line in lines: - s = line.strip() - if not s: - continue - try: - d = json.loads(s) - except json.JSONDecodeError: - out.append(line) # preserve unparseable lines verbatim - skipped += 1 - continue - # Skip tombstones and already-scored entries (idempotent). - if d.get("superseded_by") or "importance" in d or not d.get("decision"): - out.append(json.dumps(d)) - skipped += 1 - continue - score = _importance_score( - d.get("decision", ""), d.get("context", ""), - d.get("alternatives", ""), d.get("rationale", ""), - ) - d["importance"] = score - if score < IMPORTANCE_FLOOR: - d["low_signal"] = True - flagged += 1 - out.append(json.dumps(d)) - scored += 1 - - if scored: - path.with_suffix(".jsonl.bak").write_text("\n".join(lines) + "\n") - tmp = path.with_suffix(".jsonl.tmp") - tmp.write_text("\n".join(out) + "\n") - tmp.replace(path) # atomic - - return {"job": "backfill_importance", "status": "ok", - "scored": scored, "skipped": skipped, "flagged_low_signal": flagged} - - -def main() -> int: - import argparse - - p = argparse.ArgumentParser(description="Cortex memory-loop maintenance") - p.add_argument("--lookback-minutes", type=int, default=90) - p.add_argument("--backfill-importance", action="store_true", - help="One-time P1 backfill: score decisions lacking an importance key (idempotent)") - args = p.parse_args() - if args.backfill_importance: - print(json.dumps(backfill_importance(), indent=2)) - return 0 - print(json.dumps(run_once(lookback_minutes=args.lookback_minutes), indent=2)) - return 0 - +from intelligence.memory.maintenance import ( # noqa: E402,F401 + _decision_index, + _emit_failures, + _regenerate_edges, + backfill_importance, + main, + run_once, +) if __name__ == "__main__": raise SystemExit(main()) diff --git a/tests/test_correction_acceptance.py b/tests/test_correction_acceptance.py new file mode 100644 index 0000000..1d7b382 --- /dev/null +++ b/tests/test_correction_acceptance.py @@ -0,0 +1,190 @@ +"""Acceptance: corrected advice stays out of graph context and live recall.""" + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from engines.synthesis import ContextGraph, Edge, EdgeType, Node, NodeType +from mcp_handlers import record_learning_decision + + +@pytest.fixture +def state(tmp_path, monkeypatch): + monkeypatch.setenv("CORTEX_STATE_DIR", str(tmp_path)) + monkeypatch.setenv("CORTEX_HOME", str(tmp_path / "wrong")) + return tmp_path + + +def seeded(state): + old = record_learning_decision("Use obsolete quasar advice", project="pilot") + graph = ContextGraph() + graph.import_decisions() + old_id = "decision:" + old["decision_id"] + graph.add_node(Node("pattern:pilot", NodeType.PATTERN, "quasar pattern", {"project": "pilot"})) + graph.add_node(Node("goal:pilot", NodeType.GOAL, "quasar goal", {})) + graph.add_edge(Edge("goal:pilot", old_id, EdgeType.RELATES_TO)) + graph.add_edge(Edge(old_id, "pattern:pilot", EdgeType.RELATES_TO)) + return graph, old + + +def assert_absent(graph, old_id): + assert graph.get_node(old_id) is None + assert old_id not in [n.id for n in graph.query("quasar")] + assert old_id not in [n.id for n in graph.get_nodes_by_type(NodeType.DECISION)] + assert graph.get_related(old_id) == [] + assert graph.get_pointing_to(old_id) == [] + assert graph.get_related("goal:pilot") == [] + assert graph.get_pointing_to("pattern:pilot") == [] + assert graph.find_path("goal:pilot", "pattern:pilot") is None + assert graph.get_subgraph(old_id) == {"nodes": [], "edges": []} + assert [n["id"] for n in graph.get_subgraph("goal:pilot")["nodes"]] == ["goal:pilot"] + + +def test_correction_filters_all_graph_read_paths_without_save(state): + graph, old = seeded(state) + new = record_learning_decision( + "Use corrected quasar advice", project="pilot", supersedes=old["decision_id"] + ) + old_id = "decision:" + old["decision_id"] + assert_absent(graph, old_id) + assert graph.get_node("decision:" + new["decision_id"]).name == "Use corrected quasar advice" + assert old_id in graph.nodes # audit history remains intact + assert graph.nodes[old_id].data["superseded_by"] == new["decision_id"] + # Disk still contains the old graph; a new process view must reconcile it. + assert_absent(ContextGraph(), old_id) + + +def test_graph_recovery_after_interrupted_save_and_retry(state, monkeypatch): + graph, old = seeded(state) + new = record_learning_decision("corrected quasar advice", supersedes=old["decision_id"]) + write = graph._atomic_write_json + + def fail_edges(path, payload): + if path.name == "edges.json": + raise OSError("interrupted between graph files") + write(path, payload) + + with monkeypatch.context() as patch: + patch.setattr(graph, "_atomic_write_json", fail_edges) + with pytest.raises(OSError, match="synchronization"): + graph.import_decisions() + assert_absent(ContextGraph(), "decision:" + old["decision_id"]) + graph.import_decisions() # dirty sync must retry even when content is unchanged + restored = ContextGraph() + assert restored.get_node("decision:" + new["decision_id"]) is not None + + +def test_maintenance_reports_save_failure(state, monkeypatch): + from intelligence.memory.maintenance import _regenerate_edges + + record_learning_decision("quasar durable advice") + monkeypatch.setattr(ContextGraph, "_save", lambda self: False) + result = _regenerate_edges() + assert result["ok"] is False + assert "persisted" in result["error"] + + +def test_edge_regeneration_reports_save_failure(state, monkeypatch): + graph = ContextGraph() + graph.nodes["pattern:p"] = Node("pattern:p", NodeType.PATTERN, "pattern", {}) + monkeypatch.setattr(graph, "_save", lambda: False) + with pytest.raises(OSError, match="regeneration"): + graph.regenerate_edges() + + +def test_tombstone_only_correction_is_persisted_and_idempotent(state): + graph, old = seeded(state) + with (state / "decisions.jsonl").open("a") as stream: + stream.write( + json.dumps({"decision_id": old["decision_id"], "superseded_by": "external"}) + "\n" + ) + assert graph.import_decisions() == 1 + assert graph.import_decisions() == 0 + assert_absent(ContextGraph(), "decision:" + old["decision_id"]) + persisted = json.loads((state / "graph" / "nodes.json").read_text()) + assert ( + next(n for n in persisted if n["id"] == "decision:" + old["decision_id"])["data"][ + "superseded_by" + ] + == "external" + ) + + +def test_unavailable_journal_hides_imported_decisions(state): + graph, old = seeded(state) + (state / "decisions.jsonl").unlink() + assert_absent(graph, "decision:" + old["decision_id"]) + assert graph.get_node("pattern:pilot") is not None + + +def test_live_retriever_refreshes_corrected_advice(state, monkeypatch): + import intelligence.memory.hybrid_retriever as hr + + monkeypatch.setattr(hr, "_DECISIONS_PATH", None) + old = record_learning_decision("Use obsolete quasar advice") + retriever = hr.HybridRetriever([], include_conversation_digests=False) + assert [p.id for p, _ in retriever.search("quasar", alpha=0)] == [ + "decision:" + old["decision_id"] + ] + new = record_learning_decision("Use corrected quasar advice", supersedes=old["decision_id"]) + assert [p.id for p, _ in retriever.search("quasar", alpha=0)] == [ + "decision:" + new["decision_id"] + ] + + +def test_fresh_process_maintenance_and_query_acceptance(state): + graph, old = seeded(state) + new = record_learning_decision( + "Use corrected quasar advice", project="pilot", supersedes=old["decision_id"] + ) + script = """import json +from intelligence.memory.maintenance import _regenerate_edges +from engines.synthesis import ContextGraph +from intelligence.memory.hybrid_retriever import HybridRetriever +report = _regenerate_edges() +graph = ContextGraph() +print(json.dumps({"report": report, "graph": [n.id for n in graph.query("advice")], + "recall": [p.id for p, _ in HybridRetriever([], include_conversation_digests=False).search("advice", alpha=0)]})) +""" + env = { + "PATH": os.environ["PATH"], + "CORTEX_STATE_DIR": str(state), + "CORTEX_HOME": str(state / "wrong"), + "PYTHONPATH": str(Path(__file__).resolve().parents[1]), + } + result = subprocess.run( + [sys.executable, "-c", script], + cwd="/tmp", + env=env, + capture_output=True, + text=True, + check=True, + timeout=30, + ) + data = json.loads(result.stdout) + assert data["report"]["ok"] is True + assert data["graph"] == data["recall"] == ["decision:" + new["decision_id"]] + assert not (state / "wrong" / "graph").exists() + + +def test_live_semantic_retriever_discards_stale_vectors(state, monkeypatch): + import intelligence.memory.hybrid_retriever as hr + from intelligence.embeddings_client import EmbeddingsClient + + monkeypatch.setattr(hr, "_DECISIONS_PATH", None) + monkeypatch.setenv("CORTEX_EMBED_BACKEND", "hashing") + old = record_learning_decision("Use obsolete quasar advice") + retriever = hr.HybridRetriever( + [], embeddings_client=EmbeddingsClient(), include_conversation_digests=False + ) + assert retriever.pattern_embeddings is not None + new = record_learning_decision("Use corrected quasar advice", supersedes=old["decision_id"]) + assert [p.id for p, _ in retriever.search("quasar", alpha=1)] == [ + "decision:" + new["decision_id"] + ] + assert retriever.pattern_embeddings is None + assert retriever.embeddings_available is False diff --git a/tests/test_evidence_memory_integrity.py b/tests/test_evidence_memory_integrity.py new file mode 100644 index 0000000..1ce6bfc --- /dev/null +++ b/tests/test_evidence_memory_integrity.py @@ -0,0 +1,271 @@ +"""Failure/replay contracts using actual disk state and separate processes.""" + +import json +import os +import subprocess +import sys +from concurrent.futures import ProcessPoolExecutor +from pathlib import Path + +import pytest + +import intelligence.durable_jsonl as durable +import intelligence.outcome_linker as linker +import mcp_handlers + + +def event(kind, second, **fields): + return dict( + type=kind, + queued_at=f"2026-09-07T00:{second // 60:02}:{second % 60:02}+00:00", + session_id="a", + project="cortex", + **fields, + ) + + +def link(tmp_path, events): + queue = tmp_path / "queue.jsonl" + queue.write_text("".join(json.dumps(e) + "\n" for e in events)) + return linker.link_outcomes(queue) + + +def test_scope_types_and_unknown_sessions(tmp_path): + prompt = event("prompt_received", 0, prompt="repair", prompt_id="p") + wrong = event("git_commit", 2) + wrong["session_id"] = "b" + other_project = event("git_commit", 3) + other_project["project"] = "other" + assert link(tmp_path, [prompt, wrong, other_project, event("prompt_received", 5)]) == [] + prompt["session_id"] = None + wrong["session_id"] = None + assert link(tmp_path, [prompt, wrong]) == [] + + +def test_stable_id_and_late_explicit_evidence(tmp_path): + prompt = event("prompt_received", 5, prompt="repair", prompt_id="p") + early = event("git_commit", 15, event_id="commit") + original = link(tmp_path, [prompt, early]) + older = event("prompt_received", 0, prompt="unrelated") + late = event("test_result", 600, prompt_id="p", passed=0, failed=1) + refreshed = link(tmp_path, [older, prompt, early, late]) + assert len(refreshed) == 1 + assert original[0]["prompt_id"] == refreshed[0]["prompt_id"] + assert len(refreshed[0]["outcomes"]) == 2 + assert refreshed[0]["learning_eligible"] is False + assert refreshed[0]["verification_status"] == "unverified" + dest = tmp_path / "outcomes.jsonl" + linker.write_linked_outcomes(original * 2, dest) + linker.write_linked_outcomes(refreshed, dest) + linker.write_linked_outcomes(original, dest) # stale worker must not revert evidence + rows = list(durable.read_records(dest.read_text())) + assert len(rows) == 1 + assert len(rows[0]["outcomes"]) == 2 + assert rows[0]["outcome_score"] == 0.6 + assert len(rows[0]["evidence_revisions"]) == 2 + + +def test_wrong_explicit_reference_never_falls_back(tmp_path): + assert ( + link( + tmp_path, + [ + event("prompt_received", 0, prompt_id="p"), + event("git_commit", 2, prompt_id="different"), + ], + ) + == [] + ) + + +def test_ambiguous_temporal_prompt_is_not_credited(tmp_path): + assert ( + link( + tmp_path, + [ + event("prompt_received", 0, prompt="one"), + event("prompt_received", 0, prompt="two"), + event("git_commit", 2), + ], + ) + == [] + ) + + +def test_conflicting_evidence_rolls_back_entire_update(tmp_path): + rows = link(tmp_path, [event("prompt_received", 0), event("git_commit", 1, event_id="e")]) + dest = tmp_path / "outcomes.jsonl" + linker.write_linked_outcomes(rows, dest) + before = dest.read_bytes() + rows[0]["outcomes"][0]["hash"] = "conflicting payload" + with pytest.raises(ValueError, match="Conflicting evidence"): + linker.write_linked_outcomes(rows, dest) + assert dest.read_bytes() == before + + +def _write_link(args): + path, row = args + linker.write_linked_outcomes([row], Path(path)) + + +def test_concurrent_linkers_merge_without_lost_updates(tmp_path): + rows = link(tmp_path, [event("prompt_received", 0), event("git_commit", 1)]) + variants = [] + for i in range(8): + row = dict(rows[0], outcomes=[event("git_commit", i + 1, event_id=str(i))]) + variants.append((str(tmp_path / "outcomes.jsonl"), row)) + with ProcessPoolExecutor(max_workers=4) as pool: + list(pool.map(_write_link, variants)) + result = list(durable.read_records((tmp_path / "outcomes.jsonl").read_text())) + assert len(result) == 1 + assert len(result[0]["outcomes"]) == 8 + + +def test_interrupted_replace_preserves_original(tmp_path, monkeypatch): + path = tmp_path / "decisions.jsonl" + durable.append_decision(path, {"decision_id": "old", "decision": "old advice"}) + before = path.read_bytes() + + def fail(*args): + raise OSError("interrupted before commit") + + monkeypatch.setattr(durable.os, "replace", fail) + with pytest.raises(OSError): + durable.append_decision( + path, {"decision_id": "new", "decision": "corrected", "supersedes": "old"} + ) + assert path.read_bytes() == before + assert list(tmp_path.glob(".decisions.jsonl-*")) == [] + + +def test_failed_primary_correction_replays_atomically(tmp_path, monkeypatch): + monkeypatch.setenv("CORTEX_STATE_DIR", str(tmp_path)) + old = mcp_handlers.record_learning_decision("obsolete advice") + + def fail(*args): + raise OSError("primary unavailable") + + with monkeypatch.context() as patch: + patch.setattr(mcp_handlers, "_append_line", fail) + new = mcp_handlers.record_learning_decision( + "corrected advice", supersedes=old["decision_id"] + ) + assert new["spooled"] is True + assert mcp_handlers.flush_spool() == {"flushed": 1, "skipped": 0, "remaining": 0} + records = list(durable.read_records((tmp_path / "decisions.jsonl").read_text())) + assert [d["decision_id"] for d in durable.active_decisions(records)] == [new["decision_id"]] + assert len(records) == 3 + + +def test_conflicting_spool_is_retained(tmp_path, monkeypatch): + monkeypatch.setenv("CORTEX_STATE_DIR", str(tmp_path)) + saved = mcp_handlers.record_learning_decision("original") + spool = tmp_path / "spool" + spool.mkdir() + (spool / "decision-conflict.json").write_text(json.dumps(dict(saved, decision="different"))) + assert mcp_handlers.flush_spool() == {"flushed": 0, "skipped": 0, "remaining": 1} + + +def test_no_acknowledgement_when_primary_and_spool_fail(tmp_path, monkeypatch): + monkeypatch.setenv("CORTEX_STATE_DIR", str(tmp_path)) + + def fail(*args): + raise OSError("storage unavailable") + + monkeypatch.setattr(durable, "atomic_write", fail) + with pytest.raises(OSError, match="storage unavailable"): + mcp_handlers.record_learning_decision("not safely stored") + + +def run_process(script, state): + env = { + "PATH": os.environ["PATH"], + "CORTEX_STATE_DIR": str(state), + "CORTEX_HOME": str(state / "wrong"), + "CORTEX_EMBED_BACKEND": "hashing", + "PYTHONPATH": str(Path(__file__).resolve().parents[1]), + } + result = subprocess.run( + [sys.executable, "-c", script], + env=env, + cwd="/tmp", + capture_output=True, + text=True, + check=True, + timeout=30, + ) + return json.loads(result.stdout) + + +def test_fresh_process_record_correct_reindex_and_recall(tmp_path): + ids = run_process( + """import json +from mcp_handlers import record_learning_decision +old = record_learning_decision("Use obsolete quasar database advice") +new = record_learning_decision("Use corrected quasar database advice", supersedes=old["decision_id"]) +print(json.dumps([old["decision_id"], new["decision_id"]])) +""", + tmp_path, + ) + result = run_process( + """import json +from intelligence.memory.hybrid_retriever import HybridRetriever +from intelligence.memory.pattern_indexer import PatternIndexer +patterns = PatternIndexer._load_decisions() +r = HybridRetriever(patterns, include_conversation_digests=False) +print(json.dumps({"indexed": [p.id for p in patterns], "recalled": [p.id for p, _ in r.search("quasar database", alpha=0)]})) +""", + tmp_path, + ) + assert result == {"indexed": ["decision:" + ids[1]], "recalled": ["decision:" + ids[1]]} + assert not (tmp_path / "wrong" / "decisions.jsonl").exists() + + +def test_self_contained_correction_suppresses_legacy_partial_write(tmp_path, monkeypatch): + monkeypatch.setenv("CORTEX_STATE_DIR", str(tmp_path)) + import intelligence.memory.hybrid_retriever as hr + + monkeypatch.setattr(hr, "_DECISIONS_PATH", None) + path = tmp_path / "decisions.jsonl" + path.write_text(json.dumps({"decision_id": "old", "decision": "old advice"}) + "\n") + stale = hr._load_decision_patterns() + with path.open("a") as stream: + stream.write( + json.dumps({"decision_id": "new", "decision": "new advice", "supersedes": "old"}) + "\n" + ) + retriever = hr.HybridRetriever(stale, include_conversation_digests=False) + assert [p.id for p in retriever.patterns] == ["decision:new"] + + +def test_cache_does_not_cross_state_directories(tmp_path, monkeypatch): + import intelligence.memory.hybrid_retriever as hr + + monkeypatch.setattr(hr, "_DECISIONS_PATH", None) + for name in ("one", "two"): + root = tmp_path / name + root.mkdir() + (root / "decisions.jsonl").write_text(json.dumps({"decision_id": name, "decision": name})) + os.utime(root / "decisions.jsonl", (100, 100)) + monkeypatch.setenv("CORTEX_STATE_DIR", str(root)) + assert [p.id for p in hr._load_decision_patterns()] == ["decision:" + name] + + +def test_failure_after_replace_replays_without_duplicate(tmp_path, monkeypatch): + monkeypatch.setenv("CORTEX_STATE_DIR", str(tmp_path)) + fsync = durable.os.fsync + calls = [] + + def fail_directory_once(fd): + calls.append(fd) + if len(calls) == 2: + raise OSError("directory sync failed after replace") + return fsync(fd) + + with monkeypatch.context() as patch: + patch.setattr(durable.os, "fsync", fail_directory_once) + result = mcp_handlers.record_learning_decision("survive ambiguous acknowledgement") + assert result["spooled"] is True + assert mcp_handlers.flush_spool() == {"flushed": 0, "skipped": 1, "remaining": 0} + rows = list(durable.read_records((tmp_path / "decisions.jsonl").read_text())) + assert len(rows) == 1 + assert rows[0]["decision_id"] == result["decision_id"] diff --git a/tests/test_hybrid_retriever.py b/tests/test_hybrid_retriever.py index 067aebe..08810af 100644 --- a/tests/test_hybrid_retriever.py +++ b/tests/test_hybrid_retriever.py @@ -21,8 +21,11 @@ # integration, so both auto-loaded sources are pointed at nonexistent paths. import intelligence.memory.hybrid_retriever as _hr_module -_hr_module._DIGESTS_PATH = Path("/nonexistent/digests.jsonl") -_hr_module._DECISIONS_PATH = Path("/nonexistent/decisions.jsonl") + +@pytest.fixture(autouse=True) +def isolate_memory_sources(monkeypatch): + monkeypatch.setattr(_hr_module, "_DIGESTS_PATH", Path("/nonexistent/digests.jsonl")) + monkeypatch.setattr(_hr_module, "_DECISIONS_PATH", Path("/nonexistent/decisions.jsonl")) # Fixtures @@ -451,6 +454,7 @@ class TestDecisionCuration: def _seed(self, tmp_path, entries): import json as _json + p = tmp_path / "decisions.jsonl" p.write_text("\n".join(_json.dumps(e) for e in entries) + "\n") return p @@ -469,9 +473,23 @@ def _reload_with(self, path): def test_superseded_decision_excluded(self, tmp_path): entries = [ - {"decision_id": "dec_old", "decision": "original call about the schema", "timestamp": datetime.now().isoformat()}, - {"decision_id": "dec_new", "decision": "revised call about the schema", "supersedes": "dec_old", "timestamp": datetime.now().isoformat()}, - {"decision_id": "dec_old", "superseded_by": "dec_new", "timestamp": datetime.now().isoformat(), "source": "supersede"}, + { + "decision_id": "dec_old", + "decision": "original call about the schema", + "timestamp": datetime.now().isoformat(), + }, + { + "decision_id": "dec_new", + "decision": "revised call about the schema", + "supersedes": "dec_old", + "timestamp": datetime.now().isoformat(), + }, + { + "decision_id": "dec_old", + "superseded_by": "dec_new", + "timestamp": datetime.now().isoformat(), + "source": "supersede", + }, ] patterns = self._reload_with(self._seed(tmp_path, entries)) ids = {p.id for p in patterns} @@ -483,8 +501,19 @@ def test_superseded_decision_excluded(self, tmp_path): def test_low_importance_weighted_below_high(self, tmp_path): ts = datetime.now().isoformat() entries = [ - {"decision_id": "dec_low", "decision": "low signal decision text here", "importance": 2, "low_signal": True, "timestamp": ts}, - {"decision_id": "dec_high", "decision": "high signal decision text here", "importance": 9, "timestamp": ts}, + { + "decision_id": "dec_low", + "decision": "low signal decision text here", + "importance": 2, + "low_signal": True, + "timestamp": ts, + }, + { + "decision_id": "dec_high", + "decision": "high signal decision text here", + "importance": 9, + "timestamp": ts, + }, ] self._reload_with(self._seed(tmp_path, entries)) w = _hr_module._decision_weights @@ -494,7 +523,13 @@ def test_missing_importance_defaults_neutral(self, tmp_path): # Pre-P1 entries (no importance key) get a neutral weight, so # un-backfilled history isn't penalised. ts = datetime.now().isoformat() - entries = [{"decision_id": "dec_old", "decision": "legacy decision without a score", "timestamp": ts}] + entries = [ + { + "decision_id": "dec_old", + "decision": "legacy decision without a score", + "timestamp": ts, + } + ] self._reload_with(self._seed(tmp_path, entries)) # importance 5 → imp_factor 0.8; fresh → decay ~1.0 w = _hr_module._decision_weights["decision:dec_old"] diff --git a/tests/test_mcp_handlers.py b/tests/test_mcp_handlers.py index 46ea416..a95df8c 100644 --- a/tests/test_mcp_handlers.py +++ b/tests/test_mcp_handlers.py @@ -30,7 +30,7 @@ def state_dir(tmp_path, monkeypatch): def _lines(path: Path): - return [json.loads(l) for l in path.read_text().splitlines() if l.strip()] + return [json.loads(line) for line in path.read_text().splitlines() if line.strip()] # ─── Schema ──────────────────────────────────────────────────────────── @@ -106,8 +106,7 @@ def test_rich_decision_not_low_signal(state_dir): def test_decision_ids_unique_at_second_resolution(state_dir): """uuid ids fix the old dec_{int(time.time())} collision.""" ids = { - mcp_handlers.record_learning_decision(decision=f"d{i}")["decision_id"] - for i in range(20) + mcp_handlers.record_learning_decision(decision=f"d{i}")["decision_id"] for i in range(20) } assert len(ids) == 20 @@ -156,7 +155,15 @@ def boom(path, entry): # A stale spool file whose id already landed is skipped and removed. spool = state_dir / "spool" stale = spool / f"decision-{r1['decision_id']}.json" - stale.write_text(json.dumps({"decision_id": r1["decision_id"], "decision": "dupe"})) + stale.write_text( + json.dumps( + next( + e + for e in _lines(state_dir / "decisions.jsonl") + if e["decision_id"] == r1["decision_id"] + ) + ) + ) result = mcp_handlers.flush_spool() assert result == {"flushed": 0, "skipped": 1, "remaining": 0} assert len(_lines(state_dir / "decisions.jsonl")) == 2 # no dupe appended @@ -258,13 +265,31 @@ def test_outcome_stats_derives_accuracy_from_real_log(state_dir): now = datetime.now().isoformat() rows = [ # followed: 1 success + 1 partial → accuracy (1.0 + 0.5)/2 = 0.75 - {"timestamp": now, "recommendation_type": "goal_progress", "followed": True, - "outcome": "success", "source": "human", "context": {"project": "cortex"}}, - {"timestamp": now, "recommendation_type": "goal_progress", "followed": True, - "outcome": "partial", "source": "auto", "context": {"project": "cortex"}}, + { + "timestamp": now, + "recommendation_type": "goal_progress", + "followed": True, + "outcome": "success", + "source": "human", + "context": {"project": "cortex"}, + }, + { + "timestamp": now, + "recommendation_type": "goal_progress", + "followed": True, + "outcome": "partial", + "source": "auto", + "context": {"project": "cortex"}, + }, # not followed → excluded from accuracy, counted in total - {"timestamp": now, "recommendation_type": "blocker", "followed": False, - "outcome": "unknown", "source": "human", "context": {"project": "other"}}, + { + "timestamp": now, + "recommendation_type": "blocker", + "followed": False, + "outcome": "unknown", + "source": "human", + "context": {"project": "other"}, + }, ] (state_dir / "outcomes.jsonl").write_text("\n".join(json.dumps(r) for r in rows) + "\n") diff --git a/tests/test_outcome_linker.py b/tests/test_outcome_linker.py index e2af3d2..9dcb990 100644 --- a/tests/test_outcome_linker.py +++ b/tests/test_outcome_linker.py @@ -4,9 +4,8 @@ against synthesized real-shape interaction queues. They use tmp_path to keep production ~/.cortex paths untouched and module-level QUEUE/OUTCOMES intact. -Together they prove the headline "compounding intelligence" claim is live in -this build: prompts → outcomes are joined by session_id within a 90-second -window, scored, and persisted idempotently. +These exercise advisory session-scoped associations and persistence. They do +not demonstrate verified task success or compounding intelligence. """ from __future__ import annotations @@ -191,9 +190,7 @@ def test_isolated_paths_do_not_touch_module_globals(tmp_path: Path) -> None: def test_existing_prompt_ids_accepts_kwarg(tmp_path: Path) -> None: """The dedup helper must read from the provided path, not the module global.""" outcomes_path = tmp_path / "prompt_outcomes.jsonl" - outcomes_path.write_text( - json.dumps({"prompt_id": "sess_K_0", "outcome_score": 1.0}) + "\n" - ) + outcomes_path.write_text(json.dumps({"prompt_id": "sess_K_0", "outcome_score": 1.0}) + "\n") existing = _existing_prompt_ids(outcomes_path=outcomes_path) assert existing == {"sess_K_0"}