From 0ed28836056c83eb8118fc48770a2e6b6bbcf114 Mon Sep 17 00:00:00 2001 From: zfy0701 <1646270+zfy0701@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:05:59 +0800 Subject: [PATCH 1/3] fix(daemon): own the session-metadata outbox per member MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #1023. On a daemon pool the session-metadata outbox is one shared table, but an `event/session-sync` frame is scoped by the agent's organization, which only a member serving that agent can resolve. Every member drained every row, and a snapshot whose organization it could not resolve was raised locally as a non-retryable SCOPE_DENIED, deferred, and retried forever on every member. The outbox now carries the same ownership model as the hook-completion outbox and the purge receipts: `ownerId` / `claimedAt` (schema v10 plus its migration step), a member is offered its own rows and unowned or lapsed rows only for the agents it serves, and it claims before every emit. A snapshot this member cannot scope is parked — the claim is released for whichever member serves the agent, the body and the failure count survive, and a short backoff keeps it out of this member's next pass instead of head-of-line blocking the rows behind it. Gaining a duty re-arms the parked snapshots of the gained agents and re-runs the drain, so a takeover replays them at once. A local single-daemon store is unchanged: it lists and settles unfenced, the claim is a no-op, and parking is refused so the existing retry/defer path still applies. Co-Authored-By: Claude Fable 5 --- packages/daemon/src/daemon.ts | 92 ++++++- packages/daemon/src/store/local-store.ts | 161 ++++++++++-- ...aemon-session-metadata-outbox-pool.test.ts | 233 ++++++++++++++++++ packages/daemon/test/local-store.test.ts | 15 +- 4 files changed, 460 insertions(+), 41 deletions(-) create mode 100644 packages/daemon/test/daemon-session-metadata-outbox-pool.test.ts diff --git a/packages/daemon/src/daemon.ts b/packages/daemon/src/daemon.ts index 71a3578d4..de279441a 100644 --- a/packages/daemon/src/daemon.ts +++ b/packages/daemon/src/daemon.ts @@ -836,6 +836,8 @@ function foreignHookDispatch(report: HookReport, daemonId?: string): boolean { const SESSION_METADATA_RETRY_MS = 5_000 const SESSION_METADATA_FAILURES_BEFORE_DEFER = 5 const SESSION_METADATA_DEFER_MS = 5 * 60_000 +// A snapshot this member cannot scope waits this long before it is offered again. +const SESSION_METADATA_PARK_MS = 60_000 /** Connectable when there is a URL and a credential: an API key, or — in-cluster — the * projected ServiceAccount token this pod presents instead of one. */ @@ -13017,6 +13019,8 @@ export class Daemon { this.catchUpMissedSchedules(result.agentsGained) // The purge-receipt drain is holder-scoped, so a receipt a prior holder left is owed by this member now. if (result.agentsGained.length) void this.drainSessionPurges() + // Same for the session-metadata outbox: a snapshot the previous holder parked is this member's to emit. + this.replayGainedSessionMetadata(result.agentsGained) } /** @@ -16625,7 +16629,8 @@ export class Daemon { input.sessionId, JSON.stringify(snapshot), input.phase !== 'plan', - this.clock.now() + this.clock.now(), + this.cfg.daemonId ) !== undefined } catch (err) { // Preserve the pre-outbox behavior if the local write fails: a live CP may @@ -16699,25 +16704,52 @@ export class Daemon { while (!this.draining && (cp.state === 'READY' || cp.state === 'DRAINING')) { let row: SessionMetadataOutboxRow | undefined try { - row = this.store.nextSessionMetadataSnapshot(this.clock.now()) + row = this.store.nextSessionMetadataSnapshot(this.clock.now(), this.cfg.daemonId, this.servedAgentIds()) if (!row) return + // Emit only under a live claim: on a pool this outbox is one shared table, so a + // row a peer holds is that member's to report, never ours to duplicate or drop. + if ( + !this.store.claimSessionMetadataSnapshot( + row.agentId, + row.sessionId, + row.revision, + this.cfg.daemonId, + this.clock.now() + ) + ) { + continue + } + // The frame is scoped by the agent's organization, and only a member serving the + // agent can resolve it — leave the row for that member instead of failing it here. + if (!this.servesAgent(row.agentId) && this.parkSessionMetadataRow(row, 'agent served elsewhere')) continue const parsed = EventSessionSchema.safeParse(JSON.parse(row.snapshot)) if (!parsed.success || parsed.data.agentId !== row.agentId || parsed.data.sessionId !== row.sessionId) { this.log.warn(`event/session outbox dropped an invalid snapshot for session ${row.sessionId}`) - this.store.acknowledgeSessionMetadataSnapshot(row.agentId, row.sessionId, row.revision) + this.store.acknowledgeSessionMetadataSnapshot(row.agentId, row.sessionId, row.revision, this.cfg.daemonId) continue } const result = await cp.syncEventSession(parsed.data) if (result === 'unsupported') return // Revision fencing: an event produced while this request was in flight // remains pending instead of being cleared by the older ACK. - this.store.acknowledgeSessionMetadataSnapshot(row.agentId, row.sessionId, row.revision) + this.store.acknowledgeSessionMetadataSnapshot(row.agentId, row.sessionId, row.revision, this.cfg.daemonId) } catch (err) { if (!row) { this.log.warn(`event/session outbox read failed (${formatErr(err)})`) this.scheduleSessionMetadataRetry() return } + // SCOPE_DENIED says this member cannot name the agent's organization, not that the + // snapshot is bad. Park it for the member that can rather than burning a failure. + if ( + typeof err === 'object' && + err !== null && + 'code' in err && + err.code === 'SCOPE_DENIED' && + this.parkSessionMetadataRow(row, 'organization unresolvable here') + ) { + continue + } const nextFailure = row.failedAttempts + 1 const explicitlyPermanent = typeof err === 'object' && err !== null && 'retryable' in err && err.retryable === false @@ -16728,7 +16760,8 @@ export class Daemon { row.agentId, row.sessionId, row.revision, - defer ? this.clock.now() + SESSION_METADATA_DEFER_MS : null + defer ? this.clock.now() + SESSION_METADATA_DEFER_MS : null, + this.cfg.daemonId ) } catch (storeErr) { this.log.warn(`event/session outbox failure record failed (${formatErr(storeErr)})`) @@ -16752,15 +16785,48 @@ export class Daemon { } } - private schedulePendingSessionMetadataDrain(): void { - if ( - this.draining || - !this.cpClient?.supportsServerFeature?.(SESSION_METADATA_ACK_FEATURE) || - !this.store.hasPendingSessionMetadata() - ) { - return + /** Agents whose shared-outbox rows this member may work on: the ones it serves. */ + private servedAgentIds(): string[] { + return [...this.agents.keys()].filter((agentId) => this.servesAgent(agentId)) + } + + /** Release a snapshot this member cannot scope, so the member serving the agent drains + * it. The body and the failure count survive; the backoff only keeps it out of this + * member's next pass. False on a local store, where there is no other member. */ + private parkSessionMetadataRow(row: SessionMetadataOutboxRow, why: string): boolean { + let parked = false + try { + parked = this.store.parkSessionMetadataSnapshot( + row.agentId, + row.sessionId, + row.revision, + this.clock.now() + SESSION_METADATA_PARK_MS + ) + } catch (err) { + this.log.warn(`event/session outbox park failed (${formatErr(err)})`) + return false } - const attemptAt = this.store.nextSessionMetadataAttemptAt() + if (parked) this.log.debug(`event/session snapshot parked for session ${row.sessionId} (${why})`) + return parked + } + + /** A duty newly held here owns that agent's parked snapshots: re-arm them so the + * takeover replays them now instead of waiting out a departed holder's backoff. */ + private replayGainedSessionMetadata(agentIds: readonly string[]): void { + if (!agentIds.length) return + try { + this.store.resumeSessionMetadataSnapshots(agentIds) + } catch (err) { + this.log.warn(`event/session outbox resume failed (${formatErr(err)})`) + } + void this.drainSessionMetadataSnapshots() + } + + private schedulePendingSessionMetadataDrain(): void { + if (this.draining || !this.cpClient?.supportsServerFeature?.(SESSION_METADATA_ACK_FEATURE)) return + const served = this.servedAgentIds() + if (!this.store.hasPendingSessionMetadata(this.clock.now(), this.cfg.daemonId, served)) return + const attemptAt = this.store.nextSessionMetadataAttemptAt(this.clock.now(), this.cfg.daemonId, served) if (attemptAt !== undefined) this.scheduleSessionMetadataRetry(Math.max(0, attemptAt - this.clock.now())) } diff --git a/packages/daemon/src/store/local-store.ts b/packages/daemon/src/store/local-store.ts index 3f91e4c42..1629fd26e 100644 --- a/packages/daemon/src/store/local-store.ts +++ b/packages/daemon/src/store/local-store.ts @@ -636,7 +636,7 @@ function restrictPath(path: string, mode: number): void { * change that edits a `CREATE TABLE` below, and append the matching step to * {@link SCHEMA_MIGRATIONS}. */ -const SCHEMA_VERSION = 9 +const SCHEMA_VERSION = 10 /** * Ordered in-place upgrades for a store created by an EARLIER daemon. @@ -717,6 +717,11 @@ const SCHEMA_MIGRATIONS: ((db: StoreDatabase) => void)[] = [ db.exec(` ALTER TABLE session_purges ADD COLUMN ownerId TEXT; ALTER TABLE session_purges ADD COLUMN claimedAt INTEGER; + `), + (db) => + db.exec(` + ALTER TABLE session_metadata_outbox ADD COLUMN ownerId TEXT; + ALTER TABLE session_metadata_outbox ADD COLUMN claimedAt INTEGER; `) ] @@ -807,6 +812,8 @@ export class LocalStore { -- Latest-wins session metadata awaiting a correlated CP persistence ACK. -- This is deliberately separate from sessions: an upgrade starts with an -- empty outbox and never treats historical session rows as pending work. + -- On a shared pool store ownerId / claimedAt lease each snapshot to one + -- member the way inbox.reportOwnerId and session_purges.ownerId do. CREATE TABLE IF NOT EXISTS session_metadata_outbox ( agentId TEXT NOT NULL, sessionId TEXT NOT NULL, @@ -815,6 +822,8 @@ export class LocalStore { queuedAt INTEGER NOT NULL, failedAttempts INTEGER NOT NULL DEFAULT 0, nextAttemptAt INTEGER, + ownerId TEXT, + claimedAt INTEGER, PRIMARY KEY (agentId, sessionId) ); CREATE INDEX IF NOT EXISTS session_metadata_outbox_fifo @@ -2388,31 +2397,38 @@ export class LocalStore { sessionId: string, snapshot: string, enqueue: boolean, - queuedAt: number + queuedAt: number, + ownerId?: string ): number | undefined { + // Stamped to the writing member: the daemon that produced the snapshot serves the + // agent, so it is the one that can scope the frame's organization right now. + const owner = ownerId ?? null const row = enqueue ? (this.db .prepare( `INSERT INTO session_metadata_outbox - (agentId, sessionId, revision, snapshot, queuedAt, failedAttempts, nextAttemptAt) - VALUES (?, ?, 1, ?, ?, 0, NULL) + (agentId, sessionId, revision, snapshot, queuedAt, failedAttempts, nextAttemptAt, ownerId, claimedAt) + VALUES (?, ?, 1, ?, ?, 0, NULL, ?, ?) ON CONFLICT (agentId, sessionId) DO UPDATE SET revision = session_metadata_outbox.revision + 1, snapshot = excluded.snapshot, queuedAt = excluded.queuedAt, failedAttempts = 0, - nextAttemptAt = NULL + nextAttemptAt = NULL, + ownerId = excluded.ownerId, + claimedAt = excluded.claimedAt RETURNING revision` ) - .get(agentId, sessionId, snapshot, queuedAt) as { revision: number } | undefined) + .get(agentId, sessionId, snapshot, queuedAt, owner, queuedAt) as { revision: number } | undefined) : (this.db .prepare( `UPDATE session_metadata_outbox - SET revision = revision + 1, snapshot = ?, queuedAt = ?, failedAttempts = 0, nextAttemptAt = NULL + SET revision = revision + 1, snapshot = ?, queuedAt = ?, failedAttempts = 0, nextAttemptAt = NULL, + ownerId = ?, claimedAt = ? WHERE agentId = ? AND sessionId = ? RETURNING revision` ) - .get(snapshot, queuedAt, agentId, sessionId) as { revision: number } | undefined) + .get(snapshot, queuedAt, owner, queuedAt, agentId, sessionId) as { revision: number } | undefined) return row?.revision } @@ -2425,21 +2441,106 @@ export class LocalStore { .get(agentId, sessionId) as unknown as SessionMetadataOutboxRow | undefined } - nextSessionMetadataSnapshot(now = Date.now()): SessionMetadataOutboxRow | undefined { + /** The scope of the outbox this member may work on. A local store owns every row + * outright. On a shared pool store a row is offered when this member owns it, or + * when it is unowned / its owner's claim lapsed AND this member serves the agent. + * Unlike the hook outbox, an unowned row is NOT offered install-wide: the frame + * carries the agent's organization, which only a serving member can resolve, so a + * parked snapshot must wait for that member instead of circling the pool (#1023). */ + private sessionMetadataScope( + now: number, + ownerId?: string, + agentIds?: readonly string[] + ): { sql: string; params: SqlParams } { + if (!this.shared) return { sql: '', params: {} } + const scope = idScope('agentId', agentIds) + return { + sql: ` AND (ownerId = @ownerId OR ((ownerId IS NULL OR COALESCE(claimedAt, 0) <= @staleBefore)${scope.sql}))`, + params: { ownerId: ownerId ?? null, staleBefore: now - SHARED_OUTBOX_LEASE_MS, ...scope.params } + } + } + + nextSessionMetadataSnapshot( + now = Date.now(), + ownerId?: string, + agentIds?: readonly string[] + ): SessionMetadataOutboxRow | undefined { + const scope = this.sessionMetadataScope(now, ownerId, agentIds) return this.db .prepare( `SELECT agentId, sessionId, revision, snapshot, queuedAt, failedAttempts, nextAttemptAt FROM session_metadata_outbox - WHERE nextAttemptAt IS NULL OR nextAttemptAt <= ? + WHERE (nextAttemptAt IS NULL OR nextAttemptAt <= @now)${scope.sql} ORDER BY queuedAt ASC LIMIT 1` ) - .get(now) as unknown as SessionMetadataOutboxRow | undefined + .get({ now, ...scope.params }) as unknown as SessionMetadataOutboxRow | undefined + } + + /** Take or renew this member's claim on one snapshot before emitting it. Local + * stores never lease: the single owner claims everything. */ + claimSessionMetadataSnapshot( + agentId: string, + sessionId: string, + revision: number, + ownerId: string | undefined, + now: number + ): boolean { + if (!this.shared) return true + return ( + this.db + .prepare( + `UPDATE session_metadata_outbox + SET ownerId = @ownerId, claimedAt = @now + WHERE agentId = @agentId AND sessionId = @sessionId AND revision = @revision + AND (ownerId IS NULL OR ownerId = @ownerId OR COALESCE(claimedAt, 0) <= @staleBefore)` + ) + .run({ + agentId, + sessionId, + revision, + ownerId: ownerId ?? null, + now, + staleBefore: now - SHARED_OUTBOX_LEASE_MS + }).changes === 1 + ) + } + + /** Hand a snapshot this member cannot scope back to the pool: the claim is released + * so the member serving the agent picks it up, the body and the failure count are + * untouched, and the backoff keeps it out of this member's next pass. Returns false + * on a local store, where there is no other member to park it for. */ + parkSessionMetadataSnapshot(agentId: string, sessionId: string, revision: number, retryAt: number): boolean { + if (!this.shared) return false + return ( + this.db + .prepare( + `UPDATE session_metadata_outbox + SET ownerId = NULL, claimedAt = NULL, nextAttemptAt = @retryAt + WHERE agentId = @agentId AND sessionId = @sessionId AND revision = @revision` + ) + .run({ agentId, sessionId, revision, retryAt }).changes === 1 + ) + } + + /** Re-arm parked snapshots for agents a member has just gained: a takeover replays + * them now instead of waiting out the backoff a departed holder wrote. */ + resumeSessionMetadataSnapshots(agentIds: readonly string[]): number { + if (!this.shared || agentIds.length === 0) return 0 + const scope = idScope('agentId', agentIds) + return Number( + this.db + .prepare(`UPDATE session_metadata_outbox SET nextAttemptAt = NULL WHERE ownerId IS NULL${scope.sql}`) + .run(scope.params).changes + ) } - nextSessionMetadataAttemptAt(): number | undefined { + nextSessionMetadataAttemptAt(now = Date.now(), ownerId?: string, agentIds?: readonly string[]): number | undefined { + const scope = this.sessionMetadataScope(now, ownerId, agentIds) const row = this.db - .prepare('SELECT MIN(COALESCE(nextAttemptAt, 0)) AS attemptAt FROM session_metadata_outbox') - .get() as { attemptAt: number | null } | undefined + .prepare( + `SELECT MIN(COALESCE(nextAttemptAt, 0)) AS attemptAt FROM session_metadata_outbox WHERE 1 = 1${scope.sql}` + ) + .get(scope.params) as { attemptAt: number | null } | undefined return row?.attemptAt === null || row?.attemptAt === undefined ? undefined : Number(row.attemptAt) } @@ -2447,28 +2548,40 @@ export class LocalStore { agentId: string, sessionId: string, revision: number, - nextAttemptAt: number | null + nextAttemptAt: number | null, + ownerId?: string ): Pick | undefined { + const fence = this.shared ? ' AND (ownerId IS NULL OR ownerId = @ownerId)' : '' return this.db .prepare( `UPDATE session_metadata_outbox - SET failedAttempts = failedAttempts + 1, nextAttemptAt = ? - WHERE agentId = ? AND sessionId = ? AND revision = ? + SET failedAttempts = failedAttempts + 1, nextAttemptAt = @nextAttemptAt + WHERE agentId = @agentId AND sessionId = @sessionId AND revision = @revision${fence} RETURNING failedAttempts, nextAttemptAt` ) - .get(nextAttemptAt, agentId, sessionId, revision) as + .get({ agentId, sessionId, revision, nextAttemptAt, ...(fence ? { ownerId: ownerId ?? null } : {}) }) as Pick | undefined } - hasPendingSessionMetadata(): boolean { - return this.db.prepare('SELECT 1 AS pending FROM session_metadata_outbox LIMIT 1').get() !== undefined + hasPendingSessionMetadata(now = Date.now(), ownerId?: string, agentIds?: readonly string[]): boolean { + const scope = this.sessionMetadataScope(now, ownerId, agentIds) + return ( + this.db + .prepare(`SELECT 1 AS pending FROM session_metadata_outbox WHERE 1 = 1${scope.sql} LIMIT 1`) + .get(scope.params) !== undefined + ) } - /** Clear exactly the revision the CP ACKed. A newer coalesced snapshot wins. */ - acknowledgeSessionMetadataSnapshot(agentId: string, sessionId: string, revision: number): boolean { + /** Clear exactly the revision the CP ACKed. A newer coalesced snapshot wins. On a + * shared store only the claim holder may drop a row — never a peer's. */ + acknowledgeSessionMetadataSnapshot(agentId: string, sessionId: string, revision: number, ownerId?: string): boolean { + const fence = this.shared ? ' AND (ownerId IS NULL OR ownerId = @ownerId)' : '' const result = this.db - .prepare('DELETE FROM session_metadata_outbox WHERE agentId = ? AND sessionId = ? AND revision = ?') - .run(agentId, sessionId, revision) + .prepare( + `DELETE FROM session_metadata_outbox + WHERE agentId = @agentId AND sessionId = @sessionId AND revision = @revision${fence}` + ) + .run({ agentId, sessionId, revision, ...(fence ? { ownerId: ownerId ?? null } : {}) }) return result.changes === 1 } diff --git a/packages/daemon/test/daemon-session-metadata-outbox-pool.test.ts b/packages/daemon/test/daemon-session-metadata-outbox-pool.test.ts new file mode 100644 index 000000000..2ac5e2d31 --- /dev/null +++ b/packages/daemon/test/daemon-session-metadata-outbox-pool.test.ts @@ -0,0 +1,233 @@ +import { describe, it, expect, vi } from 'vitest' +import { mkdtempSync, mkdirSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { DatabaseSync } from 'node:sqlite' +import { Daemon } from '../src/daemon.js' +import { LocalStore } from '../src/store/local-store.js' +import { statePath } from '../src/paths.js' +import { FakeClock } from './cp/fake-clock.js' + +/** + * #1023 — on a daemon pool the session-metadata outbox is one shared table, but the + * `event/session-sync` frame is scoped by the agent's organization, which only a member + * serving that agent can resolve. Every member used to drain every row and defer the + * ones it could not scope forever. A member now drains only what it owns or serves, + * claims before it emits, and parks what it cannot scope for the member that can. + */ + +const AGENT_A = '33333333-3333-4333-8333-333333333333' +const AGENT_B = '44444444-4444-4444-8444-444444444444' +const GROUP_A = '11111111-1111-4111-8111-111111111111' +const GROUP_B = '22222222-2222-4222-8222-222222222222' +const ORG = 'org-1' +const PARK_MS = 60_000 + +function scaffold(): string { + const root = mkdtempSync(join(tmpdir(), 'ac-metadata-pool-')) + writeFileSync( + join(root, 'config.json'), + JSON.stringify({ + version: 1, + controlPlane: { enabled: false }, + runtimes: { claude: { command: 'node', args: ['unused'] } } + }) + ) + for (const id of [AGENT_A, AGENT_B]) { + const adir = join(root, 'agents', id) + mkdirSync(adir, { recursive: true }) + writeFileSync( + join(adir, 'agent.json'), + JSON.stringify({ + id, + name: id, + status: 'active', + runtime: 'claude', + workspace: { mode: 'from-scratch', path: join(adir, 'workspace') }, + integrations: [], + output: { mode: 'low' } + }) + ) + } + return root +} + +type Member = Awaited> + +/** One member whose CP stub scopes exactly like the real client: the organization comes + * from the agent registry, which on a pool carries only the agents this member serves. */ +async function boot(root: string, daemonId: string, scope: 'frame' | 'install' = 'frame') { + const clock = new FakeClock() + const daemon = new Daemon({ root, hostFactory: () => ({}) as any, clock }) + await daemon.start() + const inner = daemon as any + inner.cfg.daemonId = daemonId + const warn = vi.fn() + inner.log.warn = warn + const scopeBlind = new Set() + const synced: { orgId: string; agentId: string; sessionId: string }[] = [] + const syncEventSession = vi.fn(async (event: { agentId: string; sessionId: string }) => { + const known = scope === 'install' || inner.duties.holdsAgent(event.agentId) + if (!known || scopeBlind.has(event.agentId)) { + throw Object.assign(new Error('cannot resolve organization for event/session-sync'), { + name: 'WireError', + code: 'SCOPE_DENIED', + retryable: false + }) + } + synced.push({ orgId: ORG, agentId: event.agentId, sessionId: event.sessionId }) + return 'acknowledged' as const + }) + inner.cpClient = { + organizationScope: () => scope, + state: 'READY', + supportsServerFeature: (feature: string) => feature === 'session-metadata-ack-v1', + syncEventSession, + emitSessionPurged: vi.fn(async () => 'acknowledged' as const), + stop: async () => {}, + releaseDuties: vi.fn(async () => {}), + reportDutiesNow: vi.fn(() => {}), + fetchDutyAgent: vi.fn() + } + return { daemon, inner, clock, warn, synced, syncEventSession, scopeBlind } +} + +/** Two members over ONE store, each with its own clock and its own duty leases. */ +async function bootPool() { + const root = scaffold() + const a = await boot(root, 'daemon-a') + const b = await boot(root, 'daemon-b') + const locals: LocalStore[] = [a.inner.store, b.inner.store] + const path = statePath(root) + a.inner.store = new LocalStore({ database: new DatabaseSync(path), shared: true, ownerId: 'daemon-a' }) + b.inner.store = new LocalStore({ database: new DatabaseSync(path), shared: true, ownerId: 'daemon-b' }) + const shared: LocalStore = a.inner.store + return { + a, + b, + shared, + stop: async () => { + await Promise.all([a.daemon.stop(), b.daemon.stop()]) + for (const local of locals) local.close() + } + } +} + +const grant = (groupId: string, agentId: string) => ({ + groupId, + orgId: ORG, + term: '1', + members: [{ kind: 'agent' as const, refId: agentId }] +}) +const hold = (inner: any, groupId: string, agentId: string) => inner.duties.applyGrant([grant(groupId, agentId)]) + +/** One unacknowledged terminal milestone, stamped to the member that produced it. */ +function seedSnapshot(store: LocalStore, agentId: string, sessionId: string, queuedAt: number, ownerId: string): void { + const event = { + sessionId, + agentId, + phase: 'end', + platform: 'slack', + channel: 'C1', + ts: new Date(queuedAt).toISOString() + } + store.saveSessionMetadataSnapshot(agentId, sessionId, JSON.stringify(event), true, queuedAt, ownerId) +} + +const deferred = (member: Member) => + member.warn.mock.calls.filter(([message]: [string]) => String(message).includes('snapshot deferred')) + +describe('session-metadata outbox ownership on a daemon pool (#1023)', () => { + it("a member drains its own rows and never a peer's", async () => { + const { a, b, shared, stop } = await bootPool() + hold(a.inner, GROUP_A, AGENT_A) + hold(b.inner, GROUP_B, AGENT_B) + seedSnapshot(shared, AGENT_A, 'acp-a-1', 1, 'daemon-a') + seedSnapshot(shared, AGENT_B, 'acp-b-1', 2, 'daemon-b') + + await b.inner.drainSessionMetadataSnapshots() + expect(b.synced).toEqual([{ orgId: ORG, agentId: AGENT_B, sessionId: 'acp-b-1' }]) + // A's row is untouched: still pending, still unfailed, still A's to report. + expect(shared.pendingSessionMetadataSnapshot(AGENT_A, 'acp-a-1')).toMatchObject({ revision: 1, failedAttempts: 0 }) + expect(deferred(b)).toEqual([]) + + await a.inner.drainSessionMetadataSnapshots() + expect(a.synced).toEqual([{ orgId: ORG, agentId: AGENT_A, sessionId: 'acp-a-1' }]) + expect(shared.hasPendingSessionMetadata()).toBe(false) + await stop() + }, 15_000) + + it('parks a row for an agent no member here serves instead of counting a failure', async () => { + const { a, b, shared, stop } = await bootPool() + // The duty for AGENT_A moved off this member after it wrote the snapshot. + hold(b.inner, GROUP_B, AGENT_B) + seedSnapshot(shared, AGENT_A, 'acp-moved', 1, 'daemon-a') + + await a.inner.drainSessionMetadataSnapshots() + expect(a.syncEventSession).not.toHaveBeenCalled() + // Claim released, body and failure count intact, backoff only keeps it out of this pass. + expect(shared.pendingSessionMetadataSnapshot(AGENT_A, 'acp-moved')).toMatchObject({ + revision: 1, + failedAttempts: 0, + nextAttemptAt: PARK_MS + }) + expect(deferred(a)).toEqual([]) + // The peer does not serve AGENT_A either, so it leaves the parked row alone. + await b.inner.drainSessionMetadataSnapshots() + expect(b.syncEventSession).not.toHaveBeenCalled() + expect(shared.pendingSessionMetadataSnapshot(AGENT_A, 'acp-moved')).toBeDefined() + await stop() + }, 15_000) + + it('gaining the duty replays the parked row exactly once, scoped to the agent org', async () => { + const { a, b, shared, stop } = await bootPool() + seedSnapshot(shared, AGENT_A, 'acp-moved', 1, 'daemon-a') + await a.inner.drainSessionMetadataSnapshots() + expect(shared.pendingSessionMetadataSnapshot(AGENT_A, 'acp-moved')?.nextAttemptAt).toBe(PARK_MS) + + // The grant lands on B well before the parked backoff would have expired. + b.inner.settleDutyChange(b.inner.duties.applyGrant([grant(GROUP_A, AGENT_A)])) + await vi.waitFor(() => expect(b.syncEventSession).toHaveBeenCalledOnce()) + expect(b.synced).toEqual([{ orgId: ORG, agentId: AGENT_A, sessionId: 'acp-moved' }]) + expect(shared.hasPendingSessionMetadata()).toBe(false) + await b.inner.drainSessionMetadataSnapshots() + expect(b.syncEventSession).toHaveBeenCalledOnce() + await stop() + }, 15_000) + + it('parks a served row whose organization is not resolvable yet, and drains it once it is', async () => { + const { a, shared, stop } = await bootPool() + hold(a.inner, GROUP_A, AGENT_A) + a.scopeBlind.add(AGENT_A) + seedSnapshot(shared, AGENT_A, 'acp-cold', 1, 'daemon-a') + + await a.inner.drainSessionMetadataSnapshots() + expect(a.syncEventSession).toHaveBeenCalledOnce() + // A local SCOPE_DENIED is not a rejection of the snapshot: no failure, no defer warning. + expect(shared.pendingSessionMetadataSnapshot(AGENT_A, 'acp-cold')).toMatchObject({ + failedAttempts: 0, + nextAttemptAt: PARK_MS + }) + expect(deferred(a)).toEqual([]) + + a.scopeBlind.clear() + a.clock.advance(PARK_MS + 1) + await a.inner.drainSessionMetadataSnapshots() + expect(a.synced).toEqual([{ orgId: ORG, agentId: AGENT_A, sessionId: 'acp-cold' }]) + expect(shared.hasPendingSessionMetadata()).toBe(false) + await stop() + }, 15_000) + + it('a single daemon on its own store drains every row unfenced', async () => { + const root = scaffold() + const solo = await boot(root, 'daemon-solo', 'install') + const store: LocalStore = solo.inner.store + seedSnapshot(store, AGENT_A, 'acp-1', 1, 'daemon-solo') + seedSnapshot(store, AGENT_B, 'acp-2', 2, 'daemon-solo') + + await solo.inner.drainSessionMetadataSnapshots() + expect(solo.synced.map((entry) => entry.sessionId)).toEqual(['acp-1', 'acp-2']) + expect(store.hasPendingSessionMetadata()).toBe(false) + await solo.daemon.stop() + }, 15_000) +}) diff --git a/packages/daemon/test/local-store.test.ts b/packages/daemon/test/local-store.test.ts index 434f72ef8..73fc903a1 100644 --- a/packages/daemon/test/local-store.test.ts +++ b/packages/daemon/test/local-store.test.ts @@ -62,6 +62,8 @@ describe('LocalStore schema versioning', () => { old.exec('ALTER TABLE cron_runs DROP COLUMN definition') old.exec('ALTER TABLE session_purges DROP COLUMN ownerId') old.exec('ALTER TABLE session_purges DROP COLUMN claimedAt') + old.exec('ALTER TABLE session_metadata_outbox DROP COLUMN ownerId') + old.exec('ALTER TABLE session_metadata_outbox DROP COLUMN claimedAt') old.exec('PRAGMA user_version = 1') old.close() @@ -79,7 +81,8 @@ describe('LocalStore schema versioning', () => { const purgeColumns = columnsOf('session_purges') upgraded.close() expect(columns).toContain('ownerId') - expect(outboxColumns).toEqual(expect.arrayContaining(['failedAttempts', 'nextAttemptAt'])) + // Session-metadata snapshots are leased per pool member too (#1023). + expect(outboxColumns).toEqual(expect.arrayContaining(['failedAttempts', 'nextAttemptAt', 'ownerId', 'claimedAt'])) // Recovery ownership: a pool member must be able to tell its own rows from a peer's. expect(dreamColumns).toContain('ownerId') expect(grantColumns).toContain('ownerId') @@ -88,7 +91,7 @@ describe('LocalStore schema versioning', () => { expect(cronColumns).toContain('definition') // Purge receipts are leased per pool member (#1032). expect(purgeColumns).toEqual(expect.arrayContaining(['ownerId', 'claimedAt'])) - expect(userVersion(path)).toBe(9) + expect(userVersion(path)).toBe(10) }) it('never persists the CP routing map on a shared store, and still does on an owned one', () => { @@ -131,6 +134,8 @@ describe('LocalStore schema versioning', () => { old.exec('ALTER TABLE cron_runs DROP COLUMN definition') old.exec('ALTER TABLE session_purges DROP COLUMN ownerId') old.exec('ALTER TABLE session_purges DROP COLUMN claimedAt') + old.exec('ALTER TABLE session_metadata_outbox DROP COLUMN ownerId') + old.exec('ALTER TABLE session_metadata_outbox DROP COLUMN claimedAt') old.exec('PRAGMA user_version = 5') old.close() @@ -146,7 +151,7 @@ describe('LocalStore schema versioning', () => { expect(upgraded.isCaptureExcluded('bot-c', 'acp-2')).toBe(true) upgraded.close() - expect(userVersion(path)).toBe(9) + expect(userVersion(path)).toBe(10) }) it('re-keys the runtime catalog cache on its owning member when upgrading a v7 store', () => { @@ -172,6 +177,8 @@ describe('LocalStore schema versioning', () => { `) old.exec('ALTER TABLE session_purges DROP COLUMN ownerId') old.exec('ALTER TABLE session_purges DROP COLUMN claimedAt') + old.exec('ALTER TABLE session_metadata_outbox DROP COLUMN ownerId') + old.exec('ALTER TABLE session_metadata_outbox DROP COLUMN claimedAt') old.exec('PRAGMA user_version = 7') old.close() @@ -193,7 +200,7 @@ describe('LocalStore schema versioning', () => { .map((column) => column.name) expect(primaryKey(metaColumns)).toEqual(['ownerId', 'runtimeId']) expect(primaryKey(capColumns)).toEqual(['ownerId', 'runtimeId', 'modelId']) - expect(userVersion(path)).toBe(9) + expect(userVersion(path)).toBe(10) }) it('refuses a store written by a newer daemon WITHOUT touching it first', () => { From 9e86debaf73925b788736cdeabcb3def5b74c5aa Mon Sep 17 00:00:00 2001 From: zfy0701 <1646270+zfy0701@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:25:09 +0800 Subject: [PATCH 2/3] fix(daemon): outlast a departed holder's session-metadata claim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up on #1023. A graceful handoff could strand a snapshot: member A persists it, enters shutdown (draining, retry timer cleared) and releases the duty before parking the row, so the row still names A with a fresh claim. The successor's scope hid it, the duty-gain replay only re-armed unowned rows, and nothing was scheduled for the moment the claim lapsed. Three ways out, all of them cheap: - `reclaimSessionMetadataSnapshots` now releases a previous holder's claim for the gained agents, not just the parked rows. The duty ledger has already proved that member no longer serves the agent; this member's own claims are left alone, since only it knows whether one is in flight. - `nextSessionMetadataAttemptAt` and `hasPendingSessionMetadata` read a wider work scope — own rows plus every row of a served agent — and the wake is armed at `claimedAt + SHARED_OUTBOX_LEASE_MS` for a row a peer still holds, so the drain runs when the claim lapses even without a duty change. - Shutdown releases every claim this member still holds, once its drain has been joined and before the store is closed, so a successor does not wait out a lease nobody will renew. Co-Authored-By: Claude Fable 5 --- packages/daemon/src/daemon.ts | 26 +++++-- packages/daemon/src/store/local-store.ts | 67 +++++++++++++++---- ...aemon-session-metadata-outbox-pool.test.ts | 53 ++++++++++++++- 3 files changed, 126 insertions(+), 20 deletions(-) diff --git a/packages/daemon/src/daemon.ts b/packages/daemon/src/daemon.ts index de279441a..1b06f105e 100644 --- a/packages/daemon/src/daemon.ts +++ b/packages/daemon/src/daemon.ts @@ -16810,23 +16810,35 @@ export class Daemon { return parked } - /** A duty newly held here owns that agent's parked snapshots: re-arm them so the - * takeover replays them now instead of waiting out a departed holder's backoff. */ + /** A duty newly held here owns that agent's snapshots: take the parked ones off their + * backoff and the previous holder's claim off the rest — it released the duty, so it + * will never emit them — then replay at once instead of waiting out the lease. */ private replayGainedSessionMetadata(agentIds: readonly string[]): void { if (!agentIds.length) return try { - this.store.resumeSessionMetadataSnapshots(agentIds) + this.store.reclaimSessionMetadataSnapshots(agentIds, this.cfg.daemonId) } catch (err) { - this.log.warn(`event/session outbox resume failed (${formatErr(err)})`) + this.log.warn(`event/session outbox reclaim failed (${formatErr(err)})`) } void this.drainSessionMetadataSnapshots() } + /** Shutdown counterpart: this member will not emit again, so every claim it still holds + * goes back to the pool for its successor instead of blocking on the lease. */ + private releaseOwnedSessionMetadata(): void { + try { + const released = this.store.releaseOwnedSessionMetadataSnapshots(this.cfg.daemonId) + if (released) this.log.debug(`event/session outbox released ${released} claim(s) for the pool`) + } catch (err) { + this.log.warn(`event/session outbox release failed (${formatErr(err)})`) + } + } + private schedulePendingSessionMetadataDrain(): void { if (this.draining || !this.cpClient?.supportsServerFeature?.(SESSION_METADATA_ACK_FEATURE)) return const served = this.servedAgentIds() - if (!this.store.hasPendingSessionMetadata(this.clock.now(), this.cfg.daemonId, served)) return - const attemptAt = this.store.nextSessionMetadataAttemptAt(this.clock.now(), this.cfg.daemonId, served) + if (!this.store.hasPendingSessionMetadata(this.cfg.daemonId, served)) return + const attemptAt = this.store.nextSessionMetadataAttemptAt(this.cfg.daemonId, served) if (attemptAt !== undefined) this.scheduleSessionMetadataRetry(Math.max(0, attemptAt - this.clock.now())) } @@ -23165,6 +23177,8 @@ export class Daemon { this.shutdownDutyDrain = undefined await Promise.resolve(this.cpClient?.stop()).catch((e) => errors.push(e)) await Promise.resolve(this.sessionMetadataDrain).catch((e) => errors.push(e)) + // Nothing here can emit after this point; hand any claim this member still holds back. + this.releaseOwnedSessionMetadata() // The closed CP transport cannot admit another lifecycle frame. Drain every // remove/upsert/move already published into its per-agent queue before any // store or registry it may still touch is closed. diff --git a/packages/daemon/src/store/local-store.ts b/packages/daemon/src/store/local-store.ts index 1629fd26e..9a041998d 100644 --- a/packages/daemon/src/store/local-store.ts +++ b/packages/daemon/src/store/local-store.ts @@ -2460,6 +2460,19 @@ export class LocalStore { } } + /** Everything this member is eventually answerable for: its own rows plus every row of + * an agent it serves, whoever holds the claim right now. Wider than the claimable-now + * scope on purpose — a peer's live claim on a served agent's row still has to arm this + * member's wake, or nothing would run when that claim lapses. */ + private sessionMetadataWorkScope(ownerId?: string, agentIds?: readonly string[]): { sql: string; params: SqlParams } { + if (!this.shared) return { sql: '', params: {} } + const scope = idScope('agentId', agentIds) + return { + sql: ` AND (ownerId = @ownerId OR 1 = 1${scope.sql})`, + params: { ownerId: ownerId ?? null, ...scope.params } + } + } + nextSessionMetadataSnapshot( now = Date.now(), ownerId?: string, @@ -2522,25 +2535,52 @@ export class LocalStore { ) } - /** Re-arm parked snapshots for agents a member has just gained: a takeover replays - * them now instead of waiting out the backoff a departed holder wrote. */ - resumeSessionMetadataSnapshots(agentIds: readonly string[]): number { + /** Hand the snapshots of agents a member has just gained back to the pool: the parked + * ones lose their backoff and a previous holder's claim — live or not — is released, + * because the duty ledger has already proved that member no longer serves the agent. + * This member's own claims are left alone; only it knows whether they are in flight. */ + reclaimSessionMetadataSnapshots(agentIds: readonly string[], ownerId?: string): number { if (!this.shared || agentIds.length === 0) return 0 const scope = idScope('agentId', agentIds) return Number( this.db - .prepare(`UPDATE session_metadata_outbox SET nextAttemptAt = NULL WHERE ownerId IS NULL${scope.sql}`) - .run(scope.params).changes + .prepare( + `UPDATE session_metadata_outbox + SET ownerId = NULL, claimedAt = NULL, nextAttemptAt = NULL + WHERE (ownerId IS NULL OR ownerId <> @ownerId)${scope.sql}` + ) + .run({ ownerId: ownerId ?? null, ...scope.params }).changes ) } - nextSessionMetadataAttemptAt(now = Date.now(), ownerId?: string, agentIds?: readonly string[]): number | undefined { - const scope = this.sessionMetadataScope(now, ownerId, agentIds) + /** Release every claim this member still holds, at shutdown: it will not emit again, so + * a successor must not wait out the lease. Bodies, revisions and backoffs survive. */ + releaseOwnedSessionMetadataSnapshots(ownerId?: string): number { + if (!this.shared) return 0 + return Number( + this.db + .prepare( + `UPDATE session_metadata_outbox + SET ownerId = NULL, claimedAt = NULL, nextAttemptAt = NULL + WHERE ownerId = @ownerId` + ) + .run({ ownerId: ownerId ?? null }).changes + ) + } + + /** When the earliest row this member is answerable for becomes workable: its own backoff, + * or — for a row a peer still holds — the moment that claim lapses. Without the second + * half a graceful handoff leaves a live foreign claim with nothing armed to outlast it. */ + nextSessionMetadataAttemptAt(ownerId?: string, agentIds?: readonly string[]): number | undefined { + const scope = this.sessionMetadataWorkScope(ownerId, agentIds) + const lapse = this.shared + ? `MAX(COALESCE(nextAttemptAt, 0), CASE WHEN ownerId IS NOT NULL AND ownerId <> @ownerId + THEN COALESCE(claimedAt, 0) + @lease ELSE 0 END)` + : 'COALESCE(nextAttemptAt, 0)' const row = this.db - .prepare( - `SELECT MIN(COALESCE(nextAttemptAt, 0)) AS attemptAt FROM session_metadata_outbox WHERE 1 = 1${scope.sql}` - ) - .get(scope.params) as { attemptAt: number | null } | undefined + .prepare(`SELECT MIN(${lapse}) AS attemptAt FROM session_metadata_outbox WHERE 1 = 1${scope.sql}`) + .get({ ...scope.params, ...(this.shared ? { lease: SHARED_OUTBOX_LEASE_MS } : {}) }) as + { attemptAt: number | null } | undefined return row?.attemptAt === null || row?.attemptAt === undefined ? undefined : Number(row.attemptAt) } @@ -2563,8 +2603,9 @@ export class LocalStore { Pick | undefined } - hasPendingSessionMetadata(now = Date.now(), ownerId?: string, agentIds?: readonly string[]): boolean { - const scope = this.sessionMetadataScope(now, ownerId, agentIds) + /** Any row this member is answerable for, workable now or once a peer's claim lapses. */ + hasPendingSessionMetadata(ownerId?: string, agentIds?: readonly string[]): boolean { + const scope = this.sessionMetadataWorkScope(ownerId, agentIds) return ( this.db .prepare(`SELECT 1 AS pending FROM session_metadata_outbox WHERE 1 = 1${scope.sql} LIMIT 1`) diff --git a/packages/daemon/test/daemon-session-metadata-outbox-pool.test.ts b/packages/daemon/test/daemon-session-metadata-outbox-pool.test.ts index 2ac5e2d31..98dcc4dea 100644 --- a/packages/daemon/test/daemon-session-metadata-outbox-pool.test.ts +++ b/packages/daemon/test/daemon-session-metadata-outbox-pool.test.ts @@ -22,6 +22,7 @@ const GROUP_A = '11111111-1111-4111-8111-111111111111' const GROUP_B = '22222222-2222-4222-8222-222222222222' const ORG = 'org-1' const PARK_MS = 60_000 +const LEASE_MS = 2 * 60_000 function scaffold(): string { const root = mkdtempSync(join(tmpdir(), 'ac-metadata-pool-')) @@ -107,7 +108,7 @@ async function bootPool() { b, shared, stop: async () => { - await Promise.all([a.daemon.stop(), b.daemon.stop()]) + await Promise.all([a.daemon.stop().catch(() => {}), b.daemon.stop().catch(() => {})]) for (const local of locals) local.close() } } @@ -218,6 +219,56 @@ describe('session-metadata outbox ownership on a daemon pool (#1023)', () => { await stop() }, 15_000) + it('takes over a claim the departed holder never released when the duty is gained', async () => { + const { a, b, shared, stop } = await bootPool() + // A wrote the snapshot, then released the duty on a graceful shutdown before parking it: + // the row still names A with a fresh claim, so nothing in the pool could touch it. + hold(a.inner, GROUP_A, AGENT_A) + seedSnapshot(shared, AGENT_A, 'acp-handoff', 1, 'daemon-a') + + b.inner.settleDutyChange(b.inner.duties.applyGrant([grant(GROUP_A, AGENT_A)])) + await vi.waitFor(() => expect(b.syncEventSession).toHaveBeenCalledOnce()) + expect(b.synced).toEqual([{ orgId: ORG, agentId: AGENT_A, sessionId: 'acp-handoff' }]) + expect(shared.hasPendingSessionMetadata()).toBe(false) + await stop() + }, 15_000) + + it("arms a wake at the lease expiry of a peer's claim on a served agent", async () => { + const { a, b, stop } = await bootPool() + // Both members read the duty as theirs — the takeover already happened, but no duty change + // fires here, so only the armed wake can outlast A's claim. + hold(a.inner, GROUP_A, AGENT_A) + hold(b.inner, GROUP_A, AGENT_A) + seedSnapshot(a.inner.store, AGENT_A, 'acp-lease', 1, 'daemon-a') + + await b.inner.drainSessionMetadataSnapshots() + expect(b.syncEventSession).not.toHaveBeenCalled() + expect(b.clock.pending()).toContain(1 + LEASE_MS) + + b.clock.advance(1 + LEASE_MS) + await vi.waitFor(() => expect(b.syncEventSession).toHaveBeenCalledOnce()) + expect(b.synced).toEqual([{ orgId: ORG, agentId: AGENT_A, sessionId: 'acp-lease' }]) + await stop() + }, 15_000) + + it('releases the claims it still holds when it stops, so a successor drains at once', async () => { + const { a, b, stop } = await bootPool() + hold(a.inner, GROUP_A, AGENT_A) + hold(b.inner, GROUP_A, AGENT_A) + // Asserted through B's handle: stopping A closes the store handle A opened. + const survivor: LocalStore = b.inner.store + seedSnapshot(survivor, AGENT_A, 'acp-exit', 1, 'daemon-a') + + await b.inner.drainSessionMetadataSnapshots() + expect(b.syncEventSession).not.toHaveBeenCalled() + + await a.daemon.stop() + await b.inner.drainSessionMetadataSnapshots() + expect(b.synced).toEqual([{ orgId: ORG, agentId: AGENT_A, sessionId: 'acp-exit' }]) + expect(survivor.hasPendingSessionMetadata()).toBe(false) + await stop() + }, 15_000) + it('a single daemon on its own store drains every row unfenced', async () => { const root = scaffold() const solo = await boot(root, 'daemon-solo', 'install') From 3afa7cce5e42db61ef4391201f83f3b467c838b6 Mon Sep 17 00:00:00 2001 From: zfy0701 <1646270+zfy0701@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:48:36 +0800 Subject: [PATCH 3/3] fix(daemon): keep the outbox wake query portable to the pool store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up on #1023. `nextSessionMetadataAttemptAt` reached for SQLite's scalar two-argument `max(x, y)`. The same statement text runs through PostgreSQL on a pool, where `MAX` is only an aggregate, so the query raised `function max(bigint, bigint) does not exist`. The refill check only logs a failed read, so the lease-expiry wake was never armed there — exactly the recovery path the previous commit added. The later-of is expressed as one CASE instead, which both engines parse the same way. No rewrite rule was added to the worker: `MAX(` → `GREATEST(` would also rewrite every legitimate one-argument aggregate in this file. `postgres-pool-store.int.test.ts` gains a session-metadata case that drives the lease, the claim CAS, the park, the reclaim and this query through `PostgresSyncDatabase`. It fails with the two-argument form and passes with the CASE. The stale `getCronLastRun` assertion in the first case is repaired in passing — that method was renamed to `cronRun` and the suite only runs with a database, so nothing caught it. Co-Authored-By: Claude Fable 5 --- packages/daemon/src/store/local-store.ts | 15 +++--- .../test/postgres-pool-store.int.test.ts | 54 ++++++++++++++++++- 2 files changed, 61 insertions(+), 8 deletions(-) diff --git a/packages/daemon/src/store/local-store.ts b/packages/daemon/src/store/local-store.ts index 9a041998d..4510aca91 100644 --- a/packages/daemon/src/store/local-store.ts +++ b/packages/daemon/src/store/local-store.ts @@ -2569,16 +2569,19 @@ export class LocalStore { } /** When the earliest row this member is answerable for becomes workable: its own backoff, - * or — for a row a peer still holds — the moment that claim lapses. Without the second - * half a graceful handoff leaves a live foreign claim with nothing armed to outlast it. */ + * or — for a row a peer still holds — the later of that backoff and the moment the claim + * lapses. Without the second half a graceful handoff leaves a live foreign claim with + * nothing armed to outlast it. Written as one CASE rather than a two-argument `max()`: the + * same statement text runs on the pool's PostgreSQL, where `MAX` is only an aggregate. */ nextSessionMetadataAttemptAt(ownerId?: string, agentIds?: readonly string[]): number | undefined { const scope = this.sessionMetadataWorkScope(ownerId, agentIds) - const lapse = this.shared - ? `MAX(COALESCE(nextAttemptAt, 0), CASE WHEN ownerId IS NOT NULL AND ownerId <> @ownerId - THEN COALESCE(claimedAt, 0) + @lease ELSE 0 END)` + const workableAt = this.shared + ? `CASE WHEN ownerId IS NOT NULL AND ownerId <> @ownerId + AND COALESCE(claimedAt, 0) + @lease > COALESCE(nextAttemptAt, 0) + THEN COALESCE(claimedAt, 0) + @lease ELSE COALESCE(nextAttemptAt, 0) END` : 'COALESCE(nextAttemptAt, 0)' const row = this.db - .prepare(`SELECT MIN(${lapse}) AS attemptAt FROM session_metadata_outbox WHERE 1 = 1${scope.sql}`) + .prepare(`SELECT MIN(${workableAt}) AS attemptAt FROM session_metadata_outbox WHERE 1 = 1${scope.sql}`) .get({ ...scope.params, ...(this.shared ? { lease: SHARED_OUTBOX_LEASE_MS } : {}) }) as { attemptAt: number | null } | undefined return row?.attemptAt === null || row?.attemptAt === undefined ? undefined : Number(row.attemptAt) diff --git a/packages/daemon/test/postgres-pool-store.int.test.ts b/packages/daemon/test/postgres-pool-store.int.test.ts index 5811cbd6a..e860374c4 100644 --- a/packages/daemon/test/postgres-pool-store.int.test.ts +++ b/packages/daemon/test/postgres-pool-store.int.test.ts @@ -81,7 +81,7 @@ describe.skipIf(!databaseUrl)('PostgreSQL pool member store', () => { createdAt: 1, updatedAt: 1 }) - first.store.setCronLastRun(`${agentId}:cron`, 42) + first.store.setCronLastRun(`${agentId}:cron`, 42, '{}') first.store.setDisplayName(`U-${suffix}`, 'Cloud user', 1) first.store.saveSessionMetadataSnapshot(agentId, `session-${suffix}`, '{"title":"Cloud"}', true, 7) first.store.insertDream({ @@ -116,7 +116,7 @@ describe.skipIf(!databaseUrl)('PostgreSQL pool member store', () => { expect.objectContaining({ id: `delivery-${suffix}`, sessionKey, loopGuardCounted: 1 }) ) expect(second.store.nextMemoryCaptureDueAt()).toBe(1234) - expect(second.store.getCronLastRun(`${agentId}:cron`)).toBe(42) + expect(second.store.cronRun(`${agentId}:cron`)?.lastRunAt).toBe(42) expect(second.store.getDisplayNames([`U-${suffix}`]).get(`U-${suffix}`)).toBe('Cloud user') expect(second.store.pendingSessionMetadataSnapshot(agentId, `session-${suffix}`)?.snapshot).toBe( '{"title":"Cloud"}' @@ -257,6 +257,56 @@ describe.skipIf(!databaseUrl)('PostgreSQL pool member store', () => { } }) + it('leases session-metadata snapshots per member and wakes when the claim lapses', async () => { + // #1023 against the real engine: the outbox is install-wide here, and the refill check's + // "when does this become workable" query must run on PostgreSQL, not just SQLite. + const suffix = randomUUID() + const agentId = `agent-${suffix}` + const sessionId = `session-${suffix}` + const ownerA = `daemon-a-${suffix}` + const ownerB = `daemon-b-${suffix}` + const lease = 2 * 60 * 1_000 + const config = { version: 1 as const, databaseUrl: databaseUrl!, maxConnections: 2 } + const orgForAgent = (id: string) => (id === agentId ? `org-${suffix}` : undefined) + const first = await PostgresDataPlane.open(config, orgForAgent) + const second = await PostgresDataPlane.open(config, orgForAgent) + try { + expect(first.store.saveSessionMetadataSnapshot(agentId, sessionId, '{"phase":"end"}', true, 1_000, ownerA)).toBe( + 1 + ) + // A's claim is live: B is not offered the row, cannot take it, and cannot release it. + expect(second.store.nextSessionMetadataSnapshot(1_500, ownerB, [agentId])).toBeUndefined() + expect(second.store.claimSessionMetadataSnapshot(agentId, sessionId, 1, ownerB, 1_500)).toBe(false) + expect(second.store.acknowledgeSessionMetadataSnapshot(agentId, sessionId, 1, ownerB)).toBe(false) + // ...but B's wake is armed for the moment it lapses, so nothing waits on a duty change. + expect(second.store.nextSessionMetadataAttemptAt(ownerB, [agentId])).toBe(1_000 + lease) + + const lapsed = 1_000 + lease + 1 + expect(second.store.nextSessionMetadataSnapshot(lapsed, ownerB, [agentId])?.sessionId).toBe(sessionId) + expect(second.store.claimSessionMetadataSnapshot(agentId, sessionId, 1, ownerB, lapsed)).toBe(true) + // Parking returns the row to the pool with its body and failure count intact. + expect(second.store.parkSessionMetadataSnapshot(agentId, sessionId, 1, lapsed + 60_000)).toBe(true) + expect(second.store.pendingSessionMetadataSnapshot(agentId, sessionId)).toMatchObject({ + failedAttempts: 0, + snapshot: '{"phase":"end"}' + }) + expect(second.store.nextSessionMetadataSnapshot(lapsed, ownerB, [agentId])).toBeUndefined() + + // The duty comes back to A: the reclaim drops the backoff and settles under A's fence. + expect(first.store.reclaimSessionMetadataSnapshots([agentId], ownerA)).toBe(1) + expect(first.store.nextSessionMetadataAttemptAt(ownerA, [agentId])).toBe(0) + expect(first.store.nextSessionMetadataSnapshot(lapsed, ownerA, [agentId])?.sessionId).toBe(sessionId) + expect(first.store.claimSessionMetadataSnapshot(agentId, sessionId, 1, ownerA, lapsed)).toBe(true) + expect(first.store.releaseOwnedSessionMetadataSnapshots(ownerA)).toBe(1) + expect(first.store.acknowledgeSessionMetadataSnapshot(agentId, sessionId, 1, ownerA)).toBe(true) + expect(first.store.hasPendingSessionMetadata(ownerA, [agentId])).toBe(false) + expect(first.store.nextSessionMetadataAttemptAt(ownerA, [agentId])).toBeUndefined() + } finally { + await second.close() + await first.close() + } + }) + it('keeps each member on its own runtime model catalog', async () => { // The rollout case against the real schema: two members, two fingerprints, one table. const suffix = randomUUID()