feat(memory): add Claude-style memory runtime (#267)

* Add Claude-style memory runtime

* fix(memory): trim relevance EOF whitespace

src/openharness/commands/registry.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import importlib.metadata
 import json
+import os
 import re
 import shutil
 import subprocess
@@ -38,8 +39,28 @@ from openharness.memory import (
     list_memory_files,
     migrate_memory,
     remove_memory_entry,
+    scan_memory_files,
+)
+from openharness.memory.agent import (
+    ensure_agent_memory_vault,
+    get_agent_memory_entrypoint,
+    initialize_agent_memory_from_snapshot,
+)
+from openharness.memory.schema import (
+    DEFAULT_MEMORY_SCOPE,
+    DEFAULT_MEMORY_TYPE,
+    MEMORY_TYPES,
+    is_disabled_metadata,
+    is_memory_expired,
+    parse_memory_scope,
+    parse_memory_type,
+    split_memory_file,
+)
+from openharness.memory.team import (
+    check_team_memory_secrets,
+    ensure_team_memory_vault,
+    get_team_memory_dir,
 )
-from openharness.memory.schema import is_disabled_metadata, is_memory_expired, split_memory_file
 from openharness.output_styles import load_output_styles
 from openharness.permissions import PermissionChecker, PermissionMode
 from openharness.plugins import load_plugins
@@ -60,6 +81,12 @@ from openharness.services.autodream import (
     restore_memory_backup,
     start_dream_now,
 )
+from openharness.services.memory_extract import extract_memories_from_turn
+from openharness.services.session_memory import (
+    get_session_memory_content,
+    get_session_memory_path,
+    update_session_memory_file,
+)
 from openharness.services.session_backend import DEFAULT_SESSION_BACKEND, SessionBackend
 from openharness.skills import load_skill_registry
 from openharness.skills.types import SkillDefinition
@@ -654,7 +681,10 @@ def create_default_command_registry(
                 message=(
                     f"Memory store: {backend.label}\n"
                     f"Memory directory: {backend.get_memory_dir()}\n"
-                    f"Entrypoint: {backend.get_entrypoint()}"
+                    f"Entrypoint: {backend.get_entrypoint()}\n"
+                    "Commands: list, show NAME, add TITLE :: CONTENT, remove NAME, "
+                    "edit [NAME], validate, extract, session, team, agent, "
+                    "migrate --dry-run, migrate --apply"
                 )
             )
         action = tokens[0]
@@ -710,19 +740,55 @@ def create_default_command_registry(
                 return CommandResult(message=f"Memory entry not found: {rest}")
             return CommandResult(message=content)
         if action == "add" and rest:
-            title, separator, content = rest.partition("::")
+            memory_type, scope, cleaned_rest = _parse_memory_add_flags(rest)
+            title, separator, content = cleaned_rest.partition("::")
             if not separator or not title.strip() or not content.strip():
-                return CommandResult(message="Usage: /memory add TITLE :: CONTENT")
-            path = backend.add_entry(title.strip(), content.strip())
+                return CommandResult(message="Usage: /memory add [--type TYPE] [--scope SCOPE] TITLE :: CONTENT")
+            if context.memory_backend is None:
+                path = add_memory_entry(
+                    context.cwd,
+                    title.strip(),
+                    content.strip(),
+                    memory_type=memory_type,
+                    scope=scope,
+                )
+            else:
+                path = backend.add_entry(title.strip(), content.strip())
             return CommandResult(message=f"Added memory entry {path.name}")
         if action == "remove" and rest:
             if backend.remove_entry(rest.strip()):
                 return CommandResult(message=f"Removed memory entry {rest.strip()}")
             return CommandResult(message=f"Memory entry not found: {rest.strip()}")
+        if action == "edit":
+            return _handle_memory_edit_command(rest, context, backend)
+        if action == "validate":
+            return _handle_memory_validate_command(context)
+        if action == "extract":
+            if context.memory_backend is not None:
+                return CommandResult(message="Memory extraction is only supported for OpenHarness project memory.")
+            result = await extract_memories_from_turn(
+                cwd=context.cwd,
+                api_client=context.engine.api_client,
+                model=context.engine.model,
+                messages=context.engine.messages,
+                max_records=load_settings().memory.auto_extract_max_records,
+            )
+            if result.skipped:
+                return CommandResult(message=f"Memory extraction skipped: {result.reason}")
+            return CommandResult(
+                message="Memory extraction wrote:\n" + "\n".join(f"- {path}" for path in result.written_paths)
+            )
+        if action == "session":
+            return _handle_memory_session_command(rest, context)
+        if action == "team":
+            return _handle_memory_team_command(rest, context)
+        if action == "agent":
+            return _handle_memory_agent_command(rest, context)
         return CommandResult(
             message=(
                 "Usage: /memory "
-                "[list|show NAME|add TITLE :: CONTENT|remove NAME|"
+                "[list|show NAME|add TITLE :: CONTENT|remove NAME|edit [NAME]|"
+                "validate|extract|session|team|agent|"
                 "migrate --dry-run|migrate --apply]"
             )
         )
@@ -2478,6 +2544,158 @@ def create_default_command_registry(
     return registry
 
 
+def _handle_memory_edit_command(
+    args: str,
+    context: CommandContext,
+    backend: MemoryCommandBackend,
+) -> CommandResult:
+    memory_dir = backend.get_memory_dir()
+    target = backend.get_entrypoint()
+    if args.strip():
+        path, invalid = _resolve_memory_entry_path(memory_dir, args.strip())
+        if invalid:
+            return CommandResult(message="Memory entry path must stay within the configured memory directory.")
+        if path is None:
+            return CommandResult(message=f"Memory entry not found: {args.strip()}")
+        target = path
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.touch(exist_ok=True)
+    editor = os.environ.get("VISUAL") or os.environ.get("EDITOR")
+    if not editor:
+        return CommandResult(message=f"Memory file ready: {target}\nSet $VISUAL or $EDITOR to open it from /memory edit.")
+    result = subprocess.run([editor, str(target)], cwd=context.cwd, check=False)
+    if result.returncode != 0:
+        return CommandResult(message=f"Editor exited with status {result.returncode}: {editor}")
+    return CommandResult(message=f"Edited memory file: {target}")
+
+
+def _parse_memory_add_flags(args: str):
+    """Parse optional ``/memory add`` type/scope flags."""
+
+    memory_type = DEFAULT_MEMORY_TYPE
+    scope = DEFAULT_MEMORY_SCOPE
+    rest = args.strip()
+    changed = True
+    while changed:
+        changed = False
+        if rest.startswith("--type "):
+            _, _, tail = rest.partition(" ")
+            raw, _, rest = tail.partition(" ")
+            parsed = parse_memory_type(raw, default=DEFAULT_MEMORY_TYPE)
+            if parsed is not None:
+                memory_type = parsed
+            changed = True
+        if rest.startswith("--scope "):
+            _, _, tail = rest.partition(" ")
+            raw, _, rest = tail.partition(" ")
+            parsed_scope = parse_memory_scope(raw, default=DEFAULT_MEMORY_SCOPE)
+            if parsed_scope is not None:
+                scope = parsed_scope
+            changed = True
+    return memory_type, scope, rest
+
+
+def _handle_memory_validate_command(context: CommandContext) -> CommandResult:
+    memory_dir = get_project_memory_dir(context.cwd)
+    headers = scan_memory_files(context.cwd, max_files=500)
+    issues: list[str] = []
+    for header in headers:
+        raw_type = header.frontmatter.get("type") or header.frontmatter.get("memory_type")
+        if parse_memory_type(raw_type) is None:
+            issues.append(
+                f"- {header.relative_path}: invalid or missing type {raw_type!r}; expected {', '.join(MEMORY_TYPES)}"
+            )
+        if "team" in Path(header.relative_path).parts:
+            try:
+                text = header.path.read_text(encoding="utf-8", errors="replace")
+            except OSError:
+                text = ""
+            secret_error = check_team_memory_secrets(text)
+            if secret_error:
+                issues.append(f"- {header.relative_path}: {secret_error}")
+    if not issues:
+        return CommandResult(
+            message=(
+                "Memory validation passed.\n"
+                f"- files: {len(headers)}\n"
+                f"- memory_dir: {memory_dir}"
+            )
+        )
+    return CommandResult(message="Memory validation issues:\n" + "\n".join(issues))
+
+
+def _handle_memory_session_command(args: str, context: CommandContext) -> CommandResult:
+    action = args.split(maxsplit=1)[0] if args.strip() else "status"
+    path = get_session_memory_path(context.cwd, context.session_id or "default")
+    if action == "update":
+        path = update_session_memory_file(
+            context.cwd,
+            context.engine.messages,
+            tool_metadata=context.engine.tool_metadata,
+            session_id=context.session_id or "default",
+        )
+        return CommandResult(message=f"Updated session memory: {path}")
+    if action == "show":
+        content = get_session_memory_content(path)
+        return CommandResult(message=content or f"No session memory at {path}")
+    return CommandResult(
+        message=(
+            "Session memory:\n"
+            f"- path: {path}\n"
+            f"- exists: {path.exists()}\n"
+            "Commands: /memory session [status|show|update]"
+        )
+    )
+
+
+def _handle_memory_team_command(args: str, context: CommandContext) -> CommandResult:
+    action = args.split(maxsplit=1)[0] if args.strip() else "status"
+    team_dir = ensure_team_memory_vault(context.cwd)
+    if action == "list":
+        files = sorted(path for path in team_dir.rglob("*.md") if path.name != "MEMORY.md")
+        return CommandResult(message="\n".join(str(path.relative_to(team_dir)) for path in files) or "No team memory files.")
+    if action == "validate":
+        issues: list[str] = []
+        for path in sorted(team_dir.rglob("*.md")):
+            if path.name == "MEMORY.md":
+                continue
+            text = path.read_text(encoding="utf-8", errors="replace")
+            secret_error = check_team_memory_secrets(text)
+            if secret_error:
+                issues.append(f"- {path.relative_to(team_dir)}: {secret_error}")
+        return CommandResult(message="Team memory validation passed." if not issues else "\n".join(issues))
+    return CommandResult(
+        message=(
+            "Team memory:\n"
+            f"- directory: {get_team_memory_dir(context.cwd)}\n"
+            f"- exists: {team_dir.exists()}\n"
+            "Commands: /memory team [status|list|validate]"
+        )
+    )
+
+
+def _handle_memory_agent_command(args: str, context: CommandContext) -> CommandResult:
+    parts = args.split()
+    action = parts[0] if parts else "status"
+    agent_type = parts[1] if len(parts) > 1 else "default"
+    scope = parts[2] if len(parts) > 2 else "project"
+    if scope not in {"user", "project", "local"}:
+        return CommandResult(message="Agent memory scope must be one of: user, project, local")
+    if action == "snapshot":
+        target = initialize_agent_memory_from_snapshot(context.cwd, agent_type, scope)  # type: ignore[arg-type]
+        return CommandResult(message=f"Initialized agent memory from snapshot: {target}" if target else "No snapshot found.")
+    vault = ensure_agent_memory_vault(context.cwd, agent_type, scope)  # type: ignore[arg-type]
+    return CommandResult(
+        message=(
+            "Agent memory:\n"
+            f"- agent_type: {agent_type}\n"
+            f"- scope: {scope}\n"
+            f"- directory: {vault}\n"
+            f"- entrypoint: {get_agent_memory_entrypoint(context.cwd, agent_type, scope)}"
+        )
+    )
+
+
 def _resolve_memory_entry_path(memory_dir: Path, candidate: str) -> tuple[Path | None, bool]:
     """Resolve a memory entry path while enforcing containment under ``memory_dir``."""
 

src/openharness/config/settings.py

@@ -63,8 +63,12 @@ class MemorySettings(BaseModel):
     enabled: bool = True
     max_files: int = 5
     max_entrypoint_lines: int = 200
+    max_entrypoint_bytes: int = 25_000
     context_window_tokens: int | None = None
     auto_compact_threshold_tokens: int | None = None
+    auto_extract_enabled: bool = False
+    auto_extract_max_records: int = 3
+    session_memory_enabled: bool = True
     auto_dream_enabled: bool = False
     auto_dream_min_hours: float = 24.0
     auto_dream_min_sessions: int = 5

src/openharness/engine/query_engine.py

@@ -152,6 +152,63 @@ class QueryEngine:
             **kwargs,
         )
 
+    def _prepare_session_memory(self) -> None:
+        """Expose file-backed session memory to compaction when enabled."""
+
+        if self._settings is None or not self._settings.memory.session_memory_enabled:
+            return
+        if not self._settings.memory.enabled:
+            return
+        from openharness.services.session_memory import prepare_session_memory_metadata
+
+        prepare_session_memory_metadata(
+            self._cwd,
+            self._tool_metadata,
+            session_id=str(self._tool_metadata.get("session_id") or "default"),
+        )
+
+    async def _update_session_memory(self) -> None:
+        """Persist a session checkpoint after a user turn."""
+
+        if self._settings is None or not self._settings.memory.session_memory_enabled:
+            return
+        if not self._settings.memory.enabled:
+            return
+        from openharness.services.session_memory import update_session_memory_file
+
+        update_session_memory_file(
+            self._cwd,
+            list(self._messages),
+            tool_metadata=self._tool_metadata,
+            session_id=str(self._tool_metadata.get("session_id") or "default"),
+        )
+
+    async def _extract_durable_memories(self) -> None:
+        """Run the optional durable memory extraction pass."""
+
+        if self._settings is None or not self._settings.memory.auto_extract_enabled:
+            return
+        if not self._settings.memory.enabled:
+            return
+        from openharness.services.memory_extract import extract_memories_from_turn
+
+        try:
+            result = await extract_memories_from_turn(
+                cwd=self._cwd,
+                api_client=self._api_client,
+                model=self._model,
+                messages=list(self._messages),
+                max_records=self._settings.memory.auto_extract_max_records,
+            )
+        except Exception as exc:
+            self._tool_metadata["memory_extract_last_error"] = str(exc)
+            return
+        self._tool_metadata["memory_extract_last"] = {
+            "skipped": result.skipped,
+            "reason": result.reason,
+            "written_paths": [str(path) for path in result.written_paths],
+        }
+
     def has_pending_continuation(self) -> bool:
         """Return True when the conversation ends with tool results awaiting a follow-up model turn."""
         if not self._messages:
@@ -176,6 +233,7 @@ class QueryEngine:
         )
         if user_message.text.strip() and not self._tool_metadata.pop("_suppress_next_user_goal", False):
             remember_user_goal(self._tool_metadata, user_message.text)
+        self._prepare_session_memory()
         self._messages = sanitize_conversation_messages(self._messages)
         self._messages.append(user_message)
         if self._hook_executor is not None:
@@ -215,10 +273,13 @@ class QueryEngine:
                     self._cost_tracker.add(usage)
                 yield event
         finally:
+            await self._update_session_memory()
+            await self._extract_durable_memories()
             self._schedule_auto_dream()
 
     async def continue_pending(self, *, max_turns: int | None = None) -> AsyncIterator[StreamEvent]:
         """Continue an interrupted tool loop without appending a new user message."""
+        self._prepare_session_memory()
         self._messages = sanitize_conversation_messages(self._messages)
         context = QueryContext(
             api_client=self._api_client,
@@ -241,3 +302,5 @@ class QueryEngine:
             if usage is not None:
                 self._cost_tracker.add(usage)
             yield event
+        await self._update_session_memory()
+        await self._extract_durable_memories()

src/openharness/memory/__init__.py

@@ -7,10 +7,12 @@ from openharness.memory.paths import get_memory_entrypoint, get_project_memory_d
 from openharness.memory.scan import scan_memory_files
 from openharness.memory.search import find_relevant_memories
 from openharness.memory.usage import mark_memory_used
+from openharness.memory.relevance import format_relevant_memories, select_relevant_memories
 
 __all__ = [
     "add_memory_entry",
     "find_relevant_memories",
+    "format_relevant_memories",
     "get_memory_entrypoint",
     "get_project_memory_dir",
     "list_memory_files",
@@ -19,4 +21,5 @@ __all__ = [
     "migrate_memory",
     "remove_memory_entry",
     "scan_memory_files",
+    "select_relevant_memories",
 ]

src/openharness/memory/agent.py

@@ -0,0 +1,91 @@
+"""Agent-scoped memory paths and snapshots."""
+
+from __future__ import annotations
+
+import re
+import shutil
+from pathlib import Path
+from typing import Literal
+
+from openharness.config.paths import get_data_dir
+from openharness.memory.paths import get_project_memory_dir
+
+AgentMemoryScope = Literal["user", "project", "local"]
+
+MEMORY_INDEX = "MEMORY.md"
+SNAPSHOT_DIR_NAME = "agent-memory-snapshots"
+
+
+def sanitize_agent_type(agent_type: str) -> str:
+    """Return a path-safe agent type."""
+
+    return re.sub(r"[^a-zA-Z0-9_.-]+", "_", agent_type.strip()).strip("._") or "default"
+
+
+def get_agent_memory_dir(cwd: str | Path, agent_type: str, scope: AgentMemoryScope) -> Path:
+    """Return an agent memory vault for the requested scope."""
+
+    safe = sanitize_agent_type(agent_type)
+    if scope == "project":
+        return get_project_memory_dir(cwd) / "agent" / safe
+    if scope == "local":
+        return Path(cwd).resolve() / ".openharness" / "agent-memory-local" / safe
+    return get_data_dir() / "agent-memory" / safe
+
+
+def ensure_agent_memory_vault(cwd: str | Path, agent_type: str, scope: AgentMemoryScope) -> Path:
+    """Create and return an agent-scoped memory vault."""
+
+    memory_dir = get_agent_memory_dir(cwd, agent_type, scope)
+    memory_dir.mkdir(parents=True, exist_ok=True)
+    entrypoint = memory_dir / MEMORY_INDEX
+    if not entrypoint.exists():
+        entrypoint.write_text("# Memory Index\n", encoding="utf-8")
+    return memory_dir
+
+
+def get_agent_memory_entrypoint(cwd: str | Path, agent_type: str, scope: AgentMemoryScope) -> Path:
+    """Return an agent memory ``MEMORY.md`` path."""
+
+    return ensure_agent_memory_vault(cwd, agent_type, scope) / MEMORY_INDEX
+
+
+def get_agent_snapshot_dir(cwd: str | Path, agent_type: str) -> Path:
+    """Return the project snapshot directory for an agent type."""
+
+    return Path(cwd).resolve() / ".openharness" / SNAPSHOT_DIR_NAME / sanitize_agent_type(agent_type)
+
+
+def initialize_agent_memory_from_snapshot(
+    cwd: str | Path,
+    agent_type: str,
+    scope: AgentMemoryScope,
+    *,
+    replace: bool = False,
+) -> Path | None:
+    """Initialize local agent memory from a project snapshot if present."""
+
+    snapshot_dir = get_agent_snapshot_dir(cwd, agent_type)
+    if not snapshot_dir.exists():
+        return None
+    target = ensure_agent_memory_vault(cwd, agent_type, scope)
+    if replace and target.exists():
+        shutil.rmtree(target)
+        target.mkdir(parents=True, exist_ok=True)
+    for src in snapshot_dir.rglob("*.md"):
+        rel = src.relative_to(snapshot_dir)
+        dest = target / rel
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        if replace or not dest.exists() or _is_default_agent_index(dest):
+            shutil.copy2(src, dest)
+    return target
+
+
+def _is_default_agent_index(path: Path) -> bool:
+    if path.name != MEMORY_INDEX or not path.exists():
+        return False
+    try:
+        text = path.read_text(encoding="utf-8", errors="replace")
+    except OSError:
+        return False
+    return text.startswith("# Memory Index\n")

src/openharness/memory/manager.py

@@ -8,6 +8,10 @@ from re import sub
 from openharness.memory.paths import get_memory_entrypoint, get_project_memory_dir
 from openharness.memory.scan import scan_memory_files
 from openharness.memory.schema import (
+    DEFAULT_MEMORY_SCOPE,
+    DEFAULT_MEMORY_TYPE,
+    MemoryScope,
+    MemoryType,
     SCHEMA_VERSION,
     compute_memory_signature,
     first_content_line,
@@ -32,12 +36,28 @@ def list_memory_files(cwd: str | Path) -> list[Path]:
     return sorted(header.path for header in scan_memory_files(cwd, max_files=None))
 
 
-def add_memory_entry(cwd: str | Path, title: str, content: str) -> Path:
+def add_memory_entry(
+    cwd: str | Path,
+    title: str,
+    content: str,
+    *,
+    memory_type: MemoryType = DEFAULT_MEMORY_TYPE,
+    scope: MemoryScope = DEFAULT_MEMORY_SCOPE,
+    description: str = "",
+    tags: tuple[str, ...] = (),
+) -> Path:
     """Create or refresh a memory file and append it to MEMORY.md."""
-    memory_dir = get_project_memory_dir(cwd)
+    if scope == "team":
+        from openharness.memory.team import check_team_memory_secrets, ensure_team_memory_vault
+
+        secret_error = check_team_memory_secrets(content)
+        if secret_error:
+            raise ValueError(secret_error)
+        memory_dir = ensure_team_memory_vault(cwd)
+    else:
+        memory_dir = get_project_memory_dir(cwd)
     slug = sub(r"[^a-zA-Z0-9]+", "_", title.strip().lower()).strip("_") or "memory"
-    with exclusive_file_lock(_memory_lock_path(cwd)):
-        memory_type = "project"
+    with exclusive_file_lock(memory_dir / ".memory.lock"):
         category = "knowledge"
         body = content.strip() + "\n"
         signature = compute_memory_signature(body, memory_type, category)
@@ -46,6 +66,7 @@ def add_memory_entry(cwd: str | Path, title: str, content: str) -> Path:
             max_files=None,
             include_disabled=True,
             include_expired=True,
+            memory_dir=memory_dir,
         )
         duplicate = next(
             (header for header in existing if _effective_signature(header.path, header.signature) == signature),
@@ -75,8 +96,9 @@ def add_memory_entry(cwd: str | Path, title: str, content: str) -> Path:
                 "schema_version": SCHEMA_VERSION,
                 "id": memory_id,
                 "name": title.strip(),
-                "description": first_content_line(body) or title.strip(),
+                "description": description.strip() or first_content_line(body) or title.strip(),
                 "type": str(metadata.get("type") or memory_type),
+                "scope": str(metadata.get("scope") or scope),
                 "category": str(metadata.get("category") or category),
                 "importance": max(coerce_int(metadata.get("importance"), default=0), 1),
                 "source": "manual",
@@ -86,11 +108,12 @@ def add_memory_entry(cwd: str | Path, title: str, content: str) -> Path:
                 "ttl_days": metadata.get("ttl_days"),
                 "disabled": False,
                 "supersedes": metadata.get("supersedes") or [],
+                "tags": list(dict.fromkeys(str(tag).strip() for tag in tags if str(tag).strip())),
             }
         )
         atomic_write_text(path, render_memory_file(metadata, body))
 
-        entrypoint = get_memory_entrypoint(cwd)
+        entrypoint = memory_dir / "MEMORY.md"
         index_text = entrypoint.read_text(encoding="utf-8") if entrypoint.exists() else "# Memory Index\n"
         if path.name not in index_text:
             index_text = index_text.rstrip() + f"\n- [{title}]({path.name})\n"

src/openharness/memory/memdir.py

@@ -5,9 +5,19 @@ from __future__ import annotations
 from pathlib import Path
 
 from openharness.memory.paths import get_memory_entrypoint, get_project_memory_dir
+from openharness.memory.schema import (
+    MAX_ENTRYPOINT_BYTES,
+    MEMORY_POLICY_LINES,
+    truncate_entrypoint_content,
+)
 
 
-def load_memory_prompt(cwd: str | Path, *, max_entrypoint_lines: int = 200) -> str | None:
+def load_memory_prompt(
+    cwd: str | Path,
+    *,
+    max_entrypoint_lines: int = 200,
+    max_entrypoint_bytes: int = MAX_ENTRYPOINT_BYTES,
+) -> str | None:
     """Return the memory prompt section for the current project."""
     memory_dir = get_project_memory_dir(cwd)
     entrypoint = get_memory_entrypoint(cwd)
@@ -16,12 +26,20 @@ def load_memory_prompt(cwd: str | Path, *, max_entrypoint_lines: int = 200) -> s
         f"- Persistent memory directory: {memory_dir}",
         "- Use this directory to store durable project and repository context that should survive future sessions.",
         "- Prefer concise topic files plus an index entry in MEMORY.md.",
+        "",
+        *MEMORY_POLICY_LINES,
     ]
 
     if entrypoint.exists():
-        content_lines = entrypoint.read_text(encoding="utf-8").splitlines()[:max_entrypoint_lines]
-        if content_lines:
-            lines.extend(["", "## MEMORY.md", "```md", *content_lines, "```"])
+        raw = entrypoint.read_text(encoding="utf-8", errors="replace")
+        view = truncate_entrypoint_content(
+            raw,
+            max_lines=max_entrypoint_lines,
+            max_bytes=max_entrypoint_bytes,
+        )
+        content = view.content.strip()
+        if content:
+            lines.extend(["", "## MEMORY.md", "```md", content, "```"])
     else:
         lines.extend(
             [

src/openharness/memory/relevance.py

@@ -0,0 +1,145 @@
+"""Relevant memory selection and formatting."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Iterable
+
+from openharness.memory.scan import scan_memory_files
+from openharness.memory.schema import memory_age_label, memory_freshness_text
+from openharness.memory.search import find_relevant_memories
+from openharness.memory.types import MemoryHeader
+
+
+@dataclass(frozen=True)
+class RelevantMemory:
+    """A memory selected for prompt injection."""
+
+    header: MemoryHeader
+    freshness: str = ""
+
+
+MemorySelector = Callable[[str, list[MemoryHeader]], list[str]]
+
+
+def build_memory_manifest(headers: Iterable[MemoryHeader]) -> str:
+    """Render a compact manifest for selector prompts and diagnostics."""
+
+    lines: list[str] = []
+    for header in headers:
+        prefix = f"[{header.memory_type or 'memory'}]"
+        bits = [
+            prefix,
+            header.relative_path or header.path.name,
+            f"({memory_age_label(header.modified_at)})",
+        ]
+        if header.description:
+            bits.append(f"- {header.description}")
+        lines.append(" ".join(bits))
+    return "\n".join(lines)
+
+
+def select_relevant_memories(
+    query: str,
+    cwd: str | Path,
+    *,
+    max_results: int = 5,
+    already_surfaced: set[str] | None = None,
+    selector: MemorySelector | None = None,
+) -> list[RelevantMemory]:
+    """Return relevant memories with duplicate and freshness handling.
+
+    ``selector`` is an optional side-query style reranker. It receives the query
+    and a heuristic shortlist, and returns relative paths in desired order.
+    """
+
+    surfaced = already_surfaced or set()
+    heuristic = [
+        header
+        for header in find_relevant_memories(query, cwd, max_results=max(10, max_results * 3))
+        if (header.relative_path or str(header.path)) not in surfaced
+    ]
+    selected = _apply_selector(query, heuristic, selector=selector, max_results=max_results)
+    result: list[RelevantMemory] = []
+    for header in selected[:max_results]:
+        result.append(RelevantMemory(header=header, freshness=memory_freshness_text(header.modified_at)))
+    return result
+
+
+def select_manifest_memories(
+    query: str,
+    cwd: str | Path,
+    *,
+    max_results: int = 5,
+    selector: MemorySelector | None = None,
+) -> list[RelevantMemory]:
+    """Select from the full manifest instead of heuristic matches only."""
+
+    headers = scan_memory_files(cwd, max_files=200)
+    selected = _apply_selector(query, headers, selector=selector, max_results=max_results)
+    return [
+        RelevantMemory(header=header, freshness=memory_freshness_text(header.modified_at))
+        for header in selected[:max_results]
+    ]
+
+
+def format_relevant_memories(memories: Iterable[RelevantMemory], *, max_chars: int = 8000) -> str:
+    """Render selected memories for prompt context."""
+
+    lines = ["# Relevant Memories"]
+    for item in memories:
+        header = item.header
+        content = header.path.read_text(encoding="utf-8", errors="replace").strip()
+        if item.freshness:
+            lines.extend(["", f"## {header.relative_path or header.path.name}", f"> {item.freshness}"])
+        else:
+            lines.extend(["", f"## {header.relative_path or header.path.name}"])
+        lines.extend(["```md", content[:max_chars], "```"])
+    return "\n".join(lines)
+
+
+def json_selector_from_text(text: str) -> list[str]:
+    """Parse a selector response as either JSON list or newline paths."""
+
+    stripped = text.strip()
+    if not stripped:
+        return []
+    try:
+        payload = json.loads(stripped)
+    except json.JSONDecodeError:
+        return [line.strip("- ").strip() for line in stripped.splitlines() if line.strip()]
+    if isinstance(payload, list):
+        return [str(item).strip() for item in payload if str(item).strip()]
+    if isinstance(payload, dict) and isinstance(payload.get("paths"), list):
+        return [str(item).strip() for item in payload["paths"] if str(item).strip()]
+    return []
+
+
+def _apply_selector(
+    query: str,
+    headers: list[MemoryHeader],
+    *,
+    selector: MemorySelector | None,
+    max_results: int,
+) -> list[MemoryHeader]:
+    if not headers or selector is None:
+        return headers[:max_results]
+    requested = selector(query, headers)
+    by_path = {header.relative_path or header.path.name: header for header in headers}
+    selected: list[MemoryHeader] = []
+    seen: set[str] = set()
+    for path in requested:
+        header = by_path.get(path)
+        if header is None or path in seen:
+            continue
+        selected.append(header)
+        seen.add(path)
+    if len(selected) < max_results:
+        selected.extend(
+            header
+            for header in headers
+            if (header.relative_path or header.path.name) not in seen
+        )
+    return selected[:max_results]

src/openharness/memory/scan.py

@@ -102,6 +102,9 @@ def _parse_memory_file(path: Path, content: str) -> MemoryHeader:
         ttl_days=coerce_optional_int(metadata.get("ttl_days")),
         disabled=coerce_bool(metadata.get("disabled"), default=False),
         supersedes=coerce_str_list(metadata.get("supersedes")),
+        relative_path=path.name,
+        tags=coerce_str_list(metadata.get("tags")),
+        frontmatter=metadata,
     )
 
 

src/openharness/memory/schema.py

@@ -7,20 +7,35 @@ import json
 import re
 import secrets
 import string
+from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
 import yaml
 
 SCHEMA_VERSION = 1
 
+MemoryType = Literal["user", "feedback", "project", "reference"]
+MemoryScope = Literal["private", "project", "team"]
+
+MEMORY_TYPES: tuple[MemoryType, ...] = ("user", "feedback", "project", "reference")
+MEMORY_SCOPES: tuple[MemoryScope, ...] = ("private", "project", "team")
+
+DEFAULT_MEMORY_TYPE: MemoryType = "project"
+DEFAULT_MEMORY_SCOPE: MemoryScope = "project"
+
+MAX_ENTRYPOINT_LINES = 200
+MAX_ENTRYPOINT_BYTES = 25_000
+MAX_MANIFEST_FILES = 200
+
 FRONTMATTER_FIELDS = (
     "schema_version",
     "id",
     "name",
     "description",
     "type",
+    "scope",
     "category",
     "importance",
     "source",
@@ -30,9 +45,19 @@ FRONTMATTER_FIELDS = (
     "ttl_days",
     "disabled",
     "supersedes",
+    "tags",
 )
 
 
+@dataclass(frozen=True)
+class EntrypointView:
+    """A bounded view of ``MEMORY.md`` plus truncation diagnostics."""
+
+    content: str
+    was_truncated: bool
+    reason: str = ""
+
+
 def utc_now() -> datetime:
     """Return the current UTC time without sub-second noise."""
 
@@ -83,6 +108,121 @@ def compute_memory_signature(content: str, memory_type: str, category: str) -> s
     return hashlib.sha256(payload.encode("utf-8")).hexdigest()
 
 
+def parse_memory_type(raw: Any, *, default: MemoryType | None = None) -> MemoryType | None:
+    """Parse a frontmatter ``type`` value into the canonical runtime taxonomy."""
+
+    if isinstance(raw, str):
+        value = raw.strip().lower()
+        if value in MEMORY_TYPES:
+            return value  # type: ignore[return-value]
+        if value in {"note", "memory", "core", "knowledge"}:
+            return default
+    return default
+
+
+def parse_memory_scope(raw: Any, *, default: MemoryScope | None = None) -> MemoryScope | None:
+    """Parse a frontmatter ``scope`` value into the canonical scope taxonomy."""
+
+    if isinstance(raw, str):
+        value = raw.strip().lower()
+        if value in MEMORY_SCOPES:
+            return value  # type: ignore[return-value]
+        if value in {"personal", "user"}:
+            return "private"
+        if value in {"shared"}:
+            return "team"
+    return default
+
+
+def truncate_entrypoint_content(
+    raw: str,
+    *,
+    max_lines: int = MAX_ENTRYPOINT_LINES,
+    max_bytes: int = MAX_ENTRYPOINT_BYTES,
+) -> EntrypointView:
+    """Bound ``MEMORY.md`` by line count and UTF-8 byte count."""
+
+    lines = raw.splitlines()
+    was_line_truncated = len(lines) > max_lines
+    text = "\n".join(lines[:max_lines])
+    encoded = text.encode("utf-8")
+    was_byte_truncated = len(encoded) > max_bytes
+    if was_byte_truncated:
+        encoded = encoded[:max_bytes]
+        text = encoded.decode("utf-8", errors="ignore")
+        cut_at = text.rfind("\n")
+        if cut_at > 0:
+            text = text[:cut_at]
+    if raw.endswith("\n") and not text.endswith("\n"):
+        text += "\n"
+    if not was_line_truncated and not was_byte_truncated:
+        return EntrypointView(content=text, was_truncated=False)
+    reason = (
+        f"{len(raw.encode('utf-8'))} bytes (limit: {max_bytes})"
+        if was_byte_truncated
+        else f"{len(lines)} lines (limit: {max_lines})"
+    )
+    warning = (
+        f"\n\n> WARNING: MEMORY.md is {reason}. Only part of it was loaded. "
+        "Keep index entries one line and move detail into topic notes.\n"
+    )
+    return EntrypointView(content=text.rstrip() + warning, was_truncated=True, reason=reason)
+
+
+def memory_age_days(mtime: float, *, now: float | None = None) -> int:
+    """Return floor-rounded days elapsed since a file modification time."""
+
+    import time
+
+    current = time.time() if now is None else now
+    return max(0, int((current - mtime) // 86_400))
+
+
+def memory_age_label(mtime: float, *, now: float | None = None) -> str:
+    """Return a model-friendly age label."""
+
+    days = memory_age_days(mtime, now=now)
+    if days == 0:
+        return "today"
+    if days == 1:
+        return "yesterday"
+    return f"{days} days ago"
+
+
+def memory_freshness_text(mtime: float, *, now: float | None = None) -> str:
+    """Return a staleness warning for older memories."""
+
+    days = memory_age_days(mtime, now=now)
+    if days <= 1:
+        return ""
+    return (
+        f"This memory is {days} days old. Memories are point-in-time observations; "
+        "verify claims against the current project state before treating them as facts."
+    )
+
+
+def path_is_relative_to(path: str | Path, root: str | Path) -> bool:
+    """Compatibility helper for containment checks."""
+
+    try:
+        Path(path).resolve().relative_to(Path(root).resolve())
+    except ValueError:
+        return False
+    return True
+
+
+MEMORY_POLICY_LINES: tuple[str, ...] = (
+    "## Durable memory policy",
+    "- Store durable memory only when the information is not cheaply derivable from current files, docs, git history, or tool output.",
+    "- Use `type: user|feedback|project|reference` and optional `scope: private|project|team` frontmatter.",
+    "- `MEMORY.md` is an index, not a memory body. Keep each pointer one line.",
+    "- Update or remove stale contradictions instead of duplicating notes.",
+    "- If the user says to ignore memory, proceed as if no memory was loaded and do not cite, apply, or mention memory contents.",
+    "- Memory can be stale. Verify remembered project/code state against current files before acting on it.",
+    "- Do not save secrets, credentials, private personal context in team memory, or temporary task chatter.",
+)
+
+
 def generate_memory_id(now: datetime | None = None) -> str:
     """Generate a stable-looking memory id for a new memory file."""
 

src/openharness/memory/team.py

@@ -0,0 +1,90 @@
+"""Local team-memory vault helpers and safety guards."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from pathlib import Path
+
+from openharness.memory.paths import get_project_memory_dir
+from openharness.memory.schema import path_is_relative_to
+
+TEAM_DIR_NAME = "team"
+MEMORY_INDEX = "MEMORY.md"
+
+
+@dataclass(frozen=True)
+class SecretMatch:
+    """A possible secret found in shared memory content."""
+
+    rule_id: str
+    label: str
+
+
+SECRET_RULES: tuple[tuple[str, str, re.Pattern[str]], ...] = (
+    ("private-key", "private key", re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----")),
+    ("aws-access-key", "AWS access key", re.compile(r"\bAKIA[0-9A-Z]{16}\b")),
+    ("github-token", "GitHub token", re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b")),
+    ("openai-key", "OpenAI API key", re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b")),
+    ("anthropic-key", "Anthropic API key", re.compile(r"\bsk-ant-[A-Za-z0-9_-]{20,}\b")),
+    ("generic-secret", "secret assignment", re.compile(r"(?i)\b(secret|token|api[_-]?key|password)\s*[:=]\s*['\"]?[^'\"\s]{12,}")),
+)
+
+
+def get_team_memory_dir(cwd: str | Path) -> Path:
+    """Return the project-local shared team memory vault."""
+
+    return get_project_memory_dir(cwd) / TEAM_DIR_NAME
+
+
+def ensure_team_memory_vault(cwd: str | Path) -> Path:
+    """Create and return the team memory vault."""
+
+    team_dir = get_team_memory_dir(cwd)
+    team_dir.mkdir(parents=True, exist_ok=True)
+    entrypoint = team_dir / MEMORY_INDEX
+    if not entrypoint.exists():
+        entrypoint.write_text("# Memory Index\n", encoding="utf-8")
+    return team_dir
+
+
+def validate_team_memory_write_path(cwd: str | Path, candidate: str | Path) -> tuple[Path | None, str | None]:
+    """Validate a write target against traversal and symlink escape."""
+
+    team_dir = ensure_team_memory_vault(cwd).resolve()
+    path = Path(candidate).expanduser()
+    if not path.is_absolute():
+        path = team_dir / path
+    resolved = path.resolve()
+    if not path_is_relative_to(resolved, team_dir):
+        return None, f"Path escapes team memory directory: {candidate}"
+    parent = resolved.parent
+    deepest = parent
+    while not deepest.exists() and deepest != deepest.parent:
+        deepest = deepest.parent
+    if deepest.exists() and not path_is_relative_to(deepest.resolve(), team_dir):
+        return None, f"Path escapes team memory directory via symlink: {candidate}"
+    return resolved, None
+
+
+def scan_for_secrets(content: str) -> list[SecretMatch]:
+    """Return possible secrets in content without exposing matched values."""
+
+    matches: list[SecretMatch] = []
+    for rule_id, label, pattern in SECRET_RULES:
+        if pattern.search(content):
+            matches.append(SecretMatch(rule_id=rule_id, label=label))
+    return matches
+
+
+def check_team_memory_secrets(content: str) -> str | None:
+    """Return an error message when shared memory content appears sensitive."""
+
+    matches = scan_for_secrets(content)
+    if not matches:
+        return None
+    labels = ", ".join(match.label for match in matches)
+    return (
+        f"Content contains potential secrets ({labels}) and cannot be written to team memory. "
+        "Team memory is shared with project collaborators."
+    )

src/openharness/memory/types.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
+from typing import Any
 
 
 @dataclass(frozen=True)
@@ -27,3 +28,6 @@ class MemoryHeader:
     ttl_days: int | None = None
     disabled: bool = False
     supersedes: tuple[str, ...] = ()
+    relative_path: str = ""
+    tags: tuple[str, ...] = field(default_factory=tuple)
+    frontmatter: dict[str, Any] = field(default_factory=dict)

src/openharness/prompts/context.py

@@ -12,7 +12,8 @@ from openharness.config.paths import (
 )
 from openharness.config.settings import Settings
 from openharness.coordinator.coordinator_mode import get_coordinator_system_prompt, is_coordinator_mode
-from openharness.memory import find_relevant_memories, load_memory_prompt
+from openharness.memory import load_memory_prompt
+from openharness.memory.relevance import format_relevant_memories, select_relevant_memories
 from openharness.memory.usage import mark_memory_used
 from openharness.personalization.rules import load_local_rules
 from openharness.prompts.claudemd import load_claude_md_prompt
@@ -139,33 +140,23 @@ def build_runtime_system_prompt(
         memory_section = load_memory_prompt(
             cwd,
             max_entrypoint_lines=settings.memory.max_entrypoint_lines,
+            max_entrypoint_bytes=settings.memory.max_entrypoint_bytes,
         )
         if memory_section:
             sections.append(memory_section)
 
         if latest_user_prompt:
-            relevant = find_relevant_memories(
+            relevant = select_relevant_memories(
                 latest_user_prompt,
                 cwd,
                 max_results=settings.memory.max_files,
             )
             if relevant:
                 try:
-                    mark_memory_used(cwd, relevant, memory_dir=relevant[0].path.parent)
+                    headers = [item.header for item in relevant]
+                    mark_memory_used(cwd, headers, memory_dir=headers[0].path.parent)
                 except OSError:
                     pass
-                lines = ["# Relevant Memories"]
-                for header in relevant:
-                    content = header.path.read_text(encoding="utf-8", errors="replace").strip()
-                    lines.extend(
-                        [
-                            "",
-                            f"## {header.path.name}",
-                            "```md",
-                            content[:8000],
-                            "```",
-                        ]
-                    )
-                sections.append("\n".join(lines))
+                sections.append(format_relevant_memories(relevant))
 
     return "\n\n".join(section for section in sections if section.strip())

src/openharness/services/compact/__init__.py

@@ -890,6 +890,28 @@ def _build_session_memory_message(messages: list[ConversationMessage]) -> Conver
     )
 
 
+def _build_file_session_memory_message(metadata: dict[str, Any] | None) -> ConversationMessage | None:
+    """Build a compaction message from the persisted session-memory file."""
+
+    if not metadata:
+        return None
+    path = metadata.get("session_memory_path")
+    if not path:
+        return None
+    try:
+        from openharness.services.session_memory import (
+            get_session_memory_content,
+            session_memory_to_compact_text,
+        )
+
+        text = session_memory_to_compact_text(get_session_memory_content(str(path)))
+    except Exception:
+        return None
+    if not text.strip():
+        return None
+    return ConversationMessage.from_user_text(text)
+
+
 def try_session_memory_compaction(
     messages: list[ConversationMessage],
     *,
@@ -901,7 +923,8 @@ def try_session_memory_compaction(
     if len(messages) <= preserve_recent + 4:
         return None
     older, newer = _split_preserving_tool_pairs(messages, preserve_recent=preserve_recent)
-    summary_message = _build_session_memory_message(older)
+    file_summary_message = _build_file_session_memory_message(metadata)
+    summary_message = file_summary_message or _build_session_memory_message(older)
     if summary_message is None:
         return None
     provisional = [summary_message, *newer]
@@ -917,6 +940,7 @@ def try_session_memory_compaction(
         "pre_compact_token_count": estimate_message_tokens(messages),
         "preserve_recent": preserve_recent,
         "used_session_memory": True,
+        "used_file_session_memory": file_summary_message is not None,
         "pre_compact_discovered_tools": _extract_discovered_tools(older),
         "attachments": _extract_attachment_paths(older),
     }

src/openharness/services/memory_extract/__init__.py

@@ -0,0 +1,261 @@
+"""Durable memory extraction from completed turns."""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from openharness.api.client import ApiMessageCompleteEvent, ApiMessageRequest, SupportsStreamingMessages
+from openharness.engine.messages import ConversationMessage, ToolUseBlock
+from openharness.memory.manager import add_memory_entry
+from openharness.memory.paths import get_project_memory_dir
+from openharness.memory.relevance import build_memory_manifest
+from openharness.memory.scan import scan_memory_files
+from openharness.memory.schema import (
+    DEFAULT_MEMORY_SCOPE,
+    DEFAULT_MEMORY_TYPE,
+    MemoryScope,
+    MemoryType,
+    parse_memory_scope,
+    parse_memory_type,
+)
+from openharness.memory.team import check_team_memory_secrets, validate_team_memory_write_path
+
+log = logging.getLogger(__name__)
+
+MEMORY_WRITE_TOOLS = {"write_file", "edit_file"}
+
+
+@dataclass(frozen=True)
+class ExtractionRecord:
+    """Structured memory record proposed by the extraction pass."""
+
+    title: str
+    body: str
+    memory_type: MemoryType = DEFAULT_MEMORY_TYPE
+    scope: MemoryScope = DEFAULT_MEMORY_SCOPE
+    description: str = ""
+    tags: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class ExtractionResult:
+    """Outcome of a durable memory extraction run."""
+
+    skipped: bool
+    reason: str = ""
+    records: tuple[ExtractionRecord, ...] = ()
+    written_paths: tuple[Path, ...] = ()
+
+
+def has_memory_writes_since(
+    messages: list[ConversationMessage],
+    memory_dir: str | Path,
+    *,
+    cwd: str | Path | None = None,
+) -> bool:
+    """Return whether the visible turn already wrote memory files."""
+
+    root = Path(memory_dir).expanduser().resolve()
+    write_base = Path(cwd).expanduser().resolve() if cwd is not None else root
+    for message in messages:
+        for block in message.content:
+            if not isinstance(block, ToolUseBlock):
+                continue
+            if block.name not in MEMORY_WRITE_TOOLS:
+                continue
+            raw_path = block.input.get("path") or block.input.get("file_path")
+            if not raw_path:
+                continue
+            path = Path(str(raw_path)).expanduser()
+            if not path.is_absolute():
+                path = write_base / path
+            try:
+                path.resolve().relative_to(root)
+            except ValueError:
+                continue
+            return True
+    return False
+
+
+async def extract_memories_from_turn(
+    *,
+    cwd: str | Path,
+    api_client: SupportsStreamingMessages,
+    model: str,
+    messages: list[ConversationMessage],
+    max_records: int = 3,
+) -> ExtractionResult:
+    """Ask the model for durable memory candidates and apply them."""
+
+    memory_dir = get_project_memory_dir(cwd)
+    if len(messages) < 2:
+        return ExtractionResult(skipped=True, reason="not enough messages")
+    if has_memory_writes_since(messages, memory_dir, cwd=cwd):
+        return ExtractionResult(skipped=True, reason="main conversation already wrote memory")
+
+    prompt = build_extraction_prompt(cwd, messages, max_records=max_records)
+    final_text = ""
+    async for event in api_client.stream_message(
+        ApiMessageRequest(
+            model=model,
+            messages=[ConversationMessage.from_user_text(prompt)],
+            system_prompt=EXTRACTION_SYSTEM_PROMPT,
+            max_tokens=2048,
+            tools=[],
+        )
+    ):
+        if isinstance(event, ApiMessageCompleteEvent):
+            final_text = event.message.text
+            break
+    records = parse_extraction_records(final_text, max_records=max_records)
+    if not records:
+        return ExtractionResult(skipped=True, reason="no durable memories proposed")
+    return apply_extraction_records(cwd, records)
+
+
+def build_extraction_prompt(cwd: str | Path, messages: list[ConversationMessage], *, max_records: int) -> str:
+    """Build the extraction request from recent messages and manifest."""
+
+    manifest = build_memory_manifest(scan_memory_files(cwd, max_files=80))
+    transcript = "\n".join(_summarize_message(message) for message in messages[-12:])
+    return (
+        "Extract only durable memories from the recent conversation.\n"
+        f"Return JSON with at most {max_records} records. Existing memory manifest:\n"
+        f"{manifest or '(empty)'}\n\n"
+        "Recent conversation:\n"
+        f"{transcript}\n\n"
+        "JSON schema: {\"memories\":[{\"title\":\"...\",\"type\":\"user|feedback|project|reference\","
+        "\"scope\":\"private|project|team\",\"description\":\"...\",\"body\":\"...\",\"tags\":[\"...\"]}]}"
+    )
+
+
+EXTRACTION_SYSTEM_PROMPT = """You maintain OpenHarness durable memory.
+Save only stable, future-useful facts that are not derivable from current files,
+git history, or documentation. Prefer updating existing memories conceptually
+over duplicating them. Do not save secrets. If nothing is worth saving, return
+{"memories": []}.
+"""
+
+
+def parse_extraction_records(text: str, *, max_records: int = 3) -> tuple[ExtractionRecord, ...]:
+    """Parse JSON memory extraction output."""
+
+    try:
+        payload = json.loads(_extract_json_object(text))
+    except json.JSONDecodeError:
+        return ()
+    raw_records = payload.get("memories") if isinstance(payload, dict) else None
+    if not isinstance(raw_records, list):
+        return ()
+    records: list[ExtractionRecord] = []
+    for item in raw_records[:max_records]:
+        if not isinstance(item, dict):
+            continue
+        title = str(item.get("title") or "").strip()
+        body = str(item.get("body") or "").strip()
+        if not title or not body:
+            continue
+        memory_type = parse_memory_type(item.get("type"), default=DEFAULT_MEMORY_TYPE) or DEFAULT_MEMORY_TYPE
+        scope = parse_memory_scope(item.get("scope"), default=DEFAULT_MEMORY_SCOPE) or DEFAULT_MEMORY_SCOPE
+        tags_raw = item.get("tags") or ()
+        tags = tuple(str(tag).strip() for tag in tags_raw if str(tag).strip()) if isinstance(tags_raw, list) else ()
+        records.append(
+            ExtractionRecord(
+                title=title,
+                body=body,
+                memory_type=memory_type,
+                scope=scope,
+                description=str(item.get("description") or "").strip(),
+                tags=tags,
+            )
+        )
+    return tuple(records)
+
+
+def apply_extraction_records(cwd: str | Path, records: tuple[ExtractionRecord, ...]) -> ExtractionResult:
+    """Write accepted records to durable memory."""
+
+    written: list[Path] = []
+    for record in records:
+        if record.scope == "team":
+            secret_error = check_team_memory_secrets(record.body)
+            if secret_error:
+                log.warning("memory extraction skipped team record %r: %s", record.title, secret_error)
+                continue
+            path, error = validate_team_memory_write_path(cwd, f"{record.title}.md")
+            if error or path is None:
+                log.warning("memory extraction skipped team record %r: %s", record.title, error)
+                continue
+        written.append(
+            add_memory_entry(
+                cwd,
+                record.title,
+                record.body,
+                memory_type=record.memory_type,
+                scope=record.scope,
+                description=record.description,
+                tags=record.tags,
+            )
+        )
+    return ExtractionResult(skipped=not bool(written), reason="" if written else "all records rejected", records=records, written_paths=tuple(written))
+
+
+def validate_extraction_tool_request(tool_name: str, tool_input: dict[str, Any], memory_dir: str | Path) -> tuple[bool, str]:
+    """Permission guard for extraction-like agents."""
+
+    if tool_name in {"read_file", "grep", "glob"}:
+        return True, ""
+    if tool_name == "bash":
+        command = str(tool_input.get("command") or "")
+        if _is_read_only_shell(command):
+            return True, ""
+        return False, "memory extraction may only run read-only shell commands"
+    if tool_name in {"write_file", "edit_file"}:
+        raw_path = str(tool_input.get("path") or tool_input.get("file_path") or "")
+        if not raw_path:
+            return False, "memory extraction write requires a path"
+        root = Path(memory_dir).expanduser().resolve()
+        path = Path(raw_path).expanduser()
+        if not path.is_absolute():
+            path = root / path
+        try:
+            path.resolve().relative_to(root)
+        except ValueError:
+            return False, f"memory extraction writes must stay within {root}"
+        return True, ""
+    return False, f"memory extraction cannot use tool {tool_name}"
+
+
+def _extract_json_object(text: str) -> str:
+    stripped = text.strip()
+    if stripped.startswith("{") and stripped.endswith("}"):
+        return stripped
+    start = stripped.find("{")
+    end = stripped.rfind("}")
+    if start >= 0 and end > start:
+        return stripped[start : end + 1]
+    return stripped
+
+
+def _summarize_message(message: ConversationMessage) -> str:
+    text = " ".join(message.text.split())
+    if text:
+        return f"{message.role}: {text[:1200]}"
+    if message.tool_uses:
+        return f"{message.role}: tool calls -> {', '.join(block.name for block in message.tool_uses)}"
+    return f"{message.role}: [non-text content]"
+
+
+def _is_read_only_shell(command: str) -> bool:
+    lowered = command.strip().lower()
+    if not lowered:
+        return False
+    denied = (" > ", ">>", " rm ", " mv ", " cp ", " sed -i", " tee ", "python -c", "python3 -c")
+    if any(marker in f" {lowered} " for marker in denied):
+        return False
+    first = lowered.split(maxsplit=1)[0]
+    return first in {"ls", "pwd", "cat", "head", "tail", "rg", "grep", "find", "git", "wc", "sed", "awk", "stat"}

src/openharness/services/session_memory/__init__.py

@@ -0,0 +1,139 @@
+"""File-backed session memory for compact continuity."""
+
+from __future__ import annotations
+
+from hashlib import sha1
+from pathlib import Path
+
+from openharness.config.paths import get_data_dir
+from openharness.engine.messages import ConversationMessage, ToolResultBlock
+from openharness.services.token_estimation import estimate_tokens
+from openharness.utils.fs import atomic_write_text
+
+MAX_SESSION_MEMORY_CHARS = 12_000
+MAX_RECENT_LINES = 80
+
+
+def get_session_memory_dir(cwd: str | Path) -> Path:
+    """Return the project session-memory directory."""
+
+    root = Path(cwd).resolve()
+    digest = sha1(str(root).encode("utf-8")).hexdigest()[:12]
+    path = get_data_dir() / "session-memory" / f"{root.name}-{digest}"
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def get_session_memory_path(cwd: str | Path, session_id: str | None = None) -> Path:
+    """Return the markdown session-memory path."""
+
+    safe_session = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in (session_id or "default"))
+    return get_session_memory_dir(cwd) / f"{safe_session or 'default'}.md"
+
+
+def prepare_session_memory_metadata(
+    cwd: str | Path,
+    tool_metadata: dict[str, object],
+    *,
+    session_id: str | None = None,
+) -> Path:
+    """Ensure metadata points compaction to the session-memory file."""
+
+    sid = session_id or str(tool_metadata.get("session_id") or "default")
+    path = get_session_memory_path(cwd, sid)
+    tool_metadata["session_memory_path"] = str(path)
+    return path
+
+
+def get_session_memory_content(path: str | Path | None) -> str:
+    """Read session memory content if available."""
+
+    if not path:
+        return ""
+    candidate = Path(path).expanduser()
+    if not candidate.exists():
+        return ""
+    try:
+        return candidate.read_text(encoding="utf-8", errors="replace")
+    except OSError:
+        return ""
+
+
+def update_session_memory_file(
+    cwd: str | Path,
+    messages: list[ConversationMessage],
+    *,
+    tool_metadata: dict[str, object] | None = None,
+    session_id: str | None = None,
+) -> Path:
+    """Update the deterministic session-memory checkpoint."""
+
+    path = prepare_session_memory_metadata(cwd, tool_metadata or {}, session_id=session_id)
+    body = build_session_memory_document(messages, tool_metadata=tool_metadata)
+    atomic_write_text(path, body)
+    return path
+
+
+def build_session_memory_document(
+    messages: list[ConversationMessage],
+    *,
+    tool_metadata: dict[str, object] | None = None,
+) -> str:
+    """Build a compact markdown checkpoint for the current session."""
+
+    state = tool_metadata.get("task_focus_state") if isinstance(tool_metadata, dict) else None
+    goal = ""
+    next_step = ""
+    verified: list[str] = []
+    artifacts: list[str] = []
+    if isinstance(state, dict):
+        goal = str(state.get("goal") or "").strip()
+        next_step = str(state.get("next_step") or "").strip()
+        verified = [str(item).strip() for item in state.get("verified_state", []) if str(item).strip()]
+        artifacts = [str(item).strip() for item in state.get("active_artifacts", []) if str(item).strip()]
+
+    lines = ["# Session Memory", ""]
+    lines.extend(["## Current State", goal or "(no current goal recorded)", ""])
+    if next_step:
+        lines.extend(["## Next Step", next_step, ""])
+    if verified:
+        lines.extend(["## Verified Work", *[f"- {item}" for item in verified[-10:]], ""])
+    if artifacts:
+        lines.extend(["## Active Artifacts", *[f"- {item}" for item in artifacts[-10:]], ""])
+    lines.extend(["## Recent Conversation", *_recent_message_lines(messages), ""])
+    text = "\n".join(lines).strip() + "\n"
+    if len(text) > MAX_SESSION_MEMORY_CHARS:
+        text = text[:MAX_SESSION_MEMORY_CHARS].rsplit("\n", 1)[0]
+        text += "\n\n> Session memory was truncated to stay within budget.\n"
+    return text
+
+
+def session_memory_to_compact_text(content: str) -> str:
+    """Prepare persisted session memory for insertion across compact boundaries."""
+
+    stripped = content.strip()
+    if not stripped:
+        return ""
+    if estimate_tokens(stripped) > 4_000:
+        stripped = stripped[:MAX_SESSION_MEMORY_CHARS].rsplit("\n", 1)[0]
+    return "Session memory checkpoint from earlier in this conversation:\n" + stripped
+
+
+def _recent_message_lines(messages: list[ConversationMessage]) -> list[str]:
+    lines: list[str] = []
+    for message in messages[-MAX_RECENT_LINES:]:
+        line = _summarize_message(message)
+        if line:
+            lines.append(f"- {line}")
+    return lines or ["- (no recent messages)"]
+
+
+def _summarize_message(message: ConversationMessage) -> str:
+    text = " ".join(message.text.split())
+    if text:
+        return f"{message.role}: {text[:220]}"
+    if message.tool_uses:
+        return f"{message.role}: tool calls -> {', '.join(block.name for block in message.tool_uses[:6])}"
+    if any(isinstance(block, ToolResultBlock) for block in message.content):
+        return f"{message.role}: tool results returned"
+    return f"{message.role}: [non-text content]"

tests/test_memory/test_claude_runtime_memory.py

@@ -0,0 +1,223 @@
+"""Tests for Claude Code-inspired memory runtime behavior."""
+
+from __future__ import annotations
+
+import os
+import time
+from pathlib import Path
+
+import pytest
+
+from openharness.api.client import ApiMessageCompleteEvent
+from openharness.api.usage import UsageSnapshot
+from openharness.commands.registry import CommandContext, create_default_command_registry
+from openharness.config.settings import Settings
+from openharness.engine.messages import ConversationMessage, ToolUseBlock
+from openharness.engine.query_engine import QueryEngine
+from openharness.memory import add_memory_entry, get_project_memory_dir
+from openharness.memory.agent import (
+    ensure_agent_memory_vault,
+    get_agent_memory_entrypoint,
+    initialize_agent_memory_from_snapshot,
+)
+from openharness.memory.relevance import format_relevant_memories, select_relevant_memories
+from openharness.memory.schema import (
+    memory_freshness_text,
+    parse_memory_scope,
+    parse_memory_type,
+    truncate_entrypoint_content,
+)
+from openharness.memory.team import (
+    check_team_memory_secrets,
+    ensure_team_memory_vault,
+    validate_team_memory_write_path,
+)
+from openharness.permissions import PermissionChecker
+from openharness.services.memory_extract import (
+    extract_memories_from_turn,
+    has_memory_writes_since,
+    parse_extraction_records,
+    validate_extraction_tool_request,
+)
+from openharness.services.session_memory import (
+    get_session_memory_content,
+    get_session_memory_path,
+    update_session_memory_file,
+)
+from openharness.tools import create_default_tool_registry
+
+
+class _FakeApiClient:
+    def __init__(self, text: str = "done") -> None:
+        self.text = text
+        self.requests = []
+
+    async def stream_message(self, request):
+        self.requests.append(request)
+        yield ApiMessageCompleteEvent(
+            message=ConversationMessage.from_user_text(self.text).model_copy(update={"role": "assistant"}),
+            usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+            stop_reason="end_turn",
+        )
+
+
+def test_schema_truncation_and_freshness() -> None:
+    raw = "\n".join(f"- item {idx}" for idx in range(240))
+    view = truncate_entrypoint_content(raw, max_lines=10, max_bytes=1_000)
+
+    assert view.was_truncated is True
+    assert "WARNING" in view.content
+    assert parse_memory_type("note", default="project") == "project"
+    assert parse_memory_scope("personal") == "private"
+    assert "2 days old" in memory_freshness_text(time.time() - 2 * 86_400)
+
+
+def test_relevance_formats_staleness(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    project = tmp_path / "repo"
+    project.mkdir()
+    path = add_memory_entry(project, "Redis Rule", "Redis cache decisions matter.", memory_type="project")
+    old = time.time() - 3 * 86_400
+    os.utime(path, (old, old))
+
+    selected = select_relevant_memories("redis cache", project)
+    rendered = format_relevant_memories(selected)
+
+    assert selected
+    assert "3 days old" in rendered
+    assert "Redis cache decisions" in rendered
+
+
+@pytest.mark.asyncio
+async def test_memory_extraction_writes_typed_note(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    project = tmp_path / "repo"
+    project.mkdir()
+    api = _FakeApiClient(
+        '{"memories":[{"title":"Testing Policy","type":"feedback","scope":"project",'
+        '"description":"real database tests","body":"Use real database tests for migrations.",'
+        '"tags":["testing"]}]}'
+    )
+    messages = [
+        ConversationMessage.from_user_text("do not mock database migrations"),
+        ConversationMessage.from_user_text("noted").model_copy(update={"role": "assistant"}),
+    ]
+
+    result = await extract_memories_from_turn(cwd=project, api_client=api, model="claude-test", messages=messages)
+
+    assert result.skipped is False
+    assert len(result.written_paths) == 1
+    text = result.written_paths[0].read_text(encoding="utf-8")
+    assert 'type: "feedback"' in text
+    assert 'scope: "project"' in text
+    assert "Use real database tests" in text
+
+
+def test_memory_extraction_parser_and_tool_guard(tmp_path: Path) -> None:
+    records = parse_extraction_records(
+        '{"memories":[{"title":"User Role","type":"user","scope":"private","body":"User knows Go."}]}'
+    )
+    assert records[0].memory_type == "user"
+    assert records[0].scope == "private"
+
+    ok, _ = validate_extraction_tool_request("bash", {"command": "rg memory src"}, tmp_path)
+    denied, reason = validate_extraction_tool_request("write_file", {"path": "../x", "content": "x"}, tmp_path)
+
+    assert ok is True
+    assert denied is False
+    assert "within" in reason
+
+
+def test_memory_write_detection_resolves_relative_paths_from_cwd(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    project = tmp_path / "repo"
+    project.mkdir()
+    memory_dir = get_project_memory_dir(project)
+
+    normal_relative_write = ConversationMessage(
+        role="assistant",
+        content=[ToolUseBlock(name="write_file", input={"path": "REPORT.md", "content": "ok"})],
+    )
+    memory_absolute_write = ConversationMessage(
+        role="assistant",
+        content=[ToolUseBlock(name="write_file", input={"path": str(memory_dir / "report.md")})],
+    )
+
+    assert has_memory_writes_since([normal_relative_write], memory_dir, cwd=project) is False
+    assert has_memory_writes_since([memory_absolute_write], memory_dir, cwd=project) is True
+
+
+def test_session_memory_file_round_trip(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    project = tmp_path / "repo"
+    project.mkdir()
+    messages = [ConversationMessage.from_user_text("current task: finish memory runtime")]
+
+    path = update_session_memory_file(project, messages, session_id="abc")
+
+    assert path == get_session_memory_path(project, "abc")
+    assert "finish memory runtime" in get_session_memory_content(path)
+
+
+def test_team_memory_guards_and_agent_memory(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    project = tmp_path / "repo"
+    project.mkdir()
+
+    team_dir = ensure_team_memory_vault(project)
+    valid_path, error = validate_team_memory_write_path(project, "notes/shared.md")
+    escaped_path, escaped_error = validate_team_memory_write_path(project, "../outside.md")
+    agent_dir = ensure_agent_memory_vault(project, "reviewer/bot", "project")
+
+    assert team_dir.exists()
+    assert valid_path is not None and error is None
+    assert escaped_path is None and escaped_error
+    assert check_team_memory_secrets("OPENAI_API_KEY=sk-12345678901234567890")
+    assert agent_dir.exists()
+    assert get_agent_memory_entrypoint(project, "reviewer/bot", "project").exists()
+
+
+def test_agent_memory_snapshot_initializes(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    project = tmp_path / "repo"
+    project.mkdir()
+    snapshot = project / ".openharness" / "agent-memory-snapshots" / "reviewer"
+    snapshot.mkdir(parents=True)
+    (snapshot / "MEMORY.md").write_text("# Snapshot\n", encoding="utf-8")
+
+    target = initialize_agent_memory_from_snapshot(project, "reviewer", "project")
+
+    assert target is not None
+    assert (target / "MEMORY.md").read_text(encoding="utf-8") == "# Snapshot\n"
+
+
+@pytest.mark.asyncio
+async def test_memory_commands_expose_session_team_and_agent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    project = tmp_path / "repo"
+    project.mkdir()
+    engine = QueryEngine(
+        api_client=_FakeApiClient(),
+        tool_registry=create_default_tool_registry(),
+        permission_checker=PermissionChecker(Settings().permission),
+        cwd=project,
+        model="claude-test",
+        system_prompt="system",
+    )
+    context = CommandContext(engine=engine, cwd=str(project), session_id="s1")
+    registry = create_default_command_registry()
+
+    for raw in ("/memory session", "/memory team", "/memory agent status reviewer project"):
+        command, args = registry.lookup(raw)
+        assert command is not None
+        result = await command.handler(args, context)
+        assert result.message
+
+    command, args = registry.lookup("/memory add --type feedback --scope private Style :: Keep answers concise.")
+    assert command is not None
+    result = await command.handler(args, context)
+    assert "Added memory entry" in (result.message or "")
+    assert 'type: "feedback"' in (get_project_memory_dir(project) / "style.md").read_text(encoding="utf-8")