From 314164762b9107d87311fbd2061100db7694df75 Mon Sep 17 00:00:00 2001 From: Sunny Fung Date: Wed, 23 Sep 2026 06:04:47 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20make=20wake=20re-drives=20identifiable?= =?UTF-8?q?=20at=20the=20presentation=20layer=20=E2=80=94=20re-served=20bi?= =?UTF-8?q?t=20+=20original=20message=20id=20(H-64=20leg=20B)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wake scheduler re-drives a silent conversation's row by flipping it back to unprocessed (fireDuePrivateChatA2AWakes), so the re-driven row re-enters the inbound path exactly like a true new peer message: ECDH-ready log, session presentation, renderer emit — with no marker distinguishing a re-service from fresh inbound (the 09-22 triple-display incident: one cast, one DB row, three UI presentations at 0s/+900s/+7200s). Fix (dispatch-point disposition, both sides decided): - fireDuePrivateChatA2AWakes returns a PrivateChatWakeReDriveDisposition map ({ reServed: true, originalRowId }) written at throw time, never retroactively; the poll tick hands each re-driven row its disposition as a new optional processOne parameter. - processOne derives isWakeTurn from the parameter (falling back to the in-memory wake map for rows re-picked on a later tick, e.g. deferred behind a busy runner turn); wake scheduling semantics untouched (900s/7200s ladder, re-arm on silence, budget exhaustion). - When a re-drive actually re-presents the row (session-scoped dedup missed), the inbound bubble carries metadata privateChatReServed + privateChatReServedForMessageId so handler and UI can tell re-service from true inbound; the existing presentation is never retroactively marked; peer protocol surface unchanged (local metadata only, no chain fields). Tests: new tests/privateChatWakeReservedDisposition.test.mjs (harness mirrors privateChatNoReplyWake; discriminating vector proven red on the unfixed build, green on the fix) wired into package.json (test:private-chat-wake-disposition) and .github/workflows/build.yml. Full privateChat cluster: 154 tests / 147 pass / 7 fail — the 7 are pre-existing on main (stale harness stubs in privateChatSkillTurnDeliveryInvariant.test.mjs missing getRecentA2AThreadMessages), fail-set identical before and after the fix. --- .github/workflows/build.yml | 9 + package.json | 1 + src/main/services/privateChatDaemon.ts | 45 +- ...rivateChatWakeReservedDisposition.test.mjs | 422 ++++++++++++++++++ 4 files changed, 471 insertions(+), 6 deletions(-) create mode 100644 tests/privateChatWakeReservedDisposition.test.mjs diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 2b22af57..9b48f98d 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -136,6 +136,15 @@ jobs: - name: Run group-task / skill-assignment / cowork-proxy regression tests run: pnpm run test:release-regressions + # H-64 leg B: a wake re-drive must be identifiable at the presentation + # layer — the re-presented row carries the re-served bit plus the + # original message id, never retroactively, with the wake scheduling + # semantics unchanged. Pins the disposition contract and the + # [NO_REPLY] -> wake -> fired -> re-drive flow (the script runs + # compile:electron first, project convention). + - name: Run private-chat wake re-drive disposition tests + run: pnpm run test:private-chat-wake-disposition + - name: Run memory and team-culture tests run: pnpm run test:memory diff --git a/package.json b/package.json index ea0c9503..5a751831 100644 --- a/package.json +++ b/package.json @@ -32,6 +32,7 @@ "test:chain-history": "pnpm run compile:electron && node --test tests/chainContentHistoryStore.test.mjs tests/chainWriteLedger.test.mjs tests/chainReadLedger.test.mjs tests/contentSummaryService.test.mjs tests/chainHistoryRecallBlocks.test.mjs", "test:release-regressions": "pnpm run test:group-tasks && pnpm run test:skill-assignment && pnpm run test:cowork-proxy && pnpm run test:proposals && pnpm run test:bot-browser && pnpm run test:release-regressions-extra", "test:release-regressions-extra": "pnpm run compile:electron && node --test tests/privateChatNoReplySentinel.test.mjs tests/privateChatClosedReopen.test.mjs tests/coworkDshWarmupStaleSessionRoute.test.mjs tests/p2pRuntimeConfigMode.test.mjs && npx tsx --test tests/coworkSessionBrainProvider.test.ts tests/coworkSessionSortAnchor.test.ts && pnpm run test:simplelog", + "test:private-chat-wake-disposition": "pnpm run compile:electron && node --test tests/privateChatWakeReservedDisposition.test.mjs", "test:workspace-instructions": "node --test tests/workspaceInstructionsBoundary.test.mjs tests/coworkSessionDetailGitBranch.test.mjs", "test:group-tasks": "pnpm run compile:electron && node --test tests/chainIdentifierPrompt.test.mjs tests/groupTaskDaemon.test.mjs tests/groupTaskDaemonProtocol.test.mjs tests/groupTaskCheckpoints.test.mjs tests/groupTaskPlanChange.test.mjs tests/groupTaskRemotePresence.test.mjs tests/groupTaskService.test.mjs tests/groupTaskStaffing.test.mjs tests/groupTaskCandidateSearch.test.mjs tests/groupTaskEntropyP0.test.mjs tests/groupTaskComposerEnterSend.test.mjs tests/groupTaskUtils.test.mjs tests/groupTaskDeliverableParser.test.mjs tests/trackedTaskClosureAgentTools.test.mjs", "test:skill-assignment": "pnpm run compile:electron && node --test tests/skillAssignmentStore.test.mjs tests/metabotManageService.test.mjs tests/metabotManageAgentTools.test.mjs tests/metabotAvatarAgentTool.test.mjs tests/avatarNormalize.test.mjs tests/chatSkillAuthorization.test.mjs tests/skillAgentTools.test.mjs", diff --git a/src/main/services/privateChatDaemon.ts b/src/main/services/privateChatDaemon.ts index ce9d12c5..8083d593 100644 --- a/src/main/services/privateChatDaemon.ts +++ b/src/main/services/privateChatDaemon.ts @@ -379,14 +379,28 @@ function cancelPrivateChatA2AWakesForConversation( emitLog?.(`[PrivateChat] Cancelled pending wake for ${externalConversationId.slice(0, 30)}… (${reason}).`); } -/** Fire due wakes: re-drive their inbound row and let the poll pick it up. */ +/** + * The disposition a fired wake attaches to the row it re-drives: the handler + * and the presentation layer can then tell a re-served old message apart from + * a true new inbound (H-64 leg B). Written at throw time, never retroactively. + */ +export interface PrivateChatWakeReDriveDisposition { + reServed: true; + /** private_chat_messages row id of the original message being re-served. */ + originalRowId: number; +} + +/** Fire due wakes: re-drive their inbound row and let the poll pick it up. + * Returns the re-drive disposition for every row it actually threw back so + * the poll can hand it straight to processOne and the presentation layer. */ function fireDuePrivateChatA2AWakes(deps: { db: Pick; saveDb: SaveDbFn; coworkStore: Pick; emitLog: (msg: string) => void; emitToRenderer?: (channel: string, data: unknown) => void; -}): void { +}): Map { + const reDrivenDispositions = new Map(); const now = Date.now(); for (const [taskKey, wake] of [...privateChatA2AWakes]) { if (wake.running || wake.fireAt > now) continue; @@ -428,6 +442,9 @@ function fireDuePrivateChatA2AWakes(deps: { deps.emitLog(`[PrivateChat] Wake for ${wake.externalConversationId.slice(0, 30)}… failed to re-drive message ${wake.rowId}: ${error instanceof Error ? error.message : String(error)}`); continue; } + // H-64 leg B: the disposition is written when the row is thrown back + // into the pipeline — never retroactively after presentation. + reDrivenDispositions.set(wake.rowId, { reServed: true, originalRowId: wake.rowId }); appendPrivateChatA2AMessage({ coworkStore: deps.coworkStore, sessionId: wake.sessionId, @@ -444,6 +461,7 @@ function fireDuePrivateChatA2AWakes(deps: { }); deps.emitLog(`[PrivateChat] Wake ${wake.fires} fired for ${wake.externalConversationId.slice(0, 30)}…; re-driving message ${wake.rowId}.`); } + return reDrivenDispositions; } export function shouldDeferForBusyRunnerSession( @@ -3651,6 +3669,8 @@ async function processOne( experienceStore?: MetaIDExperienceStore, getMetaIDCognitionPromptBlock?: GetMetaIDCognitionPromptBlockFn, isSessionTurnActive?: (sessionId: string) => boolean, + /** Written by the wake dispatch point at throw time (H-64 leg B). */ + wakeDisposition?: PrivateChatWakeReDriveDisposition, ): Promise { const taskKey = row.pin_id; if (thinkingTasks.has(taskKey)) return; @@ -3660,9 +3680,12 @@ async function processOne( } // A wake turn re-drives a previously processed row whose conversation went // silent without a bye; it re-decides under a host wake notice instead of a - // new peer message. + // new peer message. The disposition arrives as a parameter straight from + // the wake dispatch point; the in-memory wake map stays as the carrier for + // rows re-picked on a later tick (e.g. deferred behind a busy runner turn), + // whose fire happened in an earlier tick. const wakeEntry = privateChatA2AWakes.get(taskKey); - const isWakeTurn = wakeEntry?.running === true; + const isWakeTurn = wakeDisposition?.reServed === true || wakeEntry?.running === true; thinkingTasks.add(taskKey); try { const toGlobalMetaId = (row.to_global_metaid ?? row.to_metaid ?? '').trim(); @@ -5091,6 +5114,14 @@ async function processOne( txId: row.tx_id, pinId: row.pin_id, }), + // H-64 leg B: a re-drive that actually re-presents this row (the + // session-scoped dedup above missed) must not masquerade as a true + // new inbound — carry the wake disposition to the renderer. The + // already-presented original bubble is never retroactively marked. + ...(isWakeTurn ? { + privateChatReServed: true, + privateChatReServedForMessageId: String(wakeDisposition?.originalRowId ?? wakeEntry?.rowId ?? row.id), + } : {}), ...(episodeStarted ? { refreshSessionSummary: true, a2aEpisodeStarted: true, @@ -5840,8 +5871,9 @@ export function startPrivateChatDaemon( const runActiveTickWork = async (): Promise => { try { // Fire due wakes first: a fired wake resets its row to unprocessed so - // the query below re-drives it in this same tick. - fireDuePrivateChatA2AWakes({ + // the query below re-drives it in this same tick. The returned + // dispositions ride along with those rows into processOne. + const reDrivenWakeDispositions = fireDuePrivateChatA2AWakes({ db, saveDb, coworkStore, @@ -5890,6 +5922,7 @@ export function startPrivateChatDaemon( experienceStore, getMetaIDCognitionPromptBlock, isSessionTurnActive, + reDrivenWakeDispositions.get(row.id), ); } catch (e) { console.error('[PrivateChat] processOne error:', e); diff --git a/tests/privateChatWakeReservedDisposition.test.mjs b/tests/privateChatWakeReservedDisposition.test.mjs new file mode 100644 index 00000000..f399e7de --- /dev/null +++ b/tests/privateChatWakeReservedDisposition.test.mjs @@ -0,0 +1,422 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import crypto from 'node:crypto'; + +// H-64 腿 B (wake re-drive identifiability): the wake scheduler re-drives a +// silent conversation's row by flipping it back to unprocessed. The re-driven +// row then re-enters the inbound path (ECDH ready + presentation) exactly like +// a true new peer message. The fix hands a wake disposition — the ORIGINAL +// message id + a re-served bit — from the dispatch point down to the +// presentation layer, so handler and UI can tell "re-served old message" +// apart from "true new inbound". These tests pin that contract: +// 1. a re-drive that actually re-presents the row carries +// privateChatReServed + privateChatReServedForMessageId; +// 2. a true new inbound never carries them; +// 3. the existing presentation is never retroactively marked; +// 4. the [NO_REPLY] -> wake -> fired -> re-drive flow itself is unchanged. + +let startPrivateChatDaemon; +let stopPrivateChatDaemon; +let buildPrivateChatA2AWakeNotice; +let nextPrivateChatA2AWakeAt; +let setPrivateChatA2AWakeDelaysForTests; +let PRIVATE_CHAT_A2A_DEFAULT_WAKE_DELAYS_MS; +({ + startPrivateChatDaemon, + stopPrivateChatDaemon, + buildPrivateChatA2AWakeNotice, + nextPrivateChatA2AWakeAt, + setPrivateChatA2AWakeDelaysForTests, + PRIVATE_CHAT_A2A_DEFAULT_WAKE_DELAYS_MS, +} = await import('../dist-electron/main/services/privateChatDaemon.js')); + +const TEST_MNEMONIC = 'abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about'; + +function createPeerPublicKey() { + const ecdh = crypto.createECDH('prime256v1'); + ecdh.generateKeys(); + return ecdh.getPublicKey('hex', 'uncompressed'); +} + +function createWakeDispositionHarness() { + // The row id mirrors the incident log (`Wake 1 fired … re-driving message + // 21856`): the disposition must carry THIS id to the presentation layer. + const row = { + id: 21856, + pin_id: 'incoming-pin-wake-disposition-1', + tx_id: 'b'.repeat(64), + from_metaid: 'peer-metaid', + from_global_metaid: 'peer-global', + from_name: 'Peer Bot', + from_avatar: null, + from_chat_pubkey: createPeerPublicKey(), + to_metaid: 'local-metaid', + to_global_metaid: 'local-global', + content: 'The verification anchors are ready — I will wait silently for your complete reply.', + encryption: null, + reply_pin: '', + raw_data: null, + is_processed: 0, + }; + const columns = [ + 'id', 'pin_id', 'tx_id', 'from_metaid', 'from_global_metaid', 'from_name', + 'from_avatar', 'from_chat_pubkey', 'to_metaid', 'to_global_metaid', + 'content', 'encryption', 'reply_pin', 'raw_data', + ]; + const externalConversationId = 'metaweb-private:peer-global'; + const session = { + id: 'session-private-1', + sessionType: 'a2a', + metabotId: 1, + peerGlobalMetaId: 'peer-global', + messages: [ + // Established conversation: a prior delivered outbound so the daemon + // treats this peer as one we are actively talking to (wake arming + // requires prior local outbound). + { + id: 'msg-seed-outbound', + timestamp: 1_770_000_000_000, + type: 'assistant', + content: 'Received — give me a moment to verify these points and I will come back with the complete reply.', + metadata: { + sourceChannel: 'metaweb_private', + externalConversationId, + direction: 'outgoing', + privateChatDeliveryStatus: 'sent', + }, + }, + ], + }; + const mapping = { + channel: 'metaweb_private', + externalConversationId, + metabotId: 1, + coworkSessionId: session.id, + metadataJson: JSON.stringify({ peerGlobalMetaId: 'peer-global' }), + }; + const metabot = { + id: 1, + name: 'Local Bot', + enabled: true, + metaid: 'local-metaid', + globalmetaid: 'local-global', + allow_chat_skills: ['metaid-master-wiki'], + }; + const db = { + exec(sql) { + if (/FROM private_chat_messages WHERE is_processed = 0/i.test(sql)) { + return row.is_processed + ? [] + : [{ columns, values: [columns.map((column) => row[column])] }]; + } + return [{ columns: ['found'], values: [] }]; + }, + run(sql, params) { + if (/UPDATE private_chat_messages SET is_processed = 1 WHERE id = \?/i.test(sql)) { + row.is_processed = 1; + assert.deepEqual(params, [row.id]); + } else if (/UPDATE private_chat_messages SET is_processed = 0 WHERE id = \?/i.test(sql)) { + row.is_processed = 0; + assert.deepEqual(params, [row.id]); + } + }, + }; + const coworkStore = { + getConversationMapping(channel, conversationId, metabotId) { + return channel === 'metaweb_private' && conversationId === externalConversationId && metabotId === 1 + ? mapping + : null; + }, + getSession(sessionId) { + return sessionId === session.id ? session : null; + }, + getSessionWithoutMessages(sessionId) { + return sessionId === session.id ? session : null; + }, + getSessionMessagesMatchingMetadataValues(sessionId, values, limit = 50) { + const needles = (Array.isArray(values) ? values : [values]) + .map((value) => String(value ?? '').trim()) + .filter(Boolean); + if (needles.length === 0) return []; + return session.messages + .filter((message) => needles.some((needle) => JSON.stringify(message.metadata ?? {}).includes(needle))) + .reverse() + .slice(0, limit); + }, + getRecentA2AThreadMessages(sessionId, limit = 400) { + return this.getRecentPrivateA2AMessages(sessionId, limit); + }, + listA2AConversationEpisodes() { + return []; + }, + getRecentPrivateA2AMessages(sessionId, requestedLimit = 100) { + const limit = Number.isFinite(requestedLimit) + ? Math.max(1, Math.min(1000, Math.floor(requestedLimit))) + : 100; + return session.messages + .filter((message) => ( + (message.type === 'user' || message.type === 'assistant') + && message.metadata?.sourceChannel === 'metaweb_private' + && message.metadata?.orderExecutionTrace !== true + )) + .slice(-limit); + }, + getMessageById(sessionId, messageId) { + return session.messages.find((message) => message.id === messageId) ?? null; + }, + getConversationSourceContextBySession() { + return { sourceChannel: 'metaweb_private', externalConversationId }; + }, + isSessionArchived() { + return false; + }, + unarchiveSession() {}, + registerA2AEpisode() {}, + isDelegationBlocking() { + return false; + }, + setDelegationBlocking() {}, + updateSession(sessionId, updates) { + if (sessionId === session.id) Object.assign(session, updates); + }, + upsertConversationMapping() {}, + touchConversationMapping() {}, + deleteConversationMapping() {}, + createSession() { + return { id: 'session-created-1', messages: [] }; + }, + findOrderSessionByOrderPinId() { + return null; + }, + findOrderSessionByOrderTxid() { + return null; + }, + findOrderSessionByPeer() { + return null; + }, + hasPriorPrivateA2AOutboundMessage(sessionId) { + return session.messages.some((message) => ( + message.type === 'assistant' + && message.metadata?.sourceChannel === 'metaweb_private' + && String(message.content ?? '').trim() !== '' + )); + }, + ensureCanonicalPeerSessionShape() { + return true; + }, + addMessage(sessionId, message) { + const created = { + id: `msg-${session.messages.length + 1}`, + timestamp: 1_770_000_000_000 + session.messages.length, + ...message, + }; + session.messages.push(created); + return created; + }, + updateMessage(sessionId, messageId, updates) { + const message = session.messages.find((item) => item.id === messageId); + if (message) Object.assign(message, updates); + }, + updateConversationMappingMetadata() {}, + getConfig() { + return { workingDirectory: '/tmp/idbots-test' }; + }, + getMemoryBackend() { + return { + getEffectiveMemoryPolicyForMetabot() { + return { memoryEnabled: false }; + }, + }; + }, + }; + const metabotStore = { + getMetabotByGlobalMetaId(globalMetaId) { + return globalMetaId === metabot.globalmetaid ? metabot : null; + }, + getMetabotById() { + return null; + }, + getMetabotWalletByMetabotId(id) { + assert.equal(id, metabot.id); + return { mnemonic: TEST_MNEMONIC, path: "m/44'/10001'/0'/0/0" }; + }, + }; + return { + db, + row, + coworkStore, + metabotStore, + metabot, + session, + externalConversationId, + inboundBubbles: () => session.messages.filter((message) => ( + message.type === 'user' + && message.metadata?.sourceChannel === 'metaweb_private' + )), + }; +} + +async function waitFor(predicate, timeoutMs = 15_000) { + const startedAt = Date.now(); + while (Date.now() - startedAt < timeoutMs) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + assert.fail('timed out waiting for private chat daemon test condition'); +} + +function startDispositionHarnessDaemon(harness, logs, skillTurnImpl) { + let createPinCount = 0; + const skillTurnCalls = []; + startPrivateChatDaemon( + harness.db, + () => {}, + harness.coworkStore, + harness.metabotStore, + { on() {}, off() {} }, + async () => { + createPinCount += 1; + return { txids: ['t'.repeat(64)], pinId: 'p'.repeat(64) + 'i0' }; + }, + (message) => logs.push(message), + null, + undefined, + undefined, + () => ({ respondToStrangerPrivateChats: true }), + undefined, + undefined, + undefined, + async () => ({ + prompt: 'metaid-master-wiki', + activeSkillIds: ['metaid-master-wiki'], + }), + async (params) => { + skillTurnCalls.push(params); + return skillTurnImpl(params, skillTurnCalls.length, harness.coworkStore); + }, + async () => '我需要查询一下，请稍等。' + ); + return { + get createPinCount() { + return createPinCount; + }, + skillTurnCalls, + }; +} + +// Full flow: true new inbound -> silent [NO_REPLY] -> wake armed -> wake fires +// and re-drives the row -> the re-drive re-presents the row (the session no +// longer holds the original bubble — the session-scoped dedup guard misses, +// exactly the production triple-display precondition) -> the re-presented +// inbound carries the wake disposition (re-served bit + original message id). +test('wake re-drive presents the re-served row with the original message id; a true new inbound does not carry it', async () => { + setPrivateChatA2AWakeDelaysForTests([300]); + const harness = createWakeDispositionHarness(); + const logs = []; + let firstInboundBubble = null; + const handle = startDispositionHarnessDaemon(harness, logs, (params, call, coworkStore) => { + if (call === 1) { + coworkStore.addMessage(params.sessionId, { + type: 'assistant', + content: '[NO_REPLY]', + metadata: { isStreaming: false, isFinal: true }, + }); + return { replyText: '[NO_REPLY]', assistantMessageId: null }; + } + const persisted = coworkStore.addMessage(params.sessionId, { + type: 'assistant', + content: 'Verification complete — here is the full reply I owed you.', + metadata: { isStreaming: false, isFinal: true }, + }); + return { replyText: 'Verification complete — here is the full reply I owed you.', assistantMessageId: persisted.id }; + }); + + try { + // True new inbound: the first presentation must look like one. + await waitFor(() => harness.inboundBubbles().length === 1, 20_000); + firstInboundBubble = { ...harness.inboundBubbles()[0], metadata: { ...harness.inboundBubbles()[0].metadata } }; + await waitFor(() => logs.some((message) => message.includes('Wake 1 scheduled')), 20_000); + // Simulate the session-scoped dedup miss (episode rollover / mapping + // repair put the conversation tail into a different session shape): the + // re-drive will no longer find the original bubble and will re-present. + harness.session.messages = harness.session.messages.filter( + (message) => message.id !== firstInboundBubble.id + ); + await waitFor(() => logs.some((message) => message.includes('Wake 1 fired')), 20_000); + await waitFor(() => logs.some((message) => message.includes('Replied to')), 20_000); + // The delivered reply ends the episode: no further turns may run. + await new Promise((resolve) => setTimeout(resolve, 7_000)); + } finally { + await stopPrivateChatDaemon({ waitForTick: true }); + setPrivateChatA2AWakeDelaysForTests(PRIVATE_CHAT_A2A_DEFAULT_WAKE_DELAYS_MS); + } + + // True new inbound: no disposition bits. + assert.ok(firstInboundBubble, 'the original inbound bubble was captured'); + assert.notEqual(firstInboundBubble.metadata?.privateChatReServed, true, 'a true new inbound must not carry the re-served bit'); + assert.equal(firstInboundBubble.metadata?.privateChatReServedForMessageId, undefined); + + // Exactly one re-presented inbound bubble, carrying the disposition. + const rePresented = harness.inboundBubbles(); + assert.equal(rePresented.length, 1, 'the re-drive re-presented exactly one inbound bubble'); + assert.equal(rePresented[0].content, harness.row.content); + assert.equal(rePresented[0].metadata?.privateChatReServed, true, 'the re-served bit must reach the presentation layer'); + assert.equal( + rePresented[0].metadata?.privateChatReServedForMessageId, + String(harness.row.id), + 'the original private_chat_messages row id must ride along with the re-served bit' + ); + + // The wake turn itself is unchanged: wake notice in the system prompt. + assert.equal(handle.skillTurnCalls.length, 2, 'exactly one wake turn should follow the silent decision'); + assert.ok(!handle.skillTurnCalls[0].systemPrompt.includes('Host Wake Check'), 'original turn has no wake notice'); + assert.ok(handle.skillTurnCalls[1].systemPrompt.includes('Host Wake Check'), 'wake turn carries the wake notice'); + assert.equal(harness.row.is_processed, 1); + assert.equal(handle.createPinCount, 1, 'the owed reply is delivered on-chain'); + const delivered = harness.session.messages.find( + (message) => message.content === 'Verification complete — here is the full reply I owed you.' + ); + assert.equal(delivered?.metadata?.privateChatDeliveryStatus, 'sent'); +}); + +// The decided fix forbids retroactive marking: when the original bubble is +// still in the session, the dedup guard suppresses the re-presentation and the +// EXISTING bubble must stay untouched. The silent re-drive also still re-arms +// the wake ladder (900s/7200s semantics unchanged, shrunk here for tests). +test('wake re-drive never retroactively marks the existing presentation and still re-arms while silent', async () => { + setPrivateChatA2AWakeDelaysForTests([300, 300]); + const harness = createWakeDispositionHarness(); + const logs = []; + const handle = startDispositionHarnessDaemon(harness, logs, (params, call, coworkStore) => { + const persisted = coworkStore.addMessage(params.sessionId, { + type: 'assistant', + content: '[NO_REPLY]', + metadata: { isStreaming: false, isFinal: true }, + }); + return { replyText: '[NO_REPLY]', assistantMessageId: persisted.id }; + }); + + try { + await waitFor(() => logs.some((message) => message.includes('Wake budget exhausted')), 30_000); + // No further wake may fire after exhaustion. + await new Promise((resolve) => setTimeout(resolve, 7_000)); + } finally { + await stopPrivateChatDaemon({ waitForTick: true }); + setPrivateChatA2AWakeDelaysForTests(PRIVATE_CHAT_A2A_DEFAULT_WAKE_DELAYS_MS); + } + + // The single original inbound bubble is never duplicated nor re-marked. + const bubbles = harness.inboundBubbles(); + assert.equal(bubbles.length, 1, 'the dedup guard keeps the re-drive from duplicating the inbound bubble'); + assert.notEqual(bubbles[0].metadata?.privateChatReServed, true, 'the original presentation must not be retroactively marked'); + assert.equal(bubbles[0].metadata?.privateChatReServedForMessageId, undefined); + + // Silent re-drive chain unchanged: original turn + 2 wake fires, re-arm + // ladder walked to exhaustion, nothing delivered on-chain. + assert.equal(handle.skillTurnCalls.length, 3, 'original turn + 2 wake fires, then stop'); + const wakeTurns = handle.skillTurnCalls.filter((params) => params.systemPrompt.includes('Host Wake Check')); + assert.equal(wakeTurns.length, 2); + assert.ok(logs.some((message) => message.includes('Wake 2 scheduled')), 'a silent wake turn still re-arms the ladder'); + assert.equal(harness.row.is_processed, 1); + assert.equal(handle.createPinCount, 0, 'a permanently silent conversation delivers nothing on-chain'); +}); base-commit: 2071265da0316f64a870c83b1327cccb32802801 -- 2.53.0