Ov Dream

"""OV Dream skill package."""

"""Scripts for the OV Dream skill."""

from __future__ import annotations

import argparse
import json
import os
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
from urllib.error import HTTPError
from urllib.request import Request, urlopen

DEFAULT_BASE_URL = "http://127.0.0.1:1933"
DEFAULT_TARGET_URI = "viking://user/default"
LEGACY_TARGET_URI = "viking://user/memories"
SERVERLESS_BASE_URL = "https://api.vikingdb.cn-beijing.volces.com/openviking"


@dataclass
class Message:
    role: str
    content: str
    timestamp: str


@dataclass
class Session:
    session_id: str
    cwd: str
    created_at: str
    session_key: str = ""
    session_file: str = ""


class OpenVikingClient:
    def __init__(
        self,
        base_url: str,
        api_key: str | None = None,
        auth_mode: str = "auto",
        timeout: int = 30,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key or os.environ.get("OPENVIKING_API_KEY", "")
        self.auth_mode = self._resolve_auth_mode(auth_mode)
        self.timeout = timeout

    def _resolve_auth_mode(self, auth_mode: str) -> str:
        if auth_mode not in {"auto", "local", "serverless"}:
            raise ValueError("auth_mode must be one of: auto, local, serverless")
        if auth_mode != "auto":
            return auth_mode
        if "api.vikingdb" in self.base_url or self.base_url.endswith("/openviking"):
            return "serverless"
        return "local"

    def _headers(self) -> dict[str, str]:
        headers = {"Content-Type": "application/json"}
        if self.auth_mode == "serverless":
            if self.api_key:
                headers["Authorization"] = "Bearer " + self.api_key
        else:
            headers.update(
                {
                    "X-OpenViking-Account": os.environ.get("OPENVIKING_ACCOUNT", "default"),
                    "X-OpenViking-User": os.environ.get("OPENVIKING_USER", "default"),
                }
            )
            if self.api_key:
                headers["X-API-Key"] = self.api_key
        return headers

    def _resolve_target_uri(self, target_uri: str) -> str:
        normalized = target_uri.rstrip("/")
        if self.auth_mode == "serverless":
            return target_uri
        if normalized == DEFAULT_TARGET_URI:
            user_space = self._headers().get("X-OpenViking-User", "default") or "default"
            return f"viking://user/{user_space}"
        if normalized == LEGACY_TARGET_URI:
            user_space = self._headers().get("X-OpenViking-User", "default") or "default"
            return f"viking://user/{user_space}/memories/"
        return target_uri

    def _request(
        self, method: str, path: str, payload: dict[str, Any] | None = None
    ) -> dict[str, Any]:
        data = None if payload is None else json.dumps(payload).encode("utf-8")
        request = Request(
            f"{self.base_url}{path}",
            data=data,
            headers=self._headers(),
            method=method,
        )
        try:
            with urlopen(request, timeout=self.timeout) as response:
                body = json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            raw = exc.read().decode("utf-8", errors="replace")
            try:
                body = json.loads(raw)
            except json.JSONDecodeError as decode_exc:
                raise RuntimeError(f"HTTP {exc.code}: {raw}") from decode_exc
            error = body.get("error") or {}
            detail = body.get("detail")
            message = error.get("message") or detail or f"HTTP {exc.code}"
            raise RuntimeError(message) from exc
        if body.get("status") == "error":
            message = body.get("error", {}).get("message", "unknown error")
            raise RuntimeError(message)
        return body.get("result", body)

    def add_session_message(self, session_id: str, role: str, content: str) -> dict[str, Any]:
        if self.auth_mode == "serverless":
            payload = {
                "role": role,
                "parts": [{"type": "text", "text": content}],
            }
        else:
            payload = {"role": role, "content": content}
        return self._request(
            "POST",
            f"/api/v1/sessions/{session_id}/messages",
            payload,
        )

    def commit_session(self, session_id: str, wait: bool = True) -> dict[str, Any]:
        payload = {"telemetry": False} if self.auth_mode == "serverless" else {}
        suffix = "" if self.auth_mode == "serverless" else "?wait=true" if wait else ""
        return self._request("POST", f"/api/v1/sessions/{session_id}/commit{suffix}", payload)

    def recall(
        self, query: str, limit: int = 5, target_uri: str = DEFAULT_TARGET_URI
    ) -> dict[str, Any]:
        return self._request(
            "POST",
            "/api/v1/search/find",
            {
                "query": query,
                "limit": limit,
                "target_uri": self._resolve_target_uri(target_uri),
            },
        )


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def load_sync_state(state_root: Path) -> dict[str, Any]:
    path = state_root / "ov_dream_sync.json"
    if not path.exists():
        return {"sessions": {}}
    return json.loads(path.read_text(encoding="utf-8"))


def save_sync_state(state_root: Path, state: dict[str, Any]) -> None:
    state_root.mkdir(parents=True, exist_ok=True)
    path = state_root / "ov_dream_sync.json"
    path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")


def get_session_path(sessions_root: Path, session_id: str) -> Path:
    return sessions_root / f"{session_id}.jsonl"


def get_session_file_path(sessions_root: Path, session: Session) -> Path:
    if not session.session_file:
        return get_session_path(sessions_root, session.session_id)
    path = Path(session.session_file)
    return path if path.is_absolute() else sessions_root / path


def is_chat_session_key(key: str) -> bool:
    # OpenClaw chat session keys can vary by channel, so only filter known non-chat routes.
    blocked = (":cron:", ":heartbeat", ":subagent:", ":acp:", ":hook:")
    return bool(key) and not any(part in key for part in blocked)


def _session_from_file(path: Path, session_key: str = "") -> Session | None:
    if not path.exists():
        return None
    lines = path.read_text(encoding="utf-8").splitlines()
    if not lines:
        return None
    try:
        first = json.loads(lines[0])
    except json.JSONDecodeError:
        return None
    session_id = first.get("id")
    if not isinstance(session_id, str) or not session_id:
        session_id = path.stem
    return Session(
        session_id=session_id,
        cwd=first.get("cwd", ""),
        created_at=first.get("timestamp", ""),
        session_key=session_key,
        session_file=str(path),
    )


def _session_from_index_entry(sessions_root: Path, session_key: str, entry: Any) -> Session | None:
    if not isinstance(entry, dict):
        return None

    session_id = entry.get("sessionId")
    if not isinstance(session_id, str) or not session_id:
        return None

    session_file = entry.get("sessionFile")
    if isinstance(session_file, str) and session_file:
        raw_path = Path(session_file)
        path = raw_path if raw_path.is_absolute() else sessions_root / raw_path
    else:
        path = get_session_path(sessions_root, session_id)
    session = _session_from_file(path, session_key=session_key)
    if session is None:
        return None
    if session.session_id != session_id:
        session.session_id = session_id
    return session


def _get_indexed_chat_sessions(sessions_root: Path) -> list[Session]:
    index_path = sessions_root / "sessions.json"
    if not index_path.exists():
        return []
    try:
        index = json.loads(index_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return []
    if not isinstance(index, dict):
        return []

    sessions: list[Session] = []
    seen: set[str] = set()
    for session_key, entry in sorted(index.items()):
        if not isinstance(session_key, str) or not is_chat_session_key(session_key):
            continue
        session = _session_from_index_entry(sessions_root, session_key, entry)
        if session is None or session.session_id in seen:
            continue
        seen.add(session.session_id)
        sessions.append(session)
    return sessions


def get_active_sessions(openclaw_root: Path) -> list[Session]:
    sessions_root = openclaw_root / "agents" / "main" / "sessions"
    if not sessions_root.exists():
        return []

    # Only trust OpenClaw's session index; raw jsonl fallback can accidentally sync cron/subagent transcripts.
    return _get_indexed_chat_sessions(sessions_root)


def get_active_session(openclaw_root: Path) -> Session | None:
    sessions = get_active_sessions(openclaw_root)
    return sessions[0] if sessions else None


def parse_messages(
    sessions_root: Path, session: Session, after_timestamp: str | None
) -> Iterable[Message]:
    path = get_session_file_path(sessions_root, session)
    if not path.exists():
        return []

    messages: list[Message] = []
    for line in path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        row = json.loads(line)
        if row.get("type") != "message":
            continue
        timestamp = row.get("timestamp", "")
        if after_timestamp and timestamp <= after_timestamp:
            continue
        message = row.get("message", {})
        role = message.get("role")
        if role not in {"user", "assistant"}:
            continue
        blocks = message.get("content", [])
        text_parts = [
            block.get("text", "").strip() for block in blocks if block.get("type") == "text"
        ]
        content = "\n".join(part for part in text_parts if part)
        if not content:
            continue
        messages.append(Message(role=role, content=content, timestamp=timestamp))
    return messages


def sync_session(
    client: OpenVikingClient,
    sessions_root: Path,
    state: dict[str, Any],
    session: Session,
) -> dict[str, Any]:
    sessions = state.setdefault("sessions", {})
    session_state = sessions.get(session.session_id)
    if not isinstance(session_state, dict):
        session_state = {}

    # Cursor is tracked per source session so cron syncs only upload newly appended messages.
    last_synced_timestamp = session_state.get("last_synced_timestamp")
    messages = [
        message
        for message in parse_messages(sessions_root, session, last_synced_timestamp)
        if message.timestamp
        and (last_synced_timestamp is None or message.timestamp > last_synced_timestamp)
    ]
    messages.sort(key=lambda message: message.timestamp)

    synced_count = 0
    committed = False
    now = _utc_now_iso()
    for message in messages:
        client.add_session_message(session.session_id, message.role, message.content)
        synced_count += 1

    session_state["last_status"] = "ok"
    session_state["last_synced_count"] = synced_count
    session_state["last_sync_at"] = now
    session_state["committed"] = False
    session_state["session_key"] = session.session_key
    session_state["session_file"] = session.session_file

    if synced_count:
        client.commit_session(session.session_id, wait=True)
        committed = True
        session_state["committed"] = True
        session_state["last_commit_at"] = now
        last_synced_timestamp = messages[-1].timestamp
        session_state["last_synced_timestamp"] = last_synced_timestamp

    sessions[session.session_id] = session_state
    return {
        "session_key": session.session_key,
        "session_id": session.session_id,
        "synced_count": synced_count,
        "committed": committed,
        "last_synced_timestamp": last_synced_timestamp,
    }


def sync_active_session(
    client: OpenVikingClient, openclaw_root: Path, state_root: Path
) -> dict[str, Any]:
    sessions_root = openclaw_root / "agents" / "main" / "sessions"
    active_sessions = get_active_sessions(openclaw_root)
    if not active_sessions:
        raise RuntimeError("No active OpenClaw chat sessions found.")

    state = load_sync_state(state_root)
    summaries: list[dict[str, Any]] = []
    try:
        for session in active_sessions:
            summaries.append(sync_session(client, sessions_root, state, session))
    except Exception:
        save_sync_state(state_root, state)
        raise

    save_sync_state(state_root, state)
    last_timestamps = [
        str(summary.get("last_synced_timestamp", ""))
        for summary in summaries
        if summary.get("last_synced_timestamp")
    ]
    return {
        "session_count": len(summaries),
        "sessions": summaries,
        "synced_count": sum(int(summary.get("synced_count", 0) or 0) for summary in summaries),
        "committed": any(bool(summary.get("committed")) for summary in summaries),
        "last_synced_timestamp": max(last_timestamps) if last_timestamps else None,
    }


def _normalize_ov_command(argv: list[str] | None) -> list[str] | None:
    if argv is None:
        return None
    if not argv:
        return argv
    if len(argv) >= 2 and argv[0] == "ov":
        if argv[1] == "dream":
            return ["dream"]
        if argv[1] == "recall":
            query = " ".join(argv[2:]).strip()
            return ["recall", query] if query else ["recall", ""]
    if len(argv) == 1:
        raw = argv[0].strip()
        if raw == "ov dream":
            return ["dream"]
        if raw.startswith("ov recall "):
            return ["recall", raw[len("ov recall ") :].strip()]
    return argv


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="ov dream")
    parser.add_argument(
        "--base-url", default=os.environ.get("OPENVIKING_BASE_URL", DEFAULT_BASE_URL)
    )
    parser.add_argument("--api-key", default=None)
    parser.add_argument(
        "--auth-mode",
        choices=["auto", "local", "serverless"],
        default=os.environ.get("OPENVIKING_AUTH_MODE", "auto"),
    )
    parser.add_argument("--openclaw-root", default=str(Path.home() / ".openclaw"))
    parser.add_argument("--state-root", default=str(Path.home() / ".openclaw" / "memory"))

    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("dream", help="Sync the active OpenClaw session to OpenViking.")

    recall = subparsers.add_parser("recall", help="Recall memories from OpenViking.")
    recall.add_argument("query")
    recall.add_argument("--limit", type=int, default=5)

    return parser


def _iter_memories(result: Any) -> Iterable[dict[str, Any]]:
    if not isinstance(result, dict):
        return []
    memories = result.get("memories")
    if not isinstance(memories, list):
        return []
    return [item for item in memories if isinstance(item, dict)]


def _print_sync_summary(summary: dict[str, Any]) -> None:
    child_summaries = summary.get("sessions")
    if isinstance(child_summaries, list):
        print(
            "session_count={session_count} synced_count={synced_count} committed={committed} last_synced_timestamp={last_synced_timestamp}".format(
                session_count=summary.get("session_count", len(child_summaries)),
                synced_count=summary.get("synced_count", 0),
                committed=str(summary.get("committed", False)).lower(),
                last_synced_timestamp=summary.get("last_synced_timestamp", ""),
            )
        )
        for child in child_summaries:
            if isinstance(child, dict):
                _print_sync_summary(child)
        return

    print(
        "session_key={session_key} session_id={session_id} synced_count={synced_count} committed={committed} last_synced_timestamp={last_synced_timestamp}".format(
            session_key=summary.get("session_key", ""),
            session_id=summary.get("session_id", ""),
            synced_count=summary.get("synced_count", 0),
            committed=str(summary.get("committed", False)).lower(),
            last_synced_timestamp=summary.get("last_synced_timestamp", ""),
        )
    )


def _print_recall_results(result: Any) -> None:
    memories = list(_iter_memories(result))
    if not memories:
        print("No memories found.")
        return

    for item in memories:
        uri = item.get("uri", "")
        score = item.get("score", "")
        summary = item.get("abstract") or item.get("overview") or ""
        print(f"{uri}|{score}|{summary}")


def run_dream(args: argparse.Namespace) -> int:
    client = OpenVikingClient(
        base_url=args.base_url,
        api_key=args.api_key,
        auth_mode=args.auth_mode,
    )
    summary = sync_active_session(
        client=client,
        openclaw_root=Path(args.openclaw_root),
        state_root=Path(args.state_root),
    )
    _print_sync_summary(summary)
    return 0


def run_recall(args: argparse.Namespace) -> int:
    client = OpenVikingClient(
        base_url=args.base_url,
        api_key=args.api_key,
        auth_mode=args.auth_mode,
    )
    result = client.recall(query=args.query, limit=args.limit)
    _print_recall_results(result)
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()
    try:
        args = parser.parse_args(_normalize_ov_command(argv))
        if args.command == "dream":
            return run_dream(args)
        if args.command == "recall":
            return run_recall(args)
        raise RuntimeError(f"Unsupported command: {args.command}")
    except Exception as exc:
        print(str(exc), file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

from __future__ import annotations

import importlib.util
import json
import sys
from pathlib import Path


def _load_dream_module():
    module_path = Path("examples/skills/ov_dream/scripts/dream.py").resolve()
    spec = importlib.util.spec_from_file_location("ov_dream_cli", module_path)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


dream = _load_dream_module()


def _write_session(
    path: Path, session_id: str, messages: list[tuple[str, str, str]] | None = None
) -> None:
    rows = [
        {"id": session_id, "timestamp": "2026-04-20T00:00:00Z", "cwd": "/tmp"},
    ]
    for role, text, timestamp in messages or []:
        rows.append(
            {
                "type": "message",
                "timestamp": timestamp,
                "message": {
                    "role": role,
                    "content": [{"type": "text", "text": text}],
                },
            }
        )
    path.write_text(
        "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n", encoding="utf-8"
    )


def test_normalize_raw_ov_recall_phrase() -> None:
    assert dream._normalize_ov_command(["ov recall 小明的信息"]) == ["recall", "小明的信息"]


def test_recall_expands_default_user_root_to_explicit_user_space(monkeypatch) -> None:
    monkeypatch.setenv("OPENVIKING_USER", "default")
    client = dream.OpenVikingClient(base_url="http://127.0.0.1:1933")

    assert client._resolve_target_uri("viking://user/default") == "viking://user/default"
    assert client._resolve_target_uri("viking://user/default/") == "viking://user/default"
    assert client._resolve_target_uri("viking://user/memories") == "viking://user/default/memories/"
    assert (
        client._resolve_target_uri("viking://user/memories/") == "viking://user/default/memories/"
    )
    assert (
        client._resolve_target_uri("viking://user/default/memories/")
        == "viking://user/default/memories/"
    )


def test_recall_default_target_uri_is_user_root() -> None:
    calls = []

    class RecordingClient(dream.OpenVikingClient):
        def _request(self, method, path, payload=None):
            calls.append((method, path, payload))
            return {"memories": []}

    client = RecordingClient(base_url=dream.SERVERLESS_BASE_URL, api_key="test-key")

    client.recall("hello")

    assert calls == [
        (
            "POST",
            "/api/v1/search/find",
            {
                "query": "hello",
                "limit": 5,
                "target_uri": "viking://user/default",
            },
        )
    ]


def test_serverless_headers_use_bearer_auth() -> None:
    client = dream.OpenVikingClient(
        base_url=dream.SERVERLESS_BASE_URL,
        api_key="test-key",
    )

    assert client.auth_mode == "serverless"
    assert client._headers()["Authorization"] == "Bearer test-key"
    assert "X-API-Key" not in client._headers()
    assert "X-OpenViking-User" not in client._headers()


def test_serverless_sync_reuses_source_session_id_and_uses_parts_payload() -> None:
    calls = []

    class RecordingClient(dream.OpenVikingClient):
        def _request(self, method, path, payload=None):
            calls.append((method, path, payload))
            return {}

    client = RecordingClient(
        base_url=dream.SERVERLESS_BASE_URL,
        api_key="test-key",
    )

    client.add_session_message("source-session", "user", "hello")
    client.commit_session("source-session")

    assert calls == [
        (
            "POST",
            "/api/v1/sessions/source-session/messages",
            {"role": "user", "parts": [{"type": "text", "text": "hello"}]},
        ),
        ("POST", "/api/v1/sessions/source-session/commit", {"telemetry": False}),
    ]


def test_get_active_session_prefers_sessions_index(tmp_path: Path) -> None:
    openclaw_root = tmp_path / ".openclaw"
    sessions_root = openclaw_root / "agents" / "main" / "sessions"
    sessions_root.mkdir(parents=True)

    indexed_session = sessions_root / "indexed.jsonl"
    indexed_session.write_text(
        json.dumps({"id": "indexed", "timestamp": "2026-04-20T00:00:00Z", "cwd": "/tmp"}) + "\n",
        encoding="utf-8",
    )
    newer_fallback = sessions_root / "newer.jsonl"
    newer_fallback.write_text(
        json.dumps({"id": "newer", "timestamp": "2026-04-20T00:00:01Z", "cwd": "/tmp"}) + "\n",
        encoding="utf-8",
    )

    (sessions_root / "sessions.json").write_text(
        json.dumps(
            {
                "agent:main:main": {
                    "sessionId": "indexed",
                    "sessionFile": str(indexed_session),
                }
            }
        ),
        encoding="utf-8",
    )

    session = dream.get_active_session(openclaw_root)

    assert session is not None
    assert session.session_id == "indexed"


def test_get_active_sessions_does_not_fallback_to_raw_jsonl(tmp_path: Path) -> None:
    openclaw_root = tmp_path / ".openclaw"
    sessions_root = openclaw_root / "agents" / "main" / "sessions"
    sessions_root.mkdir(parents=True)

    cron_session = sessions_root / "cron.jsonl"
    _write_session(cron_session, "cron")
    latest_unindexed = sessions_root / "latest.jsonl"
    _write_session(latest_unindexed, "latest")

    (sessions_root / "sessions.json").write_text(
        json.dumps(
            {
                "agent:main:cron:daily": {
                    "sessionId": "cron",
                    "sessionFile": str(cron_session),
                }
            }
        ),
        encoding="utf-8",
    )

    assert dream.get_active_sessions(openclaw_root) == []
    assert dream.get_active_session(openclaw_root) is None


def test_is_chat_session_key_filters_non_chat_openclaw_sessions() -> None:
    assert dream.is_chat_session_key("agent:main:main")
    assert dream.is_chat_session_key("agent:main:web-abc")
    assert dream.is_chat_session_key("agent:main:telegram:direct:123")
    assert dream.is_chat_session_key("agent:main:discord:channel:456")
    assert dream.is_chat_session_key("agent:main:chat:group:789")
    assert dream.is_chat_session_key("agent:main:chat:room:abc")
    assert dream.is_chat_session_key("agent:other:main")
    assert dream.is_chat_session_key("plain:main")

    assert not dream.is_chat_session_key("agent:main:cron:daily")
    assert not dream.is_chat_session_key("agent:main:heartbeat")
    assert not dream.is_chat_session_key("agent:main:subagent:child")
    assert not dream.is_chat_session_key("agent:main:acp:tool")
    assert not dream.is_chat_session_key("agent:main:hook:event")
    assert not dream.is_chat_session_key("")


def test_get_active_sessions_filters_index_entries(tmp_path: Path) -> None:
    openclaw_root = tmp_path / ".openclaw"
    sessions_root = openclaw_root / "agents" / "main" / "sessions"
    sessions_root.mkdir(parents=True)

    kept_main = sessions_root / "main.jsonl"
    kept_direct = sessions_root / "direct.jsonl"
    skipped_cron = sessions_root / "cron.jsonl"
    skipped_subagent = sessions_root / "subagent.jsonl"
    _write_session(kept_main, "main")
    _write_session(kept_direct, "direct")
    _write_session(skipped_cron, "cron")
    _write_session(skipped_subagent, "subagent")

    (sessions_root / "sessions.json").write_text(
        json.dumps(
            {
                "agent:main:main": {"sessionId": "main", "sessionFile": "main.jsonl"},
                "agent:main:telegram:direct:123": {
                    "sessionId": "direct",
                    "sessionFile": str(kept_direct),
                },
                "agent:main:cron:daily": {"sessionId": "cron", "sessionFile": str(skipped_cron)},
                "agent:main:subagent:child": {
                    "sessionId": "subagent",
                    "sessionFile": str(skipped_subagent),
                },
            }
        ),
        encoding="utf-8",
    )

    sessions = dream.get_active_sessions(openclaw_root)

    assert {session.session_id for session in sessions} == {"main", "direct"}
    assert {session.session_key for session in sessions} == {
        "agent:main:main",
        "agent:main:telegram:direct:123",
    }


def test_sync_active_session_syncs_chat_sessions_with_independent_cursors(tmp_path: Path) -> None:
    openclaw_root = tmp_path / ".openclaw"
    sessions_root = openclaw_root / "agents" / "main" / "sessions"
    state_root = openclaw_root / "memory"
    sessions_root.mkdir(parents=True)

    main_file = sessions_root / "main.jsonl"
    direct_file = sessions_root / "direct.jsonl"
    cron_file = sessions_root / "cron.jsonl"
    _write_session(
        main_file,
        "main",
        [("user", "hello from main", "2026-04-20T00:01:00Z")],
    )
    _write_session(
        direct_file,
        "direct",
        [("assistant", "hello from direct", "2026-04-20T00:02:00Z")],
    )
    _write_session(
        cron_file,
        "cron",
        [("user", "cron should not sync", "2026-04-20T00:03:00Z")],
    )
    (sessions_root / "sessions.json").write_text(
        json.dumps(
            {
                "agent:main:main": {"sessionId": "main", "sessionFile": str(main_file)},
                "agent:main:telegram:direct:123": {
                    "sessionId": "direct",
                    "sessionFile": str(direct_file),
                },
                "agent:main:cron:daily": {"sessionId": "cron", "sessionFile": str(cron_file)},
            }
        ),
        encoding="utf-8",
    )

    class RecordingClient:
        def __init__(self) -> None:
            self.messages = []
            self.commits = []

        def add_session_message(self, session_id, role, content):
            self.messages.append((session_id, role, content))

        def commit_session(self, session_id, wait=True):
            self.commits.append((session_id, wait))

    client = RecordingClient()

    summary = dream.sync_active_session(client, openclaw_root, state_root)

    assert summary["session_count"] == 2
    assert summary["synced_count"] == 2
    assert client.messages == [
        ("main", "user", "hello from main"),
        ("direct", "assistant", "hello from direct"),
    ]
    assert client.commits == [("main", True), ("direct", True)]

    state = json.loads((state_root / "ov_dream_sync.json").read_text(encoding="utf-8"))
    assert state["sessions"]["main"]["session_key"] == "agent:main:main"
    assert state["sessions"]["main"]["last_synced_timestamp"] == "2026-04-20T00:01:00Z"
    assert state["sessions"]["direct"]["session_key"] == "agent:main:telegram:direct:123"
    assert state["sessions"]["direct"]["last_synced_timestamp"] == "2026-04-20T00:02:00Z"
    assert "cron" not in state["sessions"]