From b4ece3c6791f3b13e4ee886f7cac789f357aa663 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 14:46:57 -0700 Subject: [PATCH 1/5] fix(T699): one liveness answer for agents, roster and send receipts `agentmbx agents` read the session binding (liveAgents) while `mbx_agents` and the send receipt read the identity lease. When OpenCode evicted an idle MCP, the lease was released while the serve kept the session row alive, so the CLI showed "live" and the receipt showed "offline (released)" for hours. Liveness is now one answer (src/liveness.ts) composed from the lease, the live session row and the serve kind: - a released or expired lease whose session still runs is `idle` with its reason (claimable; wakeable when the T688 shared service or a registered T659 poller can start its next turn, which re-claims the mailbox); - the reason text uses only evidence every process can read, so all three surfaces agree; - `agentmbx agents` prints the shared state plus a trailing reason column and keeps shell senders visible as live; - mbx_agents/mbx_identity list follow through listIdentityStatus, and the receipt reports live-next-prompt/live-wake instead of offline; - the 30-minute lease idle TTL and the receipt's 3-minute heartbeat rule are unchanged. Adds test/liveness-agreement.test.ts covering held, released+service, released+standalone-poller, released+dead and unknown evidence. Checks: tsc --noEmit, npm run build (dist committed), identity-status, roster, opencode-poll-liveness, project-mailbox, liveness-activity, opencode-poller-wake, recipient-truth, agents-directory, chosen-identity, identity-cleanup, doctor-identities, delivery-guidance, cli-lease, mcp-leases, hooks, doctor-sessions, stranded, codex-activity, wake-dispatch, wake-contract, wake-project-mail, mcp-connection, mcp-lifecycle, cli-identity-handover. --- CHANGELOG.md | 2 + dist/cli.js | 9 +- dist/identity-status.js | 36 +++-- dist/liveness.js | 177 ++++++++++++++++++++++++ dist/node.js | 33 ++--- dist/opencode-poller.js | 38 ++++- dist/receipts.js | 106 +++++++------- dist/roster.js | 4 +- src/cli.ts | 9 +- src/identity-status.ts | 31 +++-- src/liveness.ts | 236 ++++++++++++++++++++++++++++++++ src/node.ts | 29 ++-- src/opencode-poller.ts | 56 +++++++- src/receipts.ts | 107 +++++++++------ src/roster.ts | 4 +- test/liveness-agreement.test.ts | 123 +++++++++++++++++ 16 files changed, 845 insertions(+), 155 deletions(-) create mode 100644 dist/liveness.js create mode 100644 src/liveness.ts create mode 100644 test/liveness-agreement.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index c79b4ebb..f34951cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- **One liveness answer: a released lease whose session still runs is idle, not live and not offline (T699).** `agentmbx agents` read the session binding while `mbx_agents` and the send receipt read the identity lease, so an OpenCode MCP eviction released the lease and the surfaces disagreed for hours: the CLI said `live`, the receipt said `offline (released)`. A single `agentLiveness` answer (src/liveness.ts) now composes the lease, the live session row and the serve kind, and every surface uses it. A released or expired lease whose session still runs reads as `idle` with its reason (claimable, and wakeable when the T688 shared service or a registered T659 poller can start its next turn, which re-claims the mailbox). `agentmbx agents` prints that state plus a trailing reason column and keeps shell senders visible as `live`; `mbx_agents` and `mbx_identity list` follow; a send reports `live-next-prompt`/`live-wake` instead of `offline`. The 30-minute lease idle TTL and the receipt's 3-minute heartbeat rule are unchanged. + ## 0.5.48 (2026-10-17) - **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. diff --git a/dist/cli.js b/dist/cli.js index 232662bc..3361fa37 100644 --- a/dist/cli.js +++ b/dist/cli.js @@ -41,6 +41,7 @@ import { dispatchWakes, hasWakeAuthority, humanPromptKey, inboxCommand, isHumanP import { kimiMultiHost } from "./kimi-web.js"; import { bindInstruction, issueBindTicket } from "./bind-ticket.js"; import { activityKey, identityAvailability, parseActivity } from "./identity-availability.js"; +import { agentDirectoryStates } from "./liveness.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"; @@ -1040,8 +1041,12 @@ async function run(argv) { }); } case "agents": { - const live = node.liveAgents(); - return node.agents().forEach((a) => console.log(`${a.name}@${a.host}\t${a.host === node.host ? (live.has(a.name) ? "live" : "offline") : "remote"}\t${a.role ?? ""}\t${a.cli ?? ""}\t${a.last_seen ?? ""}\t${a.description ?? ""}`)); + // T699: the same liveness answer as mbx_agents and the send receipt, with its reason as a trailing column. + const states = agentDirectoryStates(node); + return node.agents().forEach((a) => { + const local = states.get(a.name); + return console.log(`${a.name}@${a.host}\t${a.host === node.host ? local?.state ?? "offline" : "remote"}\t${a.role ?? ""}\t${a.cli ?? ""}\t${a.last_seen ?? ""}\t${a.description ?? ""}\t${a.host === node.host ? local?.reason ?? "" : ""}`); + }); } case "status": { if (o.json && str("schema") === STATUS_V2_SCHEMA) { diff --git a/dist/identity-status.js b/dist/identity-status.js index ccaf0148..2e3355cc 100644 --- a/dist/identity-status.js +++ b/dist/identity-status.js @@ -3,7 +3,8 @@ import { existsSync, readFileSync } from "node:fs"; import { join } from "node:path"; import { DatabaseSync } from "node:sqlite"; import { inspectLeaseProcess } from "./identity-leases.js"; -import { activityKey, holderProviderView, identityAvailability, parseActivity, parseProviderRecord } from "./identity-availability.js"; +import { activityKey, holderProviderView, parseActivity, parseProviderRecord } from "./identity-availability.js"; +import { livenessSnapshot } from "./liveness.js"; import { procTable, recordedStartMatches } from "./proc.js"; import { notSlot } from "./project-slot.js"; import { resolveProject } from "./project-key.js"; @@ -20,6 +21,7 @@ export function listIdentityStatus(home, options = {}) { throw new Error("mailbox host configuration is invalid"); const db = new DatabaseSync(path, { readOnly: true }); const rows = new Map(), leases = [], conflicts = new Set(), activityByName = new Map(); + const sessionRows = []; const providerByHolder = new Map(); // T656: server process start by pid, and the server pids each (cli, real session) binding names — read in the tx below. const serverStartByPid = new Map(); @@ -133,6 +135,8 @@ export function listIdentityStatus(home, options = {}) { } if (schema >= 2) leases.push(...db.prepare("SELECT * FROM identity_leases").all()); + for (const r of db.prepare("SELECT agent,cli,session_id,pid,pid_start,updated_at FROM sessions ORDER BY updated_at DESC").all()) + sessionRows.push(r); db.exec("COMMIT"); } finally { @@ -169,17 +173,30 @@ export function listIdentityStatus(home, options = {}) { } return holderProviderView((options.processTable ?? procTable)(), lease.holder_pid, caller.providerPid, recorded, evidence, caller.canonicalHarness); }; + // T699: one liveness answer for the whole inventory, from this snapshot: the caller's own lease is held by it, + // every other name gets the shared answer (a released lease whose session still runs is idle, not held or absent). + const evidenceByPid = new Map(); + for (const lease of leases) { + if (lease.released_at !== null || now - lease.heartbeat_at >= lease.idle_ttl) + continue; + try { + evidenceByPid.set(lease.holder_pid, inspect(lease.holder_pid)); + } + catch { /* failed inspection is unknown */ } + } + const evidenceOf = (pid) => evidenceByPid.get(pid) ?? { alive: null, start: null }; + const callerLease = options.caller ? leases.find((l) => l.cli === options.caller.cli && namesCallerSession(l, options.caller)) : undefined; + const caller = options.caller ? { ...options.caller, ...(callerLease ? sameSessionProvider(callerLease, options.caller) : {}) } : undefined; + const liveness = livenessSnapshot({ + names: leases.map((l) => l.name), leases, sessions: sessionRows, now, + activity: (name) => activityByName.get(name) ?? null, conflicts, evidence: evidenceOf, + ...(caller ? { caller } : {}), scope: home, + }); for (const lease of leases) { const item = row(lease.name); activity(item, lease.heartbeat_at); activity(item, lease.released_at); - let evidence = { alive: null, start: null }; - if (lease.released_at === null && now - lease.heartbeat_at < lease.idle_ttl) { - try { - evidence = inspect(lease.holder_pid); - } - catch { /* failed inspection is unknown */ } - } + const evidence = evidenceOf(lease.holder_pid); // The caller's own lease (its MCP server is the holder process) is simply held by it: not "an older process of this // same session", and not something it should claim again (T316). // A shared transport (Codex, OpenCode) serves many conversations from one process, so the process alone is not the @@ -187,8 +204,7 @@ export function listIdentityStatus(home, options = {}) { const own = options.caller?.pid !== undefined && lease.released_at === null && lease.holder_pid === options.caller.pid && lease.cli === options.caller.cli && namesCallerSession(lease, options.caller) && evidence.alive !== false; const a = own ? { state: "live", claimable: false, reason: "held by this session" } - : identityAvailability({ lease, evidence, activity: activityByName.get(lease.name) ?? null, conflict: conflicts.has(lease.name), now, - caller: options.caller && { ...options.caller, ...sameSessionProvider(lease, options.caller) } }); + : liveness.get(lease.name) ?? { state: "offline", claimable: true, reason: "no session has held it under a lease" }; item.state = a.state === "live" ? (a.claimable ? "idle" : "held") : a.state === "idle" ? "idle" : a.state === "unknown" ? "unknown" : a.state === "conflict" ? "conflict" : "available"; item.claimable = a.claimable; item.reason = a.reason; diff --git a/dist/liveness.js b/dist/liveness.js new file mode 100644 index 00000000..3b5de480 --- /dev/null +++ b/dist/liveness.js @@ -0,0 +1,177 @@ +// T699: one answer to "is this mailbox being served, and how?" for `mbx_agents`, `mbx_identity list`, +// `agentmbx agents` and the send receipt. +// +// Liveness is not claimability. The lease still decides who may claim a name (identityAvailability), but a +// released lease is not the same thing as an absent session: OpenCode evicts an idle MCP about hourly and +// releases the lease while the serve keeps hosting the conversation (T659). That mailbox is idle — claimable, +// no holder right now, but wakeable and seen on the session's next turn — not "live" (nothing is serving it) +// and not "offline" (the session is still there). Both wrong answers came from asking different questions: +// `agentmbx agents` read the session binding, the lease-based surfaces read identity_leases. +// +// The reason text is deliberately built from evidence every process can read (lease row, session row, process +// table, serve kind), never from this process's wake-queue registry, so every surface says the same thing. +import { activityKey, identityAvailability, parseActivity } from "./identity-availability.js"; +import { identityLeaseStatus, inspectLeaseProcesses } from "./identity-leases.js"; +import { sessionRowLive } from "./node.js"; +import { wakeBindingsByAgent } from "./opencode-poller.js"; +const iso = (ms) => new Date(ms).toISOString(); +const sameTarget = (b, s) => !!s && s.cli === b.cli && s.session_id === b.session_id && s.pid === b.pid; +/** + * The one mapping from lease availability + a live binding to a liveness answer. Pure: callers pass the evidence + * they already read, so the same rows always produce the same token and reason. + */ +export function livenessOf(o) { + const a = identityAvailability(o); + if (a.state === "conflict") + return { state: "conflict", reason: a.reason, claimable: false }; + if (a.state === "live" || a.state === "unknown") { + return { state: a.state, reason: a.reason, claimable: a.claimable, ...(a.takeover ? { takeover: a.takeover } : {}), ...(a.coUse ? { coUse: a.coUse } : {}) }; + } + if (a.state === "idle") { + return { state: "idle", idleKind: "held", reason: a.reason, claimable: true, ...(a.takeover ? { takeover: a.takeover } : {}) }; + } + // `offline` (a released or expired lease) and `free` (no lease) are idle only while the same session still runs. + if (o.binding) { + const serve = sameTarget(o.binding, o.serve) ? o.serve : null; + return { + state: "idle", idleKind: "released", binding: o.binding, claimable: true, + wakeable: serve?.wakeable === true, + reason: releasedIdleReason(o.lease, a.reason, o.binding, serve), + }; + } + return { state: "offline", reason: a.reason, claimable: a.claimable, ...(a.takeover ? { takeover: a.takeover } : {}) }; +} +function releasedIdleReason(lease, expiredReason, b, serve) { + const why = !lease ? "no current lease" + : lease.released_at !== null + ? `${lease.release_reason && lease.release_reason !== "released" ? `expired (${lease.release_reason})` : "released"} ${iso(lease.released_at)}` + : expiredReason; + const host = serve ? (serve.kind === "service" ? ", shared service" : ", standalone serve") : ""; + return `${why}; its ${b.cli} session ${b.session_id} is still running (pid ${b.pid}${host}), so it is idle: mail waits for its next turn`; +} +/** The live binding of `lease`'s session when there is one, else the newest live binding of this name. */ +function pickBinding(lease, rows, now) { + const usable = rows.filter((r) => !r.session_id.startsWith("mcp-") && sessionRowLive({ pid: r.pid, pid_start: r.pid_start, updated_at: r.updated_at }, now)); + const exact = lease ? usable.find((r) => r.cli === lease.cli && r.session_id === lease.session_id) : undefined; + const row = exact ?? usable[0]; + return row && row.pid ? { cli: row.cli, session_id: row.session_id, pid: row.pid } : null; +} +/** How long a wake verdict may be reused. Receipt pages ask again per message; one binding cannot change that fast. */ +const WAKE_MEMO_MS = 1_000; +const WAKE_MEMO_MAX = 512; +const wakeMemo = new Map(); +const rowKey = (r) => `${r.cli}\0${r.session_id}\0${r.pid ?? ""}`; +/** + * The one liveness mapping for a whole page of names: bindings and wake verdicts are derived here, so the agent + * list, the roster and the send receipt cannot drift apart again. The wake verdict is reused for one second so a + * receipt page costs one process read, not one per line. + */ +export function livenessSnapshot(o) { + const unique = [...new Set(o.names)]; + const leases = new Map(o.leases.map((r) => [r.name, r])); + const rowsByName = new Map(); + for (const r of o.sessions) + rowsByName.set(r.agent, [...(rowsByName.get(r.agent) ?? []), r]); + const bindings = new Map(); + for (const name of unique) + bindings.set(name, pickBinding(leases.get(name), rowsByName.get(name) ?? [], o.now)); + // A wake verdict is only meaningful when no current lease serves the name: release, expiry or no lease at all. + const candidates = new Map(); + for (const name of unique) { + const lease = leases.get(name), binding = bindings.get(name); + if (!binding) + continue; + if (lease && identityLeaseStatus(lease, o.now, o.evidence(lease.holder_pid)).state !== "expired") + continue; + candidates.set(name, (rowsByName.get(name) ?? []).filter((r) => r.cli === binding.cli && r.session_id === binding.session_id)); + } + const serves = new Map(); + const fresh = new Map(); + for (const [name, rows] of candidates) { + const lease = leases.get(name), key = `${o.scope ?? ""}\0${lease?.token ?? ""}\0${lease?.released_at ?? ""}\0${rows.map(rowKey).join(",")}`; + const memo = wakeMemo.get(name); + if (memo && memo.key === key && o.now - memo.at < WAKE_MEMO_MS) + serves.set(name, memo.value); + else + fresh.set(name, rows); + } + if (fresh.size) { + const found = wakeBindingsByAgent(new Map(fresh), o.host, o.waiting); + for (const name of fresh.keys()) { + const value = found.get(name) ?? null; + const lease = leases.get(name); + const key = `${o.scope ?? ""}\0${lease?.token ?? ""}\0${lease?.released_at ?? ""}\0${(fresh.get(name) ?? []).map(rowKey).join(",")}`; + wakeMemo.set(name, { key, at: o.now, value }); + serves.set(name, value); + } + if (wakeMemo.size > WAKE_MEMO_MAX) + for (const k of wakeMemo.keys()) + if (wakeMemo.size > WAKE_MEMO_MAX) + wakeMemo.delete(k); + } + const out = new Map(); + for (const name of unique) { + const lease = leases.get(name); + out.set(name, livenessOf({ + lease, evidence: lease ? o.evidence(lease.holder_pid) : { alive: null, start: null }, + activity: o.activity(name), conflict: o.conflicts.has(name), now: o.now, + ...(o.caller ? { caller: o.caller } : {}), + binding: bindings.get(name), serve: serves.get(name) ?? null, + })); + } + return out; +} +/** The same liveness answer for many names, read from one store (and one process snapshot). */ +export function agentLivenessBatch(node, names, o = {}) { + const now = o.now ?? Date.now(); + const unique = [...new Set(names)]; + if (!unique.length) + return new Map(); + const marks = unique.map(() => "?").join(","); + const leases = node.store.db.prepare(`SELECT * FROM identity_leases WHERE name IN (${marks})`).all(...unique); + const sessions = node.store.db.prepare(`SELECT agent,cli,session_id,pid,pid_start,updated_at FROM sessions WHERE agent IN (${marks}) ORDER BY updated_at DESC`).all(...unique); + const holders = leases.flatMap((l) => l.released_at === null ? [l.holder_pid] : []); + const evidence = new Map(); + if (o.inspect) + for (const pid of holders) + evidence.set(pid, o.inspect(pid)); + else if (holders.length) + for (const [pid, e] of inspectLeaseProcesses(holders)) + evidence.set(pid, e); + return livenessSnapshot({ + names: unique, leases, sessions, now, + activity: (name) => parseActivity(node.store.get(activityKey(name))), + conflicts: new Set(unique.filter((name) => node.store.get(`identity-conflict:${name}`) !== undefined)), + evidence: (pid) => evidence.get(pid) ?? { alive: null, start: null }, + host: o.host, waiting: o.waiting, scope: node.home, + }); +} +/** The same answer for one name; every surface uses this or the batch, never its own rule. */ +export function agentLiveness(node, name, o = {}) { + return agentLivenessBatch(node, [name], o).get(name) ?? { state: "offline", reason: "no session has held it under a lease", claimable: true }; +} +/** + * The `agentmbx agents` display for this host's directory: local names carry the shared liveness token and reason. + * A shell participant (`agentmbx send --as`) has no session and no lease; it stays visible as live for two hours + * after its last send, exactly as before (T699 decision). + */ +export function agentDirectoryStates(node, o = {}) { + const now = o.now ?? Date.now(); + const records = node.agents(); + const local = records.filter((a) => a.host === node.host); + const answers = agentLivenessBatch(node, local.map((a) => a.name), o); + const since = new Date(now - 2 * 3_600_000).toISOString(); + const bound = new Set(node.store.db.prepare("SELECT DISTINCT agent FROM sessions").all().map((r) => r.agent)); + const out = new Map(); + for (const a of local) { + const l = answers.get(a.name); + // A shell participant (--as) never bound a session; it stays visible as live for two hours after its last send. + if (a.cli === "cli" && a.last_seen !== null && a.last_seen > since && !bound.has(a.name) && l?.state === "offline") { + const mins = Math.max(0, Math.round((now - Date.parse(a.last_seen)) / 60_000)); + out.set(a.name, { state: "live", reason: `shell sender, last seen ${mins} min ago` }); + } + else if (l) + out.set(a.name, { state: l.state, reason: l.reason }); + } + return out; +} diff --git a/dist/node.js b/dist/node.js index 36168fbe..67a6b081 100644 --- a/dist/node.js +++ b/dist/node.js @@ -57,6 +57,20 @@ export const alive = (pid) => { if (!pid) catch (e) { return e.code === "EPERM"; } }; +/** + * T699: a session row still belongs to the live process it was recorded for — the same display rule `sameSession` + * uses without `proof`. A row with a recorded birth must match it exactly; an older row counts only while fresh and + * alive. Liveness only: authorization still uses provenProcess (sameSession with `{proof: true}`). + */ +export function sessionRowLive(r, now = Date.now()) { + if (!r.pid || !Number.isSafeInteger(r.pid) || r.pid <= 0) + return false; + if (now - Date.parse(r.updated_at) > LIVE_AGENT_MS) + return false; + if (r.pid_start) + return sameProcess(r.pid, r.pid_start); + return now - Date.parse(r.updated_at) < SESSION_FRESH_MS && alive(r.pid); +} const WAKEABLE = new Set(["codex", "opencode"]); /** Idle Grok has no first-party inject (T385). `grok --help` has no queue command. `~/.grok/leader.sock` * is the leader for config and MCP watches, not a prompt socket (user-guide 26-config-reference.md). @@ -335,9 +349,7 @@ export class MbxNode { return false; // live sessions rebind every minute if (o.proof) return provenProcess(pid, r.pid_start); - if (r.pid_start) - return sameProcess(pid, r.pid_start); - return Date.now() - Date.parse(r.updated_at) < SESSION_FRESH_MS && alive(pid); + return sessionRowLive({ pid, pid_start: r.pid_start, updated_at: r.updated_at }); } /** The agent of the session this process runs inside (walks up the process tree), if any. */ callerAgent(ancestors) { @@ -459,21 +471,6 @@ export class MbxNode { out.add(r.agent); return out; } - /** For display: agents with a live session, plus shell senders (--as, no session) active in the last 2 h. */ - liveAgents() { - const out = new Set(), bound = new Set(); - for (const r of this.store.db.prepare("SELECT agent, pid, pid_start, updated_at FROM sessions").all()) { - bound.add(r.agent); - if (r.pid && this.sameSession(r.pid, r)) - out.add(r.agent); - } - // shell participants (no session binding ever) count while they are active - const since = new Date(Date.now() - SHELL_AGENT_MS).toISOString(); - for (const r of this.store.db.prepare("SELECT name FROM agents WHERE host=? AND last_seen > ?").all(this.host, since)) - if (!bound.has(r.name)) - out.add(r.name); - return out; - } /** How mail reaches this agent when its session is idle. */ deliveryMode(agent, sessions = this.sessionsFor(agent)) { const ss = sessions.filter((x) => x.pid && this.sameSession(x.pid, x)); diff --git a/dist/opencode-poller.js b/dist/opencode-poller.js index 8d808e10..f8507c23 100644 --- a/dist/opencode-poller.js +++ b/dist/opencode-poller.js @@ -10,7 +10,7 @@ // the service itself still hosts the session; wakeOpencode posts to it (and keeps T646's dual-host refusal). Nothing here reads or changes // mail: a wake carries ids only, and the session claims its identity again on the turn the wake starts. import { provenProcess, withProcSnapshot } from "./proc.js"; -import { opencodeHostOf } from "./opencode-provider.js"; +import { opencodeHostClassifier, opencodeHostOf } from "./opencode-provider.js"; import { opencodeWakeWaiting } from "./opencode-wake-queue.js"; const refused = () => Object.assign(new Error("opencode serve poller or session binding changed"), { code: "WAKE_IDENTITY_LOST" }); const binding = (s) => JSON.stringify([s.agent, s.cli, s.session_id, s.pid, s.pid_start, s.session_key, s.channel]); @@ -42,6 +42,42 @@ const serviceBindingLive = (node, session, host = opencodeHostOf) => bindingLive export function pollerTargetLive(node, session, host = opencodeHostOf) { return serveBindingLive(node, session, host) && opencodeWakeWaiting(session.session_id, session.pid); } +/** + * T699: the wake bindings behind many agents from one process snapshot (and one argv read with the batched host + * classifier), for the liveness surfaces. This is the same evidence `pollerWakeTargets` uses, without its run + * closures: display never takes mail or leases. A standalone poller registration is per-process, so a caller + * outside the daemon sees the binding (`kind`, `wakeable: false`) but cannot claim the wake path. + */ +export function wakeBindingsByAgent(sessionsByAgent, host = opencodeHostClassifier(), waiting = opencodeWakeWaiting) { + return withProcSnapshot(() => { + const out = new Map(); + for (const [agent, sessions] of sessionsByAgent) { + let best = null; + for (const s of sessions) { + if (s.agent !== agent || s.cli !== "opencode" || !s.pid || s.pid <= 1 || s.session_id.startsWith("mcp-")) + continue; + // The recorded birth must still be that process: a reused pid must not inherit a row it never wrote. + if (!provenProcess(s.pid, s.pid_start)) + continue; + const kind = host(s.pid); + if (kind !== "service" && kind !== "standalone") + continue; + const polling = kind === "standalone" && waiting(s.session_id, s.pid); + const candidate = kind === "service" + ? { cli: s.cli, session_id: s.session_id, pid: s.pid, kind, wakeable: true, via: "service" } + : { cli: s.cli, session_id: s.session_id, pid: s.pid, kind, wakeable: polling, via: polling ? "standalone-poller" : null }; + // A service binding hosts every session of the service; prefer it over a standalone row of the same agent. + if (!best || candidate.kind === "service" || (candidate.wakeable && !best.wakeable)) + best = candidate; + if (best.kind === "service") + break; + } + if (best) + out.set(agent, best); + } + return out; + }); +} /** * Poller targets for `agent`: bound standalone OpenCode sessions whose MCP is gone but whose plugin still polls, and (T688) * bound sessions of the shared service, which still hosts them. diff --git a/dist/receipts.js b/dist/receipts.js index b0c2ae0b..5e0c3f05 100644 --- a/dist/receipts.js +++ b/dist/receipts.js @@ -1,77 +1,64 @@ // Send-time recipient truth (T205): who a message actually reached, decided when it is sent, so a sender never mistakes -// "stored in a mailbox" for "an agent will see it". Liveness comes from the identity lease and its existing process -// evidence. Native push eligibility uses the dispatcher's exact lease/binding guard. -import { IdentityLeases, identityLeaseStatus } from "./identity-leases.js"; +// "stored in a mailbox" for "an agent will see it". Liveness is one answer from liveness.ts (T699): the lease, the +// session binding and the serve kind together. Native push eligibility uses the dispatcher's exact lease/binding guard. +import { identityLeaseStatus, inspectLeaseProcess } from "./identity-leases.js"; +import { agentLiveness } from "./liveness.js"; import { GROK_NO_PUSH } from "./node.js"; import {} from "./store.js"; import { hasWakeAuthority, liveWatcher, queuedBehindTurnAt, watcherKey } from "./wake.js"; import { captureWakeIdentity } from "./wake-identity.js"; -import { procTable } from "./proc.js"; import { activityKey, parseActivity, SHARED_IDLE_MS } from "./identity-availability.js"; import { connectorKey, retryErrorClass } from "./diagnostics.js"; import { remoteReceipts } from "./remote-receipts.js"; import { isSlotName } from "./project-slot.js"; import { projectLabel, projectMailboxBySlot, projectReader } from "./project-mailbox.js"; import { peerProjectBySlot } from "./project-directory.js"; -/** A holder renews its lease every minute; three missed beats mean its session is suspended or gone. */ -const STALE_HEARTBEAT_MS = 3 * 60_000; const iso = (ms) => new Date(ms).toISOString(); -/** Is a session holding `name` right now? Unknown process evidence with a fresh heartbeat counts as live (it is the - * lease's own rule for "unknown"); everything else says since when and who held it last. */ -export function mailboxLiveness(node, name, now = Date.now()) { +/** A holder renews its lease every minute; three missed beats mean its session is suspended or gone. Receipts are + * stricter than the 30-minute lease idle TTL: mail nobody is polling should read as undelivered sooner (T204). */ +const STALE_HEARTBEAT_MS = 3 * 60_000; +/** Is a session holding `name` right now — or still there with its lease released? One answer from liveness.ts + * (T699), so a receipt cannot call a wakeable session "offline" while the agent list calls it "live". */ +export function mailboxLiveness(node, name, now = Date.now(), o = {}) { if (isSlotName(name)) { // T493: a project mailbox has no holder of its own. Its reader is the project's current lead, so it is live exactly when that lead is. const reader = projectReader(node, name, new Date(now)); if (!reader) - return { live: false, detail: "this project mailbox is not allocated on this host" }; + return { live: false, state: "offline", reason: "this project mailbox is not allocated on this host", detail: "this project mailbox is not allocated on this host" }; const label = `project mailbox ${projectLabel(reader.mailbox)}`; - if (!reader.lead) - return { live: false, detail: `${label}: no owner-designated lead on ${node.host} to read it` }; - const lead = mailboxLiveness(node, reader.lead.agent, now); - return { live: lead.live, detail: `${label}, read by its lead ${reader.lead.agent}: ${lead.detail}` }; + if (!reader.lead) { + const reason = `${label}: no owner-designated lead on ${node.host} to read it`; + return { live: false, state: "offline", reason, detail: reason }; + } + const lead = mailboxLiveness(node, reader.lead.agent, now, o); + return { ...lead, detail: `${label}, read by its lead ${reader.lead.agent}: ${lead.detail}` }; } + const l = agentLiveness(node, name, { now, ...(o.inspect ? { inspect: o.inspect } : {}) }); const row = node.store.db.prepare("SELECT * FROM identity_leases WHERE name=?").get(name); + if (l.state === "conflict") + return { live: false, state: "conflict", reason: l.reason, detail: l.reason }; if (!row) - return { live: false, detail: "no session has ever held this mailbox; it is read when an agent claims it" }; - let state = "unknown", reason = null; - try { - ({ state, reason } = identityLeaseStatus(row, now, new IdentityLeases(node.store).processEvidence(row.holder_pid))); - } - catch { - // Evidence can't be inspected inside a transaction: judge by the lease row, the process table snapshot taken before - // it (a vanished holder is gone) and the heartbeat, which a live holder renews every minute (T204 review). - const table = procTable(); - if (row.released_at !== null || now - row.heartbeat_at >= row.idle_ttl) { - state = "expired"; - reason = row.released_at !== null ? null : "idle"; - } - else if (table.size && !table.has(row.holder_pid)) { - state = "expired"; - reason = "dead"; - } - else if (now - row.heartbeat_at >= STALE_HEARTBEAT_MS) { - state = "expired"; - reason = `no heartbeat for ${Math.round((now - row.heartbeat_at) / 60_000)} min`; - } - // Its pid is running and it renewed within three minutes: as live as an advisory receipt can tell. - else if (table.has(row.holder_pid)) - state = "live"; - } - if ((state === "live" || state === "unknown") && now - row.heartbeat_at >= STALE_HEARTBEAT_MS) { - state = "expired"; - reason = `no heartbeat for ${Math.round((now - row.heartbeat_at) / 60_000)} min`; - } + return { live: false, state: "offline", reason: l.reason, detail: "no session has ever held this mailbox; it is read when an agent claims it" }; const holder = `${row.cli} session ${row.session_id.slice(0, 12)}`; - if (state === "live" || state === "unknown") { - const activity = parseActivity(node.store.get(activityKey(name))); - const quiet = activity?.shared && now - activity.at >= SHARED_IDLE_MS - ? `; its conversation has made no mbx call for ${Math.round((now - activity.at) / 60_000)} min and may have ended (no hook or wake-poll activity either)` : ""; - return { live: true, detail: state === "live" ? `held by ${holder}${quiet}` - : `held by ${holder} (process unconfirmed, heartbeat ${Math.round((now - row.heartbeat_at) / 1000)} s ago)${quiet}` }; + const silent = row.released_at === null && now - row.heartbeat_at >= STALE_HEARTBEAT_MS; + if (l.state === "offline" || silent) { + const status = row.released_at === null ? identityLeaseStatus(row, now, inspectLeaseProcess(row.holder_pid)) : null; + const why = row.release_reason ?? (silent ? `no heartbeat for ${Math.round((now - row.heartbeat_at) / 60_000)} min` : status?.reason ?? "released"); + return { live: false, state: "offline", reason: l.reason, + detail: `no live session since ${iso(row.released_at ?? row.heartbeat_at)} (last holder ${holder}, ${why})` }; } - const since = row.released_at ?? row.heartbeat_at; - const why = row.release_reason ?? reason ?? "released"; - return { live: false, detail: `no live session since ${iso(since)} (last holder ${holder}, ${why})` }; + const activity = parseActivity(node.store.get(activityKey(name))); + const quiet = activity?.shared ? now - activity.at : null; + const note = quiet !== null && quiet >= SHARED_IDLE_MS && (l.state === "live" || (l.state === "idle" && l.idleKind === "held")) + ? `; its conversation has made no mbx call for ${Math.round(quiet / 60_000)} min and may have ended (no hook or wake-poll activity either)` : ""; + const detail = l.idleKind === "released" ? l.reason + : l.state === "unknown" ? `held by ${holder} (process unconfirmed, heartbeat ${Math.round((now - row.heartbeat_at) / 1000)} s ago)${note}` + : `held by ${holder}${note}`; + return { + live: true, state: l.state, reason: l.reason, detail, + ...(l.idleKind === "released" ? { released: true } : {}), + ...(l.wakeable !== undefined ? { wakeable: l.wakeable } : {}), + }; } /** "addressed to the project mailbox X; " when `name` is a project slot that `host` advertises, else nothing (T493). */ const projectBoxNote = (node, name, host) => { @@ -118,6 +105,17 @@ export function recipientReceipts(node, msgId, targets, now = Date.now()) { state = "offline"; detail = live.detail; } + else if (live.released) { + // T699: the lease was released (or expired) but the session still runs. That is idle, not offline: mail waits + // for its next turn, and a proven wake path (a shared service, or a registered poller) can start that turn. + const mode = node.deliveryMode(name); + const why = !m ? "message not found" : !node.wantsWake(name, m) ? "this kind does not wake (status, or no needs_reply/mention)" + : !live.wakeable ? "the session is running without a wake path proven from this process" + : !hasWakeAuthority(node, name, m) ? "no owner policy lets this sender wake it" : null; + state = why ? "live-next-prompt" : "live-wake"; + detail = why ? `${live.detail}; seen on its next prompt: ${why}` + : `${live.detail}; eligible for wake (${mode}); the turn re-claims the mailbox; dispatcher admission pending`; + } else { const sessions = node.sessionsFor(name); const verified = sessions.filter(s => captureWakeIdentity(node, s)); @@ -196,10 +194,10 @@ export function deliveryReceipts(node, m, now = Date.now()) { const dids = new Map(node.store.db.prepare("SELECT json_extract(detail,'$.recipient') r, json_extract(detail,'$.did') d FROM audit WHERE event='peer_action' AND json_extract(detail,'$.msg')=? ORDER BY at") .all(m.id).map((x) => [x.r, x.d])); const local = node.store.db.prepare("SELECT agent,state,updated_at,note FROM deliveries WHERE msg_id=? ORDER BY agent").all(m.id).map((d) => { - const live = d.agent === "owner" ? { live: true, detail: "the owner" } : mailboxLiveness(node, d.agent, now); + const live = d.agent === "owner" ? { live: true, state: "live", detail: "the owner" } : mailboxLiveness(node, d.agent, now); const behind = d.state === "delivered" || d.state === "notified" ? queuedBehindTurnAt(node.store, d.agent, m.id) : null; return { address: `${d.agent}@${node.host}`, state: d.state, updated_at: d.updated_at, note: d.note, did: dids.get(d.agent) ?? null, - liveness: live.live ? `live: ${live.detail}` : `offline: ${live.detail}`, + liveness: `${live.state}: ${live.detail}`, ...(behind ? { wake: `${OPENCODE_BEHIND_TURN_NOTE} (admitted ${behind.slice(0, 19)}Z)` } : {}) }; }); // Remote recipients: the envelope's addressees on other hosts; a queued outbox row means not yet accepted there. diff --git a/dist/roster.js b/dist/roster.js index 8ae97126..e97be135 100644 --- a/dist/roster.js +++ b/dist/roster.js @@ -114,8 +114,10 @@ export function rosterText(r) { const lines = r.agents.map((a) => { const led = a.lead_of.length ? ` LEAD of ${a.lead_of.map((p) => basename(p) || p).join(", ")}` : ""; const cli = a.harness ?? a.cli; + // T699: an idle row now also covers a released lease whose session still runs; its reason is the point. + const reason = a.state === "idle" || a.state === "unknown" ? ` reason: ${oneLine(a.reason)}` : ""; return `${a.address} ${a.state}${led}${a.self ? " (you)" : ""}${a.seen_here ? " seen here (not a member)" : ""}${a.role ? ` role:${a.role}` : ""}${a.task ? ` task:${a.task}` : ""}${a.lane ? ` lane:${a.lane}` : ""}${cli ? ` (${cli})` : ""}` - + ` last seen ${a.last_seen ?? "never"}${a.description ? ` — ${oneLine(a.description)}` : ""}`; + + ` last seen ${a.last_seen ?? "never"}${a.description ? ` — ${oneLine(a.description)}` : ""}${reason}`; }); const leadLine = !r.scope.project ? "lead: no project" : r.lead ? `lead: ${r.lead.address} until ${r.lead.exp}` : "lead: none"; const hidden = r.scope.hidden ? `\n${r.scope.hidden} retired or generated name(s) hidden; pass all:true to list them.` : ""; diff --git a/src/cli.ts b/src/cli.ts index 7440cbcf..056a633e 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -42,6 +42,7 @@ import { dispatchWakes, hasWakeAuthority, humanPromptKey, inboxCommand, isHumanP import { kimiMultiHost } from "./kimi-web.ts"; import { bindInstruction, issueBindTicket } from "./bind-ticket.ts"; import { activityKey, identityAvailability, parseActivity } from "./identity-availability.ts"; +import { agentDirectoryStates } from "./liveness.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"; @@ -860,8 +861,12 @@ async function run(argv: string[]) { }); } case "agents": { - const live = node.liveAgents(); - return node.agents().forEach((a) => console.log(`${a.name}@${a.host}\t${a.host === node.host ? (live.has(a.name) ? "live" : "offline") : "remote"}\t${a.role ?? ""}\t${a.cli ?? ""}\t${a.last_seen ?? ""}\t${a.description ?? ""}`)); + // T699: the same liveness answer as mbx_agents and the send receipt, with its reason as a trailing column. + const states = agentDirectoryStates(node); + return node.agents().forEach((a) => { + const local = states.get(a.name); + return console.log(`${a.name}@${a.host}\t${a.host === node.host ? local?.state ?? "offline" : "remote"}\t${a.role ?? ""}\t${a.cli ?? ""}\t${a.last_seen ?? ""}\t${a.description ?? ""}\t${a.host === node.host ? local?.reason ?? "" : ""}`); + }); } case "status": { if (o.json && str("schema") === STATUS_V2_SCHEMA) { diff --git a/src/identity-status.ts b/src/identity-status.ts index 4490e266..1b96f7c6 100644 --- a/src/identity-status.ts +++ b/src/identity-status.ts @@ -3,7 +3,8 @@ import { existsSync, readFileSync } from "node:fs"; import { join } from "node:path"; import { DatabaseSync } from "node:sqlite"; import { inspectLeaseProcess, type IdentityLease, type ProcessEvidence } from "./identity-leases.ts"; -import { activityKey, holderProviderView, identityAvailability, parseActivity, parseProviderRecord, type LeaseActivity, type RecordedProvider } from "./identity-availability.ts"; +import { activityKey, holderProviderView, parseActivity, parseProviderRecord, type LeaseActivity, type RecordedProvider } from "./identity-availability.ts"; +import { livenessSnapshot, type AgentLiveness, type SessionRow } from "./liveness.ts"; import { procTable, recordedStartMatches } from "./proc.ts"; import { notSlot } from "./project-slot.ts"; import { resolveProject } from "./project-key.ts"; @@ -43,6 +44,7 @@ export function listIdentityStatus(home: string, options: { now?: number; inspec if (typeof config.host !== "string" || !config.host) throw new Error("mailbox host configuration is invalid"); const db = new DatabaseSync(path, { readOnly: true }); const rows = new Map(), leases: IdentityLease[] = [], conflicts = new Set(), activityByName = new Map(); + const sessionRows: SessionRow[] = []; const providerByHolder = new Map(); // T656: server process start by pid, and the server pids each (cli, real session) binding names — read in the tx below. const serverStartByPid = new Map(); @@ -122,6 +124,8 @@ export function listIdentityStatus(home: string, options: { now?: number; inspec } } if (schema >= 2) leases.push(...db.prepare("SELECT * FROM identity_leases").all() as unknown as IdentityLease[]); + for (const r of db.prepare("SELECT agent,cli,session_id,pid,pid_start,updated_at FROM sessions ORDER BY updated_at DESC").all() as unknown as SessionRow[]) + sessionRows.push(r); db.exec("COMMIT"); } finally { db.close(); } // Process discovery happens after releasing the read transaction. Listing is advisory; @@ -145,21 +149,32 @@ export function listIdentityStatus(home: string, options: { now?: number; inspec if (recorded) { try { evidence = inspect(recorded.pid); } catch { evidence = { alive: null, start: null }; } } return holderProviderView((options.processTable ?? procTable)(), lease.holder_pid, caller.providerPid, recorded, evidence, caller.canonicalHarness); }; + // T699: one liveness answer for the whole inventory, from this snapshot: the caller's own lease is held by it, + // every other name gets the shared answer (a released lease whose session still runs is idle, not held or absent). + const evidenceByPid = new Map(); + for (const lease of leases) { + if (lease.released_at !== null || now - lease.heartbeat_at >= lease.idle_ttl) continue; + try { evidenceByPid.set(lease.holder_pid, inspect(lease.holder_pid)); } catch { /* failed inspection is unknown */ } + } + const evidenceOf = (pid: number): ProcessEvidence => evidenceByPid.get(pid) ?? { alive: null, start: null }; + const callerLease = options.caller ? leases.find((l) => l.cli === options.caller!.cli && namesCallerSession(l, options.caller!)) : undefined; + const caller = options.caller ? { ...options.caller, ...(callerLease ? sameSessionProvider(callerLease, options.caller) : {}) } : undefined; + const liveness = livenessSnapshot({ + names: leases.map((l) => l.name), leases, sessions: sessionRows, now, + activity: (name) => activityByName.get(name) ?? null, conflicts, evidence: evidenceOf, + ...(caller ? { caller } : {}), scope: home, + }); for (const lease of leases) { const item = row(lease.name); activity(item, lease.heartbeat_at); activity(item, lease.released_at); - let evidence: ProcessEvidence = { alive: null, start: null }; - if (lease.released_at === null && now - lease.heartbeat_at < lease.idle_ttl) { - try { evidence = inspect(lease.holder_pid); } catch { /* failed inspection is unknown */ } - } + const evidence = evidenceOf(lease.holder_pid); // The caller's own lease (its MCP server is the holder process) is simply held by it: not "an older process of this // same session", and not something it should claim again (T316). // A shared transport (Codex, OpenCode) serves many conversations from one process, so the process alone is not the // caller: the lease must also name this conversation's session. const own = options.caller?.pid !== undefined && lease.released_at === null && lease.holder_pid === options.caller.pid && lease.cli === options.caller.cli && namesCallerSession(lease, options.caller) && evidence.alive !== false; - const a = own ? { state: "live" as const, claimable: false, reason: "held by this session" } - : identityAvailability({ lease, evidence, activity: activityByName.get(lease.name) ?? null, conflict: conflicts.has(lease.name), now, - caller: options.caller && { ...options.caller, ...sameSessionProvider(lease, options.caller) } }); + const a: AgentLiveness = own ? { state: "live", claimable: false, reason: "held by this session" } + : liveness.get(lease.name) ?? { state: "offline", claimable: true, reason: "no session has held it under a lease" }; item.state = a.state === "live" ? (a.claimable ? "idle" : "held") : a.state === "idle" ? "idle" : a.state === "unknown" ? "unknown" : a.state === "conflict" ? "conflict" : "available"; item.claimable = a.claimable; item.reason = a.reason; diff --git a/src/liveness.ts b/src/liveness.ts new file mode 100644 index 00000000..12f19444 --- /dev/null +++ b/src/liveness.ts @@ -0,0 +1,236 @@ +// T699: one answer to "is this mailbox being served, and how?" for `mbx_agents`, `mbx_identity list`, +// `agentmbx agents` and the send receipt. +// +// Liveness is not claimability. The lease still decides who may claim a name (identityAvailability), but a +// released lease is not the same thing as an absent session: OpenCode evicts an idle MCP about hourly and +// releases the lease while the serve keeps hosting the conversation (T659). That mailbox is idle — claimable, +// no holder right now, but wakeable and seen on the session's next turn — not "live" (nothing is serving it) +// and not "offline" (the session is still there). Both wrong answers came from asking different questions: +// `agentmbx agents` read the session binding, the lease-based surfaces read identity_leases. +// +// The reason text is deliberately built from evidence every process can read (lease row, session row, process +// table, serve kind), never from this process's wake-queue registry, so every surface says the same thing. +import { activityKey, identityAvailability, parseActivity, type LeaseActivity } from "./identity-availability.ts"; +import { identityLeaseStatus, inspectLeaseProcesses, type IdentityLease, type ProcessEvidence } from "./identity-leases.ts"; +import { sessionRowLive, type MbxNode } from "./node.ts"; +import { wakeBindingsByAgent, type WakeBinding } from "./opencode-poller.ts"; +import type { OpencodeHost } from "./opencode-provider.ts"; + +export type LivenessState = "live" | "idle" | "unknown" | "offline" | "conflict"; + +/** A bound session of this name that is still running (the display rule `sessionRowLive`). */ +export interface SessionBinding { cli: string; session_id: string; pid: number } + +export interface AgentLiveness { + state: LivenessState; + reason: string; + claimable: boolean; + /** + * `idle` has two causes, and the receipt treats them differently: `held` still has a current lease (a quiet + * shared conversation, T608/T383) so the ordinary capture/wake path applies; `released` has no usable lease + * and the session is only reachable through its live binding (T659/T688). + */ + idleKind?: "held" | "released"; + binding?: SessionBinding; + /** T659/T688: a wake path is proven for that binding (shared service, or a registered standalone poller). */ + wakeable?: boolean; + takeover?: "idle-conversation" | "same-session"; + coUse?: true; +} + +type AvailabilityInput = Parameters[0]; +export interface LivenessInput extends AvailabilityInput { + /** A live binding of this name, when one exists. Without it a released lease is plainly offline. */ + binding?: SessionBinding | null; + /** The proven serve hosting that binding (opencode only). */ + serve?: WakeBinding | null; +} + +const iso = (ms: number) => new Date(ms).toISOString(); + +const sameTarget = (b: SessionBinding, s: WakeBinding | null | undefined): s is WakeBinding => + !!s && s.cli === b.cli && s.session_id === b.session_id && s.pid === b.pid; + +/** + * The one mapping from lease availability + a live binding to a liveness answer. Pure: callers pass the evidence + * they already read, so the same rows always produce the same token and reason. + */ +export function livenessOf(o: LivenessInput): AgentLiveness { + const a = identityAvailability(o); + if (a.state === "conflict") return { state: "conflict", reason: a.reason, claimable: false }; + if (a.state === "live" || a.state === "unknown") { + return { state: a.state, reason: a.reason, claimable: a.claimable, ...(a.takeover ? { takeover: a.takeover } : {}), ...(a.coUse ? { coUse: a.coUse } : {}) }; + } + if (a.state === "idle") { + return { state: "idle", idleKind: "held", reason: a.reason, claimable: true, ...(a.takeover ? { takeover: a.takeover } : {}) }; + } + // `offline` (a released or expired lease) and `free` (no lease) are idle only while the same session still runs. + if (o.binding) { + const serve = sameTarget(o.binding, o.serve) ? o.serve : null; + return { + state: "idle", idleKind: "released", binding: o.binding, claimable: true, + wakeable: serve?.wakeable === true, + reason: releasedIdleReason(o.lease, a.reason, o.binding, serve), + }; + } + return { state: "offline", reason: a.reason, claimable: a.claimable, ...(a.takeover ? { takeover: a.takeover } : {}) }; +} + +function releasedIdleReason(lease: IdentityLease | undefined, expiredReason: string, b: SessionBinding, serve: WakeBinding | null): string { + const why = !lease ? "no current lease" + : lease.released_at !== null + ? `${lease.release_reason && lease.release_reason !== "released" ? `expired (${lease.release_reason})` : "released"} ${iso(lease.released_at)}` + : expiredReason; + const host = serve ? (serve.kind === "service" ? ", shared service" : ", standalone serve") : ""; + return `${why}; its ${b.cli} session ${b.session_id} is still running (pid ${b.pid}${host}), so it is idle: mail waits for its next turn`; +} + +/** One session row as read for a liveness snapshot (process evidence is checked separately). */ +export interface SessionRow { + agent: string; cli: string; session_id: string; pid: number | null; pid_start: string | null; updated_at: string; +} + +/** The live binding of `lease`'s session when there is one, else the newest live binding of this name. */ +function pickBinding(lease: IdentityLease | undefined, rows: readonly SessionRow[], now: number): SessionBinding | null { + const usable = rows.filter((r) => !r.session_id.startsWith("mcp-") && sessionRowLive({ pid: r.pid, pid_start: r.pid_start, updated_at: r.updated_at }, now)); + const exact = lease ? usable.find((r) => r.cli === lease.cli && r.session_id === lease.session_id) : undefined; + const row = exact ?? usable[0]; + return row && row.pid ? { cli: row.cli, session_id: row.session_id, pid: row.pid } : null; +} + +export interface LivenessOptions { + now?: number; + /** Injected evidence (tests, and identity inventory's own snapshot). Omitted, one batched process read per call. */ + inspect?: (pid: number) => ProcessEvidence; + /** Injected serve classification (tests). Omitted, one argv read per batch. */ + host?: (pid: number | null | undefined) => OpencodeHost; + /** Injected poller registration (tests). Omitted, this process's wake queue. */ + waiting?: (sessionID: string, pid: number) => boolean; +} + +/** Everything a liveness answer needs, already read. The identity inventory builds one from its own snapshot. */ +export interface LivenessSnapshot { + /** Names the caller wants answered. */ + names: readonly string[]; + leases: readonly IdentityLease[]; + sessions: readonly SessionRow[]; + now: number; + activity: (name: string) => LeaseActivity | null; + conflicts: ReadonlySet; + evidence: (pid: number) => ProcessEvidence; + caller?: AvailabilityInput["caller"]; + host?: (pid: number | null | undefined) => OpencodeHost; + waiting?: (sessionID: string, pid: number) => boolean; + /** Distinguishes stores in the wake memo (a home path). */ + scope?: string; +} + +/** How long a wake verdict may be reused. Receipt pages ask again per message; one binding cannot change that fast. */ +const WAKE_MEMO_MS = 1_000; +const WAKE_MEMO_MAX = 512; +const wakeMemo = new Map(); +const rowKey = (r: { cli: string; session_id: string; pid: number | null }) => `${r.cli}\0${r.session_id}\0${r.pid ?? ""}`; + +/** + * The one liveness mapping for a whole page of names: bindings and wake verdicts are derived here, so the agent + * list, the roster and the send receipt cannot drift apart again. The wake verdict is reused for one second so a + * receipt page costs one process read, not one per line. + */ +export function livenessSnapshot(o: LivenessSnapshot): Map { + const unique = [...new Set(o.names)]; + const leases = new Map(o.leases.map((r) => [r.name, r])); + const rowsByName = new Map(); + for (const r of o.sessions) rowsByName.set(r.agent, [...(rowsByName.get(r.agent) ?? []), r]); + const bindings = new Map(); + for (const name of unique) bindings.set(name, pickBinding(leases.get(name), rowsByName.get(name) ?? [], o.now)); + + // A wake verdict is only meaningful when no current lease serves the name: release, expiry or no lease at all. + const candidates = new Map(); + for (const name of unique) { + const lease = leases.get(name), binding = bindings.get(name); + if (!binding) continue; + if (lease && identityLeaseStatus(lease, o.now, o.evidence(lease.holder_pid)).state !== "expired") continue; + candidates.set(name, (rowsByName.get(name) ?? []).filter((r) => r.cli === binding.cli && r.session_id === binding.session_id)); + } + const serves = new Map(); + const fresh = new Map(); + for (const [name, rows] of candidates) { + const lease = leases.get(name), key = `${o.scope ?? ""}\0${lease?.token ?? ""}\0${lease?.released_at ?? ""}\0${rows.map(rowKey).join(",")}`; + const memo = wakeMemo.get(name); + if (memo && memo.key === key && o.now - memo.at < WAKE_MEMO_MS) serves.set(name, memo.value); + else fresh.set(name, rows); + } + if (fresh.size) { + const found = wakeBindingsByAgent(new Map(fresh), o.host, o.waiting); + for (const name of fresh.keys()) { + const value = found.get(name) ?? null; + const lease = leases.get(name); + const key = `${o.scope ?? ""}\0${lease?.token ?? ""}\0${lease?.released_at ?? ""}\0${(fresh.get(name) ?? []).map(rowKey).join(",")}`; + wakeMemo.set(name, { key, at: o.now, value }); + serves.set(name, value); + } + if (wakeMemo.size > WAKE_MEMO_MAX) for (const k of wakeMemo.keys()) if (wakeMemo.size > WAKE_MEMO_MAX) wakeMemo.delete(k); + } + + const out = new Map(); + for (const name of unique) { + const lease = leases.get(name); + out.set(name, livenessOf({ + lease, evidence: lease ? o.evidence(lease.holder_pid) : { alive: null, start: null }, + activity: o.activity(name), conflict: o.conflicts.has(name), now: o.now, + ...(o.caller ? { caller: o.caller } : {}), + binding: bindings.get(name), serve: serves.get(name) ?? null, + })); + } + return out; +} + +/** The same liveness answer for many names, read from one store (and one process snapshot). */ +export function agentLivenessBatch(node: MbxNode, names: readonly string[], o: LivenessOptions = {}): Map { + const now = o.now ?? Date.now(); + const unique = [...new Set(names)]; + if (!unique.length) return new Map(); + const marks = unique.map(() => "?").join(","); + const leases = node.store.db.prepare(`SELECT * FROM identity_leases WHERE name IN (${marks})`).all(...unique) as unknown as IdentityLease[]; + const sessions = node.store.db.prepare(`SELECT agent,cli,session_id,pid,pid_start,updated_at FROM sessions WHERE agent IN (${marks}) ORDER BY updated_at DESC`).all(...unique) as unknown as SessionRow[]; + const holders = leases.flatMap((l) => l.released_at === null ? [l.holder_pid] : []); + const evidence = new Map(); + if (o.inspect) for (const pid of holders) evidence.set(pid, o.inspect(pid)); + else if (holders.length) for (const [pid, e] of inspectLeaseProcesses(holders)) evidence.set(pid, e); + return livenessSnapshot({ + names: unique, leases, sessions, now, + activity: (name) => parseActivity(node.store.get(activityKey(name))), + conflicts: new Set(unique.filter((name) => node.store.get(`identity-conflict:${name}`) !== undefined)), + evidence: (pid) => evidence.get(pid) ?? { alive: null, start: null }, + host: o.host, waiting: o.waiting, scope: node.home, + }); +} + +/** The same answer for one name; every surface uses this or the batch, never its own rule. */ +export function agentLiveness(node: MbxNode, name: string, o: LivenessOptions = {}): AgentLiveness { + return agentLivenessBatch(node, [name], o).get(name) ?? { state: "offline", reason: "no session has held it under a lease", claimable: true }; +} + +/** + * The `agentmbx agents` display for this host's directory: local names carry the shared liveness token and reason. + * A shell participant (`agentmbx send --as`) has no session and no lease; it stays visible as live for two hours + * after its last send, exactly as before (T699 decision). + */ +export function agentDirectoryStates(node: MbxNode, o: LivenessOptions = {}): Map { + const now = o.now ?? Date.now(); + const records = node.agents(); + const local = records.filter((a) => a.host === node.host); + const answers = agentLivenessBatch(node, local.map((a) => a.name), o); + const since = new Date(now - 2 * 3_600_000).toISOString(); + const bound = new Set((node.store.db.prepare("SELECT DISTINCT agent FROM sessions").all() as { agent: string }[]).map((r) => r.agent)); + const out = new Map(); + for (const a of local) { + const l = answers.get(a.name); + // A shell participant (--as) never bound a session; it stays visible as live for two hours after its last send. + if (a.cli === "cli" && a.last_seen !== null && a.last_seen > since && !bound.has(a.name) && l?.state === "offline") { + const mins = Math.max(0, Math.round((now - Date.parse(a.last_seen)) / 60_000)); + out.set(a.name, { state: "live", reason: `shell sender, last seen ${mins} min ago` }); + } else if (l) out.set(a.name, { state: l.state, reason: l.reason }); + } + return out; +} diff --git a/src/node.ts b/src/node.ts index 6a9c65bd..39ef4322 100644 --- a/src/node.ts +++ b/src/node.ts @@ -69,6 +69,18 @@ export const LIVE_AGENT_MS = 24 * 3_600_000; /** Agents with no session binding at all (shell participants using --as) count as live this long after their last send. */ export const SHELL_AGENT_MS = 2 * 3_600_000; export const alive = (pid: number | null | undefined) => { if (!pid) return false; try { process.kill(pid, 0); return true; } catch (e) { return (e as NodeJS.ErrnoException).code === "EPERM"; } }; + +/** + * T699: a session row still belongs to the live process it was recorded for — the same display rule `sameSession` + * uses without `proof`. A row with a recorded birth must match it exactly; an older row counts only while fresh and + * alive. Liveness only: authorization still uses provenProcess (sameSession with `{proof: true}`). + */ +export function sessionRowLive(r: { pid: number | null; pid_start: string | null; updated_at: string }, now = Date.now()): boolean { + if (!r.pid || !Number.isSafeInteger(r.pid) || r.pid <= 0) return false; + if (now - Date.parse(r.updated_at) > LIVE_AGENT_MS) return false; + if (r.pid_start) return sameProcess(r.pid, r.pid_start); + return now - Date.parse(r.updated_at) < SESSION_FRESH_MS && alive(r.pid); +} const WAKEABLE = new Set(["codex", "opencode"]); /** Idle Grok has no first-party inject (T385). `grok --help` has no queue command. `~/.grok/leader.sock` * is the leader for config and MCP watches, not a prompt socket (user-guide 26-config-reference.md). @@ -326,8 +338,7 @@ export class MbxNode { sameSession(pid: number, r: { pid_start: string | null; updated_at: string }, o: { proof?: boolean } = {}): boolean { if (Date.now() - Date.parse(r.updated_at) > LIVE_AGENT_MS) return false; // live sessions rebind every minute if (o.proof) return provenProcess(pid, r.pid_start); - if (r.pid_start) return sameProcess(pid, r.pid_start); - return Date.now() - Date.parse(r.updated_at) < SESSION_FRESH_MS && alive(pid); + return sessionRowLive({ pid, pid_start: r.pid_start, updated_at: r.updated_at }); } /** The agent of the session this process runs inside (walks up the process tree), if any. */ @@ -445,20 +456,6 @@ export class MbxNode { return out; } - /** For display: agents with a live session, plus shell senders (--as, no session) active in the last 2 h. */ - liveAgents(): Set { - const out = new Set(), bound = new Set(); - for (const r of this.store.db.prepare("SELECT agent, pid, pid_start, updated_at FROM sessions").all() as { agent: string; pid: number | null; pid_start: string | null; updated_at: string }[]) { - bound.add(r.agent); - if (r.pid && this.sameSession(r.pid, r)) out.add(r.agent); - } - // shell participants (no session binding ever) count while they are active - const since = new Date(Date.now() - SHELL_AGENT_MS).toISOString(); - for (const r of this.store.db.prepare("SELECT name FROM agents WHERE host=? AND last_seen > ?").all(this.host, since) as { name: string }[]) - if (!bound.has(r.name)) out.add(r.name); - return out; - } - /** How mail reaches this agent when its session is idle. */ deliveryMode(agent: string, sessions = this.sessionsFor(agent)): string { const ss = sessions.filter((x) => x.pid && this.sameSession(x.pid, x)); diff --git a/src/opencode-poller.ts b/src/opencode-poller.ts index 03ff16d1..906cdf51 100644 --- a/src/opencode-poller.ts +++ b/src/opencode-poller.ts @@ -10,7 +10,7 @@ // the service itself still hosts the session; wakeOpencode posts to it (and keeps T646's dual-host refusal). Nothing here reads or changes // mail: a wake carries ids only, and the session claims its identity again on the turn the wake starts. import { provenProcess, withProcSnapshot } from "./proc.ts"; -import { opencodeHostOf, type OpencodeHost } from "./opencode-provider.ts"; +import { opencodeHostClassifier, opencodeHostOf, type OpencodeHost } from "./opencode-provider.ts"; import { opencodeWakeWaiting } from "./opencode-wake-queue.ts"; import type { MbxNode } from "./node.ts"; @@ -58,6 +58,60 @@ export function pollerTargetLive(node: MbxNode, session: WakeSession, host: Host return serveBindingLive(node, session, host) && opencodeWakeWaiting(session.session_id, session.pid!); } +/** + * T699: one bound session the dispatcher can still wake with no lease held. `kind` is the serve that hosts it; + * `wakeable` says a wake path is proven (a shared service needs no poller, T688; a standalone serve needs its + * plugin poll registered in THIS process, T659). + */ +export interface WakeBinding { + cli: string; + session_id: string; + pid: number; + kind: "service" | "standalone"; + wakeable: boolean; + via: "service" | "standalone-poller" | null; +} + +/** The session fields wake classification needs; a full session row is assignable to this. */ +export interface WakeRow { + agent: string; cli: string; session_id: string; pid: number | null; pid_start: string | null; +} + +/** + * T699: the wake bindings behind many agents from one process snapshot (and one argv read with the batched host + * classifier), for the liveness surfaces. This is the same evidence `pollerWakeTargets` uses, without its run + * closures: display never takes mail or leases. A standalone poller registration is per-process, so a caller + * outside the daemon sees the binding (`kind`, `wakeable: false`) but cannot claim the wake path. + */ +export function wakeBindingsByAgent( + sessionsByAgent: ReadonlyMap, + host: Hosts = opencodeHostClassifier(), + waiting: (sessionID: string, pid: number) => boolean = opencodeWakeWaiting, +): Map { + return withProcSnapshot(() => { + const out = new Map(); + for (const [agent, sessions] of sessionsByAgent) { + let best: WakeBinding | null = null; + for (const s of sessions) { + if (s.agent !== agent || s.cli !== "opencode" || !s.pid || s.pid <= 1 || s.session_id.startsWith("mcp-")) continue; + // The recorded birth must still be that process: a reused pid must not inherit a row it never wrote. + if (!provenProcess(s.pid, s.pid_start)) continue; + const kind = host(s.pid); + if (kind !== "service" && kind !== "standalone") continue; + const polling = kind === "standalone" && waiting(s.session_id, s.pid); + const candidate: WakeBinding = kind === "service" + ? { cli: s.cli, session_id: s.session_id, pid: s.pid, kind, wakeable: true, via: "service" } + : { cli: s.cli, session_id: s.session_id, pid: s.pid, kind, wakeable: polling, via: polling ? "standalone-poller" : null }; + // A service binding hosts every session of the service; prefer it over a standalone row of the same agent. + if (!best || candidate.kind === "service" || (candidate.wakeable && !best.wakeable)) best = candidate; + if (best.kind === "service") break; + } + if (best) out.set(agent, best); + } + return out; + }); +} + /** * Poller targets for `agent`: bound standalone OpenCode sessions whose MCP is gone but whose plugin still polls, and (T688) * bound sessions of the shared service, which still hosts them. diff --git a/src/receipts.ts b/src/receipts.ts index b729e8ba..7982204a 100644 --- a/src/receipts.ts +++ b/src/receipts.ts @@ -1,13 +1,13 @@ // Send-time recipient truth (T205): who a message actually reached, decided when it is sent, so a sender never mistakes -// "stored in a mailbox" for "an agent will see it". Liveness comes from the identity lease and its existing process -// evidence. Native push eligibility uses the dispatcher's exact lease/binding guard. -import { IdentityLeases, identityLeaseStatus, type IdentityLease } from "./identity-leases.ts"; +// "stored in a mailbox" for "an agent will see it". Liveness is one answer from liveness.ts (T699): the lease, the +// session binding and the serve kind together. Native push eligibility uses the dispatcher's exact lease/binding guard. +import { identityLeaseStatus, inspectLeaseProcess, type IdentityLease, type ProcessEvidence } from "./identity-leases.ts"; +import { agentLiveness } from "./liveness.ts"; import type { Envelope } from "./envelope.ts"; import { GROK_NO_PUSH, type MbxNode, type RouteTarget } from "./node.ts"; import { type MessageRow, type Store } from "./store.ts"; import { hasWakeAuthority, liveWatcher, queuedBehindTurnAt, watcherKey } from "./wake.ts"; import { captureWakeIdentity } from "./wake-identity.ts"; -import { procTable } from "./proc.ts"; import { activityKey, parseActivity, SHARED_IDLE_MS } from "./identity-availability.ts"; import { connectorKey, retryErrorClass } from "./diagnostics.ts"; import { remoteReceipts } from "./remote-receipts.ts"; @@ -15,54 +15,68 @@ import { isSlotName } from "./project-slot.ts"; import { projectLabel, projectMailboxBySlot, projectReader } from "./project-mailbox.ts"; import { peerProjectBySlot } from "./project-directory.ts"; -/** A holder renews its lease every minute; three missed beats mean its session is suspended or gone. */ -const STALE_HEARTBEAT_MS = 3 * 60_000; - export type RecipientState = "live-wake" | "live-next-prompt" | "offline" | "queued-no-holder" | "forwarded" | "remote"; export interface RecipientReceipt { to: string; address: string; state: RecipientState; detail: string } const iso = (ms: number) => new Date(ms).toISOString(); -/** Is a session holding `name` right now? Unknown process evidence with a fresh heartbeat counts as live (it is the - * lease's own rule for "unknown"); everything else says since when and who held it last. */ -export function mailboxLiveness(node: MbxNode, name: string, now = Date.now()): { live: boolean; detail: string } { +export interface MailboxLiveness { + /** True when a session will eventually see mail for this mailbox (live, idle, or an unverified holder). */ + live: boolean; + state: "live" | "idle" | "unknown" | "offline" | "conflict"; + /** The shared T699 answer, identical on every surface (roster, agent list, receipt). */ + reason: string; + /** Receipt prose: the holder it last knew, plus this surface's own context. */ + detail: string; + /** The idle came from a released or expired lease whose session is still running (T699). */ + released?: boolean; + /** A proven wake path exists for that binding (T688 shared service, or a registered T659 poller). */ + wakeable?: boolean; +} + +/** A holder renews its lease every minute; three missed beats mean its session is suspended or gone. Receipts are + * stricter than the 30-minute lease idle TTL: mail nobody is polling should read as undelivered sooner (T204). */ +const STALE_HEARTBEAT_MS = 3 * 60_000; + +/** Is a session holding `name` right now — or still there with its lease released? One answer from liveness.ts + * (T699), so a receipt cannot call a wakeable session "offline" while the agent list calls it "live". */ +export function mailboxLiveness(node: MbxNode, name: string, now = Date.now(), o: { inspect?: (pid: number) => ProcessEvidence } = {}): MailboxLiveness { if (isSlotName(name)) { // T493: a project mailbox has no holder of its own. Its reader is the project's current lead, so it is live exactly when that lead is. const reader = projectReader(node, name, new Date(now)); - if (!reader) return { live: false, detail: "this project mailbox is not allocated on this host" }; + if (!reader) return { live: false, state: "offline", reason: "this project mailbox is not allocated on this host", detail: "this project mailbox is not allocated on this host" }; const label = `project mailbox ${projectLabel(reader.mailbox)}`; - if (!reader.lead) return { live: false, detail: `${label}: no owner-designated lead on ${node.host} to read it` }; - const lead = mailboxLiveness(node, reader.lead.agent, now); - return { live: lead.live, detail: `${label}, read by its lead ${reader.lead.agent}: ${lead.detail}` }; + if (!reader.lead) { + const reason = `${label}: no owner-designated lead on ${node.host} to read it`; + return { live: false, state: "offline", reason, detail: reason }; + } + const lead = mailboxLiveness(node, reader.lead.agent, now, o); + return { ...lead, detail: `${label}, read by its lead ${reader.lead.agent}: ${lead.detail}` }; } + const l = agentLiveness(node, name, { now, ...(o.inspect ? { inspect: o.inspect } : {}) }); const row = node.store.db.prepare("SELECT * FROM identity_leases WHERE name=?").get(name) as unknown as IdentityLease | undefined; - if (!row) return { live: false, detail: "no session has ever held this mailbox; it is read when an agent claims it" }; - let state: "live" | "unknown" | "expired" = "unknown", reason: string | null = null; - try { ({ state, reason } = identityLeaseStatus(row, now, new IdentityLeases(node.store).processEvidence(row.holder_pid))); } - catch { - // Evidence can't be inspected inside a transaction: judge by the lease row, the process table snapshot taken before - // it (a vanished holder is gone) and the heartbeat, which a live holder renews every minute (T204 review). - const table = procTable(); - if (row.released_at !== null || now - row.heartbeat_at >= row.idle_ttl) { state = "expired"; reason = row.released_at !== null ? null : "idle"; } - else if (table.size && !table.has(row.holder_pid)) { state = "expired"; reason = "dead"; } - else if (now - row.heartbeat_at >= STALE_HEARTBEAT_MS) { state = "expired"; reason = `no heartbeat for ${Math.round((now - row.heartbeat_at) / 60_000)} min`; } - // Its pid is running and it renewed within three minutes: as live as an advisory receipt can tell. - else if (table.has(row.holder_pid)) state = "live"; - } - if ((state === "live" || state === "unknown") && now - row.heartbeat_at >= STALE_HEARTBEAT_MS) { - state = "expired"; reason = `no heartbeat for ${Math.round((now - row.heartbeat_at) / 60_000)} min`; - } + if (l.state === "conflict") return { live: false, state: "conflict", reason: l.reason, detail: l.reason }; + if (!row) return { live: false, state: "offline", reason: l.reason, detail: "no session has ever held this mailbox; it is read when an agent claims it" }; const holder = `${row.cli} session ${row.session_id.slice(0, 12)}`; - if (state === "live" || state === "unknown") { - const activity = parseActivity(node.store.get(activityKey(name))); - const quiet = activity?.shared && now - activity.at >= SHARED_IDLE_MS - ? `; its conversation has made no mbx call for ${Math.round((now - activity.at) / 60_000)} min and may have ended (no hook or wake-poll activity either)` : ""; - return { live: true, detail: state === "live" ? `held by ${holder}${quiet}` - : `held by ${holder} (process unconfirmed, heartbeat ${Math.round((now - row.heartbeat_at) / 1000)} s ago)${quiet}` }; + const silent = row.released_at === null && now - row.heartbeat_at >= STALE_HEARTBEAT_MS; + if (l.state === "offline" || silent) { + const status = row.released_at === null ? identityLeaseStatus(row, now, inspectLeaseProcess(row.holder_pid)) : null; + const why = row.release_reason ?? (silent ? `no heartbeat for ${Math.round((now - row.heartbeat_at) / 60_000)} min` : status?.reason ?? "released"); + return { live: false, state: "offline", reason: l.reason, + detail: `no live session since ${iso(row.released_at ?? row.heartbeat_at)} (last holder ${holder}, ${why})` }; } - const since = row.released_at ?? row.heartbeat_at; - const why = row.release_reason ?? reason ?? "released"; - return { live: false, detail: `no live session since ${iso(since)} (last holder ${holder}, ${why})` }; + const activity = parseActivity(node.store.get(activityKey(name))); + const quiet = activity?.shared ? now - activity.at : null; + const note = quiet !== null && quiet >= SHARED_IDLE_MS && (l.state === "live" || (l.state === "idle" && l.idleKind === "held")) + ? `; its conversation has made no mbx call for ${Math.round(quiet / 60_000)} min and may have ended (no hook or wake-poll activity either)` : ""; + const detail = l.idleKind === "released" ? l.reason + : l.state === "unknown" ? `held by ${holder} (process unconfirmed, heartbeat ${Math.round((now - row.heartbeat_at) / 1000)} s ago)${note}` + : `held by ${holder}${note}`; + return { + live: true, state: l.state, reason: l.reason, detail, + ...(l.idleKind === "released" ? { released: true } : {}), + ...(l.wakeable !== undefined ? { wakeable: l.wakeable } : {}), + }; } /** "addressed to the project mailbox X; " when `name` is a project slot that `host` advertises, else nothing (T493). */ @@ -105,6 +119,17 @@ export function recipientReceipts(node: MbxNode, msgId: string, targets: RouteTa detail = `addressed to the owner-designated lead; ${live.detail}; it waits here until a session claims ${name} or the owner forwards it, and the owner is notified`; } else if (!live.live) { state = "offline"; detail = live.detail; } + else if (live.released) { + // T699: the lease was released (or expired) but the session still runs. That is idle, not offline: mail waits + // for its next turn, and a proven wake path (a shared service, or a registered poller) can start that turn. + const mode = node.deliveryMode(name); + const why = !m ? "message not found" : !node.wantsWake(name, m) ? "this kind does not wake (status, or no needs_reply/mention)" + : !live.wakeable ? "the session is running without a wake path proven from this process" + : !hasWakeAuthority(node, name, m) ? "no owner policy lets this sender wake it" : null; + state = why ? "live-next-prompt" : "live-wake"; + detail = why ? `${live.detail}; seen on its next prompt: ${why}` + : `${live.detail}; eligible for wake (${mode}); the turn re-claims the mailbox; dispatcher admission pending`; + } else { const sessions = node.sessionsFor(name); const verified = sessions.filter(s => captureWakeIdentity(node, s)); @@ -195,10 +220,10 @@ export function deliveryReceipts(node: MbxNode, m: MessageRow, now = Date.now()) .all(m.id) as { r: string; d: string }[]).map((x) => [x.r, x.d])); const local = (node.store.db.prepare("SELECT agent,state,updated_at,note FROM deliveries WHERE msg_id=? ORDER BY agent").all(m.id) as { agent: string; state: string; updated_at: string; note: string | null }[]).map((d) => { - const live = d.agent === "owner" ? { live: true, detail: "the owner" } : mailboxLiveness(node, d.agent, now); + const live = d.agent === "owner" ? { live: true, state: "live" as const, detail: "the owner" } : mailboxLiveness(node, d.agent, now); const behind = d.state === "delivered" || d.state === "notified" ? queuedBehindTurnAt(node.store, d.agent, m.id) : null; return { address: `${d.agent}@${node.host}`, state: d.state, updated_at: d.updated_at, note: d.note, did: dids.get(d.agent) ?? null, - liveness: live.live ? `live: ${live.detail}` : `offline: ${live.detail}`, + liveness: `${live.state}: ${live.detail}`, ...(behind ? { wake: `${OPENCODE_BEHIND_TURN_NOTE} (admitted ${behind.slice(0, 19)}Z)` } : {}) }; }); // Remote recipients: the envelope's addressees on other hosts; a queued outbox row means not yet accepted there. diff --git a/src/roster.ts b/src/roster.ts index d4118693..35855c9f 100644 --- a/src/roster.ts +++ b/src/roster.ts @@ -137,8 +137,10 @@ export function rosterText(r: AgentsV1): string { const lines = r.agents.map((a) => { const led = a.lead_of.length ? ` LEAD of ${a.lead_of.map((p) => basename(p) || p).join(", ")}` : ""; const cli = a.harness ?? a.cli; + // T699: an idle row now also covers a released lease whose session still runs; its reason is the point. + const reason = a.state === "idle" || a.state === "unknown" ? ` reason: ${oneLine(a.reason)}` : ""; return `${a.address} ${a.state}${led}${a.self ? " (you)" : ""}${a.seen_here ? " seen here (not a member)" : ""}${a.role ? ` role:${a.role}` : ""}${a.task ? ` task:${a.task}` : ""}${a.lane ? ` lane:${a.lane}` : ""}${cli ? ` (${cli})` : ""}` - + ` last seen ${a.last_seen ?? "never"}${a.description ? ` — ${oneLine(a.description)}` : ""}`; + + ` last seen ${a.last_seen ?? "never"}${a.description ? ` — ${oneLine(a.description)}` : ""}${reason}`; }); const leadLine = !r.scope.project ? "lead: no project" : r.lead ? `lead: ${r.lead.address} until ${r.lead.exp}` : "lead: none"; const hidden = r.scope.hidden ? `\n${r.scope.hidden} retired or generated name(s) hidden; pass all:true to list them.` : ""; diff --git a/test/liveness-agreement.test.ts b/test/liveness-agreement.test.ts new file mode 100644 index 00000000..8deb9a26 --- /dev/null +++ b/test/liveness-agreement.test.ts @@ -0,0 +1,123 @@ +// T699: one liveness answer on every surface. A released lease whose session still runs is idle (claimable, +// wakeable, seen on its next turn) — never "live" and never "offline". The same rows must produce the same +// token and the same reason in `agentmbx agents`, `mbx_agents` (the roster) and the send receipt. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { MbxNode } from "../src/node.ts"; +import { procStart } from "../src/proc.ts"; +import { inspectLeaseProcess, type ProcessEvidence } from "../src/identity-leases.ts"; +import { opencodeHostSeams } from "../src/opencode-provider.ts"; +import { resetOpencodeWakeQueue, waitForOpencodeWake } from "../src/opencode-wake-queue.ts"; +import { agentDirectoryStates } from "../src/liveness.ts"; +import { buildRoster } from "../src/roster.ts"; +import { mailboxLiveness, recipientReceipts } from "../src/receipts.ts"; + +const MIN = 60_000; +const SERVICE_PID = process.pid; +const POLLER_PID = process.ppid; +const DEAD_PID = 2 ** 22 + 4242; +const UNKNOWN_PID = 2 ** 22 + 9999; + +interface Fixture { + name: string; + pid: number; + start: string; + session: string; + released: number | null; + row?: { pid: number; start: string }; +} + +const FIXTURES: Fixture[] = [ + // A verified holder: plainly live. The lease start is the lease layer's ps-utc form (holderStart), not procStart's lstart. + { name: "held-agent", pid: SERVICE_PID, start: inspectLeaseProcess(SERVICE_PID).start ?? "self", session: "ses_t699_held", released: null }, + // T688: the MCP was evicted, the lease released, but the shared service still hosts the conversation. + { name: "service-agent", pid: DEAD_PID, start: "gone", session: "ses_t699_service", released: Date.now() - 2 * 60 * MIN, + row: { pid: SERVICE_PID, start: procStart(SERVICE_PID) ?? "self" } }, + // T659: the same, in a standalone serve whose plugin poll is registered in this process. + { name: "poller-agent", pid: DEAD_PID, start: "gone", session: "ses_t699_poller", released: Date.now() - 2 * 60 * MIN, + row: { pid: POLLER_PID, start: procStart(POLLER_PID) ?? "parent" } }, + // Released with no session left at all: genuinely offline. + { name: "dead-agent", pid: DEAD_PID, start: "gone", session: "ses_t699_dead", released: Date.now() - 2 * 60 * MIN }, + // A holder whose process cannot be verified right now: unknown, never silently live or offline. + { name: "unknown-agent", pid: UNKNOWN_PID, start: "unreadable", session: "ses_t699_unknown", released: null }, +]; + +const EXPECTED: Record = { + "held-agent": "live", + "service-agent": "idle", + "poller-agent": "idle", + "dead-agent": "offline", + "unknown-agent": "unknown", +}; + +test("T699: agents, roster and receipt return the same liveness token and reason for held, released+service, released+poller, released+dead and unknown", async (t) => { + const home = mkdtempSync(join(tmpdir(), "mbx-t699-")); + const node = new MbxNode(home, { host: "alpha" }); + const realArgs = opencodeHostSeams.args; + t.after(() => { + resetOpencodeWakeQueue(); + opencodeHostSeams.args = realArgs; + node.close(); + rmSync(home, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }); + }); + // One argv read for both live serves: the service hosts the service fixture, the standalone one polls. + opencodeHostSeams.args = () => new Map([ + [SERVICE_PID, ["/opt/opencode/bin/opencode", "serve", "--service"]], + [POLLER_PID, ["/opt/opencode/bin/opencode", "serve", "--stdio", "--port", "0"]], + ]); + void waitForOpencodeWake("ses_t699_poller", POLLER_PID); // registers the plugin poll: T659's live evidence + + const now = Date.now(); + for (const f of FIXTURES) { + node.registerAgent(f.name, { cli: "opencode" }); + node.store.db.prepare(`INSERT INTO identity_leases (name,token,holder_pid,holder_start,key_fp,cli,session_id,claimed_at,heartbeat_at,idle_ttl,released_at,release_reason) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`) + .run(f.name, `tok-${f.name}`, f.pid, f.start, "aaaa-bbbb-cccc-dddd", "opencode", f.session, now, now, 30 * MIN, f.released, f.released === null ? null : "released"); + if (f.row) node.store.db.prepare(`INSERT INTO sessions (agent,cli,session_id,cwd,pid,session_key,channel,updated_at,pid_start) + VALUES (?,?,?,?,?,?,?,?,?)`).run(f.name, "opencode", f.session, home, f.row.pid, "k", 0, new Date(now).toISOString(), f.row.start); + } + const inspect = (pid: number): ProcessEvidence => pid === UNKNOWN_PID ? { alive: null, start: null } : inspectLeaseProcess(pid); + const opts = { now, inspect }; + + const agents = agentDirectoryStates(node, opts); + const roster = buildRoster(node, { ask: "*", all: true, now, inspect }); + for (const f of FIXTURES) { + const a = agents.get(f.name), r = roster.agents.find((x) => x.name === f.name), m = mailboxLiveness(node, f.name, now, { inspect }); + assert.ok(a, `${f.name}: the agent list has a row`); + assert.ok(r, `${f.name}: the roster has a row`); + assert.equal(a.state, EXPECTED[f.name], `${f.name}: agent list state`); + assert.equal(r.state, EXPECTED[f.name], `${f.name}: roster state`); + assert.equal(m.state, EXPECTED[f.name], `${f.name}: receipt state`); + assert.equal(a.reason, r.reason, `${f.name}: agent list and roster reasons agree`); + assert.equal(m.reason, r.reason, `${f.name}: receipt and roster reasons agree`); + } + + // The idle reasons name the evidence, and only the sessions that are actually there are wakeable. + const service = mailboxLiveness(node, "service-agent", now, { inspect }); + assert.match(service.reason, /released \d{4}-.*opencode session ses_t699_service is still running \(pid \d+, shared service\), so it is idle/); + assert.equal(service.wakeable, true); + assert.equal(service.live, true); + assert.equal(service.released, true); + const poller = mailboxLiveness(node, "poller-agent", now, { inspect }); + assert.match(poller.reason, /standalone serve/); + assert.equal(poller.wakeable, true); + assert.equal(mailboxLiveness(node, "dead-agent", now, { inspect }).live, false); + assert.equal(mailboxLiveness(node, "held-agent", now, { inspect }).live, true); + assert.equal(mailboxLiveness(node, "unknown-agent", now, { inspect }).state, "unknown"); + + // The receipt itself: an idle service target is not offline; mail waits for (or wakes) its next turn. + node.registerAgent("sender", { cli: "cli" }); + const sent = node.send({ from: "sender", to: ["service-agent"], kind: "request", subject: "T699", body: "b", needs_reply: true }); + const receipt = recipientReceipts(node, sent.envelope.id, sent.targets, now).find((x) => x.address === "service-agent@alpha"); + assert.ok(receipt, "the receipt addresses the idle service target"); + assert.notEqual(receipt.state, "offline"); + assert.equal(receipt.state, "live-next-prompt", "no owner policy lets this sender wake it, so it waits for the next prompt"); + assert.match(receipt.detail, /still running \(pid \d+, shared service\), so it is idle/); + + const dead = node.send({ from: "sender", to: ["dead-agent"], kind: "request", subject: "T699", body: "b", needs_reply: true }); + const deadReceipt = recipientReceipts(node, dead.envelope.id, dead.targets, now).find((x) => x.address === "dead-agent@alpha"); + assert.equal(deadReceipt?.state, "offline"); +}); From 9dddcd9ef70ef4c431b79ac15d725808854f28e4 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 15:16:35 -0700 Subject: [PATCH 2/5] fix(T700): OpenCode v2 places the resume note and never drops it on a receipt OpenCode 2.x drives the plugin default `setup(ctx)` path and marks a turn at `session.execution.started` before the first tool. The busy-turn guard then treated every tool as mid-turn, so a revisit notice could only fall back; at the turn's end the plugin admitted it with `resume:false`, whose msg_ receipt can arrive without a session_message (T544), and discarded the pending copy. - Track tools-per-turn so the first tool of a turn places the note even when the v2 start event already marked the turn. - Drop a kept note only after it actually lands in a tool result. A stop no longer admits it through a receipt; `resume:true` is not used because starting a turn on an idle TUI is not the job of a resume note. - Keep a fresh conversation's start note on its existing queued admission, which the host materialises as the conversation begins. - Add v2 harness tests for both failures; update the two T187 hook tests that encoded the old receipt-retry contract. Checks: the two v2 tests fail on the pre-fix template and pass after it; opencode-resume, opencode-hooks, opencode-no-duplicate-loop, opencode-poller-wake, the full opencode-* suite, tsc --noEmit, npm run build. --- CHANGELOG.md | 2 + dist/setup.js | 38 ++++++++---- src/setup.ts | 38 ++++++++---- test/opencode-hooks.test.ts | 29 +++++---- test/opencode-resume.test.ts | 117 +++++++++++++++++++++++++++++++++-- 5 files changed, 182 insertions(+), 42 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c79b4ebb..3b8ac6cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- **An OpenCode v2 resume note reaches the model instead of being silently dropped (T700).** OpenCode 2.x drives the plugin's default `setup(ctx)` path. It marks a turn at `session.execution.started`, before the first tool, so the busy-turn guard treated every tool as mid-turn and the revisit notice could only fall back; at the turn's end the plugin admitted it with `resume:false`, whose `msg_` receipt can arrive without any `session_message` (T544), and then discarded the pending copy. The first tool of a turn now places the note even after the v2 start event, and a kept note is only dropped after it actually lands in a tool result: a stop no longer acknowledges a receipt as delivery, and `resume:true` is not used because starting a turn on an idle TUI is not a resume note's job. A fresh conversation's start note keeps its queued admission, which the host materialises as the conversation begins. + ## 0.5.48 (2026-10-17) - **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. diff --git a/dist/setup.js b/dist/setup.js index 664417a1..20715f77 100644 --- a/dist/setup.js +++ b/dist/setup.js @@ -1504,9 +1504,12 @@ const watchWake = (ctx: NoteCtx, sid: string): void => { /** T642: a session is mid-turn from its first tool call (or execution start) until its turn ends (idle / execution succeeded). * Delivery "queue" holds a note or a wake until that end, so the post-tool notice rides the tool result instead. */ const turnSince = new Map(); +/** T700: tools this turn has already run. The first tool of a turn may place a resume note even on the v2 path, + * where session.execution.started marks the turn before any tool runs; later tools in that same turn are mid-turn. */ +const turnTools = new Map(); const TURN_STALE_MS = 30 * 60_000; -const markTurn = (sid: string): void => { if (!turnSince.has(sid)) turnSince.set(sid, Date.now()); }; -const endTurn = (sid: string): void => { turnSince.delete(sid); }; +const markTurn = (sid: string): void => { if (!turnSince.has(sid)) { turnSince.set(sid, Date.now()); turnTools.set(sid, 0); } }; +const endTurn = (sid: string): void => { turnSince.delete(sid); turnTools.delete(sid); }; /** T675: the turn ended. Tell the daemon (receipt id "idle"), so its next normal pass may wake this session for mail still * unhandled instead of waiting out the busy backoff. It admits nothing. True only when the daemon took it, which means the * wake path owns the post-turn alert for this session; false keeps the stop hook as the fallback. */ @@ -1582,16 +1585,17 @@ const postToolSteady = (sid: string): boolean => { /** One hook call: the JSON payload goes in on stdin; stdout is captured; a reason we recognize is * 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, +const call = async (ctx: NoteCtx, bin: string[], event: string, payload: Record, pending: Map, 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. + // Captured before this hook marks or ends the turn. T700: the v2 event loop marks the turn at + // session.execution.started, before its first tool, so "turn open" is not the same as "a tool already ran". const since = sid ? turnSince.get(sid) : undefined; const turnOpen = since !== undefined && Date.now() - since < TURN_STALE_MS; + const firstTool = event === "post-tool" && (turnTools.get(sid ?? "") ?? 0) === 0; if (sid) watchWake(ctx, sid); - if (sid && event === "post-tool") markTurn(sid); + if (sid && event === "post-tool") { markTurn(sid); turnTools.set(sid, (turnTools.get(sid) ?? 0) + 1); } 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. @@ -1603,17 +1607,25 @@ const call = async (ctx: NoteCtx, bin: string[], event: string, payload: Record< 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"); }; + let placed: string | null = null; const deliver = async () => { - if (!sid || sending.has(sid)) return; + if (!sid) return; const text = pending.get(sid); if (!text) return; + // T700: the first tool of a turn is where a resume notice belongs, even when the v2 path marked the turn + // at execution.started. A successful placement is the only proof that the note reached the model, so only + // then is the pending copy dropped. + if (text === placed) { pending.delete(sid); return; } + if (event === "post-tool" && firstTool && place && place(text)) { placed = text; if (pending.get(sid) === text) pending.delete(sid); 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; } + // T700: a note kept across a turn is never acknowledged-and-dropped on a receipt. resume:false can return a + // msg_ id without writing a session_message (T544), and resume:true would start a turn on an idle TUI, so the + // next turn's first tool places it. A fresh conversation's start note keeps its queued admission, which the + // host materialises as the conversation begins; only that event may admit without a tool result. + if (event !== "session-start") 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); } + if (await inject(ctx, sid, "resume-entry", text) && pending.get(sid) === text) pending.delete(sid); }; await deliver(); // T675: with this session's wake poll live, the daemon's wake is the one post-turn alert: it starts a turn and offers only @@ -1638,8 +1650,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(), resumeTold = new Set(); - return (event: string, payload: Record, place?: (text: string) => boolean) => call(ctx, bin, event, { cwd, ...payload }, pending, sending, resumeTold, place); + const pending = new Map(), resumeTold = new Set(); + return (event: string, payload: Record, place?: (text: string) => boolean) => call(ctx, bin, event, { cwd, ...payload }, pending, resumeTold, place); }; const sidOf = (v: unknown): string | undefined => { diff --git a/src/setup.ts b/src/setup.ts index c62efcdc..c45be2c9 100644 --- a/src/setup.ts +++ b/src/setup.ts @@ -1337,9 +1337,12 @@ const watchWake = (ctx: NoteCtx, sid: string): void => { /** T642: a session is mid-turn from its first tool call (or execution start) until its turn ends (idle / execution succeeded). * Delivery "queue" holds a note or a wake until that end, so the post-tool notice rides the tool result instead. */ const turnSince = new Map(); +/** T700: tools this turn has already run. The first tool of a turn may place a resume note even on the v2 path, + * where session.execution.started marks the turn before any tool runs; later tools in that same turn are mid-turn. */ +const turnTools = new Map(); const TURN_STALE_MS = 30 * 60_000; -const markTurn = (sid: string): void => { if (!turnSince.has(sid)) turnSince.set(sid, Date.now()); }; -const endTurn = (sid: string): void => { turnSince.delete(sid); }; +const markTurn = (sid: string): void => { if (!turnSince.has(sid)) { turnSince.set(sid, Date.now()); turnTools.set(sid, 0); } }; +const endTurn = (sid: string): void => { turnSince.delete(sid); turnTools.delete(sid); }; /** T675: the turn ended. Tell the daemon (receipt id "idle"), so its next normal pass may wake this session for mail still * unhandled instead of waiting out the busy backoff. It admits nothing. True only when the daemon took it, which means the * wake path owns the post-turn alert for this session; false keeps the stop hook as the fallback. */ @@ -1415,16 +1418,17 @@ const postToolSteady = (sid: string): boolean => { /** One hook call: the JSON payload goes in on stdin; stdout is captured; a reason we recognize is * 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, +const call = async (ctx: NoteCtx, bin: string[], event: string, payload: Record, pending: Map, 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. + // Captured before this hook marks or ends the turn. T700: the v2 event loop marks the turn at + // session.execution.started, before its first tool, so "turn open" is not the same as "a tool already ran". const since = sid ? turnSince.get(sid) : undefined; const turnOpen = since !== undefined && Date.now() - since < TURN_STALE_MS; + const firstTool = event === "post-tool" && (turnTools.get(sid ?? "") ?? 0) === 0; if (sid) watchWake(ctx, sid); - if (sid && event === "post-tool") markTurn(sid); + if (sid && event === "post-tool") { markTurn(sid); turnTools.set(sid, (turnTools.get(sid) ?? 0) + 1); } 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. @@ -1436,17 +1440,25 @@ const call = async (ctx: NoteCtx, bin: string[], event: string, payload: Record< 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"); }; + let placed: string | null = null; const deliver = async () => { - if (!sid || sending.has(sid)) return; + if (!sid) return; const text = pending.get(sid); if (!text) return; + // T700: the first tool of a turn is where a resume notice belongs, even when the v2 path marked the turn + // at execution.started. A successful placement is the only proof that the note reached the model, so only + // then is the pending copy dropped. + if (text === placed) { pending.delete(sid); return; } + if (event === "post-tool" && firstTool && place && place(text)) { placed = text; if (pending.get(sid) === text) pending.delete(sid); 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; } + // T700: a note kept across a turn is never acknowledged-and-dropped on a receipt. resume:false can return a + // msg_ id without writing a session_message (T544), and resume:true would start a turn on an idle TUI, so the + // next turn's first tool places it. A fresh conversation's start note keeps its queued admission, which the + // host materialises as the conversation begins; only that event may admit without a tool result. + if (event !== "session-start") 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); } + if (await inject(ctx, sid, "resume-entry", text) && pending.get(sid) === text) pending.delete(sid); }; await deliver(); // T675: with this session's wake poll live, the daemon's wake is the one post-turn alert: it starts a turn and offers only @@ -1471,8 +1483,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(), resumeTold = new Set(); - return (event: string, payload: Record, place?: (text: string) => boolean) => call(ctx, bin, event, { cwd, ...payload }, pending, sending, resumeTold, place); + const pending = new Map(), resumeTold = new Set(); + return (event: string, payload: Record, place?: (text: string) => boolean) => call(ctx, bin, event, { cwd, ...payload }, pending, resumeTold, place); }; const sidOf = (v: unknown): string | undefined => { diff --git a/test/opencode-hooks.test.ts b/test/opencode-hooks.test.ts index 8ef3d866..93450867 100644 --- a/test/opencode-hooks.test.ts +++ b/test/opencode-hooks.test.ts @@ -244,16 +244,19 @@ test("T187: failed admission retries on the next hook without another node spawn try { const hooks = await mod.AgentMBXHooks({ directory: "/work", session: { synthetic: async (body: Note) => { notes.push(body); if (!ready) throw new Error("not admitted"); return body; - } } }) as { event: (event: unknown) => Promise; "tool.execute.after": (event: unknown) => Promise }; + } } }) as { event: (event: unknown) => Promise; "tool.execute.after": (event: unknown, output?: { output?: string }) => Promise }; await hooks.event({ event: { type: "session.created", properties: { info: { id: "ses_retry" } } } }); assert.equal(notes.length, 1); ready = true; await hooks["tool.execute.after"]({ sessionID: "ses_sibling" }); assert.equal(notes.length, 1, "sibling never receives or drains another entry"); - await hooks["tool.execute.after"]({ sessionID: "ses_retry" }); - assert.equal(notes.length, 2); assert.equal(notes[0].id, notes[1].id); - assert.ok(notes.every(note => note.sessionID === "ses_retry" && note.resume === false)); - await hooks["tool.execute.after"]({ sessionID: "ses_retry" }); - assert.equal(notes.length, 2, "admitted entry is removed from pending"); + // T700: a failed admission is not retried through the session API. The kept note rides the next turn's first + // tool result, which is the only admission the model can be proven to have seen. + const first = { output: "tool ok" }; + await hooks["tool.execute.after"]({ sessionID: "ses_retry" }, first); + assert.match(first.output, /worker@alpha: mbx_catchup/); + assert.equal(notes.length, 1, "the tool result is the admission; no second synthetic attempt"); + await hooks["tool.execute.after"]({ sessionID: "ses_retry" }, { output: "next" }); + assert.equal(notes.length, 1, "a placed entry is removed from pending"); assert.equal(calls.filter(c => c.args[1] === "session-start").length, 1); assert.equal(calls.filter(c => c.args[1] === "post-tool" && c.input.session_id === "ses_retry").length, 0, "steady retries don't spawn node"); } finally { @@ -286,12 +289,16 @@ test("T187: a resolved native call with no receipt remains pending instead of cl try { const hooks = await mod.AgentMBXHooks({ directory: "/work", session: { synthetic: async (body: Note) => { notes.push(body); return ++attempts === 1 ? {} : { data: { id: body.id } }; - } } }) as { event: (event: unknown) => Promise; "tool.execute.after": (event: unknown) => Promise }; + } } }) as { event: (event: unknown) => Promise; "tool.execute.after": (event: unknown, output?: { output?: string }) => Promise }; await hooks.event({ event: { type: "session.created", properties: { info: { id: "ses_no_receipt" } } } }); - await hooks["tool.execute.after"]({ sessionID: "ses_no_receipt" }); - assert.equal(notes.length, 2); assert.equal(notes[0].id, notes[1].id); - await hooks["tool.execute.after"]({ sessionID: "ses_no_receipt" }); - assert.equal(notes.length, 2, "recognized data.id closes only the pending admission, not work completion"); + // T700: the no-receipt call did not prove admission, so the note stays pending; the next turn's first tool + // result carries it, and no further synthetic attempt is made. + const first = { output: "tool ok" }; + await hooks["tool.execute.after"]({ sessionID: "ses_no_receipt" }, first); + assert.match(first.output, /worker@alpha: mbx_catchup/); + assert.equal(notes.length, 1, "the session API was not asked again"); + await hooks["tool.execute.after"]({ sessionID: "ses_no_receipt" }, { output: "next" }); + assert.equal(notes.length, 1, "a placed entry is removed from pending"); } finally { restore(); rmSync(dir, { recursive: true, force: true }); } }); diff --git a/test/opencode-resume.test.ts b/test/opencode-resume.test.ts index 95cbb2b4..f7928c96 100644 --- a/test/opencode-resume.test.ts +++ b/test/opencode-resume.test.ts @@ -100,7 +100,7 @@ test("T187: a quiet resume places the same notice in that tool result and does n } 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 () => { +test("T187: a busy turn keeps the resume note, reports the fallback once, and places it on the next turn's first tool", async () => { const { mod, dir } = await loadPlugin(), notes: Note[] = []; let tools = 0; const restore = stubSpawn([], c => { @@ -119,10 +119,13 @@ test("T187: a busy turn keeps the resume note, reports the fallback once, and ad 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"); + assert.equal(notes.length, 0, "T700: ending the turn does not acknowledge a note the host never materialised"); + const next = { output: "next turn" }; + await hooks["tool.execute.after"]({ sessionID: "ses_busy" }, next); + assert.match(next.output, /You are worker@alpha/, "the next turn's first tool result carries the kept note"); + assert.doesNotMatch(next.output, /Resume note was not admitted/); + assert.equal(next.output.split(NOTICE).length - 1, 1, "exactly one copy"); + assert.equal(notes.length, 0); } finally { restore(); rmSync(dir, { recursive: true, force: true }); } }); @@ -189,3 +192,107 @@ test("T187: a host with no session note API keeps the note and a later tool resu restore(); rmSync(dir, { recursive: true, force: true }); } }); + +// ---- T700: the v2 plugin path. OpenCode 2.x validates the default export and dispatches through +// setup(ctx): ctx.event.subscribe() marks a turn at session.execution.started, and ctx.tool.hook is +// the only place a note can ride a tool result. The v1 hooks object is never consumed there. + +type V2Tool = (input: unknown) => unknown; +type V2Module = { default: { setup: (ctx: unknown) => Promise<() => void> } }; + +async function loadPluginV2() { + (globalThis as { __mbxWakePort?: string | null }).__mbxWakePort = null; + const dir = mkdtempSync(join(tmpdir(), "mbx-oc-v2-")); + const file = join(dir, "agentmbx.ts"); + writeFileSync(file, opencodePluginSource(CMD, version())); + return { mod: await import(file) as V2Module, dir }; +} + +function eventStream() { + const queue: unknown[] = []; + let wake: (() => void) | null = null, open = true; + const nudge = () => { const w = wake; wake = null; w?.(); }; + return { + iterable: { + [Symbol.asyncIterator]: () => ({ + next: async () => { + while (!queue.length && open) await new Promise((resolve) => { wake = resolve; }); + return queue.length ? { done: false, value: queue.shift() } : { done: true, value: undefined }; + }, + }), + }, + push(event: unknown) { queue.push(event); nudge(); }, + close() { open = false; nudge(); }, + }; +} + +const tick = () => new Promise((resolve) => setTimeout(resolve, 30)); +const at = (sid: string) => ({ location: { directory: "/work" }, data: { sessionID: sid } }); + +test("T700 v2: execution.started does not block the first tool of the turn from placing the resume note", async () => { + const { mod, dir } = await loadPluginV2(); + const notes: Note[] = []; + const home = process.env.MBX_HOME; + process.env.MBX_HOME = dir; + const restore = stubSpawn([], (c) => c.args[1] === "post-tool" ? envelope("ses_v2a", NOTICE) : ""); + const stream = eventStream(); + let tool: V2Tool | null = null; + const dispose = await mod.default.setup({ + location: { directory: "/work" }, event: { subscribe: () => stream.iterable }, + tool: { hook: (_name: string, fn: V2Tool) => { tool = fn; } }, session: stubSession(notes), + }); + try { + stream.push({ type: "session.execution.started", ...at("ses_v2a") }); + await tick(); + const result = { sessionID: "ses_v2a", status: "completed", result: { content: "tool output" } }; + await tool!(result); + assert.match(result.result.content, /You are worker@alpha/, "the first tool result carries the bounded notice"); + assert.doesNotMatch(result.result.content, /Resume note was not admitted/); + assert.equal(notes.length, 0, "placing it in the tool result starts no turn and queues no copy"); + } finally { + dispose(); stream.close(); + if (home === undefined) delete process.env.MBX_HOME; else process.env.MBX_HOME = home; + restore(); rmSync(dir, { recursive: true, force: true }); + } +}); + +test("T700 v2: a stop never acknowledges-and-drops a note; the next turn's first tool result places it once", async () => { + const { mod, dir } = await loadPluginV2(); + const notes: Note[] = []; + const home = process.env.MBX_HOME; + process.env.MBX_HOME = dir; + const restore = stubSpawn([], (c) => c.args[1] === "post-tool" ? envelope("ses_v2b", NOTICE) : ""); + const stream = eventStream(); + let tool: V2Tool | null = null; + const dispose = await mod.default.setup({ + location: { directory: "/work" }, event: { subscribe: () => stream.iterable }, + tool: { hook: (_name: string, fn: V2Tool) => { tool = fn; } }, session: stubSession(notes), + }); + try { + stream.push({ type: "session.execution.started", ...at("ses_v2b") }); + await tick(); + // An error result cannot carry a note, so the note stays pending and only the fallback is attempted. + const first = { sessionID: "ses_v2b", status: "error", result: { content: "boom" } }; + await tool!(first); + assert.doesNotMatch(first.result.content, /You are worker@alpha/); + + // The host may return a msg_ receipt without writing a session_message (T544). A stop must not treat that + // as delivered: the note stays pending for the next turn instead of being acknowledged and dropped. + stream.push({ type: "session.execution.succeeded", ...at("ses_v2b") }); + await tick(); + assert.equal(notes.length, 0, "no synthetic admission, and no unwanted turn on an idle session"); + + stream.push({ type: "session.execution.started", ...at("ses_v2b") }); + await tick(); + const second = { sessionID: "ses_v2b", status: "completed", result: { content: "next turn" } }; + await tool!(second); + assert.match(second.result.content, /You are worker@alpha/, "the kept note rides the next turn's first tool result"); + assert.doesNotMatch(second.result.content, /Resume note was not admitted/); + assert.equal(second.result.content.split(NOTICE).length - 1, 1, "exactly one copy"); + assert.equal(notes.length, 0); + } finally { + dispose(); stream.close(); + if (home === undefined) delete process.env.MBX_HOME; else process.env.MBX_HOME = home; + restore(); rmSync(dir, { recursive: true, force: true }); + } +}); From 991115649419a65a25460539481c1a383902192f Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 15:23:45 -0700 Subject: [PATCH 3/5] test(T699): make the roster dead-session fixture genuinely dead sessionRowLive reads the real process table, so the fixture's fabricated low pid could be alive (EPERM) on CI. An expired holder whose session row still names a live process is legitimately idle under T699, so the row kept its task and lane and the T497 assertion failed with "idle" instead of "offline". Point the session binding at a reaped pid so the fixture is offline on every host. Test-only; no product change. Checks: roster.test.ts (9/9 with liveness-agreement), identity-status, agents-directory, tsc --noEmit. --- test/roster.test.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/test/roster.test.ts b/test/roster.test.ts index 17dae4db..30bbf571 100644 --- a/test/roster.test.ts +++ b/test/roster.test.ts @@ -2,6 +2,7 @@ // council set (no counts on a row, a role label is not the lead, a remote row is never presented as verified). import { test } from "node:test"; import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; import { mkdtempSync, readFileSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -20,6 +21,9 @@ import { AGENTS_V1_SCHEMA, validateAgentsV1, type AgentsV1 } from "../src/status const PASS = "test-only-passphrase"; const FIXTURE = JSON.parse(readFileSync(join(import.meta.dirname, "fixtures", "agents-v1.json"), "utf8")) as AgentsV1; const A = "/work/proj-a", B = "/work/proj-b", C = "/work/proj-c"; +/** A pid the OS has already reaped. T699 reads the real process table for a session row, so a low fabricated pid + * can be EPERM-alive on CI; a reaped pid is deterministically dead. */ +const exitedPid = (): number => spawnSync(process.execPath, ["-e", ""]).pid!; function world(t: { after: (fn: () => unknown) => void }) { const home = mkdtempSync(join(tmpdir(), "mbx-roster-")); @@ -257,6 +261,9 @@ test("T497: a live local row shows the session task and lane; a remote row leave w.hold("plain", { project: A, cwd: A }); const gone = w.hold("was-on-task", { project: A, cwd: A }); w.evidence.set(gone, { alive: false, start: null }); + // T699: an expired holder whose session row still names a live process is idle (the T688 eviction case), so this + // fixture's session binding must be dead too for the row to be genuinely offline. + w.node.store.db.prepare("UPDATE sessions SET pid=? WHERE agent='was-on-task'").run(exitedPid()); setSessionPresence(w.node, { cli: "claude", sessionId: "s-on-task", task: "T497", lane: "implementation" }); setSessionPresence(w.node, { cli: "claude", sessionId: "s-was-on-task", task: "T497", lane: "implementation" }); w.node.store.db.prepare("INSERT INTO agents (name,host,role) VALUES ('far-dev','beta','builder')").run(); From 453f7b858bb22eaa3d5c021c1e53bda81c575852 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 16:15:34 -0700 Subject: [PATCH 4/5] chore(T705): 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 1a4d32cb..fa6e3ef1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## Unreleased - **One liveness answer: a released lease whose session still runs is idle, not live and not offline (T699).** `agentmbx agents` read the session binding while `mbx_agents` and the send receipt read the identity lease, so an OpenCode MCP eviction released the lease and the surfaces disagreed for hours: the CLI said `live`, the receipt said `offline (released)`. A single `agentLiveness` answer (src/liveness.ts) now composes the lease, the live session row and the serve kind, and every surface uses it. A released or expired lease whose session still runs reads as `idle` with its reason (claimable, and wakeable when the T688 shared service or a registered T659 poller can start its next turn, which re-claims the mailbox). `agentmbx agents` prints that state plus a trailing reason column and keeps shell senders visible as `live`; `mbx_agents` and `mbx_identity list` follow; a send reports `live-next-prompt`/`live-wake` instead of `offline`. The 30-minute lease idle TTL and the receipt's 3-minute heartbeat rule are unchanged. + - **An OpenCode v2 resume note reaches the model instead of being silently dropped (T700).** OpenCode 2.x drives the plugin's default `setup(ctx)` path. It marks a turn at `session.execution.started`, before the first tool, so the busy-turn guard treated every tool as mid-turn and the revisit notice could only fall back; at the turn's end the plugin admitted it with `resume:false`, whose `msg_` receipt can arrive without any `session_message` (T544), and then discarded the pending copy. The first tool of a turn now places the note even after the v2 start event, and a kept note is only dropped after it actually lands in a tool result: a stop no longer acknowledges a receipt as delivery, and `resume:true` is not used because starting a turn on an idle TUI is not a resume note's job. A fresh conversation's start note keeps its queued admission, which the host materialises as the conversation begins. ## 0.5.48 (2026-10-17) From 323104aac83321aceba7123b6eceb327ea20c310 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 16:47:56 -0700 Subject: [PATCH 5/5] test(T699): make reaped pids the default in persona fixtures The roster helper assigned 1000+ fabricated pids; on CI those can be live system processes (alive() counts EPERM), so an expired holder whose session row still named one read idle and the seen_here fixture leaked into the list. hold() now assigns an OS-reaped pid, making every persona fixture deterministic; the T497 fixture no longer needs its manual session-row update. The liveness agreement dead/unknown fixtures and the Grok dead-harness check use reaped pids too. Audited: identity-status (injected evidence, no session rows), agents-directory (real process.pid), project-membership (stale rows, no leases), opencode-poll-liveness (released lease only for poll activity). Test-only; no product change. Checks: roster, identity-status, agents-directory, liveness-agreement and liveness-activity (19 pass), tsc --noEmit. --- test/liveness-activity.test.ts | 5 ++++- test/liveness-agreement.test.ts | 7 +++++-- test/roster.test.ts | 8 +++----- 3 files changed, 12 insertions(+), 8 deletions(-) diff --git a/test/liveness-activity.test.ts b/test/liveness-activity.test.ts index 1904c546..ce54c0c3 100644 --- a/test/liveness-activity.test.ts +++ b/test/liveness-activity.test.ts @@ -1,5 +1,6 @@ import { test } from "node:test"; import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; import { mkdirSync, mkdtempSync, rmSync, utimesSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -7,6 +8,8 @@ import { identityAvailability, SHARED_IDLE_MS } from "../src/identity-availabili import type { IdentityLease } from "../src/identity-leases.ts"; const THREAD = "44444444-4444-4444-8444-444444444444"; +/** A pid the OS has already reaped: the Grok dead-harness check must not depend on a fabricated pid being free. */ +const exitedPid = (): number => spawnSync(process.execPath, ["-e", ""]).pid!; const live = { alive: true as const, start: "birth" }; function lease(cli: string, sessionId: string, now: number): IdentityLease { @@ -58,7 +61,7 @@ test("a live Grok session with a stale mbx call reads as active", () => { 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() }])); + writeFileSync(file, JSON.stringify([{ session_id: "grok-t583", pid: exitedPid(), 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`); diff --git a/test/liveness-agreement.test.ts b/test/liveness-agreement.test.ts index 8deb9a26..a0c6fd57 100644 --- a/test/liveness-agreement.test.ts +++ b/test/liveness-agreement.test.ts @@ -1,6 +1,7 @@ // T699: one liveness answer on every surface. A released lease whose session still runs is idle (claimable, // wakeable, seen on its next turn) — never "live" and never "offline". The same rows must produce the same // token and the same reason in `agentmbx agents`, `mbx_agents` (the roster) and the send receipt. +import { spawnSync } from "node:child_process"; import { test } from "node:test"; import assert from "node:assert/strict"; import { mkdtempSync, rmSync } from "node:fs"; @@ -18,8 +19,10 @@ import { mailboxLiveness, recipientReceipts } from "../src/receipts.ts"; const MIN = 60_000; const SERVICE_PID = process.pid; const POLLER_PID = process.ppid; -const DEAD_PID = 2 ** 22 + 4242; -const UNKNOWN_PID = 2 ** 22 + 9999; +/** Reaped pids: deterministic on every host (a fabricated low pid can be alive on CI). */ +const exitedPid = (): number => spawnSync(process.execPath, ["-e", ""]).pid!; +const DEAD_PID = exitedPid(); +const UNKNOWN_PID = exitedPid(); interface Fixture { name: string; diff --git a/test/roster.test.ts b/test/roster.test.ts index 30bbf571..5f9a47e4 100644 --- a/test/roster.test.ts +++ b/test/roster.test.ts @@ -40,10 +40,9 @@ function world(t: { after: (fn: () => unknown) => void }) { return e ?? { alive: true, start: `birth-${pid}` }; }; const leases = new IdentityLeases(node.store, { clock: () => NOW, idleTtlMs: 3_600_000, inspect }); - let nextPid = 1000; /** A persona with a lease held by a process: registered (unless told otherwise), bound to a project, in the directory. */ const hold = (name: string, o: { cli?: string; role?: string; project?: string; registered?: boolean; pid?: number; description?: string; cwd?: string } = {}) => { - const pid = o.pid ?? nextPid++; + const pid = o.pid ?? exitedPid(); leases.claim(name, { pid, start: `birth-${pid}`, cli: o.cli ?? "claude", sessionId: `s-${name}`, keyFp: "aaaa-bbbb-cccc-dddd" }); if (o.registered !== false) registerIdentity(node.store, { name, role: o.role ?? "builder", description: o.description ?? null }); node.registerAgent(name, { cli: o.cli ?? "claude", ...(o.role ? { role: o.role } : {}), ...(o.description ? { description: o.description } : {}) }); @@ -261,9 +260,8 @@ test("T497: a live local row shows the session task and lane; a remote row leave w.hold("plain", { project: A, cwd: A }); const gone = w.hold("was-on-task", { project: A, cwd: A }); w.evidence.set(gone, { alive: false, start: null }); - // T699: an expired holder whose session row still names a live process is idle (the T688 eviction case), so this - // fixture's session binding must be dead too for the row to be genuinely offline. - w.node.store.db.prepare("UPDATE sessions SET pid=? WHERE agent='was-on-task'").run(exitedPid()); + // T699: the helper's reaped pid makes the session binding dead too, so this row is genuinely offline (an expired + // holder whose session row still named a live process would be idle, the T688 eviction case). setSessionPresence(w.node, { cli: "claude", sessionId: "s-on-task", task: "T497", lane: "implementation" }); setSessionPresence(w.node, { cli: "claude", sessionId: "s-was-on-task", task: "T497", lane: "implementation" }); w.node.store.db.prepare("INSERT INTO agents (name,host,role) VALUES ('far-dev','beta','builder')").run();