diff --git a/services/platform/backend/domains/chat/store.test.ts b/services/platform/backend/domains/chat/store.test.ts index 5a9c68ec89..133a3202e0 100644 --- a/services/platform/backend/domains/chat/store.test.ts +++ b/services/platform/backend/domains/chat/store.test.ts @@ -257,10 +257,6 @@ describe('createPgTurnStore.appendMessage', () => { }); expect(appended.id).toBe('msg_1'); - // No generation row ever opens for a pre-model refusal, so without this - // NOTIFY the other viewers of the thread learn of the two rows only on a - // later invalidation. - expect(f.notified).toEqual(['chat_stream:thread_1']); }); }); diff --git a/services/platform/backend/domains/chat/store.ts b/services/platform/backend/domains/chat/store.ts index 111f04173e..3df517dcb0 100644 --- a/services/platform/backend/domains/chat/store.ts +++ b/services/platform/backend/domains/chat/store.ts @@ -7,7 +7,6 @@ import { type UsageLedger, type UsageLedgerEntry, } from '../../../lib/chat/turn.ts'; -import { settleDeferredSendOnUserAppend } from '../../core/chat/turn_store.ts'; import { getProviderCatalog } from '../../core/lib/providers/catalog_fetch.ts'; import { resolveProvidersForOrg } from '../../core/lib/providers/org_providers.ts'; import { toJson } from '../../db/sql.ts'; @@ -206,6 +205,37 @@ async function finalizeWithStreamedTail( }); } +/** + * Decorate a turn store so a deferred send's row dies the moment the turn + * persists the user message — the turn-open write, when it carries the user + * parts. Until that write the row is the parked message's only + * representation (the tray above the composer); from it on the thread shows + * the bubble, and a row that survived to the action's terminal settle would + * double-display the message for the whole generation. A settle failure is + * logged, never fatal — the terminal settle in the action retries it. + * (Re-homed from the retired `core/chat/turn_store.ts`: the PG store is its + * only consumer.) + */ +function settleDeferredSendOnUserAppend( + store: TurnStore, + settle: () => Promise, +): TurnStore { + return { + ...store, + async beginTurn(setup) { + const opened = await store.beginTurn(setup); + if (setup.userParts !== undefined) { + try { + await settle(); + } catch (error) { + console.warn('Deferred send settle at user append failed:', error); + } + } + return opened; + }, + }; +} + /** A turn store over app.messages + app.generations. With * `onUserMessageAppended` the store is decorated to run it the moment * `beginTurn` persisted the user row — the deferred-send lane settles its @@ -231,10 +261,6 @@ function pgTurnStore(sql: Sql): TurnStore { .map((part) => (part.type === 'text' ? part.text : '')) .join(''), }); - // A pre-model refusal lands its two rows through this write alone (no - // generation row ever opens), so this is the only signal the thread's - // other viewers get that the transcript moved. - await notifyThread(sql, message.threadId); return appended; }, diff --git a/services/platform/backend/domains/members/service.test.ts b/services/platform/backend/domains/members/service.test.ts index fe878f9f26..9b0415c8ed 100644 --- a/services/platform/backend/domains/members/service.test.ts +++ b/services/platform/backend/domains/members/service.test.ts @@ -77,7 +77,10 @@ function createRecordingTx(scenario: Scenario): { if (text.startsWith('DELETE FROM "passkey"')) { return scenario.passkeyDeleteReturns ?? [{ id: 'pk-1' }]; } - if (text.startsWith('INSERT INTO app.audit_chain_heads')) { + if ( + text.startsWith('SELECT pg_advisory_xact_lock(') || + text.startsWith('INSERT INTO app.audit_chain_heads') + ) { return []; } if (text.includes('FROM app.audit_chain_heads')) { diff --git a/services/platform/backend/domains/tasks/reattach.stalled.test.ts b/services/platform/backend/domains/tasks/reattach.stalled.test.ts index 80cc11afc1..3bb96e0975 100644 --- a/services/platform/backend/domains/tasks/reattach.stalled.test.ts +++ b/services/platform/backend/domains/tasks/reattach.stalled.test.ts @@ -67,7 +67,6 @@ describe('recoverStalledTaskAgentTurns — the op-less arm ages on the run row', * `integration-check.ts`. */ - interface Statement { text: string; values: unknown[]; diff --git a/services/platform/backend/domains/users/create-member.test.ts b/services/platform/backend/domains/users/create-member.test.ts index 1076a7b356..e1e9d090cb 100644 --- a/services/platform/backend/domains/users/create-member.test.ts +++ b/services/platform/backend/domains/users/create-member.test.ts @@ -77,7 +77,10 @@ function createRecordingSql(scenario: Scenario): { if (text.startsWith('INSERT INTO "member"')) { return [{ id: 'member-new' }]; } - if (text.startsWith('INSERT INTO app.audit_chain_heads')) { + if ( + text.startsWith('SELECT pg_advisory_xact_lock(') || + text.startsWith('INSERT INTO app.audit_chain_heads') + ) { return []; } if (text.includes('FROM app.audit_chain_heads')) {