2.1.1 — MCP Phase 0: every high-level op returns an envelope (return-not-print)

Internal refactor per docs/MCP-SERVER-SPEC.md §4; CLI output and exit codes
unchanged (wrappers reproduce them from the envelopes):

- echo_ops: capture_op/resolve_op/link_op — duplicate gate and offline queueing
  are data (action: duplicate-gate + candidates; queued: true); the --json
  stdout-swallow hack is deleted (helper chatter -> stderr inside the cores);
  capture_op takes body_text= directly (the MCP path, no temp file).
- echo_recall.recall_op — one structured result backs prose and --json.
- echo_triage.list_op/route_op, echo_reflect.apply_op,
  echo_session.session_end_op — call capture_op internally, count gate outcomes
  from envelopes.
- echo.scope_show_op/scope_set_op/load_op (+ shared _load_gather), and
  echo_doctor.run_op (checks as data).

New 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 six
suites green. The 2.2 server build now starts directly at the app layer.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

CHANGELOG.md

@@ -1,5 +1,91 @@
 # Changelog
 
+## 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,
+#3, #7). Three independently useful changes; no schema or layout changes.
+
+### Added — `load --brief`: a token-budgeted cold-start digest
+
+The SessionStart hook injected the full text of six files every session (measured
+15.9KB and growing without bound — Observations and Scope History never shrink).
+`load --brief` renders a digest instead: Fact / Pattern rules in full, the last ~10
+Observations (with a `+N older` note), scope + `scope_updated` + sessions-since,
+the heartbeat-pointed session log's key sections (Goal / Decisions Made / Open
+Threads / Suggested Next Step), today's Agent Log lines, and the inbox as a
+**count + oldest age** — everything the load-time reconcile needs. Over
+`ECHO_LOAD_BUDGET` (default ~8000 chars), sections trim lowest-priority-first
+(observations → agent log → session summary) with explicit `(truncated)` markers.
+The hook now injects the brief form; `/echo-load` and the manual procedure keep the
+full dump. All `load` reads (plus the sessions listing, which also feeds the
+heartbeat-absent fallback) are fetched **in parallel** over the pooled connections.
+
+### Fixed — capture is now offline-durable (the write path's inversion bug)
+
+Low-level verbs queued on an outage, but the *default* write (`capture`) died: the
+update path's POST/PATCH and `ensure_daily_log` bypassed the queue entirely, and an
+unreachable index aborted the whole op. Now: (1) a fully-offline `capture` queues
+the **whole operation as one semantic record** (title/kind/body/tags/…, plus the
+capture-time date); `flush` replays it **through `capture`** so create-vs-update,
+the duplicate gate, and aliasing re-run against the index as it is at replay time —
+byte-level replay would freeze a decision made blind. (2) A duplicate-gate stop on
+replay keeps the record, flags it `needs_attention` (surfaced by `flush` and
+`load`), and does NOT block later records. (3) Mid-capture outages: the update
+path and `ensure_daily_log` writes ride `safe_request` (accepted edge: a queued
+agent-log PATCH 400-drops loudly if the daily note never existed — the trace line,
+not the memory). Same-args offline captures dedupe by idem-key.
+
+### Added — `session-end`: the one-call session close
+
+`echo.py session-end <bundle.json> [--apply]` performs the whole end-of-session
+sequence under ONE lock acquisition, in order: session log PUT → Agent Log line →
+reflect proposals (via `capture`, gate-aware) → optional `scope set` → **heartbeat
+LAST as the commit marker** (a part-way failure leaves the previous pointer intact,
+never dangling). Dry-run by default previews the full plan including the reflect
+classification. Filename HHMM from `$ECHO_NOW` (else local clock), validated
+against the canonical session-log regex **before any write**. The Stop hook's nudge
+now names the command, and recognizes a `session-end` invocation as
+"already reflected".
+
+### Fixed — helper-module errors exited 2 instead of their intended codes
+
+When `echo.py` runs as `__main__`, helper modules' `import echo` creates a twin
+module whose `EchoError` is a different class object, so `main()`'s handler missed
+it and the raw traceback leaked. The handler now catches `RuntimeError` (both
+classes subclass it) and honors `.code`.
+
+Tests: +19 end-to-end checks (brief-load digest/budget, offline-capture queue /
+replay-through-capture / gate-on-replay / idem-key dedup, session-end dry-run /
+apply / ordering / validation-abort). All suites green.
+
 ## 2.0.0
 
 **Packaging & structure major — nothing about memory behavior changes.** The major

README.md

@@ -1,4 +1,4 @@
-# echo-memory — v2.0.0
+# echo-memory — v2.1.1
 
 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.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`. |
 | **1.5.1** | **Duplicate-gate precision.** Live use caught the gate blocking a decision note because it shared the single vault-common token "echo" with an archived project (min-normalized scoring gives any single-token entity a 1.0 match). New `gate_candidates()` blocks only on **same-kind** candidates (cross-kind name collisions — a decision titled after its project — warn instead), and a **lone shared token blocks only when unique to that entity** (vault df == 1); multi-token overlaps still block. Advisory warnings (`fuzzy_candidates`) unchanged; exit-76/`--merge-into`/`--force` unchanged. +5 tests; verified against the live vault both ways (false positive creates cleanly, true lookalike still gates). |

ROADMAP-2.0.md

@@ -35,8 +35,9 @@ coexisting (see `TODO-1.6.md`). 2.0 finishes it:
 
 - [x] **Delete the legacy `commands/` directory** — **done 2.0.0** (1.6.0's install
       verification on both surfaces cleared the gate; skills had precedence anyway).
-- [ ] Re-verify desktop + CoWork installs of the skills-only artifact (operator step,
+- [x] Re-verify desktop + CoWork installs of the skills-only artifact — **verified
-      post-2.0.0-install).
+      2026-07-28** on the baked 2.0.0 build: installs cleanly, all eight skills
+      register and work with no legacy `commands/` present.
 - [x] Update SKILL.md / README / command docs that reference `commands/` — **done 2.0.0**.
 
 ## 3. Codex packaging (from MAINTENANCE › Canonical source tree)

docs/IMPROVEMENT-PLANS.md

@@ -345,8 +345,9 @@ manifest-verification pass).
 | Release | Items | Rationale |
 |---|---|---|
 | **1.6.0** | TODO-1.6: skills-format migration (+ per-skill `allowed-tools`, `disable-model-invocation` on write-heavy entries, `$ECHO` block dedupe) · lint blind-spot fix (retired-dir/leaf-README) · plugin.json `homepage` | Restructures the command/SKILL files later work edits; allowed-tools removes prompt friction for every verb added below. Vault link repointing (TODO-1.6 §3) is vault data, not plugin code — any session. |
-| **1.6.x quick wins** | #1 brief load · #3 offline capture · #7 session-end (+ return-not-print refactor, MCP spec Phase 0, if it fits) | Small, correctness/efficiency, no schema changes. #7 unblocks the MCP tool. Users get these without a reinstall — don't gate them behind 2.0. |
+| **1.6.0 — SHIPPED 2026-07-28** | Skills migration + lint fix + homepage | Verified on desktop + CoWork. |
-| **2.0** | ROADMAP-2.0 unchanged: delete legacy `commands/` · artifact policy (Gitea releases, gitignore, tags) · Codex note · README layout | Packaging only, no feature riders — a bad feature must never force reverting a packaging break. |
+| **2.0.0 — SHIPPED 2026-07-28** | ROADMAP-2.0: commands/ deleted · release-based artifacts · gitignore · Codex note | Packaging-only major; skills-only install re-verified. |
-| **2.1** | MCP server (containerized, `echomcp.alwisp.com`) — `docs/MCP-SERVER-SPEC.md` | Needs #7 + Phase 0. Lands after 2.0 so its docs target one command format and the Gitea release channel exists. Deployed via CI + PORT, not the plugin manifest. |
+| **2.1.0 quick wins — SHIPPED 2026-07-28** | #1 brief load · #3 offline capture · #7 session-end | All three landed with +19 e2e checks; see CHANGELOG 2.1.0. The MCP spec's Phase 0 (return-not-print) did NOT ride along — it's the first phase of the MCP build session. |
-| **2.2 index train** | #2 local-first recall (CLI-path half) · #4 incremental sweep · #8 stemming+expansion | One schema-bump train; all touch the same index/sweep loop. The container already keeps its index warm in memory (spec §7), so #2 here covers the CLI fallback path; #4's fast sweep also becomes the container's background job. |
+| **2.2** | MCP server (containerized, `echomcp.alwisp.com`) — `docs/MCP-SERVER-SPEC.md` | Needs Phase 0 (return-not-print refactor, first step of the build session). #7 session-end shipped, so `echo_log_session` wraps a real verb. Deployed via CI + PORT, not the plugin manifest. |
-| **2.3 memory train** | #5 lifecycle/decay · #6 supersession · #9 resolve-miss learning | Capability tier; #6 after #5 (status vocabulary). |
+| **2.3 index train** | #2 local-first recall (CLI-path half) · #4 incremental sweep · #8 stemming+expansion | One schema-bump train; all touch the same index/sweep loop. The container already keeps its index warm in memory (spec §7), so #2 here covers the CLI fallback path; #4's fast sweep also becomes the container's background job. |
+| **2.4 memory train** | #5 lifecycle/decay · #6 supersession · #9 resolve-miss learning | Capability tier; #6 after #5 (status vocabulary). |

docs/MCP-SERVER-SPEC.md

@@ -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.1** (after the 1.6.x train and the 2.0 packaging major) | Needs #7 + Phase 0; docs then target the single post-2.0 command format. |
+| 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`. |

echo-memory.plugin.src/.claude-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "echo-memory",
-  "version": "2.0.0",
+  "version": "2.1.1",
   "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.",

echo-memory.plugin.src/skills/echo-memory/SKILL.md

@@ -140,7 +140,7 @@ Load at the start of any substantive conversation — anything beyond a single q
 
 ### Loading procedure
 
-**Run `python3 "$ECHO" load`** — one call that performs the six orientation reads below and prints each labeled section (404s on today's note / inbox show as absent, not errors; an absent marker is flagged). This replaces issuing six separate GETs by hand. The table documents what `load` reads and why:
+**Run `python3 "$ECHO" load`** — one call that performs the six orientation reads below (fetched in parallel) and prints each labeled section (404s on today's note / inbox show as absent, not errors; an absent marker is flagged). This replaces issuing six separate GETs by hand. **`load --brief`** renders a token-budgeted digest instead — Fact/Pattern rules in full, the last ~10 Observations, scope + freshness, the last session's key sections, today's Agent Log lines, and the inbox as a *count* (`ECHO_LOAD_BUDGET`, default ~8000 chars, trims lowest-priority sections first). The SessionStart hook injects the brief form; `/echo-load` and this procedure use the full form. The table documents what `load` reads and why:
 
 | # | GET | Notes |
 |---|-----|-------|
@@ -405,25 +405,40 @@ At the end of substantive conversations (ones that produced decisions, artifacts
 
 **Filename format is canonical: `_agent/sessions/YYYY-MM-DD-HHMM-<slug>.md`.** Always include the four-digit local-time HHMM component — it makes filenames lexically sort in true chronological order, which is what Step 3 of loading relies on. Older session logs without the HHMM part exist; leave them alone, but every new one must use the full form.
 
-Write the body (see `references/session-log-template.md`) to a file with the Write tool, then:
+**Preferred — one call (`session-end`).** Write a bundle JSON with the Write tool and run it; it performs the whole session-end sequence under one lock, in order, with the **heartbeat written last as the commit marker** (a failure part-way leaves the previous pointer intact):
+
+```json
+{ "slug": "<kebab-slug>",
+  "log_body": "<full session-log markdown, frontmatter included — see references/session-log-template.md>",
+  "agent_log_line": "- <currentDate>: <one-line summary>",
+  "scope": "<new scope text — omit to leave unchanged>",
+  "reflect": [ { "title": "…", "kind": "…", "body": "…", "confidence": 0.9 } ] }
+```
+
+```bash
+python3 "$ECHO" session-end bundle.json            # dry-run: previews the full plan
+ECHO_NOW=<HHMM> python3 "$ECHO" session-end bundle.json --apply
+```
+
+`--apply` writes: session log → Agent Log line → reflect proposals (via `capture`, gate-aware) → optional scope switch → heartbeat. Set `ECHO_NOW` to the conversation's local HHMM so the filename sorts truthfully. Reflect proposals still follow the reflect contract — preview with the dry-run and apply only with the operator's go-ahead. Offline, every step queues durably.
+
+**Manual fallback** (only if `session-end` is unavailable) — the same sequence by hand:
 
 ```bash
 python3 "$ECHO" put _agent/sessions/<currentDate>-HHMM-<slug>.md <bodyfile>
 ```
 
-Then add a one-line entry to today's daily note via the **Daily Note — Agent Log** procedure above.
+Then add a one-line entry to today's daily note via the **Daily Note — Agent Log** procedure above, and finally update the heartbeat pointer so the next session can orient in one read (a pointer nobody writes is a pointer nobody can read) — a single line, `<session-log-path> @ <ISO-timestamp>`:
-
-Finally, update the heartbeat pointer so the next session can orient in one read (this closes the loop with loading-procedure Step 4 — a pointer nobody writes is a pointer nobody can read). It is a single line, `<session-log-path> @ <ISO-timestamp>`; write it to a file and PUT it:
 
 ```bash
 python3 "$ECHO" put _agent/heartbeat/last-session.md <bodyfile>
 ```
 
-`last-session.md` is overwritten (PUT) each session end — never appended, so it can't grow or duplicate.
+`last-session.md` is overwritten (PUT) each session end — never appended, so it can't grow or duplicate. Keep the heartbeat write **last** in the manual sequence too.
 
 ## Vault Unreachable
 
-If the API returns a connection error, timeout, or `502`, tell the operator once that the memory vault is unreachable (a `502` usually means Obsidian/the REST plugin is not running on the backend), then proceed without memory. Do not retry repeatedly.
+If the API returns a connection error, timeout, or `502`, tell the operator once that the memory vault is unreachable (a `502` usually means Obsidian/the REST plugin is not running on the backend), then proceed — reads degrade to the last-known-good cache at load, and **writes queue durably**: the low-level verbs queue the request, and `capture` queues the whole operation as one semantic record that replays *through capture* (routing + duplicate gate re-run against the index at replay time) on the next reachable session. A queued capture that stops at the duplicate gate on replay is kept and flagged (`flush` lists it) rather than landed blind. Do not retry repeatedly.
 
 ## Style Rules
 

echo-memory.plugin.src/skills/echo-memory/scripts/echo.py

@@ -549,13 +549,12 @@ 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")
-
-    if subcommand == "show":
     scope_text = extract_heading(current, "Scope")
     scope_updated = ""
     for line in current.splitlines():
@@ -572,23 +571,22 @@ def cmd_scope(subcommand: str, text: str | None = None, as_json: bool = False) -
                 sessions_since = sum(1 for f in files if f.endswith(".md") and f[:10] > scope_updated)
             except json.JSONDecodeError:
                 pass
-        if as_json:
+    return {"ok": True, "action": "scope-show", "scope": scope_text,
-            print(json.dumps({"ok": True, "action": "scope-show", "scope": scope_text,
+            "scope_updated": scope_updated or None, "sessions_since": sessions_since}
-                              "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)"
+    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",
@@ -601,13 +599,242 @@ def cmd_scope(subcommand: str, text: str | None = None, as_json: bool = False) -
                 "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()})")
+    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
 
 
-def cmd_load() -> int:
+def _fetch_statuses(paths: list[str]) -> dict[str, tuple[int, bytes]]:
-    """Cold-start orientation: the canonical 6 reads in one call. 404s on today's
+    """Concurrently GET vault paths keeping (status, body) per path — unlike read_many,
-    daily note and the inbox are normal (printed as absent, not errors)."""
+    the 404-vs-offline distinction survives, which cmd_load's cache logic needs."""
+    if not paths:
+        return {}
+    workers = min(MAX_WORKERS, len(paths))
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        return dict(zip(paths, pool.map(lambda p: request("GET", vault_url(p)), paths)))
+
+
+def _bullet_lines(text: str) -> list[str]:
+    return [ln for ln in text.splitlines() if ln.lstrip().startswith("- ")]
+
+
+def _fm_line(text: str, field: str) -> str:
+    for ln in text.splitlines():
+        if ln.startswith(f"{field}:"):
+            return ln.split(":", 1)[1].strip().strip('"').strip("'")
+    return ""
+
+
+def _render_brief(texts: dict[str, str | None], listing_files: list[str],
+                  offline: bool) -> str:
+    """The token-budgeted cold-start digest (2.1.0). Selection over compression: Fact /
+    Pattern rules in full, recent Observations only, scope + freshness (not the whole
+    history), the last session's key sections, today's Agent Log lines, inbox COUNT.
+    Sections are trimmed lowest-priority-first when over ECHO_LOAD_BUDGET."""
+    budget = int(os.environ.get("ECHO_LOAD_BUDGET", "8000"))
+    S: list[tuple[str, str]] = []  # (section-name, text) in display order
+
+    marker = texts.get("marker")
+    if marker is None:
+        S.append(("marker", "marker: _agent/echo-vault.md ABSENT — vault not bootstrapped "
+                            "(run bootstrap.py before relying on memory)"))
+    else:
+        ver = _fm_line(marker, "schema_version") or "?"
+        S.append(("marker", f"marker: _agent/echo-vault.md — bootstrapped, schema_version {ver}"))
+
+    ctx = texts.get("context")
+    if ctx:
+        scope = extract_heading(ctx, "Scope") or "(empty)"
+        upd = _fm_line(ctx, "scope_updated") or "unknown"
+        since = sum(1 for f in listing_files if f.endswith(".md") and f[:10] > upd) \
+            if upd != "unknown" else None
+        head = f"scope (updated {upd}"
+        if since is not None:
+            head += f", {since} session(s) logged since"
+        S.append(("scope", head + "):\n" + scope))
+    else:
+        S.append(("scope", "scope: current-context.md absent"))
+
+    prefs = texts.get("preferences")
+    if prefs:
+        fact = extract_heading(prefs, "Fact / Pattern")
+        if fact:
+            S.append(("facts", "preferences — Fact / Pattern:\n" + fact))
+        obs = _bullet_lines(extract_heading(prefs, "Observations"))
+        if obs:
+            shown = obs[-10:]
+            head = f"preferences — Observations (last {len(shown)} of {len(obs)}):"
+            S.append(("observations", head + "\n" + "\n".join(shown)))
+
+    hb = texts.get("heartbeat")
+    if hb and hb.strip():
+        pointer = hb.strip().splitlines()[0]
+        sess_path = pointer.split(" @ ", 1)[0].strip()
+        part = [f"last session: {pointer}"]
+        log_text = texts.get("_pointed_session")
+        if log_text:
+            for h in ("Goal", "Decisions Made", "Open Threads", "Suggested Next Step"):
+                sec = extract_heading(log_text, h)
+                if sec and sec.strip("() \n"):
+                    part.append(f"  {h}: " + " / ".join(ln.strip() for ln in sec.splitlines()
+                                                        if ln.strip())[:400])
+        elif sess_path:
+            part.append("  (session log not readable — see the path above)")
+        S.append(("session", "\n".join(part)))
+    elif listing_files:
+        recent = sorted((f for f in listing_files if f.endswith(".md")), reverse=True)[:3]
+        S.append(("session", "last session: heartbeat absent — recent logs:\n"
+                  + "\n".join(f"  _agent/sessions/{f}" for f in recent)))
+
+    today_note = texts.get("today")
+    if today_note:
+        log_lines = _bullet_lines(extract_heading(today_note, "Agent Log"))
+        S.append(("agent-log", "today's Agent Log:\n" + ("\n".join(log_lines) or "  (empty)")))
+    else:
+        S.append(("agent-log", "today's Agent Log: (no daily note yet)"))
+
+    inbox = texts.get("inbox")
+    if inbox:
+        items = re.findall(r"(?m)^\s*-\s*(\d{4}-\d{2}-\d{2})", inbox)
+        if items:
+            try:
+                oldest = (dt.date.fromisoformat(today()) - dt.date.fromisoformat(min(items))).days
+                S.append(("inbox", f"inbox: {len(items)} capture(s), oldest {oldest}d — "
+                                   "offer triage if any are older than ~7 days"))
+            except ValueError:
+                S.append(("inbox", f"inbox: {len(items)} capture(s)"))
+        else:
+            S.append(("inbox", "inbox: empty"))
+    else:
+        S.append(("inbox", "inbox: empty"))
+
+    def total() -> int:
+        return sum(len(t) + 2 for _, t in S)
+
+    # Over budget -> trim lowest-priority sections first, with explicit markers.
+    for name, keep in (("observations", 4), ("agent-log", 4), ("session", 9)):
+        if total() <= budget:
+            break
+        for i, (n, t) in enumerate(S):
+            if n == name:
+                lines = t.splitlines()
+                if len(lines) > keep:
+                    S[i] = (n, "\n".join(lines[:keep]) + "\n  (truncated — /echo-load for full)")
+    out = f"ECHO load (brief) — {today()}\n\n" + "\n\n".join(t for _, t in S)
+    if len(out) > budget:
+        out = out[:budget] + "\n(truncated — /echo-load for full)"
+    if offline:
+        out += ("\n\nNOTE: vault unreachable — context above is last-known-good cache; "
+                "writes this session will be queued and synced when the vault returns.")
+    return out
+
+
+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"),
+]
+
+
+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 = heartbeat_absent = offline = False
+    texts: dict[str, str | None] = {}
+    for label, path in targets:
+        status, body = results[path]
+        if status == 200:
+            echo_queue.cache_put(path, body)  # refresh last-known-good for offline fallback
+            texts[label] = body.decode(errors="replace")
+        elif status == 0:  # vault unreachable -> degrade to last-known-good cache
+            offline = True
+            cached = echo_queue.cache_get(path)
+            texts[label] = cached.decode(errors="replace") if cached is not None else None
+        else:
+            texts[label] = None
+            if status == 404:
+                if label == "marker":
+                    marker_missing = True
+                if label == "heartbeat":
+                    heartbeat_absent = True
+    lst_status, lst_body = results[listing_path]
+    listing_files: list[str] = []
+    if lst_status == 200:
+        try:
+            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()}")
@@ -617,49 +844,47 @@ def cmd_load() -> int:
         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 = [
+    targets = LOAD_TARGETS + [("today", f"journal/daily/{today()}.md"),
-        ("marker",      "_agent/echo-vault.md"),
+                              ("inbox", "inbox/captures/inbox.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)
 
-    marker_missing = False
+    results, texts, listing_files, offline, marker_missing, heartbeat_absent = \
-    heartbeat_absent = False
+        _load_gather(targets)
-    offline = False
+
+    if brief:
+        _fetch_pointed_session(texts)
+        print(_render_brief(texts, listing_files, offline))
+        return 0
+
+    # ---- full mode: raw sections, output format unchanged ----
     for label, path in targets:
-        status, body = request("GET", vault_url(path))
+        status, body = results[path]
         if status == 200:
-            echo_queue.cache_put(path, body)  # refresh last-known-good for offline fallback
             print(f"===== {label}: {path} (HTTP 200) =====")
-            text = body.decode(errors="replace")
+            text = texts[label] or ""
             sys.stdout.write(text)
             if not text.endswith("\n"):
                 sys.stdout.write("\n")
         elif status == 404:
             print(f"===== {label}: {path} (HTTP 404) =====")
             print("(absent — fine)")
-            if label == "marker":
+        elif status == 0:
-                marker_missing = True
+            if texts[label] is not None:
-            if label == "heartbeat":
-                heartbeat_absent = True
-        elif status == 0:  # vault unreachable -> degrade to last-known-good cache
-            offline = True
-            cached = echo_queue.cache_get(path)
-            if cached is not None:
                 print(f"===== {label}: {path} (OFFLINE — serving stale cache) =====")
-                sys.stdout.write(cached.decode(errors="replace"))
+                sys.stdout.write(texts[label])
-                if not cached.endswith(b"\n"):
+                if not texts[label].endswith("\n"):
                     sys.stdout.write("\n")
             else:
                 print(f"===== {label}: {path} (OFFLINE — no cache) =====")
@@ -669,15 +894,9 @@ def cmd_load() -> int:
             print(f"(error HTTP {status}: {body.decode(errors='replace')[:200]})")
         print()
     # M3: heartbeat pointer missing/stale -> fall back to the recent sessions listing
-    # (matches the documented loading procedure) so orientation works without the pointer.
+    # (already fetched in the batch) so orientation works without the pointer.
-    if heartbeat_absent and not offline:
+    if heartbeat_absent and not offline and listing_files:
-        st, body = request("GET", vault_url("_agent/sessions/"))
+        files = sorted((f for f in listing_files if f.endswith(".md")), reverse=True)
-        if st == 200:
-            try:
-                files = sorted((f for f in json.loads(body).get("files", []) if f.endswith(".md")),
-                               reverse=True)
-            except json.JSONDecodeError:
-                files = []
         if files:
             print("===== recent sessions (heartbeat absent — fallback) =====")
             for f in files[:5]:
@@ -732,7 +951,8 @@ def cmd_config(args) -> int:
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="Validated cross-platform ECHO vault client")
     sub = parser.add_subparsers(dest="cmd", required=True)
-    sub.add_parser("load")
+    p = sub.add_parser("load")
+    p.add_argument("--brief", action="store_true")  # token-budgeted digest (hook default)
     sub.add_parser("flush")  # H2: replay writes queued during a prior outage
     sub.add_parser("doctor")  # M3: one-call readiness check
     sub.add_parser("get").add_argument("path")
@@ -761,6 +981,9 @@ def build_parser() -> argparse.ArgumentParser:
     p = sub.add_parser("reflect")  # H5: apply a JSON proposal set (dry-run unless --apply)
     p.add_argument("file", nargs="?")
     p.add_argument("--apply", action="store_true")
+    p = sub.add_parser("session-end")  # 2.1.0: one-call session end (log+agent-log+reflect+scope+heartbeat)
+    p.add_argument("file", nargs="?")                    # bundle JSON; - or stdin
+    p.add_argument("--apply", action="store_true")
     p = sub.add_parser("triage")   # one-tap inbox triage (reflect pipeline + audit log)
     p.add_argument("file", nargs="?")                    # proposals JSON; omit to list
     p.add_argument("--list", action="store_true", dest="list_inbox")
@@ -796,21 +1019,27 @@ def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
     try:
         if args.cmd == "load":
-            return cmd_load()
+            return cmd_load(brief=args.brief)
         if args.cmd == "doctor":
             import echo_doctor
             return echo_doctor.run()
         if args.cmd == "flush":
             import echo_queue
             n = echo_queue.flush()
-            remaining = len(echo_queue.pending())
+            remaining = echo_queue.pending()
+            flagged = [r for r in remaining if r.get("needs_attention")]
             if n:
                 print(f"ok: flushed {n} queued write(s)"
-                      + (f"; {remaining} still queued (vault unreachable)" if remaining else ""))
+                      + (f"; {len(remaining)} still queued" if remaining else ""))
             elif remaining:
-                print(f"ok: {remaining} write(s) still queued (vault unreachable)")
+                print(f"ok: {len(remaining)} write(s) still queued")
             else:
                 print("ok: queue empty")
+            for r in flagged:
+                what = (f"capture '{(r.get('args') or {}).get('title')}'" if r.get("op") == "capture"
+                        else f"{r.get('method')} {r.get('url')}")
+                print(f"  needs attention ({r['needs_attention']}): {what} — resolve "
+                      "(e.g. capture --merge-into <slug> / --force), it will not auto-land")
             return 0
         if args.cmd == "config":
             return cmd_config(args)
@@ -852,6 +1081,14 @@ def main(argv: list[str] | None = None) -> int:
             if not isinstance(proposals, list):
                 raise EchoError("reflect: proposals must be a JSON array of objects", 2)
             return echo_reflect.apply(proposals, confirm=args.apply)
+        if args.cmd == "session-end":
+            import echo_session
+            raw = read_body(args.file).decode("utf-8", errors="replace")
+            try:
+                bundle = json.loads(raw)
+            except json.JSONDecodeError as exc:
+                raise EchoError(f"session-end: bundle must be a JSON object ({exc})", 2)
+            return echo_session.session_end(bundle, apply=args.apply)
         if args.cmd == "triage":
             import echo_triage
             if args.list_inbox or not args.file:
@@ -880,9 +1117,12 @@ def main(argv: list[str] | None = None) -> int:
                 date=args.date, domain=args.domain, inbox=args.inbox, no_log=args.no_log,
                 as_json=args.json, dry_run=args.dry_run,
                 force=args.force, merge_into=args.merge_into)
-    except EchoError as exc:
+    except RuntimeError as exc:
+        # Catches EchoError from THIS module and from helper modules' own `import echo`
+        # twin (when echo.py runs as __main__ the two class objects differ, but both
+        # subclass RuntimeError and carry .code).
         print(f"echo.py: {exc}", file=sys.stderr)
-        return exc.code
+        return getattr(exc, "code", 1)
     return 2
 
 

echo-memory.plugin.src/skills/echo-memory/scripts/echo_doctor.py

@@ -22,17 +22,18 @@ import echo  # noqa: E402
 MIN_PY = (3, 9)
 
 
-def run() -> int:
+def run_op() -> dict:
-    reds = 0
+    """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
+        checks.append({"ok": ok, "label": label, "detail": detail})
-        reds += 0 if ok else 1
-        print(f"  [{'OK ' if ok else 'RED'}] {label}" + (f" — {detail}" if detail else ""))
 
     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,18 +54,16 @@ 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):
-        print("\ndoctor: not configured — ask the operator for their key file and install it "
+        fatal = "not-configured"
-              "(`echo.py config import <file>`, or `config set --owner … --endpoint … --key …`), "
+    else:
-              "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])
-        print("\ndoctor: endpoint unreachable — is Obsidian + the Local REST API running?")
+            fatal = "unreachable"
-        return 1
+        else:
-    line(status not in (401, 403), "auth accepted", f"HTTP {status}" if status in (401, 403) 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:
@@ -75,11 +74,30 @@ def run() -> int:
             else:
                 line(status < 400, "marker fetch", f"HTTP {status}")
 
-    # 4. invariants pointer.
+    reds = sum(1 for c in checks if not c["ok"])
-    print("  [i] invariants — run `/echo-health` (vault_lint.py) for the full check")
+    return {"ok": reds == 0 and fatal is None, "action": "doctor",
+            "endpoint": cfg["endpoint"], "fatal": fatal, "reds": reds, "checks": checks}
 
-    print(f"\ndoctor: {'all green' if reds == 0 else f'{reds} issue(s) — see RED above'}")
+
-    return 1 if reds else 0
+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
+    if env["fatal"] == "unreachable":
+        print("\ndoctor: endpoint unreachable — is Obsidian + the Local REST API running?")
+        return 1
+    # invariants pointer.
+    print("  [i] invariants — run `/echo-health` (vault_lint.py) for the full check")
+    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__":

echo-memory.plugin.src/skills/echo-memory/scripts/echo_hook_session_start.py

@@ -42,7 +42,7 @@ def main() -> int:
     try:
         import echo
         with contextlib.redirect_stdout(buf):
-            rc = echo.cmd_load()
+            rc = echo.cmd_load(brief=True)  # token-budgeted digest; /echo-load stays full
         text = buf.getvalue()
         if rc == 78:
             context = ("[echo-memory] ECHO is NOT CONFIGURED on this machine. Before "

echo-memory.plugin.src/skills/echo-memory/scripts/echo_hook_stop.py

@@ -30,9 +30,11 @@ MIN_TURNS = int(os.environ.get("ECHO_REFLECT_MIN_TURNS", "5"))
 REASON = (
     "[echo-memory] Session-end reflection has not run yet. If this session produced "
     "durable facts, decisions, commitments, or artifacts worth remembering: run the "
-    "/echo-reflect flow (extract -> preview -> apply only with the operator's confirm) "
+    "/echo-reflect flow (extract -> preview -> apply only with the operator's confirm), "
-    "and write the session log + heartbeat per the echo-memory skill. If nothing "
+    "then finish with ONE call — `echo.py session-end <bundle.json> --apply` — which "
-    "durable emerged, simply finish — do not invent memories. "
+    "writes the session log, Agent Log line, reflect proposals, optional scope switch, "
+    "and the heartbeat (last, as the commit marker) together. If nothing durable "
+    "emerged, simply finish — do not invent memories. "
     "(This reminder fires once per session.)"
 )
 
@@ -64,6 +66,7 @@ def already_reflected(raw: str) -> bool:
     """Transcript evidence that reflection or session logging already happened."""
     return ("/echo-reflect" in raw
             or re.search(r'echo\.py"?\s+reflect\b', raw) is not None
+            or re.search(r'echo\.py"?\s+session-end\b', raw) is not None
             or "put _agent/sessions/" in raw
             or "put _agent/heartbeat/last-session.md" in raw)
 

echo-memory.plugin.src/skills/echo-memory/scripts/echo_ops.py

@@ -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 {"match": True, "slug": slug, **e}
-        return 0
     # 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:
-    a_changed, b_changed = links.link_bidirectional(a_path, b_path)
+    """Core link: reciprocal `## Related` links, returns the change envelope."""
-    if as_json:
     import echo_output
-        env = echo_output.envelope("link", {"a": a_path, "b": b_path,
+    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:
         print(json.dumps(env, ensure_ascii=False))
         return 0
-    print(f"ok: linked {a_path} <-> {b_path} "
+    print(f"ok: linked {env['a']} <-> {env['b']} "
-          f"(added: {'A' if a_changed else '-'}{'B' if b_changed else '-'})")
+          f"(added: {'A' if env['a_changed'] else '-'}{'B' if env['b_changed'] else '-'})")
     return 0
 
 
@@ -83,8 +93,13 @@ def recall(query, limit: int = 8, as_json: bool = False) -> int:
 # ----------------------------------------------------------- agent log -------
 def ensure_daily_log(line: str) -> None:
     """Resilient: ensure today's daily note + its `## Agent Log` heading exist, then
-    idempotently append the line. Best-effort — never raises into the caller."""
+    idempotently append the line. Best-effort — never raises into the caller. Writes go
+    through the offline queue (2.1.0), so a mid-session outage queues the line instead
+    of silently dropping it. Offline edge accepted: if the daily note still doesn't
+    exist at replay time, the queued PATCH 400-drops with a loud warning — the agent-log
+    line is a convenience trace, not the memory itself."""
     try:
+        import echo_queue
         date = echo.today()
         path = f"journal/daily/{date}.md"
         status, body = echo.request("GET", echo.vault_url(path))
@@ -92,18 +107,19 @@ def ensure_daily_log(line: str) -> None:
             ts, tb = echo.request("GET", echo.vault_url("journal/templates/daily-note-template.md"))
             tmpl = tb.decode(errors="replace") if ts == 200 else f"# {date}\n\n## Agent Log\n\n## Related\n"
             tmpl = tmpl.replace("{{date:YYYY-MM-DD}}", date).replace("{{DATE}}", date)
-            echo.request("PUT", echo.vault_url(path), data=tmpl.encode(),
+            echo_queue.safe_request("PUT", echo.vault_url(path), data=tmpl.encode(),
                                     headers={"Content-Type": "text/markdown"})
             text = tmpl
         else:
-            text = body.decode(errors="replace")
+            text = body.decode(errors="replace") if status == 200 else ""
-        if not re.search(r"(?m)^## Agent Log\s*$", text):
+        if status == 200 and not re.search(r"(?m)^## Agent Log\s*$", text):
-            echo.request("POST", echo.vault_url(path), data=b"\n\n## Agent Log\n",
+            echo_queue.safe_request("POST", echo.vault_url(path), data=b"\n\n## Agent Log\n",
                                     headers={"Content-Type": "text/markdown"})
-        status, body = echo.request("GET", echo.vault_url(path))
+        st2, body2 = echo.request("GET", echo.vault_url(path))
-        if status == 200 and line in body.decode(errors="replace"):
+        if st2 == 200 and line in body2.decode(errors="replace"):
             return
-        echo.request("PATCH", echo.vault_url(path),
+        echo_queue.safe_request(
+            "PATCH", echo.vault_url(path),
             data=echo.normalize_patch_body((line + "\n").encode(), "append", "heading"),
             headers={"Operation": "append", "Target-Type": "heading",
                      "Target": f"{date}::Agent Log", "Content-Type": "text/markdown"})
@@ -148,81 +164,106 @@ def _dated_block(today_s: str, body_text: str) -> tuple[str, str]:
 
 
 def _append_to_existing(path: str, kind: str, today_s: str, body_text: str) -> None:
+    # Writes route through the offline queue (2.1.0) so a mid-capture outage queues the
+    # update instead of silently losing it. (A fully-offline capture never reaches here —
+    # the short-circuit in capture() queues the whole op as one semantic record.)
+    import echo_queue
     text = links.get_text(path) or ""
     h1 = links.first_h1(text) or path.rsplit("/", 1)[-1][:-3]
     heading = LOG_HEADING.get(kind, "Notes")
     if not re.search(rf"(?m)^##\s+{re.escape(heading)}\s*$", text):
-        echo.request("POST", echo.vault_url(path), data=f"\n\n## {heading}\n".encode(),
+        echo_queue.safe_request("POST", echo.vault_url(path), data=f"\n\n## {heading}\n".encode(),
                                 headers={"Content-Type": "text/markdown"})
     bullet, block = _dated_block(today_s, body_text)
     status, body = echo.request("GET", echo.vault_url(path))
     if status == 200 and bullet in body.decode(errors="replace"):
         return
-    echo.request("PATCH", echo.vault_url(path),
+    echo_queue.safe_request(
+        "PATCH", echo.vault_url(path),
         data=echo.normalize_patch_body((block + "\n").encode(), "append", "heading"),
         headers={"Operation": "append", "Target-Type": "heading",
                  "Target": f"{h1}::{heading}", "Content-Type": "text/markdown"})
     echo.cmd_fm(path, "updated", json.dumps(today_s))
 
 
-def capture(kind: str | None, title: str, file_arg: str | None, status_v: str = "",
+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,
-            as_json: bool = False, dry_run: bool = False, force: bool = False,
+               dry_run: bool = False, force: bool = False,
-            merge_into: str | None = None) -> int:
+               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
-    def quiet():
+        # stdout clean for the envelope by routing them to stderr wholesale.
-        # M4: in --json mode, swallow the helper "ok:" chatter so stdout is clean JSON.
+        return contextlib.redirect_stdout(sys.stderr)
-        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
 
+    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
+    # round-trips; when the vault is unreachable, queue the WHOLE capture as one semantic
+    # record — flush replays it back through this function against the index as it is at
+    # replay time, so routing, the duplicate gate, and aliasing re-run with fresh state.
+    # (Byte-level request replay would freeze a create-vs-update decision made blind.)
+    try:
         index = idx_mod.load()
+    except echo.EchoError as exc:
+        if "unreachable" not in str(exc):
+            raise
+        if dry_run:
+            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,
+            "aliases": aliases, "sources": sources, "tags": tags, "date": date,
+            "domain": domain, "inbox": False, "no_log": no_log,
+            "force": force, "merge_into": merge_into, "today": today_s,
+        })
+        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 "
+            raise echo.EchoError(f"--merge-into '{merge_into}' matches no entity in the "
-                  f"(try `resolve` first)", file=sys.stderr)
+                                 "index (try `resolve` first)", 2)
-            return 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
@@ -237,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:
+        return echo_output.envelope("duplicate-gate", {
-            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)
 
     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),
+        would_gate = (not force
-                    dry=True, near=near)
+                      and bool(idx_mod.gate_candidates(index, title, kind=kind,
-        if (not as_json and not force
+                                                       threshold=DUP_GATE)))
-                and 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),
-            # (in --json mode the near_duplicates field carries this; keep stdout clean)
+                       dry=True, near=near, would_gate=would_gate)
-            print("note: a real run would STOP at the duplicate gate — use --merge-into "
-                  "or --force.", file=real_stdout)
-        return plan
 
     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
@@ -335,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)
+        print(json.dumps(env, ensure_ascii=False))
-    else:
+        if action == "duplicate-gate":
-        print(f"ok: {action} {kind} -> {path}" + (f"; auto-linked {linked}" if linked else ""))
+            return 76
-        if near_dupes:
+        return 0 if env.get("ok") else 1
-            kind_word = "entity" if len(near_dupes) == 1 else "entities"
+
-            print(f"WARNING: similar existing {kind_word} ({', '.join(near_dupes)}) — if this is "
+    if action == "duplicate-gate":
-                  f"the same thing, merge or `echo.py link` instead of keeping a duplicate.",
+        print(f"STOP: '{title}' likely already exists — not creating a duplicate.")
-                  file=real_stdout)
+        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

echo-memory.plugin.src/skills/echo-memory/scripts/echo_queue.py

@@ -81,6 +81,23 @@ def enqueue(method: str, url: str, data: bytes | None, headers: dict, idem_key:
         fh.write(json.dumps(rec, ensure_ascii=False) + "\n")
 
 
+def enqueue_capture(args: dict) -> None:
+    """Queue a whole capture as ONE semantic record (2.1.0). Byte-level replay is wrong
+    for the routed capture path: the create-vs-update decision, duplicate gate, and
+    aliasing must re-run against the index AS IT IS AT REPLAY TIME, so flush re-invokes
+    echo_ops.capture with the recorded arguments instead of replaying stale requests.
+    `args` carries body_text inline (never a temp-file path) plus `today` (the capture-
+    time ECHO_TODAY) so replayed frontmatter/log dates reflect when it was said."""
+    key = "capture:" + hashlib.sha1(
+        json.dumps(args, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()
+    if any(rec.get("idem_key") == key for rec in pending()):
+        return
+    _ensure_dirs()
+    rec = {"op": "capture", "args": args, "idem_key": key}
+    with outbox_path().open("a", encoding="utf-8") as fh:
+        fh.write(json.dumps(rec, ensure_ascii=False) + "\n")
+
+
 def pending() -> list[dict]:
     p = outbox_path()
     if not p.exists():
@@ -88,6 +105,12 @@ def pending() -> list[dict]:
     return [json.loads(ln) for ln in p.read_text(encoding="utf-8").splitlines() if ln.strip()]
 
 
+def needs_attention() -> list[dict]:
+    """Queued records that replay could not land automatically (e.g. a duplicate-gate
+    stop) — surfaced by `flush` and `load` so the operator resolves them deliberately."""
+    return [rec for rec in pending() if rec.get("needs_attention")]
+
+
 def _rewrite(records: list[dict]) -> None:
     p = outbox_path()
     if not records:
@@ -108,9 +131,57 @@ def _rebase(url: str) -> str:
     return echo.BASE + parts.path + (("?" + parts.query) if parts.query else "")
 
 
+def _replay_capture(rec: dict) -> str:
+    """Replay a queued semantic capture through echo_ops.capture against the CURRENT
+    index. Returns 'ok', 'retry' (still offline), or 'gated' (duplicate gate / needs
+    the operator — the record is kept and flagged, later records still replay)."""
+    # Cheap reachability probe FIRST: if the vault is still down, echo_ops.capture's
+    # own offline short-circuit would try to re-enqueue this very record — probe and
+    # bail as 'retry' instead of recursing into the queue.
+    st, _ = echo.request("GET", echo.vault_url("_agent/echo-vault.md"))
+    if st == 0:
+        return "retry"
+    import echo_ops
+    a = dict(rec.get("args") or {})
+    body_text = a.pop("body_text", "") or ""
+    day = a.pop("today", None)
+    bodyfile = echo.temp_file(body_text.encode("utf-8")) if body_text else None
+    prev = os.environ.get("ECHO_TODAY")
+    try:
+        if day:
+            os.environ["ECHO_TODAY"] = day  # replayed dates reflect when it was said
+        rc = echo_ops.capture(
+            a.get("kind"), a.get("title") or "(untitled)", bodyfile,
+            status_v=a.get("status_v", ""), aliases=a.get("aliases") or [],
+            sources=a.get("sources") or [], tags=a.get("tags") or [],
+            date=a.get("date"), domain=a.get("domain", "business"),
+            inbox=bool(a.get("inbox")), no_log=bool(a.get("no_log")),
+            force=bool(a.get("force")), merge_into=a.get("merge_into"))
+    except Exception as exc:  # noqa: BLE001 — a broken record must not wedge the queue
+        print(f"echo_queue: queued capture '{a.get('title')}' failed on replay ({exc}) "
+              "— kept and flagged", file=sys.stderr)
+        return "gated"
+    finally:
+        if day:
+            if prev is None:
+                os.environ.pop("ECHO_TODAY", None)
+            else:
+                os.environ["ECHO_TODAY"] = prev
+    if rc == 0:
+        return "ok"
+    if rc == 76:
+        print(f"echo_queue: queued capture '{a.get('title')}' stopped at the duplicate "
+              "gate on replay — resolve with capture --merge-into <slug> or --force, "
+              "then remove it from the outbox", file=sys.stderr)
+    return "gated"
+
+
 def _replay(rec: dict) -> str:
     """Re-issue one queued write idempotently. Returns 'ok' (landed/already-present),
-    'drop' (permanent 4xx — can never succeed), or 'retry' (still offline / 5xx)."""
+    'drop' (permanent 4xx — can never succeed), 'gated' (kept + flagged for the
+    operator), or 'retry' (still offline / 5xx)."""
+    if rec.get("op") == "capture":
+        return _replay_capture(rec)
     method = rec["method"]
     url = _rebase(rec["url"])
     data = base64.b64decode(rec["body_b64"]) if rec.get("body_b64") else None
@@ -137,21 +208,30 @@ def _replay(rec: dict) -> str:
 
 def flush() -> int:
     """Replay queued writes in order; stop at the first still-unreachable one (keeping it
-    and the rest). Returns the number actually replayed. Safe to call when empty."""
+    and everything after). A 'gated' record (duplicate gate / replay error) is kept and
+    flagged but does NOT block later records — they are independent writes. Returns the
+    number actually replayed. Safe to call when empty."""
     items = pending()
     if not items:
         return 0
     replayed = 0
     remaining: list[dict] = []
-    for i, rec in enumerate(items):
+    stopped = False
+    for rec in items:
+        if stopped:
+            remaining.append(rec)
+            continue
         result = _replay(rec)
         if result == "ok":
             replayed += 1
         elif result == "drop":
             continue  # warned in _replay; remove from queue
+        elif result == "gated":
+            rec["needs_attention"] = rec.get("needs_attention") or "replay-gated"
+            remaining.append(rec)
         else:  # retry -> still offline; keep this and everything after, in order
-            remaining = items[i:]
+            remaining.append(rec)
-            break
+            stopped = True
     _rewrite(remaining)
     return replayed
 

echo-memory.plugin.src/skills/echo-memory/scripts/echo_recall.py

@@ -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,7 +442,6 @@ def recall(query, limit: int = 8, as_json: bool = False) -> int:
     # --- graph layer ----------------------------------------------------------
     neighbours = expand_graph(hits, nmap, base, max_hops=MAX_HOPS)
 
-    if as_json:
     import echo_output
     texts = echo.read_many(hits + [p for p, _ in neighbours[: 2 * limit]])
     primary = []
@@ -479,16 +456,35 @@ def recall(query, limit: int = 8, as_json: bool = False) -> int:
             continue
         linked.append({**_doc_summary(p, texts[p]),
                        "score": round(sc, 3), "via": via})
-        env = echo_output.envelope("recall", {"query": q, "primary": primary,
+    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:
         print(json.dumps(env, ensure_ascii=False))
         return 0
-
+    print(f"== recall: {env['query']} ==")
-    print(f"== recall: {q} ==")
+    print(f"\n# Primary hits ({len(env['primary'])})")
-    print(f"\n# Primary hits ({len(hits)})")
+    for info in env["primary"]:
-    for p in hits:
+        _print_hit(info)
-        _brief(p, score=base.get(p))
+    print(f"\n# Linked context ({len(env['linked'])})")
-    print(f"\n# Linked context ({len(neighbours)})")
+    for info in env["linked"]:
-    for p, (sc, via) in neighbours[: 2 * limit]:
+        _print_hit(info)
-        _brief(p, score=sc, via=via)
     return 0

echo-memory.plugin.src/skills/echo-memory/scripts/echo_reflect.py

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

echo-memory.plugin.src/skills/echo-memory/scripts/echo_session.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""echo_session.py — the session-end bundle: one call ends a session correctly. [2.1.0]
+
+Ending a substantive session used to take four-plus separate invocations (session-log
+PUT, Agent Log append, heartbeat PUT, optional scope set, optional reflect apply) —
+a cost per session, and a partial failure left broken orientation state (log written
+but heartbeat stale). This verb does all of it in one call, under ONE advisory-lock
+acquisition, in a fixed order with the heartbeat written LAST as the commit marker:
+a failure part-way leaves the previous pointer intact, never a dangling one.
+
+Bundle shape (JSON object):
+  {
+    "slug":           "echo-mcp-spec",          # required, kebab-case
+    "log_body":       "<full session-log markdown, frontmatter included>",  # required
+    "agent_log_line": "- 2026-07-28: ...",      # optional; derived when omitted
+    "scope":          "new scope text",          # optional -> scope set
+    "reflect":        [ { ...PROPOSAL_SCHEMA... } ]   # optional -> routed via capture
+  }
+
+Dry-run by default (previews the whole plan, reflect included); --apply writes.
+Times: the filename HHMM comes from $ECHO_NOW (HHMM) else the local clock; the date
+from ECHO_TODAY via echo.today(). Offline: every step rides the offline queue, so a
+session end during an outage queues durably instead of failing.
+
+CLI: echo.py session-end <bundle.json> [--apply]
+"""
+from __future__ import annotations
+
+import datetime as dt
+import json
+import os
+import re
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import echo  # noqa: E402
+
+SESSION_RE = re.compile(r"^\d{4}-\d{2}-\d{2}-\d{4}-[a-z0-9-]+\.md$")
+SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
+
+
+def _hhmm() -> str:
+    v = os.environ.get("ECHO_NOW", "").strip()
+    if v:
+        if not re.match(r"^\d{4}$", v):
+            raise echo.EchoError(f"session-end: $ECHO_NOW must be HHMM, got '{v}'", 2)
+        return v
+    return dt.datetime.now().strftime("%H%M")
+
+
+def validate(bundle: dict) -> tuple[str, str]:
+    """Return (session_path, agent_log_line) or raise EchoError(2). Validation runs
+    BEFORE any write — a bad bundle writes nothing at all."""
+    if not isinstance(bundle, dict):
+        raise echo.EchoError("session-end: bundle must be a JSON object", 2)
+    slug = str(bundle.get("slug", "")).strip()
+    if not SLUG_RE.match(slug):
+        raise echo.EchoError(f"session-end: slug must be kebab-case, got '{slug}'", 2)
+    if not str(bundle.get("log_body", "")).strip():
+        raise echo.EchoError("session-end: log_body is required (the full session-log markdown)", 2)
+    reflect = bundle.get("reflect")
+    if reflect is not None and not isinstance(reflect, list):
+        raise echo.EchoError("session-end: reflect must be a JSON array of proposals", 2)
+    fname = f"{echo.today()}-{_hhmm()}-{slug}.md"
+    if not SESSION_RE.match(fname):
+        raise echo.EchoError(f"session-end: derived filename '{fname}' violates the "
+                             "canonical YYYY-MM-DD-HHMM-<slug>.md form", 2)
+    path = f"_agent/sessions/{fname}"
+    line = str(bundle.get("agent_log_line") or "").strip() \
+        or f"- {echo.today()}: session logged [[{path[:-3]}]]"
+    return path, line
+
+
+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)
+    rows: list[dict] = []
+    if valid:
+        echo_reflect.classify(valid)
+        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:
+        return echo_output.envelope("session-end", data)
+
+    import echo_concurrency
+    import echo_ops
+    steps: dict[str, str] = {}
+    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.
+        echo.cmd_put(path, echo.temp_file(str(bundle["log_body"]).encode("utf-8")))
+        steps["session_log"] = "ok"
+        # 2. Agent-log line (best-effort by contract; queues itself when offline).
+        echo_ops.ensure_daily_log(line)
+        steps["agent_log"] = "ok"
+        # 3. Reflect proposals through the normal capture path (gate-aware).
+        if valid:
+            gated = 0
+            for p in valid:
+                if p.get("_action") == "error":
+                    continue
+                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 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:
+            steps["reflect"] = "skipped (none)"
+        # 4. Scope switch (optional).
+        if scope:
+            echo.cmd_scope("set", scope)
+            steps["scope"] = "ok"
+        else:
+            steps["scope"] = "unchanged"
+        # 5. Heartbeat LAST — the commit marker for the whole bundle.
+        echo.cmd_put("_agent/heartbeat/last-session.md",
+                     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 env["steps"].items()))
+    return 0

echo-memory.plugin.src/skills/echo-memory/scripts/echo_triage.py

@@ -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"))
-    if as_json:
+    return echo_output.envelope("triage-list", {"path": INBOX_PATH, "items": items,
-        import echo_output
-        env = 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:
         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:
+def route_op(proposals: list[dict], confirm: bool = False) -> dict:
-    """Validate -> classify -> preview; with confirm, route each via capture AND write
+    """Core triage routing (Phase 0): reflect pipeline + processing-log audit lines.
-    the processing-log audit line. Mirrors echo_reflect.apply's contract exactly."""
+    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:
+    rows: list[dict] = []
-        print(f"skip: {e}", file=sys.stderr)
+    counts: dict[str, int] = {}
-    if not valid:
+    if 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) — "
+        rows = [{"action": p.get("_action"), "kind": p.get("kind"),
-          f"{counts['create']} new, {counts['update']} update, {counts['inbox']} stay-in-inbox"
+                 "title": p["title"], "path": p.get("_path")} for p in valid]
-          + (f", {counts['error']} error" if counts["error"] else ""))
+    data = {"proposals": len(proposals or []), "valid": len(valid), "errors": errors,
-    print(echo_reflect.preview(valid))
+            "counts": counts, "rows": rows, "dry_run": not confirm,
-
+            "routed": 0, "gated": 0, "log": f"{LOG_DIR}/{echo.today()}.md", "results": []}
-    if not confirm:
+    if not valid or not confirm:
-        print("\ntriage: dry-run — re-run with --apply to route these and log the moves.")
+        return echo_output.envelope("triage", data)
-        return 0
 
     import echo_ops
     today_s = echo.today()
-    applied = gated = 0
+    routed = gated = 0
+    results = []
     for p in valid:
-        if p.get("_action") == "error":
+        if p.get("_action") in ("error", "inbox"):
-            continue
-        if p.get("_action") == "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
+        env = echo_ops.capture_op(
-        rc = echo_ops.capture(
+            p.get("kind"), p["title"], body_text=p.get("body") or "",
-            p.get("kind"), p["title"], bodyfile,
             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()
+        with contextlib.redirect_stdout(sys.stderr):
             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. "
+    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 ""))

echo-memory.plugin.src/skills/echo-reflect/SKILL.md

@@ -30,4 +30,9 @@ python3 "$ECHO" reflect proposals.json --apply    # write — routes each via ca
 `reflect` dedups every proposal against the entity index (so it shows create-vs-update),
 skips below-confidence items, and routes `inbox` proposals to the capture inbox. Each
 applied item goes through the normal `capture` path — canonical frontmatter, auto-linking,
-and the recall index — and the writes are lock-guarded. $ARGUMENTS
+and the recall index — and the writes are lock-guarded.
+
+**Finishing the session?** Prefer the one-call bundle — `python3 "$ECHO" session-end
+bundle.json --apply` — which applies the reflect proposals AND writes the session log,
+Agent Log line, optional scope switch, and the heartbeat (last, as the commit marker)
+together. See the echo-memory skill's **Session Logging** section for the bundle shape. $ARGUMENTS

eval/test_features.py

@@ -345,6 +345,63 @@ def main():
         check("v1.5 session-start hook stays quiet on resume",
               r.returncode == 0 and not r.stdout.strip(), r.stdout)
 
+        # 18. (2.1.0) brief load: digest form, budget respected, key facts present.
+        h.seed("_agent/memory/semantic/operator-preferences.md",
+               "---\ntype: semantic-memory\ncreated: 2026-06-01\n---\n# Operator Preferences\n\n"
+               "## Fact / Pattern\n- prefers concise output\n\n## Observations\n"
+               + "\n".join(f"- 2026-06-{i:02d}: observation {i}" for i in range(1, 15)) + "\n")
+        h.seed("inbox/captures/inbox.md", "- 2026-06-01: old capture\n- 2026-06-20: new capture\n")
+        r = h.echo(ECHO, "load", "--brief")
+        check("v2.1 brief load renders the digest header",
+              "ECHO load (brief)" in r.stdout, r.stdout[:200])
+        check("v2.1 brief load keeps Fact / Pattern",
+              "prefers concise output" in r.stdout, r.stdout[:800])
+        check("v2.1 brief load trims observations to the last 10",
+              "last 10 of 14" in r.stdout and "observation 14" in r.stdout
+              and "observation 2" not in r.stdout.replace("observation 2026", ""), r.stdout[-1200:])
+        check("v2.1 brief load reports inbox as a count",
+              "inbox: 2 capture(s), oldest 20d" in r.stdout, r.stdout[-400:])
+        check("v2.1 brief load includes the marker line",
+              "echo-vault.md" in r.stdout and "marker" in r.stdout, r.stdout[:300])
+
+        # 19. (2.1.0) session-end: dry-run writes nothing; apply lands all steps with
+        # the heartbeat LAST; a bad bundle aborts before any write.
+        se_env = dict(os.environ, ECHO_BASE=base, ECHO_KEY=KEY, ECHO_VERIFY="1",
+                      ECHO_TODAY="2026-06-21", ECHO_NOW="2359")
+        bundle = {"slug": "quickwins-ship",
+                  "log_body": "---\ntype: session-log\nstatus: complete\ncreated: 2026-06-21\n---\n"
+                              "# Session Log\n\n## Goal\nShip the quick wins.\n",
+                  "agent_log_line": "- 2026-06-21: shipped the quick wins",
+                  "reflect": [{"title": "Quark Preference", "kind": "semantic",
+                               "body": "The operator prefers quark.", "confidence": 0.9}]}
+        bfile = HERE / "_session_bundle.json"
+        bfile.write_text(json.dumps(bundle), encoding="utf-8")
+        sess_path = "_agent/sessions/2026-06-21-2359-quickwins-ship.md"
+        r = subprocess.run([sys.executable, str(ECHO), "session-end", str(bfile)],
+                           capture_output=True, text=True, env=se_env)
+        check("v2.1 session-end dry-run previews the plan",
+              "dry-run" in r.stdout and sess_path in r.stdout, r.stdout)
+        check("v2.1 session-end dry-run writes nothing", h.ground(sess_path) is None)
+        r = subprocess.run([sys.executable, str(ECHO), "session-end", str(bfile), "--apply"],
+                           capture_output=True, text=True, env=se_env)
+        check("v2.1 session-end apply lands the session log",
+              "Ship the quick wins" in (h.ground(sess_path) or ""), r.stdout + r.stderr)
+        check("v2.1 session-end apply routes the reflect proposal",
+              h.ground("_agent/memory/semantic/quark-preference.md") is not None)
+        check("v2.1 session-end apply writes the heartbeat last (commit marker)",
+              sess_path in (h.ground("_agent/heartbeat/last-session.md") or ""))
+        daily = h.ground("journal/daily/2026-06-21.md") or ""
+        check("v2.1 session-end apply appends the agent-log line",
+              "shipped the quick wins" in daily, daily[-300:])
+        bad = dict(bundle, slug="Bad_Slug!")
+        bfile.write_text(json.dumps(bad), encoding="utf-8")
+        r = subprocess.run([sys.executable, str(ECHO), "session-end", str(bfile), "--apply"],
+                           capture_output=True, text=True, env=se_env)
+        bfile.unlink()
+        check("v2.1 session-end rejects a bad slug before any write",
+              r.returncode == 2 and sess_path in (h.ground("_agent/heartbeat/last-session.md") or ""),
+              r.stderr)
+
         print(f"\n{len(failures)} failure(s)" if failures else "\nall feature tests passed")
         return 1 if failures else 0
     finally:

eval/test_offline_queue.py

@@ -115,6 +115,40 @@ def main():
         check("offline load flags OFFLINE", "OFFLINE" in r.stdout, r.stdout)
         check("offline load serves cached marker", "EVALMARK-marker" in r.stdout, r.stdout)
 
+        # --- Phase 5 (2.1.0): vault DOWN -> capture queues a SEMANTIC record ------
+        r = echo(DEAD, "capture", "Offline Person", "--kind", "person")
+        check("offline capture exits 0 (queued)", r.returncode == 0, r.stdout + r.stderr)
+        check("offline capture reports queued", "queued (offline): capture" in r.stdout, r.stdout)
+        outbox = Path(state) / "outbox.ndjson"
+        check("capture queued as an op record",
+              outbox.exists() and '"op": "capture"' in outbox.read_text(encoding="utf-8"))
+        # same capture again while offline -> deduped by idem_key, not double-queued
+        echo(DEAD, "capture", "Offline Person", "--kind", "person")
+        recs = [ln for ln in outbox.read_text(encoding="utf-8").splitlines() if '"op": "capture"' in ln]
+        check("offline capture is idempotent in the queue", len(recs) == 1, str(len(recs)))
+
+        # --- Phase 6 (2.1.0): vault UP -> flush replays capture THROUGH capture ---
+        srv = start_mock()
+        r = echo(mock_base, "flush")
+        check("flush replays the queued capture", "flushed" in r.stdout, r.stdout + r.stderr)
+        note = ground("resources/people/offline-person.md")
+        check("replayed capture created the routed note",
+              note is not None and "type: person" in note, str(note)[:200])
+        check("replayed capture used the capture-time date",
+              note is not None and "created: 2026-06-22" in note, str(note)[:200])
+
+        # --- Phase 7 (2.1.0): duplicate gate ON REPLAY keeps + flags the record ---
+        echo(mock_base, "capture", "Zebulon Quargle", "--kind", "person")  # the existing entity
+        r = echo(DEAD, "capture", "Zebulon Quargle Junior", "--kind", "person")
+        check("lookalike capture queues offline", "queued (offline)" in r.stdout, r.stdout)
+        r = echo(mock_base, "flush")
+        check("gated replay is flagged, not landed",
+              "needs attention" in r.stdout and "replay-gated" in r.stdout, r.stdout + r.stderr)
+        check("gated replay did NOT create the duplicate",
+              ground("resources/people/zebulon-quargle-junior.md") is None)
+        check("gated record survives in the outbox",
+              '"needs_attention"' in (outbox.read_text(encoding="utf-8") if outbox.exists() else ""))
+
         print(f"\n{len(failures)} failure(s)" if failures else "\nall offline-queue tests passed")
         return 1 if failures else 0
     finally:

eval/test_ops_api.py

@@ -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())