From 17516378c2f3348c026f28db91f6ea536dc576cc Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 08:21:57 -0700 Subject: [PATCH 01/12] fix(T696): align skill relay depth with policy limits --- CHANGELOG.md | 2 ++ skill/SKILL.md | 2 +- test/skill-relay-depth.test.ts | 11 +++++++++++ 3 files changed, 14 insertions(+), 1 deletion(-) create mode 100644 test/skill-relay-depth.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index a06257f..e9e40f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- **Skill relay-depth guidance matches policy enforcement (T696).** The skill now says ask allows 6 hops and collaborate, autonomous and yolo have no limit. A regression test checks the documented limits against `LEVEL_MAX_HOP`. + ## 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. diff --git a/skill/SKILL.md b/skill/SKILL.md index 635f824..ffdc4ae 100644 --- a/skill/SKILL.md +++ b/skill/SKILL.md @@ -250,7 +250,7 @@ wake only queues a short pointer to the inbox; it never means the recipient read - Check the receipt's `recipients[].state` (`live-wake` or `live-next-prompt`) rather than assuming the send woke anyone. Relay depth counts only what you read from agents other than the recipients: answering the agent you heard from never -adds depth, forwarding to someone else does (ask allows 6, collaborate 20, autonomous/yolo no limit; the policy line shows it). +adds depth, forwarding to someone else does (ask 6, collaborate no limit, autonomous no limit, yolo no limit; the policy line shows it). ## Autonomy probes (`[mbx-probe]`) diff --git a/test/skill-relay-depth.test.ts b/test/skill-relay-depth.test.ts new file mode 100644 index 0000000..e5a3818 --- /dev/null +++ b/test/skill-relay-depth.test.ts @@ -0,0 +1,11 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { LEVEL_MAX_HOP } from "../src/policy.ts"; + +test("the skill relay-depth limits agree with policy enforcement", () => { + const skill = readFileSync(new URL("../skill/SKILL.md", import.meta.url), "utf8"); + const limits = Object.entries(LEVEL_MAX_HOP) + .map(([level, limit]) => `${level} ${Number.isFinite(limit) ? limit : "no limit"}`).join(", "); + assert.ok(skill.includes(`(${limits}; the policy line shows it).`), `skill must document: ${limits}`); +}); From c346276467f330eed0958595f0b6a8175ec06ab0 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 08:24:25 -0700 Subject: [PATCH 02/12] docs(T697): update OpenCode wake and eviction guidance --- CHANGELOG.md | 2 ++ README.md | 9 ++++++++- docs/PORTABILITY.md | 9 ++++++++- 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a06257f..02914ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- **OpenCode wake documentation matches the shipped adapters (T697).** README and portability guidance describe standalone `resume: true` wakes, one shared service per machine for long-lived agents, and standalone/service-hosted recovery after MCP eviction (T659/T688). + ## 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. diff --git a/README.md b/README.md index afff988..26cab88 100644 --- a/README.md +++ b/README.md @@ -148,13 +148,20 @@ AgentMBX gives every agent the same small set of mailbox tools. It delivers mess | CLI | Wake path | Status | |---|---|---| | Codex | `codex app-server` experimental `thread/queue/add`, `thread/queue/list`, and `thread/queue/delete`; AgentMBX tracks its owned notice and withdraws it after read/ack, release, or handover while still queued, preserving unowned items | requires a Codex build with the experimental queue API; historical live receipts | - | OpenCode | service-hosted session: the local service's session API (`/synthetic`). `--standalone` session: the plugin inside that session's own serve, over a loopback queue (see [OPENCODE.md](docs/OPENCODE.md)) | historical service-path live receipts; standalone wakes are admitted in-process, whether one starts a turn on an idle session is unproven (T544) | + | OpenCode | service-hosted session: the local service's session API (`/synthetic`, `resume: true`). `--standalone` session: its own plugin/loopback queue, also `resume: true` (T544; see [OPENCODE.md](docs/OPENCODE.md)) | recorded service-path live receipts and an idle standalone turn (T544); admission alone does not prove model execution | | Claude Code | pushed by the session's own mbx MCP server through Claude Code's per-session inbox socket (`CLAUDE_CODE_MESSAGING_SOCKET`), so a plainly started `claude` wakes on mail with no flag, setting or cron; `agentmbx claude [args]` adds the research-preview mbx channel instead | historical macOS/Fedora live receipts | | Kimi | desktop app: its local control socket (setup installs an AgentMBX plugin into the app). `kimi web`: the local server's prompts API. Terminal: a background `agentmbx watch` task the session keeps running (Kimi starts a turn when it exits) | historical terminal/desktop/web live receipts | | Hermes | one background `agentmbx watch` task with completion notification; shell hooks bind the session and add inbox context before model calls | watcher requires re-arming after each wake; shared gateway hosts aren't covered | | Grok | a background `agentmbx watch --cli grok --session ` task; exact session hooks | watcher requires re-arming after each wake | | anything else | desktop notification | no automatic model turn implied | + For long-lived OpenCode agents, run plain `opencode` against the shared service, started once with + `opencode service start`: one service per machine, with each session keeping its own project folder. + After OpenCode evicts an idle session's MCP, standalone wakes remain available while that serve's plugin + polls the local daemon (T659); service-hosted wakes remain available while the recorded service process + is alive (T688). Both paths require the recorded PID and process birth, and a lease held by a different + session blocks the old binding. See [the run-mode and eviction guide](docs/OPENCODE.md#after-opencode-evicts-the-sessions-mcp). + - **Across machines:** a small daemon per host. Hosts pair with one command each: `agentmbx pair` prints a one-time token, `agentmbx join ` on the other machine finishes it (or compare a 6-digit code instead). Hosts find each other on the LAN over mDNS. Every hop is signed. Messages to a sleeping machine wait in an outbox and retry for 72 h, and each is stored exactly once. - **Wake batching:** one atomic 30-second window per agent, plus explicit wake mute. Status messages never wake anyone. The published runtime has no hourly per-thread or daily per-agent hard cap; loop detection is observational, so batching alone doesn't bound an automated conversation's token use. - **Full-text search** (SQLite FTS5) and an audit log. `mbx:@` references can be cited from tickets and notes. diff --git a/docs/PORTABILITY.md b/docs/PORTABILITY.md index 057abe1..f7319e8 100644 --- a/docs/PORTABILITY.md +++ b/docs/PORTABILITY.md @@ -43,13 +43,20 @@ detection/wiring remains a separate boundary. An arbitrary MCP client can wire |---|---|---|---| | Claude Code | MCP; SessionStart, UserPromptSubmit, PermissionRequest, Stop, PostToolUse, SessionEnd | Its own MCP server sends a no-body hint through the per-session messaging socket; optional `agentmbx claude` channel launcher | Socket/channel support and exact-session binding; historical macOS/Linux receipts aren't proof of every current frontend | | Codex | MCP; SessionStart, UserPromptSubmit, PermissionRequest, Stop | `codex app-server` experimental `thread/queue/add`, `thread/queue/list`, and `thread/queue/delete`; tracks owned notices and withdraws them after read/ack, release, or handover while still queued; unowned items are preserved | Requires a Codex build with the experimental queue API and hooks. Consumed notices cannot be recalled. Startup, same-thread resume, and new-thread isolation are separate live gates | -| OpenCode | MCP; plugin session-created, post-tool, and idle hooks; sidebar plugin | Service sessions use their API; standalone sessions use their own plugin/loopback queue | Standalone in-process admission doesn't prove an idle model turn. Shared-service permission coverage doesn't establish standalone coverage | +| OpenCode | MCP; plugin session-created, post-tool, and idle hooks; sidebar plugin | Service sessions use their API; standalone sessions use their own plugin/loopback queue. Both daemon wake paths use `resume: true` (T544) | An idle standalone turn was observed with T544; a receipt alone still doesn't prove model execution. Shared-service permission coverage doesn't establish standalone coverage | | Kimi terminal | MCP, skill, SessionStart/UserPromptSubmit/PermissionRequest/Stop hooks | One background `agentmbx watch` task exits with a hint and starts the next turn | Re-arm the watcher after handling mail. `MBX_SELF_WATCH` opts into a periodic self-check. Terminal permission hooks can't approve prompts | | Kimi web/rc/desktop | MCP and hosted-session hooks; one-time conversation bind ticket | Web prompts API or desktop control socket | Restart a hosted server started before setup; verify the current exact conversation. Approval API coverage applies to hosted sessions | | Hermes terminal/TUI | MCP; `on_session_start` and `pre_llm_call` shell hooks with exact hook approvals | Background `agentmbx watch` task with completion notifications | Re-arm after each wake. Hooks load at session start; a shared gateway/desktop host isn't covered. No Hermes permission-approval adapter is claimed | | Grok | MCP, status line, SessionStart/UserPromptSubmit/PostToolUse/Stop hooks | Background `agentmbx watch --cli grok --session ` | Re-arm each completion; verify exact session ID and provider background-task notification | | Other MCP clients | Manual MCP configuration | Desktop notification fallback | No automatic model turn or permission adapter implied | +For long-lived OpenCode agents, prefer plain `opencode` against the shared service, started once with +`opencode service start`: one service per machine, not per project. After MCP eviction, standalone +bindings remain wakeable while their plugin polls the local daemon (T659); service-hosted bindings +remain wakeable while their recorded service process lives (T688). Both paths verify the recorded PID +and process birth, and refuse an old binding when a different session holds the mailbox lease. +See [OpenCode run modes and eviction recovery](OPENCODE.md#after-opencode-evicts-the-sessions-mcp). + Setup preserves other tools' hook groups, backs up files before editing, and uninstall removes AgentMBX-owned entries. Modified/foreign configuration can require manual review. See [the wake reference](SPEC.md#waking-recipients). From 2501021bc5e3b2533739d2b124a6a7c82733e2ea Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 08:34:06 -0700 Subject: [PATCH 03/12] feat(T583): show harness last activity when the mailbox is quiet A Codex rollout mtime and a live Grok session row keep a stale mbx call from reading as idle. A dead or quiet harness still falls back to the mbx time. --- CHANGELOG.md | 2 + dist/identity-availability.js | 89 ++++++++++++++++++++++++++++++++-- src/identity-availability.ts | 83 +++++++++++++++++++++++++++++-- test/liveness-activity.test.ts | 85 ++++++++++++++++++++++++++++++++ 4 files changed, 250 insertions(+), 9 deletions(-) create mode 100644 test/liveness-activity.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index a06257f..fa35eed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- **Liveness follows the harness when the mailbox has gone quiet (T583).** `mbx_agents` and `agentmbx agents` timed a live holder by its last mbx call, so a session that was busy in its harness looked stale. A Codex lease now also uses that thread's rollout mtime (file metadata only, the same one-file rule as T677). A Grok lease uses `active_sessions.json`: a live row is a heartbeat, and a dead or reused pid is not. When that stamp is the newer one the reason says `last activity N min ago (codex rollout)` or `(grok session)`; otherwise it still says `last mbx activity`. A shared conversation becomes claimable only once both are quiet. + ## 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. diff --git a/dist/identity-availability.js b/dist/identity-availability.js index c152df4..ceddf3b 100644 --- a/dist/identity-availability.js +++ b/dist/identity-availability.js @@ -1,6 +1,10 @@ // One answer to "can this identity be claimed now?", used by both mbx_identity list and the claim itself (T204, R4.3), so // a name the list calls available is never refused by the claim for a reason the list didn't show. +import { globSync, lstatSync, readFileSync } from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; import { identityLeaseStatus } from "./identity-leases.js"; +import { procStartEpochMs } from "./proc.js"; /** * A conversation of a shared provider process (Codex or OpenCode transport, hosted Kimi) that made no mbx call and fired * no hook for this long can be taken over by an explicit claim: closing such a conversation does not end its process, so @@ -20,6 +24,77 @@ export function parseActivity(raw) { } } const minutes = (ms) => `${Math.max(1, Math.round(ms / 60_000))} min`; +const CODEX_THREAD = /^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i; +/** A stamp counts only when it is a real past moment from this holder's lease, never a future or pre-lease mtime. */ +function usableStamp(at, now, claimedAt) { + return Number.isSafeInteger(at) && at >= 0 && at <= now && at > claimedAt ? at : null; +} +function fileMtime(path) { + try { + const stat = lstatSync(path); + return stat.isFile() ? Math.floor(stat.mtimeMs) : null; + } + catch { + return null; + } +} +function pidAlive(pid) { + if (!Number.isSafeInteger(pid) || pid <= 0) + return false; + try { + process.kill(pid, 0); + return true; + } + catch (e) { + return e.code === "EPERM"; + } +} +/** Exact-thread rollout mtime. Metadata only: the same one-file rule as CodexActivity, and the body is never opened. */ +function codexRolloutActivity(sessionId, home, now, claimedAt) { + if (!CODEX_THREAD.test(sessionId)) + return null; + try { + const found = globSync(`sessions/*/*/*/rollout-*-${sessionId.toLowerCase()}.jsonl`, { cwd: home }); + if (found.length !== 1) + return null; + const at = usableStamp(fileMtime(join(home, found[0])) ?? -1, now, claimedAt); + return at === null ? null : { at, source: "codex rollout" }; + } + catch { + return null; + } +} +/** A live row in Grok's active_sessions.json is a heartbeat at decision time. A dead or reused pid is not activity. */ +function grokSessionActivity(sessionId, file, now, claimedAt) { + try { + const rows = JSON.parse(readFileSync(file, "utf8")); + if (!Array.isArray(rows)) + return null; + const matches = rows.filter((r) => !!r && typeof r === "object" && r.session_id === sessionId + && Number.isSafeInteger(r.pid) && (r.pid) > 0); + if (matches.length !== 1 || !pidAlive(matches[0].pid)) + return null; + const opened = typeof matches[0].opened_at === "string" ? Date.parse(matches[0].opened_at) : NaN; + const startMs = procStartEpochMs(matches[0].pid); + if (Number.isFinite(opened) && startMs !== null && opened < startMs) + return null; + const at = usableStamp(now, now, claimedAt); + return at === null ? null : { at, source: "grok session" }; + } + catch { + return null; + } +} +function harnessActivity(lease, now, homes) { + if (lease.cli === "codex") + return codexRolloutActivity(lease.session_id, homes?.codexHome ?? (process.env.CODEX_HOME || join(homedir(), ".codex")), now, lease.claimed_at); + if (lease.cli === "grok") { + const file = homes?.grokSessionsFile ?? (process.env.MBX_GROK_SESSIONS_FILE + || join(process.env.GROK_HOME || join(homedir(), ".grok"), "active_sessions.json")); + return grokSessionActivity(lease.session_id, file, now, lease.claimed_at); + } + return null; +} export const providerRecordKey = (mcpPid) => `mcp-provider:${mcpPid}`; /** A kv value is a provider record only when both fields are present and bounded. Anything else is "no record". */ export function parseProviderRecord(raw) { @@ -105,6 +180,9 @@ export function identityAvailability(o) { const holder = `${lease.cli} session ${lease.session_id}`; if (status.state === "unknown") return { claimable: false, state: "unknown", reason: `held by ${holder}, whose process could not be verified right now; retry shortly` }; + const harness = harnessActivity(lease, o.now, o.homes); + const mbxAt = o.activity?.at ?? null; + const effectiveAt = Math.max(mbxAt ?? 0, harness?.at ?? 0); if (o.caller && lease.cli === o.caller.cli && lease.session_id === o.caller.sessionId && !o.caller.sessionId.startsWith("mcp-")) { // T383: a provider may start a second, short-lived MCP server for a session whose own server is still live and serving // (Codex did). T439: a server under the caller's own provider co-uses the lease. T440: a different parent is takeover @@ -114,7 +192,7 @@ export function identityAvailability(o) { // T608: the same conversation's other server, quiet for the stale threshold (D039), is stuck, not serving: this // session takes its own identity back without the owner. A holder still in use keeps it, so two live servers do not // trade the lease. - const sessionQuiet = o.now - (o.activity?.at ?? lease.claimed_at); + const sessionQuiet = o.now - Math.max(o.activity?.at ?? lease.claimed_at, harness?.at ?? 0); if (holderProvider !== providerPid && sessionQuiet >= SHARED_IDLE_MS) return { claimable: true, state: "idle", reason: `held by an idle MCP server of this same ${lease.cli} session (pid ${lease.holder_pid}, no mbx activity for ${minutes(sessionQuiet)})`, takeover: "same-session" }; if (providerPid !== undefined && holderProvider != null && holderProvider !== providerPid) { @@ -128,8 +206,11 @@ export function identityAvailability(o) { + "a sibling server of this session co-uses it: it acts as that identity under the holder's lease, without taking or releasing it" }; return { claimable: false, state: "live", reason: `held by a live process of this same ${lease.cli} session (pid ${lease.holder_pid}) whose provider process could not be verified; retry shortly` }; } - const quiet = o.activity ? o.now - o.activity.at : null; - if (o.activity?.shared && quiet !== null && quiet >= SHARED_IDLE_MS) + const quiet = mbxAt !== null ? o.now - mbxAt : null; + const effectiveQuiet = mbxAt !== null || harness ? o.now - effectiveAt : null; + if (o.activity?.shared && quiet !== null && quiet >= SHARED_IDLE_MS && (effectiveQuiet ?? quiet) >= SHARED_IDLE_MS) return { claimable: true, state: "idle", reason: `held by ${holder} in a shared ${lease.cli} process with no mbx activity for ${minutes(quiet)} (that conversation has probably ended)`, takeover: "idle-conversation" }; - return { claimable: false, state: "live", reason: `held by live ${holder}${quiet !== null ? `, last mbx activity ${minutes(quiet)} ago` : ""}` }; + const showHarness = harness !== null && (mbxAt === null || harness.at >= mbxAt); + const suffix = showHarness ? `, last activity ${minutes(o.now - harness.at)} ago (${harness.source})` : quiet !== null ? `, last mbx activity ${minutes(quiet)} ago` : ""; + return { claimable: false, state: "live", reason: `held by live ${holder}${suffix}` }; } diff --git a/src/identity-availability.ts b/src/identity-availability.ts index d0da9ad..7c024fb 100644 --- a/src/identity-availability.ts +++ b/src/identity-availability.ts @@ -1,6 +1,10 @@ // One answer to "can this identity be claimed now?", used by both mbx_identity list and the claim itself (T204, R4.3), so // a name the list calls available is never refused by the claim for a reason the list didn't show. +import { globSync, lstatSync, readFileSync } from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; import { identityLeaseStatus, type IdentityLease, type ProcessEvidence } from "./identity-leases.ts"; +import { procStartEpochMs } from "./proc.ts"; /** * A conversation of a shared provider process (Codex or OpenCode transport, hosted Kimi) that made no mbx call and fired @@ -35,6 +39,66 @@ export interface Availability { } const minutes = (ms: number) => `${Math.max(1, Math.round(ms / 60_000))} min`; +const CODEX_THREAD = /^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i; + +interface HarnessHomes { codexHome?: string; grokSessionsFile?: string } +interface HarnessStamp { at: number; source: string } + +/** A stamp counts only when it is a real past moment from this holder's lease, never a future or pre-lease mtime. */ +function usableStamp(at: number, now: number, claimedAt: number): number | null { + return Number.isSafeInteger(at) && at >= 0 && at <= now && at > claimedAt ? at : null; +} + +function fileMtime(path: string): number | null { + try { + const stat = lstatSync(path); + return stat.isFile() ? Math.floor(stat.mtimeMs) : null; + } catch { return null; } +} + +function pidAlive(pid: number): boolean { + if (!Number.isSafeInteger(pid) || pid <= 0) return false; + try { process.kill(pid, 0); return true; } + catch (e) { return (e as NodeJS.ErrnoException).code === "EPERM"; } +} + +/** Exact-thread rollout mtime. Metadata only: the same one-file rule as CodexActivity, and the body is never opened. */ +function codexRolloutActivity(sessionId: string, home: string, now: number, claimedAt: number): HarnessStamp | null { + if (!CODEX_THREAD.test(sessionId)) return null; + try { + const found = globSync(`sessions/*/*/*/rollout-*-${sessionId.toLowerCase()}.jsonl`, { cwd: home }); + if (found.length !== 1) return null; + const at = usableStamp(fileMtime(join(home, found[0])) ?? -1, now, claimedAt); + return at === null ? null : { at, source: "codex rollout" }; + } catch { return null; } +} + +/** A live row in Grok's active_sessions.json is a heartbeat at decision time. A dead or reused pid is not activity. */ +function grokSessionActivity(sessionId: string, file: string, now: number, claimedAt: number): HarnessStamp | null { + try { + const rows = JSON.parse(readFileSync(file, "utf8")) as unknown; + if (!Array.isArray(rows)) return null; + const matches = rows.filter((r): r is { pid: number; opened_at?: unknown } => + !!r && typeof r === "object" && (r as { session_id?: unknown }).session_id === sessionId + && Number.isSafeInteger((r as { pid?: unknown }).pid) && ((r as { pid: number }).pid) > 0); + if (matches.length !== 1 || !pidAlive(matches[0].pid)) return null; + const opened = typeof matches[0].opened_at === "string" ? Date.parse(matches[0].opened_at) : NaN; + const startMs = procStartEpochMs(matches[0].pid); + if (Number.isFinite(opened) && startMs !== null && opened < startMs) return null; + const at = usableStamp(now, now, claimedAt); + return at === null ? null : { at, source: "grok session" }; + } catch { return null; } +} + +function harnessActivity(lease: IdentityLease, now: number, homes?: HarnessHomes): HarnessStamp | null { + if (lease.cli === "codex") return codexRolloutActivity(lease.session_id, homes?.codexHome ?? (process.env.CODEX_HOME || join(homedir(), ".codex")), now, lease.claimed_at); + if (lease.cli === "grok") { + const file = homes?.grokSessionsFile ?? (process.env.MBX_GROK_SESSIONS_FILE + || join(process.env.GROK_HOME || join(homedir(), ".grok"), "active_sessions.json")); + return grokSessionActivity(lease.session_id, file, now, lease.claimed_at); + } + return null; +} /** The provider a holder recorded at claim (`mcp-provider:`). Start is the lease-layer birth time, not a proc-table field. */ export interface RecordedProvider { pid: number; start: string; harness?: true } @@ -113,7 +177,10 @@ export function identityAvailability(o: { lease?: IdentityLease; evidence: Proce * `providerPid`: the caller's provider process. `holderProviderPid`: the provider the live holder serves under. * `holderProviderAlive`: false only when that provider was confirmed dead. A pid mismatch without that proof is not takeover (T440). */ - caller?: { cli: string; sessionId: string; providerPid?: number; holderProviderPid?: number | null; holderProviderAlive?: boolean | null } }): Availability { + caller?: { cli: string; sessionId: string; providerPid?: number; holderProviderPid?: number | null; holderProviderAlive?: boolean | null }; + /** Isolated Codex home or Grok session file. Omitted, the real harness locations are read. */ + homes?: HarnessHomes; +}): Availability { if (o.conflict) return { claimable: false, state: "conflict", reason: "historical ownership requires explicit owner recovery" }; const lease = o.lease; if (!lease) return { claimable: true, state: "free", reason: "no session has held it under a lease" }; @@ -121,6 +188,9 @@ export function identityAvailability(o: { lease?: IdentityLease; evidence: Proce if (status.state === "expired") return { claimable: true, state: "offline", reason: lease.released_at !== null ? lease.release_reason ?? "released" : status.reason ?? "expired" }; const holder = `${lease.cli} session ${lease.session_id}`; if (status.state === "unknown") return { claimable: false, state: "unknown", reason: `held by ${holder}, whose process could not be verified right now; retry shortly` }; + const harness = harnessActivity(lease, o.now, o.homes); + const mbxAt = o.activity?.at ?? null; + const effectiveAt = Math.max(mbxAt ?? 0, harness?.at ?? 0); if (o.caller && lease.cli === o.caller.cli && lease.session_id === o.caller.sessionId && !o.caller.sessionId.startsWith("mcp-")) { // T383: a provider may start a second, short-lived MCP server for a session whose own server is still live and serving // (Codex did). T439: a server under the caller's own provider co-uses the lease. T440: a different parent is takeover @@ -130,7 +200,7 @@ export function identityAvailability(o: { lease?: IdentityLease; evidence: Proce // T608: the same conversation's other server, quiet for the stale threshold (D039), is stuck, not serving: this // session takes its own identity back without the owner. A holder still in use keeps it, so two live servers do not // trade the lease. - const sessionQuiet = o.now - (o.activity?.at ?? lease.claimed_at); + const sessionQuiet = o.now - Math.max(o.activity?.at ?? lease.claimed_at, harness?.at ?? 0); if (holderProvider !== providerPid && sessionQuiet >= SHARED_IDLE_MS) return { claimable: true, state: "idle", reason: `held by an idle MCP server of this same ${lease.cli} session (pid ${lease.holder_pid}, no mbx activity for ${minutes(sessionQuiet)})`, takeover: "same-session" }; if (providerPid !== undefined && holderProvider != null && holderProvider !== providerPid) { @@ -144,8 +214,11 @@ export function identityAvailability(o: { lease?: IdentityLease; evidence: Proce + "a sibling server of this session co-uses it: it acts as that identity under the holder's lease, without taking or releasing it" }; return { claimable: false, state: "live", reason: `held by a live process of this same ${lease.cli} session (pid ${lease.holder_pid}) whose provider process could not be verified; retry shortly` }; } - const quiet = o.activity ? o.now - o.activity.at : null; - if (o.activity?.shared && quiet !== null && quiet >= SHARED_IDLE_MS) + const quiet = mbxAt !== null ? o.now - mbxAt : null; + const effectiveQuiet = mbxAt !== null || harness ? o.now - effectiveAt : null; + if (o.activity?.shared && quiet !== null && quiet >= SHARED_IDLE_MS && (effectiveQuiet ?? quiet) >= SHARED_IDLE_MS) return { claimable: true, state: "idle", reason: `held by ${holder} in a shared ${lease.cli} process with no mbx activity for ${minutes(quiet)} (that conversation has probably ended)`, takeover: "idle-conversation" }; - return { claimable: false, state: "live", reason: `held by live ${holder}${quiet !== null ? `, last mbx activity ${minutes(quiet)} ago` : ""}` }; + const showHarness = harness !== null && (mbxAt === null || harness.at >= mbxAt); + const suffix = showHarness ? `, last activity ${minutes(o.now - harness.at)} ago (${harness.source})` : quiet !== null ? `, last mbx activity ${minutes(quiet)} ago` : ""; + return { claimable: false, state: "live", reason: `held by live ${holder}${suffix}` }; } diff --git a/test/liveness-activity.test.ts b/test/liveness-activity.test.ts new file mode 100644 index 0000000..1904c54 --- /dev/null +++ b/test/liveness-activity.test.ts @@ -0,0 +1,85 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { mkdirSync, mkdtempSync, rmSync, utimesSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { identityAvailability, SHARED_IDLE_MS } from "../src/identity-availability.ts"; +import type { IdentityLease } from "../src/identity-leases.ts"; + +const THREAD = "44444444-4444-4444-8444-444444444444"; +const live = { alive: true as const, start: "birth" }; + +function lease(cli: string, sessionId: string, now: number): IdentityLease { + return { + name: "reader", token: "t", holder_pid: 10, holder_start: "birth", key_fp: "aaaa-aaaa-aaaa-aaaa", + cli, session_id: sessionId, claimed_at: now - SHARED_IDLE_MS - 60_000, heartbeat_at: now, + idle_ttl: 30 * 60_000, released_at: null, release_reason: null, + }; +} + +function dir(): string { + const home = mkdtempSync(join(tmpdir(), "mbx-liveness-")); + return home; +} + +test("a writing Codex thread with a stale mbx call reads as active", () => { + const now = Date.now(), home = dir(), at = now - 1000; + const day = join(home, "sessions", "2026", "10", "17"); + mkdirSync(day, { recursive: true }); + const path = join(day, `rollout-2026-10-17T12-00-00-${THREAD}.jsonl`); + writeFileSync(path, "metadata only", { mode: 0o000 }); + utimesSync(path, at / 1000, at / 1000); + try { + const got = identityAvailability({ + lease: lease("codex", THREAD, now), evidence: live, now, homes: { codexHome: home }, + activity: { at: now - SHARED_IDLE_MS - 60_000, shared: true }, + }); + assert.equal(got.claimable, false); + assert.equal(got.state, "live"); + assert.match(got.reason, /last activity 1 min ago \(codex rollout\)/); + } finally { rmSync(home, { recursive: true, force: true }); } +}); + +test("a live Grok session with a stale mbx call reads as active", () => { + const now = Date.now(), home = dir(); + const file = join(home, "active_sessions.json"); + writeFileSync(file, JSON.stringify([{ session_id: "grok-t583", pid: process.pid, cwd: home, opened_at: new Date(now).toISOString() }])); + try { + const got = identityAvailability({ + lease: lease("grok", "grok-t583", now), evidence: live, now, homes: { grokSessionsFile: file }, + activity: { at: now - SHARED_IDLE_MS - 60_000, shared: true }, + }); + assert.equal(got.claimable, false); + assert.equal(got.state, "live"); + assert.match(got.reason, /last activity 1 min ago \(grok session\)/); + } finally { rmSync(home, { recursive: true, force: true }); } +}); + +test("a dead harness still reads as stale", () => { + const now = Date.now(), home = dir(); + const file = join(home, "active_sessions.json"); + writeFileSync(file, JSON.stringify([{ session_id: "grok-t583", pid: 2_147_483_646, cwd: home, opened_at: new Date(now).toISOString() }])); + const day = join(home, "sessions", "2026", "10", "17"); + mkdirSync(day, { recursive: true }); + const path = join(day, `rollout-2026-10-17T12-00-00-${THREAD}.jsonl`); + writeFileSync(path, "old"); + const quietAt = now - SHARED_IDLE_MS - 1000; + utimesSync(path, quietAt / 1000, quietAt / 1000); + const stale = { at: now - SHARED_IDLE_MS - 60_000, shared: true }; + try { + const grok = identityAvailability({ + lease: lease("grok", "grok-t583", now), evidence: live, now, homes: { grokSessionsFile: file }, activity: stale, + }); + assert.equal(grok.claimable, true); + assert.equal(grok.takeover, "idle-conversation"); + assert.match(grok.reason, /no mbx activity/); + assert.doesNotMatch(grok.reason, /\(grok session\)/); + const codex = identityAvailability({ + lease: lease("codex", THREAD, now), evidence: live, now, homes: { codexHome: home }, activity: stale, + }); + assert.equal(codex.claimable, true); + assert.equal(codex.takeover, "idle-conversation"); + assert.match(codex.reason, /no mbx activity/); + assert.doesNotMatch(codex.reason, /\(codex rollout\)/); + } finally { rmSync(home, { recursive: true, force: true }); } +}); From ee1e8ba2234f2c379795c0e33512072b03b0b588 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 08:35:55 -0700 Subject: [PATCH 04/12] fix(T677): keep silent open Codex turns active --- CHANGELOG.md | 2 + dist/codex-activity.js | 65 ++++++++++++++++++++++++--- src/codex-activity.ts | 54 ++++++++++++++++++++--- test/codex-activity.test.ts | 88 ++++++++++++++++++++++++++++++++----- 4 files changed, 188 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a06257f..e433fec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- **Silent Codex turns keep their mailbox identity (T677).** The existing MCP heartbeat records activity while the exact thread's rollout has an unmatched turn start, even when it has no new writes or mbx calls. Matching completion or turn-end records stop that protection; missing, unreadable or rotated evidence falls back to rollout mtime. Lease token, thread and process-birth fences remain in force. An abandoned open turn can conservatively retain a live holder until recovery. + ## 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. diff --git a/dist/codex-activity.js b/dist/codex-activity.js index 8c6056b..33d6a47 100644 --- a/dist/codex-activity.js +++ b/dist/codex-activity.js @@ -1,13 +1,66 @@ -import { globSync, lstatSync } from "node:fs"; +import { closeSync, fstatSync, globSync, lstatSync, openSync, readFileSync } from "node:fs"; import { homedir } from "node:os"; import { join } from "node:path"; import { activityKey, parseActivity } from "./identity-availability.js"; const THREAD = /^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i; -/** Metadata-only activity for the existing MCP heartbeat; never opens a rollout or starts a process. */ +/** Exact-thread rollout activity for the existing MCP heartbeat; never starts a process. */ export class CodexActivity { home; paths = new Map(); + turns = new Map(); constructor(home = process.env.CODEX_HOME || join(homedir(), ".codex")) { this.home = home; } + openTurn(thread, path, stat) { + let fd; + try { + fd = openSync(path, "r"); + const observed = fstatSync(fd), prior = this.turns.get(thread); + if (!observed.isFile() || observed.dev !== stat.dev || observed.ino !== stat.ino + || observed.size !== stat.size || observed.mtimeMs !== stat.mtimeMs) + throw new Error("rollout changed"); + if (prior && (prior.dev !== stat.dev || prior.ino !== stat.ino || stat.size < prior.size + || stat.mtimeMs < prior.mtimeMs || (stat.size === prior.size && stat.mtimeMs !== prior.mtimeMs))) { + this.turns.delete(thread); + return false; // A replaced or rewritten rollout is not continuity evidence for the cached turn. + } + if (prior && prior.size === stat.size && prior.mtimeMs === stat.mtimeMs) + return prior.open; + const turns = new Set(); + // shortcut: changed rollouts are rescanned; use incremental reads if heartbeat parsing becomes costly. + for (const line of readFileSync(fd, "utf8").split("\n")) { + if (!line.trim()) + continue; + const row = JSON.parse(line); + if (row?.type !== "event_msg") + continue; + const payload = row.payload; + if (payload?.turn_id !== undefined && (typeof payload.turn_id !== "string" || !payload.turn_id)) + throw new Error("invalid turn id"); + const id = payload?.turn_id ?? ""; + if (payload?.type === "task_started") + turns.add(id); + else if (["task_complete", "turn_complete", "turn_completed", "turn_end", "turn_aborted"].includes(payload?.type)) { + if (id) + turns.delete(id); + else if (turns.size === 1) + turns.clear(); + } + } + const after = fstatSync(fd); + if (after.size !== stat.size || after.mtimeMs !== stat.mtimeMs) + throw new Error("rollout changed"); + const open = turns.size > 0; + this.turns.set(thread, { dev: stat.dev, ino: stat.ino, size: stat.size, mtimeMs: stat.mtimeMs, open }); + return open; + } + catch { + this.turns.delete(thread); + return false; + } + finally { + if (fd !== undefined) + closeSync(fd); + } + } record(store, leases, holder) { if (!THREAD.test(holder.session_id)) return false; @@ -21,10 +74,11 @@ export class CodexActivity { path = join(this.home, found[0]); this.paths.set(thread, path); } - const stat = lstatSync(path), at = Math.floor(stat.mtimeMs); - if (!stat.isFile() || !Number.isSafeInteger(at) || at < 0 || at > Date.now()) + const stat = lstatSync(path), modified = Math.floor(stat.mtimeMs), now = Date.now(); + if (!stat.isFile() || !Number.isSafeInteger(modified) || modified < 0 || modified > now) return false; - // shortcut: rollout writes show activity, but a silent turn remains invisible; use native thread status when available. + // An unmatched start conservatively protects a silent turn until its end is recorded. + const at = this.openTurn(thread, path, stat) ? now : modified; return leases.withHeld(holder.name, holder.token, () => { const lease = store.db.prepare("SELECT * FROM identity_leases WHERE name=?").get(holder.name); if (lease.cli !== "codex" || lease.session_id !== holder.session_id @@ -39,6 +93,7 @@ export class CodexActivity { } catch { this.paths.delete(thread); + this.turns.delete(thread); return false; // Advisory evidence: missing files, lost leases and unknown process evidence write no activity. } } diff --git a/src/codex-activity.ts b/src/codex-activity.ts index ba76ee5..11c62fd 100644 --- a/src/codex-activity.ts +++ b/src/codex-activity.ts @@ -1,4 +1,4 @@ -import { globSync, lstatSync } from "node:fs"; +import { closeSync, fstatSync, globSync, lstatSync, openSync, readFileSync, type Stats } from "node:fs"; import { homedir } from "node:os"; import { join } from "node:path"; import { activityKey, parseActivity } from "./identity-availability.ts"; @@ -8,12 +8,54 @@ import type { Store } from "./store.ts"; type Holder = Pick & { shared: boolean }; const THREAD = /^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i; -/** Metadata-only activity for the existing MCP heartbeat; never opens a rollout or starts a process. */ +/** Exact-thread rollout activity for the existing MCP heartbeat; never starts a process. */ export class CodexActivity { private home: string; private paths = new Map(); + private turns = new Map(); constructor(home = process.env.CODEX_HOME || join(homedir(), ".codex")) { this.home = home; } + private openTurn(thread: string, path: string, stat: Stats): boolean { + let fd: number | undefined; + try { + fd = openSync(path, "r"); + const observed = fstatSync(fd), prior = this.turns.get(thread); + if (!observed.isFile() || observed.dev !== stat.dev || observed.ino !== stat.ino + || observed.size !== stat.size || observed.mtimeMs !== stat.mtimeMs) throw new Error("rollout changed"); + if (prior && (prior.dev !== stat.dev || prior.ino !== stat.ino || stat.size < prior.size + || stat.mtimeMs < prior.mtimeMs || (stat.size === prior.size && stat.mtimeMs !== prior.mtimeMs))) { + this.turns.delete(thread); + return false; // A replaced or rewritten rollout is not continuity evidence for the cached turn. + } + if (prior && prior.size === stat.size && prior.mtimeMs === stat.mtimeMs) return prior.open; + const turns = new Set(); + // shortcut: changed rollouts are rescanned; use incremental reads if heartbeat parsing becomes costly. + for (const line of readFileSync(fd, "utf8").split("\n")) { + if (!line.trim()) continue; + const row = JSON.parse(line); + if (row?.type !== "event_msg") continue; + const payload = row.payload; + if (payload?.turn_id !== undefined && (typeof payload.turn_id !== "string" || !payload.turn_id)) throw new Error("invalid turn id"); + const id = payload?.turn_id ?? ""; + if (payload?.type === "task_started") turns.add(id); + else if (["task_complete", "turn_complete", "turn_completed", "turn_end", "turn_aborted"].includes(payload?.type)) { + if (id) turns.delete(id); + else if (turns.size === 1) turns.clear(); + } + } + const after = fstatSync(fd); + if (after.size !== stat.size || after.mtimeMs !== stat.mtimeMs) throw new Error("rollout changed"); + const open = turns.size > 0; + this.turns.set(thread, { dev: stat.dev, ino: stat.ino, size: stat.size, mtimeMs: stat.mtimeMs, open }); + return open; + } catch { + this.turns.delete(thread); + return false; + } finally { + if (fd !== undefined) closeSync(fd); + } + } + record(store: Store, leases: IdentityLeases, holder: Holder): boolean { if (!THREAD.test(holder.session_id)) return false; const thread = holder.session_id.toLowerCase(); @@ -25,9 +67,10 @@ export class CodexActivity { path = join(this.home, found[0]); this.paths.set(thread, path); } - const stat = lstatSync(path), at = Math.floor(stat.mtimeMs); - if (!stat.isFile() || !Number.isSafeInteger(at) || at < 0 || at > Date.now()) return false; - // shortcut: rollout writes show activity, but a silent turn remains invisible; use native thread status when available. + const stat = lstatSync(path), modified = Math.floor(stat.mtimeMs), now = Date.now(); + if (!stat.isFile() || !Number.isSafeInteger(modified) || modified < 0 || modified > now) return false; + // An unmatched start conservatively protects a silent turn until its end is recorded. + const at = this.openTurn(thread, path, stat) ? now : modified; return leases.withHeld(holder.name, holder.token, () => { const lease = store.db.prepare("SELECT * FROM identity_leases WHERE name=?").get(holder.name) as unknown as IdentityLease; if (lease.cli !== "codex" || lease.session_id !== holder.session_id @@ -39,6 +82,7 @@ export class CodexActivity { }); } catch { this.paths.delete(thread); + this.turns.delete(thread); return false; // Advisory evidence: missing files, lost leases and unknown process evidence write no activity. } } diff --git a/test/codex-activity.test.ts b/test/codex-activity.test.ts index 684a016..afe17d3 100644 --- a/test/codex-activity.test.ts +++ b/test/codex-activity.test.ts @@ -1,6 +1,6 @@ import { test, type TestContext } from "node:test"; import assert from "node:assert/strict"; -import { lstatSync, mkdirSync, mkdtempSync, rmSync, symlinkSync, utimesSync, writeFileSync } from "node:fs"; +import { appendFileSync, chmodSync, lstatSync, mkdirSync, mkdtempSync, renameSync, rmSync, symlinkSync, utimesSync, writeFileSync } from "node:fs"; import { join, resolve } from "node:path"; import { tmpdir } from "node:os"; import { CodexActivity } from "../src/codex-activity.ts"; @@ -12,6 +12,7 @@ import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js" import { MbxNode } from "../src/node.ts"; const THREAD = "44444444-4444-4444-8444-444444444444", SIBLING = "55555555-5555-4555-8555-555555555555"; +const event = (type: string, turn_id = "turn") => `${JSON.stringify({ type: "event_msg", payload: { type, turn_id } })}\n`; function fixture(t: TestContext) { const home = mkdtempSync(join(tmpdir(), "mbx-codex-activity-")), store = new Store(home); t.after(() => { store.close(); rmSync(home, { recursive: true, force: true }); }); @@ -21,10 +22,10 @@ function fixture(t: TestContext) { const lease = leases.claim("reader", { pid: 10, start: "birth", keyFp: "aaaa-aaaa-aaaa-aaaa", cli: "codex", sessionId: THREAD }); store.set(activityKey(lease.name), JSON.stringify({ at: claimed, shared: true })); const sampler = new CodexActivity(home), holder = { ...lease, shared: true }; - const rollout = (thread: string, at: number, day = "09") => { + const rollout = (thread: string, at: number, day = "09", body = "not JSON; use metadata fallback") => { const dir = join(home, "sessions", "2026", "10", day); mkdirSync(dir, { recursive: true }); const path = join(dir, `rollout-2026-10-${day}T12-00-00-${thread}.jsonl`); - writeFileSync(path, "not JSON; the rollout body must never be read", { mode: 0o000 }); + writeFileSync(path, body, { mode: body.startsWith("{") ? 0o600 : 0o000 }); utimesSync(path, at / 1000, at / 1000); return path; }; @@ -45,7 +46,67 @@ test("exact-thread rollout activity keeps a writing thread live after >10 min wi assert.equal(f.record(), false, "an unchanged rollout must not synthesize another heartbeat"); }); -test("quiet rollout and missing evidence preserve idle recovery and the silent-turn gap", t => { +test("a silent open turn stays unclaimable after >10 min without mbx calls or rollout writes", t => { + const f = fixture(t), path = f.rollout(THREAD, f.claimed - 1000, "09", event("task_started")); + const writtenAt = lstatSync(path).mtimeMs; + assert.equal(f.record(), true); + const later = f.now + SHARED_IDLE_MS + 1000; + t.mock.method(Date, "now", () => later); + assert.equal(f.record(), true, "the same open turn remains active without a new write"); + assert.equal(f.activity().at, later); + assert.equal(lstatSync(path).mtimeMs, writtenAt); + assert.equal(identityAvailability({ lease: f.lease, activity: f.activity(), evidence: { alive: true, start: "birth" }, now: later }).claimable, false); +}); + +test("a completed turn past the idle threshold remains claimable", t => { + const f = fixture(t); + f.rollout(THREAD, f.claimed - 1000, "09", event("task_started") + event("task_complete")); + assert.equal(f.record(), false); + assert.deepEqual(f.activity(), { at: f.claimed, shared: true }); + assert.equal(identityAvailability({ lease: f.lease, activity: f.activity(), evidence: { alive: true, start: "birth" }, now: f.now }).claimable, true); +}); + +for (const end of ["task_complete", "turn_end", "turn_aborted"]) + test(`${end} closes only its matching open turn and restores quiet recovery`, t => { + const f = fixture(t), path = f.rollout(THREAD, f.claimed - 1000, "09", event("task_started", "working") + event(end, "other")); + assert.equal(f.record(), true, "an unrelated turn end must not close this turn"); + appendFileSync(path, event(end, "working")); + utimesSync(path, f.claimed / 1000, f.claimed / 1000); + const later = f.now + SHARED_IDLE_MS + 1000; + t.mock.method(Date, "now", () => later); + assert.equal(f.record(), false, "a completed quiet turn must not refresh observation time"); + assert.equal(identityAvailability({ lease: f.lease, activity: f.activity(), evidence: { alive: true, start: "birth" }, now: later }).claimable, true); + }); + +for (const change of ["missing", "unreadable", "rotated", "truncated"]) + test(`${change} rollout discards cached open-turn protection and falls back to mtime`, t => { + const f = fixture(t), path = f.rollout(THREAD, f.claimed - 1000, "09", event("task_started")); + assert.equal(f.record(), true); + if (change === "missing") rmSync(path); + else if (change === "unreadable") chmodSync(path, 0o000); + else { + if (change === "rotated") renameSync(path, `${path}.old`); + writeFileSync(path, event("task_started"), { mode: 0o600 }); + if (change === "truncated") writeFileSync(path, ""); + utimesSync(path, (f.claimed - 1000) / 1000, (f.claimed - 1000) / 1000); + } + const later = f.now + SHARED_IDLE_MS + 1000; + t.mock.method(Date, "now", () => later); + assert.equal(f.record(), false); + assert.equal(identityAvailability({ lease: f.lease, activity: f.activity(), evidence: { alive: true, start: "birth" }, now: later }).claimable, true); + }); + +test("non-event payloads and malformed rollouts cannot manufacture an open turn", t => { + const f = fixture(t), path = f.rollout(THREAD, f.claimed - 1000, "09", + JSON.stringify({ type: "response_item", payload: { type: "task_started", turn_id: "fake" } })); + assert.equal(f.record(), false); + writeFileSync(path, `${event("task_started")}not JSON\n`); + utimesSync(path, (f.claimed - 1000) / 1000, (f.claimed - 1000) / 1000); + assert.equal(f.record(), false); + assert.deepEqual(f.activity(), { at: f.claimed, shared: true }); +}); + +test("quiet rollout and missing evidence preserve idle recovery", t => { const f = fixture(t); f.rollout(SIBLING, f.now - 1000); assert.equal(f.record(), false, "a sibling's rollout is not this thread's activity"); @@ -82,7 +143,7 @@ test("future, ambiguous, non-file, missing and provisional evidence writes nothi for (const changed of ["token", "holder_pid", "holder_start", "session_id", "cli", "released_at"] as const) test(`rollout evidence is fenced against a changed lease ${changed}`, t => { - const f = fixture(t); f.rollout(THREAD, f.now - 1000); + const f = fixture(t); f.rollout(THREAD, f.now - 1000, "09", event("task_started")); const value = changed === "holder_pid" ? 20 : changed === "released_at" ? f.now : changed === "session_id" ? SIBLING : "changed"; f.store.db.prepare(`UPDATE identity_leases SET ${changed}=? WHERE name=?`).run(value, f.lease.name); assert.equal(f.record(), false); assert.deepEqual(f.activity(), { at: f.claimed, shared: true }); @@ -102,7 +163,7 @@ test("unknown process evidence writes no activity and performs no synthetic rene assert.equal(lease.token, f.lease.token); assert.equal(lease.heartbeat_at, f.claimed); assert.equal(lease.released_at, null); }); -test("the existing MCP heartbeat records exact-thread mtime without another mbx call", async t => { +for (const open of [false, true]) test(`the existing MCP heartbeat records ${open ? "a silent open turn" : "exact-thread mtime"} without another mbx call`, async t => { const home = mkdtempSync(join(tmpdir(), "mbx-codex-heartbeat-")), node = new MbxNode(home, { host: "alpha" }); const client = new Client({ name: "codex-activity", version: "1" }), preload = join(home, "heartbeat.mjs"); writeFileSync(preload, "const interval=globalThis.setInterval; globalThis.setInterval=(fn,ms,...args)=>interval(fn,ms===60000?50:ms,...args);"); @@ -115,14 +176,19 @@ test("the existing MCP heartbeat records exact-thread mtime without another mbx node.store.set(activityKey(lease.name), JSON.stringify({ at: Date.now() - SHARED_IDLE_MS - 1000, shared: true })); const dir = join(home, "sessions", "2026", "10", "16"); mkdirSync(dir, { recursive: true }); const path = join(dir, `rollout-2026-10-16T12-00-00-${THREAD}.jsonl`); - writeFileSync(path, "unreadable rollout contents", { mode: 0o000 }); + writeFileSync(path, open ? event("task_started") : "unreadable rollout contents", { mode: open ? 0o600 : 0o000 }); + if (open) utimesSync(path, (lease.claimed_at - SHARED_IDLE_MS - 1000) / 1000, (lease.claimed_at - SHARED_IDLE_MS - 1000) / 1000); const at = Math.floor(lstatSync(path).mtimeMs); - assert.ok(at > lease.claimed_at); + assert.equal(at > lease.claimed_at, !open); let activity = parseActivity(node.store.get(activityKey(lease.name))); - for (let i = 0; i < 40 && activity?.at !== at; i++) { + for (let i = 0; i < 40 && !(open ? (activity?.at ?? 0) > lease.claimed_at : activity?.at === at); i++) { await new Promise(resolve => setTimeout(resolve, 50)); activity = parseActivity(node.store.get(activityKey(lease.name))); } - assert.deepEqual(activity, { at, shared: true }); + if (open) assert.ok(activity!.at > lease.claimed_at); + else assert.deepEqual(activity, { at, shared: true }); + const first = activity!.at; await new Promise(resolve => setTimeout(resolve, 200)); - assert.equal(parseActivity(node.store.get(activityKey(lease.name)))?.at, at, "an idle heartbeat does not substitute now"); + if (open) assert.ok(parseActivity(node.store.get(activityKey(lease.name)))!.at > first); + else assert.equal(parseActivity(node.store.get(activityKey(lease.name)))?.at, at, "an idle heartbeat does not substitute now"); + assert.equal(Math.floor(lstatSync(path).mtimeMs), at, "heartbeats do not write the rollout"); }); From 5c929ce73a115a5bf1db25a6745af09bfe43c9de Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 08:47:42 -0700 Subject: [PATCH 05/12] feat(T187): keep an OpenCode resume note when the turn is busy A running turn or a host with no session note API keeps the bounded resume note and reports one fallback. A fresh session and a quiet tool result still show the note. Mail is not read or acked. --- CHANGELOG.md | 2 + dist/setup.js | 22 +++- src/setup.ts | 22 +++- test/opencode-resume.test.ts | 191 +++++++++++++++++++++++++++++++++++ 4 files changed, 231 insertions(+), 6 deletions(-) create mode 100644 test/opencode-resume.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index a06257f..8f83176 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- **OpenCode resume notes wait when the turn is busy or the session API is missing (T187).** A fresh `session.created`, and the first tool result of a quiet turn, still show the bounded `[mbx]` resume note: who the session is, what it has not caught up on, and `mbx_inbox`. A turn that is already running keeps that note and adds one fallback line (`mbx_whoami`, then `mbx_inbox`); the note is admitted when the turn ends. A host with no `synthetic` or `prompt` does the same, and a later tool result can still show the note. This path does not read or ack mail. The plugin template changed: run `agentmbx setup --only opencode` and reopen OpenCode tabs to load it. + ## 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. diff --git a/dist/setup.js b/dist/setup.js index fbc506f..664417a 100644 --- a/dist/setup.js +++ b/dist/setup.js @@ -1583,18 +1583,34 @@ const postToolSteady = (sid: string): boolean => { * admitted through the hosting serve's session API. A failure never surfaces. The same call starts * the standalone wake poll once the session id is known. */ const call = async (ctx: NoteCtx, bin: string[], event: string, payload: Record, pending: Map, sending: Set, - place?: (text: string) => boolean): Promise => { + resumeTold: Set, place?: (text: string) => boolean): Promise => { try { const sid = typeof payload.session_id === "string" ? payload.session_id : null; + // Captured before this hook marks or ends the turn, so the first tool of a turn can still admit + // and a turn that is already running cannot. + const since = sid ? turnSince.get(sid) : undefined; + const turnOpen = since !== undefined && Date.now() - since < TURN_STALE_MS; if (sid) watchWake(ctx, sid); if (sid && event === "post-tool") markTurn(sid); if (sid && event === "stop") endTurn(sid); if (sid && (event === "post-tool" || event === "stop")) await releaseStartNote(ctx, sid, event, place); // T187: retry an unadmitted notice on an existing hook, without bypassing the zero-spawn fast path. + // A turn already running, or a host with no session note API, keeps the note and says so once. + const sayResumeFallback = (session: string, text: string) => { + const key = session + "\\0" + text; + if (resumeTold.has(key)) return; + resumeTold.add(key); + const line = "[mbx] Resume note was not admitted. Call mbx_whoami, then mbx_inbox. Mail is unchanged."; + if (!(place && place(line))) logWake("resume", session, "not admitted; mail unchanged"); + }; const deliver = async () => { if (!sid || sending.has(sid)) return; const text = pending.get(sid); if (!text) return; + if (turnOpen && event !== "stop") { sayResumeFallback(sid, text); return; } + if (event === "post-tool" && place && place(text)) { if (pending.get(sid) === text) pending.delete(sid); return; } + const api = typeof ctx?.session?.synthetic === "function" || typeof ctx?.session?.prompt === "function"; + if (!api) { sayResumeFallback(sid, text); return; } sending.add(sid); try { if (await inject(ctx, sid, "resume-entry", text) && pending.get(sid) === text) pending.delete(sid); } finally { sending.delete(sid); } @@ -1622,8 +1638,8 @@ const call = async (ctx: NoteCtx, bin: string[], event: string, payload: Record< }; const runFor = (ctx: NoteCtx, bin: string[], cwd: string | undefined) => { - const pending = new Map(), sending = new Set(); - return (event: string, payload: Record, place?: (text: string) => boolean) => call(ctx, bin, event, { cwd, ...payload }, pending, sending, place); + const pending = new Map(), sending = new Set(), resumeTold = new Set(); + return (event: string, payload: Record, place?: (text: string) => boolean) => call(ctx, bin, event, { cwd, ...payload }, pending, sending, resumeTold, place); }; const sidOf = (v: unknown): string | undefined => { diff --git a/src/setup.ts b/src/setup.ts index 0d7c8c5..c62efcd 100644 --- a/src/setup.ts +++ b/src/setup.ts @@ -1416,18 +1416,34 @@ const postToolSteady = (sid: string): boolean => { * admitted through the hosting serve's session API. A failure never surfaces. The same call starts * the standalone wake poll once the session id is known. */ const call = async (ctx: NoteCtx, bin: string[], event: string, payload: Record, pending: Map, sending: Set, - place?: (text: string) => boolean): Promise => { + resumeTold: Set, place?: (text: string) => boolean): Promise => { try { const sid = typeof payload.session_id === "string" ? payload.session_id : null; + // Captured before this hook marks or ends the turn, so the first tool of a turn can still admit + // and a turn that is already running cannot. + const since = sid ? turnSince.get(sid) : undefined; + const turnOpen = since !== undefined && Date.now() - since < TURN_STALE_MS; if (sid) watchWake(ctx, sid); if (sid && event === "post-tool") markTurn(sid); if (sid && event === "stop") endTurn(sid); if (sid && (event === "post-tool" || event === "stop")) await releaseStartNote(ctx, sid, event, place); // T187: retry an unadmitted notice on an existing hook, without bypassing the zero-spawn fast path. + // A turn already running, or a host with no session note API, keeps the note and says so once. + const sayResumeFallback = (session: string, text: string) => { + const key = session + "\\0" + text; + if (resumeTold.has(key)) return; + resumeTold.add(key); + const line = "[mbx] Resume note was not admitted. Call mbx_whoami, then mbx_inbox. Mail is unchanged."; + if (!(place && place(line))) logWake("resume", session, "not admitted; mail unchanged"); + }; const deliver = async () => { if (!sid || sending.has(sid)) return; const text = pending.get(sid); if (!text) return; + if (turnOpen && event !== "stop") { sayResumeFallback(sid, text); return; } + if (event === "post-tool" && place && place(text)) { if (pending.get(sid) === text) pending.delete(sid); return; } + const api = typeof ctx?.session?.synthetic === "function" || typeof ctx?.session?.prompt === "function"; + if (!api) { sayResumeFallback(sid, text); return; } sending.add(sid); try { if (await inject(ctx, sid, "resume-entry", text) && pending.get(sid) === text) pending.delete(sid); } finally { sending.delete(sid); } @@ -1455,8 +1471,8 @@ const call = async (ctx: NoteCtx, bin: string[], event: string, payload: Record< }; const runFor = (ctx: NoteCtx, bin: string[], cwd: string | undefined) => { - const pending = new Map(), sending = new Set(); - return (event: string, payload: Record, place?: (text: string) => boolean) => call(ctx, bin, event, { cwd, ...payload }, pending, sending, place); + const pending = new Map(), sending = new Set(), resumeTold = new Set(); + return (event: string, payload: Record, place?: (text: string) => boolean) => call(ctx, bin, event, { cwd, ...payload }, pending, sending, resumeTold, place); }; const sidOf = (v: unknown): string | undefined => { diff --git a/test/opencode-resume.test.ts b/test/opencode-resume.test.ts new file mode 100644 index 0000000..95cbb2b --- /dev/null +++ b/test/opencode-resume.test.ts @@ -0,0 +1,191 @@ +// T187: the OpenCode plugin's resume-entry path. Fresh and quiet-resume hooks show the bounded +// notice. A busy turn, a missing binding, a stale connector, or a host with no session note API +// keeps the note and says so, without a wake and without a fake receipt. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { opencodePluginSource } from "../src/setup.ts"; +import { version } from "../src/version.ts"; + +const CMD = ["/opt/bin/agentmbx"]; +const NOTICE = [ + "[mbx] You are worker@alpha (role: builder).", + "[mbx] 1 unread mbx message(s): call mbx_inbox.", + "[mbx] Not caught up: call mbx_catchup.", + "[mbx] 1 open thread(s) with unacked mail.", +].join("\n"); +const FALLBACK = "[mbx] Resume note was not admitted. Call mbx_whoami, then mbx_inbox. Mail is unchanged."; + +interface Note { sessionID: string; id: string; text: string; delivery: string; resume: boolean } +interface Spawned { args: string[]; input: Record } +type Hooks = { + event: (event: unknown) => Promise; + "tool.execute.after": (input: unknown, output?: { output?: string }) => Promise; +}; + +async function loadPlugin() { + (globalThis as { __mbxWakePort?: string | null }).__mbxWakePort = null; + const dir = mkdtempSync(join(tmpdir(), "mbx-oc-resume-")); + const file = join(dir, "agentmbx.ts"); + writeFileSync(file, opencodePluginSource(CMD, version())); + try { + const mod = await import(file) as { AgentMBXHooks: (ctx: unknown) => Promise }; + return { mod, dir }; + } catch (e) { rmSync(dir, { recursive: true, force: true }); throw e; } +} + +function stubSpawn(calls: Spawned[], script: (c: Spawned) => string) { + (globalThis as { __mbxSpawn?: unknown }).__mbxSpawn = (bin: string[], args: string[], input: string, cb: (out: string) => void) => { + const c = { args, input: JSON.parse(input) as Record }; + calls.push(c); + cb(script(c)); + }; + return () => { delete (globalThis as { __mbxSpawn?: unknown }).__mbxSpawn; }; +} + +function stubSession(notes: Note[]) { + return { synthetic: async (body: Note) => { notes.push(body); return body; } }; +} + +const envelope = (sid: string, text: string) => JSON.stringify({ mbx_resume: { session_id: sid, text } }); + +test("T187: a fresh session admits the bounded notice once, to that session only", async () => { + const { mod, dir } = await loadPlugin(), calls: Spawned[] = [], notes: Note[] = []; + const restore = stubSpawn(calls, c => { + if (c.args[1] === "session-start" && c.input.session_id === "ses_fresh") return envelope("ses_fresh", NOTICE); + if (c.input.session_id === "ses_other") return envelope("ses_fresh", NOTICE); + return ""; + }); + try { + const hooks = await mod.AgentMBXHooks({ directory: "/work", session: stubSession(notes) }); + await hooks.event({ event: { type: "session.created", properties: { info: { id: "ses_fresh" } } } }); + assert.equal(notes.length, 1); + assert.equal(notes[0].sessionID, "ses_fresh"); + assert.equal(notes[0].text, NOTICE); + assert.equal(notes[0].resume, false); + assert.equal(notes[0].delivery, "queue"); + assert.ok(NOTICE.length <= 1200 && NOTICE.split("\n").length <= 8); + assert.match(NOTICE, /mbx_inbox/); + assert.match(NOTICE, /mbx_catchup/); + assert.doesNotMatch(NOTICE, /PRIVATE/); + await hooks["tool.execute.after"]({ sessionID: "ses_fresh" }); + await hooks.event({ event: { type: "session.created", properties: { info: { id: "ses_other" } } } }); + assert.equal(notes.length, 1, "an empty follow-up and a mismatched envelope do not admit again"); + assert.ok(notes.every(n => n.resume === false)); + } finally { restore(); rmSync(dir, { recursive: true, force: true }); } +}); + +test("T187: a quiet resume places the same notice in that tool result and does not queue a copy", async () => { + const { mod, dir } = await loadPlugin(), notes: Note[] = []; + let once = false; + const restore = stubSpawn([], c => { + if (!once && c.args[1] === "post-tool" && c.input.session_id === "ses_resume") { once = true; return envelope("ses_resume", NOTICE); } + return ""; + }); + const output = { output: "tool ok" }; + try { + const hooks = await mod.AgentMBXHooks({ directory: "/work", session: stubSession(notes) }); + await hooks["tool.execute.after"]({ sessionID: "ses_resume" }, output); + assert.equal(notes.length, 0, "the tool result is the admission; nothing is queued behind the turn"); + assert.match(output.output, /You are worker@alpha/); + assert.match(output.output, /mbx_inbox/); + assert.match(output.output, /mbx_catchup/); + assert.doesNotMatch(output.output, /Resume note was not admitted/); + const again = { output: "next tool" }; + await hooks["tool.execute.after"]({ sessionID: "ses_resume" }, again); + assert.equal(again.output, "next tool", "the same turn does not repeat the notice"); + assert.equal(notes.length, 0); + } finally { restore(); rmSync(dir, { recursive: true, force: true }); } +}); + +test("T187: a busy turn keeps the resume note, reports the fallback once, and admits it when the turn ends", async () => { + const { mod, dir } = await loadPlugin(), notes: Note[] = []; + let tools = 0; + const restore = stubSpawn([], c => { + if (c.args[1] !== "post-tool") return ""; + return ++tools === 1 ? "" : envelope("ses_busy", NOTICE); + }); + try { + const hooks = await mod.AgentMBXHooks({ directory: "/work", session: stubSession(notes) }); + await hooks["tool.execute.after"]({ sessionID: "ses_busy" }, { output: "first" }); + const mid = { output: "mid" }; + await hooks["tool.execute.after"]({ sessionID: "ses_busy" }, mid); + assert.equal(notes.length, 0, "a running turn does not admit the note"); + assert.ok(mid.output.includes(FALLBACK)); + assert.doesNotMatch(mid.output, /mbx_catchup/, "the full note stays pending"); + const third = { output: "third" }; + await hooks["tool.execute.after"]({ sessionID: "ses_busy" }, third); + assert.equal(third.output, "third", "the fallback is once per pending note"); + await hooks.event({ event: { type: "session.idle", properties: { info: { id: "ses_busy" } } } }); + assert.equal(notes.length, 1); + assert.equal(notes[0].sessionID, "ses_busy"); + assert.equal(notes[0].text, NOTICE); + assert.equal(notes[0].resume, false, "ending the turn admits the note; it does not start a wake"); + } finally { restore(); rmSync(dir, { recursive: true, force: true }); } +}); + +test("T187: a missing binding holds the register fallback and never admits a resume note", async () => { + const { mod, dir } = await loadPlugin(), notes: Note[] = []; + const unbound = "[mbx] This session has no mailbox identity yet.\n[mbx] Register or claim, then call mbx_inbox."; + let started = false; + const restore = stubSpawn([], c => { + if (c.args[1] === "session-start" && !started) { started = true; return unbound; } + return ""; + }); + const output = { output: "tool ok" }; + try { + const hooks = await mod.AgentMBXHooks({ directory: "/work", session: stubSession(notes) }); + await hooks.event({ event: { type: "session.created", properties: { info: { id: "ses_unbound" } } } }); + assert.equal(notes.length, 0, "an unbound start note is held, not admitted"); + await hooks["tool.execute.after"]({ sessionID: "ses_unbound" }, output); + assert.match(output.output, /no mailbox identity yet/); + assert.match(output.output, /mbx_inbox/); + assert.doesNotMatch(output.output, /Resume note was not admitted/); + assert.equal(notes.length, 0); + await hooks.event({ event: { type: "session.idle", properties: { info: { id: "ses_unbound" } } } }); + assert.equal(notes.length, 0, "idle does not turn the fallback into a resume admission"); + } finally { restore(); rmSync(dir, { recursive: true, force: true }); } +}); + +test("T187: a stale connector shows the disconnect line and does not admit a resume note", async () => { + const { mod, dir } = await loadPlugin(), notes: Note[] = []; + const line = "[mbx] mbx disconnected: reconnect it with /mcps in OpenCode (select mbx); this session no longer holds stale-worker@alpha."; + const restore = stubSpawn([], c => c.args[1] === "post-tool" ? line : ""); + const output = { output: "tool ok" }; + try { + const hooks = await mod.AgentMBXHooks({ directory: "/work", session: stubSession(notes) }); + await hooks["tool.execute.after"]({ sessionID: "ses_stale" }, output); + assert.match(output.output, /mbx disconnected/); + assert.match(output.output, /stale-worker@alpha/); + assert.equal(notes.length, 0); + await hooks.event({ event: { type: "session.idle", properties: { info: { id: "ses_stale" } } } }); + assert.equal(notes.length, 0, "the disconnect line is not kept as a resume note"); + assert.ok(notes.every(n => n.resume === false)); + } finally { restore(); rmSync(dir, { recursive: true, force: true }); } +}); + +test("T187: a host with no session note API keeps the note and a later tool result can show it", async () => { + const { mod, dir } = await loadPlugin(); + const home = process.env.MBX_HOME; + process.env.MBX_HOME = dir; + const restore = stubSpawn([], c => c.args[1] === "session-start" ? envelope("ses_noapi", NOTICE) : ""); + const output = { output: "tool ok" }; + try { + const hooks = await mod.AgentMBXHooks({ directory: "/work", session: {} }); + await hooks.event({ event: { type: "session.created", properties: { info: { id: "ses_noapi" } } } }); + const log = readFileSync(join(dir, "logs", "opencode-plugin.log"), "utf8"); + assert.match(log, /resume session=ses_noapi/); + assert.match(log, /not admitted; mail unchanged/); + await hooks["tool.execute.after"]({ sessionID: "ses_noapi" }, output); + assert.match(output.output, /You are worker@alpha/); + assert.match(output.output, /mbx_inbox/); + assert.doesNotMatch(output.output, /Resume note was not admitted/, "the tool result shows the note itself"); + await hooks.event({ event: { type: "session.idle", properties: { info: { id: "ses_noapi" } } } }); + assert.equal(output.output.includes(NOTICE), true); + } finally { + if (home === undefined) delete process.env.MBX_HOME; else process.env.MBX_HOME = home; + restore(); rmSync(dir, { recursive: true, force: true }); + } +}); From 0424b22f150583e8dbecb5b1afa4f3a558ae73c3 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 09:12:19 -0700 Subject: [PATCH 06/12] chore(T698): keep integrated entries under Unreleased Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c2c70f..2df7e40 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,9 +3,13 @@ ## Unreleased - **Skill relay-depth guidance matches policy enforcement (T696).** The skill now says ask allows 6 hops and collaborate, autonomous and yolo have no limit. A regression test checks the documented limits against `LEVEL_MAX_HOP`. + - **OpenCode wake documentation matches the shipped adapters (T697).** README and portability guidance describe standalone `resume: true` wakes, one shared service per machine for long-lived agents, and standalone/service-hosted recovery after MCP eviction (T659/T688). + - **Liveness follows the harness when the mailbox has gone quiet (T583).** `mbx_agents` and `agentmbx agents` timed a live holder by its last mbx call, so a session that was busy in its harness looked stale. A Codex lease now also uses that thread's rollout mtime (file metadata only, the same one-file rule as T677). A Grok lease uses `active_sessions.json`: a live row is a heartbeat, and a dead or reused pid is not. When that stamp is the newer one the reason says `last activity N min ago (codex rollout)` or `(grok session)`; otherwise it still says `last mbx activity`. A shared conversation becomes claimable only once both are quiet. + - **Silent Codex turns keep their mailbox identity (T677).** The existing MCP heartbeat records activity while the exact thread's rollout has an unmatched turn start, even when it has no new writes or mbx calls. Matching completion or turn-end records stop that protection; missing, unreadable or rotated evidence falls back to rollout mtime. Lease token, thread and process-birth fences remain in force. An abandoned open turn can conservatively retain a live holder until recovery. + - **OpenCode resume notes wait when the turn is busy or the session API is missing (T187).** A fresh `session.created`, and the first tool result of a quiet turn, still show the bounded `[mbx]` resume note: who the session is, what it has not caught up on, and `mbx_inbox`. A turn that is already running keeps that note and adds one fallback line (`mbx_whoami`, then `mbx_inbox`); the note is admitted when the turn ends. A host with no `synthetic` or `prompt` does the same, and a later tool result can still show the note. This path does not read or ack mail. The plugin template changed: run `agentmbx setup --only opencode` and reopen OpenCode tabs to load it. ## 0.5.47 (2026-10-16) From 6f38782c3fd19fb40b82c1421621d3e80fe6496f Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 09:17:45 -0700 Subject: [PATCH 07/12] 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 08/12] 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 09/12] 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. From c01a36db88c6289859be8f5c13e2ac00284d3375 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 09:27:47 -0700 Subject: [PATCH 10/12] chore(T698): keep integrated entries under Unreleased Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b5f088..8d03475 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ - **Silent Codex turns keep their mailbox identity (T677).** The existing MCP heartbeat records activity while the exact thread's rollout has an unmatched turn start, even when it has no new writes or mbx calls. Matching completion or turn-end records stop that protection; missing, unreadable or rotated evidence falls back to rollout mtime. Lease token, thread and process-birth fences remain in force. An abandoned open turn can conservatively retain a live holder until recovery. - **OpenCode resume notes wait when the turn is busy or the session API is missing (T187).** A fresh `session.created`, and the first tool result of a quiet turn, still show the bounded `[mbx]` resume note: who the session is, what it has not caught up on, and `mbx_inbox`. A turn that is already running keeps that note and adds one fallback line (`mbx_whoami`, then `mbx_inbox`); the note is admitted when the turn ends. A host with no `synthetic` or `prompt` does the same, and a later tool result can still show the note. This path does not read or ack mail. The plugin template changed: run `agentmbx setup --only opencode` and reopen OpenCode tabs to load it. + - **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) From 6653958b7bfd05bd492235d05d92e7a7cc959d19 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 09:20:56 -0700 Subject: [PATCH 11/12] test(T185): lock Claude SessionStart fork and busy fixtures Fork must leave the parent session row and print the MCP fallback. A live holder must emit the held-by-live line with no takeover suggestion. --- test/hooks.test.ts | 52 ++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 50 insertions(+), 2 deletions(-) diff --git a/test/hooks.test.ts b/test/hooks.test.ts index 0bf74a7..80cf670 100644 --- a/test/hooks.test.ts +++ b/test/hooks.test.ts @@ -1,7 +1,7 @@ import { sendLeased } from "./helpers/leased-send.ts"; import { test, type TestContext } from "node:test"; import assert from "node:assert/strict"; -import { spawnSync } from "node:child_process"; +import { spawn, spawnSync } from "node:child_process"; import { mkdtempSync, rmSync, mkdirSync, writeFileSync, existsSync, readFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; @@ -16,6 +16,7 @@ import { MbxNode } from "../src/node.ts"; import { IdentityLeases } from "../src/identity-leases.ts"; import { applyForward, buildForward } from "../src/identity-cleanup.ts"; import { RESUME_NOTICE_MAX_CHARS, RESUME_NOTICE_MAX_LINES } from "../src/resume-notice.ts"; +import { bindWakeLease } from "./helpers/wake-lease.ts"; for (const cli of ["claude", "codex", "kimi", "opencode"]) test(`${cli} startup offers recovery choices without claiming historical mail`, t => { const home = mkdtempSync(join(tmpdir(), "mbx-startup-choice-")), node = new MbxNode(home, { host: "alpha" }); @@ -400,6 +401,7 @@ test("lean notices: a wake prompt adds no second notice, and the policy recap is // ---- T185: Claude SessionStart carries the bounded resume notice (spec t184-provider-resume-adapters, Claude section) ---- const additional = (stdout: string): string => stdout ? (JSON.parse(stdout) as { hookSpecificOutput: { additionalContext: string } }).hookSpecificOutput.additionalContext : ""; const hinted = (s: string) => (s.match(/not caught up on; call mbx_catchup/g) ?? []).length; +const FORK_FALLBACK = "[mbx] Mailbox context could not be restored. Call mbx_whoami, then mbx_catchup."; test("T185: every Claude SessionStart source but fork is an entry with one bounded notice; prompts after it add none; nothing moves", async t => { const { n, agent, run, send } = await holder(t, "claude"); @@ -428,11 +430,57 @@ test("T185: every Claude SessionStart source but fork is an entry with one bound assert.equal(hinted(prompt("after-clear")), 0); assert.equal(hinted(start("first", "resume")), 1, "/resume back to an earlier conversation"); const fork = start("forked", "fork"); - assert.equal(hinted(fork), 0, "fork is unsupported (D4/C2): no notice"); assert.match(fork, /You are builder@alpha/); + assert.equal(fork, FORK_FALLBACK, "fork prints silence plus the MCP fallback"); + assert.equal(hinted(fork), 0); + assert.doesNotMatch(fork, /You are |takeover/i); assert.equal(hinted(prompt("forked")), 0, "and none is deferred to its first prompt"); assert.equal(state(), before, "no delivery state, checkpoint or lease moved"); }); +test("T185: a Claude fork leaves the parent session row and prints the MCP fallback", async t => { + const { n, run } = await holder(t, "claude"); + assert.equal(run("prompt", "parent", "claude", {}, { prompt: "bind the parent" }).status, 0); + const rows = () => n.store.db.prepare("SELECT agent,cli,session_id,pid,session_key FROM sessions ORDER BY cli,session_id").all(); + const leases = () => n.store.db.prepare("SELECT name,cli,session_id,released_at,token FROM identity_leases ORDER BY name").all(); + const before = rows(), leaseBefore = leases(); + assert.deepEqual(before.map((r: { session_id: string }) => r.session_id), ["parent"]); + const r = run("session-start", "child", "claude", {}, { source: "fork" }); + assert.equal(r.status, 0, r.stderr); + assert.deepEqual(rows(), before, "fork leaves the sessions table and the parent row untouched"); + assert.deepEqual(leases(), leaseBefore); + const context = additional(r.stdout); + assert.equal(context, FORK_FALLBACK); + assert.equal(hinted(context), 0); + assert.doesNotMatch(context, /You are |takeover/i); + assert.equal(hinted(additional(run("prompt", "child", "claude", {}, { prompt: "go on" }).stdout)), 0); +}); + +test("T185: a Claude SessionStart held live by another session names that holder and does not suggest takeover", t => { + const home = mkdtempSync(join(tmpdir(), "mbx-t185-busy-")); + const n = new MbxNode(home, { host: "alpha" }); + const holderProc = spawn("sleep", ["600"], { stdio: "ignore" }); + t.after(() => { holderProc.kill("SIGKILL"); n.close(); rmSync(home, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }); }); + assert.ok(holderProc.pid); + bindWakeLease(n, { agent: "builder", cli: "claude", session_id: "holder-session", pid: holderProc.pid }); + n.keepName("claude", "busy-session", "builder"); + sendLeased(n, { from: "sender", to: ["builder"], subject: "PRIVATE SUBJECT", body: "SECRET BODY", kind: "request" }); + const rows = () => n.store.db.prepare("SELECT agent,cli,session_id,pid,session_key FROM sessions ORDER BY cli,session_id").all(); + const leases = () => n.store.db.prepare("SELECT name,cli,session_id,released_at,token FROM identity_leases ORDER BY name").all(); + const deliveries = () => n.store.db.prepare("SELECT msg_id,agent,state FROM deliveries ORDER BY msg_id").all(); + const before = { rows: rows(), leases: leases(), deliveries: deliveries() }; + const r = spawnSync(process.execPath, [resolve("bin/agentmbx.js"), "hook", "session-start", "--cli", "claude"], { + input: JSON.stringify({ session_id: "busy-session", cwd: process.cwd(), source: "startup" }), encoding: "utf8", timeout: 10_000, + env: { ...process.env, MBX_HOME: home, MBX_AGENT: "unrelated", AGENTMBX_DEV: "1" }, + }); + assert.equal(r.status, 0, r.stderr); + assert.deepEqual({ rows: rows(), leases: leases(), deliveries: deliveries() }, before); + const context = additional(r.stdout); + assert.match(context, /held by live claude session holder-session/); + assert.doesNotMatch(context, /takeover/i); + assert.equal(hinted(context), 0); + assert.doesNotMatch(context, /You are |PRIVATE SUBJECT|SECRET BODY|no mailbox identity yet/); +}); + test("T185: an unbound Claude SessionStart defers the notice to the first bound prompt, once; Codex SessionStart is unchanged", async t => { const { n, call, agent, run, send } = await holder(t, "claude"); send(); From ee8f88b8108cbb690d34ee6467933cb63126010a Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 09:59:56 -0700 Subject: [PATCH 12/12] fix(T185): Claude fork skips bind and a live holder is named without takeover --- CHANGELOG.md | 2 ++ dist/cli.js | 42 +++++++++++++++++++++++++++++++++++++++--- src/cli.ts | 33 ++++++++++++++++++++++++++++++--- test/hooks.test.ts | 4 ++-- 4 files changed, 73 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d03475..d9157b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- **Claude SessionStart fork keeps the parent row, and a live holder is named without a takeover suggestion (T185).** A `source: fork` start skips bind and the parent-row delete and prints only the mailbox-context fallback. A session whose name is held live by another session prints that `held by live` reason and does not suggest takeover. + - **Skill relay-depth guidance matches policy enforcement (T696).** The skill now says ask allows 6 hops and collaborate, autonomous and yolo have no limit. A regression test checks the documented limits against `LEVEL_MAX_HOP`. - **OpenCode wake documentation matches the shipped adapters (T697).** README and portability guidance describe standalone `resume: true` wakes, one shared service per machine for long-lived agents, and standalone/service-hosted recovery after MCP eviction (T659/T688). diff --git a/dist/cli.js b/dist/cli.js index 2ec5e0e..232662b 100644 --- a/dist/cli.js +++ b/dist/cli.js @@ -40,7 +40,7 @@ import { CLIS, defaultHostName, defaultWhich, formatRows, ownerStep, resolveComm 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"; +import { activityKey, identityAvailability, parseActivity } from "./identity-availability.js"; import { fillCleoPresence } from "./cleo-presence.js"; import { identityLeaseStatus, inspectLeaseProcess } from "./identity-leases.js"; import { AUTO_NAME_RE, identityProjects, linkedKey, projectOf, recordSessionHint, registeredIdentity, removeStaleProjectBindings, staleProjectBindings } from "./registry.js"; @@ -1936,6 +1936,15 @@ async function hook(node, event, cli) { const at = sessionPid(session); return !!at?.pid && takeResumeEntry(node.store, cli, at.pid, at.pid_start, session); }; + // T185: a Claude fork is a different conversation. Skip bindSession and the parent-row delete. + // Record the entry on the parent row's pid so the fork's first prompt does not add a catch-up hint. + if (event === "session-start" && cli === "claude" && sid && input.source === "fork") { + const parent = node.store.db.prepare("SELECT pid,pid_start FROM sessions WHERE cli=? AND pid=? AND session_id NOT GLOB 'mcp-*'").get(cli, process.ppid); + if (parent?.pid) + takeResumeEntry(node.store, cli, parent.pid, parent.pid_start, sid); + emit(cli, "SessionStart", "[mbx] Mailbox context could not be restored. Call mbx_whoami, then mbx_catchup."); + return; + } // Inspect provider capabilities and processes before the lease transaction. No directory-based session discovery. const host = cli === "claude" ? detectHost(process.ppid) : null; const watch = event === "session-start" && noPush(cli, !!host && (host.channel || host.socket), cli === "kimi" && kimiMultiHost(process.ppid)); @@ -2003,8 +2012,8 @@ async function hook(node, event, cli) { } if (event === "session-start") { // T185: a bound Claude SessionStart is itself the entry (startup, resume, clear, compact; T184 spec C1): it carries the - // resume notice even on an unchanged id, because /clear and /compact do not re-send the MCP instructions. A fork only - // records the entry (D4/C2), so no notice is deferred to its first prompt either. T186: a bound Codex SessionStart is + // resume notice even on an unchanged id, because /clear and /compact do not re-send the MCP instructions. A fork + // returned before this bind. T186: a bound Codex SessionStart is // the entry too (a missing source counts as startup); its payload has no event id, so clear/compact re-inject on the // same thread id and enteredNow only stops the following prompt from adding a second copy. const hookSource = typeof input.source === "string" && input.source ? input.source : "startup"; @@ -2264,6 +2273,33 @@ async function hook(node, event, cli) { emit(cli, "PostToolUse", stillResuming); return; } + // T185: a name held live by another session is a conflict, not an invitation to claim. Same-session and released + // leases fall through so T573 resuming and T503 disconnect stay as they are. Nothing here writes. + const busyHeld = () => { + if (cli !== "claude" || !sid || (event !== "session-start" && event !== "prompt")) + return null; + const name = node.store.get(`name:${cli}:${sid}`); + if (!name || !NAME_RE.test(name) || node.store.get(`identity-conflict:${name}`) !== undefined) + return null; + const lease = node.store.db.prepare("SELECT * FROM identity_leases WHERE name=?").get(name); + if (!lease || lease.released_at !== null || (lease.cli === cli && lease.session_id === sid)) + return null; + const availability = identityAvailability({ + lease, evidence: inspectLeaseProcess(lease.holder_pid), + activity: parseActivity(node.store.get(activityKey(name))), now: Date.now(), + }); + if (availability.claimable || availability.takeover || availability.coUse || !availability.reason.startsWith("held by live ")) + return null; + return `[mbx] ${availability.reason}.`; + }; + try { + const busy = busyHeld(); + if (busy) { + emit(cli, event === "session-start" ? "SessionStart" : "UserPromptSubmit", busy); + return; + } + } + catch { /* a lookup failure keeps the existing unbound guidance */ } const disconnected = grokUnboundPost ? null : hookDisconnectedNote(node, cli, sid, process.ppid); const multi = cli === "kimi" && kimiMultiHost(process.ppid); // A conversation in a multi-conversation Kimi host can't be matched to its mbx server from here: hand it a bind diff --git a/src/cli.ts b/src/cli.ts index ce1c343..7440cbc 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -41,7 +41,7 @@ import { CLIS, defaultHostName, defaultWhich, formatRows, ownerStep, resolveComm 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"; +import { activityKey, identityAvailability, parseActivity } from "./identity-availability.ts"; import { fillCleoPresence } from "./cleo-presence.ts"; import { identityLeaseStatus, inspectLeaseProcess, type IdentityLease } from "./identity-leases.ts"; import { AUTO_NAME_RE, identityProjects, linkedKey, projectOf, recordSessionHint, registeredIdentity, removeStaleProjectBindings, staleProjectBindings } from "./registry.ts"; @@ -1586,6 +1586,14 @@ async function hook(node: MbxNode, event: string | undefined, cli: string) { const at = sessionPid(session); return !!at?.pid && takeResumeEntry(node.store, cli, at.pid, at.pid_start, session); }; + // T185: a Claude fork is a different conversation. Skip bindSession and the parent-row delete. + // Record the entry on the parent row's pid so the fork's first prompt does not add a catch-up hint. + if (event === "session-start" && cli === "claude" && sid && input.source === "fork") { + const parent = node.store.db.prepare("SELECT pid,pid_start FROM sessions WHERE cli=? AND pid=? AND session_id NOT GLOB 'mcp-*'").get(cli, process.ppid) as { pid: number | null; pid_start: string | null } | undefined; + if (parent?.pid) takeResumeEntry(node.store, cli, parent.pid, parent.pid_start, sid); + emit(cli, "SessionStart", "[mbx] Mailbox context could not be restored. Call mbx_whoami, then mbx_catchup."); + return; + } // Inspect provider capabilities and processes before the lease transaction. No directory-based session discovery. const host = cli === "claude" ? detectHost(process.ppid) : null; const watch = event === "session-start" && noPush(cli, !!host && (host.channel || host.socket), @@ -1647,8 +1655,8 @@ async function hook(node: MbxNode, event: string | undefined, cli: string) { } if (event === "session-start") { // T185: a bound Claude SessionStart is itself the entry (startup, resume, clear, compact; T184 spec C1): it carries the - // resume notice even on an unchanged id, because /clear and /compact do not re-send the MCP instructions. A fork only - // records the entry (D4/C2), so no notice is deferred to its first prompt either. T186: a bound Codex SessionStart is + // resume notice even on an unchanged id, because /clear and /compact do not re-send the MCP instructions. A fork + // returned before this bind. T186: a bound Codex SessionStart is // the entry too (a missing source counts as startup); its payload has no event id, so clear/compact re-inject on the // same thread id and enteredNow only stops the following prompt from adding a second copy. const hookSource = typeof input.source === "string" && input.source ? input.source : "startup"; @@ -1866,6 +1874,25 @@ async function hook(node: MbxNode, event: string | undefined, cli: string) { emit(cli, "PostToolUse", stillResuming); return; } + // T185: a name held live by another session is a conflict, not an invitation to claim. Same-session and released + // leases fall through so T573 resuming and T503 disconnect stay as they are. Nothing here writes. + const busyHeld = (): string | null => { + if (cli !== "claude" || !sid || (event !== "session-start" && event !== "prompt")) return null; + const name = node.store.get(`name:${cli}:${sid}`); + if (!name || !NAME_RE.test(name) || node.store.get(`identity-conflict:${name}`) !== undefined) return null; + const lease = node.store.db.prepare("SELECT * FROM identity_leases WHERE name=?").get(name) as IdentityLease | undefined; + if (!lease || lease.released_at !== null || (lease.cli === cli && lease.session_id === sid)) return null; + const availability = identityAvailability({ + lease, evidence: inspectLeaseProcess(lease.holder_pid), + activity: parseActivity(node.store.get(activityKey(name))), now: Date.now(), + }); + if (availability.claimable || availability.takeover || availability.coUse || !availability.reason.startsWith("held by live ")) return null; + return `[mbx] ${availability.reason}.`; + }; + try { + const busy = busyHeld(); + if (busy) { emit(cli, event === "session-start" ? "SessionStart" : "UserPromptSubmit", busy); return; } + } catch { /* a lookup failure keeps the existing unbound guidance */ } const disconnected = grokUnboundPost ? null : hookDisconnectedNote(node, cli, sid, process.ppid); const multi = cli === "kimi" && kimiMultiHost(process.ppid); // A conversation in a multi-conversation Kimi host can't be matched to its mbx server from here: hand it a bind diff --git a/test/hooks.test.ts b/test/hooks.test.ts index 80cf670..cb44bb7 100644 --- a/test/hooks.test.ts +++ b/test/hooks.test.ts @@ -440,10 +440,10 @@ test("T185: every Claude SessionStart source but fork is an entry with one bound test("T185: a Claude fork leaves the parent session row and prints the MCP fallback", async t => { const { n, run } = await holder(t, "claude"); assert.equal(run("prompt", "parent", "claude", {}, { prompt: "bind the parent" }).status, 0); - const rows = () => n.store.db.prepare("SELECT agent,cli,session_id,pid,session_key FROM sessions ORDER BY cli,session_id").all(); + const rows = () => n.store.db.prepare("SELECT agent,cli,session_id,pid,session_key FROM sessions ORDER BY cli,session_id").all() as { agent: string; cli: string; session_id: string; pid: number | null; session_key: string | null }[]; const leases = () => n.store.db.prepare("SELECT name,cli,session_id,released_at,token FROM identity_leases ORDER BY name").all(); const before = rows(), leaseBefore = leases(); - assert.deepEqual(before.map((r: { session_id: string }) => r.session_id), ["parent"]); + assert.deepEqual(before.map((r) => r.session_id), ["parent"]); const r = run("session-start", "child", "claude", {}, { source: "fork" }); assert.equal(r.status, 0, r.stderr); assert.deepEqual(rows(), before, "fork leaves the sessions table and the parent row untouched");