echoes_vault.py

#!/usr/bin/env python3
"""Deterministic local storage engine for the EchoesVault Codex plugin."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import secrets
import shlex
import subprocess
import sys
import tempfile
import time
import unicodedata
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Iterator, Optional


MANAGED_ADAPTER_VERSION = "1.1.1"
ENGINE_VERSION = "1.1.1"
PROTOCOL_VERSION = "1.0.0"
SCHEMA_VERSION = 3
STATE_VERSION = 4
VAULT_DIRNAME = "EchoesVault"
RUNTIME_DIRNAME = ".echoes-vault"
RUNTIME_FILENAME = "echoes_vault.py"
STATE_RELATIVE_PATH = Path(RUNTIME_DIRNAME) / "state.json"
LOCK_RELATIVE_PATH = Path(RUNTIME_DIRNAME) / "lock"
OPENCODE_STATE_RELATIVE_PATH = Path(".opencode") / "echoes-state.json"
LEGACY_STATE_RELATIVE_PATH = Path(".codex") / "echoes-vault-state.json"
MARKER_FILENAME = ".echoes-vault.json"
PROTOCOL_FILENAME = "AGENT_PROTOCOL.md"
SUMMARY_MAX_LENGTH = 160
LOCK_WAIT_SECONDS = 8.0
STALE_LOCK_SECONDS = 60.0
REQUIRED_FRONTMATTER = ("type", "stack", "status", "summary")
INDEX_ENTRY_RE = re.compile(r"^- \[\[([^\]]+)\]\]:\s*(.*)$")
ENTRY_HEADER_RE = re.compile(r"^### (?:Scratchpad|Session) — ", flags=re.MULTILINE)
CONFLICT_MARKER_RE = re.compile(
    r"^(?:<<<<<<<(?: .*)?|=======|>>>>>>>(?: .*)?)$", flags=re.MULTILINE
)
DEFAULT_INDEX = """# EchoesVault Index

<!-- Generated by EchoesVault. Do not edit manually. -->

This registry tracks all structured pages in the project knowledge vault.

## Pages
"""
VAULT_MARKER = {
    "schemaVersion": SCHEMA_VERSION,
    "protocolVersion": PROTOCOL_VERSION,
    "generatedIndex": True,
    "dailyLayout": "unique-files-v1",
    "runtime": f"{RUNTIME_DIRNAME}/{RUNTIME_FILENAME}",
    "requiredFrontmatter": list(REQUIRED_FRONTMATTER),
}

AGENT_GUIDE_START = "<!-- echoes-vault:start -->"
AGENT_GUIDE_END = "<!-- echoes-vault:end -->"
AGENT_GUIDE_BLOCK = f"""{AGENT_GUIDE_START}
## EchoesVault project memory

This repository uses the agent-neutral EchoesVault protocol {PROTOCOL_VERSION}.
Managed adapter version: {MANAGED_ADAPTER_VERSION}. Reference engine: {ENGINE_VERSION}.

Before accessing persistent project memory, read `EchoesVault/{PROTOCOL_FILENAME}`. Use the project
runtime with `--workspace . --agent <agent-name> --adapter-version <adapter-version> <command>` for
all mutations.
Never edit `EchoesVault/index.md` or append to a shared date-level daily file manually.
Use `status` or `inspect` for read-only health checks; use `hydrate` only to refresh ignored local
state and the generated index. Final session saving requires an explicit user request.
{AGENT_GUIDE_END}"""

AGENT_ADAPTER_SKILL = f"""---
name: echoes-vault
description: Use repository-local EchoesVault memory when asked to initialize, restore, search, remember, document, inspect status, or explicitly save a session.
---

<!-- Generated by EchoesVault protocol {PROTOCOL_VERSION}. Do not edit manually. -->
<!-- Managed adapter version: {MANAGED_ADAPTER_VERSION}. -->

# EchoesVault adapter

Read `EchoesVault/{PROTOCOL_FILENAME}` before the first vault operation in a session. Run the
portable engine with:

```text
python3 {RUNTIME_DIRNAME}/{RUNTIME_FILENAME} --workspace . \\
  --agent <agent-name> --adapter-version {MANAGED_ADAPTER_VERSION} <command>
```

Use `init`, `inspect`, `hydrate`, `start --recent 3`, `status --format card`, `search`, `append`,
`hash`, `upsert`, and `end --confirm-explicit-user-end` according to the protocol. Pass write
payloads as JSON through stdin or a temporary JSON file. Never interpolate Markdown into a shell
command.

Do not use legacy EchoesVault tools that directly edit `index.md` or append to
`daily/YYYY-MM-DD.md`. Never finalize memory unless the user explicitly asks to end or save the
session.
"""

AGENT_PROTOCOL = f"""# EchoesVault Agent Protocol

<!-- Generated by EchoesVault protocol {PROTOCOL_VERSION}. Do not edit manually. -->

Reference engine: `{ENGINE_VERSION}`

EchoesVault is repository-local, agent-neutral project memory. Codex, OpenCode, Claude, and other
agents must use the same portable storage engine and the rules below.

## Compatibility

- Protocol version: `{PROTOCOL_VERSION}`
- Runtime: `{RUNTIME_DIRNAME}/{RUNTIME_FILENAME}`
- Engine version: `{ENGINE_VERSION}`
- Initialization marker: `EchoesVault/{MARKER_FILENAME}`
- Source of truth: `EchoesVault/pages/*.md` and `EchoesVault/daily/**/*.md`
- Generated local view: `EchoesVault/index.md`

Before writing, verify that the marker's `protocolVersion` equals the runtime protocol version.
Stop on an unsupported version; never guess a migration or write through an adapter that bypasses
the portable runtime.

## Required behavior

1. Invoke `{RUNTIME_DIRNAME}/{RUNTIME_FILENAME}` for every mutation, passing `--workspace .`,
   `--agent <agent-name>`, `--adapter-version <adapter-version>`, and the command.
2. Read before updating an existing page and use `hash <filename>` immediately before `upsert`.
3. Every page must begin with frontmatter containing `type`, `stack`, `status`, and `summary`.
4. `summary` must be non-empty, single-line, and no longer than {SUMMARY_MAX_LENGTH} characters.
5. Never edit `index.md`; the runtime derives it deterministically from page metadata.
6. Never append to a shared `daily/YYYY-MM-DD.md`; the runtime creates one unique file per entry.
7. Store durable technical facts, decisions, contracts, verified fixes, blockers, and next steps,
   not transcripts.
8. Deprecate instead of deleting. Use `status: deprecated`, a warning in the body, and a link to
   the replacement.
9. Finalize memory only after an explicit user request to end, wrap up, finalize, or save the
   session.
10. On a concurrency error, reread, reconcile, obtain a new hash, and retry. On conflict markers or
    an unsupported protocol, stop and report the problem.

## Page format

```yaml
---
type: architecture
stack: [python]
status: active
summary: Authentication boundaries and token flow.
---
```

Use `[[page-slug]]` for knowledge links and `![[asset.png]]` for files in `EchoesVault/assets/`.

## Command contract

- `init`: idempotently initialize or migrate the vault and install agent adapters.
- `migrate`: explicitly migrate a recognized legacy vault.
- `upgrade`: explicitly upgrade the project runtime and tracked adapters.
- `protocol`: report the supported protocol and managed paths.
- `configure-agents`: repair protocol documentation and agent adapters without changing knowledge.
- `inspect`: report health without writing any file.
- `hydrate`: refresh only ignored `index.md` and `state.json`.
- `status --format card`: read-only alias for `inspect` with a compact card.
- `start --recent 3`: restore the generated index and latest session entries.
- `search <query>`: search page bodies without loading the whole vault.
- `append --payload -`: write `{{"entry": "...", "agent": "optional-name"}}` to a unique log.
- `hash <filename>`: obtain `expectedSha256` before updating an existing page.
- `upsert --payload -`: create or replace one complete page.
- `end --confirm-explicit-user-end --payload -`: explicitly save a final summary and page updates.
- `rebuild-index`: validate page metadata and reconstruct the local generated index.

An `upsert` page payload contains `filename`, complete `content`, and `expectedSha256` for an
existing page. An `end` payload contains `dailySummary`, a `pages` array, and optional `agent`.

## Git contract

Commit the marker, protocol, portable runtime, agent adapters, pages, unique daily files, assets,
and raw sources. Do not commit `EchoesVault/index.md`, `{RUNTIME_DIRNAME}/state.json`,
`{RUNTIME_DIRNAME}/lock`, `.opencode/echoes-state.json`, or
`.codex/echoes-vault-state.json`.

Different pages and unique daily files normally merge cleanly. If two branches edit the same page,
resolve the Markdown conflict manually, retain valid frontmatter, remove all conflict markers, and
run `hydrate`, followed by read-only `status`.
"""

OPENCODE_COMMANDS = {
    "echoes-init.md": f"""---
description: Initialize or upgrade the agent-neutral EchoesVault
agent: build
---

<!-- Generated by EchoesVault protocol {PROTOCOL_VERSION}. Do not edit manually. -->
<!-- Managed adapter version: {MANAGED_ADAPTER_VERSION}. -->

Read `EchoesVault/{PROTOCOL_FILENAME}` when present, then initialize with:

<echoes_result>
!`python3 {RUNTIME_DIRNAME}/{RUNTIME_FILENAME} --workspace . --agent opencode --adapter-version {MANAGED_ADAPTER_VERSION} init`
</echoes_result>

Read the generated index. Briefly report the protocol version, installed agent adapters, and known
concepts. Do not edit the index manually.
""",
    "echoes-start.md": f"""---
description: Restore context through the agent-neutral EchoesVault runtime
agent: build
---

<!-- Generated by EchoesVault protocol {PROTOCOL_VERSION}. Do not edit manually. -->
<!-- Managed adapter version: {MANAGED_ADAPTER_VERSION}. -->

Read `EchoesVault/{PROTOCOL_FILENAME}`, then analyze this runtime-produced context:

<echoes_context>
!`python3 {RUNTIME_DIRNAME}/{RUNTIME_FILENAME} --workspace . --agent opencode --adapter-version {MANAGED_ADAPTER_VERSION} start --recent 3`
</echoes_context>

Summarize completed outcomes, blockers, and immediate next steps. Use targeted search for details.
""",
    "echoes-status.md": f"""---
description: Show agent-neutral EchoesVault health and integrity
agent: build
---

<!-- Generated by EchoesVault protocol {PROTOCOL_VERSION}. Do not edit manually. -->
<!-- Managed adapter version: {MANAGED_ADAPTER_VERSION}. -->

Return this card without inspecting the architectural meaning of pages:

<echoes_status>
!`python3 {RUNTIME_DIRNAME}/{RUNTIME_FILENAME} --workspace . --agent opencode --adapter-version {MANAGED_ADAPTER_VERSION} status --format card`
</echoes_status>
""",
    "echoes-end.md": f"""---
description: Explicitly distill and save the session through the agent-neutral EchoesVault runtime
agent: build
---

<!-- Generated by EchoesVault protocol {PROTOCOL_VERSION}. Do not edit manually. -->
<!-- Managed adapter version: {MANAGED_ADAPTER_VERSION}. -->

This command is the user's explicit request to finalize memory. Read
`EchoesVault/{PROTOCOL_FILENAME}`, search and read relevant existing pages, obtain hashes for every
existing page update, and distill outcomes, blockers, decisions, and next steps rather than a
transcript. Submit a JSON payload with `dailySummary`, `agent: "opencode"`, and `pages` to:

```text
python3 {RUNTIME_DIRNAME}/{RUNTIME_FILENAME} --workspace . \\
  --agent opencode --adapter-version {MANAGED_ADAPTER_VERSION} end \\
  --confirm-explicit-user-end --payload -
```

Never edit `index.md`, use legacy index mutation arguments, or claim success if the runtime rejects
the payload.
""",
}

LEGACY_OPENCODE_SKILLS = {
    "echoes-append-to-daily-log": {
        "signatures": (
            "echoes_append_to_daily_log",
            "EchoesVault/daily/YYYY-MM-DD.md",
        ),
        "content": f"""---
name: echoes-append-to-daily-log
description: Redirect legacy OpenCode scratchpad writes to the shared EchoesVault runtime.
---

<!-- Generated by EchoesVault protocol {PROTOCOL_VERSION}. Do not edit manually. -->
<!-- Managed adapter version: {MANAGED_ADAPTER_VERSION}. Legacy redirect. -->

# EchoesVault legacy append redirect

Do not call the legacy `echoes_append_to_daily_log` tool and do not append to a shared date-level
file. Send `{{"entry": "...", "agent": "opencode"}}` as JSON through stdin to:

```text
python3 {RUNTIME_DIRNAME}/{RUNTIME_FILENAME} --workspace . \\
  --agent opencode --adapter-version {MANAGED_ADAPTER_VERSION} append --payload -
```
""",
    },
    "echoes-search-vault-pages": {
        "signatures": (
            "echoes_search_vault_pages",
            "Read-Before-Write",
        ),
        "content": f"""---
name: echoes-search-vault-pages
description: Redirect legacy OpenCode search to the shared EchoesVault runtime.
---

<!-- Generated by EchoesVault protocol {PROTOCOL_VERSION}. Do not edit manually. -->
<!-- Managed adapter version: {MANAGED_ADAPTER_VERSION}. Legacy redirect. -->

# EchoesVault legacy search redirect

Do not call the legacy `echoes_search_vault_pages` tool. Search through the project runtime:

```text
python3 {RUNTIME_DIRNAME}/{RUNTIME_FILENAME} --workspace . \\
  --agent opencode --adapter-version {MANAGED_ADAPTER_VERSION} search <specific-query>
```

Read only the relevant returned pages and follow replacement links from deprecated pages.
""",
    },
    "echoes-create-or-update-page": {
        "signatures": (
            "echoes_create_or_update_page",
            "automatically updating the index",
        ),
        "content": f"""---
name: echoes-create-or-update-page
description: Redirect legacy OpenCode page writes to the shared EchoesVault runtime.
---

<!-- Generated by EchoesVault protocol {PROTOCOL_VERSION}. Do not edit manually. -->
<!-- Managed adapter version: {MANAGED_ADAPTER_VERSION}. Legacy redirect. -->

# EchoesVault legacy page redirect

Do not call the legacy `echoes_create_or_update_page` tool and never edit `index.md` directly.
Read an existing page, obtain its hash, then submit the complete page as JSON through stdin:

```text
python3 {RUNTIME_DIRNAME}/{RUNTIME_FILENAME} --workspace . \\
  --agent opencode --adapter-version {MANAGED_ADAPTER_VERSION} hash <filename>
python3 {RUNTIME_DIRNAME}/{RUNTIME_FILENAME} --workspace . \\
  --agent opencode --adapter-version {MANAGED_ADAPTER_VERSION} upsert --payload -
```
""",
    },
}


class EchoesError(RuntimeError):
    """An expected, user-actionable vault error."""


def now() -> datetime:
    return datetime.now().astimezone()


def utc_now() -> datetime:
    return datetime.now(timezone.utc)


def timestamp() -> str:
    return now().isoformat(timespec="seconds")


def today() -> str:
    return now().date().isoformat()


def utc_today() -> str:
    return utc_now().date().isoformat()


def resolve_workspace(value: Optional[str]) -> Path:
    candidate = Path(value or os.getcwd()).expanduser().resolve()
    if not candidate.is_dir():
        raise EchoesError(f"Workspace is not a directory: {candidate}")
    try:
        result = subprocess.run(
            ["git", "-C", str(candidate), "rev-parse", "--show-toplevel"],
            check=True,
            capture_output=True,
            text=True,
            timeout=3,
        )
        root = Path(result.stdout.strip()).resolve()
        if root.is_dir():
            return root
    except (FileNotFoundError, subprocess.SubprocessError):
        pass
    return candidate


def vault_paths(workspace: Path) -> dict[str, Path]:
    vault = workspace / VAULT_DIRNAME
    paths = {
        "workspace": workspace,
        "vault": vault,
        "raw": vault / "raw",
        "pages": vault / "pages",
        "daily": vault / "daily",
        "assets": vault / "assets",
        "index": vault / "index.md",
        "marker": vault / MARKER_FILENAME,
        "protocol": vault / PROTOCOL_FILENAME,
        "vaultIgnore": vault / ".gitignore",
        "state": workspace / STATE_RELATIVE_PATH,
        "openCodeState": workspace / OPENCODE_STATE_RELATIVE_PATH,
        "legacyState": workspace / LEGACY_STATE_RELATIVE_PATH,
        "lock": workspace / LOCK_RELATIVE_PATH,
        "runtimeDir": workspace / RUNTIME_DIRNAME,
        "runtime": workspace / RUNTIME_DIRNAME / RUNTIME_FILENAME,
        "runtimeIgnore": workspace / RUNTIME_DIRNAME / ".gitignore",
        "agentsGuide": workspace / "AGENTS.md",
        "claudeGuide": workspace / "CLAUDE.md",
        "claudeSkill": workspace / ".claude" / "skills" / "echoes-vault" / "SKILL.md",
        "openCodeSkill": workspace / ".opencode" / "skills" / "echoes-vault" / "SKILL.md",
        "openCodeCommands": workspace / ".opencode" / "commands",
    }
    for key, candidate in paths.items():
        if key == "workspace":
            continue
        try:
            candidate.resolve(strict=False).relative_to(workspace)
        except ValueError as exc:
            raise EchoesError(
                f"Managed path escapes the workspace through a symlink: {candidate}"
            ) from exc
    return paths


def default_state() -> dict[str, Any]:
    return {
        "version": STATE_VERSION,
        "protocolVersion": PROTOCOL_VERSION,
        "engineVersion": ENGINE_VERSION,
        "initialized": False,
        "session": {
            "started": False,
            "saved": False,
            "lastStart": None,
            "lastSave": None,
        },
        "stats": {"totalPages": 0, "totalDailyLogs": 0, "deprecatedPages": 0},
        "lastWriter": {"agent": None, "adapterVersion": None},
    }


def merge_state(raw: Any, legacy_agent: Optional[str] = None) -> dict[str, Any]:
    state = default_state()
    if not isinstance(raw, dict):
        return state
    state["initialized"] = bool(raw.get("initialized", False))
    session = raw.get("session")
    if isinstance(session, dict):
        for key in ("started", "saved", "lastStart", "lastSave"):
            if key in session:
                state["session"][key] = session[key]
    last_writer = raw.get("lastWriter")
    if isinstance(last_writer, dict):
        agent = last_writer.get("agent")
        adapter_version = last_writer.get("adapterVersion")
        state["lastWriter"] = {
            "agent": agent if isinstance(agent, str) and agent else None,
            "adapterVersion": (
                adapter_version
                if isinstance(adapter_version, str) and adapter_version
                else None
            ),
        }
    elif legacy_agent:
        legacy_version = raw.get("pluginVersion")
        state["lastWriter"] = {
            "agent": legacy_agent,
            "adapterVersion": legacy_version if isinstance(legacy_version, str) else None,
        }
    state["engineVersion"] = ENGINE_VERSION
    state["protocolVersion"] = PROTOCOL_VERSION
    return state


def read_state(paths: dict[str, Path]) -> dict[str, Any]:
    candidates = (
        (paths["state"], None),
        (paths["openCodeState"], "opencode"),
        (paths["legacyState"], "codex"),
    )
    for candidate, legacy_agent in candidates:
        try:
            return merge_state(
                json.loads(candidate.read_text(encoding="utf-8")), legacy_agent
            )
        except (FileNotFoundError, json.JSONDecodeError, OSError):
            continue
    return default_state()


def atomic_write(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.is_symlink():
        raise EchoesError(f"Refusing to replace a symbolic link: {path}")
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
    )
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    finally:
        if temporary.exists():
            temporary.unlink()


def append_ignore_lines(path: Path, lines: tuple[str, ...]) -> None:
    if path.is_symlink():
        raise EchoesError(f"Refusing to update a symbolic-link ignore file: {path}")
    try:
        existing = path.read_text(encoding="utf-8")
    except FileNotFoundError:
        existing = ""
    except OSError as exc:
        raise EchoesError(f"Cannot read ignore file {path}: {exc}") from exc
    present = set(existing.splitlines())
    missing = [line for line in lines if line not in present]
    if not missing:
        return
    prefix = existing.rstrip()
    block = "\n".join(missing)
    atomic_write(path, f"{prefix}\n{block}\n" if prefix else f"{block}\n")


def ensure_ignore_rules(paths: dict[str, Path]) -> None:
    append_ignore_lines(
        paths["vaultIgnore"],
        ("# Generated locally by EchoesVault", "/index.md"),
    )
    append_ignore_lines(
        paths["runtimeIgnore"],
        (
            "# EchoesVault runtime files",
            "/state.json",
            "/lock",
        ),
    )


def version_tuple(content: str, constant: str) -> Optional[tuple[int, int, int]]:
    match = re.search(
        rf'^{re.escape(constant)} = "(\d+)\.(\d+)\.(\d+)"$',
        content,
        re.MULTILINE,
    )
    if not match:
        return None
    return tuple(int(part) for part in match.groups())


def engine_version_from_content(content: str) -> Optional[tuple[int, int, int]]:
    return version_tuple(content, "ENGINE_VERSION") or version_tuple(
        content, "PLUGIN_VERSION"
    )


def protocol_version_from_content(content: str) -> Optional[tuple[int, int, int]]:
    return version_tuple(content, "PROTOCOL_VERSION")


def ensure_portable_runtime(paths: dict[str, Path]) -> bool:
    source = Path(__file__).resolve()
    target = paths["runtime"]
    if target.is_symlink():
        raise EchoesError(f"Refusing to replace a symbolic-link runtime: {target}")
    if source == target.resolve(strict=False):
        return False
    source_content = source.read_text(encoding="utf-8")
    if target.exists():
        try:
            target_content = target.read_text(encoding="utf-8")
        except (OSError, UnicodeError) as exc:
            raise EchoesError(f"Cannot read portable runtime {target}: {exc}") from exc
        target_version = engine_version_from_content(target_content)
        source_version = engine_version_from_content(source_content)
        if target_version is None or source_version is None:
            raise EchoesError(
                f"Refusing to overwrite an unrecognized portable runtime: {target}"
            )
        source_protocol = protocol_version_from_content(source_content)
        target_protocol = protocol_version_from_content(target_content)
        if source_protocol is None or target_protocol is None:
            raise EchoesError(
                f"Refusing to overwrite a runtime with unknown protocol metadata: {target}"
            )
        if source_protocol != target_protocol:
            raise EchoesError(
                f"Refusing to replace runtime protocol {target_protocol} with "
                f"incompatible protocol {source_protocol}."
            )
        if target_version > source_version:
            return False
        if target_version == source_version and target_content == source_content:
            return False
    atomic_write(target, source_content)
    return True


def write_generated_file(
    path: Path, content: str, marker: str, refuse_unknown: bool = True
) -> bool:
    if path.is_symlink():
        raise EchoesError(f"Refusing to replace a symbolic link: {path}")
    try:
        current = path.read_text(encoding="utf-8")
    except FileNotFoundError:
        current = ""
    except (OSError, UnicodeError) as exc:
        raise EchoesError(f"Cannot read managed file {path}: {exc}") from exc
    if current == content:
        return False
    if current and marker not in current and refuse_unknown:
        raise EchoesError(
            f"Refusing to overwrite non-EchoesVault file: {path}. Move it or add the protocol manually."
        )
    if current and marker not in current:
        return False
    atomic_write(path, content)
    return True


def write_opencode_command(path: Path, content: str, marker: str) -> bool:
    if path.is_symlink():
        raise EchoesError(f"Refusing to replace a symbolic-link OpenCode command: {path}")
    try:
        current = path.read_text(encoding="utf-8")
    except FileNotFoundError:
        current = ""
    except (OSError, UnicodeError) as exc:
        raise EchoesError(f"Cannot read OpenCode command {path}: {exc}") from exc
    if current == content:
        return False
    legacy_signatures = (
        "echoes_activate_vault",
        "echoes_start_session",
        "commit_memory_to_echoes_vault",
        "SYSTEM MESSAGE: Vault Status Report",
    )
    if current and marker not in current and not any(
        signature in current for signature in legacy_signatures
    ):
        return False
    atomic_write(path, content)
    return True


def reconcile_legacy_opencode_skills(
    paths: dict[str, Path], protocol_marker: str
) -> tuple[list[str], list[str]]:
    redirected: list[str] = []
    conflicts: list[str] = []
    skill_root = paths["workspace"] / ".opencode" / "skills"
    for name, definition in LEGACY_OPENCODE_SKILLS.items():
        skill_path = skill_root / name / "SKILL.md"
        relative = str(skill_path.relative_to(paths["workspace"]))
        if not skill_path.exists():
            continue
        if skill_path.is_symlink() or not skill_path.is_file():
            conflicts.append(relative)
            continue
        try:
            current = skill_path.read_text(encoding="utf-8")
        except (OSError, UnicodeError):
            conflicts.append(relative)
            continue
        replacement = definition["content"]
        signatures = definition["signatures"]
        owned = protocol_marker in current and "Legacy redirect" in current
        recognized_legacy = any(signature in current for signature in signatures)
        if not owned and not recognized_legacy:
            conflicts.append(relative)
            continue
        if current != replacement:
            atomic_write(skill_path, replacement)
            redirected.append(relative)
    return redirected, conflicts


def adapter_configuration_conflicts(paths: dict[str, Path]) -> list[str]:
    protocol_marker = f"Generated by EchoesVault protocol {PROTOCOL_VERSION}"
    conflicts: list[str] = []
    managed = [paths["claudeSkill"], paths["openCodeSkill"]]
    managed.extend(
        paths["openCodeCommands"] / filename for filename in OPENCODE_COMMANDS
    )
    for path in managed:
        if not path.exists():
            continue
        relative = str(path.relative_to(paths["workspace"]))
        if path.is_symlink() or not path.is_file():
            conflicts.append(relative)
            continue
        try:
            if protocol_marker not in path.read_text(encoding="utf-8"):
                conflicts.append(relative)
        except (OSError, UnicodeError):
            conflicts.append(relative)
    skill_root = paths["workspace"] / ".opencode" / "skills"
    for name in LEGACY_OPENCODE_SKILLS:
        skill_path = skill_root / name / "SKILL.md"
        if not skill_path.exists():
            continue
        relative = str(skill_path.relative_to(paths["workspace"]))
        if skill_path.is_symlink() or not skill_path.is_file():
            conflicts.append(relative)
            continue
        try:
            content = skill_path.read_text(encoding="utf-8")
        except (OSError, UnicodeError):
            conflicts.append(relative)
            continue
        if protocol_marker not in content or "Legacy redirect" not in content:
            conflicts.append(relative)
    return sorted(set(conflicts))


def ensure_root_guide(path: Path) -> bool:
    if path.is_symlink():
        raise EchoesError(f"Refusing to update a symbolic-link agent guide: {path}")
    try:
        current = path.read_text(encoding="utf-8")
    except FileNotFoundError:
        current = ""
    except (OSError, UnicodeError) as exc:
        raise EchoesError(f"Cannot read agent guide {path}: {exc}") from exc
    start_count = current.count(AGENT_GUIDE_START)
    end_count = current.count(AGENT_GUIDE_END)
    if start_count != end_count or start_count > 1:
        raise EchoesError(f"Malformed EchoesVault managed block in {path}.")
    if start_count == 1:
        pattern = re.compile(
            re.escape(AGENT_GUIDE_START) + r".*?" + re.escape(AGENT_GUIDE_END),
            re.DOTALL,
        )
        updated = pattern.sub(AGENT_GUIDE_BLOCK, current)
    else:
        prefix = current.rstrip()
        updated = f"{prefix}\n\n{AGENT_GUIDE_BLOCK}\n" if prefix else AGENT_GUIDE_BLOCK + "\n"
    if updated == current:
        return False
    atomic_write(path, updated)
    return True


def configure_agent_adapters(paths: dict[str, Path]) -> dict[str, Any]:
    ensure_ignore_rules(paths)
    runtime_updated = ensure_portable_runtime(paths)
    protocol_marker = f"Generated by EchoesVault protocol {PROTOCOL_VERSION}"
    protocol_updated = write_generated_file(
        paths["protocol"], AGENT_PROTOCOL, protocol_marker
    )
    guides_updated = [
        str(path.relative_to(paths["workspace"]))
        for path in (paths["agentsGuide"], paths["claudeGuide"])
        if ensure_root_guide(path)
    ]
    skills_updated = [
        str(path.relative_to(paths["workspace"]))
        for path in (paths["claudeSkill"], paths["openCodeSkill"])
        if write_generated_file(
            path, AGENT_ADAPTER_SKILL, protocol_marker, refuse_unknown=False
        )
    ]
    commands_updated = [
        str(path.relative_to(paths["workspace"]))
        for filename, content in OPENCODE_COMMANDS.items()
        for path in (paths["openCodeCommands"] / filename,)
        if write_opencode_command(path, content, protocol_marker)
    ]
    legacy_skills_redirected, adapter_conflicts = reconcile_legacy_opencode_skills(
        paths, protocol_marker
    )
    return {
        "protocolVersion": PROTOCOL_VERSION,
        "engineVersion": ENGINE_VERSION,
        "managedAdapterVersion": MANAGED_ADAPTER_VERSION,
        "codexAdapterVersion": MANAGED_ADAPTER_VERSION,
        "runtimeUpdated": runtime_updated,
        "protocolUpdated": protocol_updated,
        "guidesUpdated": guides_updated,
        "skillsUpdated": skills_updated,
        "commandsUpdated": commands_updated,
        "legacySkillsRedirected": legacy_skills_redirected,
        "adapterConflicts": adapter_conflicts,
    }


def read_marker(paths: dict[str, Path]) -> Optional[dict[str, Any]]:
    try:
        marker = json.loads(paths["marker"].read_text(encoding="utf-8"))
    except FileNotFoundError:
        return None
    except (OSError, json.JSONDecodeError) as exc:
        raise EchoesError(f"Invalid EchoesVault marker: {exc}") from exc
    if not isinstance(marker, dict):
        raise EchoesError("Invalid EchoesVault marker: root must be an object.")
    return marker


def ensure_protocol_marker(paths: dict[str, Path]) -> bool:
    marker = read_marker(paths)
    if marker is not None:
        version = marker.get("protocolVersion")
        if version not in (None, PROTOCOL_VERSION):
            raise EchoesError(
                f"Unsupported EchoesVault protocol {version!r}; this runtime supports "
                f"{PROTOCOL_VERSION}. Upgrade the agent adapter before writing."
            )
        if marker == VAULT_MARKER:
            return False
    atomic_write(
        paths["marker"], json.dumps(VAULT_MARKER, ensure_ascii=False, indent=2) + "\n"
    )
    return True


def vault_is_initialized(paths: dict[str, Path]) -> bool:
    return (
        paths["marker"].is_file()
        or paths["index"].is_file()
        or (paths["pages"].is_dir() and read_state(paths)["initialized"])
    )


def ensure_structure(paths: dict[str, Path], create_marker: bool = False) -> None:
    for key in ("raw", "pages", "daily", "assets"):
        paths[key].mkdir(parents=True, exist_ok=True)
    ensure_ignore_rules(paths)
    if create_marker:
        ensure_protocol_marker(paths)


def lock_is_stale(path: Path) -> bool:
    try:
        return time.time() - path.stat().st_mtime > STALE_LOCK_SECONDS
    except OSError:
        return False


@contextmanager
def vault_lock(paths: dict[str, Path]) -> Iterator[None]:
    path = paths["lock"]
    path.parent.mkdir(parents=True, exist_ok=True)
    token = f"{os.getpid()}-{secrets.token_hex(8)}"
    deadline = time.monotonic() + LOCK_WAIT_SECONDS
    while True:
        try:
            descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
            with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
                handle.write(token + "\n")
            break
        except FileExistsError:
            if lock_is_stale(path):
                try:
                    path.unlink()
                except FileNotFoundError:
                    pass
                continue
            if time.monotonic() >= deadline:
                raise EchoesError(
                    "Another EchoesVault operation is still running. Retry after it finishes."
                )
            time.sleep(0.05)
    try:
        yield
    finally:
        try:
            if path.read_text(encoding="utf-8").strip() == token:
                path.unlink()
        except FileNotFoundError:
            pass


def stable_name(value: str) -> str:
    return unicodedata.normalize("NFC", value)


def stable_path_key(path: Path) -> tuple[str, str]:
    normalized = stable_name(path.name)
    return normalized.casefold(), normalized


def markdown_files(directory: Path) -> list[Path]:
    if not directory.is_dir():
        return []
    return sorted(
        (
            item
            for item in directory.iterdir()
            if item.is_file() and not item.is_symlink() and item.suffix.casefold() == ".md"
        ),
        key=stable_path_key,
    )


def daily_markdown_files(directory: Path) -> list[Path]:
    if not directory.is_dir():
        return []
    files = [
        item
        for item in directory.rglob("*.md")
        if item.is_file() and not item.is_symlink()
    ]
    return sorted(
        files,
        key=lambda item: stable_name(item.relative_to(directory).as_posix()),
    )


def decode_scalar(value: str, key: str) -> str:
    stripped = value.strip()
    if not stripped:
        raise EchoesError(f"YAML frontmatter key {key!r} must have an inline value.")
    if stripped.startswith('"'):
        try:
            decoded = json.loads(stripped)
        except json.JSONDecodeError as exc:
            raise EchoesError(f"Invalid quoted YAML value for {key}: {exc}") from exc
        if not isinstance(decoded, str):
            raise EchoesError(f"YAML frontmatter key {key!r} must be a string.")
        return decoded
    if stripped.startswith("'") and stripped.endswith("'") and len(stripped) >= 2:
        return stripped[1:-1].replace("''", "'")
    return stripped


def normalize_summary(value: Any, slug: Optional[str] = None) -> str:
    if not isinstance(value, str) or not value.strip():
        raise EchoesError("Page summary must be a non-empty single-line string.")
    candidate = value.strip()
    match = INDEX_ENTRY_RE.match(candidate)
    if match:
        if slug is not None and match.group(1) != slug:
            raise EchoesError(
                f"Index link [[{match.group(1)}]] does not match page [[{slug}]]."
            )
        candidate = match.group(2).strip()
    if "\n" in candidate or "\r" in candidate:
        raise EchoesError("Page summary must fit on one line.")
    summary = " ".join(candidate.split())
    if not summary:
        raise EchoesError("Page summary cannot be empty.")
    if len(summary) > SUMMARY_MAX_LENGTH:
        raise EchoesError(
            f"Page summary exceeds {SUMMARY_MAX_LENGTH} characters ({len(summary)})."
        )
    if CONFLICT_MARKER_RE.search(summary):
        raise EchoesError("Page summary contains a Git conflict marker.")
    return summary


def parse_frontmatter(
    content: Any, require_summary: bool = True
) -> tuple[str, dict[str, str]]:
    if not isinstance(content, str) or not content.strip():
        raise EchoesError("Page content must be a non-empty string.")
    normalized = content.strip() + "\n"
    if CONFLICT_MARKER_RE.search(normalized):
        raise EchoesError("Page contains unresolved Git conflict markers.")
    lines = normalized.splitlines()
    if not lines or lines[0] != "---":
        raise EchoesError("Every page must begin with YAML frontmatter.")
    try:
        closing = lines.index("---", 1)
    except ValueError as exc:
        raise EchoesError("YAML frontmatter is missing its closing '---'.") from exc
    if closing == 1:
        raise EchoesError("YAML frontmatter cannot be empty.")
    raw_values: dict[str, str] = {}
    valued_keys: set[str] = set()
    frontmatter_lines = lines[1:closing]
    for position, line in enumerate(frontmatter_lines):
        if not line.strip() or line.lstrip().startswith("#") or line[0].isspace():
            continue
        match = re.match(r"^([A-Za-z][A-Za-z0-9_-]*):\s*(.*)$", line)
        if not match:
            continue
        key, inline = match.group(1), match.group(2)
        if key in raw_values:
            raise EchoesError(f"YAML frontmatter contains duplicate key: {key}")
        raw_values[key] = inline
        if inline.strip():
            valued_keys.add(key)
            continue
        for following in frontmatter_lines[position + 1 :]:
            if not following.strip() or following.lstrip().startswith("#"):
                continue
            if following[0].isspace():
                valued_keys.add(key)
            break
    required = REQUIRED_FRONTMATTER if require_summary else REQUIRED_FRONTMATTER[:-1]
    missing = [key for key in required if key not in valued_keys]
    if missing:
        raise EchoesError(f"YAML frontmatter is missing required keys: {', '.join(missing)}")
    metadata = {
        "type": decode_scalar(raw_values["type"], "type"),
        "status": decode_scalar(raw_values["status"], "status").casefold(),
    }
    if "summary" in valued_keys:
        metadata["summary"] = normalize_summary(
            decode_scalar(raw_values["summary"], "summary")
        )
    return normalized, metadata


def validate_frontmatter(content: Any) -> str:
    normalized, _metadata = parse_frontmatter(content, require_summary=True)
    return normalized


def inject_summary(content: str, summary: str) -> str:
    normalized, metadata = parse_frontmatter(content, require_summary=False)
    if "summary" in metadata:
        return normalized
    lines = normalized.splitlines()
    closing = lines.index("---", 1)
    lines.insert(closing, f"summary: {json.dumps(summary, ensure_ascii=False)}")
    migrated = "\n".join(lines).rstrip() + "\n"
    validate_frontmatter(migrated)
    return migrated


def parse_index(content: str) -> list[tuple[str, str, str]]:
    entries: list[tuple[str, str, str]] = []
    for line in content.splitlines():
        match = INDEX_ENTRY_RE.match(line)
        if match:
            entries.append((match.group(1), match.group(2).strip(), line))
    return entries


def legacy_index_descriptions(paths: dict[str, Path]) -> dict[str, str]:
    try:
        content = paths["index"].read_text(encoding="utf-8")
    except OSError:
        return {}
    descriptions: dict[str, str] = {}
    duplicates: set[str] = set()
    for slug, description, _line in parse_index(content):
        if slug in descriptions:
            duplicates.add(slug)
        descriptions[slug] = description
    for slug in duplicates:
        descriptions.pop(slug, None)
    return descriptions


def page_source_map(
    paths: dict[str, Path], overrides: Optional[dict[Path, str]] = None
) -> dict[Path, str]:
    sources: dict[Path, str] = {}
    for page in markdown_files(paths["pages"]):
        try:
            sources[page] = page.read_text(encoding="utf-8")
        except (OSError, UnicodeError) as exc:
            raise EchoesError(f"Cannot read page {page.name}: {exc}") from exc
    if overrides:
        sources.update(overrides)
    return sources


def validate_unique_page_names(paths: Iterable[Path]) -> None:
    seen: dict[str, str] = {}
    for path in paths:
        normalized = stable_name(path.stem)
        folded = normalized.casefold()
        if folded in seen:
            raise EchoesError(
                f"Page filenames collide across filesystems: {seen[folded]!r} and {normalized!r}."
            )
        seen[folded] = normalized


def render_index_from_sources(sources: dict[Path, str]) -> str:
    validate_unique_page_names(sources)
    rows: list[tuple[tuple[str, str], str]] = []
    for page, content in sources.items():
        try:
            _normalized, metadata = parse_frontmatter(content, require_summary=True)
        except EchoesError as exc:
            raise EchoesError(f"Cannot generate index from {page.name}: {exc}") from exc
        slug = stable_name(page.stem)
        summary = metadata["summary"]
        if metadata["status"] == "deprecated" and not summary.casefold().startswith("deprecated"):
            summary = f"DEPRECATED — {summary}"
        rows.append((stable_path_key(page), f"- [[{slug}]]: {summary}"))
    rows.sort(key=lambda item: item[0])
    body = "\n".join(row for _key, row in rows)
    return DEFAULT_INDEX + ("\n" + body + "\n" if body else "")


def migrate_legacy_summaries(paths: dict[str, Path]) -> list[str]:
    sources = page_source_map(paths)
    validate_unique_page_names(sources)
    legacy = legacy_index_descriptions(paths)
    migrations: list[tuple[Path, str]] = []
    errors: list[str] = []
    for page, content in sources.items():
        try:
            normalized, metadata = parse_frontmatter(content, require_summary=False)
        except EchoesError as exc:
            errors.append(f"{page.name}: {exc}")
            continue
        if "summary" in metadata:
            continue
        description = legacy.get(page.stem)
        if not description:
            errors.append(
                f"{page.name}: missing summary and no legacy index description is available"
            )
            continue
        try:
            summary = normalize_summary(description, page.stem)
            migrations.append((page, inject_summary(normalized, summary)))
        except EchoesError as exc:
            errors.append(f"{page.name}: {exc}")
    if errors:
        raise EchoesError("Index metadata errors: " + "; ".join(errors))
    for page, content in migrations:
        atomic_write(page, content)
    return [page.name for page, _content in migrations]


def refresh_index(
    paths: dict[str, Path], allow_legacy_migration: bool = False
) -> dict[str, Any]:
    migrated = migrate_legacy_summaries(paths) if allow_legacy_migration else []
    generated = render_index_from_sources(page_source_map(paths))
    try:
        current = paths["index"].read_text(encoding="utf-8")
    except OSError:
        current = ""
    rebuilt = current != generated
    if rebuilt:
        atomic_write(paths["index"], generated)
    return {
        "rebuilt": rebuilt,
        "migratedSummaries": migrated,
        "pageCount": len(parse_index(generated)),
        "sha256": hashlib.sha256(generated.encode("utf-8")).hexdigest(),
    }


def inspect_index(paths: dict[str, Path]) -> tuple[dict[str, Any], Optional[str]]:
    """Compare the index with its deterministic rendering without writing anything."""
    try:
        generated = render_index_from_sources(page_source_map(paths))
    except EchoesError as exc:
        return {
            "rebuilt": False,
            "needsRefresh": False,
            "migratedSummaries": [],
            "pageCount": len(markdown_files(paths["pages"])),
            "sha256": None,
        }, str(exc)
    try:
        current = paths["index"].read_text(encoding="utf-8")
    except OSError:
        current = ""
    needs_refresh = current != generated
    return {
        "rebuilt": False,
        "needsRefresh": needs_refresh,
        "migratedSummaries": [],
        "pageCount": len(parse_index(generated)),
        "sha256": hashlib.sha256(generated.encode("utf-8")).hexdigest(),
    }, None


def legacy_vault_detected(paths: dict[str, Path]) -> bool:
    if paths["marker"].is_file():
        return False
    return (
        paths["index"].is_file()
        or bool(markdown_files(paths["pages"]))
        or bool(daily_markdown_files(paths["daily"]))
    )


def require_initialized(paths: dict[str, Path]) -> dict[str, Any]:
    marker = read_marker(paths)
    if marker is None:
        if legacy_vault_detected(paths):
            raise EchoesError(
                "Legacy EchoesVault detected. Run the echoes-init workflow to migrate it."
            )
        raise EchoesError("EchoesVault is not initialized. Run the echoes-init workflow first.")
    version = marker.get("protocolVersion")
    if version != PROTOCOL_VERSION:
        raise EchoesError(
            f"Unsupported EchoesVault protocol {version!r}; this runtime supports "
            f"{PROTOCOL_VERSION}. Upgrade the agent adapter before writing."
        )
    if marker.get("schemaVersion") != SCHEMA_VERSION:
        raise EchoesError(
            f"Unsupported EchoesVault marker schema {marker.get('schemaVersion')!r}; "
            f"this engine supports schema {SCHEMA_VERSION}. Run init or migrate explicitly."
        )
    for key in ("generatedIndex", "dailyLayout", "runtime", "requiredFrontmatter"):
        if marker.get(key) != VAULT_MARKER[key]:
            raise EchoesError(
                f"Invalid EchoesVault marker field {key!r}. Run init or migrate explicitly."
            )
    state = read_state(paths)
    state["initialized"] = True
    return state


def vault_inventory(vault: Path) -> dict[str, Any]:
    """Collect size and filesystem integrity without following symbolic links."""
    total_bytes = 0
    total_files = 0
    markdown_count = 0
    symlinks: list[str] = []
    unreadable: list[str] = []
    latest_mtime: Optional[float] = None
    if not vault.is_dir():
        return {
            "totalBytes": 0,
            "totalFiles": 0,
            "markdownFiles": 0,
            "symlinks": [],
            "unreadableFiles": [],
            "lastModified": None,
        }
    for root_name, directory_names, file_names in os.walk(vault, followlinks=False):
        root = Path(root_name)
        safe_directories: list[str] = []
        for directory_name in directory_names:
            directory = root / directory_name
            if directory.is_symlink():
                symlinks.append(str(directory.relative_to(vault)))
            else:
                safe_directories.append(directory_name)
        directory_names[:] = safe_directories
        for file_name in file_names:
            file_path = root / file_name
            relative = str(file_path.relative_to(vault))
            if file_path.is_symlink():
                symlinks.append(relative)
                continue
            try:
                stat = file_path.stat()
            except OSError:
                unreadable.append(relative)
                continue
            if not file_path.is_file():
                continue
            total_files += 1
            total_bytes += stat.st_size
            markdown_count += int(file_path.suffix.casefold() == ".md")
            latest_mtime = max(latest_mtime or stat.st_mtime, stat.st_mtime)
    return {
        "totalBytes": total_bytes,
        "totalFiles": total_files,
        "markdownFiles": markdown_count,
        "symlinks": sorted(symlinks),
        "unreadableFiles": sorted(unreadable),
        "lastModified": (
            datetime.fromtimestamp(latest_mtime).astimezone().isoformat(timespec="seconds")
            if latest_mtime is not None
            else None
        ),
    }


def format_bytes(value: int) -> str:
    amount = float(max(0, value))
    units = ("B", "KB", "MB", "GB", "TB")
    unit = units[0]
    for unit in units:
        if amount < 1024 or unit == units[-1]:
            break
        amount /= 1024
    return f"{int(amount)} {unit}" if unit == "B" else f"{amount:.1f} {unit}"


def state_file_health(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {"exists": False, "valid": False, "error": "state file is missing"}
    if path.is_symlink():
        return {"exists": True, "valid": False, "error": "state file is a symbolic link"}
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        return {"exists": True, "valid": False, "error": str(exc)}
    if not isinstance(raw, dict):
        return {"exists": True, "valid": False, "error": "state root is not an object"}
    if raw.get("version") != STATE_VERSION:
        return {
            "exists": True,
            "valid": False,
            "error": f"unsupported state version: {raw.get('version')!r}",
        }
    if raw.get("protocolVersion") != PROTOCOL_VERSION:
        return {
            "exists": True,
            "valid": False,
            "error": f"state protocol mismatch: {raw.get('protocolVersion')!r}",
        }
    if raw.get("engineVersion") != ENGINE_VERSION:
        return {
            "exists": True,
            "valid": False,
            "error": f"state engine mismatch: {raw.get('engineVersion')!r}",
        }
    last_writer = raw.get("lastWriter")
    if not isinstance(last_writer, dict):
        return {
            "exists": True,
            "valid": False,
            "error": "state lastWriter must be an object",
        }
    return {"exists": True, "valid": True, "error": None}


def regular_file_contains(path: Path, marker: str) -> bool:
    if not path.is_file() or path.is_symlink():
        return False
    try:
        return marker in path.read_text(encoding="utf-8")
    except (OSError, UnicodeError):
        return False


def regular_file_contains_all(path: Path, markers: Iterable[str]) -> bool:
    if not path.is_file() or path.is_symlink():
        return False
    try:
        content = path.read_text(encoding="utf-8")
    except (OSError, UnicodeError):
        return False
    return all(marker in content for marker in markers)


def format_version(value: Optional[tuple[int, int, int]]) -> Optional[str]:
    return ".".join(str(part) for part in value) if value is not None else None


def configured_version(value: str) -> tuple[int, int, int]:
    major, minor, patch = value.split(".")
    return int(major), int(minor), int(patch)


def project_runtime_health(paths: dict[str, Path]) -> dict[str, Any]:
    runtime = paths["runtime"]
    if not runtime.exists():
        return {
            "exists": False,
            "recognized": False,
            "engineVersion": None,
            "protocolVersion": None,
            "compatible": False,
            "current": False,
            "error": "project runtime is missing",
        }
    if runtime.is_symlink() or not runtime.is_file():
        return {
            "exists": True,
            "recognized": False,
            "engineVersion": None,
            "protocolVersion": None,
            "compatible": False,
            "current": False,
            "error": "project runtime is not a regular file",
        }
    try:
        content = runtime.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as exc:
        return {
            "exists": True,
            "recognized": False,
            "engineVersion": None,
            "protocolVersion": None,
            "compatible": False,
            "current": False,
            "error": str(exc),
        }
    engine = engine_version_from_content(content)
    protocol = protocol_version_from_content(content)
    recognized = engine is not None and protocol is not None
    compatible = protocol == configured_version(PROTOCOL_VERSION)
    current = compatible and engine == configured_version(ENGINE_VERSION)
    return {
        "exists": True,
        "recognized": recognized,
        "engineVersion": format_version(engine),
        "protocolVersion": format_version(protocol),
        "compatible": compatible,
        "current": current,
        "error": None if recognized else "project runtime version metadata is missing",
    }


def find_conflict_markers(vault: Path) -> list[str]:
    conflicts: list[str] = []
    if not vault.is_dir():
        return conflicts
    for path in sorted(vault.rglob("*.md")):
        if not path.is_file() or path.is_symlink():
            continue
        try:
            content = path.read_text(encoding="utf-8")
        except (OSError, UnicodeError):
            continue
        if CONFLICT_MARKER_RE.search(content):
            conflicts.append(path.relative_to(vault).as_posix())
    return conflicts


def today_log_content(paths: dict[str, Path]) -> str:
    candidates: list[Path] = []
    for date_value in dict.fromkeys((today(), utc_today())):
        legacy = paths["daily"] / f"{date_value}.md"
        if legacy.is_file() and not legacy.is_symlink():
            candidates.append(legacy)
        directory = paths["daily"] / date_value
        if directory.is_dir():
            candidates.extend(daily_markdown_files(directory))
    contents: list[str] = []
    for path in candidates:
        try:
            contents.append(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeError):
            continue
    return "\n".join(contents)


def repository_relative_files(root: Path) -> list[Path]:
    if not root.is_dir() or root.is_symlink():
        return []
    return sorted(
        (
            path
            for path in root.rglob("*")
            if path.is_file() and not path.is_symlink()
        ),
        key=lambda path: stable_name(path.as_posix()),
    )


def durable_git_candidates(paths: dict[str, Path]) -> list[str]:
    candidates = [
        paths["marker"],
        paths["vaultIgnore"],
        paths["protocol"],
        paths["runtimeIgnore"],
        paths["runtime"],
        paths["agentsGuide"],
        paths["claudeGuide"],
        paths["claudeSkill"],
        paths["openCodeSkill"],
    ]
    candidates.extend(paths["openCodeCommands"] / name for name in OPENCODE_COMMANDS)
    legacy_skill_root = paths["workspace"] / ".opencode" / "skills"
    candidates.extend(
        legacy_skill_root / name / "SKILL.md" for name in LEGACY_OPENCODE_SKILLS
    )
    for key in ("pages", "daily", "assets", "raw"):
        candidates.extend(repository_relative_files(paths[key]))
    relative = {
        path.relative_to(paths["workspace"]).as_posix()
        for path in candidates
        if path.is_file() and not path.is_symlink()
    }
    return sorted(relative)


def parse_nul_paths(value: str) -> set[str]:
    return {item for item in value.split("\0") if item}


def git_readiness(paths: dict[str, Path]) -> dict[str, Any]:
    workspace = paths["workspace"]
    try:
        inside = subprocess.run(
            ["git", "-C", str(workspace), "rev-parse", "--is-inside-work-tree"],
            check=True,
            capture_output=True,
            text=True,
            timeout=3,
        )
    except (FileNotFoundError, subprocess.SubprocessError):
        return {
            "available": False,
            "ready": True,
            "trackedLocalFiles": [],
            "untrackedDurableFiles": [],
            "ignoredDurableFiles": [],
            "runtimeIgnored": False,
            "openCodeCommandsIgnored": [],
            "openCodeSkillsIgnored": [],
            "suggestedCommands": [],
        }
    if inside.stdout.strip() != "true":
        return {
            "available": False,
            "ready": True,
            "trackedLocalFiles": [],
            "untrackedDurableFiles": [],
            "ignoredDurableFiles": [],
            "runtimeIgnored": False,
            "openCodeCommandsIgnored": [],
            "openCodeSkillsIgnored": [],
            "suggestedCommands": [],
        }
    try:
        tracked_result = subprocess.run(
            ["git", "-C", str(workspace), "ls-files", "-z"],
            check=True,
            capture_output=True,
            text=True,
            timeout=5,
        )
        untracked_result = subprocess.run(
            [
                "git",
                "-C",
                str(workspace),
                "ls-files",
                "--others",
                "--exclude-standard",
                "-z",
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=5,
        )
    except (FileNotFoundError, subprocess.SubprocessError):
        return {
            "available": False,
            "ready": True,
            "trackedLocalFiles": [],
            "untrackedDurableFiles": [],
            "ignoredDurableFiles": [],
            "runtimeIgnored": False,
            "openCodeCommandsIgnored": [],
            "openCodeSkillsIgnored": [],
            "suggestedCommands": [],
        }
    tracked = parse_nul_paths(tracked_result.stdout)
    untracked = parse_nul_paths(untracked_result.stdout)
    durable = durable_git_candidates(paths)
    ignored: set[str] = set()
    if durable:
        try:
            ignored_result = subprocess.run(
                [
                    "git",
                    "-C",
                    str(workspace),
                    "check-ignore",
                    "--no-index",
                    "--stdin",
                    "-z",
                ],
                input="\0".join(durable) + "\0",
                check=False,
                capture_output=True,
                text=True,
                timeout=5,
            )
            ignored = parse_nul_paths(ignored_result.stdout)
        except (FileNotFoundError, subprocess.SubprocessError):
            ignored = set()
    local_only = {
        paths["index"].relative_to(workspace).as_posix(),
        paths["state"].relative_to(workspace).as_posix(),
        paths["lock"].relative_to(workspace).as_posix(),
        paths["openCodeState"].relative_to(workspace).as_posix(),
        paths["legacyState"].relative_to(workspace).as_posix(),
    }
    tracked_local = sorted(local_only & tracked)
    untracked_durable = sorted(set(durable) & untracked)
    ignored_durable = sorted(set(durable) & ignored)
    runtime_relative = paths["runtime"].relative_to(workspace).as_posix()
    ignored_commands = [
        path for path in ignored_durable if path.startswith(".opencode/commands/")
    ]
    ignored_skills = [
        path for path in ignored_durable if path.startswith(".opencode/skills/")
    ]
    suggestions: list[str] = []
    if tracked_local:
        suggestions.append(
            "git rm --cached -- "
            + " ".join(shlex.quote(path) for path in tracked_local)
        )
    if ignored_durable:
        suggestions.append(
            "git add -f -- "
            + " ".join(shlex.quote(path) for path in ignored_durable)
        )
    addable = [path for path in untracked_durable if path not in ignored]
    if addable:
        suggestions.append(
            "git add -- " + " ".join(shlex.quote(path) for path in addable)
        )
    ready = not tracked_local and not untracked_durable and not ignored_durable
    return {
        "available": True,
        "ready": ready,
        "trackedLocalFiles": tracked_local,
        "untrackedDurableFiles": untracked_durable,
        "ignoredDurableFiles": ignored_durable,
        "runtimeIgnored": runtime_relative in ignored,
        "openCodeCommandsIgnored": ignored_commands,
        "openCodeSkillsIgnored": ignored_skills,
        "suggestedCommands": suggestions,
    }


def marker_health(paths: dict[str, Path]) -> dict[str, Any]:
    try:
        marker = read_marker(paths)
    except EchoesError as exc:
        return {"exists": True, "valid": False, "value": None, "error": str(exc)}
    if marker is None:
        return {
            "exists": False,
            "valid": False,
            "value": None,
            "error": "initialization marker is missing",
        }
    if marker.get("protocolVersion") != PROTOCOL_VERSION:
        return {
            "exists": True,
            "valid": False,
            "value": marker,
            "error": f"unsupported marker protocol: {marker.get('protocolVersion')!r}",
        }
    if marker.get("schemaVersion") != SCHEMA_VERSION:
        return {
            "exists": True,
            "valid": False,
            "value": marker,
            "error": f"unsupported marker schema: {marker.get('schemaVersion')!r}",
        }
    for key in ("generatedIndex", "dailyLayout", "runtime", "requiredFrontmatter"):
        if marker.get(key) != VAULT_MARKER[key]:
            return {
                "exists": True,
                "valid": False,
                "value": marker,
                "error": f"invalid marker field: {key}",
            }
    return {"exists": True, "valid": True, "value": marker, "error": None}


def collect_health(
    paths: dict[str, Path], index_error: Optional[str] = None
) -> dict[str, Any]:
    index_inspection, detected_index_error = inspect_index(paths)
    if index_error is None:
        index_error = detected_index_error
    pages = markdown_files(paths["pages"])
    daily = daily_markdown_files(paths["daily"])
    deprecated: list[str] = []
    invalid_frontmatter: list[str] = []
    page_slugs = {stable_name(page.stem) for page in pages}
    for page in pages:
        try:
            content = page.read_text(encoding="utf-8")
            _normalized, metadata = parse_frontmatter(content, require_summary=True)
            if metadata["status"] == "deprecated":
                deprecated.append(page.stem)
        except (EchoesError, OSError, UnicodeError):
            invalid_frontmatter.append(page.stem)
    try:
        index_content = paths["index"].read_text(encoding="utf-8")
    except OSError:
        index_content = ""
    entries = parse_index(index_content)
    indexed_slugs = [slug for slug, _description, _line in entries]
    seen: set[str] = set()
    duplicates: list[str] = []
    for slug in indexed_slugs:
        if slug in seen and slug not in duplicates:
            duplicates.append(slug)
        seen.add(slug)
    runtime_health = project_runtime_health(paths)
    marker_status = marker_health(paths)
    adapter_conflicts = adapter_configuration_conflicts(paths)
    git_status = git_readiness(paths)
    required_paths = {
        "index.md": paths["index"].is_file() and not paths["index"].is_symlink(),
        MARKER_FILENAME: marker_status["valid"],
        PROTOCOL_FILENAME: regular_file_contains(
            paths["protocol"], f"Generated by EchoesVault protocol {PROTOCOL_VERSION}"
        ),
        "pages/": paths["pages"].is_dir() and not paths["pages"].is_symlink(),
        "daily/": paths["daily"].is_dir() and not paths["daily"].is_symlink(),
        "assets/": paths["assets"].is_dir() and not paths["assets"].is_symlink(),
        "raw/": paths["raw"].is_dir() and not paths["raw"].is_symlink(),
        f"{RUNTIME_DIRNAME}/{RUNTIME_FILENAME}": runtime_health["compatible"],
        "AGENTS.md": regular_file_contains_all(
            paths["agentsGuide"],
            (AGENT_GUIDE_START, f"Managed adapter version: {MANAGED_ADAPTER_VERSION}"),
        ),
        "CLAUDE.md": regular_file_contains_all(
            paths["claudeGuide"],
            (AGENT_GUIDE_START, f"Managed adapter version: {MANAGED_ADAPTER_VERSION}"),
        ),
        ".claude/skills/echoes-vault/SKILL.md": regular_file_contains_all(
            paths["claudeSkill"],
            (
                f"Generated by EchoesVault protocol {PROTOCOL_VERSION}",
                f"Managed adapter version: {MANAGED_ADAPTER_VERSION}",
            ),
        ),
        ".opencode/skills/echoes-vault/SKILL.md": regular_file_contains_all(
            paths["openCodeSkill"],
            (
                f"Generated by EchoesVault protocol {PROTOCOL_VERSION}",
                f"Managed adapter version: {MANAGED_ADAPTER_VERSION}",
            ),
        ),
        **{
            f".opencode/commands/{filename}": (
                regular_file_contains_all(
                    paths["openCodeCommands"] / filename,
                    (
                        f"Generated by EchoesVault protocol {PROTOCOL_VERSION}",
                        f"Managed adapter version: {MANAGED_ADAPTER_VERSION}",
                    ),
                )
            )
            for filename in OPENCODE_COMMANDS
        },
    }
    inventory = vault_inventory(paths["vault"])
    state_health = state_file_health(paths["state"])
    git_issues = [
        *(f"tracked local file: {item}" for item in git_status["trackedLocalFiles"]),
        *(f"untracked durable file: {item}" for item in git_status["untrackedDurableFiles"]),
        *(f"ignored durable file: {item}" for item in git_status["ignoredDurableFiles"]),
    ]
    problems = {
        "missingStructure": sorted(name for name, present in required_paths.items() if not present),
        "invalidFrontmatter": sorted(set(invalid_frontmatter)),
        "duplicateIndexEntries": duplicates,
        "emptyDescriptions": [slug for slug, description, _line in entries if not description],
        "orphanPages": sorted(page_slugs - set(indexed_slugs)),
        "missingPages": sorted(set(indexed_slugs) - page_slugs),
        "conflictMarkers": find_conflict_markers(paths["vault"]),
        "indexBuildErrors": [index_error] if index_error else [],
        "indexOutOfDate": ["index.md"] if index_inspection["needsRefresh"] else [],
        "symbolicLinks": inventory["symlinks"],
        "unreadableFiles": inventory["unreadableFiles"],
        "adapterConfiguration": adapter_conflicts,
        "gitReadiness": git_issues,
        "outdatedRuntime": (
            [
                f"project engine {runtime_health['engineVersion']} is older than "
                f"available engine {ENGINE_VERSION}"
            ]
            if runtime_health["compatible"] and not runtime_health["current"]
            else []
        ),
    }
    issue_count = sum(len(items) for items in problems.values())
    return {
        "totalPages": len(pages),
        "totalDailyLogs": len(daily),
        "deprecatedPages": len(deprecated),
        "deprecatedSlugs": sorted(deprecated),
        "indexTopics": len(entries),
        "generatedIndex": True,
        "protocolVersion": PROTOCOL_VERSION,
        "engineVersion": ENGINE_VERSION,
        "managedAdapterVersion": MANAGED_ADAPTER_VERSION,
        "codexAdapterVersion": MANAGED_ADAPTER_VERSION,
        "initialized": marker_status["valid"],
        "legacyVaultDetected": legacy_vault_detected(paths),
        **problems,
        "todayEntries": len(ENTRY_HEADER_RE.findall(today_log_content(paths))),
        "scaleAlert": len(pages) > 200,
        "integrity": "healthy" if issue_count == 0 and state_health["valid"] else "attention",
        "issueCount": issue_count + int(not state_health["valid"]),
        "stateFile": state_health,
        "marker": marker_status,
        "projectRuntime": runtime_health,
        "git": git_status,
        "storage": inventory,
    }


def format_status_card(status: dict[str, Any]) -> str:
    state = status["state"]
    health = status["health"]
    initialized = bool(health["initialized"])
    legacy_detected = bool(health["legacyVaultDetected"])
    session = state.get("session") if isinstance(state.get("session"), dict) else {}
    if legacy_detected and not initialized:
        badge = "△ Legacy vault detected"
    elif not initialized:
        badge = "○ Not initialized"
    elif health["integrity"] == "healthy":
        badge = "✓ Healthy"
    else:
        badge = f"△ Needs attention ({health['issueCount']})"
    session_label = "not started"
    if session.get("started"):
        session_label = "saved" if session.get("saved") else "active"
    storage = health["storage"]
    lines = [
        f"### EchoesVault · {badge}",
        "",
        "| Storage | Knowledge | Session |",
        "|---|---|---|",
        (
            f"| {format_bytes(int(storage['totalBytes']))} · {storage['totalFiles']} files "
            f"| {health['totalPages']} pages · {health['totalDailyLogs']} logs "
            f"| {session_label} |"
        ),
    ]
    if initialized and health["integrity"] == "healthy":
        lines.extend(
            [
                "",
                f"Protocol: {health['protocolVersion']} · engine: {health['engineVersion']} · adapters ready.",
                "",
                "Integrity: generated index, structure, metadata, and local paths are consistent.",
            ]
        )
    elif initialized:
        details = []
        labels = (
            ("missingStructure", "missing structure"),
            ("invalidFrontmatter", "invalid frontmatter"),
            ("duplicateIndexEntries", "duplicate index entries"),
            ("emptyDescriptions", "empty descriptions"),
            ("orphanPages", "orphan pages"),
            ("missingPages", "missing pages"),
            ("conflictMarkers", "Git conflict markers"),
            ("indexBuildErrors", "index build errors"),
            ("indexOutOfDate", "stale generated index"),
            ("symbolicLinks", "symbolic links"),
            ("unreadableFiles", "unreadable files"),
            ("adapterConfiguration", "adapter conflicts"),
            ("gitReadiness", "Git readiness issues"),
            ("outdatedRuntime", "outdated project runtime"),
        )
        for key, label in labels:
            if health[key]:
                details.append(f"{label}: {len(health[key])}")
        if not health["stateFile"]["valid"]:
            details.append("invalid runtime state")
        lines.extend(
            [
                "",
                f"Protocol: {health['protocolVersion']} · engine: {health['engineVersion']}.",
                "",
                "Integrity: " + "; ".join(details) + ".",
            ]
        )
    elif legacy_detected:
        lines.extend(
            [
                "",
                "Legacy vault detected. Choose **Initialize or restore EchoesVault** to migrate it.",
            ]
        )
    else:
        lines.extend(["", "Choose **Initialize or restore EchoesVault** to create local Markdown memory."])
    if health["scaleAlert"]:
        lines.extend(["", "> Scale alert: more than 200 pages; prefer targeted search."])
    lines.extend(
        [
            "",
            "Actions: **Initialize or restore** · **Show vault status** · **Save this session**",
        ]
    )
    return "\n".join(lines) + "\n"


def write_state(
    paths: dict[str, Path],
    state: dict[str, Any],
    args: Optional[argparse.Namespace] = None,
    writer_agent: Optional[str] = None,
) -> None:
    state["version"] = STATE_VERSION
    state["protocolVersion"] = PROTOCOL_VERSION
    state["engineVersion"] = ENGINE_VERSION
    state.pop("pluginVersion", None)
    state.pop("runtimeVersion", None)
    requested_agent = writer_agent or (getattr(args, "agent", None) if args else None)
    adapter_version = getattr(args, "adapter_version", None) if args else None
    if requested_agent:
        state["lastWriter"] = {
            "agent": normalize_agent_name(requested_agent),
            "adapterVersion": (
                adapter_version
                if isinstance(adapter_version, str) and adapter_version.strip()
                else None
            ),
        }
    elif not isinstance(state.get("lastWriter"), dict):
        state["lastWriter"] = {"agent": None, "adapterVersion": None}
    health = collect_health(paths)
    state["stats"] = {
        "totalPages": health["totalPages"],
        "totalDailyLogs": health["totalDailyLogs"],
        "deprecatedPages": health["deprecatedPages"],
    }
    atomic_write(paths["state"], json.dumps(state, ensure_ascii=False, indent=2) + "\n")


def normalize_filename(value: Any) -> str:
    if not isinstance(value, str) or not value.strip():
        raise EchoesError("Page filename must be a non-empty string.")
    name = stable_name(value.strip())
    if name.endswith(".md"):
        name = name[:-3]
    if not name or name in {".", ".."} or "/" in name or "\\" in name or ".." in name:
        raise EchoesError(f"Unsafe page filename: {value!r}")
    if name.startswith("."):
        raise EchoesError("Hidden page filenames are not allowed.")
    return f"{name}.md"


def sha256_text(content: str) -> str:
    return hashlib.sha256(content.encode("utf-8")).hexdigest()


def load_payload(path_value: str) -> dict[str, Any]:
    try:
        raw = (
            sys.stdin.read()
            if path_value == "-"
            else Path(path_value).expanduser().read_text(encoding="utf-8")
        )
        payload = json.loads(raw)
    except (OSError, json.JSONDecodeError) as exc:
        raise EchoesError(f"Cannot read JSON payload: {exc}") from exc
    if not isinstance(payload, dict):
        raise EchoesError("JSON payload must be an object.")
    return payload


def prepare_page(paths: dict[str, Path], item: dict[str, Any]) -> tuple[Path, str, bool]:
    filename = normalize_filename(item.get("filename"))
    page_path = paths["pages"] / filename
    if page_path.is_symlink():
        raise EchoesError(f"Refusing to read or replace a symbolic-link page: {filename}")
    normalized, metadata = parse_frontmatter(item.get("content"), require_summary=False)
    legacy_description = item.get("indexDescription")
    if "summary" not in metadata:
        summary = normalize_summary(legacy_description, page_path.stem)
        normalized = inject_summary(normalized, summary)
        _normalized, metadata = parse_frontmatter(normalized, require_summary=True)
    elif legacy_description is not None:
        legacy_summary = normalize_summary(legacy_description, page_path.stem)
        if legacy_summary != metadata["summary"]:
            raise EchoesError(
                f"indexDescription conflicts with frontmatter summary for {filename}."
            )
    content = validate_frontmatter(normalized)
    existing = page_path.exists()
    if existing:
        expected = item.get("expectedSha256")
        if not isinstance(expected, str) or not expected:
            raise EchoesError(
                f"Updating {filename} requires expectedSha256 from the content read before writing."
            )
        current = page_path.read_text(encoding="utf-8")
        actual = sha256_text(current)
        if expected != actual:
            raise EchoesError(
                f"Concurrent change detected for {filename}: expected {expected}, found {actual}."
            )
    return page_path, content, existing


def normalize_agent_name(value: Any) -> Optional[str]:
    if value is None:
        return None
    if not isinstance(value, str) or not value.strip():
        raise EchoesError("Payload agent must be a non-empty string when provided.")
    normalized = re.sub(r"[^a-z0-9._-]+", "-", value.strip().casefold()).strip("-.")
    if not normalized:
        raise EchoesError("Payload agent must contain at least one letter or number.")
    return normalized[:40]


def new_daily_log(
    paths: dict[str, Path], kind: str, content: str, agent: Optional[str] = None
) -> Path:
    moment = utc_now()
    directory = paths["daily"] / moment.date().isoformat()
    agent_part = f"-{agent}" if agent else ""
    name = (
        f"{moment.strftime('%Y%m%dT%H%M%S%f')}Z-{kind}{agent_part}-"
        f"{secrets.token_hex(4)}.md"
    )
    path = directory / name
    atomic_write(path, content)
    return path


def cmd_init(paths: dict[str, Path], args: argparse.Namespace) -> dict[str, Any]:
    existed = marker_health(paths)["valid"]
    ensure_structure(paths, create_marker=True)
    state = read_state(paths)
    state["initialized"] = True
    adapters = configure_agent_adapters(paths)
    index = refresh_index(paths, allow_legacy_migration=True)
    write_state(paths, state, args)
    return {
        "ok": True,
        "operation": args.command,
        "created": not existed,
        "vault": str(paths["vault"]),
        "index": str(paths["index"]),
        "indexRefresh": index,
        "agentAdapters": adapters,
        "state": state,
    }


def cmd_protocol(paths: dict[str, Path], _args: argparse.Namespace) -> dict[str, Any]:
    marker_status = marker_health(paths)
    marker = marker_status["value"] or {}
    return {
        "ok": True,
        "managedAdapterVersion": MANAGED_ADAPTER_VERSION,
        "codexAdapterVersion": MANAGED_ADAPTER_VERSION,
        "engineVersion": ENGINE_VERSION,
        "supportedProtocolVersion": PROTOCOL_VERSION,
        "vaultProtocolVersion": marker.get("protocolVersion"),
        "marker": marker_status,
        "projectRuntime": project_runtime_health(paths),
        "runtime": str(paths["runtime"]),
        "protocol": str(paths["protocol"]),
        "commands": [
            "init",
            "migrate",
            "upgrade",
            "protocol",
            "configure-agents",
            "inspect",
            "hydrate",
            "start",
            "status",
            "search",
            "append",
            "upsert",
            "end",
            "hash",
            "rebuild-index",
        ],
    }


def cmd_configure_agents(
    paths: dict[str, Path], args: argparse.Namespace
) -> dict[str, Any]:
    state = require_initialized(paths)
    ensure_structure(paths, create_marker=True)
    state["initialized"] = True
    adapters = configure_agent_adapters(paths)
    refresh_index(paths)
    write_state(paths, state, args)
    return {"ok": True, **adapters}


def cmd_hydrate(paths: dict[str, Path], args: argparse.Namespace) -> dict[str, Any]:
    state = require_initialized(paths)
    index = refresh_index(paths)
    write_state(paths, state, args)
    return {
        "ok": True,
        "index": str(paths["index"]),
        "indexRefresh": index,
        "state": state,
    }


def cmd_start(paths: dict[str, Path], args: argparse.Namespace) -> str:
    state = require_initialized(paths)
    refresh_index(paths)
    recent_count = max(0, min(args.recent, 10))
    selected = list(reversed(daily_markdown_files(paths["daily"])))[:recent_count]
    index_content = paths["index"].read_text(encoding="utf-8")
    sections = ["# EchoesVault session context", "", "## Index", "", index_content.rstrip()]
    sections.extend(["", f"## Recent session entries ({len(selected)})"])
    if selected:
        for daily_file in selected:
            relative = daily_file.relative_to(paths["daily"]).as_posix()
            sections.extend(
                ["", f"### {relative}", "", daily_file.read_text(encoding="utf-8").rstrip()]
            )
    else:
        sections.extend(["", "No session entries found."])
    health = collect_health(paths)
    if health["scaleAlert"]:
        sections.extend(
            [
                "",
                "> [!warning] SCALE ALERT",
                "> The vault exceeds 200 pages. Prefer targeted search over loading every page.",
            ]
        )
    state["session"]["started"] = True
    state["session"]["saved"] = False
    state["session"]["lastStart"] = timestamp()
    write_state(paths, state, args)
    return "\n".join(sections).rstrip() + "\n"


def cmd_inspect(paths: dict[str, Path], args: argparse.Namespace) -> Any:
    state = read_state(paths)
    index_refresh, index_error = inspect_index(paths)
    health = collect_health(paths, index_error=index_error)
    state["stats"] = {
        "totalPages": health["totalPages"],
        "totalDailyLogs": health["totalDailyLogs"],
        "deprecatedPages": health["deprecatedPages"],
    }
    status = {
        "ok": True,
        "workspace": str(paths["workspace"]),
        "vault": str(paths["vault"]),
        "state": state,
        "indexRefresh": index_refresh,
        "health": health,
    }
    return format_status_card(status) if args.format == "card" else status


def cmd_status(paths: dict[str, Path], args: argparse.Namespace) -> Any:
    return cmd_inspect(paths, args)


def cmd_search(paths: dict[str, Path], args: argparse.Namespace) -> dict[str, Any]:
    require_initialized(paths)
    query = args.query.strip()
    if not query:
        raise EchoesError("Search query cannot be empty.")
    folded = query.casefold()
    results = []
    limit = max(1, min(args.limit, 500))
    for page in markdown_files(paths["pages"]):
        for line_number, line in enumerate(page.read_text(encoding="utf-8").splitlines(), 1):
            if folded in line.casefold():
                results.append(
                    {
                        "file": str(page.relative_to(paths["workspace"])),
                        "line": line_number,
                        "text": line.strip()[:300],
                    }
                )
                if len(results) >= limit:
                    return {"ok": True, "query": query, "truncated": True, "results": results}
    return {"ok": True, "query": query, "truncated": False, "results": results}


def cmd_append(paths: dict[str, Path], args: argparse.Namespace) -> dict[str, Any]:
    state = require_initialized(paths)
    payload = load_payload(args.payload)
    entry = payload.get("entry")
    if not isinstance(entry, str) or not entry.strip():
        raise EchoesError("Append payload requires a non-empty string field named 'entry'.")
    agent = normalize_agent_name(payload.get("agent") or args.agent)
    agent_line = f"\nAgent: `{agent}`\n" if agent else ""
    block = f"### Scratchpad — {timestamp()}\n{agent_line}\n{entry.strip()}\n"
    daily_file = new_daily_log(paths, "scratchpad", block, agent)
    write_state(paths, state, args, writer_agent=agent)
    return {
        "ok": True,
        "dailyLog": str(daily_file),
        "kind": "scratchpad",
        "agent": agent,
    }


def cmd_upsert(paths: dict[str, Path], args: argparse.Namespace) -> dict[str, Any]:
    state = require_initialized(paths)
    payload = load_payload(args.payload)
    page_path, content, existed = prepare_page(paths, payload)
    generated = render_index_from_sources(page_source_map(paths, {page_path: content}))
    atomic_write(page_path, content)
    try:
        current_index = paths["index"].read_text(encoding="utf-8")
    except OSError:
        current_index = ""
    index_changed = current_index != generated
    if index_changed:
        atomic_write(paths["index"], generated)
    write_state(paths, state, args)
    return {
        "ok": True,
        "action": "updated" if existed else "created",
        "page": str(page_path),
        "sha256": sha256_text(content),
        "indexChanged": index_changed,
    }


def cmd_end(paths: dict[str, Path], args: argparse.Namespace) -> dict[str, Any]:
    if not args.confirm_explicit_user_end:
        raise EchoesError(
            "Final memory save requires --confirm-explicit-user-end from the echoes-end workflow."
        )
    state = require_initialized(paths)
    payload = load_payload(args.payload)
    summary = payload.get("dailySummary")
    if not isinstance(summary, str) or not summary.strip():
        raise EchoesError("End payload requires a non-empty dailySummary.")
    pages = payload.get("pages", [])
    if not isinstance(pages, list) or not all(isinstance(item, dict) for item in pages):
        raise EchoesError("End payload pages must be an array of objects.")
    if payload.get("indexUpdates") not in (None, []):
        raise EchoesError(
            "indexUpdates are no longer accepted because index.md is generated from page summaries."
        )
    prepared: list[tuple[Path, str]] = []
    overrides: dict[Path, str] = {}
    for item in pages:
        page_path, content, _existed = prepare_page(paths, item)
        if page_path in overrides:
            raise EchoesError(f"End payload contains duplicate page: {page_path.name}")
        overrides[page_path] = content
        prepared.append((page_path, content))
    generated = render_index_from_sources(page_source_map(paths, overrides))
    for page_path, content in prepared:
        atomic_write(page_path, content)
    atomic_write(paths["index"], generated)
    agent = normalize_agent_name(payload.get("agent") or args.agent)
    agent_line = f"\nAgent: `{agent}`\n" if agent else ""
    block = f"### Session — {timestamp()}\n{agent_line}\n{summary.strip()}\n"
    daily_file = new_daily_log(paths, "session", block, agent)
    state["session"]["saved"] = True
    state["session"]["lastSave"] = timestamp()
    write_state(paths, state, args, writer_agent=agent)
    return {
        "ok": True,
        "dailyLog": str(daily_file),
        "pagesWritten": len(prepared),
        "index": str(paths["index"]),
        "memorySaved": True,
        "agent": agent,
    }


def cmd_hash(paths: dict[str, Path], args: argparse.Namespace) -> dict[str, Any]:
    require_initialized(paths)
    filename = normalize_filename(args.filename)
    page = paths["pages"] / filename
    if not page.is_file():
        raise EchoesError(f"Page does not exist: {filename}")
    content = page.read_text(encoding="utf-8")
    return {"ok": True, "page": str(page), "sha256": sha256_text(content)}


def cmd_rebuild_index(paths: dict[str, Path], args: argparse.Namespace) -> dict[str, Any]:
    state = require_initialized(paths)
    index = refresh_index(paths)
    write_state(paths, state, args)
    return {"ok": True, "index": str(paths["index"]), **index}


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--workspace",
        help="Project directory. Defaults to the current Git root, then the current directory.",
    )
    parser.add_argument(
        "--agent",
        help="Agent writing state or knowledge, for example codex, opencode, or claude.",
    )
    parser.add_argument(
        "--adapter-version",
        help="Version of the invoking agent adapter; it is metadata, not the engine version.",
    )
    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("init", help="Create the vault idempotently.")
    subparsers.add_parser("migrate", help="Explicitly migrate a recognized legacy vault.")
    subparsers.add_parser("upgrade", help="Upgrade the project runtime and tracked adapters.")
    subparsers.add_parser("protocol", help="Report protocol compatibility and managed paths.")
    subparsers.add_parser(
        "configure-agents",
        help="Install or repair agent-neutral protocol documentation and adapters.",
    )
    start_parser = subparsers.add_parser("start", help="Restore index and recent session entries.")
    start_parser.add_argument("--recent", type=int, default=3)
    inspect_parser = subparsers.add_parser(
        "inspect", help="Inspect health without changing any file."
    )
    inspect_parser.add_argument("--format", choices=("json", "card"), default="json")
    status_parser = subparsers.add_parser(
        "status", help="Read-only alias for inspect."
    )
    status_parser.add_argument("--format", choices=("json", "card"), default="json")
    subparsers.add_parser(
        "hydrate", help="Refresh only ignored index.md and local state.json."
    )
    search_parser = subparsers.add_parser("search", help="Search page contents literally.")
    search_parser.add_argument("query")
    search_parser.add_argument("--limit", type=int, default=100)
    append_parser = subparsers.add_parser("append", help="Append a scratchpad entry from JSON.")
    append_parser.add_argument("--payload", default="-", help="JSON file path, or '-' for stdin.")
    upsert_parser = subparsers.add_parser("upsert", help="Create or safely update one page.")
    upsert_parser.add_argument("--payload", default="-", help="JSON file path, or '-' for stdin.")
    end_parser = subparsers.add_parser("end", help="Commit final session memory from JSON.")
    end_parser.add_argument("--payload", default="-", help="JSON file path, or '-' for stdin.")
    end_parser.add_argument("--confirm-explicit-user-end", action="store_true")
    hash_parser = subparsers.add_parser("hash", help="Hash a page for optimistic concurrency.")
    hash_parser.add_argument("filename")
    subparsers.add_parser(
        "rebuild-index", help="Force validation and deterministic index regeneration."
    )
    return parser


def emit(value: Any) -> None:
    if isinstance(value, str):
        sys.stdout.write(value)
        return
    sys.stdout.write(json.dumps(value, ensure_ascii=False, indent=2) + "\n")


READ_ONLY_COMMANDS = {"inspect", "status", "protocol", "search", "hash"}
INSPECTION_COMMANDS = {"inspect", "status", "protocol"}
FULL_INSTALL_COMMANDS = {"init", "migrate", "upgrade"}


def execute_project_runtime(
    target: Path, original_argv: list[str], args: argparse.Namespace
) -> int:
    input_value: Optional[str] = None
    if args.command in {"append", "upsert", "end"} and args.payload == "-":
        input_value = sys.stdin.read()
    try:
        result = subprocess.run(
            [sys.executable, str(target), *original_argv],
            input=input_value,
            capture_output=True,
            text=True,
            check=False,
        )
    except OSError as exc:
        raise EchoesError(f"Cannot execute project runtime {target}: {exc}") from exc
    sys.stdout.write(result.stdout)
    sys.stderr.write(result.stderr)
    return result.returncode


def delegate_to_project_runtime(
    paths: dict[str, Path], args: argparse.Namespace, original_argv: list[str]
) -> Optional[int]:
    source = Path(__file__).resolve()
    target = paths["runtime"]
    if source == target.resolve(strict=False):
        return None

    health = project_runtime_health(paths)
    source_engine = configured_version(ENGINE_VERSION)
    target_engine = engine_version_from_content(target.read_text(encoding="utf-8")) if health["recognized"] else None

    if health["recognized"] and health["compatible"] and target_engine is not None:
        if args.command in FULL_INSTALL_COMMANDS:
            with vault_lock(paths):
                ensure_portable_runtime(paths)
            return execute_project_runtime(target, original_argv, args)
        if target_engine >= source_engine:
            return execute_project_runtime(target, original_argv, args)
        if args.command in INSPECTION_COMMANDS:
            return None
        raise EchoesError(
            f"Project runtime {format_version(target_engine)} is older than engine "
            f"{ENGINE_VERSION}. Run the echoes-init workflow or `upgrade` explicitly."
        )
    elif health["exists"]:
        if args.command in INSPECTION_COMMANDS:
            return None
        if args.command not in FULL_INSTALL_COMMANDS:
            raise EchoesError(
                f"Project runtime is incompatible or unrecognized: {health['error'] or health['protocolVersion']}. "
                "Run the echoes-init workflow or `upgrade` explicitly."
            )
    elif args.command in INSPECTION_COMMANDS:
        return None
    elif args.command == "hydrate":
        raise EchoesError(
            "Project runtime is missing. Run the echoes-init workflow before hydrate."
        )
    elif args.command not in FULL_INSTALL_COMMANDS and not marker_health(paths)["valid"]:
        if legacy_vault_detected(paths):
            raise EchoesError(
                "Legacy EchoesVault detected. Run the echoes-init workflow to migrate it."
            )
        raise EchoesError(
            "EchoesVault is not initialized. Run the echoes-init workflow first."
        )

    with vault_lock(paths):
        ensure_portable_runtime(paths)
    installed = project_runtime_health(paths)
    if not installed["recognized"] or not installed["compatible"]:
        raise EchoesError(
            f"Could not install a compatible project runtime: {installed['error']}"
        )
    return execute_project_runtime(target, original_argv, args)


def main(argv: Optional[Iterable[str]] = None) -> int:
    parser = build_parser()
    original_argv = list(argv) if argv is not None else sys.argv[1:]
    args = parser.parse_args(original_argv)
    try:
        workspace = resolve_workspace(args.workspace)
        paths = vault_paths(workspace)
        delegated = delegate_to_project_runtime(paths, args, original_argv)
        if delegated is not None:
            return delegated
        commands = {
            "init": cmd_init,
            "migrate": cmd_init,
            "upgrade": cmd_init,
            "protocol": cmd_protocol,
            "configure-agents": cmd_configure_agents,
            "inspect": cmd_inspect,
            "hydrate": cmd_hydrate,
            "start": cmd_start,
            "status": cmd_status,
            "search": cmd_search,
            "append": cmd_append,
            "upsert": cmd_upsert,
            "end": cmd_end,
            "hash": cmd_hash,
            "rebuild-index": cmd_rebuild_index,
        }
        if args.command in READ_ONLY_COMMANDS:
            emit(commands[args.command](paths, args))
        else:
            with vault_lock(paths):
                emit(commands[args.command](paths, args))
        return 0
    except EchoesError as exc:
        sys.stderr.write(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False) + "\n")
        return 2


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