fix(T667): requeue a watcher-exit wake dropped mid-turn instead of stranding it

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- **A watcher-exit wake dropped mid-turn is not lost with its hint (T667).** A terminal Kimi session wakes itself: `agentmbx watch` exits when mail that wants it arrives, and the harness turns that background-task completion into the next turn. A completion that lands while a turn is running is dropped, not queued, and mail the watcher already reported sat `notified` and invisible to the dispatcher — after an exit-2 Stop continuation (Kimi runs one Stop per turn) nothing surfaced it until the next owner prompt. The exiting watcher now records which message ids its hint named, and the Stop hook reconciles that marker for mail the turn surfaces, so the two never double-fire. A marked delivery still unread once the batching window has passed is requeued to `delivered` exactly once — no requeue while a live watcher, a mute, or the brake would handle it — so it rides the next watcher report, prompt notice or notice fallback instead of stranding. Mail read or acked before the turn ends is never re-driven. Sessions whose hooks do not dispatch at all (T611) keep their documented MCP fallback.
+
 ## 0.5.47 (2026-10-16)
 
 - **Service-hosted OpenCode sessions stay wakeable after MCP eviction (T688).** OpenCode stops an idle session's MCP about hourly, which releases its lease. T659 kept standalone sessions wakeable through their plugin's poll; a session hosted by the shared `opencode serve --service` is now a wake target too while that service process (same pid and birth) lives, and the wake posts to the service with `resume: true` and a verified receipt. A lease held by another session of the agent still blocks it. docs/OPENCODE.md recommends plain `opencode` (one service per machine, every project) for long-lived agents.

dist/cli.js

@@ -37,7 +37,7 @@ import { periodicUpdateCheck, updateAvailable, updateCommand } from "./update.js
 import { installKind, version } from "./version.js";
 import { installService, serviceLabel, uninstallService } from "./service.js";
 import { CLIS, defaultHostName, defaultWhich, formatRows, ownerStep, resolveCommand, runSetup, selfHealSkill, shJoin } from "./setup.js";
-import { dispatchWakes, hasWakeAuthority, humanPromptKey, inboxCommand, isHumanPrompt, macNotifierPath, muteWakes, notifyDesktop, opencodeService, liveWatcher, wakeMutedUntil, wakeText, watcherKey, which } from "./wake.js";
+import { dispatchWakes, hasWakeAuthority, humanPromptKey, inboxCommand, isHumanPrompt, macNotifierPath, muteWakes, notifyDesktop, opencodeService, liveWatcher, wakeMutedUntil, wakeText, watcherKey, which, noteWatchReported, reconcileWatchReported } from "./wake.js";
 import { kimiMultiHost } from "./kimi-web.js";
 import { bindInstruction, issueBindTicket } from "./bind-ticket.js";
 import { activityKey } from "./identity-availability.js";
@@ -2173,6 +2173,9 @@ async function hook(node, event, cli) {
                 const digest = stopDigest();
                 const fresh = node.inbox(agent, { limit: 50 })
                     .filter(m => m.state !== "read" && m.received_at > seen && m.from_addr !== `${agent}@${node.host}` && node.wantsWake(agent, m) && hasWakeAuthority(node, agent, m) && (!digest || WAKE_KINDS.has(m.kind)));
+                // T667: this turn ended. The mail this Stop surfaces rides the continuation, and read/acked mail is
+                // handled: both leave the dropped-exit marker so the daemon's one re-drive never double-fires.
+                reconcileWatchReported(node, agent, fresh.map(m => m.id));
                 // T435: mail stays first. After it is consumed, one continuation re-arms a down watcher.
                 // Mail that arrives after the watcher exits and before the next tool or Stop still only
                 // gets a desktop notice — compaction does not fire Stop, and this gate cannot close that gap.
@@ -2457,6 +2460,10 @@ async function watch(node, selection) {
                         : mailbox === agent ? "" : ` This is your linked mailbox: use agentmbx inbox --as ${mailbox} and agentmbx ack --as ${mailbox} <id>.`;
                     if (!reservation.brake)
                         lines.push(wakeText(subject, wanted, box ? projectLabel(box) : undefined) + linked);
+                    // T667: the printed hint may be dropped mid-turn (the harness does not queue the completion).
+                    // Mark the ids so the daemon requeues the mail once if nothing ever surfaces it.
+                    if (!reservation.brake)
+                        noteWatchReported(node.store, mailbox, wanted.map(r => r.id));
                 }
                 for (const r of rows) {
                     node.setDelivery(r.id, mailbox, "notified");

dist/wake.js

@@ -7,7 +7,7 @@ import { join } from "node:path";
 import { promisify } from "node:util";
 import { kimiHostedServer } from "./kimi-web.js";
 import { desktopRpc, kimiDesktop, RpcError } from "./kimi-desktop.js";
-import { GROK_NO_PUSH, MbxNode, trustLabel } from "./node.js";
+import { GROK_NO_PUSH, MbxNode, trustLabel, WAKE_LIMITS } from "./node.js";
 import { captureWakeIdentity } from "./wake-identity.js";
 import { policyBrief } from "./policy.js";
 import { ulid } from "./crypto.js";
@@ -165,6 +165,103 @@ export function queuedBehindTurnAt(store, agent, messageId) {
         return null;
     }
 }
+// ---- T667: a watcher-exit hint dropped mid-turn --------------------------------------------------
+/** T667: the watcher printed a wake hint naming these deliveries. Observation is unconfirmed: the harness
+ *  drops a background-task completion that lands mid-turn. If nothing ever surfaces the mail, the daemon
+ *  requeues it once (requeueWatchReported); the Stop hook reconciles the marker for mail a turn surfaced. */
+export const watchReportedKey = (mailbox) => `watch-reported:${mailbox}`;
+const WATCH_REPORTED_CAP = 100;
+export function noteWatchReported(store, mailbox, ids, at = new Date().toISOString()) {
+    let map = {};
+    try {
+        map = JSON.parse(store.get(watchReportedKey(mailbox)) ?? "{}");
+    }
+    catch {
+        map = {};
+    }
+    for (const id of ids) {
+        delete map[id];
+        map[id] = at;
+    }
+    const keys = Object.keys(map);
+    for (const id of keys.slice(0, Math.max(0, keys.length - WATCH_REPORTED_CAP)))
+        delete map[id];
+    store.set(watchReportedKey(mailbox), JSON.stringify(map));
+}
+/** Drop marker ids a turn surfaced (the Stop continuation takes them) or that are no longer unread. */
+export function reconcileWatchReported(node, agent, surfacedIds) {
+    const key = watchReportedKey(agent), drop = new Set(surfacedIds);
+    let map = {};
+    try {
+        map = JSON.parse(node.store.get(key) ?? "{}");
+    }
+    catch {
+        return;
+    }
+    const state = node.store.db.prepare("SELECT state FROM deliveries WHERE msg_id=? AND agent=?");
+    let changed = false;
+    for (const id of Object.keys(map)) {
+        const row = state.get(id, agent);
+        if (drop.has(id) || !row || row.state === "read" || row.state === "acked" || row.state === "delivered") {
+            delete map[id];
+            changed = true;
+        }
+    }
+    if (!changed)
+        return;
+    if (Object.keys(map).length)
+        node.store.set(key, JSON.stringify(map));
+    else
+        node.store.db.prepare("DELETE FROM kv WHERE k=?").run(key);
+}
+/** One re-drive per message: a watcher-reported delivery nothing surfaced goes back to 'delivered' once,
+ *  so the normal dispatch checks (brake, mute, liveWatcher, wantsWake) drive it again instead of leaving it
+ *  stranded as 'notified' after the dropped exit. Gated so the usual surfaces never double-fire: a live
+ *  watcher re-reports delivered mail itself, a mute forbids hints and notices, and the wake brake paces the
+ *  requeue past the batching window the watcher's own exit spent. Exactly once: the id leaves the marker
+ *  with the requeue, so a handled (read/acked) message is never woken again. */
+export function requeueWatchReported(node, now = Date.now()) {
+    const since = new Date(now - WAKE_LIMITS.perAgentSeconds * 1000).toISOString();
+    for (const { k, v } of node.store.db.prepare("SELECT k, v FROM kv WHERE k GLOB 'watch-reported:*'").all()) {
+        const mailbox = k.slice(watchReportedKey("").length);
+        let map = {};
+        try {
+            map = JSON.parse(v);
+        }
+        catch {
+            node.store.db.prepare("DELETE FROM kv WHERE k=?").run(k);
+            continue;
+        }
+        if (liveWatcher(node, mailbox, now))
+            continue; // the watcher chain re-reports delivered mail itself
+        if (wakeMutedUntil(node, mailbox, now))
+            continue; // no hint and no notice until the mute ends; stays marked
+        if (node.store.db.prepare("SELECT count(*) n FROM wakes WHERE agent=? AND at>?").get(mailbox, since).n)
+            continue; // the brake paces the requeue
+        const delivery = node.store.db.prepare("SELECT state, note FROM deliveries WHERE msg_id=? AND agent=?");
+        const requeued = [];
+        node.store.tx(() => {
+            for (const id of Object.keys(map)) {
+                const d = delivery.get(id, mailbox);
+                if (!d || d.state === "read" || d.state === "acked" || d.state === "delivered") {
+                    delete map[id];
+                    continue;
+                }
+                if (d.state === "notified" && d.note === null) {
+                    node.store.db.prepare("UPDATE deliveries SET state='delivered', note=NULL, updated_at=? WHERE msg_id=? AND agent=?").run(new Date(now).toISOString(), id, mailbox);
+                    requeued.push(id);
+                    delete map[id];
+                }
+            }
+            if (Object.keys(map).length)
+                node.store.set(k, JSON.stringify(map));
+            else
+                node.store.db.prepare("DELETE FROM kv WHERE k=?").run(k);
+        });
+        if (requeued.length)
+            node.store.audit("wake.reenqueue", { agent: mailbox, ids: requeued.join(","), count: requeued.length });
+    }
+}
 /** T524: why a session that the shared service does not host is never pushed through it. */
 /** T646: the same session is also loaded in a standalone serve whose plugin could not take the wake right now. */
 export const OPENCODE_DUAL_HOST_NO_PUSH = "session is loaded by a standalone OpenCode serve as well as the shared service; not posting to the service, which would start a duplicate agent loop";
@@ -523,6 +620,7 @@ function noticeAllowed(node, agent, now) {
 export async function dispatchWakes(node, now = Date.now()) {
     const codexBlocked = await reconcileCodexQueue(node.store, CODEX(), p => keepCodexNotice(node, p));
     reconcileWakes(node, now);
+    requeueWatchReported(node, now); // T667: a watcher-exit hint dropped mid-turn leaves mail 'notified'; requeue it once
     // T557: a project mailbox slot has no session of its own. Its mail is woken for the project's CURRENT lead on this host (the subject: whose
     // session, policy, mute and batching window apply) while every delivery operation stays on the slot (the row's own mailbox, r.agent).
     // A slot with no such lead (unallocated, none designated, designated elsewhere, expired, revoked) is left alone: it stays delivered.

src/cli.ts

@@ -38,7 +38,7 @@ import { periodicUpdateCheck, updateAvailable, updateCommand } from "./update.ts
 import { installKind, version } from "./version.ts";
 import { installService, serviceLabel, uninstallService } from "./service.ts";
 import { CLIS, defaultHostName, defaultWhich, formatRows, ownerStep, resolveCommand, runSetup, selfHealSkill, shJoin, type SetupCtx } from "./setup.ts";
-import { dispatchWakes, hasWakeAuthority, humanPromptKey, inboxCommand, isHumanPrompt, macNotifierPath, muteWakes, notifyDesktop, opencodeService, liveWatcher, wakeMutedUntil, wakeText, watcherKey, which } from "./wake.ts";
+import { dispatchWakes, hasWakeAuthority, humanPromptKey, inboxCommand, isHumanPrompt, macNotifierPath, muteWakes, notifyDesktop, opencodeService, liveWatcher, wakeMutedUntil, wakeText, watcherKey, which, noteWatchReported, reconcileWatchReported } from "./wake.ts";
 import { kimiMultiHost } from "./kimi-web.ts";
 import { bindInstruction, issueBindTicket } from "./bind-ticket.ts";
 import { activityKey } from "./identity-availability.ts";
@@ -1787,6 +1787,9 @@ async function hook(node: MbxNode, event: string | undefined, cli: string) {
         const digest = stopDigest();
         const fresh = node.inbox(agent, { limit: 50 })
           .filter(m => m.state !== "read" && m.received_at > seen && m.from_addr !== `${agent}@${node.host}` && node.wantsWake(agent, m) && hasWakeAuthority(node, agent, m) && (!digest || WAKE_KINDS.has(m.kind)));
+        // T667: this turn ended. The mail this Stop surfaces rides the continuation, and read/acked mail is
+        // handled: both leave the dropped-exit marker so the daemon's one re-drive never double-fires.
+        reconcileWatchReported(node, agent, fresh.map(m => m.id));
         // T435: mail stays first. After it is consumed, one continuation re-arms a down watcher.
         // Mail that arrives after the watcher exits and before the next tool or Stop still only
         // gets a desktop notice — compaction does not fire Stop, and this gate cannot close that gap.
@@ -2019,6 +2022,9 @@ async function watch(node: MbxNode, selection: CliIdentitySelection) {
           const linked = box ? ` Read it in mbx_inbox (project section) and acknowledge with mbx_ack project=${box.alias ?? box.slot}.`
             : mailbox === agent ? "" : ` This is your linked mailbox: use agentmbx inbox --as ${mailbox} and agentmbx ack --as ${mailbox} <id>.`;
           if (!reservation.brake) lines.push(wakeText(subject, wanted, box ? projectLabel(box) : undefined) + linked);
+          // T667: the printed hint may be dropped mid-turn (the harness does not queue the completion).
+          // Mark the ids so the daemon requeues the mail once if nothing ever surfaces it.
+          if (!reservation.brake) noteWatchReported(node.store, mailbox, wanted.map(r => r.id));
         }
         for (const r of rows) {
           node.setDelivery(r.id, mailbox, "notified");