jason/echo
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

2.1.0 — quick-wins train: brief load, offline-durable capture, session-end verb

Browse Source 
Three independently useful changes (IMPROVEMENT-PLANS #1/#3/#7), no schema changes:

- load --brief: token-budgeted cold-start digest (facts full, last-10 observations,
  scope+freshness, last session's key sections, agent-log lines, inbox count;
  ECHO_LOAD_BUDGET ~8000 chars, lowest-priority-first trimming). SessionStart hook
  injects the brief form; /echo-load keeps the full dump. All load reads (+ the
  sessions listing, which also feeds the heartbeat fallback) fetched in parallel.
- Offline capture durability: a fully-offline capture queues the WHOLE op 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
  (needs_attention, surfaced by flush and load), never landed blind; update-path
  and ensure_daily_log writes ride safe_request; same-args captures dedupe.
- session-end: one call, one lock — session log -> agent-log line -> reflect
  proposals (gate-aware) -> optional scope set -> heartbeat LAST as the commit
  marker; dry-run default; ECHO_NOW pins HHMM; validation runs before any write.
  Stop hook nudge names the command and recognizes it as reflected.
- Fix: main() now catches the __main__ twin-module EchoError (via RuntimeError +
  .code) so helper-module errors exit with their intended codes, not tracebacks.

+19 e2e checks; all suites green. Plans renumbered: MCP container is 2.2.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jason Stedwell
2026-07-28 22:34:42 -05:00
parent 1deef7299e
commit 446cd9a0ff
16 changed files with 702 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
echo-memory.plugin.src/skills/echo-memory/scripts/echo.py
+220 -36
View File
@@ -605,9 +605,136 @@ def cmd_scope(subcommand: str, text: str | None = None, as_json: bool = False) -
return 0
def cmd_load() -> int:
"""Cold-start orientation: the canonical 6 reads in one call. 404s on today's
daily note and the inbox are normal (printed as absent, not errors)."""
def _fetch_statuses(paths: list[str]) -> dict[str, tuple[int, bytes]]:
"""Concurrently GET vault paths keeping (status, body) per path — unlike read_many,
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
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()}")
@@ -631,35 +758,77 @@ def cmd_load() -> int:
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)
# All orientation reads (+ the sessions listing) fetched in parallel over the
# warm connection pool — the listing feeds both the brief digest's scope-freshness
# count and the heartbeat-absent fallback, so it's no longer a second round-trip.
listing_path = "_agent/sessions/"
results = _fetch_statuses([p for _, p in targets] + [listing_path])
marker_missing = False
heartbeat_absent = False
offline = False
texts: dict[str, str | None] = {}
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
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
if brief:
hb = texts.get("heartbeat")
if hb and hb.strip(): # follow-up: the pointed session log's key sections
sess_path = hb.strip().splitlines()[0].split(" @ ", 1)[0].strip()
if sess_path:
st2, b2 = request("GET", vault_url(sess_path))
if st2 == 200:
texts["_pointed_session"] = b2.decode(errors="replace")
print(_render_brief(texts, listing_files, offline))
return 0
# ---- full mode: raw sections, output format unchanged ----
for label, path in targets:
status, body = results[path]
if status == 200:
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":
marker_missing = True
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:
elif status == 0:
if texts[label] is not None:
print(f"===== {label}: {path} (OFFLINE — serving stale cache) =====")
sys.stdout.write(cached.decode(errors="replace"))
if not cached.endswith(b"\n"):
sys.stdout.write(texts[label])
if not texts[label].endswith("\n"):
sys.stdout.write("\n")
else:
print(f"===== {label}: {path} (OFFLINE — no cache) =====")
@@ -669,20 +838,14 @@ 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.
if heartbeat_absent and not offline:
st, body = request("GET", vault_url("_agent/sessions/"))
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]:
print(f" _agent/sessions/{f}")
print()
# (already fetched in the batch) so orientation works without the pointer.
if heartbeat_absent and not offline and listing_files:
files = sorted((f for f in listing_files if f.endswith(".md")), reverse=True)
if files:
print("===== recent sessions (heartbeat absent — fallback) =====")
for f in files[:5]:
print(f" _agent/sessions/{f}")
print()
if offline:
print("NOTE: vault unreachable — context above is last-known-good cache; writes this "
"session will be queued and synced when the vault returns.")
@@ -732,7 +895,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 +925,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 +963,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 +1025,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 +1061,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_hook_session_start.py
+1 -1
View File
@@ -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
+6 -3
View File
@@ -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) "
"and write the session log + heartbeat per the echo-memory skill. If nothing "
"durable emerged, simply finish — do not invent memories. "
"/echo-reflect flow (extract -> preview -> apply only with the operator's confirm), "
"then finish with ONE call — `echo.py session-end <bundle.json> --apply` — which "
"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
+59 -20
View File
@@ -83,8 +83,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,21 +97,22 @@ 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(),
headers={"Content-Type": "text/markdown"})
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")
if not re.search(r"(?m)^## Agent Log\s*$", text):
echo.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))
if status == 200 and line in body.decode(errors="replace"):
text = body.decode(errors="replace") if status == 200 else ""
if status == 200 and not re.search(r"(?m)^## Agent Log\s*$", text):
echo_queue.safe_request("POST", echo.vault_url(path), data=b"\n\n## Agent Log\n",
headers={"Content-Type": "text/markdown"})
st2, body2 = echo.request("GET", echo.vault_url(path))
if st2 == 200 and line in body2.decode(errors="replace"):
return
echo.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"})
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"})
except Exception as exc:
print(f"echo_ops: agent-log skipped ({exc})", file=sys.stderr)
@@ -148,20 +154,25 @@ 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(),
headers={"Content-Type": "text/markdown"})
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),
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_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))
@@ -213,7 +224,35 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
return done("inbox", "inbox/captures/inbox.md", ok=rc == 0)
slug = idx_mod.slugify(title)
index = idx_mod.load()
# 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:
print("offline: vault unreachable — a real run would queue this capture "
"for replay on the next reachable session.", file=real_stdout)
return 0
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,
})
if as_json:
env = echo_output.envelope("queued:capture",
{"kind": kind, "title": title, "queued": True})
print(json.dumps(env, ensure_ascii=False), file=real_stdout)
else:
print(f"queued (offline): capture {kind} '{title}' — will replay through "
"capture on the next reachable session", file=real_stdout)
return 0
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.
echo-memory.plugin.src/skills/echo-memory/scripts/echo_queue.py
+85 -5
View File
@@ -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:]
break
remaining.append(rec)
stopped = True
_rewrite(remaining)
return replayed
echo-memory.plugin.src/skills/echo-memory/scripts/echo_session.py
+144
View File
@@ -0,0 +1,144 @@
#!/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(bundle: dict, apply: bool = False) -> int:
import echo_reflect
path, line = validate(bundle)
scope = str(bundle.get("scope") or "").strip()
proposals = bundle.get("reflect") or []
valid, errors = echo_reflect.validate(proposals)
for e in errors:
print(f"skip: {e}", file=sys.stderr)
print(f"session-end plan ({'APPLY' if apply else 'dry-run'}):")
print(f" 1. session log -> {path}")
print(f" 2. agent-log line: {line}")
if valid:
echo_reflect.classify(valid)
print(f" 3. reflect: {len(valid)} proposal(s)")
print(echo_reflect.preview(valid))
else:
print(" 3. reflect: (none)")
print(f" 4. scope set: {scope!r}" if scope else " 4. scope: (unchanged)")
print(f" 5. heartbeat -> {path} @ <now> (written LAST — the commit marker)")
if not apply:
print("\nsession-end: dry-run — re-run with --apply to write.")
return 0
import echo_concurrency
import echo_ops
steps: dict[str, str] = {}
with echo_concurrency.vault_lock():
# 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
bodyfile = echo.temp_file((p.get("body") or "").encode()) if p.get("body") else None
rc = echo_ops.capture(
p.get("kind"), p["title"], bodyfile,
aliases=p.get("aliases") or [], sources=p.get("sources") or [],
tags=p.get("tags") or [], date=p.get("date"),
domain=p.get("domain", "business"),
inbox=bool(p.get("inbox")) or not p.get("kind"))
gated += 1 if rc == 76 else 0
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"
print("\nsession-end: done — "
+ "; ".join(f"{k}: {v}" for k, v in steps.items()))
return 0