"""Background auto-memory consolidation ("dream"). OpenSpace consolidates recently written memory through a provider-neutral lightweight tool loop and emits runtime events for UI progress. Implemented behavior: - enabled/auto-memory/subagent gates - time gate using ``.consolidate-lock`` mtime - session-count gate with current-session exclusion - 10-minute scan throttle when time gate passes but session gate does not - live-process lock and rollback on failure - memory-dir-scoped tool gate - best-effort fire-and-forget execution with drain support Current scope: - Progress is exposed through ``auto_dream_*`` runtime events. - Daily-log paths are supported; the prompt also reviews ``logs/`` when present. - User control is exposed through explicit environment variables. """ from __future__ import annotations import asyncio import inspect import json import logging import os import re 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.extract import ( BASH_TOOL_NAME, FILE_EDIT_TOOL_NAME, FILE_READ_TOOL_NAME, FILE_WRITE_TOOL_NAME, GLOB_TOOL_NAME, GREP_TOOL_NAME, MEMORY_READ_TOOL_NAME, MEMORY_WRITE_TOOL_NAME, create_auto_mem_can_use_tool, extract_written_paths, ) from openspace.services.memory.daily_log import ( build_daily_log_consolidation_prompt, get_memory_mode, mark_log_entries_consolidated, scan_unconsolidated_logs, ) from openspace.services.memory.memdir import ( DIR_EXISTS_GUIDANCE, ENTRYPOINT_NAME, MAX_ENTRYPOINT_LINES, ensure_memory_dir_exists, get_auto_mem_path, is_auto_memory_enabled, ) from openspace.services.memory.paths import find_project_root, get_openspace_config_home_dir from openspace.services.memory.memory_scan import format_memory_manifest, scan_memory_files 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__) OPENSPACE_AUTO_DREAM_ENABLED_ENV = "OPENSPACE_AUTO_DREAM_ENABLED" OPENSPACE_DISABLE_AUTO_DREAM_ENV = "OPENSPACE_DISABLE_AUTO_DREAM" OPENSPACE_AUTO_DREAM_MIN_HOURS_ENV = "OPENSPACE_AUTO_DREAM_MIN_HOURS" OPENSPACE_AUTO_DREAM_MIN_SESSIONS_ENV = "OPENSPACE_AUTO_DREAM_MIN_SESSIONS" OPENSPACE_AUTO_DREAM_SCAN_INTERVAL_SECONDS_ENV = ( "OPENSPACE_AUTO_DREAM_SCAN_INTERVAL_SECONDS" ) OPENSPACE_AUTO_DREAM_SESSIONS_DIR_ENV = "OPENSPACE_AUTO_DREAM_SESSIONS_DIR" OPENSPACE_MEMORY_DREAM_MODEL_ENV = "OPENSPACE_MEMORY_DREAM_MODEL" OPENSPACE_REMOTE_ENV = "OPENSPACE_REMOTE" OPENSPACE_KAIROS_ACTIVE_ENV = "OPENSPACE_KAIROS_ACTIVE" DEFAULT_AUTO_DREAM_MIN_HOURS = 24.0 DEFAULT_AUTO_DREAM_MIN_SESSIONS = 5 DEFAULT_SESSION_SCAN_INTERVAL_MS = 10 * 60 * 1000 DEFAULT_MAX_DREAM_TURNS = 5 LOCK_FILE = ".consolidate-lock" HOLDER_STALE_MS = 60 * 60 * 1000 SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_.:-]+$") AppendSystemMessageFn = Callable[[dict[str, Any]], Awaitable[None] | None] @dataclass(frozen=True, slots=True) class AutoDreamConfig: """Scheduling knobs equivalent to OpenSpace ``tengu_onyx_plover`` fields.""" min_hours: float = DEFAULT_AUTO_DREAM_MIN_HOURS min_sessions: int = DEFAULT_AUTO_DREAM_MIN_SESSIONS scan_interval_ms: int = DEFAULT_SESSION_SCAN_INTERVAL_MS model: str | None = None @dataclass(frozen=True, slots=True) class DreamTurn: """A single assistant turn from the dream agent.""" text: str tool_use_count: int @dataclass(slots=True) class DreamTaskState: """OS event-side equivalent of OpenSpace ``DreamTaskState``.""" task_id: str phase: str = "starting" sessions_reviewing: int = 0 files_touched: list[str] = field(default_factory=list) turns: list[DreamTurn] = field(default_factory=list) @dataclass(slots=True) class DreamResult: """Auditable result for tests and event payloads. OpenSpace ``executeAutoDream`` resolves ``void``. OS returns a result when called directly while keeping the stop-hook submit path fire-and-forget. """ ran: bool = False skipped_reason: str | None = None sessions_reviewed: int = 0 files_touched: list[str] = field(default_factory=list) turn_count: int = 0 error: str | None = None duration_ms: float = 0.0 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 is_auto_dream_enabled() -> bool: """Return whether background consolidation is enabled. OpenSpace reads ``settings.json:autoDreamEnabled`` first, then GrowthBook. OS keeps only the canonical ``autoDream.enabled`` setting with env overrides. """ if ( _env_truthy(os.environ.get(OPENSPACE_REMOTE_ENV)) and not os.environ.get("OPENSPACE_REMOTE_MEMORY_DIR") ): return False disabled = os.environ.get(OPENSPACE_DISABLE_AUTO_DREAM_ENV) if _env_truthy(disabled): return False if _env_defined_falsy(disabled): return True explicit = os.environ.get(OPENSPACE_AUTO_DREAM_ENABLED_ENV) if explicit is not None: return _env_truthy(explicit) try: from openspace.services.runtime_support.settings import get_setting return bool(get_setting("autoDream.enabled", False)) except Exception: return False def get_auto_dream_config() -> AutoDreamConfig: """Parse Auto Dream scheduling config with OpenSpace's defensive validation.""" def setting(path: str, default: Any) -> Any: try: from openspace.services.runtime_support.settings import get_setting return get_setting(path, default) except Exception: return default return AutoDreamConfig( min_hours=_parse_positive_float( os.environ.get(OPENSPACE_AUTO_DREAM_MIN_HOURS_ENV), float(setting("autoDream.minHours", DEFAULT_AUTO_DREAM_MIN_HOURS)), ), min_sessions=_parse_positive_int( os.environ.get(OPENSPACE_AUTO_DREAM_MIN_SESSIONS_ENV), int(setting("autoDream.minSessions", DEFAULT_AUTO_DREAM_MIN_SESSIONS)), ), scan_interval_ms=int( _parse_positive_float( os.environ.get(OPENSPACE_AUTO_DREAM_SCAN_INTERVAL_SECONDS_ENV), float( setting( "autoDream.scanIntervalSeconds", DEFAULT_SESSION_SCAN_INTERVAL_MS / 1000, ) ), ) * 1000 ), model=os.environ.get(OPENSPACE_MEMORY_DREAM_MODEL_ENV) or setting("autoDream.model", None), ) def _parse_positive_float(raw: str | None, default: float) -> float: if not raw: return default try: parsed = float(raw) except ValueError: return default return parsed if parsed > 0 and parsed < float("inf") else default def _parse_positive_int(raw: str | None, default: int) -> int: if not raw: return default try: parsed = int(raw) except ValueError: return default return parsed if parsed > 0 else default def build_consolidation_prompt( memory_root: str | Path, transcript_dir: str | Path, extra: str = "", ) -> str: """Build OpenSpace's consolidation prompt with OpenSpace path values.""" prompt = f"""# Dream: Memory Consolidation You are performing a dream - a reflective pass over your memory files. Synthesize what you've learned recently into durable, well-organized memories so that future sessions can orient quickly. Memory directory: `{memory_root}` {DIR_EXISTS_GUIDANCE} Session transcripts: `{transcript_dir}` (large JSONL/NDJSON files - grep narrowly, don't read whole files) --- ## Phase 1 - Orient - `ls` the memory directory to see what already exists - Read `{ENTRYPOINT_NAME}` to understand the current index - Skim existing topic files so you improve them rather than creating duplicates - If `logs/` or `sessions/` subdirectories exist (assistant-mode layout), review recent entries there ## Phase 2 - Gather recent signal Look for new information worth persisting. Sources in rough priority order: 1. **Daily logs** (`logs/YYYY/MM/YYYY-MM-DD.md`) if present - these are the append-only stream 2. **Existing memories that drifted** - facts that contradict something you see in the codebase now 3. **Transcript search** - if you need specific context (e.g., "what was the error message from yesterday's build failure?"), grep the transcripts for narrow terms: `grep -rn "" {transcript_dir}/ --include="*.jsonl" --include="*.messages" | tail -50` Don't exhaustively read transcripts. Look only for things you already suspect matter. ## Phase 3 - Consolidate For each thing worth remembering, write or update a memory file at the top level of the memory directory. Use the memory file format and type conventions from your system prompt's auto-memory section - it's the source of truth for what to save, how to structure it, and what NOT to save. Focus on: - Merging new signal into existing topic files rather than creating near-duplicates - Converting relative dates ("yesterday", "last week") to absolute dates so they remain interpretable after time passes - Deleting contradicted facts - if today's investigation disproves an old memory, fix it at the source ## Phase 4 - Prune and index Update `{ENTRYPOINT_NAME}` so it stays under {MAX_ENTRYPOINT_LINES} lines AND under ~25KB. It's an **index**, not a dump - each entry should be one line under ~150 characters: `- [Title](file.md) - one-line hook`. Never write memory content directly into it. - Remove pointers to memories that are now stale, wrong, or superseded - Demote verbose entries: if an index line is over ~200 chars, it's carrying content that belongs in the topic file - shorten the line, move the detail - Add pointers to newly important memories - Resolve contradictions - if two files disagree, fix the wrong one --- Return a brief summary of what you consolidated, updated, or pruned. If nothing changed (memories are already tight), say so.""" if extra: prompt += f"\n\n## Additional context\n\n{extra}" return prompt def read_last_consolidated_at(memory_dir: str | Path) -> float: """Return lock mtime in milliseconds, or 0 if absent.""" try: return _lock_path(memory_dir).stat().st_mtime * 1000 except OSError: return 0.0 def try_acquire_consolidation_lock(memory_dir: str | Path) -> float | None: """Acquire the Auto Dream lock. Returns the pre-acquire mtime in milliseconds, or ``None`` if another live process owns a fresh lock. """ root = Path(memory_dir).expanduser().resolve() path = _lock_path(root) prior_mtime: float | None = None holder_pid: int | None = None try: stat = path.stat() prior_mtime = stat.st_mtime * 1000 raw = path.read_text(encoding="utf-8").strip() holder_pid = int(raw) if raw else None except (OSError, ValueError): pass if prior_mtime is not None and (time.time() * 1000 - prior_mtime) < HOLDER_STALE_MS: if holder_pid is not None and _is_process_running(holder_pid): logger.debug("[autoDream] lock held by live PID %s", holder_pid) return None root.mkdir(parents=True, exist_ok=True) path.write_text(str(os.getpid()), encoding="utf-8") try: verify = int(path.read_text(encoding="utf-8").strip()) except (OSError, ValueError): return None if verify != os.getpid(): return None return prior_mtime or 0.0 def rollback_consolidation_lock(memory_dir: str | Path, prior_mtime: float) -> None: """Rewind the lock mtime after a failed consolidation.""" path = _lock_path(memory_dir) try: if prior_mtime == 0: path.unlink(missing_ok=True) return path.write_text("", encoding="utf-8") seconds = prior_mtime / 1000 os.utime(path, (seconds, seconds)) except OSError: logger.debug("[autoDream] rollback failed", exc_info=True) def _lock_path(memory_dir: str | Path) -> Path: return Path(memory_dir).expanduser().resolve() / LOCK_FILE def _is_process_running(pid: int) -> bool: if pid <= 0: return False try: os.kill(pid, 0) return True except ProcessLookupError: return False except PermissionError: return True except OSError: return False def get_session_transcript_dir(sessions_dir: str | Path | None = None) -> Path: """Return the directory Auto Dream should scan for session transcripts.""" raw = sessions_dir or os.environ.get(OPENSPACE_AUTO_DREAM_SESSIONS_DIR_ENV) if raw: return Path(raw).expanduser().resolve() return get_openspace_config_home_dir() / "sessions" def list_sessions_touched_since( since_ms: float, *, cwd: str | Path | None = None, sessions_dir: str | Path | None = None, current_session_id: str | None = None, ) -> list[str]: """Return session ids with transcript mtime after ``since_ms``. OpenSpace scans per-project JSONL transcripts. OS stores ``.messages`` plus ``.json`` metadata under ``~/.openspace/sessions``; we use the transcript mtime and filter metadata to the current project when possible. """ root = get_session_transcript_dir(sessions_dir) if not root.exists(): return [] candidates: dict[str, float] = {} for path in root.iterdir(): if path.suffix not in {".messages", ".json"}: continue session_id = path.stem if not session_id or not SESSION_ID_RE.match(session_id): continue if session_id == current_session_id: continue try: mtime_ms = path.stat().st_mtime * 1000 except OSError: continue if mtime_ms <= since_ms: continue if cwd is not None and not _session_matches_project(root, session_id, cwd): continue candidates[session_id] = max(candidates.get(session_id, 0.0), mtime_ms) return [ session_id for session_id, _mtime in sorted( candidates.items(), key=lambda item: item[1], reverse=True, ) ] def _session_matches_project(root: Path, session_id: str, cwd: str | Path) -> bool: meta_path = root / f"{session_id}.json" if not meta_path.exists(): return True try: raw = json.loads(meta_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): return True current = Path(cwd).expanduser().resolve() current_root = find_project_root(current) for key in ("project_path", "workspace_dir", "worktree_path"): value = raw.get(key) if not isinstance(value, str) or not value.strip(): continue try: candidate = Path(value).expanduser().resolve() except OSError: continue if candidate == current or candidate == current_root: return True if find_project_root(candidate) == current_root: return True return False def should_schedule_auto_dream(context: ToolUseContext) -> bool: """Cheap gate used by stop hooks before creating a background task.""" if _should_skip_context(context): return False if not _is_gate_open(): return False return True class AutoDreamer: """Stateful background memory consolidator.""" def __init__( self, *, max_turns: int = DEFAULT_MAX_DREAM_TURNS, config: AutoDreamConfig | None = None, ) -> None: self.max_turns = max(1, int(max_turns)) self.config = config self._last_session_scan_at_by_dir: dict[str, float] = {} self._in_flight: set[asyncio.Task[DreamResult]] = set() self._task_scope_keys: dict[asyncio.Task[DreamResult], str | None] = {} async def execute( self, context: ToolUseContext, append_system_message: AppendSystemMessageFn | None = None, ) -> DreamResult: 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, manual=False, ) finally: if task is not None: self._in_flight.discard(task) # type: ignore[arg-type] self._task_scope_keys.pop(task, None) async def execute_manual( self, context: ToolUseContext, append_system_message: AppendSystemMessageFn | None = None, *, extra_context: str = "", logs_mode: bool = False, ) -> DreamResult: """Run a user-triggered dream. Manual ``/dream`` shares the consolidation lock, memory-scoped tool gate, progress events, and memory-saved notification path with background Auto Dream. It intentionally bypasses only the automatic scheduler gates: AutoDream enabled setting, KAIROS/remote skip, time gate, session-count gate, and scan throttle. """ 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, manual=True, manual_extra_context=extra_context, force_logs_mode=logs_mode, ) 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[DreamResult]: 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[DreamResult]) -> 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 auto dream 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: 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 auto dream failed during drain", exc_info=True) return len(_pending) async def _execute_impl( self, context: ToolUseContext, append_system_message: AppendSystemMessageFn | None, *, manual: bool = False, manual_extra_context: str = "", force_logs_mode: bool = False, ) -> DreamResult: if _should_skip_context(context): await _emit_dream_skipped(context, "subagent", manual=manual) return DreamResult(skipped_reason="subagent") if not manual: if _env_truthy(os.environ.get(OPENSPACE_REMOTE_ENV)): await _emit_dream_skipped(context, "remote_mode", manual=manual) return DreamResult(skipped_reason="remote_mode") if not is_auto_dream_enabled(): await _emit_dream_skipped(context, "disabled", manual=manual) return DreamResult(skipped_reason="disabled") if not is_auto_memory_enabled(): await _emit_dream_skipped(context, "auto_memory_disabled", manual=manual) return DreamResult(skipped_reason="auto_memory_disabled") memory_mode = get_memory_mode(getattr(context, "memory_mode", None)) daily_log_mode = force_logs_mode or memory_mode == "daily_log" cfg = self.config or get_auto_dream_config() memory_dir = get_auto_mem_path(cwd=context.cwd) memory_dir_key = str(Path(memory_dir).expanduser().resolve()) try: last_at = read_last_consolidated_at(memory_dir) except Exception as exc: logger.debug("[autoDream] readLastConsolidatedAt failed: %s", exc) await _emit_dream_skipped( context, "read_last_failed", manual=manual, memory_dir=str(memory_dir), error=str(exc), ) return DreamResult(skipped_reason="read_last_failed", error=str(exc)) hours_since = (time.time() * 1000 - last_at) / 3_600_000 if not manual and hours_since < cfg.min_hours: await _emit_dream_skipped( context, "time_gate", manual=manual, memory_dir=str(memory_dir), hours_since=hours_since, ) return DreamResult(skipped_reason="time_gate") since_scan_ms = ( time.time() * 1000 - self._last_session_scan_at_by_dir.get(memory_dir_key, 0.0) ) if not manual and since_scan_ms < cfg.scan_interval_ms: logger.debug( "[autoDream] scan throttle - last scan was %.1fs ago", since_scan_ms / 1000, ) await _emit_dream_skipped( context, "scan_throttle", manual=manual, memory_dir=str(memory_dir), since_scan_ms=since_scan_ms, ) return DreamResult(skipped_reason="scan_throttle") if not manual: self._last_session_scan_at_by_dir[memory_dir_key] = time.time() * 1000 try: session_ids = list_sessions_touched_since( last_at, cwd=context.cwd, current_session_id=getattr(context, "session_id", None), ) except Exception as exc: logger.debug("[autoDream] listSessionsTouchedSince failed: %s", exc) await _emit_dream_skipped( context, "session_scan_failed", manual=manual, memory_dir=str(memory_dir), error=str(exc), ) return DreamResult(skipped_reason="session_scan_failed", error=str(exc)) if not manual and len(session_ids) < cfg.min_sessions: logger.debug( "[autoDream] skip - %s sessions since last consolidation, need %s", len(session_ids), cfg.min_sessions, ) await _emit_dream_skipped( context, "session_gate", manual=manual, memory_dir=str(memory_dir), sessions_reviewed=len(session_ids), ) return DreamResult( skipped_reason="session_gate", sessions_reviewed=len(session_ids), ) llm_client = context.llm_client if llm_client is None or not hasattr(llm_client, "call_model"): await _emit_dream_skipped( context, "missing_llm_client", manual=manual, memory_dir=str(memory_dir), sessions_reviewed=len(session_ids), ) return DreamResult( skipped_reason="missing_llm_client", sessions_reviewed=len(session_ids), ) tools = _select_dream_tools(context) if not tools: await _emit_dream_skipped( context, "missing_tools", manual=manual, memory_dir=str(memory_dir), sessions_reviewed=len(session_ids), ) return DreamResult( skipped_reason="missing_tools", sessions_reviewed=len(session_ids), ) daily_log_scan = None if daily_log_mode: daily_log_scan = scan_unconsolidated_logs(memory_dir) if not daily_log_scan.entries: await _emit_dream_skipped( context, "no_daily_log_entries", manual=manual, memory_dir=str(memory_dir), sessions_reviewed=len(session_ids), ) return DreamResult( skipped_reason="no_daily_log_entries", sessions_reviewed=len(session_ids), ) prior_mtime = try_acquire_consolidation_lock(memory_dir) if prior_mtime is None: await _emit_dream_skipped( context, "lock_busy", manual=manual, memory_dir=str(memory_dir), sessions_reviewed=len(session_ids), ) return DreamResult( skipped_reason="lock_busy", sessions_reviewed=len(session_ids), ) start_time = time.time() task_state = DreamTaskState( task_id=( _background_task_id(context, "auto_dream") or f"dream-{int(start_time * 1000)}" ), sessions_reviewing=len(session_ids), ) await context.emit_event( "auto_dream_start", { "task_id": task_state.task_id, "sessions_reviewing": len(session_ids), "memory_dir": str(memory_dir), "manual": manual, }, ) try: ensure_memory_dir_exists(memory_dir) transcript_dir = get_session_transcript_dir() extra = _build_auto_dream_extra( session_ids, manual=manual, extra_context=manual_extra_context, ) if daily_log_mode and daily_log_scan is not None: manifest = format_memory_manifest(scan_memory_files(memory_dir)) daily_extra = build_daily_log_consolidation_prompt( memory_dir, daily_log_scan.log_paths, manifest, daily_log_scan.entries, ) extra = "\n\n".join(part for part in (extra, daily_extra) if part) prompt = build_consolidation_prompt(memory_dir, transcript_dir, extra) can_use_tool = create_auto_mem_can_use_tool(memory_dir) result_messages: list[dict[str, Any]] = [] turn_count = 0 async def on_side_message( message: dict[str, Any], _side_context: Any, ) -> None: if message.get("role") != "assistant": return await _record_dream_progress( task_state, message, memory_dir=memory_dir, context=context, ) model_override = os.environ.get(OPENSPACE_MEMORY_DREAM_MODEL_ENV) or cfg.model side_result = await run_side_query( prompt, tools=tools, model=model_override, parent_context=context, llm_client=llm_client, messages=list(context.messages or []), max_turns=self.max_turns, can_use_tool=can_use_tool, query_source="auto_dream", fork_label="auto_dream", agent_type="auto_dream", denied_result_type="auto_dream_tool_denied", tui_available=False, is_async_agent=True, on_message=on_side_message, ) if side_result.aborted: raise asyncio.CancelledError() result_messages = side_result.messages total_usage = side_result.total_usage turn_count = side_result.turn_count touched_paths = _uniq( [ *task_state.files_touched, *extract_written_paths(result_messages, memory_dir=memory_dir), ] ) log_entry_ids: list[str] = [] if daily_log_mode and daily_log_scan is not None: log_entry_ids = daily_log_scan.entry_ids mark_log_entries_consolidated( memory_dir, log_entry_ids, consolidated_to=touched_paths or "dropped", ) duration_ms = (time.time() - start_time) * 1000 await context.emit_event( "auto_dream_complete", { "task_id": task_state.task_id, "sessions_reviewed": len(session_ids), "files_touched": touched_paths, "turn_count": turn_count, "input_tokens": total_usage.input_tokens, "output_tokens": total_usage.output_tokens, "duration_ms": duration_ms, "source": "daily_log" if daily_log_mode else "direct", "manual": manual, "log_entry_ids": log_entry_ids, }, ) if touched_paths: await _append_memory_improved_message( context, touched_paths, append_system_message, source="daily_log" if daily_log_mode else "direct", log_entry_ids=log_entry_ids, ) return DreamResult( ran=True, sessions_reviewed=len(session_ids), files_touched=touched_paths, turn_count=turn_count, duration_ms=duration_ms, ) except asyncio.CancelledError: rollback_consolidation_lock(memory_dir, prior_mtime) await context.emit_event( "auto_dream_cancelled", {"task_id": task_state.task_id, "manual": manual}, ) return DreamResult( ran=True, skipped_reason="aborted", sessions_reviewed=len(session_ids), files_touched=list(task_state.files_touched), turn_count=len(task_state.turns), duration_ms=(time.time() - start_time) * 1000, ) except Exception as exc: rollback_consolidation_lock(memory_dir, prior_mtime) duration_ms = (time.time() - start_time) * 1000 logger.debug("[autoDream] fork failed: %s", exc, exc_info=True) await context.emit_event( "auto_dream_error", { "task_id": task_state.task_id, "sessions_reviewed": len(session_ids), "duration_ms": duration_ms, "error": str(exc), "manual": manual, }, ) return DreamResult( ran=True, sessions_reviewed=len(session_ids), files_touched=list(task_state.files_touched), turn_count=len(task_state.turns), duration_ms=duration_ms, error=str(exc), ) 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_dream_skipped( context: ToolUseContext, reason: str, *, manual: bool, **extra: Any, ) -> None: event_type = "manual_dream_skipped" if manual else "auto_dream_skipped" await context.emit_event( event_type, { "task_id": _background_task_id(context, "auto_dream"), "reason": reason, "manual": manual, **extra, }, ) def _is_gate_open() -> bool: """OpenSpace ``isGateOpen`` equivalent with OS env-backed state.""" if _env_truthy(os.environ.get(OPENSPACE_REMOTE_ENV)): return False if not is_auto_memory_enabled(): return False return is_auto_dream_enabled() def _build_auto_dream_extra( session_ids: Sequence[str], *, manual: bool = False, extra_context: str = "", ) -> str: lines = [ "", "**Tool constraints for this run:** Bash is restricted to read-only commands (`ls`, `find`, `grep`, `cat`, `stat`, `wc`, `head`, `tail`, and similar). Anything that writes, redirects to a file, or modifies state will be denied. Plan your exploration with this in mind - no need to probe.", "", ] if manual: lines.append("This dream was explicitly triggered by the user, so do a focused consolidation pass even if automatic scheduling thresholds have not been met.") lines.append("") lines.extend( [ f"Sessions since last consolidation ({len(session_ids)}):", *[f"- {session_id}" for session_id in session_ids], ] ) if extra_context.strip(): lines.extend(["", "User-supplied context for this dream:", extra_context.strip()]) return "\n".join(lines) def _select_dream_tools(context: ToolUseContext) -> list[BaseTool]: 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) try: if 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 async def _record_dream_progress( task_state: DreamTaskState, assistant_message: Mapping[str, Any], *, memory_dir: str | Path, context: ToolUseContext, ) -> None: text = "" tool_use_count = 0 touched_paths: list[str] = [] content = assistant_message.get("content") if isinstance(content, str): text += content elif isinstance(content, list): for block in content: if not isinstance(block, Mapping): continue if block.get("type") == "text": raw = block.get("text") if isinstance(raw, str): text += raw elif block.get("type") == "tool_use": tool_use_count += 1 touched_paths.extend( _touched_paths_from_tool_call(block, memory_dir=memory_dir) ) tool_calls = assistant_message.get("tool_calls") if isinstance(tool_calls, list): for call in tool_calls: if isinstance(call, Mapping): tool_use_count += 1 touched_paths.extend( _touched_paths_from_tool_call(call, memory_dir=memory_dir) ) touched_paths = _uniq(touched_paths) if not text.strip() and tool_use_count == 0 and not touched_paths: return task_state.turns.append(DreamTurn(text=text.strip(), tool_use_count=tool_use_count)) if touched_paths: task_state.phase = "updating" task_state.files_touched = _uniq([*task_state.files_touched, *touched_paths]) task_state.turns = task_state.turns[-30:] await context.emit_event( "auto_dream_progress", { "task_id": task_state.task_id, "phase": task_state.phase, "turn": { "text": text.strip(), "tool_use_count": tool_use_count, }, "files_touched": list(task_state.files_touched), }, ) def _touched_paths_from_tool_call( tool_call: Mapping[str, Any], *, memory_dir: str | Path, ) -> list[str]: 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 _path_is_inside(path, Path(memory_dir)): return [str(Path(path).expanduser().resolve())] if tool_name == MEMORY_WRITE_TOOL_NAME: 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" return [str(candidate.expanduser().resolve())] return [] 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 _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.expanduser().resolve()) return True except (OSError, ValueError): return False async def _append_memory_improved_message( context: ToolUseContext, memory_paths: Sequence[str], append_system_message: AppendSystemMessageFn | None, *, source: str = "direct", log_entry_ids: Sequence[str] | None = None, ) -> None: paths = [str(path) for path in memory_paths] if len(paths) == 1: content = f"Memory improved: {paths[0]}" else: rendered = "\n".join(f"- {path}" for path in paths) content = f"Memories improved:\n{rendered}" message = { "role": "system", "content": content, "_meta": { "type": "memory_improved", "memory_paths": paths, "source": source, "log_entry_ids": list(log_entry_ids or []), "timestamp": time.time(), }, } 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": paths, "message": message, "verb": "Improved", "source": source, "log_entry_ids": list(log_entry_ids or []), }, ) 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 _default_dreamer: AutoDreamer = AutoDreamer() def init_auto_dream(dreamer: AutoDreamer | None = None) -> AutoDreamer: """Initialize global Auto Dream state, matching OpenSpace ``initAutoDream``.""" global _default_dreamer _default_dreamer = dreamer or AutoDreamer() return _default_dreamer def get_auto_dreamer() -> AutoDreamer: return _default_dreamer async def execute_auto_dream( context: ToolUseContext, append_system_message: AppendSystemMessageFn | None = None, ) -> DreamResult: return await _default_dreamer.execute(context, append_system_message) async def execute_manual_auto_dream( context: ToolUseContext, append_system_message: AppendSystemMessageFn | None = None, *, extra_context: str = "", logs_mode: bool = False, ) -> DreamResult: return await _default_dreamer.execute_manual( context, append_system_message, extra_context=extra_context, logs_mode=logs_mode, ) def submit_auto_dream( context: ToolUseContext, append_system_message: AppendSystemMessageFn | None = None, ) -> asyncio.Task[DreamResult]: return _default_dreamer.submit(context, append_system_message) async def drain_pending_auto_dream( timeout_s: float = 60.0, *, context: Any | None = None, scope_key: str | None = None, ) -> int: return await _default_dreamer.drain( timeout_s, context=context, scope_key=scope_key, ) __all__ = [ "AutoDreamConfig", "AutoDreamer", "DreamResult", "DreamTaskState", "DreamTurn", "build_consolidation_prompt", "drain_pending_auto_dream", "execute_auto_dream", "execute_manual_auto_dream", "get_auto_dream_config", "get_auto_dreamer", "get_session_transcript_dir", "init_auto_dream", "is_auto_dream_enabled", "list_sessions_touched_since", "read_last_consolidated_at", "rollback_consolidation_lock", "should_schedule_auto_dream", "submit_auto_dream", "try_acquire_consolidation_lock", ]