"""Background auto-memory extraction. OpenSpace runs a small isolated tool loop through ``services.side_query.run_side_query`` to extract durable memory in the background. Implemented behavior: - cursor based incremental extraction - skip when the main agent already wrote auto memory - per-N-turn throttling - coalesced trailing run while extraction is in progress - restricted tool gate for memory-related reads, searches, and writes - best-effort fire-and-forget execution with drain support """ from __future__ import annotations import asyncio import hashlib import inspect import json import logging import os import time from dataclasses import dataclass, field from pathlib import Path from typing import Any, Awaitable, Callable, Iterable, Mapping, Sequence from openspace.grounding.core.tool.base import BaseTool from openspace.services.memory.memdir import ( ENTRYPOINT_NAME, ensure_memory_dir_exists, get_auto_mem_path, is_auto_mem_path, is_auto_memory_enabled, is_extract_mode_active, ) from openspace.services.memory.daily_log import ( MEMORY_LOG_TOOL_NAME, MemoryLogTool, build_extract_daily_log_prompt, extract_logged_entries, get_memory_mode, ) from openspace.services.memory.memory_scan import format_memory_manifest, scan_memory_files from openspace.services.memory.memory_types import ( MEMORY_FRONTMATTER_EXAMPLE, TYPES_SECTION_COMBINED, TYPES_SECTION_INDIVIDUAL, WHAT_NOT_TO_SAVE_SECTION, ) from openspace.services.memory.task_scope import maybe_memory_task_scope_key from openspace.services.conversation.side_query import run_side_query from openspace.services.tooling.context import ToolUseContext logger = logging.getLogger(__name__) FILE_READ_TOOL_NAME = "read" GREP_TOOL_NAME = "grep" GLOB_TOOL_NAME = "glob" BASH_TOOL_NAME = "bash" FILE_EDIT_TOOL_NAME = "edit" FILE_WRITE_TOOL_NAME = "write" MEMORY_READ_TOOL_NAME = "memory_read" MEMORY_WRITE_TOOL_NAME = "memory_write" OPENSPACE_EXTRACT_MEMORIES_ENABLED_ENV = "OPENSPACE_EXTRACT_MEMORIES_ENABLED" OPENSPACE_DISABLE_EXTRACT_MEMORIES_ENV = "OPENSPACE_DISABLE_EXTRACT_MEMORIES" OPENSPACE_EXTRACT_MEMORIES_EVERY_N_TURNS_ENV = "OPENSPACE_EXTRACT_MEMORIES_EVERY_N_TURNS" OPENSPACE_EXTRACT_MEMORIES_SKIP_INDEX_ENV = "OPENSPACE_EXTRACT_MEMORIES_SKIP_INDEX" OPENSPACE_MEMORY_EXTRACT_MODEL_ENV = "OPENSPACE_MEMORY_EXTRACT_MODEL" OPENSPACE_EXTRACT_MEMORIES_ALLOW_NON_INTERACTIVE_ENV = ( "OPENSPACE_EXTRACT_MEMORIES_ALLOW_NON_INTERACTIVE" ) DEFAULT_MAX_EXTRACT_TURNS = 5 AppendSystemMessageFn = Callable[[dict[str, Any]], Awaitable[None] | None] AutoMemCanUseToolFn = Callable[ [BaseTool | None, Mapping[str, Any]], Awaitable[dict[str, Any]], ] @dataclass(frozen=True) class MemoryCursor: """Cursor for the last parent message processed by extraction. OpenSpace stores a UUID. OpenSpace messages do not always carry UUIDs, so the cursor stores a stable fingerprint plus its occurrence number. If the fingerprint cannot be found later, we fall back to "all messages", matching OpenSpace's compaction fallback. """ fingerprint: str occurrence: int @dataclass(slots=True) class MemoryExtractionResult: """Result object for tests and observability. OpenSpace's public API resolves ``void``. Returning this value in OS does not affect the fire-and-forget call path, but makes the behavior auditable. """ ran: bool = False skipped_reason: str | None = None new_message_count: int = 0 written_paths: list[str] = field(default_factory=list) memory_paths: list[str] = field(default_factory=list) turn_count: int = 0 error: str | None = None duration_ms: float = 0.0 @dataclass(slots=True) class MemoryExtractorRuntimeState: last_cursor: MemoryCursor | None = None in_progress: bool = False turns_since_last_extraction: int = 0 pending_context: tuple[ToolUseContext, AppendSystemMessageFn | None] | None = None def _env_truthy(value: str | None) -> bool: return value is not None and value.lower() in {"1", "true", "yes", "on"} def _env_defined_falsy(value: str | None) -> bool: return value is not None and value.lower() in {"0", "false", "no", "off", ""} def _extract_feature_enabled() -> bool: if _env_truthy(os.environ.get(OPENSPACE_DISABLE_EXTRACT_MEMORIES_ENV)): return False enabled = os.environ.get(OPENSPACE_EXTRACT_MEMORIES_ENABLED_ENV) # OpenSpace gates this behind GrowthBook with default ``false``. OpenSpace has # no GrowthBook layer, so an explicit env/settings opt-in is the equivalent # feature decision. return _env_truthy(enabled) def _extract_every_n_turns(default: int = 1) -> int: raw = os.environ.get(OPENSPACE_EXTRACT_MEMORIES_EVERY_N_TURNS_ENV) if not raw: return default try: parsed = int(raw) except ValueError: return default return max(1, parsed) def _extract_skip_index() -> bool: return _env_truthy(os.environ.get(OPENSPACE_EXTRACT_MEMORIES_SKIP_INDEX_ENV)) def _extract_allow_non_interactive() -> bool: return _env_truthy( os.environ.get(OPENSPACE_EXTRACT_MEMORIES_ALLOW_NON_INTERACTIVE_ENV) ) def _opener(new_message_count: int, existing_memories: str) -> str: manifest = ( "\n\n## Existing memory files\n\n" + existing_memories + "\n\nCheck this list before writing - update an existing file rather than creating a duplicate." if existing_memories else "" ) return "\n".join( [ ( "You are now acting as the memory extraction subagent. " f"Analyze the most recent ~{new_message_count} messages above and use them to update your persistent memory systems." ), "", ( f"Available tools: {FILE_READ_TOOL_NAME}, {GREP_TOOL_NAME}, {GLOB_TOOL_NAME}, " f"read-only {BASH_TOOL_NAME} (ls/find/cat/stat/wc/head/tail and similar), " f"and {FILE_EDIT_TOOL_NAME}/{FILE_WRITE_TOOL_NAME} for paths inside the memory directory only. " f"{MEMORY_READ_TOOL_NAME}/{MEMORY_WRITE_TOOL_NAME} are also available in OpenSpace and are already scoped to the memory directory. " f"{BASH_TOOL_NAME} rm is not permitted. All other tools - MCP, Agent, write-capable {BASH_TOOL_NAME}, etc - will be denied." ), "", ( f"You have a limited turn budget. {FILE_EDIT_TOOL_NAME} requires a prior " f"{FILE_READ_TOOL_NAME} of the same file, so the efficient strategy is: " f"turn 1 - issue all {FILE_READ_TOOL_NAME} calls in parallel for every file you might update; " f"turn 2 - issue all {FILE_WRITE_TOOL_NAME}/{FILE_EDIT_TOOL_NAME} or {MEMORY_WRITE_TOOL_NAME} calls in parallel. " "Do not interleave reads and writes across multiple turns." ), "", ( f"You MUST only use content from the last ~{new_message_count} messages to update your persistent memories. " "Do not waste any turns attempting to investigate or verify that content further - no grepping source files, " "no reading code to confirm a pattern exists, no git commands." + manifest ), ] ) def build_extract_auto_only_prompt( new_message_count: int, existing_memories: str, skip_index: bool = False, ) -> str: """Build OpenSpace's auto-only extraction prompt, adapted to OS tool names.""" if skip_index: how_to_save = [ "## How to save memories", "", "Write each memory to its own file (e.g., `user_role.md`, `feedback_testing.md`) using this frontmatter format:", "", *MEMORY_FRONTMATTER_EXAMPLE, "", "- Organize memory semantically by topic, not chronologically", "- Update or remove memories that turn out to be wrong or outdated", "- Do not write duplicate memories. First check if there is an existing memory you can update before writing a new one.", ] else: how_to_save = [ "## How to save memories", "", "Saving a memory is a two-step process:", "", "**Step 1** - write the memory to its own file (e.g., `user_role.md`, `feedback_testing.md`) using this frontmatter format:", "", *MEMORY_FRONTMATTER_EXAMPLE, "", ( f"**Step 2** - add a pointer to that file in `{ENTRYPOINT_NAME}`. " f"`{ENTRYPOINT_NAME}` is an index, not a memory - each entry should be one line, under ~150 characters: " "`- [Title](file.md) - one-line hook`. It has no frontmatter. Never write memory content directly into " f"`{ENTRYPOINT_NAME}`." ), "", f"- `{ENTRYPOINT_NAME}` is always loaded into your system prompt - lines after 200 will be truncated, so keep the index concise", "- Organize memory semantically by topic, not chronologically", "- Update or remove memories that turn out to be wrong or outdated", "- Do not write duplicate memories. First check if there is an existing memory you can update before writing a new one.", ] return "\n".join( [ _opener(new_message_count, existing_memories), "", "If the user explicitly asks you to remember something, save it immediately as whichever type fits best. If they ask you to forget something, find and remove the relevant entry.", "", *TYPES_SECTION_INDIVIDUAL, *WHAT_NOT_TO_SAVE_SECTION, "", *how_to_save, ] ) def build_extract_combined_prompt( new_message_count: int, existing_memories: str, skip_index: bool = False, *, team_memory_enabled: bool = False, ) -> str: """Build OpenSpace's combined auto+team prompt. Team memory is intentionally disabled in OpenSpace because it needs a backend/API surface (DEC-025). The function keeps the OpenSpace return structure so later TeamMem work has a single place to connect. """ if not team_memory_enabled: return build_extract_auto_only_prompt( new_message_count, existing_memories, skip_index, ) if skip_index: how_to_save = [ "## How to save memories", "", "Write each memory to its own file in the chosen directory (private or team, per the type's scope guidance) using this frontmatter format:", "", *MEMORY_FRONTMATTER_EXAMPLE, "", "- Organize memory semantically by topic, not chronologically", "- Update or remove memories that turn out to be wrong or outdated", "- Do not write duplicate memories. First check if there is an existing memory you can update before writing a new one.", ] else: how_to_save = [ "## How to save memories", "", "Saving a memory is a two-step process:", "", "**Step 1** - write the memory to its own file in the chosen directory (private or team, per the type's scope guidance) using this frontmatter format:", "", *MEMORY_FRONTMATTER_EXAMPLE, "", "**Step 2** - add a pointer to that file in the same directory's `MEMORY.md`. Each directory (private and team) has its own `MEMORY.md` index - each entry should be one line, under ~150 characters: `- [Title](file.md) - one-line hook`. They have no frontmatter. Never write memory content directly into a `MEMORY.md`.", "", "- Both `MEMORY.md` indexes are loaded into your system prompt - lines after 200 will be truncated, so keep them concise", "- Organize memory semantically by topic, not chronologically", "- Update or remove memories that turn out to be wrong or outdated", "- Do not write duplicate memories. First check if there is an existing memory you can update before writing a new one.", ] return "\n".join( [ _opener(new_message_count, existing_memories), "", "If the user explicitly asks you to remember something, save it immediately as whichever type fits best. If they ask you to forget something, find and remove the relevant entry.", "", *TYPES_SECTION_COMBINED, *WHAT_NOT_TO_SAVE_SECTION, "- You MUST avoid saving sensitive data within shared team memories. For example, never save API keys or user credentials.", "", *how_to_save, ] ) def create_auto_mem_can_use_tool(memory_dir: str | Path) -> AutoMemCanUseToolFn: """Return the tool gate for auto-memory background agents.""" root = Path(memory_dir).expanduser().resolve() async def can_use_tool( tool: BaseTool | None, input: Mapping[str, Any], ) -> dict[str, Any]: if tool is None: return _deny_auto_mem_tool( "unknown", "Tool is not available to the auto-memory extraction agent.", ) name = tool.name data = dict(input) if name in {MEMORY_READ_TOOL_NAME, MEMORY_WRITE_TOOL_NAME}: return {"behavior": "allow", "updated_input": data} if name == MEMORY_LOG_TOOL_NAME: return {"behavior": "allow", "updated_input": data} if name in {FILE_READ_TOOL_NAME, GREP_TOOL_NAME, GLOB_TOOL_NAME}: return {"behavior": "allow", "updated_input": data} if name == BASH_TOOL_NAME: try: if tool.is_read_only(data): return {"behavior": "allow", "updated_input": data} except Exception: pass return _deny_auto_mem_tool( name, ( "Only read-only shell commands are permitted in this context " "(ls, find, grep, cat, stat, wc, head, tail, and similar)." ), ) if name in {FILE_EDIT_TOOL_NAME, FILE_WRITE_TOOL_NAME}: file_path = data.get("file_path") if isinstance(file_path, str) and _path_is_inside(file_path, root): return {"behavior": "allow", "updated_input": data} return _deny_auto_mem_tool( name, f"{name} is only permitted for paths inside {root}.", ) return _deny_auto_mem_tool( name, ( f"Only {FILE_READ_TOOL_NAME}, {GREP_TOOL_NAME}, {GLOB_TOOL_NAME}, " f"read-only {BASH_TOOL_NAME}, {FILE_EDIT_TOOL_NAME}/{FILE_WRITE_TOOL_NAME} within {root}, " f"{MEMORY_READ_TOOL_NAME}/{MEMORY_WRITE_TOOL_NAME}, and {MEMORY_LOG_TOOL_NAME} are allowed." ), ) return can_use_tool def _deny_auto_mem_tool(tool_name: str, reason: str) -> dict[str, Any]: logger.debug("[autoMem] denied %s: %s", tool_name, reason) return { "behavior": "deny", "message": reason, "decision_reason": {"type": "other", "reason": reason}, } def _path_is_inside(file_path: str, root: Path) -> bool: try: candidate = Path(file_path).expanduser() if not candidate.is_absolute(): candidate = root / candidate candidate.resolve().relative_to(root) return True except (OSError, ValueError): return False class MemoryExtractor: """Stateful background memory extractor. A single extractor instance mirrors OpenSpace's closure-scoped state created by ``initExtractMemories``. """ def __init__( self, *, max_turns: int = DEFAULT_MAX_EXTRACT_TURNS, throttle_turns: int | None = None, ) -> None: self.max_turns = max(1, int(max_turns)) self.throttle_turns = throttle_turns self._states: dict[str, MemoryExtractorRuntimeState] = {} self._default_state = MemoryExtractorRuntimeState() self._in_flight: set[asyncio.Task[MemoryExtractionResult]] = set() self._task_scope_keys: dict[asyncio.Task[MemoryExtractionResult], str | None] = {} @property def last_cursor(self) -> MemoryCursor | None: return self._default_state.last_cursor def _runtime_state(self, context: ToolUseContext) -> MemoryExtractorRuntimeState: scope_key = maybe_memory_task_scope_key(context) if scope_key is None: return self._default_state state = self._states.get(scope_key) if state is None: state = MemoryExtractorRuntimeState() self._states[scope_key] = state return state async def execute( self, context: ToolUseContext, append_system_message: AppendSystemMessageFn | None = None, ) -> MemoryExtractionResult: """Run extraction once, including trailing-run coalescing.""" task = asyncio.current_task() if task is not None: self._in_flight.add(task) # type: ignore[arg-type] self._task_scope_keys[task] = maybe_memory_task_scope_key(context) # type: ignore[index] try: return await self._execute_impl(context, append_system_message) finally: if task is not None: self._in_flight.discard(task) # type: ignore[arg-type] self._task_scope_keys.pop(task, None) def submit( self, context: ToolUseContext, append_system_message: AppendSystemMessageFn | None = None, ) -> asyncio.Task[MemoryExtractionResult]: """Schedule fire-and-forget extraction and track it for draining.""" task = asyncio.create_task(self.execute(context, append_system_message)) self._in_flight.add(task) self._task_scope_keys[task] = maybe_memory_task_scope_key(context) def _done(done: asyncio.Task[MemoryExtractionResult]) -> None: self._in_flight.discard(done) self._task_scope_keys.pop(done, None) try: done.result() except asyncio.CancelledError: pass except Exception: logger.debug("Background memory extraction task failed", exc_info=True) task.add_done_callback(_done) return task async def drain( self, timeout_s: float = 60.0, *, context: Any | None = None, scope_key: str | None = None, ) -> int: """Wait for in-flight extractions with a soft timeout.""" scope_key = scope_key or maybe_memory_task_scope_key(context) tasks = [ task for task in self._in_flight if scope_key is None or self._task_scope_keys.get(task) == scope_key ] if not tasks: return 0 done, _pending = await asyncio.wait( tasks, timeout=max(0.0, timeout_s), ) for task in done: try: task.result() except asyncio.CancelledError: pass except Exception: logger.debug("Pending memory extraction failed during drain", exc_info=True) return len(_pending) async def _execute_impl( self, context: ToolUseContext, append_system_message: AppendSystemMessageFn | None, ) -> MemoryExtractionResult: if _should_skip_context(context): await _emit_memory_extraction_skipped(context, "subagent") return MemoryExtractionResult(skipped_reason="subagent") feature_enabled = _extract_feature_enabled() non_interactive = not bool(getattr(context, "tui_available", False)) if not is_extract_mode_active( feature_enabled=feature_enabled, non_interactive=non_interactive, allow_non_interactive=_extract_allow_non_interactive(), ): await _emit_memory_extraction_skipped(context, "disabled") return MemoryExtractionResult(skipped_reason="disabled") if not is_auto_memory_enabled(): await _emit_memory_extraction_skipped(context, "auto_memory_disabled") return MemoryExtractionResult(skipped_reason="auto_memory_disabled") state = self._runtime_state(context) if state.in_progress: state.pending_context = (context, append_system_message) await context.emit_event("memory_extraction_coalesced", {}) await _emit_memory_extraction_skipped(context, "coalesced") return MemoryExtractionResult(skipped_reason="coalesced") return await self._run_extraction_chain(context, append_system_message, state) async def _run_extraction_chain( self, context: ToolUseContext, append_system_message: AppendSystemMessageFn | None, state: MemoryExtractorRuntimeState, *, is_trailing_run: bool = False, ) -> MemoryExtractionResult: state.in_progress = True try: result = await self._run_extraction( context, append_system_message, state, is_trailing_run=is_trailing_run, ) finally: state.in_progress = False trailing = state.pending_context state.pending_context = None if trailing is not None: await context.emit_event("memory_extraction_trailing_start", {}) await self._run_extraction_chain( trailing[0], trailing[1], state, is_trailing_run=True, ) return result async def _run_extraction( self, context: ToolUseContext, append_system_message: AppendSystemMessageFn | None, state: MemoryExtractorRuntimeState, *, is_trailing_run: bool, ) -> MemoryExtractionResult: start_time = time.time() messages = list(context.messages or []) memory_dir = get_auto_mem_path(cwd=context.cwd) new_message_count = count_model_visible_messages_since( messages, state.last_cursor, ) if new_message_count <= 0: await _emit_memory_extraction_skipped( context, "no_new_messages", memory_dir=str(memory_dir), message_count=new_message_count, ) return MemoryExtractionResult( skipped_reason="no_new_messages", new_message_count=0, ) if has_memory_writes_since(messages, state.last_cursor, memory_dir=memory_dir): self._advance_cursor(messages, state) await context.emit_event( "memory_extraction_skipped", { "task_id": _background_task_id(context, "extract_memories"), "reason": "main_agent_memory_write", "message_count": new_message_count, }, ) return MemoryExtractionResult( skipped_reason="main_agent_memory_write", new_message_count=new_message_count, ) if not is_trailing_run: state.turns_since_last_extraction += 1 required_turns = self.throttle_turns or _extract_every_n_turns() if state.turns_since_last_extraction < required_turns: await _emit_memory_extraction_skipped( context, "throttled", memory_dir=str(memory_dir), message_count=new_message_count, required_turns=required_turns, turns_since_last_extraction=state.turns_since_last_extraction, ) return MemoryExtractionResult( skipped_reason="throttled", new_message_count=new_message_count, ) state.turns_since_last_extraction = 0 llm_client = context.llm_client if llm_client is None or not hasattr(llm_client, "call_model"): await _emit_memory_extraction_skipped( context, "missing_llm_client", memory_dir=str(memory_dir), message_count=new_message_count, ) return MemoryExtractionResult( skipped_reason="missing_llm_client", new_message_count=new_message_count, ) ensure_memory_dir_exists(memory_dir) existing_memories = format_memory_manifest(scan_memory_files(memory_dir)) memory_mode = get_memory_mode(getattr(context, "memory_mode", None)) daily_log_mode = memory_mode == "daily_log" if daily_log_mode: prompt = build_extract_daily_log_prompt( new_message_count, existing_memories, ) else: prompt = build_extract_auto_only_prompt( new_message_count, existing_memories, skip_index=_extract_skip_index(), ) tools = _select_auto_mem_tools( context, memory_dir, daily_log_mode=daily_log_mode, ) if not tools: await _emit_memory_extraction_skipped( context, "missing_tools", memory_dir=str(memory_dir), message_count=new_message_count, memory_mode=memory_mode, ) return MemoryExtractionResult( skipped_reason="missing_tools", new_message_count=new_message_count, ) can_use_tool = create_auto_mem_can_use_tool(memory_dir) task_id = _background_task_id(context, "extract_memories") await context.emit_event( "memory_extraction_start", { "task_id": task_id, "message_count": new_message_count, "memory_mode": memory_mode, "memory_dir": str(memory_dir), "is_trailing_run": is_trailing_run, }, ) try: model_override = os.environ.get(OPENSPACE_MEMORY_EXTRACT_MODEL_ENV) or None side_result = await run_side_query( prompt, tools=tools, model=model_override, parent_context=context, llm_client=llm_client, messages=messages, max_turns=self.max_turns, can_use_tool=can_use_tool, query_source="extract_memories", fork_label="extract_memories", agent_type="extract_memories", denied_result_type="auto_mem_tool_denied", tui_available=False, is_async_agent=True, ) result_messages = side_result.messages total_usage = side_result.total_usage turn_count = side_result.turn_count self._advance_cursor(messages, state) written_paths = extract_written_paths(result_messages, memory_dir=memory_dir) logged_entry_ids, logged_paths = extract_logged_entries(result_messages) if daily_log_mode: written_paths = logged_paths memory_paths = [ path for path in written_paths if Path(path).name != ENTRYPOINT_NAME ] if daily_log_mode: memory_paths = [] duration_ms = (time.time() - start_time) * 1000 extraction_result = MemoryExtractionResult( ran=True, new_message_count=new_message_count, written_paths=written_paths, memory_paths=memory_paths, turn_count=turn_count, duration_ms=duration_ms, ) await context.emit_event( "memory_extraction_complete", { "task_id": task_id, "message_count": new_message_count, "turn_count": turn_count, "files_written": len(written_paths), "memories_saved": len(memory_paths), "input_tokens": total_usage.input_tokens, "output_tokens": total_usage.output_tokens, "duration_ms": duration_ms, "memory_mode": memory_mode, "written_paths": written_paths, "memory_paths": memory_paths, "log_paths": logged_paths, "logged_entry_ids": logged_entry_ids, "logs_written": len(logged_paths), }, ) if memory_paths: await _append_memory_saved_message( context, memory_paths, append_system_message, ) return extraction_result except Exception as exc: duration_ms = (time.time() - start_time) * 1000 logger.debug("[extractMemories] error: %s", exc, exc_info=True) await context.emit_event( "memory_extraction_error", { "task_id": task_id, "memory_dir": str(memory_dir), "duration_ms": duration_ms, "error": str(exc), }, ) return MemoryExtractionResult( ran=True, new_message_count=new_message_count, turn_count=turn_count, duration_ms=duration_ms, error=str(exc), ) def _advance_cursor( self, messages: Sequence[Mapping[str, Any]], state: MemoryExtractorRuntimeState, ) -> None: if not messages: return state.last_cursor = _cursor_for_index(messages, len(messages) - 1) def _should_skip_context(context: ToolUseContext) -> bool: if getattr(context, "is_async_agent", False): return True if getattr(context, "parent_task_id", None): return True if getattr(context, "agent_type", None) in {"extract_memories", "auto_dream"}: return True return False def _background_task_id(context: ToolUseContext, key: str) -> str | None: task_ids = getattr(context, "background_task_ids", None) if isinstance(task_ids, dict): value = task_ids.get(key) return str(value) if value else None return None async def _emit_memory_extraction_skipped( context: ToolUseContext, reason: str, **extra: Any, ) -> None: await context.emit_event( "memory_extraction_skipped", { "task_id": _background_task_id(context, "extract_memories"), "reason": reason, **extra, }, ) def should_schedule_extract_memories(context: ToolUseContext) -> bool: """Cheap gate used by stop hooks before creating a background task.""" if _should_skip_context(context): return False if not is_auto_memory_enabled(): return False feature_enabled = _extract_feature_enabled() non_interactive = not bool(getattr(context, "tui_available", False)) return is_extract_mode_active( feature_enabled=feature_enabled, non_interactive=non_interactive, allow_non_interactive=_extract_allow_non_interactive(), ) def _select_auto_mem_tools( context: ToolUseContext, memory_dir: str | Path, *, daily_log_mode: bool = False, ) -> list[BaseTool]: if daily_log_mode: allowed = { FILE_READ_TOOL_NAME, GREP_TOOL_NAME, GLOB_TOOL_NAME, BASH_TOOL_NAME, MEMORY_READ_TOOL_NAME, MEMORY_LOG_TOOL_NAME, } else: allowed = { FILE_READ_TOOL_NAME, GREP_TOOL_NAME, GLOB_TOOL_NAME, BASH_TOOL_NAME, FILE_EDIT_TOOL_NAME, FILE_WRITE_TOOL_NAME, MEMORY_READ_TOOL_NAME, MEMORY_WRITE_TOOL_NAME, } selected: list[BaseTool] = [] seen: set[str] = set() for tool in [*(context.all_tools or []), *(context.tools or [])]: name = getattr(tool, "name", "") if name in allowed and name not in seen: selected.append(tool) seen.add(name) # OpenSpace-specific safe default: if the main tool set omitted the memory # helpers, use them as a scoped fallback. OpenSpace lacks these dedicated tools; # OS added them in 15.5 to avoid fragile hand-written frontmatter/index # edits while preserving the same memdir storage shape. try: if daily_log_mode and MEMORY_LOG_TOOL_NAME not in seen: selected.append(MemoryLogTool()) seen.add(MEMORY_LOG_TOOL_NAME) if not daily_log_mode and MEMORY_WRITE_TOOL_NAME not in seen: from openspace.tools.memory_tools import MemoryWriteTool selected.append(MemoryWriteTool()) seen.add(MEMORY_WRITE_TOOL_NAME) if MEMORY_READ_TOOL_NAME not in seen: from openspace.tools.memory_tools import MemoryReadTool selected.append(MemoryReadTool()) seen.add(MEMORY_READ_TOOL_NAME) except Exception: pass return selected def _tool_call_name(tool_call: Mapping[str, Any]) -> str: function = tool_call.get("function") if isinstance(function, Mapping): name = function.get("name") if isinstance(name, str): return name name = tool_call.get("name") return name if isinstance(name, str) else "" def _tool_call_input(tool_call: Mapping[str, Any]) -> dict[str, Any]: function = tool_call.get("function") raw: Any = None if isinstance(function, Mapping): raw = function.get("arguments") elif "input" in tool_call: raw = tool_call.get("input") if isinstance(raw, dict): return dict(raw) if isinstance(raw, str): try: parsed = json.loads(raw) except (json.JSONDecodeError, TypeError): return {} return parsed if isinstance(parsed, dict) else {} return {} def _message_fingerprint(message: Mapping[str, Any]) -> str: meta = message.get("_meta") if isinstance(meta, Mapping): for key in ("uuid", "message_uuid", "id"): value = meta.get(key) if isinstance(value, str) and value: return f"meta:{key}:{value}" for key in ("uuid", "id"): value = message.get(key) if isinstance(value, str) and value: return f"{key}:{value}" payload = { "role": message.get("role"), "content": message.get("content"), "tool_calls": message.get("tool_calls"), "tool_call_id": message.get("tool_call_id"), "name": message.get("name"), } raw = json.dumps(payload, sort_keys=True, default=str, ensure_ascii=False) return "sha256:" + hashlib.sha256(raw.encode("utf-8")).hexdigest() def _cursor_for_index( messages: Sequence[Mapping[str, Any]], index: int, ) -> MemoryCursor: fingerprint = _message_fingerprint(messages[index]) occurrence = 0 for item in messages[: index + 1]: if _message_fingerprint(item) == fingerprint: occurrence += 1 return MemoryCursor(fingerprint=fingerprint, occurrence=occurrence) def _find_cursor_index( messages: Sequence[Mapping[str, Any]], cursor: MemoryCursor | None, ) -> int: if cursor is None: return -1 seen = 0 for index, message in enumerate(messages): if _message_fingerprint(message) == cursor.fingerprint: seen += 1 if seen == cursor.occurrence: return index return -1 def _is_model_visible_message(message: Mapping[str, Any]) -> bool: return message.get("role") in {"user", "assistant"} def count_model_visible_messages_since( messages: Sequence[Mapping[str, Any]], cursor: MemoryCursor | None, ) -> int: start = _find_cursor_index(messages, cursor) return sum(1 for message in messages[start + 1 :] if _is_model_visible_message(message)) def has_memory_writes_since( messages: Sequence[Mapping[str, Any]], cursor: MemoryCursor | None, *, memory_dir: str | Path | None = None, ) -> bool: start = _find_cursor_index(messages, cursor) for message in messages[start + 1 :]: if _message_contains_auto_memory_write(message, memory_dir=memory_dir): return True return False def _message_contains_auto_memory_write( message: Mapping[str, Any], *, memory_dir: str | Path | None, ) -> bool: meta = message.get("_meta") if isinstance(meta, Mapping): if meta.get("tool_name") in {MEMORY_WRITE_TOOL_NAME, MEMORY_LOG_TOOL_NAME}: return True result_meta = meta.get("tool_result_metadata") if isinstance(result_meta, Mapping): if result_meta.get("type") in {"memory_write", "memory_log"}: return True path = result_meta.get("file_path") or result_meta.get("entrypoint_path") if isinstance(path, str) and _is_auto_memory_path(path, memory_dir): return True if message.get("role") == "assistant": for tool_call in _iter_assistant_tool_calls(message): tool_name = _tool_call_name(tool_call) if tool_name in {MEMORY_WRITE_TOOL_NAME, MEMORY_LOG_TOOL_NAME}: return True if tool_name in {FILE_EDIT_TOOL_NAME, FILE_WRITE_TOOL_NAME}: path = _tool_call_input(tool_call).get("file_path") if isinstance(path, str) and _is_auto_memory_path(path, memory_dir): return True return False def _iter_assistant_tool_calls(message: Mapping[str, Any]) -> Iterable[Mapping[str, Any]]: tool_calls = message.get("tool_calls") if isinstance(tool_calls, list): for call in tool_calls: if isinstance(call, Mapping): yield call content = message.get("content") if isinstance(content, list): for block in content: if isinstance(block, Mapping) and block.get("type") == "tool_use": yield block def _is_auto_memory_path(path: str, memory_dir: str | Path | None) -> bool: if memory_dir is not None: return _path_is_inside(path, Path(memory_dir).expanduser().resolve()) return is_auto_mem_path(path) def extract_written_paths( agent_messages: Sequence[Mapping[str, Any]], *, memory_dir: str | Path | None = None, ) -> list[str]: paths: list[str] = [] for message in agent_messages: meta = message.get("_meta") if isinstance(meta, Mapping): result_meta = meta.get("tool_result_metadata") if isinstance(result_meta, Mapping): for key in ("file_path", "entrypoint_path"): path = result_meta.get(key) if isinstance(path, str) and _is_auto_memory_path(path, memory_dir): paths.append(str(Path(path).expanduser().resolve())) if message.get("role") == "assistant": for tool_call in _iter_assistant_tool_calls(message): tool_name = _tool_call_name(tool_call) tool_input = _tool_call_input(tool_call) if tool_name in {FILE_EDIT_TOOL_NAME, FILE_WRITE_TOOL_NAME}: path = tool_input.get("file_path") if isinstance(path, str) and _is_auto_memory_path(path, memory_dir): paths.append(str(Path(path).expanduser().resolve())) elif tool_name == MEMORY_WRITE_TOOL_NAME and memory_dir is not None: filename = tool_input.get("filename") title = tool_input.get("title") or "memory" if isinstance(filename, str) and filename.strip(): candidate = Path(memory_dir) / filename else: slug = "".join( ch.lower() if ch.isalnum() else "_" for ch in str(title).strip() ).strip("_")[:80] or "memory" candidate = Path(memory_dir) / f"{slug}.md" paths.append(str(candidate.expanduser().resolve())) return _uniq(paths) def _uniq(items: Iterable[str]) -> list[str]: seen: set[str] = set() out: list[str] = [] for item in items: if item not in seen: out.append(item) seen.add(item) return out def create_memory_saved_message(memory_paths: Sequence[str]) -> dict[str, Any]: """Build a user-visible system message equivalent to OpenSpace memory saved note.""" paths = [str(path) for path in memory_paths] if len(paths) == 1: content = f"Memory saved: {paths[0]}" else: rendered = "\n".join(f"- {path}" for path in paths) content = f"Memories saved:\n{rendered}" return { "role": "system", "content": content, "_meta": { "type": "memory_saved", "memory_paths": paths, "timestamp": time.time(), }, } async def _append_memory_saved_message( context: ToolUseContext, memory_paths: Sequence[str], append_system_message: AppendSystemMessageFn | None, ) -> None: message = create_memory_saved_message(memory_paths) if append_system_message is not None: result = append_system_message(message) if inspect.isawaitable(result): await result else: context.messages.append(message) await context.emit_event( "memory_saved", {"memory_paths": list(memory_paths), "message": message}, ) _default_extractor: MemoryExtractor = MemoryExtractor() def init_extract_memories(extractor: MemoryExtractor | None = None) -> MemoryExtractor: """Initialize global extraction state, matching OpenSpace ``initExtractMemories``.""" global _default_extractor _default_extractor = extractor or MemoryExtractor() return _default_extractor def get_memory_extractor() -> MemoryExtractor: return _default_extractor async def execute_extract_memories( context: ToolUseContext, append_system_message: AppendSystemMessageFn | None = None, ) -> MemoryExtractionResult: """Run memory extraction. This no-ops until gates pass and is safe to schedule fire-and-forget. """ return await _default_extractor.execute(context, append_system_message) def submit_extract_memories( context: ToolUseContext, append_system_message: AppendSystemMessageFn | None = None, ) -> asyncio.Task[MemoryExtractionResult]: """Schedule background extraction and track it for draining.""" return _default_extractor.submit(context, append_system_message) async def drain_pending_extraction( timeout_s: float = 60.0, *, context: Any | None = None, scope_key: str | None = None, ) -> int: """Await all in-flight extraction tasks with a soft timeout.""" return await _default_extractor.drain( timeout_s, context=context, scope_key=scope_key, ) __all__ = [ "AutoMemCanUseToolFn", "MemoryCursor", "MemoryExtractionResult", "MemoryExtractor", "build_extract_auto_only_prompt", "build_extract_combined_prompt", "count_model_visible_messages_since", "create_auto_mem_can_use_tool", "create_memory_saved_message", "drain_pending_extraction", "execute_extract_memories", "extract_written_paths", "get_memory_extractor", "has_memory_writes_since", "init_extract_memories", "should_schedule_extract_memories", "submit_extract_memories", ]