From df75f200b0a235de63fc0f381ef06743b0fd131e Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:03:21 -0700 Subject: [PATCH 1/9] feat(agents): backfill lastListenedAt from lastUsedAt for existing runtime tokens, dry run first (TASK-721) Agent state is about to read listener check-ins. Tokens that predate lastListenedAt have no value, so without this every long-gone seat would flip from gone-dark to never-connected on deploy. The script stamps lastListenedAt = lastUsedAt per token element, insert-only against live writers, and its apply run touches only the id set a dry run froze. Installation tokens gain the field so the script (and a later stamp) can write it. Co-Authored-By: Claude Sonnet 5.5 Co-authored-by: Keel Agent-Model: claude/claude-opus-5-5/high --- .../backfill-listener-check-in.db.test.js | 95 +++++++++ .../backfill-listener-check-in.test.js | 56 +++++ backend/models/AgentRegistry.ts | 3 + backend/package.json | 1 + backend/scripts/backfill-listener-check-in.ts | 193 ++++++++++++++++++ 5 files changed, 348 insertions(+) create mode 100644 backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js create mode 100644 backend/__tests__/unit/scripts/backfill-listener-check-in.test.js create mode 100644 backend/scripts/backfill-listener-check-in.ts diff --git a/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js b/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js new file mode 100644 index 0000000000..505cae7e16 --- /dev/null +++ b/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js @@ -0,0 +1,95 @@ +// The production deps against the real User and AgentInstallation schemas. The +// sibling suite stubs the deps, so it cannot see a field the schema strips or a +// filter that does not match. +/* eslint-disable global-require */ +const { MongoMemoryServer } = require('mongodb-memory-server'); +const mongoose = require('mongoose'); + +let mongod; +let User; +let AgentInstallation; +let backfillListenerCheckIn; +let productionDeps; + +const USED = new Date('2026-10-01T00:00:00Z'); +const LISTENED = new Date('2026-10-30T00:00:00Z'); + +beforeAll(async () => { + mongod = await MongoMemoryServer.create(); + await mongoose.connect(mongod.getUri()); + User = require('../../../models/User'); + ({ AgentInstallation } = require('../../../models/AgentRegistry')); + ({ backfillListenerCheckIn, productionDeps } = require('../../../scripts/backfill-listener-check-in')); +}); +afterEach(async () => { + await User.deleteMany({}); + await AgentInstallation.deleteMany({}); +}); +afterAll(async () => { + await mongoose.disconnect(); + await mongod.stop(); +}); + +const makeAgentUser = (tokens) => User.create({ + username: `bot${Math.floor(Math.random() * 1e9)}`, + email: `bot${Math.floor(Math.random() * 1e9)}@agents.commonly.local`, + password: 'x'.repeat(12), + isBot: true, + agentRuntimeTokens: tokens, +}); + +const makeInstall = (tokens) => AgentInstallation.collection.insertOne({ + agentName: 'a', instanceId: 'default', podId: new mongoose.Types.ObjectId(), status: 'active', runtimeTokens: tokens, +}); + +describe('backfill-listener-check-in against the real schemas', () => { + it('stamps only used, unstamped tokens in both stores, and never touches a token hash', async () => { + const user = await makeAgentUser([ + { tokenHash: 'secret-1', lastUsedAt: USED }, + { tokenHash: 'secret-2' }, + { tokenHash: 'secret-3', lastUsedAt: USED, lastListenedAt: LISTENED }, + ]); + const inst = await makeInstall([ + { _id: new mongoose.Types.ObjectId(), tokenHash: 'secret-4', lastUsedAt: USED }, + ]); + + const deps = productionDeps(); + const listed = await deps.listCandidates(); + expect(listed).toHaveLength(2); + expect(JSON.stringify(listed)).not.toContain('secret'); + + const report = await backfillListenerCheckIn(deps, { apply: true, frozen: listed }); + expect(report).toMatchObject({ applied: 2, skippedAlreadyStamped: 0, failed: 0 }); + + const after = await User.findById(user._id).lean(); + expect(after.agentRuntimeTokens[0].lastListenedAt).toEqual(USED); + expect(after.agentRuntimeTokens[1].lastListenedAt).toBeUndefined(); + expect(after.agentRuntimeTokens[2].lastListenedAt).toEqual(LISTENED); + const installAfter = await AgentInstallation.collection.findOne({ _id: inst.insertedId }); + expect(installAfter.runtimeTokens[0].lastListenedAt).toEqual(USED); + }); + + it('never overwrites a stamp a running listener wrote after the dry run', async () => { + const user = await makeAgentUser([{ tokenHash: 'secret-1', lastUsedAt: USED }]); + const deps = productionDeps(); + const frozen = await deps.listCandidates(); + await User.updateOne( + { _id: user._id }, + { $set: { 'agentRuntimeTokens.0.lastListenedAt': LISTENED } }, + ); + const report = await backfillListenerCheckIn(deps, { apply: true, frozen }); + expect(report).toMatchObject({ applied: 0, skippedAlreadyStamped: 1 }); + const after = await User.findById(user._id).lean(); + expect(after.agentRuntimeTokens[0].lastListenedAt).toEqual(LISTENED); + }); + + it('is idempotent: a second apply of the same set changes nothing', async () => { + await makeAgentUser([{ tokenHash: 'secret-1', lastUsedAt: USED }]); + const deps = productionDeps(); + const frozen = await deps.listCandidates(); + await backfillListenerCheckIn(deps, { apply: true, frozen }); + const second = await backfillListenerCheckIn(deps, { apply: true, frozen }); + expect(second).toMatchObject({ applied: 0, skippedAlreadyStamped: 1 }); + expect(await deps.listCandidates()).toHaveLength(0); + }); +}); diff --git a/backend/__tests__/unit/scripts/backfill-listener-check-in.test.js b/backend/__tests__/unit/scripts/backfill-listener-check-in.test.js new file mode 100644 index 0000000000..1e9b602406 --- /dev/null +++ b/backend/__tests__/unit/scripts/backfill-listener-check-in.test.js @@ -0,0 +1,56 @@ +const { backfillListenerCheckIn } = require('../../../scripts/backfill-listener-check-in'); + +const candidate = (store, tokenId) => ({ + store, ownerId: `${store}-owner`, tokenId, lastUsedAt: '2026-10-01T00:00:00.000Z', +}); + +const makeDeps = (candidates, stamp = async () => 1) => ({ + listCandidates: jest.fn(async () => candidates), + stamp: jest.fn(stamp), +}); + +describe('backfill-listener-check-in', () => { + it('reports the candidates by store and writes nothing in the default dry run', async () => { + const deps = makeDeps([candidate('user', 'a'), candidate('user', 'b'), candidate('installation', 'c')]); + const report = await backfillListenerCheckIn(deps); + expect(report).toMatchObject({ + dryRun: true, candidates: 3, byStore: { user: 2, installation: 1 }, frozen: null, applied: 0, + }); + expect(deps.stamp).not.toHaveBeenCalled(); + }); + + it('refuses to apply without the frozen set from a dry run', async () => { + await expect(backfillListenerCheckIn(makeDeps([candidate('user', 'a')]), { apply: true })) + .rejects.toThrow(/frozen id set/); + }); + + it('touches only the frozen set, even when more tokens became candidates since', async () => { + const frozen = [candidate('user', 'a')]; + const deps = makeDeps([candidate('user', 'a'), candidate('user', 'new-since')]); + const report = await backfillListenerCheckIn(deps, { apply: true, frozen }); + expect(deps.stamp).toHaveBeenCalledTimes(1); + expect(deps.stamp).toHaveBeenCalledWith(frozen[0]); + expect(report).toMatchObject({ + dryRun: false, frozen: 1, applied: 1, skippedAlreadyStamped: 0, + }); + }); + + it('counts a token a listener already stamped as skipped, not applied', async () => { + const frozen = [candidate('user', 'a'), candidate('installation', 'b')]; + const deps = makeDeps(frozen, async (c) => (c.tokenId === 'a' ? 0 : 1)); + const report = await backfillListenerCheckIn(deps, { apply: true, frozen }); + expect(report).toMatchObject({ applied: 1, skippedAlreadyStamped: 1, failed: 0 }); + }); + + it('keeps going and reports a failed write', async () => { + const frozen = [candidate('user', 'a'), candidate('user', 'b')]; + const errSpy = jest.spyOn(console, 'error').mockImplementation(() => {}); + const deps = makeDeps(frozen, async (c) => { + if (c.tokenId === 'a') throw new Error('boom'); + return 1; + }); + const report = await backfillListenerCheckIn(deps, { apply: true, frozen }); + expect(report).toMatchObject({ applied: 1, failed: 1 }); + errSpy.mockRestore(); + }); +}); diff --git a/backend/models/AgentRegistry.ts b/backend/models/AgentRegistry.ts index fd0b926558..d022726b6a 100644 --- a/backend/models/AgentRegistry.ts +++ b/backend/models/AgentRegistry.ts @@ -226,6 +226,8 @@ export interface IAgentInstallationRegistry extends Document { label?: string; createdAt: Date; lastUsedAt?: Date; + /** Last listener check-in (events poll or socket connect); see agentListenerCheckIn. */ + lastListenedAt?: Date; }>; // Server-granted because GitHub issue writes use the instance credential. // Never map this to the client-editable `scopes` array. @@ -277,6 +279,7 @@ const AgentInstallationSchema = new Schema( label: String, createdAt: { type: Date, default: Date.now }, lastUsedAt: Date, + lastListenedAt: Date, }, ], githubIssueWrite: { type: Boolean, default: false }, diff --git a/backend/package.json b/backend/package.json index 9be5fd39e4..4de9aa0214 100644 --- a/backend/package.json +++ b/backend/package.json @@ -25,6 +25,7 @@ "bootstrap:clawd-bot": "node scripts/bootstrap-clawd-bot.js", "backfill:attention-items": "node dist/scripts/backfill-attention-items.js", "backfill:message-author-kind": "node dist/scripts/backfill-message-author-kind.js", + "backfill:listener-check-in": "node dist/scripts/backfill-listener-check-in.js", "backfill:local-messages": "node dist/scripts/backfill-local-messages-to-pg.js", "migrate:task-priority": "node dist/scripts/migrate-task-priority.js", "migrate:task-source-ref-identity": "node dist/scripts/migrate-task-source-ref-identity.js", diff --git a/backend/scripts/backfill-listener-check-in.ts b/backend/scripts/backfill-listener-check-in.ts new file mode 100644 index 0000000000..13f3a495c8 --- /dev/null +++ b/backend/scripts/backfill-listener-check-in.ts @@ -0,0 +1,193 @@ +/** + * Give every token that has been used a `lastListenedAt` equal to its + * `lastUsedAt` (TASK-721, ahead of the listener-based agent state). + * + * Why. Agent state is about to read listener check-ins instead of any token + * use. Tokens that existed before `lastListenedAt` shipped have no value, so + * without this every long-gone seat would flip from gone-dark to never-connected + * on the deploy and the stalled-connect nudge could post into old rooms. This + * preserves the verdict the derivation gave before the change. It over-claims + * for a token that was only ever used by MCP tool calls; a live listener's next + * poll stamps the real value, and an MCP-only token stays as it was judged. + * + * Safety. + * - Insert-only against live writers. Each write is guarded by + * `lastListenedAt: { $exists: false }` on the one token element, so a stamp a + * running listener wrote in the meantime is never overwritten. + * - Per token, by its subdocument `_id`: no array is replaced, so a concurrent + * `lastUsedAt` write on another element or field is not lost. + * - Token hashes are never read or printed. + * - Dry run by default. A dry run can write the candidate set to `--out`; the + * apply run refuses to start without that frozen set (`--ids`) and touches + * only it. Re-running is harmless: a stamped token is skipped. + * + * Usage: + * cd backend && node dist/scripts/backfill-listener-check-in.js --out ids.json # dry run + * cd backend && node dist/scripts/backfill-listener-check-in.js --apply --ids ids.json + */ +/* eslint-disable no-console */ + +import fs from 'fs'; +import mongoose from 'mongoose'; + +// eslint-disable-next-line @typescript-eslint/no-require-imports +const User = require('../models/User'); +// eslint-disable-next-line @typescript-eslint/no-require-imports +const { AgentInstallation } = require('../models/AgentRegistry'); + +export type TokenStore = 'user' | 'installation'; + +export interface TokenCandidate { + store: TokenStore; + ownerId: string; + tokenId: string; + lastUsedAt: string; +} + +export interface ListenerBackfillDeps { + listCandidates: () => Promise; + /** Returns how many rows the database changed (0 when already stamped). */ + stamp: (candidate: TokenCandidate) => Promise; +} + +export interface ListenerBackfillReport { + dryRun: boolean; + candidates: number; + byStore: Record; + frozen: number | null; + applied: number; + skippedAlreadyStamped: number; + failed: number; +} + +export async function backfillListenerCheckIn( + deps: ListenerBackfillDeps, + options: { apply?: boolean; frozen?: TokenCandidate[] } = {}, +): Promise { + const apply = Boolean(options.apply); + if (apply && !options.frozen) { + throw new Error('apply needs the frozen id set from a dry run (--ids)'); + } + const current = await deps.listCandidates(); + const byStore: Record = { user: 0, installation: 0 }; + current.forEach((candidate) => { byStore[candidate.store] += 1; }); + const report: ListenerBackfillReport = { + dryRun: !apply, + candidates: current.length, + byStore, + frozen: options.frozen ? options.frozen.length : null, + applied: 0, + skippedAlreadyStamped: 0, + failed: 0, + }; + if (!apply) return report; + + for (const candidate of options.frozen as TokenCandidate[]) { + try { + // eslint-disable-next-line no-await-in-loop + const changed = await deps.stamp(candidate); + if (changed > 0) report.applied += changed; + else report.skippedAlreadyStamped += 1; + } catch (error) { + report.failed += 1; + const who = `${candidate.store}:${candidate.ownerId}:${candidate.tokenId}`; + console.error(`stamp failed for ${who}:`, (error as Error).message); + } + } + return report; +} + +const toIso = (value: unknown): string => new Date(value as string | number | Date).toISOString(); + +export const productionDeps = (): ListenerBackfillDeps => ({ + listCandidates: async () => { + const out: TokenCandidate[] = []; + // Only the ids and the two timestamps are selected: no token hash leaves the database. + const users = await User.find({ 'agentRuntimeTokens.lastUsedAt': { $exists: true } }) + .select('agentRuntimeTokens._id agentRuntimeTokens.lastUsedAt agentRuntimeTokens.lastListenedAt') + .lean(); + for (const user of users as Array>) { + for (const token of user.agentRuntimeTokens || []) { + if (token.lastUsedAt && !token.lastListenedAt) { + out.push({ + store: 'user', + ownerId: String(user._id), + tokenId: String(token._id), + lastUsedAt: toIso(token.lastUsedAt), + }); + } + } + } + const installs = await AgentInstallation.find({ 'runtimeTokens.lastUsedAt': { $exists: true } }) + .select('runtimeTokens._id runtimeTokens.lastUsedAt runtimeTokens.lastListenedAt') + .lean(); + for (const install of installs as Array>) { + for (const token of install.runtimeTokens || []) { + if (token.lastUsedAt && !token.lastListenedAt) { + out.push({ + store: 'installation', + ownerId: String(install._id), + tokenId: String(token._id), + lastUsedAt: toIso(token.lastUsedAt), + }); + } + } + } + return out; + }, + stamp: async (candidate) => { + const field = candidate.store === 'user' ? 'agentRuntimeTokens' : 'runtimeTokens'; + const Model = candidate.store === 'user' ? User : AgentInstallation; + const res = await Model.updateOne( + { + _id: candidate.ownerId, + [field]: { + $elemMatch: { + _id: new mongoose.Types.ObjectId(candidate.tokenId), + lastListenedAt: { $exists: false }, + }, + }, + }, + { $set: { [`${field}.$.lastListenedAt`]: new Date(candidate.lastUsedAt) } }, + ); + return res?.modifiedCount ?? res?.nModified ?? 0; + }, +}); + +const argValue = (flag: string): string | undefined => { + const at = process.argv.indexOf(flag); + return at >= 0 ? process.argv[at + 1] : undefined; +}; + +export async function main(): Promise { + const apply = process.argv.includes('--apply'); + const idsPath = argValue('--ids'); + const outPath = argValue('--out'); + if (!process.env.MONGO_URI) throw new Error('MONGO_URI is required'); + if (apply && !idsPath) throw new Error('--apply needs --ids '); + const frozen = idsPath + ? (JSON.parse(fs.readFileSync(idsPath, 'utf8')) as { candidates: TokenCandidate[] }).candidates + : undefined; + + await mongoose.connect(process.env.MONGO_URI); + try { + const deps = productionDeps(); + if (!apply && outPath) { + const candidates = await deps.listCandidates(); + fs.writeFileSync(outPath, JSON.stringify({ createdAt: new Date().toISOString(), candidates }, null, 2)); + console.log(`[backfill-listener-check-in] wrote ${candidates.length} candidates to ${outPath}`); + } + const report = await backfillListenerCheckIn(deps, { apply, frozen }); + console.log(`[backfill-listener-check-in] ${JSON.stringify(report)}`); + if (!apply) console.log('DRY RUN — nothing written. Re-run with --apply --ids .'); + } finally { + await mongoose.disconnect(); + } +} + +if (require.main === module) { + main().catch((error) => { + console.error('listener check-in backfill failed:', error); + process.exit(1); + }); +} From 781429aa5e84ed39f48f884446ac4f495470543a Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:03:37 -0700 Subject: [PATCH 2/9] docs(runbooks): how to run the listener check-in backfill (TASK-721) Co-Authored-By: Claude Sonnet 5.5 Co-authored-by: Keel Agent-Model: claude/claude-opus-5-5/high --- docs/runbooks/listener-check-in-backfill.md | 43 +++++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 docs/runbooks/listener-check-in-backfill.md diff --git a/docs/runbooks/listener-check-in-backfill.md b/docs/runbooks/listener-check-in-backfill.md new file mode 100644 index 0000000000..164b18ed09 --- /dev/null +++ b/docs/runbooks/listener-check-in-backfill.md @@ -0,0 +1,43 @@ +# Listener check-in backfill + +`backend/scripts/backfill-listener-check-in.ts` gives every runtime token that has +been used a `lastListenedAt` equal to its `lastUsedAt`. It runs on prod BEFORE +agent state starts reading listener check-ins (TASK-721). Without it, every +seat whose last call was before `lastListenedAt` existed would flip from +gone-dark to never-connected on the deploy, and the stalled-connect nudge could +post into old rooms. + +It covers both token stores: `User.agentRuntimeTokens` and +`AgentInstallation.runtimeTokens`. + +## Run it + +```bash +cd backend +# 1. Dry run. Prints the counts and freezes the candidate set (no token hashes). +node dist/scripts/backfill-listener-check-in.js --out ids.json +# 2. Apply only that set. +node dist/scripts/backfill-listener-check-in.js --apply --ids ids.json +``` + +`--apply` refuses to start without `--ids`. Re-running is harmless: a token that +already has `lastListenedAt` is counted as skipped. + +## What it will not do + +- Overwrite a stamp a running listener wrote after the dry run. Each write is + guarded by `lastListenedAt: { $exists: false }` on that one token element. +- Replace an array or touch another field, so a concurrent `lastUsedAt` write is kept. +- Read, print or store a token hash. + +## What it over-claims + +A token that was only ever used for MCP tool calls gets a `lastListenedAt` too. +That is deliberate: it keeps the verdict agent state gave before the change. A +live listener's next poll writes the real value. + +## Check it worked + +The dry run's candidate count after the apply should be 0. Then confirm the +state derivation change is safe to deploy: no old seat should read +never-connected that read gone-dark before. From bbdf790b04c7209aefab6cfa45700a9c769d8f9d Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:08:44 -0700 Subject: [PATCH 3/9] fix(agents): the listener backfill reports tokens it cannot address and also stamps explicit nulls (TASK-721) The dry run now counts used tokens whose subdocument has no ObjectId _id instead of listing them as ids that apply would fail on. The write guard is lastListenedAt: null, which matches an absent field and an explicit null but never a live Date, so it agrees with the listing. An unacknowledged write is a failure, not "already stamped". The runbook says what to check before the derivation switch ships. Co-Authored-By: Claude Sonnet 5.5 Co-authored-by: Keel Agent-Model: claude/claude-opus-5-5/high --- .../backfill-listener-check-in.db.test.js | 34 ++++++++-- .../backfill-listener-check-in.test.js | 9 ++- backend/scripts/backfill-listener-check-in.ts | 62 +++++++++++-------- docs/runbooks/listener-check-in-backfill.md | 17 +++-- 4 files changed, 86 insertions(+), 36 deletions(-) diff --git a/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js b/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js index 505cae7e16..054a256aaa 100644 --- a/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js +++ b/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js @@ -54,7 +54,7 @@ describe('backfill-listener-check-in against the real schemas', () => { ]); const deps = productionDeps(); - const listed = await deps.listCandidates(); + const { candidates: listed } = await deps.listCandidates(); expect(listed).toHaveLength(2); expect(JSON.stringify(listed)).not.toContain('secret'); @@ -72,7 +72,7 @@ describe('backfill-listener-check-in against the real schemas', () => { it('never overwrites a stamp a running listener wrote after the dry run', async () => { const user = await makeAgentUser([{ tokenHash: 'secret-1', lastUsedAt: USED }]); const deps = productionDeps(); - const frozen = await deps.listCandidates(); + const { candidates: frozen } = await deps.listCandidates(); await User.updateOne( { _id: user._id }, { $set: { 'agentRuntimeTokens.0.lastListenedAt': LISTENED } }, @@ -86,10 +86,36 @@ describe('backfill-listener-check-in against the real schemas', () => { it('is idempotent: a second apply of the same set changes nothing', async () => { await makeAgentUser([{ tokenHash: 'secret-1', lastUsedAt: USED }]); const deps = productionDeps(); - const frozen = await deps.listCandidates(); + const { candidates: frozen } = await deps.listCandidates(); await backfillListenerCheckIn(deps, { apply: true, frozen }); const second = await backfillListenerCheckIn(deps, { apply: true, frozen }); expect(second).toMatchObject({ applied: 0, skippedAlreadyStamped: 1 }); - expect(await deps.listCandidates()).toHaveLength(0); + expect((await deps.listCandidates()).candidates).toHaveLength(0); + }); + + it('stamps a token whose lastListenedAt is an explicit null', async () => { + const inst = await makeInstall([ + { + _id: new mongoose.Types.ObjectId(), tokenHash: 'secret-n', lastUsedAt: USED, lastListenedAt: null, + }, + ]); + const deps = productionDeps(); + const { candidates } = await deps.listCandidates(); + expect(candidates).toHaveLength(1); + const report = await backfillListenerCheckIn(deps, { apply: true, frozen: candidates }); + expect(report).toMatchObject({ applied: 1, skippedAlreadyStamped: 0 }); + const after = await AgentInstallation.collection.findOne({ _id: inst.insertedId }); + expect(after.runtimeTokens[0].lastListenedAt).toEqual(USED); + }); + + it('counts a token with no ObjectId _id as unaddressable instead of listing it', async () => { + await makeInstall([ + { _id: 'not-an-object-id', tokenHash: 'secret-s', lastUsedAt: USED }, + { tokenHash: 'secret-m', lastUsedAt: USED }, + ]); + const listing = await productionDeps().listCandidates(); + expect(listing.candidates).toHaveLength(0); + expect(listing.unaddressable).toBe(1 + 1); + expect(JSON.stringify(listing)).not.toContain('secret'); }); }); diff --git a/backend/__tests__/unit/scripts/backfill-listener-check-in.test.js b/backend/__tests__/unit/scripts/backfill-listener-check-in.test.js index 1e9b602406..04b30286d7 100644 --- a/backend/__tests__/unit/scripts/backfill-listener-check-in.test.js +++ b/backend/__tests__/unit/scripts/backfill-listener-check-in.test.js @@ -4,8 +4,8 @@ const candidate = (store, tokenId) => ({ store, ownerId: `${store}-owner`, tokenId, lastUsedAt: '2026-10-01T00:00:00.000Z', }); -const makeDeps = (candidates, stamp = async () => 1) => ({ - listCandidates: jest.fn(async () => candidates), +const makeDeps = (candidates, stamp = async () => 1, unaddressable = 0) => ({ + listCandidates: jest.fn(async () => ({ candidates, unaddressable })), stamp: jest.fn(stamp), }); @@ -19,6 +19,11 @@ describe('backfill-listener-check-in', () => { expect(deps.stamp).not.toHaveBeenCalled(); }); + it('reports tokens it cannot address so the operator sees them before the switch', async () => { + const report = await backfillListenerCheckIn(makeDeps([candidate('user', 'a')], undefined, 2)); + expect(report).toMatchObject({ candidates: 1, unaddressable: 2 }); + }); + it('refuses to apply without the frozen set from a dry run', async () => { await expect(backfillListenerCheckIn(makeDeps([candidate('user', 'a')]), { apply: true })) .rejects.toThrow(/frozen id set/); diff --git a/backend/scripts/backfill-listener-check-in.ts b/backend/scripts/backfill-listener-check-in.ts index 13f3a495c8..2ecf441741 100644 --- a/backend/scripts/backfill-listener-check-in.ts +++ b/backend/scripts/backfill-listener-check-in.ts @@ -44,8 +44,18 @@ export interface TokenCandidate { lastUsedAt: string; } +export interface CandidateListing { + candidates: TokenCandidate[]; + /** + * Used, unstamped tokens whose subdocument has no ObjectId `_id`. They cannot + * be addressed by this script, so the operator must see them before the + * derivation switch ships; they would flip from gone-dark to never-connected. + */ + unaddressable: number; +} + export interface ListenerBackfillDeps { - listCandidates: () => Promise; + listCandidates: () => Promise; /** Returns how many rows the database changed (0 when already stamped). */ stamp: (candidate: TokenCandidate) => Promise; } @@ -53,6 +63,7 @@ export interface ListenerBackfillDeps { export interface ListenerBackfillReport { dryRun: boolean; candidates: number; + unaddressable: number; byStore: Record; frozen: number | null; applied: number; @@ -68,12 +79,13 @@ export async function backfillListenerCheckIn( if (apply && !options.frozen) { throw new Error('apply needs the frozen id set from a dry run (--ids)'); } - const current = await deps.listCandidates(); + const { candidates: current, unaddressable } = await deps.listCandidates(); const byStore: Record = { user: 0, installation: 0 }; current.forEach((candidate) => { byStore[candidate.store] += 1; }); const report: ListenerBackfillReport = { dryRun: !apply, candidates: current.length, + unaddressable, byStore, frozen: options.frozen ? options.frozen.length : null, applied: 0, @@ -102,38 +114,34 @@ const toIso = (value: unknown): string => new Date(value as string | number | Da export const productionDeps = (): ListenerBackfillDeps => ({ listCandidates: async () => { const out: TokenCandidate[] = []; + let unaddressable = 0; + const take = (store: TokenStore, ownerId: unknown, token: Record) => { + if (!token.lastUsedAt || token.lastListenedAt) return; + if (!/^[0-9a-f]{24}$/i.test(String(token._id || ''))) { + unaddressable += 1; + return; + } + out.push({ + store, + ownerId: String(ownerId), + tokenId: String(token._id), + lastUsedAt: toIso(token.lastUsedAt), + }); + }; // Only the ids and the two timestamps are selected: no token hash leaves the database. const users = await User.find({ 'agentRuntimeTokens.lastUsedAt': { $exists: true } }) .select('agentRuntimeTokens._id agentRuntimeTokens.lastUsedAt agentRuntimeTokens.lastListenedAt') .lean(); for (const user of users as Array>) { - for (const token of user.agentRuntimeTokens || []) { - if (token.lastUsedAt && !token.lastListenedAt) { - out.push({ - store: 'user', - ownerId: String(user._id), - tokenId: String(token._id), - lastUsedAt: toIso(token.lastUsedAt), - }); - } - } + for (const token of user.agentRuntimeTokens || []) take('user', user._id, token); } const installs = await AgentInstallation.find({ 'runtimeTokens.lastUsedAt': { $exists: true } }) .select('runtimeTokens._id runtimeTokens.lastUsedAt runtimeTokens.lastListenedAt') .lean(); for (const install of installs as Array>) { - for (const token of install.runtimeTokens || []) { - if (token.lastUsedAt && !token.lastListenedAt) { - out.push({ - store: 'installation', - ownerId: String(install._id), - tokenId: String(token._id), - lastUsedAt: toIso(token.lastUsedAt), - }); - } - } + for (const token of install.runtimeTokens || []) take('installation', install._id, token); } - return out; + return { candidates: out, unaddressable }; }, stamp: async (candidate) => { const field = candidate.store === 'user' ? 'agentRuntimeTokens' : 'runtimeTokens'; @@ -144,12 +152,16 @@ export const productionDeps = (): ListenerBackfillDeps => ({ [field]: { $elemMatch: { _id: new mongoose.Types.ObjectId(candidate.tokenId), - lastListenedAt: { $exists: false }, + // null matches an absent field and an explicit null, never a live Date. + lastListenedAt: null, }, }, }, { $set: { [`${field}.$.lastListenedAt`]: new Date(candidate.lastUsedAt) } }, ); + // A write Mongoose dropped (a field missing from the schema) comes back + // unacknowledged: report it as a failure, never as "already stamped". + if (res?.acknowledged === false) throw new Error('write was not acknowledged'); return res?.modifiedCount ?? res?.nModified ?? 0; }, }); @@ -173,7 +185,7 @@ export async function main(): Promise { try { const deps = productionDeps(); if (!apply && outPath) { - const candidates = await deps.listCandidates(); + const { candidates } = await deps.listCandidates(); fs.writeFileSync(outPath, JSON.stringify({ createdAt: new Date().toISOString(), candidates }, null, 2)); console.log(`[backfill-listener-check-in] wrote ${candidates.length} candidates to ${outPath}`); } diff --git a/docs/runbooks/listener-check-in-backfill.md b/docs/runbooks/listener-check-in-backfill.md index 164b18ed09..3c2008a27f 100644 --- a/docs/runbooks/listener-check-in-backfill.md +++ b/docs/runbooks/listener-check-in-backfill.md @@ -36,8 +36,15 @@ A token that was only ever used for MCP tool calls gets a `lastListenedAt` too. That is deliberate: it keeps the verdict agent state gave before the change. A live listener's next poll writes the real value. -## Check it worked - -The dry run's candidate count after the apply should be 0. Then confirm the -state derivation change is safe to deploy: no old seat should read -never-connected that read gone-dark before. +## Before the derivation switch ships + +- The listener check-in writer (#3114) must already be live on prod. Otherwise a + live listener keeps only its backfilled stamp and ages into gone-dark. +- Run the dry run once more after the apply. Some candidates will remain on a + live system: tokens first used by an MCP call after the freeze, and any + **unaddressable** ones. The report's `unaddressable` count is tokens whose + 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. +- Installation-only tokens (no copy on the agent's User row) keep their backfilled + value forever until the switch PR adds a live stamp for them. From dc4b5508f4a00ccf8dfd3a9175190cef6b0e66ba Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:10:20 -0700 Subject: [PATCH 4/9] fix(agents): the listener backfill treats a string _id as unaddressable and lists only null or absent stamps (TASK-721) Check the id's type, not its shape, so a 24-hex string id is counted as unaddressable instead of being listed and silently skipped. List with `!= null` so the listing and the write guard agree for every stored value. The runbook and the script header name the current guard. Co-Authored-By: Claude Sonnet 5.5 Co-authored-by: Keel Agent-Model: claude/claude-opus-5-5/high --- .../unit/scripts/backfill-listener-check-in.db.test.js | 4 +++- backend/scripts/backfill-listener-check-in.ts | 6 +++--- docs/runbooks/listener-check-in-backfill.md | 2 +- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js b/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js index 054a256aaa..2c892c5424 100644 --- a/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js +++ b/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js @@ -111,11 +111,13 @@ describe('backfill-listener-check-in against the real schemas', () => { it('counts a token with no ObjectId _id as unaddressable instead of listing it', async () => { await makeInstall([ { _id: 'not-an-object-id', tokenHash: 'secret-s', lastUsedAt: USED }, + // A 24-hex id stored as a string is still not an ObjectId and would never match. + { _id: '0123456789abcdef01234567', tokenHash: 'secret-h', lastUsedAt: USED }, { tokenHash: 'secret-m', lastUsedAt: USED }, ]); const listing = await productionDeps().listCandidates(); expect(listing.candidates).toHaveLength(0); - expect(listing.unaddressable).toBe(1 + 1); + expect(listing.unaddressable).toBe(3); expect(JSON.stringify(listing)).not.toContain('secret'); }); }); diff --git a/backend/scripts/backfill-listener-check-in.ts b/backend/scripts/backfill-listener-check-in.ts index 2ecf441741..2e5f40f09f 100644 --- a/backend/scripts/backfill-listener-check-in.ts +++ b/backend/scripts/backfill-listener-check-in.ts @@ -12,7 +12,7 @@ * * Safety. * - Insert-only against live writers. Each write is guarded by - * `lastListenedAt: { $exists: false }` on the one token element, so a stamp a + * `lastListenedAt: null` on the one token element, so a stamp a * running listener wrote in the meantime is never overwritten. * - Per token, by its subdocument `_id`: no array is replaced, so a concurrent * `lastUsedAt` write on another element or field is not lost. @@ -116,8 +116,8 @@ export const productionDeps = (): ListenerBackfillDeps => ({ const out: TokenCandidate[] = []; let unaddressable = 0; const take = (store: TokenStore, ownerId: unknown, token: Record) => { - if (!token.lastUsedAt || token.lastListenedAt) return; - if (!/^[0-9a-f]{24}$/i.test(String(token._id || ''))) { + if (!token.lastUsedAt || token.lastListenedAt != null) return; + if (!(token._id instanceof mongoose.Types.ObjectId)) { unaddressable += 1; return; } diff --git a/docs/runbooks/listener-check-in-backfill.md b/docs/runbooks/listener-check-in-backfill.md index 3c2008a27f..518dfb1832 100644 --- a/docs/runbooks/listener-check-in-backfill.md +++ b/docs/runbooks/listener-check-in-backfill.md @@ -26,7 +26,7 @@ already has `lastListenedAt` is counted as skipped. ## What it will not do - Overwrite a stamp a running listener wrote after the dry run. Each write is - guarded by `lastListenedAt: { $exists: false }` on that one token element. + guarded by `lastListenedAt: null` (absent or null, never a live Date) on that one token element. - Replace an array or touch another field, so a concurrent `lastUsedAt` write is kept. - Read, print or store a token hash. From 4d1c1e67275d25ac18446c3b9899ee6a0a7b00aa Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:12:27 -0700 Subject: [PATCH 5/9] fix(agents): the listener backfill exits nonzero after a partial apply, and its file handoff is tested (TASK-721) Move the --out write, the --ids read and the report into runBackfill so they can be tested without a second connection. A failed stamp now fails the run after the whole set has been tried, and the message says to re-run the same ids file. The runbook names the installation-token live stamp as a deploy gate for the derivation change. Co-Authored-By: Claude Sonnet 5.5 Co-authored-by: Keel Agent-Model: claude/claude-opus-5-5/high --- .../backfill-listener-check-in.db.test.js | 61 ++++++++++++++++++- backend/scripts/backfill-listener-check-in.ts | 51 +++++++++++----- docs/runbooks/listener-check-in-backfill.md | 9 ++- 3 files changed, 104 insertions(+), 17 deletions(-) diff --git a/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js b/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js index 2c892c5424..66120b837c 100644 --- a/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js +++ b/backend/__tests__/unit/scripts/backfill-listener-check-in.db.test.js @@ -2,6 +2,9 @@ // sibling suite stubs the deps, so it cannot see a field the schema strips or a // filter that does not match. /* eslint-disable global-require */ +const fs = require('fs'); +const os = require('os'); +const path = require('path'); const { MongoMemoryServer } = require('mongodb-memory-server'); const mongoose = require('mongoose'); @@ -10,6 +13,7 @@ let User; let AgentInstallation; let backfillListenerCheckIn; let productionDeps; +let runBackfill; const USED = new Date('2026-10-01T00:00:00Z'); const LISTENED = new Date('2026-10-30T00:00:00Z'); @@ -19,7 +23,7 @@ beforeAll(async () => { await mongoose.connect(mongod.getUri()); User = require('../../../models/User'); ({ AgentInstallation } = require('../../../models/AgentRegistry')); - ({ backfillListenerCheckIn, productionDeps } = require('../../../scripts/backfill-listener-check-in')); + ({ backfillListenerCheckIn, productionDeps, runBackfill } = require('../../../scripts/backfill-listener-check-in')); }); afterEach(async () => { await User.deleteMany({}); @@ -120,4 +124,59 @@ describe('backfill-listener-check-in against the real schemas', () => { expect(listing.unaddressable).toBe(3); expect(JSON.stringify(listing)).not.toContain('secret'); }); + + describe('runBackfill: the dry-run file handoff', () => { + let dir; + beforeEach(() => { dir = fs.mkdtempSync(path.join(os.tmpdir(), 'listener-backfill-')); }); + afterEach(() => { fs.rmSync(dir, { recursive: true, force: true }); }); + const quiet = () => jest.spyOn(console, 'log').mockImplementation(() => {}); + + it('freezes the set with --out and applies only that set with --ids', async () => { + const log = quiet(); + const frozenUser = await makeAgentUser([{ tokenHash: 'secret-a', lastUsedAt: USED }]); + const outPath = path.join(dir, 'ids.json'); + const dry = await runBackfill(productionDeps(), { apply: false, outPath }); + expect(dry).toMatchObject({ dryRun: true, candidates: 1, applied: 0 }); + const file = JSON.parse(fs.readFileSync(outPath, 'utf8')); + expect(file.candidates).toHaveLength(1); + expect(fs.readFileSync(outPath, 'utf8')).not.toContain('secret'); + expect((await User.findById(frozenUser._id).lean()).agentRuntimeTokens[0].lastListenedAt).toBeUndefined(); + + // A token that becomes a candidate after the freeze is not part of it. + const later = await makeAgentUser([{ tokenHash: 'secret-b', lastUsedAt: USED }]); + const applied = await runBackfill(productionDeps(), { apply: true, idsPath: outPath }); + expect(applied).toMatchObject({ dryRun: false, frozen: 1, applied: 1 }); + expect((await User.findById(frozenUser._id).lean()).agentRuntimeTokens[0].lastListenedAt).toEqual(USED); + expect((await User.findById(later._id).lean()).agentRuntimeTokens[0].lastListenedAt).toBeUndefined(); + log.mockRestore(); + }); + + it('refuses --apply without --ids and writes nothing', async () => { + const log = quiet(); + const user = await makeAgentUser([{ tokenHash: 'secret-a', lastUsedAt: USED }]); + await expect(runBackfill(productionDeps(), { apply: true })).rejects.toThrow(/--ids/); + expect((await User.findById(user._id).lean()).agentRuntimeTokens[0].lastListenedAt).toBeUndefined(); + log.mockRestore(); + }); + + it('fails the run after trying the whole set when a stamp fails', async () => { + const log = quiet(); + const err = jest.spyOn(console, 'error').mockImplementation(() => {}); + await makeAgentUser([{ tokenHash: 'secret-a', lastUsedAt: USED }]); + const idsPath = path.join(dir, 'ids.json'); + fs.writeFileSync(idsPath, JSON.stringify({ + candidates: [ + { + store: 'user', ownerId: 'not-an-id', tokenId: 'nope', lastUsedAt: USED.toISOString(), + }, + ...(await productionDeps().listCandidates()).candidates, + ], + })); + await expect(runBackfill(productionDeps(), { apply: true, idsPath })).rejects.toThrow(/1 token\(s\) failed/); + // The good token after the bad one was still stamped. + expect((await productionDeps().listCandidates()).candidates).toHaveLength(0); + log.mockRestore(); + err.mockRestore(); + }); + }); }); diff --git a/backend/scripts/backfill-listener-check-in.ts b/backend/scripts/backfill-listener-check-in.ts index 2e5f40f09f..251ca0fefd 100644 --- a/backend/scripts/backfill-listener-check-in.ts +++ b/backend/scripts/backfill-listener-check-in.ts @@ -171,27 +171,50 @@ const argValue = (flag: string): string | undefined => { return at >= 0 ? process.argv[at + 1] : undefined; }; -export async function main(): Promise { - const apply = process.argv.includes('--apply'); - const idsPath = argValue('--ids'); - const outPath = argValue('--out'); - if (!process.env.MONGO_URI) throw new Error('MONGO_URI is required'); +export interface RunOptions { + apply: boolean; + idsPath?: string; + outPath?: string; +} + +/** + * Everything `main` does once a connection exists: the `--out` write of a dry + * run's frozen set, the `--ids` read for an apply, the report. A failed stamp + * makes the run fail, after the whole set has been tried, so a scripted caller + * cannot read a partial apply as success; the per-token errors are already + * printed for the retry. + */ +export async function runBackfill(deps: ListenerBackfillDeps, options: RunOptions): Promise { + const { apply, idsPath, outPath } = options; if (apply && !idsPath) throw new Error('--apply needs --ids '); const frozen = idsPath ? (JSON.parse(fs.readFileSync(idsPath, 'utf8')) as { candidates: TokenCandidate[] }).candidates : undefined; + if (!apply && outPath) { + const { candidates } = await deps.listCandidates(); + fs.writeFileSync(outPath, JSON.stringify({ createdAt: new Date().toISOString(), candidates }, null, 2)); + console.log(`[backfill-listener-check-in] wrote ${candidates.length} candidates to ${outPath}`); + } + const report = await backfillListenerCheckIn(deps, { apply, frozen }); + console.log(`[backfill-listener-check-in] ${JSON.stringify(report)}`); + if (!apply) console.log('DRY RUN — nothing written. Re-run with --apply --ids .'); + if (report.failed > 0) { + throw new Error(`${report.failed} token(s) failed to stamp; re-run the same --ids file to retry them`); + } + return report; +} +export async function main(): Promise { + if (!process.env.MONGO_URI) throw new Error('MONGO_URI is required'); + const options: RunOptions = { + apply: process.argv.includes('--apply'), + idsPath: argValue('--ids'), + outPath: argValue('--out'), + }; + if (options.apply && !options.idsPath) throw new Error('--apply needs --ids '); await mongoose.connect(process.env.MONGO_URI); try { - const deps = productionDeps(); - if (!apply && outPath) { - const { candidates } = await deps.listCandidates(); - fs.writeFileSync(outPath, JSON.stringify({ createdAt: new Date().toISOString(), candidates }, null, 2)); - console.log(`[backfill-listener-check-in] wrote ${candidates.length} candidates to ${outPath}`); - } - const report = await backfillListenerCheckIn(deps, { apply, frozen }); - console.log(`[backfill-listener-check-in] ${JSON.stringify(report)}`); - if (!apply) console.log('DRY RUN — nothing written. Re-run with --apply --ids .'); + await runBackfill(productionDeps(), options); } finally { await mongoose.disconnect(); } diff --git a/docs/runbooks/listener-check-in-backfill.md b/docs/runbooks/listener-check-in-backfill.md index 518dfb1832..8ca95c80f8 100644 --- a/docs/runbooks/listener-check-in-backfill.md +++ b/docs/runbooks/listener-check-in-backfill.md @@ -46,5 +46,10 @@ 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. -- Installation-only tokens (no copy on the agent's User row) keep their backfilled - value forever until the switch PR adds a live stamp for them. +- **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. From 98dfff2106de7fb0739b62e1ae2d947b35b61fb5 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:27:59 -0700 Subject: [PATCH 6/9] fix(agents): agent state reads listener check-ins, so an MCP call alone is never-connected (TASK-721) deriveAgentState and its callers now read lastListenedAt, not lastUsedAt. A seat that has only made MCP calls reads never-connected and the stalled-connect scan nudges it. The check-in writer stamps the installation copy of a token as well, and the WebSocket connect stamps an installation-only token, so a token that lives only on AgentInstallation no longer reads never-connected. Deploy order: check-in writer (#3114), the backfill (#3120), then this change. Co-Authored-By: Claude Sonnet 5.5 Co-authored-by: Keel Agent-Model: claude/claude-opus-5-5/high --- .../services/agentWebSocketService.test.js | 8 ++- .../routes/registry.byo-host-stamp.test.js | 2 +- .../services/agentListenerCheckIn.test.js | 31 ++++++++++- .../unit/services/agentStateService.test.js | 55 ++++++++++++++++--- .../neverConnectedMentionService.test.js | 2 +- .../services/stalledConnectService.test.js | 34 ++++++++++-- backend/routes/pods.ts | 4 +- backend/routes/registry/install.ts | 6 +- backend/services/agentListenerCheckIn.ts | 37 ++++++++----- backend/services/agentStateService.ts | 29 ++++++---- backend/services/agentWebSocketService.ts | 11 +++- .../services/neverConnectedMentionService.ts | 2 +- backend/services/stalledConnectService.ts | 6 +- docs/runbooks/listener-check-in-backfill.md | 19 ++++--- 14 files changed, 186 insertions(+), 60 deletions(-) diff --git a/backend/__tests__/services/agentWebSocketService.test.js b/backend/__tests__/services/agentWebSocketService.test.js index 6d39c5ff77..1ab5dc5572 100644 --- a/backend/__tests__/services/agentWebSocketService.test.js +++ b/backend/__tests__/services/agentWebSocketService.test.js @@ -147,7 +147,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', 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/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 921610ca98..3be6bd94a8 100644 --- a/backend/__tests__/unit/services/stalledConnectService.test.js +++ b/backend/__tests__/unit/services/stalledConnectService.test.js @@ -230,6 +230,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) }], @@ -237,8 +249,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 () => { @@ -359,7 +385,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 }); @@ -380,7 +406,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 }); @@ -395,7 +421,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 }); @@ -407,7 +433,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/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/agentListenerCheckIn.ts b/backend/services/agentListenerCheckIn.ts index 8342cc5ddd..1d42cc98bf 100644 --- a/backend/services/agentListenerCheckIn.ts +++ b/backend/services/agentListenerCheckIn.ts @@ -1,4 +1,5 @@ import User from '../models/User'; +import { AgentInstallation } from '../models/AgentRegistry'; /** * A listener check-in is the one signal that "mentions will wake this agent": @@ -12,33 +13,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..89c74aaa86 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 and the + * WebSocket connect) 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..6d6964d3e7 100644 --- a/backend/services/agentWebSocketService.ts +++ b/backend/services/agentWebSocketService.ts @@ -82,7 +82,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 => ( @@ -449,9 +449,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/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 597fb30b70..79951eb284 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 || []; }; @@ -377,7 +377,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/runbooks/listener-check-in-backfill.md b/docs/runbooks/listener-check-in-backfill.md index 8ca95c80f8..3af8c409bc 100644 --- a/docs/runbooks/listener-check-in-backfill.md +++ b/docs/runbooks/listener-check-in-backfill.md @@ -46,10 +46,15 @@ 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. From c95d7cdd6d544016e109f81f677859d214bd4fc8 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:39:29 -0700 Subject: [PATCH 7/9] fix(agents): a delivery renew counts as a listener check-in, so a seat inside a long turn stays listening (TASK-721) The runner polls /events only between turns. During a turn the only seat traffic is POST /events/renew, which stamped lastUsedAt alone, so a turn past five minutes would have aged a working seat into gone-dark once state read lastListenedAt. Renew now stamps the check-in. Also asserts the stalled-connect projection in its test mock and corrects two comments and the status doc. Co-Authored-By: Claude Sonnet 5.5 Co-authored-by: Keel Agent-Model: claude/claude-opus-5-5/high --- .../routes/agentsRuntime.dm-events.test.js | 18 ++++++++++++++++++ .../services/stalledConnectService.test.js | 7 ++++++- backend/routes/agentsRuntime.ts | 5 +++++ backend/services/agentListenerCheckIn.ts | 5 +++-- backend/services/agentStateService.ts | 4 ++-- backend/services/kernelWorkSweepService.ts | 6 ++++-- docs/design/agent-status-honesty.md | 2 +- docs/runbooks/listener-check-in-backfill.md | 9 +++++++++ 8 files changed, 48 insertions(+), 8 deletions(-) diff --git a/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js b/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js index 1472cef81c..e8d3b75e5b 100644 --- a/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js +++ b/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js @@ -178,6 +178,24 @@ 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 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/services/stalledConnectService.test.js b/backend/__tests__/unit/services/stalledConnectService.test.js index 3be6bd94a8..a4290dfc8a 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' }) }), diff --git a/backend/routes/agentsRuntime.ts b/backend/routes/agentsRuntime.ts index 6fde8107c4..da8b8e51f3 100644 --- a/backend/routes/agentsRuntime.ts +++ b/backend/routes/agentsRuntime.ts @@ -1394,6 +1394,11 @@ 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. The MCP checkpoint tool uses /ack, not this route. + recordListenerCheckIn(req.agentUser?._id, req.agentTokenHash); try { const renewed = await Promise.all(events.map(async (entry: { eventId: string; deliveryId: string }) => ( await AgentEventService.renewDelivery(entry.eventId, agentName, instanceId, entry.deliveryId) diff --git a/backend/services/agentListenerCheckIn.ts b/backend/services/agentListenerCheckIn.ts index 1d42cc98bf..dac0a88e73 100644 --- a/backend/services/agentListenerCheckIn.ts +++ b/backend/services/agentListenerCheckIn.ts @@ -3,8 +3,9 @@ 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. diff --git a/backend/services/agentStateService.ts b/backend/services/agentStateService.ts index 89c74aaa86..499ece74bb 100644 --- a/backend/services/agentStateService.ts +++ b/backend/services/agentStateService.ts @@ -7,8 +7,8 @@ * * - BYO wrapper/MCP installs (host 'byo' or legacy local-cli, no * webhookUrl): the wrapper polls every ~5s while alive, so the last - * LISTENER check-in (`lastListenedAt`, stamped by the events poll and the - * WebSocket connect) is an honest reachability signal. null → + * 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). 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/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 3af8c409bc..f5b90f3911 100644 --- a/docs/runbooks/listener-check-in-backfill.md +++ b/docs/runbooks/listener-check-in-backfill.md @@ -58,3 +58,12 @@ live listener's next poll writes the real value. 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. From 6656346cf9689ee835d77bf90da6b463bb675eeb Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 31 Oct 2026 19:12:41 -0700 Subject: [PATCH 8/9] fix(agents): a renew counts as a check-in only when it renews, a live socket's pong counts, /agent-states has a round-trip test (TASK-721) Co-Authored-By: Claude Sonnet 5.5 Co-authored-by: Keel Agent-Model: claude/claude-opus-5-5/high --- .../agent-states-listener-check-in.test.js | 81 +++++++++++++++++++ .../services/agentWebSocketService.test.js | 23 ++++++ .../routes/agentsRuntime.dm-events.test.js | 13 +++ backend/routes/agentsRuntime.ts | 9 ++- backend/services/agentWebSocketService.ts | 8 ++ 5 files changed, 131 insertions(+), 3 deletions(-) create mode 100644 backend/__tests__/service/agent-states-listener-check-in.test.js 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 1ab5dc5572..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), @@ -360,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 e8d3b75e5b..1f88da3f7c 100644 --- a/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js +++ b/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js @@ -189,6 +189,19 @@ describe('inbox delivery renewal route', () => { }, response()); expect(recordListenerCheckIn).toHaveBeenCalledWith('bot-user-1', 'hashed-token'); }); + test('a renew that renews nothing is not a check-in', async () => { + AgentEventService.renewDelivery.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 malformed renew is not a check-in', async () => { recordListenerCheckIn.mockClear(); await getRouteHandler('/events/renew', 'post')({ diff --git a/backend/routes/agentsRuntime.ts b/backend/routes/agentsRuntime.ts index da8b8e51f3..815d2fd1d5 100644 --- a/backend/routes/agentsRuntime.ts +++ b/backend/routes/agentsRuntime.ts @@ -1397,14 +1397,17 @@ router.post('/events/renew', phase4RateLimit, agentRuntimeAuth, async (req: any, // 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. The MCP checkpoint tool uses /ack, not this route. - recordListenerCheckIn(req.agentUser?._id, req.agentTokenHash); + // as a check-in, but only when a delivery it holds actually renewed: stale or + // made-up nonces renew nothing, and must not move an MCP-only seat out of + // never-connected. 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 renewedEventIds = renewed.filter(Boolean); + if (renewedEventIds.length > 0) recordListenerCheckIn(req.agentUser?._id, req.agentTokenHash); + return res.json({ renewed: renewedEventIds }); } 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/services/agentWebSocketService.ts b/backend/services/agentWebSocketService.ts index 6d6964d3e7..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 } }; @@ -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); } }); From 3506c7ee5f04907a9db5b8ca1f0d26dbf759d221 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 31 Oct 2026 19:16:41 -0700 Subject: [PATCH 9/9] fix(agents): a renew after the turn checkpointed the event still checks in (TASK-721) Co-Authored-By: Claude Sonnet 5.5 Co-authored-by: Keel Agent-Model: claude/claude-opus-5-5/high --- .../routes/agentsRuntime.dm-events.test.js | 21 +++++++++++++++++- .../agentEventService.deliveryNonce.test.js | 20 +++++++++++++++++ backend/routes/agentsRuntime.ts | 22 ++++++++++--------- backend/services/agentEventService.ts | 16 ++++++++++++++ 4 files changed, 68 insertions(+), 11 deletions(-) diff --git a/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js b/backend/__tests__/unit/routes/agentsRuntime.dm-events.test.js index 1f88da3f7c..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') => ( @@ -189,8 +190,9 @@ describe('inbox delivery renewal route', () => { }, response()); expect(recordListenerCheckIn).toHaveBeenCalledWith('bot-user-1', 'hashed-token'); }); - test('a renew that renews nothing is not a check-in', async () => { + 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')({ @@ -202,6 +204,23 @@ describe('inbox delivery renewal route', () => { 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')({ 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/routes/agentsRuntime.ts b/backend/routes/agentsRuntime.ts index 815d2fd1d5..d8b1dce3ba 100644 --- a/backend/routes/agentsRuntime.ts +++ b/backend/routes/agentsRuntime.ts @@ -1397,17 +1397,19 @@ router.post('/events/renew', phase4RateLimit, agentRuntimeAuth, async (req: any, // 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 a delivery it holds actually renewed: stale or - // made-up nonces renew nothing, and must not move an MCP-only seat out of - // never-connected. The MCP checkpoint tool uses /ack, not this route. + // 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 - ))); - const renewedEventIds = renewed.filter(Boolean); - if (renewedEventIds.length > 0) recordListenerCheckIn(req.agentUser?._id, req.agentTokenHash); - return res.json({ renewed: renewedEventIds }); + 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/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)