diff --git a/CHANGELOG.md b/CHANGELOG.md index c79b4eb..fa6e3ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## 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) - **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/dist/setup.js b/dist/setup.js index 664417a..20715f7 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/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/src/setup.ts b/src/setup.ts index c62efcd..c45be2c 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/liveness-activity.test.ts b/test/liveness-activity.test.ts index 1904c54..ce54c0c 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 new file mode 100644 index 0000000..a0c6fd5 --- /dev/null +++ b/test/liveness-agreement.test.ts @@ -0,0 +1,126 @@ +// 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"; +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; +/** 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; + 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"); +}); diff --git a/test/opencode-hooks.test.ts b/test/opencode-hooks.test.ts index 8ef3d86..9345086 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 95cbb2b..f7928c9 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 }); + } +}); diff --git a/test/roster.test.ts b/test/roster.test.ts index 17dae4d..5f9a47e 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-")); @@ -36,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 } : {}) }); @@ -257,6 +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: 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();