From b4ece3c6791f3b13e4ee886f7cac789f357aa663 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 14:46:57 -0700 Subject: [PATCH 1/2] 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 c79b4eb..f34951c 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 232662b..3361fa3 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 ccaf014..2e3355c 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 0000000..3b5de48 --- /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 36168fb..67a6b08 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 8d808e1..f8507c2 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 b0c2ae0..5e0c3f0 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 8ae9712..e97be13 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 7440cbc..056a633 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 4490e26..1b96f7c 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 0000000..12f1944 --- /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 6a9c65b..39ef432 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 03ff16d..906cdf5 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 b729e8b..7982204 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 d411869..35855c9 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 0000000..8deb9a2 --- /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 991115649419a65a25460539481c1a383902192f Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 15:23:45 -0700 Subject: [PATCH 2/2] 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 17dae4d..30bbf57 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();