2.2.0 — echo-mcp: the containerized MCP server

ECHO as 14 typed MCP tools (streamable HTTP, stateless JSON, bearer auth, open /health) wrapping the 2.1.1 *_op cores in-process: - mcp-server/app.py: FastMCP app; duplicate gate + offline queueing surface as DATA (merge_into/force are parameters, never blind retries); recall packs excerpts into budget_chars by score; get_note is traversal-guarded with section/max_chars; patch_note enriches invalid-target errors with the note's actual headings; log_session wraps the session-end bundle (heartbeat-last); ECHO_MCP_TOOLS=core exposes only the six daily drivers; all MCP writes serialize in-process; startup fails fast on missing env. - Dockerfile (legacy format — no BuildKit on the CI runner), python:3.12-slim, healthcheck probes 127.0.0.1; .dockerignore keeps the context lean. - .gitea/workflows/docker-build.yml: standard image build; PORT redeploy trigger targets the echo-mcp container explicitly (repo name is echo). - deploy.unraid.yml: br0/auto-IP, /data volume, vault adjacency (ECHO_BASE=http://10.2.0.35:27123), secrets as SECRET: refs. - SKILL.md: prefer the echo_* tools when the connector is present; CLI recipes stay as the fallback. Spec header marked BUILT (tool count corrected to 14). - eval/test_mcp_server.py: e2e over real streamable HTTP (health/auth/ initialize/tools-list + capture->gate->merge->recall->log_session); skips cleanly when the SDK is absent. All seven suites green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2.1.1 — MCP Phase 0: every high-level op returns an envelope (return-not-print)
2026-07-28 23:01:47 -05:00 · 2026-07-28 22:51:33 -05:00
21 changed files with 1497 additions and 349 deletions
@@ -0,0 +1,11 @@
+# Only mcp-server/ + the plugin scripts reach the image; keep the context lean.
+.git
+*.plugin
+dist/
+CODEX/
+docs/
+eval/
+echo-icon*
+*.pdf
+*.html
+__pycache__/
@@ -0,0 +1,50 @@
+name: Build and Push Docker Image
+
+on:
+  push:
+    branches: [main]
+  workflow_dispatch:
+
+jobs:
+  build:
+    # Runs on the forgerunner host: bundled Docker CLI + mounted /var/run/docker.sock.
+    runs-on: host
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Log in to Gitea Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: registry.alwisp.com
+          username: ${{ secrets.REGISTRY_USER }}
+          password: ${{ secrets.REGISTRY_TOKEN }}
+
+      - name: Build and Push
+        run: |
+          IMAGE="registry.alwisp.com/${{ gitea.repository }}"
+          GIT_SHA="$(git rev-parse --short HEAD)"
+          COMMIT_COUNT="$(git rev-list --count HEAD)"
+          docker build \
+            --label org.alwisp.git-sha="${{ gitea.sha }}" \
+            --label org.alwisp.version="v2.${COMMIT_COUNT}" \
+            --label org.alwisp.repo="${{ gitea.repository }}" \
+            -t "${IMAGE}:latest" .
+          docker push "${IMAGE}:latest"
+
+      # Dangling-only prune: removes untagged leftovers from previous builds. Never
+      # removes the tagged :latest or any image referenced by a running container.
+      - name: Prune dangling images on host
+        if: always()
+        run: docker image prune -f 2>/dev/null || true
+
+      - name: Trigger PORT redeploy
+        if: success()
+        run: |
+          # Repo is 'echo' but the container is 'echo-mcp' — target it explicitly.
+          curl -fsS -X POST https://port.alwisp.com/hooks/gitea \
+            -H "X-Deploy-Token: ${{ secrets.WEBHOOK_SECRET }}" \
+            -H "Content-Type: application/json" \
+            -d '{"container":"echo-mcp"}' || echo "PORT redeploy trigger failed (non-fatal)"
@@ -1,5 +1,67 @@
 # Changelog

+## 2.2.0
+
+### Added — echo-mcp: the containerized MCP server
+
+ECHO's operations are now reachable as **typed MCP tools** from any surface with
+an MCP connector (Claude Code, CoWork, claude.ai) — no Python-capable shell, no
+`$ECHO` path resolution, no output re-parsing. New `mcp-server/` + `Dockerfile`
+in this repo; deployed on ALPHA as the `echo-mcp` container behind
+`https://echomcp.alwisp.com` (streamable HTTP, stateless JSON, bearer auth,
+open `/health` for Docker/Kuma).
+
+- **14 tools** wrapping the 2.1.1 `*_op` cores: `echo_load`, `echo_recall`
+  (score-packed `budget_chars`), `echo_resolve`, `echo_get_note`
+  (`section`/`max_chars`, traversal-guarded), `echo_get_scope`/`echo_set_scope`,
+  `echo_health` (`deep` runs the linter), `echo_capture` (gate as data —
+  `merge_into`/`force` are parameters), `echo_link` (paths OR resolvable names),
+  `echo_append_note`/`echo_patch_note` (invalid-target errors include the note's
+  actual headings), `echo_triage_inbox` (list/preview/apply in one tool),
+  `echo_reflect`, `echo_log_session` (the session-end bundle, heartbeat-last).
+- **Tool profiles**: `ECHO_MCP_TOOLS=core` exposes only the six daily drivers.
+- **Vault adjacency**: the container talks to the Obsidian REST API's HTTP
+  binding on the same box (`ECHO_BASE=http://10.2.0.35:27123`) — vault ops stop
+  depending on Cloudflare/NPM/DNS; the public chain only fronts the `echomcp`
+  ingress. Server-side offline queue at `/data`.
+- **All MCP writes serialize** through the server process; the vault advisory
+  lock still coordinates with CLI clients.
+- SKILL.md: prefer the `echo_*` tools when present; CLI recipes are the fallback.
+- New suite `eval/test_mcp_server.py` (skips without the SDK): health/auth/
+  initialize/tools-list + the capture→gate→merge→recall→log_session flow over
+  real streamable HTTP.
+
+The plugin itself is unchanged apart from the SKILL.md note — the container
+vendors `skills/echo-memory/scripts/` at build time (one canonical tree).
+
+## 2.1.1
+
+### Changed — Phase 0 of the MCP build: every high-level op returns an envelope
+
+Internal refactor (MCP-SERVER-SPEC §4), no behavior change intended: each
+high-level operation now has a core `*_op` function that **returns an envelope
+dict and never prints to stdout** — the transport-agnostic seam the upcoming MCP
+server (2.2) wraps. The CLI verbs are thin wrappers that print exactly what they
+printed before and map envelope outcomes to the same exit codes.
+
+- `echo_ops`: `capture_op` / `resolve_op` / `link_op` — the duplicate gate and
+  offline queueing are now **data** (`action: duplicate-gate` with candidates;
+  `queued: true`), mapped to exit 76 / "queued" text only in the wrapper. The
+  `--json` stdout-swallowing hack is gone: helper chatter from the low-level
+  verbs routes to **stderr** wholesale inside the cores. `capture_op` accepts
+  `body_text=` directly (no temp file needed — the MCP path).
+- `echo_recall.recall_op` — one structured result backs both the prose and
+  `--json` renderings (the duplicated prose path was deleted).
+- `echo_triage.list_op` / `route_op`, `echo_reflect.apply_op`,
+  `echo_session.session_end_op` — same pattern; reflect/triage/session now call
+  `capture_op` internally and count gate outcomes from envelopes.
+- `echo.scope_show_op` / `scope_set_op` / `load_op` (sections + brief digest as
+  data), `echo_doctor.run_op` (checks as a list).
+
+New suite `eval/test_ops_api.py` asserts the contract for every core: envelope
+shape AND stdout purity (a stray print would corrupt an MCP response stream).
+All existing suites pass unchanged.
+
 ## 2.1.0

 The "quick-wins train" from the 2026-07-28 review (`docs/IMPROVEMENT-PLANS.md` #1,
@@ -0,0 +1,26 @@
+# echo-mcp — containerized MCP server over the ECHO vault (docs/MCP-SERVER-SPEC.md).
+# LEGACY Dockerfile format on purpose: the git.alwisp.com CI runner has no BuildKit
+# (no `# syntax=` line, no RUN --mount).
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY mcp-server/requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+
+# The canonical plugin scripts ARE the server's ops layer — vendored at build time,
+# never a second source tree.
+COPY echo-memory.plugin.src/skills/echo-memory/scripts /app/scripts
+COPY mcp-server/app.py /app/app.py
+
+ENV ECHO_STATE_DIR=/data \
+    ECHO_MCP_PORT=8765 \
+    PYTHONUNBUFFERED=1
+VOLUME /data
+EXPOSE 8765
+
+# Probe 127.0.0.1, NOT localhost (::1-vs-IPv4 lesson from cpas/memer/breedr).
+HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
+  CMD python3 -c "import urllib.request;urllib.request.urlopen('http://127.0.0.1:8765/health', timeout=4)" || exit 1
+
+CMD ["python3", "/app/app.py"]
@@ -1,4 +1,4 @@
-# echo-memory — v2.1.0
+# echo-memory — v2.2.0

 Persistent memory for Claude / CoWork sessions via the **ECHO** Obsidian vault, driven over the [Obsidian Local REST API](https://github.com/coddingtonbear/obsidian-local-rest-api). The skill makes direct REST calls through a bundled validated client (`scripts/echo.py`). The whole toolchain is **pure Python** (stdlib only), so it runs identically on Windows, macOS, and Linux — no bash, no platform-specific `date`.

@@ -421,6 +421,8 @@ From the credential-free harness (`eval/run_eval.py` against the deterministic m

 | Version | Highlights |
 |---------|-----------|
+| **2.2.0** | **echo-mcp — the containerized MCP server.** ECHO as 14 typed MCP tools from any connector-capable surface (Claude Code, CoWork, claude.ai): `mcp-server/` + Dockerfile in-repo, deployed on ALPHA as `echo-mcp` behind `https://echomcp.alwisp.com` (streamable HTTP, stateless JSON, bearer auth, open `/health`). Talks to the Obsidian REST API directly on the LAN (`10.2.0.35:27123`) — one client round trip per tool call, all vault chatter host-local. Duplicate gate & offline queueing surface as data; `ECHO_MCP_TOOLS=core` trims the surface to six tools; writes serialize server-side. New `eval/test_mcp_server.py` e2e suite. Full spec: `docs/MCP-SERVER-SPEC.md`. |
+| **2.1.1** | **MCP Phase 0 — return-not-print.** Every high-level op gains a core `*_op` function returning an envelope dict with a stdout-purity guarantee (helper chatter → stderr); CLI verbs become thin wrappers with identical output and exit codes. Duplicate gate and offline queueing become data (`action: duplicate-gate` / `queued: true`). New `eval/test_ops_api.py` contract suite. This is the seam the 2.2 containerized MCP server wraps. |
 | **2.1.0** | **Quick-wins train: cheaper sessions, durable capture, one-call session end.** (1) **`load --brief`** — a token-budgeted cold-start digest (Fact/Pattern in full, last ~10 Observations, scope+freshness, last session's key sections, Agent-Log lines, inbox *count*; `ECHO_LOAD_BUDGET` default ~8000 chars) now injected by the SessionStart hook, replacing the full six-file dump that grew unboundedly; all `load` reads are fetched in parallel. (2) **Offline capture durability** — `capture` on an unreachable vault queues the *whole operation* as one semantic record; `flush` replays it **through capture** so routing/gate/aliasing re-run against the current index; a gate stop on replay is kept + flagged, never landed blind; `ensure_daily_log`/update-path writes ride the queue too. (3) **`session-end`** — one call (one lock) writes session log → Agent-Log line → reflect proposals → optional scope switch → **heartbeat last as the commit marker**; dry-run by default; `ECHO_NOW` pins the HHMM. Also fixes the `__main__` twin-module trap so helper-module `EchoError`s exit with their intended codes. +19 end-to-end checks. |
 | **2.0.0** | **Packaging & structure major — memory behavior unchanged.** Skills-only: the legacy `commands/` directory is deleted (breaking for pre-skills clients; the 1.6.0-verified skills are the sole entry points). Artifact policy: the repo tracks only the `echo-memory.plugin` pointer; the 15 historical versioned zips leave the tree and versioned builds ship as **Gitea releases** (one per `v<version>` tag) from `v2.0.0` on; `.gitignore` blocks `*.plugin` except the pointer. README gains the "packaging for other agent runtimes" port note (build-target rule, never a second source tree). |
 | **1.6.0** | **Skills-format migration + lint blind-spot fix.** The eight slash commands gain `skills/<name>/SKILL.md` twins (same names, byte-identical bodies; a same-name skill takes precedence, so `commands/` coexists until its 2.0 deletion): per-skill `allowed-tools` end the permission prompts, `disable-model-invocation: true` makes `/echo-sweep` + `/echo-triage` operator-only. `vault_lint` checks retired patterns **before** routes, so a seed README inside a retired tree (`archive/…`) is flagged instead of being absolved by the leaf-README route (+ regression test). `plugin.json` gains `homepage`/`repository`. |
@@ -0,0 +1,28 @@
+# echo-mcp — PORT deploy manifest (ALPHA / br0). See docs/MCP-SERVER-SPEC.md §8.
+# Container name deliberately differs from the repo (echo): the repo ships the
+# plugin AND this server; only the server is a container.
+name: echo-mcp
+image: registry.alwisp.com/jason/echo:latest
+network: br0
+ip: auto
+ports: []            # br0 static IP — app serves :8765 directly
+volumes:
+  - /mnt/user/appdata/echo-mcp:/data     # outbox + read cache (+ backups/history, v1.1)
+env:
+  # Vault adjacency: the Obsidian Local REST API's HTTP binding on the same box —
+  # never the public echoapi.alwisp.com hairpin (spec §8; cleartext stays on the LAN).
+  ECHO_BASE: http://10.2.0.35:27123
+  ECHO_OWNER: Jason Stedwell
+  ECHO_STATE_DIR: /data
+  ECHO_MCP_PORT: "8765"
+  ECHO_MCP_TOOLS: full
+  # Secrets from the PORT secret store — values never appear in this file or chat.
+  ECHO_KEY: SECRET:echo-vault-key
+  ECHO_MCP_TOKEN: SECRET:echo-mcp-token
+health_check_path: /health
+health_check_port: 8765
+webui: https://echomcp.alwisp.com
+proxy:
+  forward_port: 8765
+  forward_scheme: http
+  websockets: true
@@ -1,9 +1,9 @@
 # echo-mcp — MCP Server Build Spec (containerized)

-> Status: **spec for a future build session** (written 2026-07-28 against v1.5.1;
-> revised same day: **remote container architecture**, operator decision — heavy
-> lifting belongs in a deployed container, not on any one machine).
-> Companion plan for the other nine review items: `docs/IMPROVEMENT-PLANS.md`.
+> Status: **BUILT — shipped as 2.2.0, 2026-07-28** (`mcp-server/app.py`, 14 tools —
+> the count below saying 13 undercounted the append/patch pair; e2e suite
+> `eval/test_mcp_server.py`). This document remains the design record; §7.2 is the
+> live v1.1 backlog. Companion plan: `docs/IMPROVEMENT-PLANS.md`.
 >
 > **Prerequisites before starting this build:**
 > 1. The `session-end` verb (IMPROVEMENT-PLANS #7) — the MCP tool wraps it.
@@ -75,7 +75,7 @@ monitor, PORT deploy/rollback. Server down ⇒ exactly today's behavior.
 | Tool prefix | `echo_` | Namespace safety next to other servers. |
 | Result shape | `structuredContent` (the `echo_output.envelope` dict) + a 1–3 line human text block | Envelope shape already exists. |
 | Statefulness | Stateless protocol; warm in-process caches | See §7. |
-| Version target | **2.2** (1.6.0, 2.0.0, and the 2.1.0 quick-wins train all shipped 2026-07-28; `session-end` exists) | Remaining prerequisite: Phase 0 (return-not-print), the build session's first step. |
+| Version target | **2.2** — ALL prerequisites shipped 2026-07-28 (`session-end` in 2.1.0; **Phase 0 in 2.1.1**, contract-tested by `eval/test_ops_api.py`) | The build session starts directly at §5 (the server app). |
 | Local stdio variant | **Dropped for v1** (documented, not built) | The CLI/skill fallback covers the no-server case; two transports = two test matrices for little gain. Phase 0 keeps the door open — the `*_op` core is transport-agnostic. |
 | Result budgets | Every read tool takes an explicit size budget (`budget_chars` / `max_chars`) and truncates with a marker + "call X for more" | The single biggest token lever: one right-sized answer instead of follow-up fetches. |
 | Tool profiles | `ECHO_MCP_TOOLS=core\|full` env: `core` exposes only load/recall/capture/triage_inbox/log_session/health | Tool schemas cost context on every surface that lists them; the claude.ai connector doesn't need the escape hatches. Desktop registers `full`. |
@@ -1,6 +1,6 @@
 {
  "name": "echo-memory",
-  "version": "2.1.0",
+  "version": "2.2.0",
  "homepage": "https://git.alwisp.com/jason/echo",
  "repository": "https://git.alwisp.com/jason/echo",
  "description": "Persistent memory via the ECHO Obsidian vault over the Obsidian Local REST API. Cross-platform Python client: one-call capture/resolve/recall/link/triage over an entity index, hybrid BM25 + graph recall spanning entities + sessions/journal (recency/status-aware), a pre-write duplicate gate, complete-frontmatter capture, session hooks that self-fire load/reflect, offline write-ahead queue, lock-guarded concurrency, linter-enforced routing, and /echo-* commands.",
@@ -52,6 +52,19 @@ Executable logic ships under `scripts/` — **pure Python**, so the whole toolch
 - `scripts/check_routing.py` — verifies the routing docs stay in sync with `routing.json` (dev/CI; offline)
 - `scripts/bootstrap.py` / `scripts/migrate.py` — deterministic vault setup/repair and schema migration

+## MCP tools first (when the echo-mcp connector is available)
+
+When this session has the **`echo_*` MCP tools** (the deployed echo-mcp server —
+`echo_load`, `echo_recall`, `echo_resolve`, `echo_get_note`, `echo_get_scope`/`echo_set_scope`,
+`echo_health`, `echo_capture`, `echo_link`, `echo_append_note`/`echo_patch_note`,
+`echo_triage_inbox`, `echo_reflect`, `echo_log_session`), **prefer them over the CLI
+for every operation they cover** — typed calls, structured results, no path
+resolution or quoting. The procedures in this skill (reconcile at load, search-first,
+scope discipline, third person, preview-before-apply) are unchanged and apply to both
+surfaces. A `duplicate-gate` tool result is the same contract as capture exit 76:
+`merge_into` or confirmed `force`, never a blind retry. The CLI recipes below are the
+fallback for hosts without the connector or when the server is unreachable.
+
 ## Bundled Tooling (prefer over raw curl)

 All paths below are under `${CLAUDE_PLUGIN_ROOT}/skills/echo-memory/`. **Invoke with `python3`** (on Windows where that isn't on PATH, use `python` or `py -3`).
@@ -549,59 +549,76 @@ def extract_heading(markdown: str, heading: str) -> str:
    return "\n".join(out).strip()


-def cmd_scope(subcommand: str, text: str | None = None, as_json: bool = False) -> int:
+def scope_show_op() -> dict:
+    """Core scope-show (Phase 0): active scope + freshness + sessions-since, as data."""
    path = "_agent/context/current-context.md"
    status, body = request("GET", vault_url(path))
-    check(status, body, f"scope {subcommand}")
+    check(status, body, "scope show")
    current = body.decode(errors="replace")
+    scope_text = extract_heading(current, "Scope")
+    scope_updated = ""
+    for line in current.splitlines():
+        if line.startswith("scope_updated:"):
+            scope_updated = line.split(":", 1)[1].strip().strip('"').strip("'")
+            break
+    sessions_since = None
+    # Count session logs dated after scope_updated (the drift signal).
+    if scope_updated:
+        status, body = request("GET", vault_url("_agent/sessions/"))
+        if status == 200:
+            try:
+                files = json.loads(body).get("files", [])
+                sessions_since = sum(1 for f in files if f.endswith(".md") and f[:10] > scope_updated)
+            except json.JSONDecodeError:
+                pass
+    return {"ok": True, "action": "scope-show", "scope": scope_text,
+            "scope_updated": scope_updated or None, "sessions_since": sessions_since}

-    if subcommand == "show":
-        scope_text = extract_heading(current, "Scope")
-        scope_updated = ""
-        for line in current.splitlines():
-            if line.startswith("scope_updated:"):
-                scope_updated = line.split(":", 1)[1].strip().strip('"').strip("'")
-                break
-        sessions_since = None
-        # Count session logs dated after scope_updated (the drift signal).
-        if scope_updated:
-            status, body = request("GET", vault_url("_agent/sessions/"))
-            if status == 200:
-                try:
-                    files = json.loads(body).get("files", [])
-                    sessions_since = sum(1 for f in files if f.endswith(".md") and f[:10] > scope_updated)
-                except json.JSONDecodeError:
-                    pass
-        if as_json:
-            print(json.dumps({"ok": True, "action": "scope-show", "scope": scope_text,
-                              "scope_updated": scope_updated or None,
-                              "sessions_since": sessions_since}, ensure_ascii=False))
-            return 0
-        print("-- Active scope --")
-        print(scope_text)
-        print(f"scope_updated: {scope_updated or '<missing — drift cannot be detected; run scope set or repair>'}")
-        if sessions_since is not None:
-            print(f"sessions logged since: {sessions_since}")
-        return 0

-    if subcommand != "set":
-        raise EchoError("scope: use 'show' or 'set \"<text>\"'", 2)
+def scope_set_op(text: str) -> dict:
+    """Core scope-set (Phase 0): atomic switch (history + replace + stamp). Chatter
+    from the underlying PATCHes routes to stderr; returns the switch envelope."""
+    import contextlib
    if not text:
        raise EchoError("scope set needs the new scope text", 2)
+    path = "_agent/context/current-context.md"
+    status, body = request("GET", vault_url(path))
+    check(status, body, "scope set")
+    current = body.decode(errors="replace")
    prior = extract_heading(current, "Scope").replace("\n", " ").strip()[:140] or "(prior scope)"
-    cmd_patch(path, "prepend", "heading", "Current Context::Scope History",
-              temp_file(f"- {today()}: {prior}\n".encode()))
-    cmd_patch(path, "replace", "heading", "Current Context::Scope",
-              temp_file(f"{text}\n".encode()))
-    try:
-        cmd_patch(path, "replace", "frontmatter", "scope_updated",
-                  temp_file(json.dumps(today()).encode()))
-    except EchoError as exc:
-        raise EchoError(
-            "scope set: body switched, but scope_updated frontmatter is missing "
-            f"(run bootstrap.py to add it) [{exc}]"
-        )
-    print(f"ok: scope switched (prior archived to Scope History; scope_updated={today()})")
+    with contextlib.redirect_stdout(sys.stderr):
+        cmd_patch(path, "prepend", "heading", "Current Context::Scope History",
+                  temp_file(f"- {today()}: {prior}\n".encode()))
+        cmd_patch(path, "replace", "heading", "Current Context::Scope",
+                  temp_file(f"{text}\n".encode()))
+        try:
+            cmd_patch(path, "replace", "frontmatter", "scope_updated",
+                      temp_file(json.dumps(today()).encode()))
+        except EchoError as exc:
+            raise EchoError(
+                "scope set: body switched, but scope_updated frontmatter is missing "
+                f"(run bootstrap.py to add it) [{exc}]"
+            )
+    return {"ok": True, "action": "scope-set", "scope": text, "prior": prior,
+            "scope_updated": today()}
+
+
+def cmd_scope(subcommand: str, text: str | None = None, as_json: bool = False) -> int:
+    if subcommand == "show":
+        env = scope_show_op()
+        if as_json:
+            print(json.dumps(env, ensure_ascii=False))
+            return 0
+        print("-- Active scope --")
+        print(env["scope"])
+        print(f"scope_updated: {env['scope_updated'] or '<missing — drift cannot be detected; run scope set or repair>'}")
+        if env["sessions_since"] is not None:
+            print(f"sessions logged since: {env['sessions_since']}")
+        return 0
+    if subcommand != "set":
+        raise EchoError("scope: use 'show' or 'set \"<text>\"'", 2)
+    env = scope_set_op(text or "")
+    print(f"ok: scope switched (prior archived to Scope History; scope_updated={env['scope_updated']})")
    return 0


@@ -730,51 +747,22 @@ def _render_brief(texts: dict[str, str | None], listing_files: list[str],
    return out


-def cmd_load(brief: bool = False) -> int:
-    """Cold-start orientation: the canonical 6 reads in one call (fetched in parallel).
-    404s on today's daily note and the inbox are normal (printed as absent, not errors).
-    `brief` renders the token-budgeted digest the SessionStart hook injects; the default
-    full mode (and /echo-load) prints the raw sections unchanged."""
-    if not echo_config.is_configured(_CFG):
-        print("echo: NOT CONFIGURED — this machine has no usable ECHO key file yet.")
-        print(f"Expected at: {echo_config.config_path()}")
-        print("ACTION: ask the operator for their echo-memory config file (it holds the")
-        print("        vault owner, endpoint, and API key), then install it with ONE of:")
-        print("  python3 echo.py config import <path-to-their-file>")
-        print("  python3 echo.py config set --owner \"…\" --endpoint \"https://…\" --key \"…\"")
-        print("Then re-run load. (Memory is unavailable until configured.)")
-        return 78  # distinct: configuration required
-    targets = [
-        ("marker",      "_agent/echo-vault.md"),
-        ("preferences", "_agent/memory/semantic/operator-preferences.md"),
-        ("context",     "_agent/context/current-context.md"),
-        ("heartbeat",   "_agent/heartbeat/last-session.md"),
-        ("today",       f"journal/daily/{today()}.md"),
-        ("inbox",       "inbox/captures/inbox.md"),
-    ]
-    import echo_queue
-    # H2: sync any writes queued during a prior outage, best-effort and quiet on empty.
-    try:
-        synced = echo_queue.flush()
-        if synced:
-            print(f"(synced {synced} write(s) queued during a prior offline session)\n")
-        flagged = echo_queue.needs_attention()
-        if flagged:
-            print(f"NOTE: {len(flagged)} queued write(s) need attention (e.g. a capture "
-                  "stopped at the duplicate gate on replay) — resolve with capture "
-                  "--merge-into/--force; `echo.py flush` re-lists them.\n")
-    except Exception as exc:  # noqa: BLE001 — never let queue upkeep block a load
-        print(f"(queue flush skipped: {exc})\n", file=sys.stderr)
+LOAD_TARGETS = [
+    ("marker",      "_agent/echo-vault.md"),
+    ("preferences", "_agent/memory/semantic/operator-preferences.md"),
+    ("context",     "_agent/context/current-context.md"),
+    ("heartbeat",   "_agent/heartbeat/last-session.md"),
+]

-    # All orientation reads (+ the sessions listing) fetched in parallel over the
-    # warm connection pool — the listing feeds both the brief digest's scope-freshness
-    # count and the heartbeat-absent fallback, so it's no longer a second round-trip.
+
+def _load_gather(targets):
+    """Fetch the orientation reads (+ the sessions listing) in parallel and resolve
+    each to text via status/cache. Shared by cmd_load and load_op. Returns
+    (results, texts, listing_files, offline, marker_missing, heartbeat_absent)."""
+    import echo_queue
    listing_path = "_agent/sessions/"
    results = _fetch_statuses([p for _, p in targets] + [listing_path])
-
-    marker_missing = False
-    heartbeat_absent = False
-    offline = False
+    marker_missing = heartbeat_absent = offline = False
    texts: dict[str, str | None] = {}
    for label, path in targets:
        status, body = results[path]
@@ -792,7 +780,6 @@ def cmd_load(brief: bool = False) -> int:
                    marker_missing = True
                if label == "heartbeat":
                    heartbeat_absent = True
-
    lst_status, lst_body = results[listing_path]
    listing_files: list[str] = []
    if lst_status == 200:
@@ -800,15 +787,84 @@ def cmd_load(brief: bool = False) -> int:
            listing_files = list(json.loads(lst_body).get("files", []))
        except json.JSONDecodeError:
            pass
+    return results, texts, listing_files, offline, marker_missing, heartbeat_absent
+
+
+def _fetch_pointed_session(texts: dict) -> None:
+    """Follow the heartbeat pointer and stash the pointed session log for the digest."""
+    hb = texts.get("heartbeat")
+    if hb and hb.strip():
+        sess_path = hb.strip().splitlines()[0].split(" @ ", 1)[0].strip()
+        if sess_path:
+            st2, b2 = request("GET", vault_url(sess_path))
+            if st2 == 200:
+                texts["_pointed_session"] = b2.decode(errors="replace")
+
+
+def load_op(brief: bool = True) -> dict:
+    """Core load (Phase 0): the orientation reads as data — sections keyed by label,
+    plus offline/bootstrap flags, queue counts, recent sessions, and (with brief) the
+    rendered digest. Raises EchoError(78) when the machine is not configured."""
+    if not echo_config.is_configured(_CFG):
+        raise EchoError("NOT CONFIGURED — no usable ECHO key file on this machine "
+                        f"(expected at {echo_config.config_path()})", 78)
+    import echo_queue
+    synced = flagged = 0
+    try:
+        synced = echo_queue.flush()
+        flagged = len(echo_queue.needs_attention())
+    except Exception:  # noqa: BLE001 — queue upkeep must never block a load
+        pass
+    targets = LOAD_TARGETS + [("today", f"journal/daily/{today()}.md"),
+                              ("inbox", "inbox/captures/inbox.md")]
+    _, texts, listing_files, offline, marker_missing, _ = _load_gather(targets)
+    data = {"ok": True, "action": "load", "offline": offline,
+            "marker_missing": marker_missing, "synced": synced,
+            "needs_attention": flagged,
+            "sections": {k: v for k, v in texts.items() if not k.startswith("_")},
+            "recent_sessions": sorted((f for f in listing_files if f.endswith(".md")),
+                                      reverse=True)[:5]}
+    if brief:
+        _fetch_pointed_session(texts)
+        data["brief"] = _render_brief(texts, listing_files, offline)
+    return data
+
+
+def cmd_load(brief: bool = False) -> int:
+    """Cold-start orientation: the canonical 6 reads in one call (fetched in parallel).
+    404s on today's daily note and the inbox are normal (printed as absent, not errors).
+    `brief` renders the token-budgeted digest the SessionStart hook injects; the default
+    full mode (and /echo-load) prints the raw sections unchanged."""
+    if not echo_config.is_configured(_CFG):
+        print("echo: NOT CONFIGURED — this machine has no usable ECHO key file yet.")
+        print(f"Expected at: {echo_config.config_path()}")
+        print("ACTION: ask the operator for their echo-memory config file (it holds the")
+        print("        vault owner, endpoint, and API key), then install it with ONE of:")
+        print("  python3 echo.py config import <path-to-their-file>")
+        print("  python3 echo.py config set --owner \"…\" --endpoint \"https://…\" --key \"…\"")
+        print("Then re-run load. (Memory is unavailable until configured.)")
+        return 78  # distinct: configuration required
+    targets = LOAD_TARGETS + [("today", f"journal/daily/{today()}.md"),
+                              ("inbox", "inbox/captures/inbox.md")]
+    import echo_queue
+    # H2: sync any writes queued during a prior outage, best-effort and quiet on empty.
+    try:
+        synced = echo_queue.flush()
+        if synced:
+            print(f"(synced {synced} write(s) queued during a prior offline session)\n")
+        flagged = echo_queue.needs_attention()
+        if flagged:
+            print(f"NOTE: {len(flagged)} queued write(s) need attention (e.g. a capture "
+                  "stopped at the duplicate gate on replay) — resolve with capture "
+                  "--merge-into/--force; `echo.py flush` re-lists them.\n")
+    except Exception as exc:  # noqa: BLE001 — never let queue upkeep block a load
+        print(f"(queue flush skipped: {exc})\n", file=sys.stderr)
+
+    results, texts, listing_files, offline, marker_missing, heartbeat_absent = \
+        _load_gather(targets)

    if brief:
-        hb = texts.get("heartbeat")
-        if hb and hb.strip():  # follow-up: the pointed session log's key sections
-            sess_path = hb.strip().splitlines()[0].split(" @ ", 1)[0].strip()
-            if sess_path:
-                st2, b2 = request("GET", vault_url(sess_path))
-                if st2 == 200:
-                    texts["_pointed_session"] = b2.decode(errors="replace")
+        _fetch_pointed_session(texts)
        print(_render_brief(texts, listing_files, offline))
        return 0

@@ -22,17 +22,18 @@ import echo  # noqa: E402
 MIN_PY = (3, 9)


-def run() -> int:
-    reds = 0
+def run_op() -> dict:
+    """Core doctor (Phase 0): the readiness checks as data — {checks: [{ok, label,
+    detail}], endpoint, fatal, ok}. `fatal` names an early-exit condition (not
+    configured / unreachable) after which later checks were skipped."""
+    checks: list[dict] = []

    def line(ok: bool, label: str, detail: str = "") -> None:
-        nonlocal reds
-        reds += 0 if ok else 1
-        print(f"  [{'OK ' if ok else 'RED'}] {label}" + (f" — {detail}" if detail else ""))
+        checks.append({"ok": ok, "label": label, "detail": detail})

    import echo_config
    cfg = echo_config.load()
-    print(f"echo doctor — endpoint {cfg['endpoint'] or '(not configured)'}")
+    fatal = None

    # 1. interpreter
    line(sys.version_info >= MIN_PY, f"python >= {MIN_PY[0]}.{MIN_PY[1]}",
@@ -53,33 +54,50 @@ def run() -> int:
    line(bool(cfg["owner"]), "vault owner set",
         cfg["owner"] if cfg["owner"] else "optional, but recommended for third-person writes")
    if not echo_config.is_configured(cfg):
+        fatal = "not-configured"
+    else:
+        # 3. reachability + auth + marker, in one GET of the bootstrap marker
+        status, body = echo.request("GET", echo.vault_url("_agent/echo-vault.md"))
+        if status == 0:
+            line(False, "vault reachable", body.decode(errors="replace")[:120])
+            fatal = "unreachable"
+        else:
+            line(status not in (401, 403), "auth accepted",
+                 f"HTTP {status}" if status in (401, 403) else "")
+            if status == 404:
+                line(False, "vault bootstrapped", "marker absent — run bootstrap.py")
+            elif status == 200:
+                text = body.decode(errors="replace")
+                ver = next((ln.split(":", 1)[1].strip() for ln in text.splitlines()
+                            if ln.startswith("schema_version:")), "?")
+                line(True, "vault bootstrapped", f"schema_version={ver}")
+            else:
+                line(status < 400, "marker fetch", f"HTTP {status}")
+
+    reds = sum(1 for c in checks if not c["ok"])
+    return {"ok": reds == 0 and fatal is None, "action": "doctor",
+            "endpoint": cfg["endpoint"], "fatal": fatal, "reds": reds, "checks": checks}
+
+
+def run() -> int:
+    env = run_op()
+    print(f"echo doctor — endpoint {env['endpoint'] or '(not configured)'}")
+    for c in env["checks"]:
+        print(f"  [{'OK ' if c['ok'] else 'RED'}] {c['label']}"
+              + (f" — {c['detail']}" if c["detail"] else ""))
+    if env["fatal"] == "not-configured":
        print("\ndoctor: not configured — ask the operator for their key file and install it "
              "(`echo.py config import <file>`, or `config set --owner … --endpoint … --key …`), "
              "then re-run.")
        return 1
-
-    # 3. reachability + auth + marker, in one GET of the bootstrap marker
-    status, body = echo.request("GET", echo.vault_url("_agent/echo-vault.md"))
-    if status == 0:
-        line(False, "vault reachable", body.decode(errors="replace")[:120])
+    if env["fatal"] == "unreachable":
        print("\ndoctor: endpoint unreachable — is Obsidian + the Local REST API running?")
        return 1
-    line(status not in (401, 403), "auth accepted", f"HTTP {status}" if status in (401, 403) else "")
-    if status == 404:
-        line(False, "vault bootstrapped", "marker absent — run bootstrap.py")
-    elif status == 200:
-        text = body.decode(errors="replace")
-        ver = next((ln.split(":", 1)[1].strip() for ln in text.splitlines()
-                    if ln.startswith("schema_version:")), "?")
-        line(True, "vault bootstrapped", f"schema_version={ver}")
-    else:
-        line(status < 400, "marker fetch", f"HTTP {status}")
-
-    # 4. invariants pointer.
+    # invariants pointer.
    print("  [i] invariants — run `/echo-health` (vault_lint.py) for the full check")
-
-    print(f"\ndoctor: {'all green' if reds == 0 else f'{reds} issue(s) — see RED above'}")
-    return 1 if reds else 0
+    summary = "all green" if env["reds"] == 0 else f"{env['reds']} issue(s) — see RED above"
+    print(f"\ndoctor: {summary}")
+    return 1 if env["reds"] else 0


 if __name__ == "__main__":
@@ -37,12 +37,13 @@ LOG_HEADING = {


 # ---------------------------------------------------------------- resolve -----
-def resolve(mention: str) -> int:
+def resolve_op(mention: str) -> dict:
+    """Core resolve: mention -> match/candidates dict. No printing (Phase 0 — the same
+    return value backs the CLI wrapper and the MCP `echo_resolve` tool)."""
    index = idx_mod.load()
    slug, e = idx_mod.resolve(index, mention)
    if e:
-        print(json.dumps({"match": True, "slug": slug, **e}, ensure_ascii=False, indent=2))
-        return 0
+        return {"match": True, "slug": slug, **e}
    # No exact match — surface fuzzy candidates so a shortened/expanded name (e.g. "echo
    # memory" for the project "echo") reveals the existing note instead of looking absent.
    cands = idx_mod.fuzzy_candidates(index, mention)
@@ -54,21 +55,30 @@ def resolve(mention: str) -> int:
                       "these before creating a new note; reuse the right path or `echo.py link`.")
    else:
        out["note"] = "no index entry — derive a path with the right --kind and create it"
-    print(json.dumps(out, ensure_ascii=False, indent=2))
+    return out
+
+
+def resolve(mention: str) -> int:
+    print(json.dumps(resolve_op(mention), ensure_ascii=False, indent=2))
    return 0


 # ------------------------------------------------------------------- link -----
-def link(a_path: str, b_path: str, as_json: bool = False) -> int:
+def link_op(a_path: str, b_path: str) -> dict:
+    """Core link: reciprocal `## Related` links, returns the change envelope."""
+    import echo_output
    a_changed, b_changed = links.link_bidirectional(a_path, b_path)
+    return echo_output.envelope("link", {"a": a_path, "b": b_path,
+                                         "a_changed": a_changed, "b_changed": b_changed})
+
+
+def link(a_path: str, b_path: str, as_json: bool = False) -> int:
+    env = link_op(a_path, b_path)
    if as_json:
-        import echo_output
-        env = echo_output.envelope("link", {"a": a_path, "b": b_path,
-                                            "a_changed": a_changed, "b_changed": b_changed})
        print(json.dumps(env, ensure_ascii=False))
        return 0
-    print(f"ok: linked {a_path} <-> {b_path} "
-          f"(added: {'A' if a_changed else '-'}{'B' if b_changed else '-'})")
+    print(f"ok: linked {env['a']} <-> {env['b']} "
+          f"(added: {'A' if env['a_changed'] else '-'}{'B' if env['b_changed'] else '-'})")
    return 0


@@ -176,52 +186,52 @@ def _append_to_existing(path: str, kind: str, today_s: str, body_text: str) -> N
    echo.cmd_fm(path, "updated", json.dumps(today_s))


-def capture(kind: str | None, title: str, file_arg: str | None, status_v: str = "",
-            aliases=None, sources=None, tags=None, date: str | None = None,
-            domain: str = "business", inbox: bool = False, no_log: bool = False,
-            as_json: bool = False, dry_run: bool = False, force: bool = False,
-            merge_into: str | None = None) -> int:
+def capture_op(kind: str | None, title: str, file_arg: str | None = None, status_v: str = "",
+               aliases=None, sources=None, tags=None, date: str | None = None,
+               domain: str = "business", inbox: bool = False, no_log: bool = False,
+               dry_run: bool = False, force: bool = False,
+               merge_into: str | None = None, body_text: str | None = None) -> dict:
+    """Core capture (Phase 0): route + frontmatter + index + auto-link + agent-log,
+    returning an envelope dict — never printing to stdout (helper chatter from the
+    low-level verbs is redirected to stderr). Special outcomes are DATA, not exit
+    codes: `duplicate-gate` (ok=false, candidates), `queued:capture` (queued=true),
+    `dry-run:*`. Raises echo.EchoError on hard failure. `body_text` may be passed
+    directly (MCP path); otherwise it is read from file_arg/stdin."""
    import contextlib
-    import io
    import echo_output
    import echo_quality

-    real_stdout = sys.stdout
+    def chatter():
+        # Low-level verbs (cmd_put/cmd_append/cmd_fm) print progress lines; keep
+        # stdout clean for the envelope by routing them to stderr wholesale.
+        return contextlib.redirect_stdout(sys.stderr)

-    def quiet():
-        # M4: in --json mode, swallow the helper "ok:" chatter so stdout is clean JSON.
-        return contextlib.redirect_stdout(io.StringIO()) if as_json else contextlib.nullcontext()
-
-    def done(action: str, path: str, links: int = 0, ok: bool = True, dry: bool = False,
-             near=None) -> int:
-        if as_json:
-            act = f"dry-run:{action}" if dry else action
-            data = {"kind": kind, "path": path, "title": title, "links_added": links}
-            if near:
-                data["near_duplicates"] = near
-            env = echo_output.envelope(act, data, ok=ok)
-            print(json.dumps(env, ensure_ascii=False), file=real_stdout)
-        elif dry:
-            print(f"would {action} {kind or '-'} -> {path}")
-        # non-dry human output is the helper "ok:" lines + the summary printed by the caller.
-        return 0 if ok else 1
-
-    body_text = echo.read_body(file_arg).decode("utf-8", errors="replace")
+    if body_text is None:
+        body_text = echo.read_body(file_arg).decode("utf-8", errors="replace")
    today_s = echo.today()
    aliases = [a.strip() for a in (aliases or []) if a.strip()]
    sources = [s.strip() for s in (sources or []) if s.strip()]
    tags = [t.strip() for t in (tags or []) if t.strip()]

+    def env_for(action: str, path: str, links: int = 0, ok: bool = True, dry: bool = False,
+                near=None, **extra) -> dict:
+        act = f"dry-run:{action}" if dry else action
+        data = {"kind": kind, "path": path, "title": title, "links_added": links}
+        if near:
+            data["near_duplicates"] = near
+        data.update(extra)
+        return echo_output.envelope(act, data, ok=ok)
+
    # Unknown home -> defer to the inbox (a single idempotent capture line).
    if inbox or not kind:
        if dry_run:
-            return done("inbox", "inbox/captures/inbox.md", dry=True)
+            return env_for("inbox", "inbox/captures/inbox.md", dry=True)
        line = f"- {today_s}: {title}"
        if body_text.strip():
            line += f" — {body_text.strip().splitlines()[0]}"
-        with quiet():
+        with chatter():
            rc = echo.cmd_append("inbox/captures/inbox.md", line)
-        return done("inbox", "inbox/captures/inbox.md", ok=rc == 0)
+        return env_for("inbox", "inbox/captures/inbox.md", ok=rc == 0)

    slug = idx_mod.slugify(title)
    # Offline short-circuit (2.1.0). The routed path needs the entity index and several
@@ -235,9 +245,8 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
        if "unreachable" not in str(exc):
            raise
        if dry_run:
-            print("offline: vault unreachable — a real run would queue this capture "
-                  "for replay on the next reachable session.", file=real_stdout)
-            return 0
+            return echo_output.envelope("dry-run:queued:capture",
+                                        {"kind": kind, "title": title, "queued": True})
        import echo_queue
        echo_queue.enqueue_capture({
            "kind": kind, "title": title, "body_text": body_text, "status_v": status_v,
@@ -245,23 +254,16 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
            "domain": domain, "inbox": False, "no_log": no_log,
            "force": force, "merge_into": merge_into, "today": today_s,
        })
-        if as_json:
-            env = echo_output.envelope("queued:capture",
-                                       {"kind": kind, "title": title, "queued": True})
-            print(json.dumps(env, ensure_ascii=False), file=real_stdout)
-        else:
-            print(f"queued (offline): capture {kind} '{title}' — will replay through "
-                  "capture on the next reachable session", file=real_stdout)
-        return 0
+        return echo_output.envelope("queued:capture",
+                                    {"kind": kind, "title": title, "queued": True})
    match_slug, existing = idx_mod.resolve(index, title)
    # --merge-into: the operator has already identified the canonical entity (e.g. after
    # a duplicate-gate stop) — route this capture as an UPDATE to it, whatever the title.
    if merge_into and not existing:
        match_slug, existing = idx_mod.resolve(index, merge_into)
        if not existing:
-            print(f"echo_ops: --merge-into '{merge_into}' matches no entity in the index "
-                  f"(try `resolve` first)", file=sys.stderr)
-            return 2
+            raise echo.EchoError(f"--merge-into '{merge_into}' matches no entity in the "
+                                 "index (try `resolve` first)", 2)
    existing_reachable = bool(existing and echo.request("GET", echo.vault_url(existing["path"]))[0] == 200)

    # Pre-write duplicate gate: a strong fuzzy candidate means this title is very likely
@@ -276,40 +278,26 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
                     for s, c, sc in idx_mod.gate_candidates(index, title, kind=kind,
                                                             threshold=DUP_GATE)]
    if gate_hits:
-        if as_json:
-            env = echo_output.envelope("duplicate-gate", {
-                "kind": kind, "title": title, "candidates": gate_hits,
-                "note": "likely duplicate — re-run with --merge-into <slug> to update the "
-                        "existing entity, or --force to create anyway"}, ok=False)
-            print(json.dumps(env, ensure_ascii=False), file=real_stdout)
-        else:
-            print(f"STOP: '{title}' likely already exists — not creating a duplicate.",
-                  file=real_stdout)
-            for c in gate_hits:
-                print(f"  candidate: {c['slug']} -> {c['path']} (score {c['score']})",
-                      file=real_stdout)
-            print("  re-run with --merge-into <slug> to update the existing entity, "
-                  "or --force to create anyway.", file=real_stdout)
-        return 76  # distinct exit: duplicate gate (cf. 75 lock, 78 config)
+        return echo_output.envelope("duplicate-gate", {
+            "kind": kind, "title": title, "candidates": gate_hits,
+            "note": "likely duplicate — re-run with --merge-into <slug> to update the "
+                    "existing entity, or --force to create anyway"}, ok=False)

    if dry_run:
        if existing_reachable:
-            return done("update", existing["path"], dry=True)
+            return env_for("update", existing["path"], dry=True)
        s2 = echo_quality.safe_slug(set(index.get("entities", {}).keys()), slug)
        near = [c.get("path") for _, c, _ in idx_mod.fuzzy_candidates(index, title)][:3]
-        plan = done("create", idx_mod.derive_path(kind, s2, date=date, domain=domain),
-                    dry=True, near=near)
-        if (not as_json and not force
-                and idx_mod.gate_candidates(index, title, kind=kind, threshold=DUP_GATE)):
-            # (in --json mode the near_duplicates field carries this; keep stdout clean)
-            print("note: a real run would STOP at the duplicate gate — use --merge-into "
-                  "or --force.", file=real_stdout)
-        return plan
+        would_gate = (not force
+                      and bool(idx_mod.gate_candidates(index, title, kind=kind,
+                                                       threshold=DUP_GATE)))
+        return env_for("create", idx_mod.derive_path(kind, s2, date=date, domain=domain),
+                       dry=True, near=near, would_gate=would_gate)

    near_dupes: list[str] = []
    index_title = title          # title to record in the index entry
    index_aliases = list(aliases)
-    with quiet():
+    with chatter():
        if existing_reachable:
            # Reuse the matched entity's CANONICAL slug — never re-slug from the mention, or
            # two index slugs end up pointing at one note. Keep its title; learn the mention
@@ -374,13 +362,61 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
        if not no_log:
            ensure_daily_log(f"- {today_s}: {action} {kind} [[{links.link_token(path)}]]")

+    return env_for(action, path, links=linked, near=near_dupes)
+
+
+def capture(kind: str | None, title: str, file_arg: str | None, status_v: str = "",
+            aliases=None, sources=None, tags=None, date: str | None = None,
+            domain: str = "business", inbox: bool = False, no_log: bool = False,
+            as_json: bool = False, dry_run: bool = False, force: bool = False,
+            merge_into: str | None = None) -> int:
+    """CLI wrapper around capture_op: prints exactly what pre-Phase-0 capture printed
+    (human prose or the --json envelope) and maps envelope outcomes to exit codes
+    (76 duplicate-gate, 2 usage-class errors, 0 otherwise)."""
+    try:
+        env = capture_op(kind, title, file_arg, status_v=status_v, aliases=aliases,
+                         sources=sources, tags=tags, date=date, domain=domain,
+                         inbox=inbox, no_log=no_log, dry_run=dry_run, force=force,
+                         merge_into=merge_into)
+    except echo.EchoError as exc:
+        print(f"echo_ops: {exc}", file=sys.stderr)
+        return getattr(exc, "code", 1)
+    action = env.get("action", "")
+
    if as_json:
-        done(action, path, links=linked, near=near_dupes)
-    else:
-        print(f"ok: {action} {kind} -> {path}" + (f"; auto-linked {linked}" if linked else ""))
-        if near_dupes:
-            kind_word = "entity" if len(near_dupes) == 1 else "entities"
-            print(f"WARNING: similar existing {kind_word} ({', '.join(near_dupes)}) — if this is "
-                  f"the same thing, merge or `echo.py link` instead of keeping a duplicate.",
-                  file=real_stdout)
-    return 0
+        print(json.dumps(env, ensure_ascii=False))
+        if action == "duplicate-gate":
+            return 76
+        return 0 if env.get("ok") else 1
+
+    if action == "duplicate-gate":
+        print(f"STOP: '{title}' likely already exists — not creating a duplicate.")
+        for c in env.get("candidates", []):
+            print(f"  candidate: {c['slug']} -> {c['path']} (score {c['score']})")
+        print("  re-run with --merge-into <slug> to update the existing entity, "
+              "or --force to create anyway.")
+        return 76  # distinct exit: duplicate gate (cf. 75 lock, 78 config)
+    if action == "queued:capture":
+        print(f"queued (offline): capture {kind} '{title}' — will replay through "
+              "capture on the next reachable session")
+        return 0
+    if action == "dry-run:queued:capture":
+        print("offline: vault unreachable — a real run would queue this capture "
+              "for replay on the next reachable session.")
+        return 0
+    if action.startswith("dry-run:"):
+        print(f"would {action.split(':', 1)[1]} {kind or '-'} -> {env.get('path')}")
+        if env.get("would_gate"):
+            print("note: a real run would STOP at the duplicate gate — use --merge-into "
+                  "or --force.")
+        return 0
+    if action == "inbox":
+        return 0 if env.get("ok") else 1
+    print(f"ok: {action} {kind} -> {env.get('path')}"
+          + (f"; auto-linked {env['links_added']}" if env.get("links_added") else ""))
+    near = env.get("near_duplicates")
+    if near:
+        kind_word = "entity" if len(near) == 1 else "entities"
+        print(f"WARNING: similar existing {kind_word} ({', '.join(near)}) — if this is "
+              f"the same thing, merge or `echo.py link` instead of keeping a duplicate.")
+    return 0 if env.get("ok") else 1
@@ -396,32 +396,10 @@ def _doc_summary(path: str, text: str) -> dict:
    return out


-def _brief(path: str, score: float | None = None, via: str | None = None) -> None:
-    text = links.get_text(path)
-    if text is None:
-        return
-    info = _doc_summary(path, text)
-    head = f"\n### {path}"
-    if score is not None:
-        head += f"   (score {score:.2f})"
-    if via:
-        head += f"   (via {via})"
-    print(head)
-    stamps = []
-    if info.get("type"):
-        stamps.append(f"type: {info['type']}")
-    if info.get("updated"):
-        stamps.append(f"updated: {info['updated']}")
-    if info.get("status"):
-        stamps.append(f"status: {info['status']}")
-    if stamps:
-        print("_" + " · ".join(stamps) + "_")   # staleness is visible, not guessed
-    if info.get("excerpt"):
-        print(info["excerpt"])
-
-
 # ------------------------------------------------------------------- entrypoint
-def recall(query, limit: int = 8, as_json: bool = False) -> int:
+def recall_op(query, limit: int = 8) -> dict:
+    """Core recall (Phase 0): hybrid BM25 + graph, returning the structured envelope
+    ({query, primary, linked}) that backs both the CLI renderings and the MCP tool."""
    q = " ".join(query) if isinstance(query, list) else query
    today_s = echo.today()
    index = idx_mod.load()
@@ -464,31 +442,49 @@ def recall(query, limit: int = 8, as_json: bool = False) -> int:
    # --- graph layer ----------------------------------------------------------
    neighbours = expand_graph(hits, nmap, base, max_hops=MAX_HOPS)

+    import echo_output
+    texts = echo.read_many(hits + [p for p, _ in neighbours[: 2 * limit]])
+    primary = []
+    for p in hits:
+        if texts.get(p) is None:
+            continue
+        primary.append({**_doc_summary(p, texts[p]),
+                        "score": round(base.get(p, 0.0), 3)})
+    linked = []
+    for p, (sc, via) in neighbours[: 2 * limit]:
+        if texts.get(p) is None:
+            continue
+        linked.append({**_doc_summary(p, texts[p]),
+                       "score": round(sc, 3), "via": via})
+    return echo_output.envelope("recall", {"query": q, "primary": primary,
+                                           "linked": linked})
+
+
+def _print_hit(info: dict) -> None:
+    """Prose rendering of one recall hit — same fields the envelope carries."""
+    head = f"\n### {info['path']}"
+    if info.get("score") is not None:
+        head += f"   (score {info['score']:.2f})"
+    if info.get("via"):
+        head += f"   (via {info['via']})"
+    print(head)
+    stamps = [f"{k}: {info[k]}" for k in ("type", "updated", "status") if info.get(k)]
+    if stamps:
+        print("_" + " · ".join(stamps) + "_")   # staleness is visible, not guessed
+    if info.get("excerpt"):
+        print(info["excerpt"])
+
+
+def recall(query, limit: int = 8, as_json: bool = False) -> int:
+    env = recall_op(query, limit=limit)
    if as_json:
-        import echo_output
-        texts = echo.read_many(hits + [p for p, _ in neighbours[: 2 * limit]])
-        primary = []
-        for p in hits:
-            if texts.get(p) is None:
-                continue
-            primary.append({**_doc_summary(p, texts[p]),
-                            "score": round(base.get(p, 0.0), 3)})
-        linked = []
-        for p, (sc, via) in neighbours[: 2 * limit]:
-            if texts.get(p) is None:
-                continue
-            linked.append({**_doc_summary(p, texts[p]),
-                           "score": round(sc, 3), "via": via})
-        env = echo_output.envelope("recall", {"query": q, "primary": primary,
-                                              "linked": linked})
        print(json.dumps(env, ensure_ascii=False))
        return 0
-
-    print(f"== recall: {q} ==")
-    print(f"\n# Primary hits ({len(hits)})")
-    for p in hits:
-        _brief(p, score=base.get(p))
-    print(f"\n# Linked context ({len(neighbours)})")
-    for p, (sc, via) in neighbours[: 2 * limit]:
-        _brief(p, score=sc, via=via)
+    print(f"== recall: {env['query']} ==")
+    print(f"\n# Primary hits ({len(env['primary'])})")
+    for info in env["primary"]:
+        _print_hit(info)
+    print(f"\n# Linked context ({len(env['linked'])})")
+    for info in env["linked"]:
+        _print_hit(info)
    return 0
@@ -99,43 +99,69 @@ def preview(proposals: list[dict]) -> str:
    return "\n".join(rows)


-def apply(proposals: list[dict], confirm: bool = False) -> int:
-    """Validate -> classify -> preview; with confirm=True, apply each via echo_ops.capture
-    (which routes/indexes/links/logs, each self-lock-guarded). Without confirm it is a
-    dry-run that writes nothing — the preview IS the confirmation step."""
+def apply_op(proposals: list[dict], confirm: bool = False) -> dict:
+    """Core reflect (Phase 0): validate -> classify -> (apply via capture_op). Returns
+    an envelope — rows carry the classified preview; with confirm, `results` carries
+    each proposal's capture outcome. Never prints to stdout."""
+    import echo_output
    valid, errors = validate(proposals)
-    for e in errors:
-        print(f"skip: {e}", file=sys.stderr)
-    if not valid:
-        print("reflect: no valid proposals to apply.")
-        return 0
-
-    classify(valid)
-    counts = {a: sum(1 for p in valid if p.get("_action") == a) for a in ("create", "update", "inbox", "error")}
-    print(f"reflect: {len(valid)} proposal(s) — "
-          f"{counts['create']} new, {counts['update']} update, {counts['inbox']} inbox"
-          + (f", {counts['error']} error" if counts["error"] else ""))
-    print(preview(valid))
-
-    if not confirm:
-        print("\nreflect: dry-run — re-run with --apply to write these to memory.")
-        return 0
+    rows: list[dict] = []
+    counts: dict[str, int] = {}
+    if valid:
+        classify(valid)
+        counts = {a: sum(1 for p in valid if p.get("_action") == a)
+                  for a in ("create", "update", "inbox", "error")}
+        rows = [{"action": p.get("_action"), "kind": p.get("kind"),
+                 "title": p["title"], "path": p.get("_path")} for p in valid]
+    data = {"proposals": len(proposals or []), "valid": len(valid), "errors": errors,
+            "counts": counts, "rows": rows, "dry_run": not confirm,
+            "applied": 0, "gated": 0, "results": []}
+    if not valid or not confirm:
+        return echo_output.envelope("reflect", data)

    import echo_ops  # lazy: the apply path pulls in the capture/index/link stack
    applied = gated = 0
+    results = []
    for p in valid:
        if p.get("_action") == "error":
            continue
-        bodyfile = echo.temp_file((p.get("body") or "").encode()) if p.get("body") else None
-        rc = echo_ops.capture(
-            p.get("kind"), p["title"], bodyfile,
+        env = echo_ops.capture_op(
+            p.get("kind"), p["title"], body_text=p.get("body") or "",
            aliases=p.get("aliases") or [], sources=p.get("sources") or [],
            tags=p.get("tags") or [],
            date=p.get("date"), domain=p.get("domain", "business"),
            inbox=bool(p.get("inbox")) or not p.get("kind"))
-        applied += 1 if rc == 0 else 0
-        gated += 1 if rc == 76 else 0
-    print(f"reflect: applied {applied}/{len(valid)} proposal(s)."
+        act = env.get("action", "")
+        if act == "duplicate-gate":
+            gated += 1
+        elif env.get("ok"):
+            applied += 1
+        results.append({"title": p["title"], "action": act,
+                        "path": env.get("path"), "ok": bool(env.get("ok"))})
+    data.update(applied=applied, gated=gated, results=results)
+    return echo_output.envelope("reflect", data)
+
+
+def apply(proposals: list[dict], confirm: bool = False) -> int:
+    """CLI wrapper: same preview/summary text as pre-Phase-0. Without confirm it is a
+    dry-run that writes nothing — the preview IS the confirmation step."""
+    env = apply_op(proposals, confirm=confirm)
+    for e in env["errors"]:
+        print(f"skip: {e}", file=sys.stderr)
+    if not env["valid"]:
+        print("reflect: no valid proposals to apply.")
+        return 0
+    counts = env["counts"]
+    print(f"reflect: {env['valid']} proposal(s) — "
+          f"{counts['create']} new, {counts['update']} update, {counts['inbox']} inbox"
+          + (f", {counts['error']} error" if counts["error"] else ""))
+    print("\n".join(f"  {r['action'] or '?':7} | {('-' if r['action'] == 'inbox' else r['kind'] or '?'):9} "
+                    f"| {r['title']}  ->  {r['path'] or '?'}" for r in env["rows"]))
+    if env["dry_run"]:
+        print("\nreflect: dry-run — re-run with --apply to write these to memory.")
+        return 0
+    gated = env["gated"]
+    print(f"reflect: applied {env['applied']}/{env['valid']} proposal(s)."
          + (f" {gated} stopped at the duplicate gate — re-propose with the existing "
             f"entity's title (or capture --merge-into <slug>)." if gated else ""))
    return 0
@@ -72,36 +72,34 @@ def validate(bundle: dict) -> tuple[str, str]:
    return path, line


-def session_end(bundle: dict, apply: bool = False) -> int:
+def session_end_op(bundle: dict, apply: bool = False) -> dict:
+    """Core session-end (Phase 0): validate -> plan -> (apply). Returns an envelope
+    (plan rows + per-step results); never prints to stdout — helper chatter from the
+    underlying verbs routes to stderr. Raises echo.EchoError(2) on a bad bundle,
+    BEFORE any write."""
+    import contextlib
+    import echo_output
    import echo_reflect
    path, line = validate(bundle)
    scope = str(bundle.get("scope") or "").strip()
    proposals = bundle.get("reflect") or []

    valid, errors = echo_reflect.validate(proposals)
-    for e in errors:
-        print(f"skip: {e}", file=sys.stderr)
-
-    print(f"session-end plan ({'APPLY' if apply else 'dry-run'}):")
-    print(f"  1. session log -> {path}")
-    print(f"  2. agent-log line: {line}")
+    rows: list[dict] = []
    if valid:
        echo_reflect.classify(valid)
-        print(f"  3. reflect: {len(valid)} proposal(s)")
-        print(echo_reflect.preview(valid))
-    else:
-        print("  3. reflect: (none)")
-    print(f"  4. scope set: {scope!r}" if scope else "  4. scope: (unchanged)")
-    print(f"  5. heartbeat -> {path} @ <now>   (written LAST — the commit marker)")
-
+        rows = [{"action": p.get("_action"), "kind": p.get("kind"),
+                 "title": p["title"], "path": p.get("_path")} for p in valid]
+    data = {"path": path, "agent_log_line": line, "scope": scope or None,
+            "reflect_rows": rows, "reflect_errors": errors,
+            "dry_run": not apply, "steps": {}}
    if not apply:
-        print("\nsession-end: dry-run — re-run with --apply to write.")
-        return 0
+        return echo_output.envelope("session-end", data)

    import echo_concurrency
    import echo_ops
    steps: dict[str, str] = {}
-    with echo_concurrency.vault_lock():
+    with echo_concurrency.vault_lock(), contextlib.redirect_stdout(sys.stderr):
        # 1. The session log itself. A hard failure here aborts the whole bundle —
        #    nothing after it (including the heartbeat) runs, so orientation state
        #    can never point at a log that was never written.
@@ -116,14 +114,13 @@ def session_end(bundle: dict, apply: bool = False) -> int:
            for p in valid:
                if p.get("_action") == "error":
                    continue
-                bodyfile = echo.temp_file((p.get("body") or "").encode()) if p.get("body") else None
-                rc = echo_ops.capture(
-                    p.get("kind"), p["title"], bodyfile,
+                env = echo_ops.capture_op(
+                    p.get("kind"), p["title"], body_text=p.get("body") or "",
                    aliases=p.get("aliases") or [], sources=p.get("sources") or [],
                    tags=p.get("tags") or [], date=p.get("date"),
                    domain=p.get("domain", "business"),
                    inbox=bool(p.get("inbox")) or not p.get("kind"))
-                gated += 1 if rc == 76 else 0
+                gated += 1 if env.get("action") == "duplicate-gate" else 0
            steps["reflect"] = f"{len(valid) - gated}/{len(valid)} applied" \
                + (f", {gated} gated (re-propose with --merge-into)" if gated else "")
        else:
@@ -139,6 +136,30 @@ def session_end(bundle: dict, apply: bool = False) -> int:
                     echo.temp_file(f"{path} @ {echo.now_iso()}\n".encode("utf-8")))
        steps["heartbeat"] = "ok"

+    data["steps"] = steps
+    return echo_output.envelope("session-end", data)
+
+
+def session_end(bundle: dict, apply: bool = False) -> int:
+    """CLI wrapper: same plan/summary text as before; envelope logic in session_end_op."""
+    env = session_end_op(bundle, apply=apply)
+    for e in env["reflect_errors"]:
+        print(f"skip: {e}", file=sys.stderr)
+    print(f"session-end plan ({'APPLY' if apply else 'dry-run'}):")
+    print(f"  1. session log -> {env['path']}")
+    print(f"  2. agent-log line: {env['agent_log_line']}")
+    rows = env["reflect_rows"]
+    if rows:
+        print(f"  3. reflect: {len(rows)} proposal(s)")
+        print("\n".join(f"  {r['action'] or '?':7} | {('-' if r['action'] == 'inbox' else r['kind'] or '?'):9} "
+                        f"| {r['title']}  ->  {r['path'] or '?'}" for r in rows))
+    else:
+        print("  3. reflect: (none)")
+    print(f"  4. scope set: {env['scope']!r}" if env["scope"] else "  4. scope: (unchanged)")
+    print(f"  5. heartbeat -> {env['path']} @ <now>   (written LAST — the commit marker)")
+    if env["dry_run"]:
+        print("\nsession-end: dry-run — re-run with --apply to write.")
+        return 0
    print("\nsession-end: done — "
-          + "; ".join(f"{k}: {v}" for k, v in steps.items()))
+          + "; ".join(f"{k}: {v}" for k, v in env["steps"].items()))
    return 0
@@ -55,19 +55,25 @@ def parse_inbox(text: str) -> list[dict]:
    return items


-def list_inbox(as_json: bool = False) -> int:
+def list_op() -> dict:
+    """Core inbox listing (Phase 0): structured captures envelope, no printing."""
+    import echo_output
    status, body = echo.request("GET", echo.vault_url(INBOX_PATH))
    if status == 404:
        items = []
    else:
        echo.check(status, body, f"triage list {INBOX_PATH}")
        items = parse_inbox(body.decode(errors="replace"))
+    return echo_output.envelope("triage-list", {"path": INBOX_PATH, "items": items,
+                                                "count": len(items)})
+
+
+def list_inbox(as_json: bool = False) -> int:
+    env = list_op()
    if as_json:
-        import echo_output
-        env = echo_output.envelope("triage-list", {"path": INBOX_PATH, "items": items,
-                                                   "count": len(items)})
        print(json.dumps(env, ensure_ascii=False))
        return 0
+    items = env["items"]
    if not items:
        print("triage: inbox is empty — nothing to route.")
        return 0
@@ -80,53 +86,76 @@ def list_inbox(as_json: bool = False) -> int:
    return 0


-def apply(proposals: list[dict], confirm: bool = False) -> int:
-    """Validate -> classify -> preview; with confirm, route each via capture AND write
-    the processing-log audit line. Mirrors echo_reflect.apply's contract exactly."""
+def route_op(proposals: list[dict], confirm: bool = False) -> dict:
+    """Core triage routing (Phase 0): reflect pipeline + processing-log audit lines.
+    Returns an envelope (rows = classified preview; results = per-item outcomes);
+    never prints to stdout (audit-append chatter routes to stderr)."""
+    import contextlib
+    import echo_output
    import echo_reflect
    valid, errors = echo_reflect.validate(proposals)
-    for e in errors:
-        print(f"skip: {e}", file=sys.stderr)
-    if not valid:
-        print("triage: no valid proposals to route.")
-        return 0
-
-    echo_reflect.classify(valid)
-    counts = {a: sum(1 for p in valid if p.get("_action") == a)
-              for a in ("create", "update", "inbox", "error")}
-    print(f"triage: {len(valid)} proposal(s) — "
-          f"{counts['create']} new, {counts['update']} update, {counts['inbox']} stay-in-inbox"
-          + (f", {counts['error']} error" if counts["error"] else ""))
-    print(echo_reflect.preview(valid))
-
-    if not confirm:
-        print("\ntriage: dry-run — re-run with --apply to route these and log the moves.")
-        return 0
+    rows: list[dict] = []
+    counts: dict[str, int] = {}
+    if valid:
+        echo_reflect.classify(valid)
+        counts = {a: sum(1 for p in valid if p.get("_action") == a)
+                  for a in ("create", "update", "inbox", "error")}
+        rows = [{"action": p.get("_action"), "kind": p.get("kind"),
+                 "title": p["title"], "path": p.get("_path")} for p in valid]
+    data = {"proposals": len(proposals or []), "valid": len(valid), "errors": errors,
+            "counts": counts, "rows": rows, "dry_run": not confirm,
+            "routed": 0, "gated": 0, "log": f"{LOG_DIR}/{echo.today()}.md", "results": []}
+    if not valid or not confirm:
+        return echo_output.envelope("triage", data)

    import echo_ops
    today_s = echo.today()
-    applied = gated = 0
+    routed = gated = 0
+    results = []
    for p in valid:
-        if p.get("_action") == "error":
-            continue
-        if p.get("_action") == "inbox":
+        if p.get("_action") in ("error", "inbox"):
            continue  # routing an inbox line back to the inbox is a no-op, not a move
-        bodyfile = echo.temp_file((p.get("body") or "").encode()) if p.get("body") else None
-        rc = echo_ops.capture(
-            p.get("kind"), p["title"], bodyfile,
+        env = echo_ops.capture_op(
+            p.get("kind"), p["title"], body_text=p.get("body") or "",
            aliases=p.get("aliases") or [], sources=p.get("sources") or [],
            tags=p.get("tags") or [],
            date=p.get("date"), domain=p.get("domain", "business"))
-        if rc == 76:
+        act = env.get("action", "")
+        results.append({"title": p["title"], "action": act,
+                        "path": env.get("path"), "ok": bool(env.get("ok"))})
+        if act == "duplicate-gate":
            gated += 1
            continue
-        if rc != 0:
+        if not env.get("ok"):
            continue
-        applied += 1
+        routed += 1
        original = (p.get("line") or p["title"]).strip()
-        echo.cmd_append(f"{LOG_DIR}/{today_s}.md",
-                        f"- {original} → {p.get('_path', '?')}")
-    print(f"triage: routed {applied}/{len(valid)} item(s); audit in {LOG_DIR}/{today_s}.md. "
+        with contextlib.redirect_stdout(sys.stderr):
+            echo.cmd_append(f"{LOG_DIR}/{today_s}.md",
+                            f"- {original} → {p.get('_path', '?')}")
+    data.update(routed=routed, gated=gated, results=results)
+    return echo_output.envelope("triage", data)
+
+
+def apply(proposals: list[dict], confirm: bool = False) -> int:
+    """CLI wrapper: same preview/summary text as pre-Phase-0. Mirrors reflect's contract."""
+    env = route_op(proposals, confirm=confirm)
+    for e in env["errors"]:
+        print(f"skip: {e}", file=sys.stderr)
+    if not env["valid"]:
+        print("triage: no valid proposals to route.")
+        return 0
+    counts = env["counts"]
+    print(f"triage: {env['valid']} proposal(s) — "
+          f"{counts['create']} new, {counts['update']} update, {counts['inbox']} stay-in-inbox"
+          + (f", {counts['error']} error" if counts["error"] else ""))
+    print("\n".join(f"  {r['action'] or '?':7} | {('-' if r['action'] == 'inbox' else r['kind'] or '?'):9} "
+                    f"| {r['title']}  ->  {r['path'] or '?'}" for r in env["rows"]))
+    if env["dry_run"]:
+        print("\ntriage: dry-run — re-run with --apply to route these and log the moves.")
+        return 0
+    gated = env["gated"]
+    print(f"triage: routed {env['routed']}/{env['valid']} item(s); audit in {env['log']}. "
          "Originals kept in the inbox (deletion is explicit-only)."
          + (f" {gated} stopped at the duplicate gate — re-propose with the existing "
             f"entity's title or use capture --merge-into." if gated else ""))
@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""test_mcp_server.py — echo-mcp (2.2) end-to-end against the mock vault.
+
+Spawns mcp-server/app.py + mock_olrapi and drives the real streamable-HTTP MCP
+protocol: health (open), auth (401), initialize, tools/list, and the memory-day
+tools/call flow (capture -> duplicate gate as DATA -> merge_into -> recall ->
+get_note -> log_session with heartbeat commit).
+
+Requires the `mcp` package in the interpreter that runs the SERVER. Set
+ECHO_MCP_PYTHON to a venv python that has it; otherwise the current interpreter
+is tried and the suite SKIPS (exit 0) when the SDK is absent — the other suites
+don't depend on it.
+
+Run:  python test_mcp_server.py [--port 8862] [--mcp-port 8767]
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+APP = HERE.parent / "mcp-server" / "app.py"
+KEY = "test-key-not-a-real-secret"
+TOKEN = "local-test-token"
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    print(f"{'ok  ' if cond else 'FAIL'} {name}" + (f"  -- {detail}" if not cond else ""))
+    if not cond:
+        failures.append(name)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--port", type=int, default=8862)
+    ap.add_argument("--mcp-port", type=int, default=8767)
+    a = ap.parse_args()
+    mock_base = f"http://127.0.0.1:{a.port}"
+    mcp_url = f"http://127.0.0.1:{a.mcp_port}/mcp"
+
+    server_py = os.environ.get("ECHO_MCP_PYTHON") or sys.executable
+    probe = subprocess.run([server_py, "-c", "import mcp"], capture_output=True)
+    if probe.returncode != 0:
+        print("SKIP: `mcp` SDK not installed for the server interpreter "
+              "(set ECHO_MCP_PYTHON to a venv python that has it)")
+        return 0
+
+    def http(method, url, body=None, headers=None, timeout=30):
+        data = body.encode() if isinstance(body, str) else body
+        req = urllib.request.Request(url, data=data, method=method, headers=headers or {})
+        try:
+            with urllib.request.urlopen(req, timeout=timeout) as r:
+                return r.status, r.read().decode("utf-8", "replace")
+        except urllib.error.HTTPError as e:
+            return e.code, e.read().decode("utf-8", "replace")
+        except Exception as e:  # noqa: BLE001
+            return 0, str(e)
+
+    _id = [0]
+
+    def rpc(method, params):
+        _id[0] += 1
+        st, body = http("POST", mcp_url, json.dumps(
+            {"jsonrpc": "2.0", "id": _id[0], "method": method, "params": params}),
+            headers={"Content-Type": "application/json",
+                     "Accept": "application/json, text/event-stream",
+                     "Authorization": f"Bearer {TOKEN}"})
+        return st, (json.loads(body) if body.strip().startswith("{") else {})
+
+    def call(name, args):
+        st, d = rpc("tools/call", {"name": name, "arguments": args})
+        res = d.get("result") or {}
+        payload = {}
+        if res.get("content"):
+            try:
+                payload = json.loads(res["content"][0]["text"])
+            except Exception:  # noqa: BLE001
+                payload = {}
+        return res.get("isError", False), payload
+
+    mock = subprocess.Popen([sys.executable, str(HERE / "mock_olrapi.py"), "--port", str(a.port)],
+                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    srv = None
+    try:
+        for _ in range(50):
+            try:
+                urllib.request.urlopen(f"{mock_base}/__debug__reset", data=b"", timeout=1)
+                break
+            except Exception:
+                time.sleep(0.1)
+        http("PUT", f"{mock_base}/vault/_agent/echo-vault.md",
+             "---\nschema_version: 4\n---\n# marker\n",
+             headers={"Authorization": f"Bearer {KEY}"})
+
+        env = dict(os.environ, ECHO_BASE=mock_base, ECHO_KEY=KEY, ECHO_MCP_TOKEN=TOKEN,
+                   ECHO_MCP_PORT=str(a.mcp_port), ECHO_STATE_DIR=tempfile.mkdtemp(),
+                   ECHO_TODAY="2026-07-28")
+        srv = subprocess.Popen([server_py, str(APP)], env=env,
+                               stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+        for _ in range(60):
+            st, _b = http("GET", f"http://127.0.0.1:{a.mcp_port}/health", timeout=2)
+            if st == 200:
+                break
+            time.sleep(0.5)
+
+        st, body = http("GET", f"http://127.0.0.1:{a.mcp_port}/health")
+        try:
+            hb = json.loads(body)
+        except Exception:  # noqa: BLE001
+            hb = {}
+        check("health is open and green", st == 200 and hb.get("ok") is True
+              and hb.get("vault_reachable") is True, body[:200])
+        st, _b = http("POST", mcp_url, "{}", headers={"Content-Type": "application/json"})
+        check("MCP endpoint requires the bearer token (401)", st == 401, str(st))
+
+        st, d = rpc("initialize", {"protocolVersion": "2025-06-18", "capabilities": {},
+                                   "clientInfo": {"name": "eval", "version": "0"}})
+        check("initialize answers with serverInfo",
+              d.get("result", {}).get("serverInfo", {}).get("name") == "echo-mcp", json.dumps(d)[:200])
+        st, d = rpc("tools/list", {})
+        tools = [t["name"] for t in d.get("result", {}).get("tools", [])]
+        check("tools/list exposes the full profile (14 tools)", len(tools) == 14, str(tools))
+
+        e, b = call("echo_capture", {"title": "Vera Lumen", "kind": "person",
+                                     "body": "CTO at Fluxcorp, met at the summit."})
+        check("capture create over MCP", not e and b.get("action") == "created"
+              and b.get("path") == "resources/people/vera-lumen.md", json.dumps(b))
+        e, b = call("echo_capture", {"title": "Vera Lumen Jr", "kind": "person"})
+        check("duplicate gate is data, not isError", not e
+              and b.get("action") == "duplicate-gate" and b.get("candidates"), json.dumps(b))
+        e, b = call("echo_capture", {"title": "Vera Lumen Jr", "kind": "person",
+                                     "merge_into": "vera-lumen", "body": "Follow-up."})
+        check("merge_into resolves the gate as an update", not e and b.get("action") == "updated",
+              json.dumps(b))
+        e, b = call("echo_recall", {"query": "fluxcorp", "budget_chars": 900})
+        check("recall over MCP finds by body term", not e and any(
+            h["path"] == "resources/people/vera-lumen.md" for h in b.get("primary", [])),
+            json.dumps(b)[:300])
+        e, b = call("echo_get_note", {"path": "resources/people/vera-lumen.md"})
+        check("get_note returns frontmatter + content", not e
+              and b.get("frontmatter", {}).get("type") == "person", json.dumps(b)[:200])
+        e, b = call("echo_get_note", {"path": "../etc/passwd"})
+        check("get_note rejects path traversal", not b.get("ok"))
+        e, b = call("echo_log_session", {"slug": "mcp-e2e", "hhmm": "2345", "apply": True,
+                                         "log_body": "---\ntype: session-log\n---\n# S\n\n## Goal\ne2e\n"})
+        check("log_session commits with the heartbeat", not e
+              and b.get("steps", {}).get("heartbeat") == "ok", json.dumps(b))
+        e, b = call("echo_capture", {"title": "Bad Kind", "kind": "wizard"})
+        check("unknown kind rejected with the valid list", not b.get("ok")
+              and "wizard" in b.get("error", ""), json.dumps(b))
+
+        print(f"\n{len(failures)} failure(s)" if failures else "\nall mcp-server tests passed")
+        return 1 if failures else 0
+    finally:
+        if srv:
+            srv.terminate()
+        mock.terminate()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""test_ops_api.py — Phase 0 (return-not-print) contract: every high-level op has a
+core `*_op` function that RETURNS an envelope dict and prints nothing to stdout.
+
+This is the seam the MCP server wraps (docs/MCP-SERVER-SPEC.md §4): the CLI wrappers
+are tested by the other suites; here we call the cores in-process against the mock and
+assert (a) the envelope shapes and (b) stdout purity — a stray print() would corrupt
+an MCP stdio/HTTP response stream.
+
+Run:  python test_ops_api.py [--port 8850]
+"""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import io
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.request
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+SCRIPTS = HERE.parent / "echo-memory.plugin.src" / "skills" / "echo-memory" / "scripts"
+KEY = "test-key-not-a-real-secret"
+
+ap = argparse.ArgumentParser()
+ap.add_argument("--port", type=int, default=8850)
+a = ap.parse_args()
+BASE = f"http://127.0.0.1:{a.port}"
+
+# env must be set BEFORE importing echo (it resolves config at import time)
+os.environ.update(ECHO_BASE=BASE, ECHO_KEY=KEY, ECHO_TODAY="2026-06-21",
+                  ECHO_NOW="2300", ECHO_STATE_DIR=tempfile.mkdtemp(), ECHO_VERIFY="0")
+sys.path.insert(0, str(SCRIPTS))
+
+failures = []
+
+
+def check(name, cond, detail=""):
+    print(f"{'ok  ' if cond else 'FAIL'} {name}" + (f"  -- {detail}" if not cond else ""))
+    if not cond:
+        failures.append(name)
+
+
+def http(method, url, body=None):
+    data = body.encode() if isinstance(body, str) else body
+    req = urllib.request.Request(url, data=data, method=method,
+                                 headers={"Authorization": f"Bearer {KEY}"})
+    try:
+        with urllib.request.urlopen(req, timeout=10) as r:
+            return r.status, r.read().decode("utf-8", "replace")
+    except Exception as e:  # noqa: BLE001
+        return getattr(e, "code", 0), ""
+
+
+def pure(fn, *args, **kw):
+    """Call fn capturing stdout; return (result, captured_stdout)."""
+    buf = io.StringIO()
+    with contextlib.redirect_stdout(buf):
+        out = fn(*args, **kw)
+    return out, buf.getvalue()
+
+
+def main():
+    srv = subprocess.Popen([sys.executable, str(HERE / "mock_olrapi.py"), "--port", str(a.port)],
+                           stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    try:
+        for _ in range(50):
+            try:
+                urllib.request.urlopen(f"{BASE}/__debug__reset", data=b"", timeout=1)
+                break
+            except Exception:
+                time.sleep(0.1)
+        http("PUT", f"{BASE}/vault/_agent/echo-vault.md", "---\nschema_version: 4\n---\n# marker\n")
+        http("PUT", f"{BASE}/vault/_agent/context/current-context.md",
+             "---\ntype: context-bundle\nscope_updated: \"2026-06-20\"\ncreated: 2026-06-01\n---\n"
+             "# Current Context\n\n## Scope\ntesting phase 0\n\n## Scope History\n")
+
+        import echo
+        import echo_doctor
+        import echo_ops
+        import echo_recall
+        import echo_reflect
+        import echo_session
+        import echo_triage
+
+        # capture_op: create / duplicate-gate / dry-run — envelopes, stdout pure
+        env, out = pure(echo_ops.capture_op, "person", "Zara Quix", body_text="Met at the expo.")
+        check("capture_op create envelope", env.get("ok") is True and env.get("action") == "created"
+              and env.get("path") == "resources/people/zara-quix.md", json.dumps(env))
+        check("capture_op prints nothing to stdout", out == "", out[:200])
+        env, out = pure(echo_ops.capture_op, "person", "Zara Quix Junior", body_text="")
+        check("capture_op duplicate-gate is data, not an exit code",
+              env.get("ok") is False and env.get("action") == "duplicate-gate"
+              and env.get("candidates"), json.dumps(env))
+        check("capture_op gate prints nothing", out == "", out[:200])
+        env, out = pure(echo_ops.capture_op, "company", "Plan Co", body_text="", dry_run=True)
+        check("capture_op dry-run envelope", env.get("action") == "dry-run:create"
+              and env.get("path") == "resources/companies/plan-co.md", json.dumps(env))
+
+        # resolve_op / link_op
+        env, out = pure(echo_ops.resolve_op, "zara quix")
+        check("resolve_op returns the match dict", env.get("match") is True
+              and env.get("path") == "resources/people/zara-quix.md", json.dumps(env))
+        pure(echo_ops.capture_op, "concept", "Gizmo", body_text="")
+        env, out = pure(echo_ops.link_op, "resources/people/zara-quix.md", "resources/concepts/gizmo.md")
+        check("link_op envelope", env.get("ok") is True and env.get("a_changed") in (True, False),
+              json.dumps(env))
+        check("link_op prints nothing", out == "", out[:200])
+
+        # recall_op
+        env, out = pure(echo_recall.recall_op, "expo")
+        check("recall_op envelope shape", env.get("action") == "recall"
+              and isinstance(env.get("primary"), list) and isinstance(env.get("linked"), list),
+              json.dumps(env)[:200])
+        check("recall_op prints nothing", out == "", out[:200])
+
+        # triage list_op / route_op (dry-run)
+        http("PUT", f"{BASE}/vault/inbox/captures/inbox.md", "- 2026-06-10: try the quorlab tool\n")
+        env, out = pure(echo_triage.list_op)
+        check("triage list_op envelope", env.get("count") == 1
+              and env["items"][0]["age_days"] == 11, json.dumps(env))
+        props = [{"title": "Quorlab Tool", "kind": "reference", "confidence": 0.9,
+                  "line": "- 2026-06-10: try the quorlab tool"}]
+        env, out = pure(echo_triage.route_op, props, False)
+        check("triage route_op dry-run rows", env.get("dry_run") is True
+              and env["rows"][0]["action"] == "create", json.dumps(env))
+        check("triage route_op prints nothing", out == "", out[:200])
+
+        # reflect apply_op (applied path, using capture_op internally)
+        env, out = pure(echo_reflect.apply_op,
+                        [{"title": "Blorp Pattern", "kind": "semantic",
+                          "body": "The operator prefers blorp.", "confidence": 0.9}], True)
+        check("reflect apply_op applies via capture_op", env.get("applied") == 1
+              and env["results"][0]["action"] == "created", json.dumps(env))
+        check("reflect apply_op prints nothing", out == "", out[:200])
+
+        # scope ops
+        env, out = pure(echo.scope_show_op)
+        check("scope_show_op returns scope + freshness", env.get("scope") == "testing phase 0"
+              and env.get("scope_updated") == "2026-06-20", json.dumps(env))
+        env, out = pure(echo.scope_set_op, "phase zero refactor")
+        check("scope_set_op envelope", env.get("action") == "scope-set"
+              and env.get("scope_updated") == "2026-06-21", json.dumps(env))
+        check("scope_set_op prints nothing", out == "", out[:200])
+
+        # load_op (brief)
+        env, out = pure(echo.load_op, True)
+        check("load_op returns sections + brief", "marker" in env.get("sections", {})
+              and "ECHO load (brief)" in env.get("brief", ""), json.dumps(env)[:200])
+        check("load_op prints nothing", out == "", out[:200])
+
+        # doctor run_op
+        env, out = pure(echo_doctor.run_op)
+        check("doctor run_op checks list", isinstance(env.get("checks"), list)
+              and env.get("fatal") is None and env.get("ok") is True, json.dumps(env))
+        check("doctor run_op prints nothing", out == "", out[:200])
+
+        # session_end_op: dry-run envelope; bad bundle raises EchoError(2) pre-write
+        bundle = {"slug": "phase-zero", "log_body": "---\ntype: session-log\n---\n# S\n\n## Goal\nx\n"}
+        env, out = pure(echo_session.session_end_op, bundle, False)
+        check("session_end_op dry-run envelope", env.get("dry_run") is True
+              and env.get("path") == "_agent/sessions/2026-06-21-2300-phase-zero.md", json.dumps(env))
+        check("session_end_op prints nothing", out == "", out[:200])
+        try:
+            echo_session.session_end_op({"slug": "Bad Slug!", "log_body": "x"}, True)
+            check("session_end_op rejects a bad slug", False, "no exception raised")
+        except RuntimeError as exc:
+            check("session_end_op rejects a bad slug", getattr(exc, "code", 0) == 2, str(exc))
+
+        print(f"\n{len(failures)} failure(s)" if failures else "\nall ops-api tests passed")
+        return 1 if failures else 0
+    finally:
+        srv.terminate()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,417 @@
+#!/usr/bin/env python3
+"""echo-mcp — the containerized MCP server over the ECHO ops layer.  [2.2]
+
+Streamable-HTTP (stateless JSON) MCP server exposing the plugin's `*_op` cores
+(Phase 0, shipped 2.1.1) as typed tools. Runs next to the Obsidian REST API on
+the same box, so every vault round-trip behind a tool call is LAN-local.
+
+Spec: docs/MCP-SERVER-SPEC.md. Highlights implemented here:
+  * bearer-token auth on everything except /health (constant-time compare);
+  * tool results = the op envelopes, returned as structured JSON;
+  * duplicate gate / offline queueing are DATA (never protocol errors), so the
+    model deliberately chooses merge_into/force instead of blind-retrying;
+  * tool profiles: ECHO_MCP_TOOLS=core exposes only the six daily drivers
+    (schema tokens cost context on every surface that lists them);
+  * result budgets: echo_recall(budget_chars), echo_get_note(section/max_chars);
+  * per-write serialization (one process mediates all MCP writes, so a plain
+    lock removes the advisory-lock race window for server-mediated writes);
+  * startup validation: fail fast (crash-loop visibly) on missing env.
+
+Deliberately NOT here in v1 (spec §7.2 backlog): entity-index TTL cache — the
+atomic_index_update correctness fix depends on a FRESH re-read under the lock,
+so caching idx_mod.load would reintroduce the clobber race it closed. LAN
+adjacency + the warm keep-alive pool make the read ~ms anyway.
+
+Env: ECHO_BASE, ECHO_KEY, ECHO_MCP_TOKEN (required) · ECHO_OWNER ·
+     ECHO_MCP_PORT=8765 · ECHO_MCP_TOOLS=full|core · ECHO_STATE_DIR=/data
+"""
+from __future__ import annotations
+
+import contextlib
+import hmac
+import json
+import os
+import re
+import subprocess
+import sys
+import threading
+from pathlib import Path
+
+# --- locate the plugin scripts (image layout first, repo layout for local dev) ---
+_HERE = Path(__file__).resolve().parent
+for _cand in (_HERE / "scripts",
+              _HERE.parent / "echo-memory.plugin.src" / "skills" / "echo-memory" / "scripts"):
+    if _cand.is_dir():
+        SCRIPTS = _cand
+        break
+else:  # pragma: no cover
+    sys.exit("echo-mcp: cannot locate the ECHO scripts directory")
+sys.path.insert(0, str(SCRIPTS))
+
+# --- startup validation: a misdeployed container must crash-loop visibly ---------
+_missing = [k for k in ("ECHO_BASE", "ECHO_KEY", "ECHO_MCP_TOKEN") if not os.environ.get(k)]
+if _missing:
+    sys.exit(f"echo-mcp: missing required env: {', '.join(_missing)} "
+             "— check the PORT template / secret refs")
+
+# Modules aliased *_mod: several tool functions below share a module's name
+# (echo_recall, echo_reflect, ...) and would shadow it at module scope otherwise.
+import echo                        # noqa: E402
+import echo_doctor as doctor_mod   # noqa: E402
+import echo_index as idx_mod       # noqa: E402
+import echo_ops as ops_mod         # noqa: E402
+import echo_queue as queue_mod     # noqa: E402
+import echo_recall as recall_mod   # noqa: E402
+import echo_reflect as reflect_mod # noqa: E402
+import echo_session as session_mod # noqa: E402
+import echo_triage as triage_mod   # noqa: E402
+
+from mcp.server.fastmcp import FastMCP  # noqa: E402
+from starlette.requests import Request  # noqa: E402
+from starlette.responses import JSONResponse, Response  # noqa: E402
+from starlette.middleware.base import BaseHTTPMiddleware  # noqa: E402
+
+PORT = int(os.environ.get("ECHO_MCP_PORT", "8765"))
+TOKEN = os.environ["ECHO_MCP_TOKEN"]
+PROFILE = os.environ.get("ECHO_MCP_TOOLS", "full").strip().lower()
+KINDS = sorted(idx_mod.KIND_FOLDER)
+
+mcp = FastMCP(
+    "echo-mcp",
+    instructions=(
+        "Persistent memory over the operator's ECHO Obsidian vault. Use echo_load at "
+        "the start of substantive sessions, echo_recall/echo_resolve to read, "
+        "echo_capture as the default write, and echo_log_session to end a session. "
+        "A duplicate-gate result is a decision point, not an error — merge_into or "
+        "(confirmed-distinct) force. Write about the operator in third person."),
+    host="0.0.0.0",
+    port=PORT,
+    stateless_http=True,
+    json_response=True,
+)
+
+WRITE_LOCK = threading.Lock()   # all MCP-path writes serialize through this process
+
+
+def _guard(write: bool, fn, *args, **kw) -> dict:
+    """Run an op core with stdout redirected to stderr (belt-and-braces stream
+    purity) and EchoError mapped to an actionable error envelope, not a protocol
+    error. Write ops serialize on WRITE_LOCK."""
+    lock = WRITE_LOCK if write else contextlib.nullcontext()
+    try:
+        with lock, contextlib.redirect_stdout(sys.stderr):
+            return fn(*args, **kw)
+    except RuntimeError as exc:  # EchoError (either module instance) subclasses it
+        code = getattr(exc, "code", 1)
+        err = {"ok": False, "error": str(exc), "code": code}
+        if code == 78:
+            err["hint"] = ("server deployment is missing vault credentials — operator: "
+                           "check the PORT template env (SECRET: refs)")
+        elif code == 44:
+            err["code_name"] = "not-found"
+        elif "unreachable" in str(exc):
+            err["hint"] = ("vault unreachable (Obsidian/REST plugin likely down) — "
+                           "proceed without memory; writes queue durably")
+        return err
+
+
+_SAFE_PATH = re.compile(r"^[^/][^\0]*$")
+
+
+def _check_path(path: str) -> str | None:
+    if not path or path.startswith("/") or ".." in path.split("/") or not _SAFE_PATH.match(path):
+        return "invalid path: must be vault-relative, no leading '/' and no '..'"
+    return None
+
+
+def _frontmatter(text: str) -> dict:
+    out: dict = {}
+    if text.startswith("---"):
+        end = text.find("\n---", 3)
+        for ln in text[3:end if end != -1 else len(text)].splitlines():
+            m = re.match(r"^([A-Za-z_][\w-]*):\s*(.*)$", ln)
+            if m:
+                out[m.group(1)] = m.group(2).strip().strip('"').strip("'")
+    return out
+
+
+# ------------------------------------------------------------------ tools -------
+CORE_TOOLS = {"echo_load", "echo_recall", "echo_capture", "echo_triage_inbox",
+              "echo_log_session", "echo_health"}
+
+
+def tool(fn):
+    """Register `fn` as an MCP tool unless the core profile excludes it."""
+    if PROFILE == "core" and fn.__name__ not in CORE_TOOLS:
+        return fn
+    return mcp.tool()(fn)
+
+
+@tool
+def echo_load(brief: bool = True) -> dict:
+    """Cold-start memory orientation — call at the start of a substantive session.
+    brief=true (default) returns a token-budgeted digest plus structured sections;
+    brief=false returns the six raw sections. Also flushes writes queued offline."""
+    return _guard(True, echo.load_op, brief)
+
+
+@tool
+def echo_recall(query: str, limit: int = 6, budget_chars: int = 4000,
+                include_linked: bool = True) -> dict:
+    """Search memory: ranked matches for a topic/person/project PLUS their linked
+    neighbourhood. Excerpts are packed into budget_chars by score — call
+    echo_get_note for any hit's full content. Freshness/status-aware ranking."""
+    limit = max(1, min(int(limit), 20))
+    budget = max(500, min(int(budget_chars), 20000))
+    env = _guard(False, recall_mod.recall_op, query, limit)
+    if not env.get("ok"):
+        return env
+    if not include_linked:
+        env["linked"] = []
+    used = 0
+    for hit in (env.get("primary") or []) + (env.get("linked") or []):
+        ex = hit.get("excerpt") or ""
+        room = max(0, budget - used)
+        if len(ex) > room:
+            hit["excerpt"] = ex[:room] + ("… (truncated — echo_get_note for full)" if room else "")
+            hit["truncated"] = True
+        used += len(hit.get("excerpt") or "")
+    return env
+
+
+@tool
+def echo_resolve(mention: str) -> dict:
+    """Resolve a name/mention to its canonical vault note (alias-aware), or get
+    did-you-mean candidates. Call before creating any note by hand; echo_capture
+    does this automatically."""
+    return _guard(False, ops_mod.resolve_op, mention)
+
+
+@tool
+def echo_get_note(path: str, section: str = "", max_chars: int = 8000) -> dict:
+    """Read one vault note (vault-relative path). section='Status' returns only that
+    ## section. Content over max_chars is truncated with a marker."""
+    bad = _check_path(path)
+    if bad:
+        return {"ok": False, "error": bad}
+
+    def _read() -> dict:
+        status, body = echo.request("GET", echo.vault_url(path))
+        if status == 404:
+            return {"ok": False, "code_name": "not-found", "path": path,
+                    "error": f"{path}: not found"}
+        echo.check(status, body, f"get {path}")
+        text = body.decode(errors="replace")
+        fm = _frontmatter(text)
+        if section:
+            text = echo.extract_heading(text, section)
+            if not text:
+                return {"ok": False, "path": path,
+                        "error": f"no '## {section}' section in {path}"}
+        truncated = len(text) > max_chars
+        return {"ok": True, "path": path, "frontmatter": fm,
+                "content": text[:max_chars] + ("\n… (truncated)" if truncated else ""),
+                "truncated": truncated}
+    return _guard(False, _read)
+
+
+@tool
+def echo_get_scope() -> dict:
+    """The operator's active scope + freshness. If sessions_since >= 3, treat the
+    recorded scope as suspect and confirm with the operator before working."""
+    return _guard(False, echo.scope_show_op)
+
+
+@tool
+def echo_set_scope(scope: str) -> dict:
+    """Switch the active scope atomically (archives the prior scope to Scope History
+    and stamps freshness). Use when the session's work diverges from the recorded
+    scope."""
+    return _guard(True, echo.scope_set_op, scope)
+
+
+@tool
+def echo_health(deep: bool = False) -> dict:
+    """ECHO readiness: config, vault reachability, auth, bootstrap/schema, and the
+    offline-queue depth. deep=true also runs the full vault-invariant linter."""
+    env = _guard(False, doctor_mod.run_op)
+    try:
+        env["outbox_depth"] = len(queue_mod.pending())
+        env["needs_attention"] = len(queue_mod.needs_attention())
+    except Exception:  # noqa: BLE001
+        pass
+    if deep and env.get("ok"):
+        r = subprocess.run([sys.executable, str(SCRIPTS / "vault_lint.py")],
+                           capture_output=True, text=True,
+                           env=dict(os.environ), timeout=120)
+        env["lint_exit"] = r.returncode
+        env["lint_report"] = (r.stdout or r.stderr)[-6000:]
+    return env
+
+
+@tool
+def echo_capture(title: str, kind: str = "", body: str = "", tags: list[str] = [],
+                 aliases: list[str] = [], sources: list[str] = [], status: str = "",
+                 date: str = "", domain: str = "business", merge_into: str = "",
+                 force: bool = False, dry_run: bool = False) -> dict:
+    """The default memory write: routes by kind (person, company, concept, reference,
+    meeting, project, area, semantic, episodic, working, skill, decision), stamps
+    complete frontmatter, indexes, auto-links, and writes the Agent-Log line — one
+    call. Omit kind for an inbox capture. If the result is action='duplicate-gate',
+    do NOT retry blindly: call again with merge_into=<candidate slug> to update the
+    existing entity, or force=true only after confirming they are genuinely distinct.
+    Offline, the whole capture queues durably (queued=true)."""
+    if kind and kind not in KINDS:
+        return {"ok": False, "error": f"unknown kind '{kind}' — one of: {', '.join(KINDS)}"}
+    return _guard(True, ops_mod.capture_op, kind or None, title,
+                  body_text=body or "", status_v=status, aliases=aliases,
+                  sources=sources, tags=tags, date=date or None, domain=domain,
+                  inbox=not kind, dry_run=dry_run, force=force,
+                  merge_into=merge_into or None)
+
+
+@tool
+def echo_link(a: str, b: str) -> dict:
+    """Add reciprocal '## Related' links between two notes. Accepts vault paths or
+    resolvable entity names (aliases work)."""
+    def _to_path(x: str) -> str | None:
+        if "/" in x:
+            return x if x.endswith(".md") else x + ".md"
+        nmap = idx_mod.name_map(idx_mod.load())
+        return nmap.get(idx_mod.slugify(x))
+    def _do() -> dict:
+        pa, pb = _to_path(a), _to_path(b)
+        if not pa or not pb:
+            missing = a if not pa else b
+            return {"ok": False, "error": f"'{missing}' resolves to no note — "
+                                          "pass a vault path or a known entity name"}
+        return ops_mod.link_op(pa, pb)
+    return _guard(True, _do)
+
+
+@tool
+def echo_append_note(path: str, line: str) -> dict:
+    """Append one line to a note, idempotently (skipped if the exact line already
+    exists). For inbox lines, Agent-Log entries, Observations bullets."""
+    bad = _check_path(path)
+    if bad:
+        return {"ok": False, "error": bad}
+    def _do() -> dict:
+        rc = echo.cmd_append(path, line)
+        return {"ok": rc == 0, "action": "append", "path": path}
+    return _guard(True, _do)
+
+
+@tool
+def echo_patch_note(path: str, operation: str, target_type: str, target: str,
+                    content: str) -> dict:
+    """Targeted edit: append/prepend/replace under a heading, frontmatter field, or
+    block. HEADING TARGETS must be the full '::'-delimited path from the top-level
+    heading (e.g. 'Operator Preferences::Fact / Pattern') — on an invalid target the
+    error includes the note's actual headings. replace overwrites the section."""
+    bad = _check_path(path)
+    if bad:
+        return {"ok": False, "error": bad}
+    def _do() -> dict:
+        rc = echo.cmd_patch(path, operation, target_type, target,
+                            echo.temp_file(content.encode("utf-8")))
+        return {"ok": rc == 0, "action": f"patch:{operation}", "path": path,
+                "target": target}
+    env = _guard(True, _do)
+    if not env.get("ok") and "HTTP 400" in str(env.get("error", "")) and target_type == "heading":
+        st, body = echo.request("GET", echo.vault_url(path),
+                                headers={"Accept": "application/vnd.olrapi.document-map+json"})
+        if st == 200:
+            try:
+                headings = [h.get("heading") or h for h in
+                            json.loads(body).get("headings", [])][:40]
+                env["available_headings"] = headings
+                env["hint"] = "use one of available_headings verbatim as the Target"
+            except Exception:  # noqa: BLE001
+                pass
+    return env
+
+
+@tool
+def echo_triage_inbox(proposals: list[dict] = [], apply: bool = False) -> dict:
+    """Inbox triage. No proposals => structured listing of captures (line, date, text,
+    age_days). With proposals (reflect schema + optional 'line'): apply=false previews
+    the routing; apply=true routes via capture and writes the processing-log audit.
+    List first, propose, preview, then apply only after the operator confirms."""
+    if not proposals:
+        return _guard(False, triage_mod.list_op)
+    return _guard(True, triage_mod.route_op, proposals, apply)
+
+
+@tool
+def echo_reflect(proposals: list[dict], apply: bool = False) -> dict:
+    """Session-reflection proposals: validate -> classify against the entity index ->
+    preview (apply=false) -> apply (routes each via capture). Never apply without the
+    operator's go-ahead; never invent memories to have something to save."""
+    return _guard(True, reflect_mod.apply_op, proposals, apply)
+
+
+@tool
+def echo_log_session(slug: str, log_body: str, agent_log_line: str = "",
+                     scope: str = "", reflect: list[dict] = [],
+                     apply: bool = False, hhmm: str = "") -> dict:
+    """End a substantive session in ONE call: session log -> Agent-Log line -> reflect
+    proposals -> optional scope switch -> heartbeat LAST (the commit marker).
+    apply=false previews the plan. Pass hhmm (local time, e.g. '1430') so the log
+    filename sorts truthfully."""
+    bundle = {"slug": slug, "log_body": log_body}
+    if agent_log_line:
+        bundle["agent_log_line"] = agent_log_line
+    if scope:
+        bundle["scope"] = scope
+    if reflect:
+        bundle["reflect"] = reflect
+    def _do() -> dict:
+        prev = os.environ.get("ECHO_NOW")
+        try:
+            if hhmm:
+                os.environ["ECHO_NOW"] = hhmm
+            return session_mod.session_end_op(bundle, apply=apply)
+        finally:
+            if hhmm:
+                if prev is None:
+                    os.environ.pop("ECHO_NOW", None)
+                else:
+                    os.environ["ECHO_NOW"] = prev
+    return _guard(True, _do)
+
+
+# ------------------------------------------------------------- health + auth ----
+@mcp.custom_route("/health", methods=["GET"])
+async def health(_request: Request) -> JSONResponse:
+    """Unauthenticated liveness for Docker HEALTHCHECK + Kuma. No vault data."""
+    st, _ = echo.request("GET", echo.vault_url("_agent/echo-vault.md"))
+    try:
+        outbox = len(queue_mod.pending())
+    except Exception:  # noqa: BLE001
+        outbox = -1
+    return JSONResponse({"ok": True, "vault_reachable": st != 0,
+                         "outbox_depth": outbox, "profile": PROFILE})
+
+
+class BearerAuth(BaseHTTPMiddleware):
+    async def dispatch(self, request, call_next):
+        if request.url.path == "/health":
+            return await call_next(request)
+        auth = request.headers.get("authorization", "")
+        if not hmac.compare_digest(auth, f"Bearer {TOKEN}"):
+            return Response(status_code=401)
+        return await call_next(request)
+
+
+def main() -> None:
+    import uvicorn
+    app = mcp.streamable_http_app()
+    app.add_middleware(BearerAuth)
+    print(f"echo-mcp: serving on :{PORT} (profile={PROFILE}, vault={echo.BASE})",
+          file=sys.stderr)
+    uvicorn.run(app, host="0.0.0.0", port=PORT, log_level="info")
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,4 @@
+# echo-mcp server deps. The PLUGIN stays pure-stdlib; the dependency budget for the
+# container is deliberately tiny: the official MCP SDK (brings starlette/uvicorn/
+# pydantic) and nothing else.
+mcp>=1.9,<2