fix(plugin/tether): don't drop replies on async parse; add ask (email-and-wait)

.claude-plugin/marketplace.json

@@ -7,7 +7,7 @@
   },
   "metadata": {
     "description": "e2a plugins for Claude Code — authenticated email for AI agents",
-    "version": "0.4.2"
+    "version": "0.4.3"
   },
   "plugins": [
     {

.cursor-plugin/marketplace.json

@@ -6,7 +6,7 @@
   },
   "metadata": {
     "description": "e2a — authenticated email gateway for AI agents (MCP server + operate-well skill).",
-    "version": "0.4.2"
+    "version": "0.4.3"
   },
   "plugins": [
     {

plugins/e2a/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "e2a",
   "displayName": "e2a",
-  "version": "0.4.2",
+  "version": "0.4.3",
   "description": "Authenticated email gateway for AI agents — per-agent inboxes, HITL approval, SPF/DKIM verification, and a queryable, replayable event log. 37 MCP tools over hosted streamable HTTP with OAuth.",
   "author": {
     "name": "Mnexa AI",

plugins/e2a/.codex-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "e2a",
   "displayName": "e2a",
-  "version": "0.4.2",
+  "version": "0.4.3",
   "description": "Authenticated email gateway for AI agents — per-agent inboxes, HITL approval, SPF/DKIM verification, and a queryable, replayable event log. 37 MCP tools over hosted streamable HTTP with OAuth.",
   "author": {
     "name": "Mnexa AI"

plugins/e2a/.cursor-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "e2a",
   "displayName": "e2a",
-  "version": "0.4.2",
+  "version": "0.4.3",
   "description": "Authenticated email gateway for AI agents — per-agent inboxes, HITL approval, SPF/DKIM verification, and a queryable, replayable event log. 37 MCP tools over hosted streamable HTTP with OAuth.",
   "author": {
     "name": "Mnexa AI",

plugins/e2a/skills/tether/SKILL.md

@@ -26,8 +26,18 @@ the two directions use different mechanisms:
   fetch replies. **Keep the session alive with `/loop`** so polling continues
   while the agent would otherwise be idle — this is the only way a reply sent
   *after* the agent goes idle gets picked up (no hook fires while idle).
+- **Questions = ask by email.** When the agent needs a decision or clarification
+  from you, it must **not** use a terminal prompt / AskUserQuestion — you're AFK
+  and can't see it, which would stall the whole session. It calls
+  `tether.sh ask "<question>"`, which emails the question into the thread and
+  **blocks until you reply**, then prints your answer. This is the hard rule:
+  **while tethered, every question goes over email, never the terminal.**
 - **Blocked-alert = hook (optional).** A `Notification` hook emails you when the
-  agent is stuck on a permission prompt and can't proceed at all.
+  agent is stuck on a *permission prompt* it can't proceed past. Note: an emailed
+  reply cannot answer a CLI permission prompt (there's no native way to inject
+  approval), so for unattended runs **pre-authorize** the tools the session needs
+  (a permission allowlist / a less-prompting mode). The hook is the safety net,
+  not the approval channel.
 
 ## Setup (once)
 
@@ -51,11 +61,17 @@ Let `T="${CLAUDE_PLUGIN_ROOT}/skills/tether/tether.sh"`.
 3. **Work**, and **send updates as you see fit**: `"$T" update "<what changed / what you need>"`.
    Good moments: finished a slice, made a decision that's worth surfacing, hit a
    blocker, or before a long unattended stretch. Skip trivial turns.
-4. **Poll on an interval**: run `"$T" poll`; if it prints a reply, treat it as a
+4. **Need a decision from the user? Ask by email — never the terminal.** Run
+   `"$T" ask "<question>"` (in the background); it emails the question and blocks
+   until the user replies, then prints the answer. **Do not** use AskUserQuestion
+   or a bare terminal prompt while tethered — an AFK user can't answer it and the
+   session stalls.
+5. **Poll on an interval**: run `"$T" poll`; if it prints a reply, treat it as a
    new instruction and act on it (then `update` with the result). To keep polling
    while idle, run the session under **`/loop <interval>`** (or self-schedule the
-   next poll). See the interval guidance below.
-5. **Stop** when the user replies `stop`/`done`, or the work is complete:
+   next poll). See the interval guidance below. (Replies are deduped by
+   message-id and survive e2a's async parse, so none are dropped or repeated.)
+6. **Stop** when the user replies `stop`/`done`, or the work is complete:
    `"$T" stop`.
 
 ## Interval guidance
@@ -87,7 +103,7 @@ working context). Left as a follow-on.
 
 | file | role |
 |---|---|
-| `tether.sh` | runtime CLI: `start` / `update` / `poll` / `status` / `stop` |
+| `tether.sh` | runtime CLI: `start` / `update` / `ask` / `poll` / `status` / `stop` |
 | `lib.sh` | config + e2a send/reply/poll helpers |
 | `hooks/tether-notify.sh` | optional Notification hook (blocked-alert) |
 | `install.sh` | wire/unwire the Notification hook; `_selftest` |

plugins/e2a/skills/tether/lib.sh

@@ -119,3 +119,65 @@ try:
   print((p or b).strip())
 except Exception:print("")'
 }
+
+# --- dedup + poll core -------------------------------------------------------
+# Replies are deduped by message-id (a `seen` set in state), NOT a bare time
+# cursor. Email parsing is async: a just-arrived reply can have an empty
+# parsed/body for a moment. We therefore do NOT mark such a message seen or
+# advance the watermark past it — we retry it next poll (up to a max age),
+# so a reply can never be silently skipped (the bug that dropped a real reply).
+
+# seconds since an RFC3339 timestamp
+t_age_seconds() {
+  python3 -c 'import sys,datetime
+try:
+  t=datetime.datetime.fromisoformat(sys.argv[1].replace("Z","+00:00"))
+  print(int((datetime.datetime.now(datetime.timezone.utc)-t).total_seconds()))
+except Exception:print(0)' "$1"
+}
+
+t_seen_has() {  # <id> → 0 if already processed
+  local f; f="$(t_state_path)"; [ -f "$f" ] || return 1
+  python3 -c 'import json,sys
+try:sys.exit(0 if sys.argv[2] in (json.load(open(sys.argv[1])).get("seen") or []) else 1)
+except Exception:sys.exit(1)' "$f" "$1"
+}
+
+t_seen_add() {  # <id> → record as processed (cap at last 500)
+  local f; f="$(t_state_path)"; mkdir -p "$(dirname "$f")"
+  python3 -c 'import json,sys,os
+f,i=sys.argv[1],sys.argv[2]
+d={}
+if os.path.exists(f):
+  try:d=json.load(open(f))
+  except Exception:d={}
+s=d.get("seen") or []
+if i not in s:s.append(i)
+d["seen"]=s[-500:]
+json.dump(d,open(f,"w"),indent=2)' "$f" "$1"
+}
+
+# t_poll_once → print any new replies (dedup + parse-race safe), else "(no new replies)"
+# Advances the `last_poll` watermark only through the contiguous processed prefix,
+# stopping at the first not-yet-parsed message so it is retried, never lost.
+t_poll_once() {
+  local conv since rows n advance id from created body age
+  conv="$(t_state_get conversation_id)"; since="$(t_state_get last_poll)"
+  rows="$(t_api_poll "$conv" "$since")" || return 1
+  n=0; advance="$since"
+  while IFS=$'\t' read -r id from created; do
+    [ -n "$id" ] || continue
+    if t_seen_has "$id"; then advance="$created"; continue; fi
+    body="$(t_api_body "$id")"
+    if [ -z "$body" ]; then
+      age="$(t_age_seconds "$created")"
+      if [ "$age" -gt 120 ]; then t_seen_add "$id"; advance="$created"; continue
+      else break; fi   # not parsed yet — retry next poll, don't advance past it
+    fi
+    t_seen_add "$id"; advance="$created"; n=$((n+1))
+    printf '── reply from %s @ %s ──\n%s\n\n' "$from" "$created" "$body"
+    t_state_set last_message_id "$id"
+  done <<< "$rows"
+  t_state_set last_poll "$advance"
+  [ "$n" -gt 0 ] || echo "(no new replies)"
+}

plugins/e2a/skills/tether/tether.sh

@@ -3,6 +3,7 @@
 #
 #   tether.sh start <your-email>   send the intro email, open the thread, arm
 #   tether.sh update "<message>"   send a threaded update ("as you see fit")
+#   tether.sh ask "<question>"     email a question and BLOCK until the reply
 #   tether.sh poll                 print any new replies since last poll (exit 0)
 #   tether.sh status               show tether state
 #   tether.sh stop                 disarm and clear state
@@ -53,24 +54,30 @@ minutes). Reply any time with a question or instruction; reply \"stop\" to end.
 
   poll)
     need_config; need_armed
-    conv="$(t_state_get conversation_id)"; since="$(t_state_get last_poll)"
-    checkpoint="$(t_now_iso)"
-    rows="$(t_api_poll "$conv" "$since")"
-    # advance the cursor regardless so we never re-report the same reply
-    t_state_set last_poll "$checkpoint"
-    [ -n "$rows" ] || { echo "(no new replies)"; exit 0; }
-    n=0
-    while IFS=$'\t' read -r id from created; do
-      [ -n "$id" ] || continue
-      body="$(t_api_body "$id")"
-      [ -n "$body" ] || continue
-      n=$((n+1))
-      echo "── reply from ${from} @ ${created} ──"
-      echo "$body"
-      echo
-      t_state_set last_message_id "$id"   # thread the next update off the user's latest
-    done <<< "$rows"
-    [ "$n" -gt 0 ] || echo "(no new replies)"
+    t_poll_once
+    ;;
+
+  ask)
+    # Email a question into the thread and BLOCK until the user replies, then
+    # print the answer. This is how a tethered agent asks the user anything —
+    # over email, never a terminal prompt the AFK user can't see. Run it in the
+    # background and wait for the completion notification.
+    need_config; need_armed
+    q="${1:-}"; [ -n "$q" ] || { echo "usage: tether.sh ask \"<question>\""; exit 2; }
+    rid="$(t_state_get last_message_id)"
+    mid="$(t_api_reply "$rid" "❓ ${q}
+
+(Reply to this email with your answer — I'll wait for it.)")"
+    [ -n "$mid" ] || { echo "tether: ask send failed"; exit 1; }
+    t_state_set last_message_id "$mid"
+    echo "tether: question sent (${mid}); waiting for your reply…"
+    max="${E2A_TETHER_ASK_TIMEOUT:-1800}"; interval="${E2A_TETHER_POLL_INTERVAL:-20}"; elapsed=0
+    while [ "$elapsed" -lt "$max" ]; do
+      sleep "$interval"; elapsed=$((elapsed + interval))
+      out="$(t_poll_once)"
+      [ "$out" = "(no new replies)" ] || { echo "$out"; exit 0; }
+    done
+    echo "tether: ask timed out after ${max}s with no answer"; exit 3
     ;;
 
   status)