feat(memory): add auto-dream consolidation

ohmo/gateway/runtime.py

@@ -39,7 +39,7 @@ from ohmo.group_registry import load_managed_group_record, normalize_cwd
 from ohmo.memory import create_memory_command_backend
 from ohmo.prompts import build_ohmo_system_prompt
 from ohmo.session_storage import OhmoSessionBackend
-from ohmo.workspace import get_plugins_dir, get_skills_dir, initialize_workspace
+from ohmo.workspace import get_memory_dir, get_plugins_dir, get_sessions_dir, get_skills_dir, initialize_workspace
 
 logger = logging.getLogger(__name__)
 
@@ -194,6 +194,12 @@ class OhmoSessionRuntimePool:
             extra_plugin_roots=(str(get_plugins_dir(self._workspace)),),
             memory_backend=create_memory_command_backend(self._workspace),
             include_project_memory=False,
+            autodream_context={
+                "memory_dir": str(get_memory_dir(self._workspace)),
+                "session_dir": str(get_sessions_dir(self._workspace)),
+                "app_label": "ohmo personal memory",
+                "runner_module": "ohmo",
+            },
         )
         if snapshot and snapshot.get("session_id"):
             bundle.session_id = str(snapshot["session_id"])
@@ -643,6 +649,12 @@ class OhmoSessionRuntimePool:
             extra_plugin_roots=(str(get_plugins_dir(self._workspace)),),
             memory_backend=create_memory_command_backend(self._workspace),
             include_project_memory=False,
+            autodream_context={
+                "memory_dir": str(get_memory_dir(self._workspace)),
+                "session_dir": str(get_sessions_dir(self._workspace)),
+                "app_label": "ohmo personal memory",
+                "runner_module": "ohmo",
+            },
         )
         refreshed.session_id = prior_session_id
         self._register_gateway_tools(refreshed)

ohmo/runtime.py

@@ -17,7 +17,7 @@ from openharness.ui.react_launcher import _resolve_npm, _resolve_tsx, get_fronte
 from ohmo.memory import create_memory_command_backend
 from ohmo.prompts import build_ohmo_system_prompt
 from ohmo.session_storage import OhmoSessionBackend
-from ohmo.workspace import get_plugins_dir, get_skills_dir, initialize_workspace
+from ohmo.workspace import get_memory_dir, get_plugins_dir, get_sessions_dir, get_skills_dir, initialize_workspace
 
 
 def _ohmo_extra_roots(workspace: str | Path | None) -> tuple[tuple[str, ...], tuple[str, ...]]:
@@ -57,6 +57,12 @@ async def run_ohmo_backend(
         extra_plugin_roots=extra_plugin_roots,
         memory_backend=create_memory_command_backend(workspace_root),
         include_project_memory=False,
+        autodream_context={
+            "memory_dir": str(get_memory_dir(workspace_root)),
+            "session_dir": str(get_sessions_dir(workspace_root)),
+            "app_label": "ohmo personal memory",
+            "runner_module": "ohmo",
+        },
     )
 
 
@@ -165,6 +171,12 @@ async def run_ohmo_print_mode(
             extra_plugin_roots=extra_plugin_roots,
             memory_backend=create_memory_command_backend(workspace_root),
             include_project_memory=False,
+            autodream_context={
+                "memory_dir": str(get_memory_dir(workspace_root)),
+                "session_dir": str(get_sessions_dir(workspace_root)),
+                "app_label": "ohmo personal memory",
+                "runner_module": "ohmo",
+            },
         )
         await start_runtime(bundle)
 

src/openharness/commands/registry.py

@@ -50,6 +50,14 @@ from openharness.services import (
     estimate_conversation_tokens,
     summarize_messages,
 )
+from openharness.services.autodream import (
+    diff_memory_dirs,
+    format_memory_diff,
+    latest_memory_backup,
+    read_last_consolidated_at,
+    restore_memory_backup,
+    start_dream_now,
+)
 from openharness.services.session_backend import DEFAULT_SESSION_BACKEND, SessionBackend
 from openharness.skills import load_skill_registry
 from openharness.skills.types import SkillDefinition
@@ -543,6 +551,97 @@ def create_default_command_registry(
             )
         )
 
+    async def _dream_handler(args: str, context: CommandContext) -> CommandResult:
+        settings = getattr(context.engine, "_settings", None) or load_settings().materialize_active_profile()
+        parts = args.split()
+        action = parts[0] if parts else "run"
+        backend = context.memory_backend if context.memory_backend is not None else None
+        memory_dir = backend.get_memory_dir() if backend is not None else get_project_memory_dir(context.cwd)
+        session_dir = context.session_backend.get_session_dir(context.cwd)
+        app_label = backend.label if backend is not None else "openharness project memory"
+        runner_module = "ohmo" if backend is not None and "ohmo" in backend.label.lower() else "openharness"
+        if action == "status":
+            last_at = read_last_consolidated_at(context.cwd, memory_dir=memory_dir)
+            last = "never" if last_at <= 0 else datetime.fromtimestamp(last_at).isoformat(timespec="seconds")
+            return CommandResult(
+                message=(
+                    f"Auto-dream: {'on' if settings.memory.auto_dream_enabled else 'off'}\n"
+                    f"Memory store: {app_label}\n"
+                    f"Memory directory: {memory_dir}\n"
+                    f"Session directory: {session_dir}\n"
+                    f"Last consolidated: {last}"
+                )
+            )
+        if action == "diff":
+            target = parts[1] if len(parts) > 1 else "latest"
+            task_memory_dir = str(memory_dir)
+            backup_dir = ""
+            label = target
+            if target == "latest":
+                latest = latest_memory_backup(memory_dir, app_label=app_label)
+                backup_dir = str(latest) if latest is not None else ""
+                label = "latest backup"
+            else:
+                task = get_task_manager().get_task(target)
+                if task is not None and task.type == "dream":
+                    backup_dir = task.metadata.get("backup_dir", "")
+                    task_memory_dir = task.metadata.get("memory_dir", str(memory_dir))
+                    label = task.id
+            if not backup_dir:
+                return CommandResult(message=f"No dream backup found for {target}.")
+            diff = diff_memory_dirs(backup_dir, task_memory_dir)
+            return CommandResult(
+                message=(
+                    f"Dream diff for {label}:\n"
+                    f"Backup: {backup_dir}\n"
+                    f"Memory: {task_memory_dir}\n"
+                    f"{format_memory_diff(diff)}"
+                )
+            )
+        if action == "rollback":
+            target = parts[1] if len(parts) > 1 else "latest"
+            task_memory_dir = str(memory_dir)
+            backup_dir = ""
+            label = target
+            if target == "latest":
+                latest = latest_memory_backup(memory_dir, app_label=app_label)
+                backup_dir = str(latest) if latest is not None else ""
+                label = "latest backup"
+            else:
+                task = get_task_manager().get_task(target)
+                if task is not None and task.type == "dream":
+                    backup_dir = task.metadata.get("backup_dir", "")
+                    task_memory_dir = task.metadata.get("memory_dir", str(memory_dir))
+                    label = task.id
+            if not backup_dir:
+                return CommandResult(message=f"No dream backup found for {target}.")
+            restore_memory_backup(backup_dir, task_memory_dir)
+            return CommandResult(message=f"Rolled back memory from backup for {label}: {backup_dir}")
+        if action not in {"run", "now", "preview"}:
+            return CommandResult(message="Usage: /dream [run|now|preview|status|diff [task_id]|rollback [task_id]]")
+        task = await start_dream_now(
+            cwd=context.cwd,
+            settings=settings,
+            model=context.engine.model,
+            current_session_id=context.session_id,
+            force=True,
+            memory_dir=memory_dir,
+            session_dir=session_dir,
+            app_label=app_label,
+            runner_module=runner_module,
+            preview=action == "preview",
+        )
+        if task is None:
+            return CommandResult(message="Dream did not start. Memory may be disabled or another dream is running.")
+        return CommandResult(
+            message=(
+                f"Started {'preview ' if action == 'preview' else ''}memory consolidation task {task.id}.\n"
+                f"Memory directory: {task.metadata.get('memory_dir', get_project_memory_dir(context.cwd))}\n"
+                f"Backup directory: {task.metadata.get('backup_dir', '') or '(preview/no backup)'}\n"
+                "Use /tasks or task_output to inspect progress. After completion: /dream diff latest or /dream rollback latest."
+            )
+        )
+
     async def _memory_handler(args: str, context: CommandContext) -> CommandResult:
         backend = _memory_backend_for_context(context)
         tokens = args.split(maxsplit=1)
@@ -2117,6 +2216,7 @@ def create_default_command_registry(
     registry.register(SlashCommand("cost", "Show token usage and estimated cost", _cost_handler))
     registry.register(SlashCommand("usage", "Show usage and token estimates", _usage_handler))
     registry.register(SlashCommand("stats", "Show session statistics", _stats_handler))
+    registry.register(SlashCommand("dream", "Consolidate memory", _dream_handler))
     registry.register(SlashCommand("memory", "Inspect and manage project memory", _memory_handler))
     registry.register(SlashCommand("hooks", "Show configured hooks", _hooks_handler))
     registry.register(SlashCommand("resume", "Restore the latest saved session", _resume_handler))

src/openharness/config/settings.py

@@ -65,6 +65,9 @@ class MemorySettings(BaseModel):
     max_entrypoint_lines: int = 200
     context_window_tokens: int | None = None
     auto_compact_threshold_tokens: int | None = None
+    auto_dream_enabled: bool = False
+    auto_dream_min_hours: float = 24.0
+    auto_dream_min_sessions: int = 5
 
 
 class SandboxNetworkSettings(BaseModel):
@@ -555,7 +558,7 @@ class Settings(BaseModel):
     def resolve_profile(self, name: str | None = None) -> tuple[str, ProviderProfile]:
         """Return the active provider profile."""
         profiles = self.merged_profiles()
-        profile_name = (name or self.active_profile or "").strip() or "claude-api"
+        profile_name = (name or self.active_profile or os.environ.get("OPENHARNESS_PROFILE") or "").strip() or "claude-api"
         if profile_name not in profiles:
             fallback_name, fallback = _profile_from_flat_settings(self)
             profiles[fallback_name] = fallback
@@ -592,9 +595,15 @@ class Settings(BaseModel):
         directly before the profile layer is used everywhere.
         """
         profile_name, profile = self.resolve_profile()
-        next_provider = (self.provider or "").strip() or profile.provider
-        next_api_format = (self.api_format or "").strip() or profile.api_format
-        next_base_url = self.base_url if self.base_url is not None else profile.base_url
+        profile_from_env = bool(os.environ.get("OPENHARNESS_PROFILE"))
+        flat_profile_fields_match_profile = profile_from_env or (
+            (self.provider or "").strip() == profile.provider
+            and (self.api_format or "").strip() == profile.api_format
+            and self.base_url == profile.base_url
+        )
+        next_provider = profile.provider if flat_profile_fields_match_profile else (self.provider or "").strip() or profile.provider
+        next_api_format = profile.api_format if flat_profile_fields_match_profile else (self.api_format or "").strip() or profile.api_format
+        next_base_url = profile.base_url if flat_profile_fields_match_profile else (self.base_url if self.base_url is not None else profile.base_url)
         next_context_window_tokens = (
             self.context_window_tokens
             if self.context_window_tokens is not None
@@ -686,6 +695,15 @@ class Settings(BaseModel):
         provider = profile.provider.strip()
         auth_source = profile.auth_source.strip() or default_auth_source_for_provider(provider, profile.api_format)
         if auth_source in {"codex_subscription", "claude_subscription"}:
+            env_auth_token = os.environ.get("ANTHROPIC_AUTH_TOKEN", "").strip()
+            if auth_source == "claude_subscription" and env_auth_token:
+                return ResolvedAuth(
+                    provider=provider,
+                    auth_kind="oauth",
+                    value=env_auth_token,
+                    source="env:ANTHROPIC_AUTH_TOKEN",
+                    state="configured",
+                )
             from openharness.auth.external import (
                 is_third_party_anthropic_endpoint,
                 load_external_credential,
@@ -928,6 +946,9 @@ def load_settings(config_path: Path | None = None) -> Settings:
     if config_path.exists():
         raw = json.loads(config_path.read_text(encoding="utf-8"))
         settings = Settings.model_validate(raw)
+        env_profile = os.environ.get("OPENHARNESS_PROFILE")
+        if env_profile:
+            settings = settings.model_copy(update={"active_profile": env_profile.strip()})
         if "profiles" not in raw or "active_profile" not in raw:
             profile_name, profile = _profile_from_flat_settings(settings)
             merged_profiles = settings.merged_profiles()
@@ -940,7 +961,11 @@ def load_settings(config_path: Path | None = None) -> Settings:
             )
         return _apply_env_overrides(settings.materialize_active_profile())
 
-    return _apply_env_overrides(Settings().materialize_active_profile())
+    settings = Settings()
+    env_profile = os.environ.get("OPENHARNESS_PROFILE")
+    if env_profile:
+        settings = settings.model_copy(update={"active_profile": env_profile.strip()})
+    return _apply_env_overrides(settings.materialize_active_profile())
 
 
 def save_settings(settings: Settings, config_path: Path | None = None) -> None:

src/openharness/engine/query_engine.py

@@ -11,8 +11,10 @@ from openharness.coordinator.coordinator_mode import get_coordinator_user_contex
 from openharness.engine.messages import ConversationMessage, TextBlock, ToolResultBlock
 from openharness.engine.query import AskUserPrompt, PermissionPrompt, QueryContext, remember_user_goal, run_query
 from openharness.engine.stream_events import AssistantTurnComplete, StreamEvent
+from openharness.config.settings import Settings
 from openharness.hooks import HookEvent, HookExecutor
 from openharness.permissions.checker import PermissionChecker
+from openharness.services.autodream.service import schedule_auto_dream
 from openharness.tools.base import ToolRegistry
 
 
@@ -36,6 +38,7 @@ class QueryEngine:
         ask_user_prompt: AskUserPrompt | None = None,
         hook_executor: HookExecutor | None = None,
         tool_metadata: dict[str, object] | None = None,
+        settings: Settings | None = None,
     ) -> None:
         self._api_client = api_client
         self._tool_registry = tool_registry
@@ -51,6 +54,7 @@ class QueryEngine:
         self._ask_user_prompt = ask_user_prompt
         self._hook_executor = hook_executor
         self._tool_metadata = tool_metadata or {}
+        self._settings = settings
         self._messages: list[ConversationMessage] = []
         self._cost_tracker = CostTracker()
 
@@ -129,6 +133,20 @@ class QueryEngine:
         """Replace the in-memory conversation history."""
         self._messages = list(messages)
 
+    def _schedule_auto_dream(self) -> None:
+        """Fire-and-forget background memory consolidation after a user turn."""
+        if self._settings is None:
+            return
+        context = self._tool_metadata.get("autodream_context")
+        kwargs = dict(context) if isinstance(context, dict) else {}
+        schedule_auto_dream(
+            cwd=self._cwd,
+            settings=self._settings,
+            model=self._model,
+            current_session_id=str(self._tool_metadata.get("session_id") or ""),
+            **kwargs,
+        )
+
     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:
@@ -182,12 +200,15 @@ class QueryEngine:
         coordinator_context = self._build_coordinator_context_message()
         if coordinator_context is not None:
             query_messages.append(coordinator_context)
-        async for event, usage in run_query(context, query_messages):
-            if isinstance(event, AssistantTurnComplete):
-                self._messages = list(query_messages)
-            if usage is not None:
-                self._cost_tracker.add(usage)
-            yield event
+        try:
+            async for event, usage in run_query(context, query_messages):
+                if isinstance(event, AssistantTurnComplete):
+                    self._messages = list(query_messages)
+                if usage is not None:
+                    self._cost_tracker.add(usage)
+                yield event
+        finally:
+            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."""

src/openharness/services/autodream/__init__.py

@@ -0,0 +1,34 @@
+"""Automatic memory consolidation (auto-dream)."""
+
+from openharness.services.autodream.backup import (
+    create_memory_backup,
+    diff_memory_dirs,
+    format_memory_diff,
+    latest_memory_backup,
+    restore_memory_backup,
+)
+from openharness.services.autodream.lock import (
+    list_sessions_touched_since,
+    read_last_consolidated_at,
+    record_consolidation,
+    rollback_consolidation_lock,
+    try_acquire_consolidation_lock,
+)
+from openharness.services.autodream.prompt import build_consolidation_prompt
+from openharness.services.autodream.service import execute_auto_dream, start_dream_now
+
+__all__ = [
+    "build_consolidation_prompt",
+    "create_memory_backup",
+    "diff_memory_dirs",
+    "execute_auto_dream",
+    "format_memory_diff",
+    "latest_memory_backup",
+    "list_sessions_touched_since",
+    "read_last_consolidated_at",
+    "record_consolidation",
+    "restore_memory_backup",
+    "rollback_consolidation_lock",
+    "start_dream_now",
+    "try_acquire_consolidation_lock",
+]

src/openharness/services/autodream/backup.py

@@ -0,0 +1,104 @@
+"""Backup, diff, and rollback helpers for auto-dream memory directories."""
+
+from __future__ import annotations
+
+import filecmp
+import shutil
+import time
+from pathlib import Path
+
+from openharness.config.paths import get_data_dir
+
+
+def default_backup_root(memory_dir: str | Path, *, app_label: str = "openharness") -> Path:
+    """Return the backup root for a memory directory."""
+
+    memory_dir = Path(memory_dir).expanduser().resolve()
+    if ".ohmo" in memory_dir.parts:
+        try:
+            idx = memory_dir.parts.index(".ohmo")
+            return Path(*memory_dir.parts[: idx + 1]) / "backups"
+        except ValueError:
+            pass
+    safe_label = "".join(ch if ch.isalnum() or ch in {"-", "_"} else "-" for ch in app_label).strip("-")
+    return get_data_dir() / "memory-backups" / (safe_label or "openharness")
+
+
+def create_memory_backup(
+    memory_dir: str | Path,
+    *,
+    backup_root: str | Path | None = None,
+    app_label: str = "openharness",
+) -> Path:
+    """Create a timestamped copy of ``memory_dir`` and return the backup path."""
+
+    memory_dir = Path(memory_dir).expanduser().resolve()
+    root = Path(backup_root).expanduser().resolve() if backup_root is not None else default_backup_root(memory_dir, app_label=app_label)
+    root.mkdir(parents=True, exist_ok=True)
+    timestamp = time.strftime("memory-%Y%m%d-%H%M%S")
+    backup = root / timestamp
+    suffix = 1
+    while backup.exists():
+        suffix += 1
+        backup = root / f"{timestamp}-{suffix}"
+    if memory_dir.exists():
+        shutil.copytree(memory_dir, backup, ignore=shutil.ignore_patterns(".consolidate-lock"))
+    else:
+        backup.mkdir(parents=True)
+    return backup
+
+
+def diff_memory_dirs(before: str | Path, after: str | Path) -> dict[str, list[str]]:
+    """Return added/removed/changed file names between two memory dirs."""
+
+    before = Path(before).expanduser().resolve()
+    after = Path(after).expanduser().resolve()
+    before_files = {p.name: p for p in before.glob("*.md")} if before.exists() else {}
+    after_files = {p.name: p for p in after.glob("*.md")} if after.exists() else {}
+    added = sorted(set(after_files) - set(before_files))
+    removed = sorted(set(before_files) - set(after_files))
+    changed = sorted(
+        name
+        for name in set(before_files) & set(after_files)
+        if not filecmp.cmp(before_files[name], after_files[name], shallow=False)
+    )
+    return {"added": added, "removed": removed, "changed": changed}
+
+
+def format_memory_diff(diff: dict[str, list[str]]) -> str:
+    """Format a compact memory diff summary."""
+
+    lines: list[str] = []
+    for label in ("added", "changed", "removed"):
+        values = diff.get(label, [])
+        if values:
+            lines.append(f"{label}: " + ", ".join(values))
+    return "\n".join(lines) if lines else "no markdown file changes"
+
+
+def latest_memory_backup(memory_dir: str | Path, *, app_label: str = "openharness") -> Path | None:
+    """Return the latest backup for a memory directory, if any."""
+
+    root = default_backup_root(memory_dir, app_label=app_label)
+    if not root.exists():
+        return None
+    backups = [path for path in root.iterdir() if path.is_dir() and path.name.startswith("memory-")]
+    if not backups:
+        return None
+    return max(backups, key=lambda path: path.stat().st_mtime)
+
+
+def restore_memory_backup(backup_dir: str | Path, memory_dir: str | Path) -> None:
+    """Restore memory_dir from a backup directory."""
+
+    backup_dir = Path(backup_dir).expanduser().resolve()
+    memory_dir = Path(memory_dir).expanduser().resolve()
+    if not backup_dir.exists() or not backup_dir.is_dir():
+        raise FileNotFoundError(f"Backup not found: {backup_dir}")
+    tmp = memory_dir.with_name(f".{memory_dir.name}.restore-tmp")
+    if tmp.exists():
+        shutil.rmtree(tmp)
+    shutil.copytree(backup_dir, tmp)
+    if memory_dir.exists():
+        shutil.rmtree(memory_dir)
+    tmp.rename(memory_dir)

src/openharness/services/autodream/lock.py

@@ -0,0 +1,138 @@
+"""Locking and session scanning for auto-dream."""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+
+from openharness.memory.paths import get_project_memory_dir
+from openharness.services.session_storage import get_project_session_dir
+from openharness.utils.fs import atomic_write_text
+
+LOCK_FILE = ".consolidate-lock"
+HOLDER_STALE_SECONDS = 60 * 60
+
+
+def _lock_path(cwd: str | Path, memory_dir: str | Path | None = None) -> Path:
+    return Path(memory_dir) / LOCK_FILE if memory_dir is not None else get_project_memory_dir(cwd) / LOCK_FILE
+
+
+def read_last_consolidated_at(cwd: str | Path, memory_dir: str | Path | None = None) -> float:
+    """Return lock mtime as the last successful consolidation timestamp."""
+
+    try:
+        return _lock_path(cwd, memory_dir).stat().st_mtime
+    except OSError:
+        return 0.0
+
+
+def _holder_pid(path: Path) -> int | None:
+    try:
+        raw = path.read_text(encoding="utf-8").strip()
+        pid = int(raw)
+    except (OSError, ValueError):
+        return None
+    return pid if pid > 0 else None
+
+
+def _is_process_running(pid: int) -> bool:
+    if pid == os.getpid():
+        return True
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    return True
+
+
+def try_acquire_consolidation_lock(cwd: str | Path, memory_dir: str | Path | None = None) -> float | None:
+    """Acquire the consolidation lock and return prior mtime, or None if held."""
+
+    path = _lock_path(cwd, memory_dir)
+    prior_mtime: float | None = None
+    try:
+        stat = path.stat()
+        prior_mtime = stat.st_mtime
+        holder = _holder_pid(path)
+    except OSError:
+        holder = None
+
+    if prior_mtime is not None and time.time() - prior_mtime < HOLDER_STALE_SECONDS:
+        if holder is not None and _is_process_running(holder):
+            return None
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    atomic_write_text(path, f"{os.getpid()}\n")
+    try:
+        if _holder_pid(path) != os.getpid():
+            return None
+    except OSError:
+        return None
+    return prior_mtime or 0.0
+
+
+def rollback_consolidation_lock(
+    cwd: str | Path,
+    prior_mtime: float,
+    memory_dir: str | Path | None = None,
+) -> None:
+    """Restore lock mtime to its pre-acquire value after failed/killed dream."""
+
+    path = _lock_path(cwd, memory_dir)
+    try:
+        if prior_mtime <= 0:
+            path.unlink(missing_ok=True)
+            return
+        atomic_write_text(path, "")
+        os.utime(path, (prior_mtime, prior_mtime))
+    except OSError:
+        # Best effort: a failed rollback only delays the next auto trigger.
+        return
+
+
+def record_consolidation(cwd: str | Path, memory_dir: str | Path | None = None) -> None:
+    """Stamp a manual consolidation time."""
+
+    path = _lock_path(cwd, memory_dir)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    atomic_write_text(path, f"{os.getpid()}\n")
+
+
+def list_sessions_touched_since(
+    cwd: str | Path,
+    since_ts: float,
+    *,
+    current_session_id: str | None = None,
+    session_dir: str | Path | None = None,
+) -> list[str]:
+    """Return saved session IDs whose snapshot files were touched after ``since_ts``."""
+
+    resolved_session_dir = Path(session_dir) if session_dir is not None else get_project_session_dir(cwd)
+    session_ids: list[str] = []
+    seen: set[str] = set()
+    for path in sorted(resolved_session_dir.glob("session-*.json"), key=lambda item: item.stat().st_mtime, reverse=True):
+        try:
+            mtime = path.stat().st_mtime
+        except OSError:
+            continue
+        if mtime <= since_ts:
+            continue
+        session_id = path.stem.removeprefix("session-")
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+            raw_id = payload.get("session_id")
+            if isinstance(raw_id, str) and raw_id.strip():
+                session_id = raw_id.strip()
+        except (OSError, json.JSONDecodeError):
+            pass
+        if current_session_id and session_id == current_session_id:
+            continue
+        if session_id in seen:
+            continue
+        seen.add(session_id)
+        session_ids.append(session_id)
+    return session_ids

src/openharness/services/autodream/prompt.py

@@ -0,0 +1,123 @@
+"""Prompt builder for memory consolidation dreams."""
+
+from __future__ import annotations
+
+from datetime import date
+from pathlib import Path
+
+MAX_ENTRYPOINT_LINES = 200
+ENTRYPOINT_NAME = "MEMORY.md"
+
+
+def build_consolidation_prompt(
+    memory_root: str | Path,
+    session_dir: str | Path,
+    extra: str = "",
+    *,
+    preview: bool = False,
+) -> str:
+    """Build the dream prompt used by manual and automatic memory consolidation."""
+
+    memory_root = Path(memory_root)
+    session_dir = Path(session_dir)
+    extra_section = f"\n\n## Additional context\n\n{extra.strip()}" if extra.strip() else ""
+    write_mode = "PREVIEW MODE: do not write files; propose a concise patch plan only." if preview else "APPLY MODE: update memory files directly when changes are clearly warranted."
+    return f"""# Dream: Memory Consolidation
+
+You are performing a dream — a reflective pass over OpenHarness/ohmo memory files. Synthesize recent signal into durable, well-organized memories so future sessions can orient quickly.
+
+Current date: {date.today().isoformat()}
+Memory directory: `{memory_root}`
+Session snapshots: `{session_dir}` (JSON files can be large; inspect narrowly, do not dump everything)
+Mode: {write_mode}
+
+---
+
+## Non-negotiable memory policy
+
+### Evidence discipline
+
+- Do not infer user mistakes, motives, personality traits, or habits from incidental logs/config.
+- Only record facts directly supported by user statements, repeated behavior, or explicit artifacts.
+- Prefer neutral safety policies over accusations.
+- If a secret appears in context, do not copy it. Record only a generic safety reminder if useful.
+- Never preserve API keys, tokens, app secrets, verification tokens, credential-bearing URLs, or bearer strings.
+
+### Classify every fact before writing
+
+Use these categories:
+
+1. **Stable Preference** — user-stated or repeatedly demonstrated durable preference.
+2. **Durable Project Context** — repo paths, canonical repos, project boundaries, validation commands.
+3. **Recent Snapshot** — active branches, current commits, temporary worktrees, recent test counts. Must include `Last observed: YYYY-MM-DD` and a reminder to verify current state.
+4. **Sensitive/Private Context** — revenue, personal identity, private repos, business metrics. Must include `Privacy: personal/private; do not share externally or in group chats unless explicitly asked.`
+5. **Operational Reminder** — short safety or workflow reminders.
+
+### Staleness and scope
+
+- Short-lived facts must be marked as snapshots, not permanent truths.
+- Prefer updating existing files over creating new ones.
+- Create at most 2 new markdown files in one dream.
+- If a topic is transient, prefer `recent_work.md` or an existing status file over a new topic file.
+- Do not move personal/business context into project memory; keep sensitive personal context in personal memory only.
+
+---
+
+## Phase 1 — Orient
+
+- List the memory directory to see what already exists.
+- Read `{ENTRYPOINT_NAME}` if present; it is the memory index.
+- Skim existing topic files so you update or merge instead of creating duplicates.
+
+## Phase 2 — Gather recent signal
+
+Look for information worth persisting. Sources in rough priority order:
+
+1. Existing memory files that may need updates or contradiction fixes.
+2. Recent session snapshots (`session-*.json`) when you need concrete context.
+3. Focused grep/search terms based on recent work; avoid exhaustive transcript reading.
+
+Skip idle chats, failed retry noise, implementation details that only matter for the current turn, and facts that cannot be supported.
+
+## Phase 3 — Consolidate
+
+For each durable thing worth remembering, write or update concise top-level markdown files in the memory directory.
+
+Focus on:
+- Merging new signal into existing topic files rather than creating near-duplicates.
+- Converting relative dates ("yesterday", "last week") to absolute dates.
+- Correcting or deleting contradicted facts at the source.
+- Keeping memories useful for future sessions, not as raw transcripts.
+- Adding `Last observed` for snapshots and `Privacy` for sensitive/private context.
+
+## Phase 4 — Prune and index
+
+Update `{ENTRYPOINT_NAME}` so it stays under {MAX_ENTRYPOINT_LINES} lines and remains an index, not a content dump.
+
+- Each entry should be one concise line: `- [Title](file.md): one-line hook`.
+- Remove pointers to memories that are stale, wrong, or superseded.
+- Add pointers to newly important memories.
+- Resolve contradictions if multiple files disagree.
+
+## Required final response
+
+Return a structured summary:
+
+```md
+## Dream Summary
+Changed:
+- file.md: what changed
+
+Confidence:
+- High: directly supported facts
+- Medium: recent snapshots that should be verified before use
+- Low: uncertain/stale candidates not written as facts
+
+Privacy:
+- Any private/sensitive context touched and how it is marked
+
+Stale candidates:
+- Items that may need review later
+```
+
+If nothing changed, say so and explain why.{extra_section}"""

src/openharness/services/autodream/service.py

@@ -0,0 +1,304 @@
+"""Auto-dream service."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+import time
+from pathlib import Path
+
+from openharness.config.settings import Settings
+from openharness.memory.paths import get_project_memory_dir
+from openharness.services.autodream.backup import create_memory_backup, diff_memory_dirs
+from openharness.services.autodream.lock import (
+    list_sessions_touched_since,
+    read_last_consolidated_at,
+    rollback_consolidation_lock,
+    try_acquire_consolidation_lock,
+)
+from openharness.services.autodream.prompt import build_consolidation_prompt
+from openharness.services.session_storage import get_project_session_dir
+from openharness.tasks.manager import get_task_manager
+from openharness.tasks.types import TaskRecord
+
+SESSION_SCAN_INTERVAL_SECONDS = 10 * 60
+_CHILD_ENV = "OPENHARNESS_AUTODREAM_CHILD"
+_last_session_scan_at: dict[str, float] = {}
+_listener_registered = False
+
+
+def _enabled(settings: Settings) -> bool:
+    return bool(settings.memory.enabled and settings.memory.auto_dream_enabled)
+
+
+def _has_dream_signal(session_ids: list[str], *, force: bool) -> bool:
+    """Return whether recent sessions are worth consolidating."""
+
+    if force:
+        return True
+    return bool(session_ids)
+
+
+def _memory_files_mtime_snapshot(memory_dir: Path) -> dict[str, float]:
+    snapshot: dict[str, float] = {}
+    for path in memory_dir.glob("*.md"):
+        try:
+            snapshot[path.name] = path.stat().st_mtime
+        except OSError:
+            continue
+    return snapshot
+
+
+def _files_changed_since(memory_dir: Path, before: dict[str, float]) -> list[str]:
+    changed: list[str] = []
+    for path in sorted(memory_dir.glob("*.md")):
+        try:
+            mtime = path.stat().st_mtime
+        except OSError:
+            continue
+        if before.get(path.name) != mtime:
+            changed.append(path.name)
+    return changed
+
+
+def _ensure_listener_registered() -> None:
+    global _listener_registered
+    if _listener_registered:
+        return
+
+    async def _listener(task: TaskRecord) -> None:
+        if task.type != "dream":
+            return
+        prior_raw = task.metadata.get("prior_mtime", "")
+        memory_dir = task.metadata.get("memory_dir") or None
+        if not prior_raw:
+            return
+        try:
+            prior_mtime = float(prior_raw)
+        except ValueError:
+            return
+        if task.status in {"failed", "killed"} or task.metadata.get("preview") == "true":
+            rollback_consolidation_lock(task.cwd, prior_mtime, memory_dir=memory_dir)
+
+    get_task_manager().register_completion_listener(_listener)
+    _listener_registered = True
+
+
+def _resolve_memory_dir(cwd: str | Path, memory_dir: str | Path | None) -> Path:
+    return Path(memory_dir).expanduser().resolve() if memory_dir is not None else get_project_memory_dir(cwd)
+
+
+def _resolve_session_dir(cwd: str | Path, session_dir: str | Path | None) -> Path:
+    return Path(session_dir).expanduser().resolve() if session_dir is not None else get_project_session_dir(cwd)
+
+
+async def start_dream_now(
+    *,
+    cwd: str | Path,
+    settings: Settings,
+    model: str | None = None,
+    current_session_id: str | None = None,
+    force: bool = False,
+    memory_dir: str | Path | None = None,
+    session_dir: str | Path | None = None,
+    app_label: str = "openharness",
+    runner_module: str = "openharness",
+    preview: bool = False,
+) -> TaskRecord | None:
+    """Start a dream task immediately, optionally bypassing time/session gates."""
+
+    if os.environ.get(_CHILD_ENV):
+        return None
+    if not settings.memory.enabled:
+        return None
+
+    cwd = Path(cwd).resolve()
+    resolved_memory_dir = _resolve_memory_dir(cwd, memory_dir)
+    resolved_session_dir = _resolve_session_dir(cwd, session_dir)
+    last_at = read_last_consolidated_at(cwd, memory_dir=resolved_memory_dir)
+    session_ids = list_sessions_touched_since(
+        cwd,
+        last_at,
+        current_session_id=current_session_id,
+        session_dir=resolved_session_dir,
+    )
+    if not force:
+        hours_since = (time.time() - last_at) / 3600
+        if hours_since < settings.memory.auto_dream_min_hours:
+            return None
+        if len(session_ids) < settings.memory.auto_dream_min_sessions:
+            return None
+    if not _has_dream_signal(session_ids, force=force):
+        return None
+
+    prior_mtime = try_acquire_consolidation_lock(cwd, memory_dir=resolved_memory_dir)
+    if prior_mtime is None:
+        return None
+
+    _ensure_listener_registered()
+    resolved_memory_dir.mkdir(parents=True, exist_ok=True)
+    resolved_session_dir.mkdir(parents=True, exist_ok=True)
+    before = _memory_files_mtime_snapshot(resolved_memory_dir)
+    backup_dir = create_memory_backup(resolved_memory_dir, app_label=app_label) if not preview else None
+    extra = (
+        f"Application context: `{app_label}`.\n"
+        "Tool constraints for this run: only modify files under the memory directory. "
+        "Use shell commands only for read-only inspection.\n\n"
+        f"Sessions since last consolidation ({len(session_ids)}):\n"
+        + "\n".join(f"- {session_id}" for session_id in session_ids)
+    )
+    prompt = build_consolidation_prompt(resolved_memory_dir, resolved_session_dir, extra, preview=preview)
+    src_root = Path(__file__).resolve().parents[3]
+    existing_pythonpath = os.environ.get("PYTHONPATH", "")
+    env = {
+        _CHILD_ENV: "1",
+        "OPENHARNESS_AUTODREAM_MEMORY_DIR": str(resolved_memory_dir),
+        "OPENHARNESS_CONFIG_DIR": str(Path.home() / ".openharness"),
+        "OPENHARNESS_PROFILE": settings.active_profile,
+        "PYTHONPATH": str(src_root) + ((os.pathsep + existing_pythonpath) if existing_pythonpath else ""),
+    }
+    try:
+        argv = [
+            sys.executable,
+            "-m",
+            runner_module,
+        ]
+        if runner_module == "openharness":
+            argv.append("--dangerously-skip-permissions")
+        if runner_module == "ohmo":
+            workspace = resolved_memory_dir.parent
+            argv.extend(["--workspace", str(workspace)])
+            if settings.active_profile:
+                argv.extend(["--profile", settings.active_profile])
+        if model:
+            argv.extend(["--model", model])
+        if runner_module == "openharness" and settings.provider != "anthropic_claude":
+            if settings.base_url:
+                argv.extend(["--base-url", settings.base_url])
+            if settings.api_format:
+                argv.extend(["--api-format", settings.api_format])
+        try:
+            auth = settings.resolve_auth()
+            if runner_module == "openharness" and auth.auth_kind == "api_key":
+                argv.extend(["--api-key", auth.value])
+            elif runner_module == "ohmo" and auth.auth_kind == "api_key":
+                env["OPENHARNESS_API_KEY"] = auth.value
+            elif auth.value:
+                env["ANTHROPIC_AUTH_TOKEN"] = auth.value
+                env.pop("ANTHROPIC_API_KEY", None)
+                env.pop("OPENAI_API_KEY", None)
+                env.pop("OPENHARNESS_API_KEY", None)
+        except Exception:
+            pass
+        argv.extend(["--print", prompt])
+        task = await get_task_manager().create_shell_task(
+            description="dreaming",
+            cwd=cwd,
+            task_type="dream",
+            env=env,
+            argv=argv,
+        )
+        task.prompt = prompt
+    except Exception:
+        rollback_consolidation_lock(cwd, prior_mtime, memory_dir=resolved_memory_dir)
+        raise
+
+    task.metadata.update(
+        {
+            "phase": "starting",
+            "sessions_reviewing": str(len(session_ids)),
+            "prior_mtime": str(prior_mtime),
+            "memory_dir": str(resolved_memory_dir),
+            "session_dir": str(resolved_session_dir),
+            "force": str(force).lower(),
+            "app_label": app_label,
+            "runner_module": runner_module,
+            "preview": str(preview).lower(),
+            "backup_dir": str(backup_dir or ""),
+        }
+    )
+
+    async def _mark_changed_on_completion(done: TaskRecord) -> None:
+        if done.id != task.id or done.status != "completed":
+            return
+        changed = _files_changed_since(resolved_memory_dir, before)
+        if backup_dir is not None:
+            diff = diff_memory_dirs(backup_dir, resolved_memory_dir)
+            done.metadata["files_added"] = "\n".join(diff["added"])
+            done.metadata["files_changed"] = "\n".join(diff["changed"])
+            done.metadata["files_removed"] = "\n".join(diff["removed"])
+        if changed:
+            done.metadata["phase"] = "updating"
+            done.metadata["files_touched"] = "\n".join(changed)
+
+    get_task_manager().register_completion_listener(_mark_changed_on_completion)
+    return task
+
+
+async def execute_auto_dream(
+    *,
+    cwd: str | Path,
+    settings: Settings,
+    model: str | None = None,
+    current_session_id: str | None = None,
+    memory_dir: str | Path | None = None,
+    session_dir: str | Path | None = None,
+    app_label: str = "openharness",
+    runner_module: str = "openharness",
+    preview: bool = False,
+) -> TaskRecord | None:
+    """Run the cheap auto-dream gates and start a background dream when eligible."""
+
+    if os.environ.get(_CHILD_ENV):
+        return None
+    if not _enabled(settings):
+        return None
+
+    cwd = Path(cwd).resolve()
+    resolved_memory_dir = _resolve_memory_dir(cwd, memory_dir)
+    resolved_session_dir = _resolve_session_dir(cwd, session_dir)
+    last_at = read_last_consolidated_at(cwd, memory_dir=resolved_memory_dir)
+    hours_since = (time.time() - last_at) / 3600
+    if hours_since < settings.memory.auto_dream_min_hours:
+        return None
+
+    key = str(resolved_memory_dir)
+    now = time.time()
+    if now - _last_session_scan_at.get(key, 0) < SESSION_SCAN_INTERVAL_SECONDS:
+        return None
+    _last_session_scan_at[key] = now
+
+    session_ids = list_sessions_touched_since(
+        cwd,
+        last_at,
+        current_session_id=current_session_id,
+        session_dir=resolved_session_dir,
+    )
+    if len(session_ids) < settings.memory.auto_dream_min_sessions:
+        return None
+    if not _has_dream_signal(session_ids, force=False):
+        return None
+
+    return await start_dream_now(
+        cwd=cwd,
+        settings=settings,
+        model=model,
+        current_session_id=current_session_id,
+        force=False,
+        memory_dir=resolved_memory_dir,
+        session_dir=resolved_session_dir,
+        app_label=app_label,
+        runner_module=runner_module,
+        preview=preview,
+    )
+
+
+def schedule_auto_dream(**kwargs: object) -> None:
+    """Fire-and-forget auto-dream scheduling."""
+
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        return
+    loop.create_task(execute_auto_dream(**kwargs))  # type: ignore[arg-type]

src/openharness/tasks/manager.py

@@ -209,6 +209,7 @@ class BackgroundTaskManager:
 
         task.status = "killed"
         task.ended_at = time.time()
+        await self._notify_completion_listeners(task)
         return task
 
     async def write_to_task(self, task_id: str, data: str) -> None:
@@ -458,6 +459,7 @@ def _task_id(task_type: TaskType) -> str:
         "local_agent": "a",
         "remote_agent": "r",
         "in_process_teammate": "t",
+        "dream": "d",
     }
     return f"{prefixes[task_type]}{uuid4().hex[:8]}"
 

src/openharness/tasks/types.py

@@ -7,7 +7,7 @@ from pathlib import Path
 from typing import Literal
 
 
-TaskType = Literal["local_bash", "local_agent", "remote_agent", "in_process_teammate"]
+TaskType = Literal["local_bash", "local_agent", "remote_agent", "in_process_teammate", "dream"]
 TaskStatus = Literal["pending", "running", "completed", "failed", "killed"]
 
 

src/openharness/tools/config_tool.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from pydantic import BaseModel, Field
 
 from openharness.config.settings import load_settings, save_settings
@@ -29,9 +31,24 @@ class ConfigTool(BaseTool):
         if arguments.action == "show":
             return ToolResult(output=settings.model_dump_json(indent=2))
         if arguments.action == "set" and arguments.key and arguments.value is not None:
-            if not hasattr(settings, arguments.key):
+            target: Any = settings
+            parts = arguments.key.split(".")
+            for part in parts[:-1]:
+                if not hasattr(target, part):
+                    return ToolResult(output=f"Unknown config key: {arguments.key}", is_error=True)
+                target = getattr(target, part)
+            leaf = parts[-1]
+            if not hasattr(target, leaf):
                 return ToolResult(output=f"Unknown config key: {arguments.key}", is_error=True)
-            setattr(settings, arguments.key, arguments.value)
+            current = getattr(target, leaf)
+            value: Any = arguments.value
+            if isinstance(current, bool):
+                value = arguments.value.strip().lower() in {"1", "true", "yes", "on"}
+            elif isinstance(current, int) and not isinstance(current, bool):
+                value = int(arguments.value)
+            elif isinstance(current, float):
+                value = float(arguments.value)
+            setattr(target, leaf, value)
             save_settings(settings)
             return ToolResult(output=f"Updated {arguments.key}")
         return ToolResult(output="Usage: action=show or action=set with key/value", is_error=True)

src/openharness/ui/runtime.py

@@ -98,6 +98,7 @@ class RuntimeBundle:
     extra_plugin_roots: tuple[str, ...] = ()
     memory_backend: MemoryCommandBackend | None = None
     include_project_memory: bool = True
+    autodream_context: dict[str, object] | None = None
 
     def current_settings(self):
         """Return the effective settings for this session.
@@ -225,6 +226,7 @@ async def build_runtime(
     extra_plugin_roots: Iterable[str | Path] | None = None,
     memory_backend: MemoryCommandBackend | None = None,
     include_project_memory: bool = True,
+    autodream_context: dict[str, object] | None = None,
 ) -> RuntimeBundle:
     """Build the shared runtime for an OpenHarness session."""
     settings_overrides: dict[str, Any] = {
@@ -341,6 +343,7 @@ async def build_runtime(
         permission_prompt=permission_prompt,
         ask_user_prompt=ask_user_prompt,
         hook_executor=hook_executor,
+        settings=settings,
         tool_metadata={
             "mcp_manager": mcp_manager,
             "bridge_manager": bridge_manager,
@@ -351,6 +354,8 @@ async def build_runtime(
             **restored_metadata,
         },
     )
+    if autodream_context is not None:
+        engine.tool_metadata["autodream_context"] = autodream_context
     # Restore messages from a saved session if provided
     if restore_messages:
         restored = sanitize_conversation_messages(
@@ -389,6 +394,7 @@ async def build_runtime(
         extra_plugin_roots=normalized_plugin_roots,
         memory_backend=memory_backend,
         include_project_memory=include_project_memory,
+        autodream_context=autodream_context,
     )
 
 

tests/test_services/test_autodream.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+
+from openharness.config.settings import Settings
+from openharness.services.autodream.backup import create_memory_backup, diff_memory_dirs, restore_memory_backup
+from openharness.services.autodream.lock import (
+    list_sessions_touched_since,
+    read_last_consolidated_at,
+    rollback_consolidation_lock,
+    try_acquire_consolidation_lock,
+)
+from openharness.services.autodream.prompt import build_consolidation_prompt
+from openharness.services.autodream.service import execute_auto_dream, start_dream_now
+from openharness.services.session_storage import get_project_session_dir
+
+
+def test_consolidation_lock_acquire_and_rollback(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    cwd = tmp_path / "repo"
+    cwd.mkdir()
+
+    assert read_last_consolidated_at(cwd) == 0
+    prior = try_acquire_consolidation_lock(cwd)
+    assert prior == 0
+    assert read_last_consolidated_at(cwd) > 0
+    assert try_acquire_consolidation_lock(cwd) is None
+
+    rollback_consolidation_lock(cwd, prior)
+    assert read_last_consolidated_at(cwd) == 0
+
+
+def test_consolidation_lock_supports_memory_dir_override(tmp_path: Path) -> None:
+    cwd = tmp_path / "repo"
+    memory_dir = tmp_path / "ohmo" / "memory"
+    cwd.mkdir()
+
+    prior = try_acquire_consolidation_lock(cwd, memory_dir=memory_dir)
+    assert prior == 0
+    assert (memory_dir / ".consolidate-lock").exists()
+    assert read_last_consolidated_at(cwd, memory_dir=memory_dir) > 0
+
+    rollback_consolidation_lock(cwd, prior, memory_dir=memory_dir)
+    assert read_last_consolidated_at(cwd, memory_dir=memory_dir) == 0
+
+
+def test_list_sessions_touched_since_excludes_current(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    cwd = tmp_path / "repo"
+    cwd.mkdir()
+    session_dir = get_project_session_dir(cwd)
+    old = session_dir / "session-old.json"
+    old.write_text(json.dumps({"session_id": "old"}), encoding="utf-8")
+    new = session_dir / "session-new.json"
+    new.write_text(json.dumps({"session_id": "new"}), encoding="utf-8")
+    current = session_dir / "session-current.json"
+    current.write_text(json.dumps({"session_id": "current"}), encoding="utf-8")
+    cutoff = time.time() - 10
+    os.utime(old, (cutoff - 20, cutoff - 20))
+
+    assert list_sessions_touched_since(cwd, cutoff, current_session_id="current") == ["new"]
+
+
+def test_list_sessions_touched_since_supports_session_dir_override(tmp_path: Path) -> None:
+    cwd = tmp_path / "repo"
+    session_dir = tmp_path / "ohmo" / "sessions"
+    cwd.mkdir()
+    session_dir.mkdir(parents=True)
+    (session_dir / "session-ohmo.json").write_text(json.dumps({"session_id": "ohmo"}), encoding="utf-8")
+
+    assert list_sessions_touched_since(cwd, 0, session_dir=session_dir) == ["ohmo"]
+
+
+def test_consolidation_prompt_contains_expected_sections(tmp_path: Path) -> None:
+    prompt = build_consolidation_prompt(tmp_path / "memory", tmp_path / "sessions", "extra")
+    assert "# Dream: Memory Consolidation" in prompt
+    assert "Phase 1" in prompt
+    assert "Phase 4" in prompt
+    assert "MEMORY.md" in prompt
+    assert "Never preserve API keys" in prompt
+    assert "Evidence discipline" in prompt
+    assert "Last observed" in prompt
+    assert "Privacy: personal/private" in prompt
+    assert "extra" in prompt
+
+
+def test_consolidation_prompt_preview_mode(tmp_path: Path) -> None:
+    prompt = build_consolidation_prompt(tmp_path / "memory", tmp_path / "sessions", preview=True)
+    assert "PREVIEW MODE" in prompt
+    assert "do not write files" in prompt
+
+
+def test_memory_backup_diff_and_restore(tmp_path: Path) -> None:
+    memory_dir = tmp_path / "memory"
+    memory_dir.mkdir()
+    (memory_dir / "MEMORY.md").write_text("# Memory\n", encoding="utf-8")
+    backup = create_memory_backup(memory_dir, backup_root=tmp_path / "backups")
+    (memory_dir / "new.md").write_text("new\n", encoding="utf-8")
+    (memory_dir / "MEMORY.md").write_text("# Memory\n- changed\n", encoding="utf-8")
+
+    diff = diff_memory_dirs(backup, memory_dir)
+    assert diff["added"] == ["new.md"]
+    assert diff["changed"] == ["MEMORY.md"]
+
+    restore_memory_backup(backup, memory_dir)
+    assert not (memory_dir / "new.md").exists()
+    assert (memory_dir / "MEMORY.md").read_text(encoding="utf-8") == "# Memory\n"
+
+
+async def test_execute_auto_dream_skips_when_disabled(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    cwd = tmp_path / "repo"
+    cwd.mkdir()
+    settings = Settings()
+    settings.memory.auto_dream_enabled = False
+
+    assert await execute_auto_dream(cwd=cwd, settings=settings, model="test") is None
+
+
+async def test_start_dream_now_uses_overrides(tmp_path: Path, monkeypatch) -> None:
+    cwd = tmp_path / "repo"
+    memory_dir = tmp_path / ".ohmo" / "memory"
+    session_dir = tmp_path / ".ohmo" / "sessions"
+    cwd.mkdir()
+    memory_dir.mkdir(parents=True)
+    session_dir.mkdir(parents=True)
+    (session_dir / "session-one.json").write_text(json.dumps({"session_id": "one"}), encoding="utf-8")
+
+    captured = {}
+
+    class _Manager:
+        def register_completion_listener(self, listener):
+            return None
+
+        async def create_shell_task(self, **kwargs):
+            from openharness.tasks.types import TaskRecord
+
+            captured.update(kwargs)
+            return TaskRecord(
+                id="dtest",
+                type="dream",
+                status="running",
+                description=kwargs["description"],
+                cwd=str(kwargs["cwd"]),
+                output_file=tmp_path / "dream.log",
+                argv=kwargs["argv"],
+                env=kwargs["env"],
+            )
+
+    monkeypatch.setattr("openharness.services.autodream.service.get_task_manager", lambda: _Manager())
+    settings = Settings()
+    task = await start_dream_now(
+        cwd=cwd,
+        settings=settings,
+        force=True,
+        memory_dir=memory_dir,
+        session_dir=session_dir,
+        app_label="ohmo personal memory",
+        runner_module="ohmo",
+    )
+
+    assert task is not None
+    assert task.metadata["memory_dir"] == str(memory_dir.resolve())
+    assert task.metadata["session_dir"] == str(session_dir.resolve())
+    assert task.metadata["app_label"] == "ohmo personal memory"
+    assert task.metadata["backup_dir"]
+    assert captured["argv"][:3][-1] == "ohmo"
+    assert "--workspace" in captured["argv"]
+    assert "--dangerously-skip-permissions" not in captured["argv"]