"""phase-memory ports — cya seam to phase-memory (markitect). This module is the explicit, inspectable boundary. All memory for assistance (preferences, project context, retrospection outcomes, etc.) flows through here. **Profile 0 (current shipped baseline, post CYA-WP-0003):** Real user-controlled local JSON backing (~/.config/cya/memory/.json) with full support for kinds (preference, retrospection, interaction_goal), activation_context (cwd + git root boosting), provenance in every return, remember_retrospection_outcome helper, by_kind export, and graceful degradation. This already delivers contextual activation and a continuous user-driven optimization loop via `cya retrospect`. See: - MemoryVision.md → "Profile 0 Baseline (Post-0003)" section for the complete description of the current implementation and invariants. - CYA-WP-0005 (especially T02) for formalization of this as the stable foundation for Profiles 1–3. - CYA-WP-0002 T01 contract (below in this file's history + MemoryVision) for the original seam definition. The public API signatures are stable. Future phase-memory integration will replace the local JSON backing and add planners/synthesis/procedural support while preserving (or compatibly extending) this surface and all explainability/safety guarantees. """ from __future__ import annotations import json import sys import time from pathlib import Path from typing import Any # Standard memory kinds used by cya (especially after CYA-WP-0003) KIND_PREFERENCE = "preference" KIND_RETROSPECTION = "retrospection" KIND_INTERACTION_GOAL = "interaction_goal" KIND_REFLECTION = "reflection" # Profile 1 (Reflexion-style verbal lessons) — T05 minimal spike def _warn_not_connected(feature: str) -> None: """Loud, visible marker that phase-memory is not yet wired (fallback path).""" msg = ( f"[phase-memory] {feature} called — phase-memory not yet connected. " "Falling back to local T02 json store (real phase graph in later slice). " "See T02 in workplan CYA-WP-0002 and MemoryVision contract." ) print(msg, file=sys.stderr) # --------------------------------------------------------------------------- # Real T02 backing (user-controlled, explainable, phase-ready) # Uses stdlib + optional phase_memory.models for structure. # Persists across cya invocations via ~/.config/cya/memory/.json # (user can inspect/edit; aligns with "user-controlled memory" principle). # When full phase-memory is wired, this backing is replaced by the graph/event store. # --------------------------------------------------------------------------- def _mem_path(scope: str = "cwd") -> Path: p = Path.home() / ".config" / "cya" / "memory" p.mkdir(parents=True, exist_ok=True) return p / f"{scope}.json" def _load(scope: str) -> list[dict[str, Any]]: f = _mem_path(scope) if f.exists(): try: data = json.loads(f.read_text(encoding="utf-8")) return data if isinstance(data, list) else [] except Exception: return [] return [] def _save(scope: str, items: list[dict[str, Any]]) -> None: f = _mem_path(scope) f.write_text(json.dumps(items, indent=2, default=str), encoding="utf-8") # --------------------------------------------------------------------------- # Explicit ports (the four capabilities from the workplan) # Refined in T01 (0003) per phase-memory architecture + interop + lifecycle. # Real (non-no-op) implementation added in T02 (0002): actual persist + recall. # Extended in T02 (0003) for contextual activation and retrospection support. # See docs/cya-memory-activation-and-retrospection-concept.md and MemoryVision.md. # --------------------------------------------------------------------------- def remember_preference( key: str, value: Any, scope: str = "cwd", *, profile: str | None = None, ttl: str | None = None, kind: str = KIND_PREFERENCE, ) -> None: """Remember a user preference, workflow pattern, retrospection outcome, or goal. `kind` defaults to "preference" for backward compatibility. Special kinds (e.g. "retrospection", "interaction_goal") are supported for higher-order memory from retrospection sessions (see CYA-WP-0003). Real T02 + 0003: persists to user-controlled json (scoped). """ try: items = _load(scope) item = { "key": key, "value": value, "ts": time.time(), "scope": scope, "profile": profile, "kind": kind, } # avoid exact dups for same key in small stores items = [i for i in items if i.get("key") != key] items.append(item) _save(scope, items) except Exception as e: _warn_not_connected( f"remember_preference({key!r}, scope={scope}, profile={profile}) err={e}" ) def recall_preferences( scope: str = "cwd", task_class: str | None = None, *, kinds: list[str] | None = None, profile: str | None = None, limit: int = 50, activation_context: dict | None = None, ) -> dict[str, Any]: """Recall relevant memory for the current context. Supports: - `kinds`: filter or boost specific kinds (e.g. ["interaction_goal", "retrospection"]) - `activation_context`: hints for smarter activation (e.g. {"cwd": "...", "project": "..."}) - Backward compatible with all previous call sites. Real T02 + 0003 extensions for contextual activation and retrospection support. """ try: items = _load(scope) # Basic kind filtering (existing behavior + 0003 enhancement) if kinds: items = [i for i in items if i.get("kind") in kinds or not i.get("kind")] # Simple activation boost: prefer items whose scope or profile matches context if activation_context: ctx_cwd = activation_context.get("cwd") or activation_context.get("scope") ctx_profile = activation_context.get("profile") boosted = [] normal = [] for item in items: if (ctx_cwd and item.get("scope") == ctx_cwd) or (ctx_profile and item.get("profile") == ctx_profile): boosted.append(item) else: normal.append(item) items = boosted + normal items = items[-limit:] # most recent first (after boosting) return { "items": items, "provenance": [ { "source": "cya-local-memory-json (T02+0003; activation + retrospection support)", "scope": scope, "count": len(items), "activation_context": activation_context, } ], "phase": "fluid", "profile": profile, "note": "real persist + contextual activation; full phase-memory in later slices", } except Exception as e: _warn_not_connected( f"recall_preferences(scope={scope}, task={task_class}, profile={profile}) err={e}" ) return {} def forget(scope: str = "cwd", keys: list[str] | None = None, *, profile: str | None = None) -> None: """Forget / reset memory (scoped). Real T02: removes from the user json. Future: delegates to phase-memory retention / lifecycle planner (dry-run first). """ try: if not keys: _save(scope, []) return items = _load(scope) items = [i for i in items if i.get("key") not in keys] _save(scope, items) except Exception as e: _warn_not_connected(f"forget(scope={scope}, keys={keys}, profile={profile}) err={e}") def export_memory(scope: str = "cwd", *, profile: str | None = None, kinds: list[str] | None = None) -> dict[str, Any]: """Inspect / export current memory for this project or user. Supports optional `kinds` filter (e.g. to export only retrospection records). Real T02 + 0003: improved transparency for different memory kinds. """ try: items = _load(scope) if kinds: items = [i for i in items if i.get("kind") in kinds] by_kind = {} for item in items: k = item.get("kind", "unknown") by_kind.setdefault(k, []).append(item) return { "status": "real (T02+0003 local json; activation + retrospection ready)", "scope": scope, "profile": profile, "count": len(items), "items": items, "by_kind": {k: len(v) for k, v in by_kind.items()}, "phase": "fluid", "provenance_summary": f"{len(items)} records from ~/.config/cya/memory/ (kinds: {list(by_kind.keys())})", "note": "User-controlled. Replace with real phase-memory when available.", "phases": ["ephemeral", "fluid", "stabilized", "rigid"], } except Exception as e: _warn_not_connected(f"export_memory(scope={scope}, profile={profile}) err={e}") return { "status": "error", "error": str(e), } def remember_retrospection_outcome( key: str, value: Any, scope: str = "cwd", *, profile: str | None = None, ) -> None: """Convenience helper to record outcomes from a retrospection session. Stores with kind="retrospection" or "interaction_goal" (caller decides via key/value). This is a thin wrapper around remember_preference for clarity in T04+. """ kind = KIND_RETROSPECTION if "retrospection" in str(key).lower() else KIND_INTERACTION_GOAL remember_preference(key, value, scope=scope, profile=profile, kind=kind) def remember_reflection( key: str, value: Any, scope: str = "cwd", *, profile: str | None = None, ) -> None: """Convenience helper for Profile 1 (Reflexion-style verbal self-improvement). Stores verbal lessons/reflections with kind="reflection" for preferential activation in future turns. Thin wrapper for the T05 minimal spike. """ remember_preference(key, value, scope=scope, profile=profile, kind=KIND_REFLECTION) __all__ = [ "remember_preference", "recall_preferences", "forget", "export_memory", "remember_retrospection_outcome", "remember_reflection", "KIND_PREFERENCE", "KIND_RETROSPECTION", "KIND_INTERACTION_GOAL", "KIND_REFLECTION", ]