From 6f38782c3fd19fb40b82c1421621d3e80fe6496f Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 09:17:45 -0700 Subject: [PATCH 1/3] fix(T667): requeue a watcher-exit wake dropped mid-turn instead of stranding it --- dist/cli.js | 9 ++++- dist/wake.js | 100 ++++++++++++++++++++++++++++++++++++++++++++++++++- src/cli.ts | 8 ++++- src/wake.ts | 66 +++++++++++++++++++++++++++++++++- 4 files changed, 179 insertions(+), 4 deletions(-) diff --git a/dist/cli.js b/dist/cli.js index 5ccb59c..2ec5e0e 100644 --- a/dist/cli.js +++ b/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} .`; 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"); diff --git a/dist/wake.js b/dist/wake.js index c50c17f..1d34461 100644 --- a/dist/wake.js +++ b/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. diff --git a/src/cli.ts b/src/cli.ts index a464c72..ce1c343 100644 --- a/src/cli.ts +++ b/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} .`; 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"); diff --git a/src/wake.ts b/src/wake.ts index c185094..321b193 100644 --- a/src/wake.ts +++ b/src/wake.ts @@ -7,7 +7,7 @@ import { join } from "node:path"; import { promisify } from "node:util"; import { kimiHostedServer, type KimiServer } from "./kimi-web.ts"; import { desktopRpc, kimiDesktop, RpcError, type KimiDesktop } from "./kimi-desktop.ts"; -import { GROK_NO_PUSH, MbxNode, trustLabel } from "./node.ts"; +import { GROK_NO_PUSH, MbxNode, trustLabel, WAKE_LIMITS } from "./node.ts"; import { captureWakeIdentity } from "./wake-identity.ts"; import { policyBrief } from "./policy.ts"; import { ulid } from "./crypto.ts"; @@ -160,6 +160,69 @@ export function queuedBehindTurnAt(store: Store, agent: string, messageId: strin try { return (JSON.parse(store.get(behindTurnKey(agent)) ?? "{}") as Record)[messageId] ?? null; } catch { 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: string) => `watch-reported:${mailbox}`; +const WATCH_REPORTED_CAP = 100; +export function noteWatchReported(store: Store, mailbox: string, ids: readonly string[], at = new Date().toISOString()): void { + let map: Record = {}; + try { map = JSON.parse(store.get(watchReportedKey(mailbox)) ?? "{}") as Record; } 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: MbxNode, agent: string, surfacedIds: readonly string[]): void { + const key = watchReportedKey(agent), drop = new Set(surfacedIds); + let map: Record = {}; + try { map = JSON.parse(node.store.get(key) ?? "{}") as Record; } 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) as { state: string } | undefined; + 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: MbxNode, now = Date.now()): void { + 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() as { k: string; v: string }[]) { + const mailbox = k.slice(watchReportedKey("").length); + let map: Record = {}; + try { map = JSON.parse(v) as Record; } 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) as { n: number }).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: string[] = []; + node.store.tx(() => { + for (const id of Object.keys(map)) { + const d = delivery.get(id, mailbox) as { state: string; note: string | null } | undefined; + 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"; @@ -475,6 +538,7 @@ function noticeAllowed(node: MbxNode, agent: string, now: number): boolean { export async function dispatchWakes(node: MbxNode, now = Date.now()): Promise<{ agent: string; result: WakeResult | { ok: false; via: "brake"; error: string } }[]> { 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. From 9bc6e5f2af735a4d4c8774eb09843e5e64dc789b Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 09:17:50 -0700 Subject: [PATCH 2/3] test(T667): mid-turn wake-drop reproduction, re-drive once, no double-surface with Stop --- test/project-mailbox.test.ts | 2 +- test/watch-midturn.test.ts | 201 +++++++++++++++++++++++++++++++++++ 2 files changed, 202 insertions(+), 1 deletion(-) create mode 100644 test/watch-midturn.test.ts diff --git a/test/project-mailbox.test.ts b/test/project-mailbox.test.ts index 5115613..2a12e0c 100644 --- a/test/project-mailbox.test.ts +++ b/test/project-mailbox.test.ts @@ -454,7 +454,7 @@ const RATCHET: Record "src/store.ts": { max: 3, category: "schema", why: "triggers and the legacy visibility backfill" }, "src/stranded.ts": { max: 3, category: "enumerator (excludes project slots, asserted in the enumeration contract)", why: "established-name gated; a slot is recognised, not claimed" }, "src/sync-daemon.ts": { max: 2, category: "enumerator (excludes project slots, asserted in the enumeration contract)", why: "cloud read model" }, - "src/wake.ts": { max: 5, category: "slot-aware (maps a project slot to its lead; test/wake-project-mail.test.ts)", why: "the dispatcher keys a slot row to the project's current lead (T557) and operates on the row's own mailbox; reconcile works per (message, mailbox) pair; T679 keepCodexNotice reads one row by (msg_id, agent) to decide whether an owned Codex queue notice is still due" }, + "src/wake.ts": { max: 8, category: "slot-aware (maps a project slot to its lead; test/wake-project-mail.test.ts)", why: "the dispatcher keys a slot row to the project's current lead (T557) and operates on the row's own mailbox; reconcile works per (message, mailbox) pair; T679 keepCodexNotice reads one row by (msg_id, agent) to decide whether an owned Codex queue notice is still due; T667 noteWatchReported/requeueWatchReported read and requeue one row by (msg_id, agent), never enumerate names" }, }; test("T493 ratchet: every query over deliveries is classified, so a new one cannot silently list a project mailbox slot", () => { diff --git a/test/watch-midturn.test.ts b/test/watch-midturn.test.ts new file mode 100644 index 0000000..546e8b1 --- /dev/null +++ b/test/watch-midturn.test.ts @@ -0,0 +1,201 @@ +// T667: a watcher-exit wake that lands while a Kimi turn is running is dropped by the harness (the +// background-task completion is not queued). The wake must not be lost with it: the pending mail +// surfaces when the current turn ends (the Stop continuation) or at the next prompt. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { spawn, spawnSync } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"; +import { MbxNode } from "../src/node.ts"; +import { dispatchWakes, liveWatcher } from "../src/wake.ts"; +import { delegateWake } from "./helpers/wake-lease.ts"; +import { sendLeased } from "./helpers/leased-send.ts"; + +const until = async (ok: () => boolean, ms = 10_000) => { for (const end = Date.now() + ms; !ok() && Date.now() < end;) await new Promise(r => setTimeout(r, 50)); return ok(); }; + +type HookResult = { status: number | null; stdout: string; stderr: string }; +type World = { n: MbxNode; c: Client; home: string; env: Record; hook: (event: string, extra?: Record, envExtra?: Record) => HookResult }; +async function world(t: { after: (fn: () => unknown) => unknown }): Promise { + const home = mkdtempSync(join(tmpdir(), "mbx-midturn-")); + const n = new MbxNode(home, { host: "alpha" }), c = new Client({ name: "midturn", version: "1" }); + t.after(async () => { await c.close(); n.close(); rmSync(home, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }); }); + const env = { ...process.env, AGENTMBX_DEV: "1", MBX_HOME: home, MBX_CLI: "kimi", MBX_AGENT: "worker", MBX_NO_DESKTOP: "1", MBX_DEBUG: "" } as Record; + await c.connect(new StdioClientTransport({ command: process.execPath, args: [resolve("bin/agentmbx.js"), "mcp"], env })); + delegateWake(n, "worker"); + const hook = (event: string, extra: Record = {}, envExtra: Record = {}): HookResult => { + const r = spawnSync(process.execPath, [resolve("bin/agentmbx.js"), "hook", event, "--cli", "kimi"], + { input: JSON.stringify({ session_id: "kimi-midturn-1", cwd: process.cwd(), ...extra }), encoding: "utf8", timeout: 10_000, + env: { ...env, MBX_AGENT: "unrelated", ...envExtra } }); + return { status: r.status, stdout: r.stdout ?? "", stderr: r.stderr ?? "" }; + }; + const bind = hook("prompt", { prompt: "hi" }); + assert.equal(bind.status, 0, bind.stderr); + return { n, c, home, env, hook }; +} + +const arm = (w: World, t: { after: (fn: () => unknown) => unknown }) => { + const child = spawn(process.execPath, [resolve("bin/agentmbx.js"), "watch", "--cli", "kimi", "--session", "kimi-midturn-1"], + { env: { ...w.env, MBX_AGENT: "", MBX_WATCH_INTERVAL_MS: "200" } }); + let hint = ""; child.stdout.on("data", d => { hint += d; }); + const exited = new Promise(r => child.on("exit", r)); + t.after(() => { child.kill(); }); + return { child, hint: () => hint, exited }; +}; + +test("T667 AC1: a watcher-exit wake dropped mid-turn starts no turn of its own and the daemon does not re-drive it", async (t) => { + const w = await world(t); + const watcher = arm(w, t); + assert.ok(await until(() => liveWatcher(w.n, "worker")), "the watcher registers"); + const m = sendLeased(w.n, { from: "boss", to: ["worker"], subject: "midturn", body: "secret body", kind: "request", needs_reply: true }).envelope; + assert.equal(await watcher.exited, 0, "the watcher exits on the wanted mail"); + assert.match(watcher.hint(), /^\[mbx\] 1 new message\(s\) for worker/); + assert.ok(watcher.hint().includes(m.id) && !watcher.hint().includes("secret body"), "ids only, never bodies"); + // The Kimi harness drops the background-task completion that arrives mid-turn: this test asserts + // the store side of that drop — the wake was spent on an exit nobody observed. + assert.equal(liveWatcher(w.n, "worker"), false, "an exited watcher no longer holds wakes back"); + const state = () => (w.n.store.db.prepare("SELECT state, note FROM deliveries WHERE msg_id=? AND agent='worker'").get(m.id) as { state: string; note: string | null }).state; + assert.equal(state(), "notified", "the watcher reported the mail"); + const pass = await dispatchWakes(w.n); + assert.equal(pass.filter(r => r.agent === "worker").length, 0, "the daemon does not re-drive a wake the watcher already spent"); + assert.equal(state(), "notified", "the dropped wake left the mail reported, not requeued"); +}); + +test("T667 AC2: the dropped wake surfaces when the running turn ends (Stop continues exactly once), then stays quiet once handled", async (t) => { + const w = await world(t); + const watcher = arm(w, t); + assert.ok(await until(() => liveWatcher(w.n, "worker")), "the watcher registers"); + const m = sendLeased(w.n, { from: "boss", to: ["worker"], subject: "midturn", body: "secret body", kind: "request", needs_reply: true }).envelope; + assert.equal(await watcher.exited, 0, "the watcher exits; the harness drops the mid-turn completion"); + // The running turn ends: Stop must continue it for the dropped wake, naming the sender (ids stay in the watcher hint). + const stop = w.hook("stop"); + assert.equal(stop.status, 2, `Stop continues the turn: ${stop.stderr}`); + assert.match(stop.stderr, /1 new message\(s\) for worker from boss@alpha arrived while you worked/); + assert.doesNotMatch(stop.stderr, /secret body/, "never bodies"); + // The agent handled the mail in the continued turn (read, not yet acked): the next Stop stays quiet. + w.n.store.db.prepare("UPDATE deliveries SET state='read' WHERE msg_id=? AND agent='worker'").run(m.id); + const quiet = w.hook("stop"); + assert.equal(quiet.status, 0, quiet.stderr); + assert.equal(quiet.stderr, "", "handled mail does not block Stop twice"); +}); + +test("T667 AC2: with the Stop digest on, mid-turn mail Stop does not cover surfaces at the next prompt instead", async (t) => { + const w = await world(t); + const watcher = arm(w, t); + assert.ok(await until(() => liveWatcher(w.n, "worker")), "the watcher registers"); + const m = sendLeased(w.n, { from: "boss", to: ["worker"], subject: "midturn", body: "secret body", kind: "message", needs_reply: true }).envelope; + assert.equal(await watcher.exited, 0, "the watcher exits; the harness drops the mid-turn completion"); + // Digest mode: only request/task/decision/alert continue Stop; a needs-reply message waits for the prompt. + // The exit 2 here is the bounded watcher re-arm nag (no live watcher), not a mail continuation. + const stop = w.hook("stop", {}, { MBX_STOP_DIGEST: "1" }); + assert.equal(stop.status, 2, `the watcher nag exits 2: ${stop.stderr}`); + assert.doesNotMatch(stop.stderr, /arrived while you worked/, "digest Stop does not continue the turn for a message kind"); + const prompt = w.hook("prompt", { prompt: "the owner typed a new turn" }); + assert.equal(prompt.status, 0, prompt.stderr); + assert.match(prompt.stdout + prompt.stderr, /1 unread for worker@alpha: mbx_inbox/, "the next prompt names the still-unread mail"); + assert.ok(!(prompt.stdout + prompt.stderr).includes("secret body"), "never bodies"); +}); + +test("T667: mail landing during an exit-2 continuation turn is requeued once and reaches the next turn without an owner prompt", { timeout: 90_000 }, async (t) => { + const w = await world(t); + const marker = () => w.n.store.get("watch-reported:worker") ?? null; + const audit = () => (w.n.store.db.prepare("SELECT detail FROM audit WHERE event='wake.reenqueue'").all() as { detail: string }[]).length; + const state = (id: string) => (w.n.store.db.prepare("SELECT state FROM deliveries WHERE msg_id=? AND agent='worker'").get(id) as { state: string }).state; + + // m1 lands mid-turn; its watcher exit is dropped; the turn ends and Stop continues for it. + const w1 = arm(w, t); + assert.ok(await until(() => liveWatcher(w.n, "worker"))); + const m1 = sendLeased(w.n, { from: "boss", to: ["worker"], subject: "one", body: "s1", kind: "request", needs_reply: true }).envelope; + assert.equal(await w1.exited, 0, "w1 exits on m1; the harness drops the completion"); + assert.ok(marker(), "the dropped exit left a marker"); + const stop1 = w.hook("stop"); + assert.equal(stop1.status, 2, "Stop continues for m1"); + assert.equal(state(m1.id), "notified"); + // the continuation turn handled m1 and re-armed the watcher; m2 lands mid-continuation and its exit is dropped too + w.n.store.db.prepare("UPDATE deliveries SET state='read' WHERE msg_id=? AND agent='worker'").run(m1.id); + assert.equal(marker(), null, "the Stop that surfaced m1 reconciled the marker"); + const w2 = arm(w, t); + assert.ok(await until(() => liveWatcher(w.n, "worker"))); + const m2 = sendLeased(w.n, { from: "boss", to: ["worker"], subject: "two", body: "s2", kind: "request", needs_reply: true }).envelope; + assert.equal(await w2.exited, 0, "w2 exits on m2; the harness drops the completion (no second Stop fires)"); + + // the turn ended without surfacing m2; once the brake window the watcher's exit spent has passed, + // the daemon requeues m2 once, and the session's next watcher reports it: no owner prompt involved. + w.n.store.db.prepare("DELETE FROM wakes").run(); // test-only: expire the 30 s batching window + const pass = await dispatchWakes(w.n); + assert.equal(audit(), 1, "exactly one re-drive"); + assert.equal(marker(), null, "the requeue consumed the marker"); + assert.ok(state(m2.id) === "delivered" || state(m2.id) === "notified", "m2 re-entered the normal wake pool"); + w.n.store.db.prepare("DELETE FROM wakes").run(); // the re-drive pass spent its own brake window; let the next watcher report now + const w3 = arm(w, t); + const deadline = Date.now() + 10_000; + let hint = ""; + w3.child.stdout.on("data", d => { hint += d; }); + while (Date.now() < deadline && !hint.includes(m2.id)) await new Promise(r => setTimeout(r, 100)); + assert.ok(hint.includes(m2.id), `the next watcher reports the requeued mail: ${JSON.stringify(hint)}`); + assert.ok(!hint.includes("s2"), "ids only, never bodies"); + assert.equal(await w3.exited, 0, "the reporting watcher exits to start that turn"); +}); + +test("T667: exactly one re-drive per message, and none while a live watcher would report it", { timeout: 90_000 }, async (t) => { + const w = await world(t); + const audit = () => (w.n.store.db.prepare("SELECT detail FROM audit WHERE event='wake.reenqueue'").all() as { detail: string }[]).length; + const state = (id: string) => (w.n.store.db.prepare("SELECT state FROM deliveries WHERE msg_id=? AND agent='worker'").get(id) as { state: string }).state; + + const w1 = arm(w, t); + assert.ok(await until(() => liveWatcher(w.n, "worker"))); + const m = sendLeased(w.n, { from: "boss", to: ["worker"], subject: "one", body: "s", kind: "request", needs_reply: true }).envelope; + assert.equal(await w1.exited, 0, "w1 exits; the completion is dropped"); + + // a live watcher holds the mail back: the daemon defers, no re-drive + const w2 = arm(w, t); + assert.ok(await until(() => liveWatcher(w.n, "worker"))); + w.n.store.db.prepare("DELETE FROM wakes").run(); + await dispatchWakes(w.n); + assert.equal(audit(), 0, "no re-drive while a live watcher owns the wake path"); + assert.equal(state(m.id), "notified", "the mail stays reported, not flipped under a live watcher"); + w2.child.kill(); + + // watcher gone, brake window expired: the first pass requeues exactly once + assert.ok(await until(() => !liveWatcher(w.n, "worker"), 20_000)); + await dispatchWakes(w.n); + assert.equal(audit(), 1, "the first eligible pass re-drives once"); + assert.equal(w.n.store.get("watch-reported:worker") ?? null, null, "the marker is consumed with the requeue"); + const after = state(m.id); + assert.ok(after === "delivered" || after === "notified", "m re-entered the normal wake pool"); + // a later pass never re-drives the same message again + await dispatchWakes(w.n); + assert.equal(audit(), 1, "no second re-drive for the same message"); +}); + +test("T667: no re-drive for mail read or acked before the turn ended, and never alongside a Stop continuation", { timeout: 90_000 }, async (t) => { + const w = await world(t); + const audit = () => (w.n.store.db.prepare("SELECT detail FROM audit WHERE event='wake.reenqueue'").all() as { detail: string }[]).length; + + // handled mail: read before any re-drive, the marker is pruned without requeueing + const w1 = arm(w, t); + assert.ok(await until(() => liveWatcher(w.n, "worker"))); + const m1 = sendLeased(w.n, { from: "boss", to: ["worker"], subject: "read case", body: "s", kind: "request", needs_reply: true }).envelope; + assert.equal(await w1.exited, 0); + w.n.store.db.prepare("UPDATE deliveries SET state='read' WHERE msg_id=? AND agent='worker'").run(m1.id); + w.n.store.db.prepare("DELETE FROM wakes").run(); + await dispatchWakes(w.n); + assert.equal(audit(), 0, "read mail is not re-driven"); + assert.equal(w.n.store.get("watch-reported:worker") ?? null, null, "read mail left the marker"); + + // Stop-surfaced mail: the continuation owns it; the daemon must not also re-drive it + const w2 = arm(w, t); + assert.ok(await until(() => liveWatcher(w.n, "worker"))); + const m2 = sendLeased(w.n, { from: "boss", to: ["worker"], subject: "stop case", body: "s", kind: "request", needs_reply: true }).envelope; + assert.equal(await w2.exited, 0); + assert.ok(w.n.store.get("watch-reported:worker"), "m2 is marked after its dropped exit"); + const stop = w.hook("stop"); + assert.equal(stop.status, 2, "Stop continues for m2: the single surface"); + assert.equal(w.n.store.get("watch-reported:worker") ?? null, null, "surfacing m2 reconciled the marker"); + w.n.store.db.prepare("DELETE FROM wakes").run(); + await dispatchWakes(w.n); + assert.equal(audit(), 0, "the re-drive never double-fires alongside the Stop continuation"); +}); + From 0720eb3bf4fe008f43ad1770401865077c2ce954 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 09:17:53 -0700 Subject: [PATCH 3/3] docs(T667): changelog entry for the mid-turn wake requeue --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a06257f..6627ff5 100644 --- a/CHANGELOG.md +++ b/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.