diff --git a/backend/__tests__/service/agent-states-listener-check-in.test.js b/backend/__tests__/service/agent-states-listener-check-in.test.js new file mode 100644 index 0000000000..197351ab99 --- /dev/null +++ b/backend/__tests__/service/agent-states-listener-check-in.test.js @@ -0,0 +1,81 @@ +// /agent-states reads listener check-ins (TASK-721), through the real models and +// the real route. The unit suite builds token rows by hand and so cannot see a +// dropped projection field or a stamp written to the wrong store: this one +// writes with `recordListenerCheckIn` and reads back over HTTP. +const express = require('express'); +const request = require('supertest'); +const jwt = require('jsonwebtoken'); +const { setupMongoDb, closeMongoDb } = require('../utils/testUtils'); +const User = require('../../models/User'); +const Pod = require('../../models/Pod'); +const { AgentInstallation } = require('../../models/AgentRegistry'); +const podsRouter = require('../../routes/pods'); +const { recordListenerCheckIn, resetListenerCheckInThrottle } = require('../../services/agentListenerCheckIn'); + +const SECRET = 'agent-states-listener-test'; +const WRAPPER = { runtimeType: 'webhook', host: 'byo' }; + +describe('GET /api/pods/:podId/agent-states reads listener check-ins', () => { + let app; let owner; let pod; let authHeader; + + const install = (agentName, runtimeTokens = []) => AgentInstallation.create({ + agentName, + instanceId: 'default', + podId: pod._id, + version: '1', + installedBy: owner._id, + config: new Map([['runtime', WRAPPER]]), + runtimeTokens, + }); + const botUser = (agentName, agentRuntimeTokens) => User.create({ + username: agentName, + email: `${agentName}@test.com`, + password: 'password123', + isBot: true, + botMetadata: { agentName, instanceId: 'default' }, + agentRuntimeTokens, + }); + const states = async () => { + const res = await request(app).get(`/api/pods/${pod._id}/agent-states`).set('Authorization', authHeader); + expect(res.status).toBe(200); + return Object.fromEntries(res.body.agents.map((a) => [a.agentName, a])); + }; + // The check-in writes are fire-and-forget, so wait for them to land. + const settle = () => new Promise((resolve) => { setTimeout(resolve, 150); }); + + beforeAll(async () => { + process.env.JWT_SECRET = SECRET; + await setupMongoDb(); + app = express(); + app.use(express.json()); + app.use('/api/pods', podsRouter); + owner = await User.create({ username: 'states-owner', email: 'states-owner@test.com', password: 'password123' }); + pod = await Pod.create({ + name: 'States', type: 'chat', createdBy: owner._id, members: [owner._id], + }); + authHeader = `Bearer ${jwt.sign({ id: String(owner._id) }, SECRET)}`; + }); + + afterAll(async () => closeMongoDb()); + beforeEach(() => resetListenerCheckInThrottle()); + + it('an MCP call alone is never-connected; a check-in on the User token flips it to listening', async () => { + const used = new Date(); + await install('mcp-only'); + await botUser('mcp-only', [{ tokenHash: 'h-user', lastUsedAt: used }]); + expect((await states())['mcp-only'].state).toBe('never-connected'); + + recordListenerCheckIn((await User.findOne({ username: 'mcp-only' }))._id, 'h-user'); + await settle(); + expect((await states())['mcp-only'].state).toBe('listening'); + }); + + it('a check-in on an installation-only token flips it to listening', async () => { + await install('install-only', [{ tokenHash: 'h-install', lastUsedAt: new Date() }]); + expect((await states())['install-only'].state).toBe('never-connected'); + + recordListenerCheckIn(null, 'h-install'); + await settle(); + expect((await states())['install-only'].state).toBe('listening'); + }); +}); diff --git a/backend/__tests__/services/agentWebSocketService.test.js b/backend/__tests__/services/agentWebSocketService.test.js index 6d39c5ff77..d539663363 100644 --- a/backend/__tests__/services/agentWebSocketService.test.js +++ b/backend/__tests__/services/agentWebSocketService.test.js @@ -11,6 +11,10 @@ jest.mock('../../models/User', () => ({ updateOne: jest.fn(), })); +const mockRecordListenerCheckIn = jest.fn(); +jest.mock('../../services/agentListenerCheckIn', () => ({ + recordListenerCheckIn: (...args) => mockRecordListenerCheckIn(...args), +})); const mockMarkByoConnectionAttempt = jest.fn(); jest.mock('../../services/onboardingMarks', () => ({ markByoConnectionAttempt: (...args) => mockMarkByoConnectionAttempt(...args), @@ -147,7 +151,13 @@ describe('agentWebSocketService', () => { }); expect(AgentInstallation.updateOne).toHaveBeenCalledWith( { _id: 'install-1', 'runtimeTokens.tokenHash': hash(token) }, - { $set: { 'runtimeTokens.$.lastUsedAt': expect.any(Date) } }, + // An installation-bound token is stamped as a listener check-in too (TASK-721). + { + $set: { + 'runtimeTokens.$.lastUsedAt': expect.any(Date), + 'runtimeTokens.$.lastListenedAt': expect.any(Date), + }, + }, ); expect(result).toEqual({ agentName: 'openclaw', @@ -354,6 +364,25 @@ describe('agentWebSocketService', () => { expect(agentData.lastPong).toBeGreaterThan(initialPong); }); + it('a pong from the live socket is a listener check-in, one from a replaced socket is not', () => { + AgentInstallation.findOne.mockResolvedValue({ agentName: 'openclaw', instanceId: 'default' }); + agentWebSocketService.init(mockIo); + if (connectionHandler) connectionHandler(mockSocket); + mockSocket.agentUserId = 'bot-user-1'; + mockSocket.tokenHash = 'hashed-token'; + const pongHandler = mockSocket.on.mock.calls.find((call) => call[0] === 'pong')?.[1]; + + mockRecordListenerCheckIn.mockClear(); + pongHandler(); + expect(mockRecordListenerCheckIn).toHaveBeenCalledWith('bot-user-1', 'hashed-token'); + + // A reconnect replaced this socket: the old one must not keep the seat listening. + agentWebSocketService.connectedAgents.get('openclaw:default').socket = { id: 'socket-2' }; + mockRecordListenerCheckIn.mockClear(); + pongHandler(); + expect(mockRecordListenerCheckIn).not.toHaveBeenCalled(); + }); + it('sends ping to all connected agents every 30 seconds', () => { AgentInstallation.findOne.mockResolvedValue({ agentName: 'openclaw', diff --git a/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js b/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js index 1472cef81c..44c34a0bc5 100644 --- a/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js +++ b/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js @@ -12,6 +12,7 @@ jest.mock('../../../services/agentEventService', () => ({ list: jest.fn(), listInboxPage: jest.fn(), renewDelivery: jest.fn(), + holdsEvent: jest.fn(), })); jest.mock('../../../services/agentIdentityService', () => ({ buildAgentUsername: jest.fn((agentName, instanceId = 'default') => ( @@ -178,6 +179,55 @@ describe('inbox delivery renewal route', () => { expect(AgentEventService.renewDelivery).toHaveBeenLastCalledWith(eventId, 'wren', 'writer-instance', 'nonce'); expect(res.json).toHaveBeenCalledWith({ renewed: [eventId] }); }); + test('a renew is a listener check-in, so a seat inside a long turn stays listening', async () => { + AgentEventService.renewDelivery.mockResolvedValue(true); + recordListenerCheckIn.mockClear(); + await getRouteHandler('/events/renew', 'post')({ + agentInstallation: { agentName: 'wren', instanceId: 'writer-instance' }, + agentUser: { _id: 'bot-user-1' }, + agentTokenHash: 'hashed-token', + body: { events: [{ eventId, deliveryId: 'nonce' }] }, + }, response()); + expect(recordListenerCheckIn).toHaveBeenCalledWith('bot-user-1', 'hashed-token'); + }); + test('a renew naming an event this agent does not own is not a check-in', async () => { + AgentEventService.renewDelivery.mockResolvedValue(false); + AgentEventService.holdsEvent.mockResolvedValue(false); + recordListenerCheckIn.mockClear(); + const res = response(); + await getRouteHandler('/events/renew', 'post')({ + agentInstallation: { agentName: 'wren', instanceId: 'writer-instance' }, + agentUser: { _id: 'bot-user-1' }, + agentTokenHash: 'hashed-token', + body: { events: [{ eventId, deliveryId: 'stale-nonce' }] }, + }, res); + expect(res.json).toHaveBeenCalledWith({ renewed: [] }); + expect(recordListenerCheckIn).not.toHaveBeenCalled(); + }); + test('a renew after the turn checkpointed the event still checks in, though nothing renews', async () => { + // commonly_checkpoint_event acks mid-turn: the row is acked, so renewDelivery + // matches nothing, but the runner is alive and still working. + AgentEventService.renewDelivery.mockResolvedValue(false); + AgentEventService.holdsEvent.mockResolvedValue(true); + recordListenerCheckIn.mockClear(); + const res = response(); + await getRouteHandler('/events/renew', 'post')({ + agentInstallation: { agentName: 'wren', instanceId: 'writer-instance' }, + agentUser: { _id: 'bot-user-1' }, + agentTokenHash: 'hashed-token', + body: { events: [{ eventId, deliveryId: 'nonce' }] }, + }, res); + expect(AgentEventService.holdsEvent).toHaveBeenLastCalledWith(eventId, 'wren', 'writer-instance'); + expect(res.json).toHaveBeenCalledWith({ renewed: [] }); + expect(recordListenerCheckIn).toHaveBeenCalledWith('bot-user-1', 'hashed-token'); + }); + test('a malformed renew is not a check-in', async () => { + recordListenerCheckIn.mockClear(); + await getRouteHandler('/events/renew', 'post')({ + agentUser: { _id: 'bot-user-1' }, agentTokenHash: 'hashed-token', body: { events: [] }, + }, response()); + expect(recordListenerCheckIn).not.toHaveBeenCalled(); + }); test.each([undefined, [], Array(51).fill({ eventId, deliveryId: 'nonce' }), [{ eventId, deliveryId: '' }], [{ eventId: {}, deliveryId: 'nonce' }]])('rejects malformed or unbounded input %j', async (events) => { AgentEventService.renewDelivery.mockClear(); diff --git a/backend/__tests__/unit/routes/registry.byo-host-stamp.test.js b/backend/__tests__/unit/routes/registry.byo-host-stamp.test.js index 8be6f76d09..d9de18de3d 100644 --- a/backend/__tests__/unit/routes/registry.byo-host-stamp.test.js +++ b/backend/__tests__/unit/routes/registry.byo-host-stamp.test.js @@ -78,7 +78,7 @@ describe('stamped and legacy polling seats both reach the honesty surface', () = // knowable in both directions. const state = deriveAgentState( install({ runtimeType: 'webhook', host: 'byo' }), - [{ lastUsedAt: new Date() }], + [{ lastListenedAt: new Date() }], 'u1', ); diff --git a/backend/__tests__/unit/services/agentEventService.deliveryNonce.test.js b/backend/__tests__/unit/services/agentEventService.deliveryNonce.test.js index 46d2aaba06..7b7a4afe5f 100644 --- a/backend/__tests__/unit/services/agentEventService.deliveryNonce.test.js +++ b/backend/__tests__/unit/services/agentEventService.deliveryNonce.test.js @@ -93,6 +93,26 @@ beforeEach(async () => { await AgentDigestState.deleteMany({}); }); +describe('holdsEvent (the listener check-in on /events/renew)', () => { + const make = (over = {}) => AgentEvent.create({ + agentName: AGENT, instanceId: INSTANCE, podId: POD_ID, type: 'chat.mention', payload: { content: 'q' }, ...over, + }); + + test('true for this agent\'s delivered and checkpointed events, false for pending, other agents and unknown ids', async () => { + const delivered = await make({ status: 'delivered', deliveryNonce: 'n1' }); + const acked = await make({ status: 'acked', deliveryNonce: null }); + const pending = await make({ status: 'pending' }); + const other = await make({ agentName: 'someone-else', status: 'acked' }); + const unknownId = String(new mongoose.Types.ObjectId()); + const holds = (id, agent = AGENT) => AgentEventService.holdsEvent(String(id), agent, INSTANCE); + expect(await holds(delivered._id)).toBe(true); + expect(await holds(acked._id)).toBe(true); + expect(await holds(pending._id)).toBe(false); + expect(await holds(other._id)).toBe(false); + expect(await holds(unknownId)).toBe(false); + }); +}); + describe('legacy ambient digest bridge', () => { test('a push hint does not consume durable inbox state before the reconciling poll', async () => { const ws = require('../../../services/agentWebSocketService'); diff --git a/backend/__tests__/unit/services/agentListenerCheckIn.test.js b/backend/__tests__/unit/services/agentListenerCheckIn.test.js index 8396130ded..7b0a7966c9 100644 --- a/backend/__tests__/unit/services/agentListenerCheckIn.test.js +++ b/backend/__tests__/unit/services/agentListenerCheckIn.test.js @@ -1,8 +1,10 @@ // TASK-719: the listener check-in stamp is its own field, written at most once // a minute per token, and a failed write retries on the next poll. jest.mock('../../../models/User', () => ({ updateOne: jest.fn() })); +jest.mock('../../../models/AgentRegistry', () => ({ AgentInstallation: { updateMany: jest.fn() } })); const User = require('../../../models/User'); +const { AgentInstallation } = require('../../../models/AgentRegistry'); const { recordListenerCheckIn, resetListenerCheckInThrottle, @@ -12,6 +14,7 @@ describe('recordListenerCheckIn', () => { beforeEach(() => { jest.useFakeTimers().setSystemTime(new Date('2026-10-31T12:00:00.000Z')); User.updateOne.mockReset().mockResolvedValue({}); + AgentInstallation.updateMany.mockReset().mockResolvedValue({}); resetListenerCheckInThrottle(); }); @@ -37,11 +40,37 @@ describe('recordListenerCheckIn', () => { expect(User.updateOne).toHaveBeenCalledTimes(3); }); - it('writes nothing without an agent id or a token hash', () => { + it('stamps the installation copy of the token too', () => { + recordListenerCheckIn('agent-1', 'hash-1'); + + expect(AgentInstallation.updateMany).toHaveBeenCalledWith( + { 'runtimeTokens.tokenHash': 'hash-1' }, + { $set: { 'runtimeTokens.$.lastListenedAt': new Date('2026-10-31T12:00:00.000Z') } }, + ); + }); + + it('stamps an installation-only token when there is no bot user row', () => { recordListenerCheckIn(undefined, 'hash-1'); + + expect(User.updateOne).not.toHaveBeenCalled(); + expect(AgentInstallation.updateMany).toHaveBeenCalledTimes(1); + }); + + it('writes nothing without a token hash', () => { recordListenerCheckIn('agent-1', undefined); recordListenerCheckIn('agent-1', ''); expect(User.updateOne).not.toHaveBeenCalled(); + expect(AgentInstallation.updateMany).not.toHaveBeenCalled(); + }); + + it('retries on the next poll when only the installation write fails', async () => { + AgentInstallation.updateMany.mockRejectedValueOnce(new Error('mongo down')); + recordListenerCheckIn('agent-1', 'hash-1'); + await Promise.resolve(); + await Promise.resolve(); + + recordListenerCheckIn('agent-1', 'hash-1'); + expect(AgentInstallation.updateMany).toHaveBeenCalledTimes(2); }); it('retries on the next poll when the write fails, and never throws', async () => { diff --git a/backend/__tests__/unit/services/agentStateService.test.js b/backend/__tests__/unit/services/agentStateService.test.js index 2cc7337385..c1779bdc5e 100644 --- a/backend/__tests__/unit/services/agentStateService.test.js +++ b/backend/__tests__/unit/services/agentStateService.test.js @@ -2,7 +2,7 @@ * agentStateService.deriveAgentState — #891 surface 1's honesty rules, pure. * * Contract under test (each rule earned by a measured review finding): - * - BYO installs derive from lastUsedAt: null → never-connected, + * - BYO installs derive from lastListenedAt: null → never-connected, * fresh → listening, stale → gone-dark * - the union of BOTH token stores decides (finding D: legacy/new split * made live agents read null) @@ -43,7 +43,7 @@ describe('deriveAgentState', () => { test('fresh use on the INSTALLATION token store → listening', () => { const row = deriveAgentState( - byoInstall({ runtimeTokens: [{ lastUsedAt: new Date(NOW - 10_000) }] }), + byoInstall({ runtimeTokens: [{ lastListenedAt: new Date(NOW - 10_000) }] }), [], 'owner-1', NOW, @@ -55,7 +55,7 @@ describe('deriveAgentState', () => { test('fresh use on the USER token store alone also counts — the union decides (finding D)', () => { const row = deriveAgentState( byoInstall(), - [{ lastUsedAt: new Date(NOW - 10_000) }], + [{ lastListenedAt: new Date(NOW - 10_000) }], 'owner-1', NOW, ); @@ -68,7 +68,7 @@ describe('deriveAgentState', () => { }); const live = deriveAgentState( installation, - [{ lastUsedAt: new Date(NOW - 10_000) }], + [{ lastListenedAt: new Date(NOW - 10_000) }], 'owner-1', NOW, ); @@ -76,16 +76,55 @@ describe('deriveAgentState', () => { const stopped = deriveAgentState( installation, - [{ lastUsedAt: new Date(NOW - AGENT_LISTENING_STALE_MS - 1000) }], + [{ lastListenedAt: new Date(NOW - AGENT_LISTENING_STALE_MS - 1000) }], 'owner-1', NOW, ); expect(stopped.state).toBe('gone-dark'); }); + test('an MCP tool call alone is not listening: a fresh lastUsedAt with no check-in is never-connected', () => { + const row = deriveAgentState( + byoInstall({ runtimeTokens: [{ lastUsedAt: new Date(NOW - 10_000) }] }), + [{ lastUsedAt: new Date(NOW - 10_000) }], + 'owner-1', + NOW, + ); + expect(row.state).toBe('never-connected'); + expect(row.lastUsedAt).toBeNull(); + expect(row.fixCommand).toBe('commonly agent run sam-agent'); + }); + + test('a check-in still decides over a later MCP call: listening, and lastUsedAt reports the check-in', () => { + const checkIn = new Date(NOW - 20_000); + const row = deriveAgentState( + byoInstall(), + [{ lastUsedAt: new Date(NOW - 1_000), lastListenedAt: checkIn }], + 'owner-1', + NOW, + ); + expect(row.state).toBe('listening'); + expect(row.lastUsedAt).toBe(checkIn.toISOString()); + }); + + test('an old seat after the backfill (lastListenedAt = lastUsedAt) keeps its verdict instead of flipping to never-connected', () => { + const lastSeen = new Date(NOW - 3 * 24 * 60 * 60 * 1000); + const before = deriveAgentState(byoInstall(), [{ lastUsedAt: lastSeen }], 'owner-1', NOW); + const after = deriveAgentState( + byoInstall(), + [{ lastUsedAt: lastSeen, lastListenedAt: lastSeen }], + 'owner-1', + NOW, + ); + // Without the backfill the deploy would read this seat as never-connected, + // and the stalled-connect scan would nudge a room that has long been quiet. + expect(before.state).toBe('never-connected'); + expect(after.state).toBe('gone-dark'); + }); + test('stale use → gone-dark, and the owner gets the wake command', () => { const row = deriveAgentState( - byoInstall({ runtimeTokens: [{ lastUsedAt: new Date(NOW - AGENT_LISTENING_STALE_MS - 1000) }] }), + byoInstall({ runtimeTokens: [{ lastListenedAt: new Date(NOW - AGENT_LISTENING_STALE_MS - 1000) }] }), [], 'owner-1', NOW, @@ -150,7 +189,7 @@ describe('deriveAgentState', () => { const row = deriveAgentState( byoInstall({ config: { runtime: { runtimeType: 'moltbot', host: 'cloud' } }, - runtimeTokens: [{ lastUsedAt: new Date(NOW - 5_000) }], // fresh, and still not trusted + runtimeTokens: [{ lastListenedAt: new Date(NOW - 5_000) }], // fresh, and still not trusted }), [], 'owner-1', @@ -164,7 +203,7 @@ describe('deriveAgentState', () => { const row = deriveAgentState( byoInstall({ config: { runtime: { runtimeType: 'webhook', host: 'cloud' } }, - runtimeTokens: [{ lastUsedAt: new Date(NOW - 5_000) }], + runtimeTokens: [{ lastListenedAt: new Date(NOW - 5_000) }], }), [], 'owner-1', diff --git a/backend/__tests__/unit/services/neverConnectedMentionService.test.js b/backend/__tests__/unit/services/neverConnectedMentionService.test.js index 77d8cca0a7..2e219390a0 100644 --- a/backend/__tests__/unit/services/neverConnectedMentionService.test.js +++ b/backend/__tests__/unit/services/neverConnectedMentionService.test.js @@ -88,7 +88,7 @@ it('stays silent for a bot sender', async () => { }); it('stays silent once the agent has a live credential use', async () => { - mockFindBot.mockReturnValue(chain(botRow([{ lastUsedAt: new Date(NOW.getTime() - 60 * 1000) }]))); + mockFindBot.mockReturnValue(chain(botRow([{ lastListenedAt: new Date(NOW.getTime() - 60 * 1000) }]))); expect(await noteNeverConnectedMention(args)).toBe(false); expect(mockEmitNotConnected).not.toHaveBeenCalled(); }); diff --git a/backend/__tests__/unit/services/stalledConnectService.test.js b/backend/__tests__/unit/services/stalledConnectService.test.js index bf010ad1df..f3cdc66618 100644 --- a/backend/__tests__/unit/services/stalledConnectService.test.js +++ b/backend/__tests__/unit/services/stalledConnectService.test.js @@ -73,7 +73,12 @@ const setup = ({ installs = [install()], userTokens = [], open = [] } = {}) => { limit: () => ({ lean: () => Promise.resolve(installs) }), }); mockUserFindOne.mockReturnValue({ - select: () => ({ lean: () => Promise.resolve({ agentRuntimeTokens: userTokens }) }), + // A mock that ignores its projection hides a dropped field: assert it. + select: (fields) => { + expect(fields).toContain('agentRuntimeTokens.lastListenedAt'); + expect(fields).toContain('agentRuntimeTokens.createdAt'); + return { lean: () => Promise.resolve({ agentRuntimeTokens: userTokens }) }; + }, }); mockUserFindById.mockReturnValue({ select: () => ({ lean: () => Promise.resolve({ username: 'alice' }) }), @@ -348,6 +353,18 @@ describe('stalledConnectService.scan', () => { }); it('says nothing when the token has been used — that is not stalled', async () => { + setup({ + installs: [install({ + runtimeTokens: [{ createdAt: OLD_TOKEN, lastListenedAt: new Date(NOW.getTime() - MINUTE) }], + })], + }); + const r = await scan({ now: NOW }); + + expect(r.nudged).toHaveLength(0); + expect(r.skippedNotStalled).toBe(1); + }); + + it('nudges a seat that made MCP calls but never checked in as a listener', async () => { setup({ installs: [install({ runtimeTokens: [{ createdAt: OLD_TOKEN, lastUsedAt: new Date(NOW.getTime() - MINUTE) }], @@ -355,8 +372,22 @@ describe('stalledConnectService.scan', () => { }); const r = await scan({ now: NOW }); + expect(r.nudged).toHaveLength(1); + expect(mockPost).toHaveBeenCalledTimes(1); + }); + + it('does not nudge an old seat whose lastListenedAt was backfilled from lastUsedAt', async () => { + const lastSeen = new Date(NOW.getTime() - 3 * 24 * 60 * MINUTE); + setup({ + installs: [install({ + runtimeTokens: [{ createdAt: OLD_TOKEN, lastUsedAt: lastSeen, lastListenedAt: lastSeen }], + })], + }); + const r = await scan({ now: NOW }); + expect(r.nudged).toHaveLength(0); expect(r.skippedNotStalled).toBe(1); + expect(mockPost).not.toHaveBeenCalled(); }); it('claims the episode BEFORE posting, so a losing race costs a nudge not a duplicate', async () => { @@ -477,7 +508,7 @@ describe('keeping the forward promise', () => { const usedAt = new Date(NOW.getTime() - 10 * MINUTE); setup({ installs: [], open: [openEpisode] }); mockInstallFindById.mockReturnValue({ - lean: () => Promise.resolve(install({ runtimeTokens: [{ createdAt: OLD_TOKEN, lastUsedAt: usedAt }] })), + lean: () => Promise.resolve(install({ runtimeTokens: [{ createdAt: OLD_TOKEN, lastListenedAt: usedAt }] })), }); const r = await scan({ now: NOW }); @@ -498,7 +529,7 @@ describe('keeping the forward promise', () => { const usedAt = new Date(NOW.getTime() - 180 * MINUTE); setup({ installs: [], open: [openEpisode] }); mockInstallFindById.mockReturnValue({ - lean: () => Promise.resolve(install({ runtimeTokens: [{ createdAt: OLD_TOKEN, lastUsedAt: usedAt }] })), + lean: () => Promise.resolve(install({ runtimeTokens: [{ createdAt: OLD_TOKEN, lastListenedAt: usedAt }] })), }); await scan({ now: NOW }); @@ -513,7 +544,7 @@ describe('keeping the forward promise', () => { const usedAt = new Date(NOW.getTime() - 10 * MINUTE); setup({ installs: [], open: [{ ...openEpisode, nudgeMessageId: 'msg-nudge-1' }] }); mockInstallFindById.mockReturnValue({ - lean: () => Promise.resolve(install({ runtimeTokens: [{ createdAt: OLD_TOKEN, lastUsedAt: usedAt }] })), + lean: () => Promise.resolve(install({ runtimeTokens: [{ createdAt: OLD_TOKEN, lastListenedAt: usedAt }] })), }); await scan({ now: NOW }); @@ -525,7 +556,7 @@ describe('keeping the forward promise', () => { const usedAt = new Date(NOW.getTime() - 10 * MINUTE); setup({ installs: [], open: [{ ...openEpisode, nudgeMessageId: 'msg-nudge-2' }] }); mockInstallFindById.mockReturnValue({ - lean: () => Promise.resolve(install({ runtimeTokens: [{ createdAt: OLD_TOKEN, lastUsedAt: usedAt }] })), + lean: () => Promise.resolve(install({ runtimeTokens: [{ createdAt: OLD_TOKEN, lastListenedAt: usedAt }] })), }); mockPost.mockRejectedValueOnce(new Error('post down')); diff --git a/backend/routes/agentsRuntime.ts b/backend/routes/agentsRuntime.ts index 6fde8107c4..d8b1dce3ba 100644 --- a/backend/routes/agentsRuntime.ts +++ b/backend/routes/agentsRuntime.ts @@ -1394,12 +1394,22 @@ router.post('/events/renew', phase4RateLimit, agentRuntimeAuth, async (req: any, || req.agentUser?.botMetadata?.agentType || req.agentUser?.username; const instanceId = req.agentInstallation?.instanceId || req.agentUser?.botMetadata?.instanceId || 'default'; if (!agentName) return res.status(403).json({ message: 'Agent token not authorized for events' }); + // The runner polls /events only between turns, so a turn longer than the + // listening window would age a working seat into gone-dark. Renewing a + // delivery is the runner saying it is alive and holding work, so it counts + // as a check-in, but only when the request names an event this agent owns: + // made-up ids must not move an MCP-only seat out of never-connected. A turn + // that already checkpointed an event (the MCP tool acks mid-turn, so the + // renew no longer matches a `delivered` row) is still that agent's own event. + // The MCP checkpoint tool uses /ack, not this route. try { - const renewed = await Promise.all(events.map(async (entry: { eventId: string; deliveryId: string }) => ( - await AgentEventService.renewDelivery(entry.eventId, agentName, instanceId, entry.deliveryId) - ? entry.eventId : null - ))); - return res.json({ renewed: renewed.filter(Boolean) }); + const results = await Promise.all(events.map(async (entry: { eventId: string; deliveryId: string }) => { + const renewedOne = await AgentEventService.renewDelivery(entry.eventId, agentName, instanceId, entry.deliveryId); + const owned = renewedOne || await AgentEventService.holdsEvent(entry.eventId, agentName, instanceId); + return { eventId: entry.eventId, renewedOne, owned }; + })); + if (results.some((r) => r.owned)) recordListenerCheckIn(req.agentUser?._id, req.agentTokenHash); + return res.json({ renewed: results.filter((r) => r.renewedOne).map((r) => r.eventId) }); } catch (error: any) { console.error('Error renewing inbox deliveries:', error.message); return res.status(500).json({ message: 'Could not renew inbox deliveries' }); diff --git a/backend/routes/pods.ts b/backend/routes/pods.ts index 6b50cf32fe..21713c2871 100644 --- a/backend/routes/pods.ts +++ b/backend/routes/pods.ts @@ -569,9 +569,9 @@ router.get('/:podId/agent-states', agentStatesRateLimit, auth, async (req: AuthR 'botMetadata.instanceId': String(inst.instanceId || 'default'), })); const agentUsers = await User.find({ isBot: true, $or: identityFilters }) - .select('botMetadata.agentName botMetadata.instanceId agentRuntimeTokens.lastUsedAt') + .select('botMetadata.agentName botMetadata.instanceId agentRuntimeTokens.lastListenedAt') .lean(); - const userTokensByIdentity = new Map>(); + const userTokensByIdentity = new Map>(); (agentUsers || []).forEach((u: any) => { const key = `${String(u.botMetadata?.agentName || '').toLowerCase()}:${String(u.botMetadata?.instanceId || 'default').toLowerCase()}`; userTokensByIdentity.set(key, u.agentRuntimeTokens || []); diff --git a/backend/routes/registry/install.ts b/backend/routes/registry/install.ts index 1540187e3d..ef78313ceb 100644 --- a/backend/routes/registry/install.ts +++ b/backend/routes/registry/install.ts @@ -1008,7 +1008,7 @@ installRouter.post('/install', logAgentInstallRefusal('/api/registry/install'), // fleet review 2026-08-14). `deriveAgentState` unions // `installation.runtimeTokens` with a USER's `agentRuntimeTokens`, and // the user it means is the agent's bot row — that is where a polling - // wrapper stamps `lastUsedAt`. An earlier draft passed the human + // wrapper stamps its check-in. An earlier draft passed the human // installer's tokens, a store a wrapper never writes, so a reused and // demonstrably live seat computed as never-connected and would have // been announced as dead. Same identity match the pod roster uses @@ -1026,8 +1026,8 @@ installRouter.post('/install', logAgentInstallRefusal('/api/registry/install'), isBot: true, 'botMetadata.agentName': agent.agentName, 'botMetadata.instanceId': normalizedInstanceId, - }).select('agentRuntimeTokens.lastUsedAt').lean() as - { agentRuntimeTokens?: { lastUsedAt?: Date | string | null }[] } | null; + }).select('agentRuntimeTokens.lastListenedAt').lean() as + { agentRuntimeTokens?: { lastListenedAt?: Date | string | null }[] } | null; // `config` is `{ type: Map, of: Mixed }` (AgentRegistry.ts:235), so // on a LIVE Mongoose document `installation.config.runtime` is // undefined — a Map needs `.get()`. `deriveAgentState` reads diff --git a/backend/services/agentEventService.ts b/backend/services/agentEventService.ts index 2fdb28f67c..75a9f829e9 100644 --- a/backend/services/agentEventService.ts +++ b/backend/services/agentEventService.ts @@ -2284,6 +2284,22 @@ class AgentEventService { return result.modifiedCount === 1; } + // True when the event is this agent's own and is mid-flight or already + // checkpointed. A turn that checkpoints (`commonly_checkpoint_event` acks + // mid-turn, which clears the nonce) keeps renewing, and `renewDelivery` then + // returns false because the row is `acked`. The listener check-in on + // /events/renew uses this to tell that runner from a request with made-up ids. + static async holdsEvent(eventId: string, agentName: string, instanceId: string): Promise { + if (!eventId) return false; + const found = await AgentEvent.exists({ + _id: eventId, + agentName: String(agentName).toLowerCase(), + instanceId: String(instanceId || 'default'), + status: { $in: ['delivered', 'acked'] }, + }); + return Boolean(found); + } + // ADR-012 §3: acknowledge is status-gated. Lifecycle: // pending → delivered (on `list()` claim) // delivered → acked (here) diff --git a/backend/services/agentListenerCheckIn.ts b/backend/services/agentListenerCheckIn.ts index 8342cc5ddd..dac0a88e73 100644 --- a/backend/services/agentListenerCheckIn.ts +++ b/backend/services/agentListenerCheckIn.ts @@ -1,9 +1,11 @@ import User from '../models/User'; +import { AgentInstallation } from '../models/AgentRegistry'; /** * A listener check-in is the one signal that "mentions will wake this agent": - * the events poll (`GET /api/agents/runtime/events`) or the agent WebSocket - * connect. `lastUsedAt` cannot say that. `agentRuntimeAuth` stamps it on any + * the events poll (`GET /api/agents/runtime/events`), a delivery renew + * (`POST .../events/renew`, which keeps a seat alive through a long turn) or the + * agent WebSocket connect. `lastUsedAt` cannot say that. `agentRuntimeAuth` stamps it on any * authenticated call, so an MCP tool call alone used to read as "listening" * (TASK-719). The stamp lives on the same embedded token record, in its own * field, so the two meanings stay separate. @@ -12,33 +14,39 @@ import User from '../models/User'; * throttled per token; the process-lifetime map is bounded by the count of * distinct listener tokens, and a restart only costs one extra write each. * - * Only tokens on the agent's User row are stamped. A token that exists only on - * `AgentInstallation.runtimeTokens` matches nothing here, and - * `/api/users/me/agent-connection` never read that list (it reads the User - * row), so such a token never showed as connected before this change either. - * Every issuer in use (`issueRuntimeTokenForAgent`) writes the same hash to - * both places; `issueRuntimeTokenForInstallation` has no callers. + * A token lives on the agent's User row, on `AgentInstallation.runtimeTokens`, + * or both (`issueRuntimeTokenForAgent` writes the same hash to both places). + * The check-in stamps whichever copies exist, so `deriveAgentState`, which + * reads both stores, sees a seat whose only token is installation-bound as + * listening. Before this, such a token could not be stamped and would have + * read as never-connected the moment the derivation moved off `lastUsedAt`. */ const CHECK_IN_INTERVAL_MS = 60 * 1000; const lastWrites = new Map(); export function recordListenerCheckIn(agentUserId: unknown, tokenHash: unknown): void { - if (!agentUserId || typeof tokenHash !== 'string' || !tokenHash) return; + if (typeof tokenHash !== 'string' || !tokenHash) return; const now = Date.now(); const prev = lastWrites.get(tokenHash); if (prev && now - prev < CHECK_IN_INTERVAL_MS) return; lastWrites.set(tokenHash, now); + const at = new Date(now); + // Retry on the next poll instead of staying "waiting" for a minute. + const retryNextPoll = () => { lastWrites.delete(tokenHash); }; try { - User.updateOne( - { _id: agentUserId, 'agentRuntimeTokens.tokenHash': tokenHash }, - { $set: { 'agentRuntimeTokens.$.lastListenedAt': new Date(now) } }, - ).catch(() => { - // Retry on the next poll instead of staying "waiting" for a minute. - lastWrites.delete(tokenHash); - }); + if (agentUserId) { + User.updateOne( + { _id: agentUserId, 'agentRuntimeTokens.tokenHash': tokenHash }, + { $set: { 'agentRuntimeTokens.$.lastListenedAt': at } }, + ).catch(retryNextPoll); + } + AgentInstallation.updateMany( + { 'runtimeTokens.tokenHash': tokenHash }, + { $set: { 'runtimeTokens.$.lastListenedAt': at } }, + ).catch(retryNextPoll); } catch { // Presence is best-effort: it must never turn a poll into an error. - lastWrites.delete(tokenHash); + retryNextPoll(); } } diff --git a/backend/services/agentStateService.ts b/backend/services/agentStateService.ts index 4548d8f290..499ece74bb 100644 --- a/backend/services/agentStateService.ts +++ b/backend/services/agentStateService.ts @@ -6,9 +6,12 @@ * `lastUsedAt` lying for every class except the polling one: * * - BYO wrapper/MCP installs (host 'byo' or legacy local-cli, no - * webhookUrl): the wrapper polls every ~5s while alive, so lastUsedAt is - * an honest reachability signal. null → never-connected · fresh → - * listening · stale → gone-dark. + * webhookUrl): the wrapper polls every ~5s while alive, so the last + * LISTENER check-in (`lastListenedAt`, stamped by the events poll, the + * WebSocket connect and a delivery renew) is an honest reachability signal. null → + * never-connected · fresh → listening · stale → gone-dark. `lastUsedAt` + * is not: any authenticated call stamps it, so one MCP tool call made a + * seat nobody was running read as listening (TASK-719, TASK-721). * - push webhooks (webhookUrl set) and native runtime: reachable by * construction — no dot, no guess. * - everything else (gateway moltbots, cloud): 'unknown'. The gateway @@ -20,9 +23,9 @@ * an instruction addressed to someone who cannot perform it is the P1/P2 * contradiction the review caught. * - * lastUsedAt is the max across BOTH token stores (User.agentRuntimeTokens - * and AgentInstallation.runtimeTokens) — the legacy/new split that made - * live agents read null (finding D). Pure function: callers fetch, this + * The check-in time is the max across BOTH token stores + * (User.agentRuntimeTokens and AgentInstallation.runtimeTokens) — the + * legacy/new split that made live agents read null (finding D). Pure function: callers fetch, this * derives; the seam is what makes the honesty rules testable without a DB. */ import { hostedRunsNative } from './hostedEngine'; @@ -32,7 +35,7 @@ export const AGENT_LISTENING_STALE_MS = 5 * 60 * 1000; export type AgentReachState = 'listening' | 'gone-dark' | 'never-connected' | 'reachable' | 'unknown'; -interface TokenRow { lastUsedAt?: Date | string | null } +interface TokenRow { lastUsedAt?: Date | string | null; lastListenedAt?: Date | string | null } interface InstallationRow { agentName?: string; @@ -48,6 +51,10 @@ export interface AgentStateRow { instanceId: string; displayName: string; state: AgentReachState; + /** + * For a polling BYO seat, the last LISTENER check-in, not the last token use. + * The key keeps its old name because the shell reads it (V2Thread "last seen"). + */ lastUsedAt: string | null; isOwner: boolean; /** A Commonly-hosted agent ("Run it here"); the beta allows one per account. */ @@ -55,12 +62,12 @@ export interface AgentStateRow { fixCommand?: string; } -const latestUse = (tokenLists: TokenRow[][]): Date | null => { +const latestListen = (tokenLists: TokenRow[][]): Date | null => { let latest: Date | null = null; for (const list of tokenLists) { for (const row of list || []) { - if (!row?.lastUsedAt) continue; - const used = new Date(row.lastUsedAt as string); + if (!row?.lastListenedAt) continue; + const used = new Date(row.lastListenedAt as string); if (Number.isNaN(used.getTime())) continue; if (!latest || used > latest) latest = used; } @@ -120,7 +127,7 @@ export function deriveAgentState( } else if (!isByo) { state = 'unknown'; } else { - lastUsedAt = latestUse([installation.runtimeTokens || [], userTokens || []]); + lastUsedAt = latestListen([installation.runtimeTokens || [], userTokens || []]); if (!lastUsedAt) state = 'never-connected'; else if (now - lastUsedAt.getTime() <= AGENT_LISTENING_STALE_MS) state = 'listening'; else state = 'gone-dark'; diff --git a/backend/services/agentWebSocketService.ts b/backend/services/agentWebSocketService.ts index a33410f429..481642b8ad 100644 --- a/backend/services/agentWebSocketService.ts +++ b/backend/services/agentWebSocketService.ts @@ -14,11 +14,14 @@ const { hash } = require('../utils/secret'); const { markByoConnectionAttempt } = require('./onboardingMarks'); // eslint-disable-next-line @typescript-eslint/no-require-imports const { embeddedRuntimeTokenExpired, runtimeCredentialRefusalReason } = require('./runtimeCredentialGate'); +// eslint-disable-next-line @typescript-eslint/no-require-imports +const { recordListenerCheckIn } = require('./agentListenerCheckIn'); interface AgentSocket { agentName: string; instanceId: string; agentUserId: string | null; + tokenHash?: string; agentKey: string; subscribedPods: Set; handshake: { auth?: { token?: string } }; @@ -82,7 +85,7 @@ interface InstallationDoc { status?: string; displayName?: string; config?: { runtime?: { runtimeType?: string; host?: string; webhookUrl?: string } }; - runtimeTokens?: Array<{ tokenHash: string; lastUsedAt?: Date }>; + runtimeTokens?: Array<{ tokenHash: string; lastUsedAt?: Date; lastListenedAt?: Date }>; } const normalizeTokenIdentityValue = (value: unknown): string => ( @@ -152,6 +155,7 @@ class AgentWebSocketService { socket.agentName = agentInfo.agentName; socket.instanceId = agentInfo.instanceId || 'default'; socket.agentUserId = agentInfo.agentUserId || null; + socket.tokenHash = token.startsWith('cm_agent_') ? hash(token) : undefined; socket.agentKey = `${socket.agentName}:${socket.instanceId}`; socket.subscribedPods = new Set(); @@ -202,6 +206,10 @@ class AgentWebSocketService { // socket refresh the replacement's liveness timestamp. if (data?.socket === socket) { data.lastPong = Date.now(); + // A socket that stays connected is still a listener. The connect + // stamp alone would age it into gone-dark after the listening window. + // The helper throttles per token, so a 30s ping costs one write a minute. + recordListenerCheckIn(socket.agentUserId, socket.tokenHash); } }); @@ -449,9 +457,16 @@ class AgentWebSocketService { .find((t: { tokenHash?: string; lastUsedAt?: Date }) => t.tokenHash === tokenHash); const isFirstLegacyUse = !wsInstallTokenRecord?.lastUsedAt; try { + // Same as the User-row copy above: a connect is a listener check-in. + const connectedAt = new Date(); await AgentInstallation.updateOne( { _id: installation._id, 'runtimeTokens.tokenHash': tokenHash }, - { $set: { 'runtimeTokens.$.lastUsedAt': new Date() } }, + { + $set: { + 'runtimeTokens.$.lastUsedAt': connectedAt, + 'runtimeTokens.$.lastListenedAt': connectedAt, + }, + }, ); } catch (err) { console.warn('Failed to update agent token usage:', (err as Error).message); diff --git a/backend/services/kernelWorkSweepService.ts b/backend/services/kernelWorkSweepService.ts index 296987551a..262b19ca0b 100644 --- a/backend/services/kernelWorkSweepService.ts +++ b/backend/services/kernelWorkSweepService.ts @@ -58,8 +58,10 @@ const TASK_CLAIM_LEASE_MS = 30 * 60 * 1000; // // LIVE means one state and one only. `deriveAgentState` returns five, and only // `listening` is backed by evidence: for BYO/wrapper installs the seat polls -// every ~5s, so a fresh `lastUsedAt` is a measurement. The other four are not -// the opposite of live, they are differently-shaped: +// every ~5s, so a fresh listener check-in (`lastListenedAt`) is a +// measurement. A holder that only makes MCP calls never checks in, so it reads +// never-connected and gets no deferral: it cannot receive the lease warning. +// The other four are not the opposite of live, they are differently-shaped: // - 'reachable' is asserted BY CONSTRUCTION for native + push-webhook // installs. Deferring on it would defer forever on a class whose liveness // was never checked. diff --git a/backend/services/neverConnectedMentionService.ts b/backend/services/neverConnectedMentionService.ts index abf06abeea..4716ae9417 100644 --- a/backend/services/neverConnectedMentionService.ts +++ b/backend/services/neverConnectedMentionService.ts @@ -54,7 +54,7 @@ export async function noteNeverConnectedMention({ }).select('username profilePicture botMetadata agentRuntimeTokens').lean() as { _id: unknown; username?: string; profilePicture?: string | null; botMetadata?: { displayName?: string; instanceId?: string }; - agentRuntimeTokens?: Array<{ lastUsedAt?: Date | string | null }>; + agentRuntimeTokens?: Array<{ lastListenedAt?: Date | string | null }>; } | null; if (!agent) return false; diff --git a/backend/services/stalledConnectService.ts b/backend/services/stalledConnectService.ts index 9498c058c3..eb9780af7b 100644 --- a/backend/services/stalledConnectService.ts +++ b/backend/services/stalledConnectService.ts @@ -126,8 +126,8 @@ const botTokensFor = async (agentName: string, instanceId: string) => { isBot: true, 'botMetadata.agentName': agentName, 'botMetadata.instanceId': instanceId, - }).select('agentRuntimeTokens.lastUsedAt agentRuntimeTokens.createdAt').lean() as - { agentRuntimeTokens?: Array<{ lastUsedAt?: Date; createdAt?: Date }> } | null; + }).select('agentRuntimeTokens.lastListenedAt agentRuntimeTokens.createdAt').lean() as + { agentRuntimeTokens?: Array<{ lastListenedAt?: Date; createdAt?: Date }> } | null; return row?.agentRuntimeTokens || []; }; @@ -381,7 +381,7 @@ const resolveConnected = async ({ const handle = buildAgentUsername ? buildAgentUsername(String(ep.agentName), String(ep.instanceId)) : String(ep.agentName); - // `lastUsedAt` leaving null is a PERMANENT transition, so this timestamp + // `lastListenedAt` leaving null is a PERMANENT transition, so this timestamp // survives a wrapper that connected and stopped again between passes — // which is why the confirmation needs no time bound (ux-lead, 53307). const firstSeenAt = derived.lastUsedAt ? new Date(derived.lastUsedAt) : now; diff --git a/docs/design/agent-status-honesty.md b/docs/design/agent-status-honesty.md index b99059d429..12c569cfd9 100644 --- a/docs/design/agent-status-honesty.md +++ b/docs/design/agent-status-honesty.md @@ -38,7 +38,7 @@ all?* — then *what does the signal say?* | runtime class | honest liveness signal | what `lastUsedAt` actually measures there | |---|---|---| -| **CLI wrapper** (BYO laptop) | `lastUsedAt` — the wrapper polls CAP continuously | reachability; the only class where "stale = gone dark" holds | +| **CLI wrapper** (BYO laptop) | `lastListenedAt` — the events poll, the WebSocket connect or a delivery renew (any other call stamps `lastUsedAt` only, TASK-721) | reachability; the only class where "stale = gone dark" holds | | **moltbot / gateway** | needs route diversity (below) | last *task*, not reachability — 23 agents share one value: the gateway's boot timestamp | | **native** (first-party apps) | **`AgentRun`** — status, trigger, error, turns, toolCalls; a complete execution log | nothing: permanently null; null is *uninformative*, never healthy (52991, replacing 52988's first reading) | | **webhook** | config completeness statically; delivery outcomes at runtime | nothing: push-only — the backend posts the reply, the agent never authenticates in | diff --git a/docs/runbooks/listener-check-in-backfill.md b/docs/runbooks/listener-check-in-backfill.md index 8ca95c80f8..f5b90f3911 100644 --- a/docs/runbooks/listener-check-in-backfill.md +++ b/docs/runbooks/listener-check-in-backfill.md @@ -46,10 +46,24 @@ live listener's next poll writes the real value. subdocument has no ObjectId `_id`; the script cannot stamp them, and they would flip from gone-dark to never-connected. Resolve them (or accept them) first. - A write that comes back unacknowledged counts as `failed`, never as skipped. -- **Deploy gate: installation tokens need a live stamp first.** Today only the - User-row copy of a token is stamped on a listener check-in - (`agentListenerCheckIn`, #3114). A token that exists only in - `AgentInstallation.runtimeTokens`, backfilled or first used after the freeze, never - gets a live `lastListenedAt`, so a derivation that reads the field would read it as - gone-dark or never-connected. The derivation PR must not ship before that stamp - exists. +- **Installation tokens now get a live stamp in the same PR as the derivation.** + `agentListenerCheckIn` stamps both copies of a token (the User row and + `AgentInstallation.runtimeTokens`), and the WebSocket connect stamps an + installation-only token as well. A token first used after the freeze therefore + gets its `lastListenedAt` from its next poll instead of reading never-connected. +- **Order on prod:** deploy the check-in writer (#3114), run the backfill, then + deploy the derivation. The derivation deploy is the only step that changes what + a seat reads, and it is held until the apply is confirmed. +- **What changes after the switch.** `deriveAgentState` reads `lastListenedAt`, so a + seat that has only made MCP calls reads never-connected and the stalled-connect + scan nudges it, which is TASK-721. `/agent-states` keeps its `lastUsedAt` key, + which now carries the last listener check-in. +- **Re-run the backfill right before the derivation deploy.** It only fills null or + absent stamps, so a second run is safe. It closes the window for an + installation-only token first used after the freeze and stopped before the deploy, + which would otherwise read never-connected. +- **Two readers still use `lastUsedAt`.** `deriveAgentConnection` + (`agentConnectionStateService`, the Your Team card) and the activity bucket read + token use, and the credential ledger has no `lastListenedAt`. For an MCP-only seat + the card can read connected while the room banner reads not connected. Tracked as a + follow-up, not part of this switch.