coulomb/can-you-assist
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(memory): complete CYA-WP-0006 Profile 1 production hardening

Browse Source 
Add guided reflection capture with preview, cya memory reflections CLI,
near-duplicate compaction, budget-capped surfacing, and expanded tests.
Profile 1 is now documented as production-ready in README and MemoryVision.
This commit is contained in:
Bernd Worsch
2026-06-22 01:39:07 +02:00
parent a0d24a31eb
commit c14d09c14d
12 changed files with 735 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
src/cya/cli/main.py
View File

@@ -11,6 +11,7 @@ This module will evolve in T06 (orchestrator) but the surface contract stays sta
from __future__ import annotations
import json
import sys
import typer
@@ -108,6 +109,61 @@ def main(
)
memory_app = typer.Typer(
help="Inspect and manage user-controlled memory (Profile 0 + Profile 1).",
rich_markup_mode="rich",
)
app.add_typer(memory_app, name="memory")
@memory_app.command("reflections")
def memory_reflections(
scope: str = typer.Option(
".",
"--scope",
"-s",
help="Directory or scope to list reflections for.",
),
export_json: bool = typer.Option(
False,
"--json",
help="Export reflections as JSON (same as export_memory with kinds filter).",
),
) -> None:
"""List or export verbal reflections for a scope (Profile 1)."""
from cya.memory import export_memory, KIND_REFLECTION
from cya.memory.reflections import list_reflections, reflection_export_stats
if export_json:
data = export_memory(scope, kinds=[KIND_REFLECTION])
console.print(json.dumps(data, indent=2, default=str))
return
items = list_reflections(scope)
stats = reflection_export_stats(scope)
if not items:
console.print(f"[yellow]No reflections in scope {scope!r}.[/yellow]")
return
lines = []
for item in items:
prov = item.get("provenance") or {}
date = prov.get("session_date", "?")
lines.append(f"• [bold]{item.get('key')}[/bold] ({date}): {item.get('value')}")
console.print(
Panel(
"\n".join(lines),
title=f"Reflections in {scope} ({stats['reflection_count']} total)",
border_style="cyan",
)
)
if stats.get("reflection_counts_by_scope"):
console.print(
f"[dim]By scope: {stats['reflection_counts_by_scope']}[/dim]"
)
@app.command()
def retrospect(
scope: str = typer.Option(

35
src/cya/memory/__init__.py
View File

@@ -99,6 +99,7 @@ def remember_preference(
profile: str | None = None,
ttl: str | None = None,
kind: str = KIND_PREFERENCE,
provenance: dict[str, Any] | None = None,
) -> None:
"""Remember a user preference, workflow pattern, retrospection outcome, or goal.
@@ -118,6 +119,8 @@ def remember_preference(
"profile": profile,
"kind": kind,
}
if provenance:
item["provenance"] = provenance
# avoid exact dups for same key in small stores
items = [i for i in items if i.get("key") != key]
items.append(item)
@@ -166,7 +169,13 @@ def recall_preferences(
normal.append(item)
items = boosted + normal
items = items[-limit:] # most recent first (after boosting)
# Profile 1: prefer reflections ahead of other kinds when requested
if kinds and KIND_REFLECTION in kinds:
reflections = [i for i in items if i.get("kind") == KIND_REFLECTION]
others = [i for i in items if i.get("kind") != KIND_REFLECTION]
items = (reflections + others)[:limit]
else:
items = items[-limit:] # most recent (after boosting)
return {
"items": items,
@@ -222,7 +231,13 @@ def export_memory(scope: str = "cwd", *, profile: str | None = None, kinds: list
k = item.get("kind", "unknown")
by_kind.setdefault(k, []).append(item)
return {
reflection_by_scope: dict[str, int] = {}
for item in items:
if item.get("kind") == KIND_REFLECTION:
s = str(item.get("scope", scope))
reflection_by_scope[s] = reflection_by_scope.get(s, 0) + 1
result = {
"status": "real (T02+0003 local json; activation + retrospection ready)",
"scope": scope,
"profile": profile,
@@ -234,6 +249,10 @@ def export_memory(scope: str = "cwd", *, profile: str | None = None, kinds: list
"note": "User-controlled. Replace with real phase-memory when available.",
"phases": ["ephemeral", "fluid", "stabilized", "rigid"],
}
if reflection_by_scope:
result["reflection_counts_by_scope"] = reflection_by_scope
result["reflection_count"] = sum(reflection_by_scope.values())
return result
except Exception as e:
_warn_not_connected(f"export_memory(scope={scope}, profile={profile}) err={e}")
return {
@@ -264,13 +283,21 @@ def remember_reflection(
scope: str = "cwd",
*,
profile: str | None = None,
provenance: dict[str, Any] | 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.
activation in future turns.
"""
remember_preference(key, value, scope=scope, profile=profile, kind=KIND_REFLECTION)
remember_preference(
key,
value,
scope=scope,
profile=profile,
kind=KIND_REFLECTION,
provenance=provenance,
)
__all__ = [

241
src/cya/memory/reflections.py Normal file
View File

@@ -0,0 +1,241 @@
"""Profile 1 reflection helpers — capture, compaction, and surfacing.
Pure, testable logic for verbal reflection management. Interactive CLI flows
in orchestrator.py call into these functions.
"""
from __future__ import annotations
import re
from datetime import datetime, timezone
from difflib import SequenceMatcher
from typing import Any
from cya.memory import (
KIND_REFLECTION,
_load,
_save,
export_memory,
remember_reflection,
)
REFLECTION_CAPTURE_PROMPTS: tuple[tuple[str, str], ...] = (
("went_well", "What went well that you want to remember? (or 'skip')"),
("remember", "What should cya remember for next time? (or 'skip')"),
("avoid", "What should cya avoid in this scope? (or 'skip')"),
)
_SKIP_VALUES = frozenset({"", "skip", "s", "n", "no"})
_MAX_SURFACING_LESSONS = 3
_MAX_LESSON_CHARS_EXPLAIN = 80
_MAX_LESSON_CHARS_RESPONSE = 60
_SIMILARITY_THRESHOLD = 0.85
def is_skip_answer(text: str) -> bool:
return not text or text.strip().lower() in _SKIP_VALUES
def collect_lessons_from_answers(answers: dict[str, str]) -> list[dict[str, str]]:
"""Build lesson records from guided prompt answers; skips empty/skip answers."""
lessons: list[dict[str, str]] = []
for prompt_key, _label in REFLECTION_CAPTURE_PROMPTS:
raw = answers.get(prompt_key, "")
if is_skip_answer(raw):
continue
text = raw.strip()
if text:
lessons.append({"prompt": prompt_key, "text": text})
return lessons
def preview_lessons(lessons: list[dict[str, str]]) -> str:
if not lessons:
return "(no lessons to save)"
lines = []
for i, lesson in enumerate(lessons, 1):
label = lesson.get("prompt", "lesson").replace("_", " ")
lines.append(f"{i}. [{label}] {lesson['text']}")
return "\n".join(lines)
def session_provenance(scope: str) -> dict[str, Any]:
return {
"session_date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
"scope": scope,
"source": "cya retrospect",
}
def save_reflection_lessons(
lessons: list[dict[str, str]],
scope: str,
*,
provenance: dict[str, Any] | None = None,
) -> int:
"""Persist confirmed lessons; returns count saved. No-op for empty input."""
if not lessons:
return 0
meta = provenance or session_provenance(scope)
saved = 0
for lesson in lessons:
key = f"reflection_{lesson['prompt']}_{meta['session_date']}"
remember_reflection(
key,
lesson["text"],
scope=scope,
provenance={**meta, "prompt": lesson["prompt"]},
)
saved += 1
return saved
def normalize_reflection_text(text: str) -> str:
collapsed = re.sub(r"\s+", " ", text.strip().lower())
return re.sub(r"[^\w\s]", "", collapsed)
def reflection_similarity(a: str, b: str) -> float:
na, nb = normalize_reflection_text(a), normalize_reflection_text(b)
if not na or not nb:
return 0.0
if na == nb:
return 1.0
return SequenceMatcher(None, na, nb).ratio()
def find_duplicate_reflection_groups(
scope: str,
*,
threshold: float = _SIMILARITY_THRESHOLD,
) -> list[list[dict[str, Any]]]:
"""Return groups of near-duplicate reflection items in the same scope."""
items = [i for i in _load(scope) if i.get("kind") == KIND_REFLECTION]
groups: list[list[dict[str, Any]]] = []
used: set[str] = set()
for i, item in enumerate(items):
key_i = item.get("key", str(i))
if key_i in used:
continue
group = [item]
val_i = str(item.get("value", ""))
for j, other in enumerate(items):
if j <= i:
continue
key_j = other.get("key", str(j))
if key_j in used:
continue
val_j = str(other.get("value", ""))
if reflection_similarity(val_i, val_j) >= threshold:
group.append(other)
used.add(key_j)
if len(group) > 1:
groups.append(group)
used.add(key_i)
return groups
def compact_reflections(
scope: str,
*,
keep_key: str,
remove_keys: list[str],
merged_value: str | None = None,
) -> dict[str, Any]:
"""Explicit opt-in compaction: update keeper and remove duplicates."""
items = _load(scope)
removed: list[str] = []
updated = False
for item in items:
if item.get("key") == keep_key and item.get("kind") == KIND_REFLECTION:
if merged_value is not None:
item["value"] = merged_value
updated = True
new_items = []
for item in items:
k = item.get("key")
if k in remove_keys and item.get("kind") == KIND_REFLECTION:
removed.append(k)
continue
new_items.append(item)
if removed or updated:
_save(scope, new_items)
return {
"kept": keep_key,
"removed": removed,
"updated": updated,
"remaining_reflections": sum(
1 for i in new_items if i.get("kind") == KIND_REFLECTION
),
}
def list_reflections(scope: str) -> list[dict[str, Any]]:
exported = export_memory(scope, kinds=[KIND_REFLECTION])
return exported.get("items", [])
def format_reflection_surfacing(
memory: dict[str, Any],
*,
for_explain: bool = False,
) -> str | None:
"""Budget-capped reflection summary for explain-context or normal responses."""
items = memory.get("items", []) if isinstance(memory, dict) else []
reflections = [i for i in items if i.get("kind") == KIND_REFLECTION]
if not reflections:
return None
max_chars = _MAX_LESSON_CHARS_EXPLAIN if for_explain else _MAX_LESSON_CHARS_RESPONSE
snippets: list[str] = []
for item in reflections[:_MAX_SURFACING_LESSONS]:
text = str(item.get("value", "")).strip()
if len(text) > max_chars:
text = text[: max_chars - 3] + "..."
prov = item.get("provenance") or {}
date = prov.get("session_date", "")
prefix = f"({date}) " if date and for_explain else ""
snippets.append(f"{prefix}{text}")
extra = len(reflections) - _MAX_SURFACING_LESSONS
suffix = f" (+{extra} more)" if extra > 0 else ""
count = len(reflections)
noun = "reflection" if count == 1 else "reflections"
return f"{count} verbal {noun} influenced this{suffix}: " + "; ".join(snippets)
def reflection_export_stats(scope: str) -> dict[str, Any]:
"""Observability: reflection counts and scope breakdown for export."""
exported = export_memory(scope, kinds=[KIND_REFLECTION])
items = exported.get("items", [])
by_scope: dict[str, int] = {}
for item in items:
s = item.get("scope", scope)
by_scope[s] = by_scope.get(s, 0) + 1
return {
"reflection_count": len(items),
"reflection_counts_by_scope": by_scope,
"scope": scope,
}
__all__ = [
"REFLECTION_CAPTURE_PROMPTS",
"collect_lessons_from_answers",
"preview_lessons",
"session_provenance",
"save_reflection_lessons",
"normalize_reflection_text",
"reflection_similarity",
"find_duplicate_reflection_groups",
"compact_reflections",
"list_reflections",
"format_reflection_surfacing",
"reflection_export_stats",
"is_skip_answer",
]

139
src/cya/orchestrator.py
View File

@@ -25,6 +25,7 @@ from __future__ import annotations
from pathlib import Path
import typer
from rich.console import Console
from rich.panel import Panel
@@ -32,11 +33,20 @@ from cya.context.collector import collect, render_explanation
from cya.memory import (
recall_preferences,
remember_retrospection_outcome,
remember_reflection,
KIND_RETROSPECTION,
KIND_INTERACTION_GOAL,
KIND_REFLECTION,
)
from cya.memory.reflections import (
REFLECTION_CAPTURE_PROMPTS,
collect_lessons_from_answers,
compact_reflections,
find_duplicate_reflection_groups,
format_reflection_surfacing,
preview_lessons,
save_reflection_lessons,
session_provenance,
)
from cya.safety.risk import classify, get_user_confirmation
from cya.llm.adapter import AssistanceRequest, FakeLLMAdapter
@@ -98,15 +108,21 @@ def handle_request(
if explain_context and memory.get("items"):
try:
prov = memory.get("provenance", [{}])[0]
# Show a couple of activated items for transparency (T03 0003)
sample = ", ".join(i.get("key", "?") for i in memory.get("items", [])[:3])
act_note = ""
if prov.get("activation_context"):
act_note = f" | ctx: {prov['activation_context']}"
body = (
f"Phase: {memory.get('phase')} | {len(memory.get('items', []))} items | "
f"{prov.get('source', 'local')}{act_note}\n"
f"Sample activated: {sample}"
)
refl_line = format_reflection_surfacing(memory, for_explain=True)
if refl_line:
body += f"\n\n[cyan]Reflections:[/cyan] {refl_line}"
console.print(
Panel(
f"Phase: {memory.get('phase')} | {len(memory.get('items', []))} items | {prov.get('source', 'local')}{act_note}\n"
f"Sample activated: {sample}",
body,
title="Memory Activated (T03)",
border_style="blue",
padding=(0, 1),
@@ -156,11 +172,9 @@ def handle_request(
mem_line = ""
if memory.get("items"):
mem_line = f"\n[dim]Memory activated: {len(memory.get('items', []))} items (phase {memory.get('phase')})[/dim]"
# Minimal Profile 1 surface (T05 spike)
reflections = [i for i in memory.get("items", []) if i.get("kind") == KIND_REFLECTION]
if reflections:
refl_text = "; ".join(str(i.get("value", ""))[:60] for i in reflections[:2])
mem_line += f"\n[cyan]Verbal reflections activated: {len(reflections)} — {refl_text}[/cyan]"
refl_line = format_reflection_surfacing(memory, for_explain=False)
if refl_line:
mem_line += f"\n[cyan]{refl_line}[/cyan]"
console.print(
Panel(
@@ -263,29 +277,15 @@ def run_retrospection(scope: str = ".", limit: int = 8) -> None:
)
console.print("[green]Recorded as safety preference.[/green]")
# Minimal Profile 1 spike (T05): optional verbal reflection / lesson capture
capture_lesson = typer.prompt(
"Capture any verbal lessons or reflections from this session? (y/n or short text)",
default="n",
show_default=False,
)
if capture_lesson and capture_lesson.lower() not in ("n", "no", "skip", "s", ""):
lesson_text = capture_lesson if len(capture_lesson) > 3 else typer.prompt(
"What is the key lesson? (1-2 sentences)",
default="",
show_default=False,
)
if lesson_text:
remember_reflection(
"verbal_lesson", lesson_text, scope=scope
)
console.print("[green]Recorded as verbal reflection (Profile 1).[/green]")
_capture_reflection_lessons(scope)
_offer_reflection_compaction(scope)
console.print(
Panel(
"Thank you. Your reflections have been stored as retrospection memory.\n"
"They will be preferentially activated in future sessions in this scope.\n\n"
"You can review them anytime with:\n"
f" [bold]cya memory reflections --scope {scope}[/bold]\n"
f" [bold]cya --explain-context \"...\"[/bold] (in this directory)\n"
f" or inspect the JSON files in [cyan]~/.config/cya/memory/[/cyan]",
title="Retrospection Complete",
@@ -295,4 +295,91 @@ def run_retrospection(scope: str = ".", limit: int = 8) -> None:
)
def _capture_reflection_lessons(scope: str) -> None:
"""Profile 1: guided verbal lesson capture with preview and confirmation."""
want = typer.prompt(
"Capture 1–3 verbal lessons from this session? (y/n)",
default="n",
show_default=False,
)
if not want or want.strip().lower() in ("n", "no", "skip", "s", ""):
return
answers: dict[str, str] = {}
for prompt_key, label in REFLECTION_CAPTURE_PROMPTS:
answers[prompt_key] = typer.prompt(label, default="", show_default=False)
lessons = collect_lessons_from_answers(answers)
if not lessons:
console.print("[yellow]No lessons captured — nothing stored.[/yellow]")
return
console.print(
Panel(
preview_lessons(lessons),
title="Preview — verbal lessons to save",
border_style="cyan",
)
)
confirm = typer.prompt("Save these lessons? (y/n)", default="n", show_default=False)
if confirm.strip().lower() not in ("y", "yes"):
console.print("[yellow]Lessons discarded — nothing stored.[/yellow]")
return
count = save_reflection_lessons(
lessons,
scope,
provenance=session_provenance(scope),
)
console.print(f"[green]Saved {count} verbal reflection(s) (Profile 1).[/green]")
def _offer_reflection_compaction(scope: str) -> None:
"""Offer explicit merge of near-duplicate reflections in this scope."""
groups = find_duplicate_reflection_groups(scope)
if not groups:
return
console.print(
Panel(
f"Found {len(groups)} group(s) of similar reflections in scope [green]{scope}[/green].\n"
"Compaction is opt-in — nothing is deleted without your confirmation.",
title="Reflection compaction available",
border_style="yellow",
)
)
do_compact = typer.prompt("Review and compact duplicates? (y/n)", default="n", show_default=False)
if do_compact.strip().lower() not in ("y", "yes"):
return
for group in groups:
console.print("\n[bold]Similar reflections:[/bold]")
for item in group:
console.print(f" • {item.get('key')}: {item.get('value')}")
keep = typer.prompt(
"Key to keep (or 'skip' this group)",
default=group[0].get("key", ""),
show_default=True,
)
if not keep or keep.strip().lower() in ("skip", "s"):
continue
remove_keys = [i.get("key") for i in group if i.get("key") != keep]
merge = typer.prompt(
"Merged text (Enter to keep existing value of kept key)",
default="",
show_default=False,
)
result = compact_reflections(
scope,
keep_key=keep,
remove_keys=[k for k in remove_keys if k],
merged_value=merge.strip() or None,
)
console.print(
f"[green]Compacted: kept {result['kept']}, removed {len(result['removed'])}[/green]"
)
__all__ = ["handle_request", "run_retrospection"]