From 683289d853495cbe44096eaebedbbdc44a967111 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 12 Nov 2026 19:57:37 -0800 Subject: [PATCH] fix(orchestration): scope sleeping mail invalidation to affected owners --- .../agent-session-crew-invalidation.ts | 52 ++++ .../agent-session-record-store-slot.ts | 8 +- .../agent-session-store-transactions.ts | 32 +-- .../orchestration/crew/crew-changes.ts | 2 + .../crew/crew-mail-mutation-scope.ts | 235 ++++++++++++++++++ .../crew/crew-mutation-notifications.ts | 43 +++- .../db/legacy/legacy-worker-completion.ts | 11 +- .../db/messages/message-inbox.ts | 3 + .../messages/message-mutation-invalidation.ts | 54 ++++ .../db/runs/run-coordinator-mail-routing.ts | 6 +- .../routing/runtime-mail-wake-admission.ts | 39 +-- .../routing/sleeping-mail-owner-mailboxes.ts | 74 ++++++ .../mailbox/routing/slept-mailbox-runtime.ts | 40 +-- .../routing/slept-mailbox-wake.test.ts | 200 +++++++++++---- .../mailbox/routing/slept-mailbox-wake.ts | 28 ++- .../persisted-surfaces/crew-operations.ts | 4 +- 16 files changed, 706 insertions(+), 125 deletions(-) create mode 100644 src/main/runtime/agent-session/records/agent-session-crew-invalidation.ts create mode 100644 src/main/runtime/orchestration/crew/crew-mail-mutation-scope.ts create mode 100644 src/main/runtime/orchestration/db/messages/message-mutation-invalidation.ts create mode 100644 src/main/runtime/orchestration/mailbox/routing/sleeping-mail-owner-mailboxes.ts diff --git a/src/main/runtime/agent-session/records/agent-session-crew-invalidation.ts b/src/main/runtime/agent-session/records/agent-session-crew-invalidation.ts new file mode 100644 index 00000000000..b58614f7e1d --- /dev/null +++ b/src/main/runtime/agent-session/records/agent-session-crew-invalidation.ts @@ -0,0 +1,52 @@ +import { sessionCrewChanges } from '../../orchestration/crew/crew-changes' +import type { AgentSessionStoreState } from './agent-session-store-state' +import type { AgentSessionStoreRowWrites } from './agent-session-store-draft' + +export function publishAgentSessionCrewChanges( + published: AgentSessionStoreState, + draft: AgentSessionStoreState, + writes: AgentSessionStoreRowWrites +): void { + const topologyChanged = + writes.sessionTabs !== null || + writes.records.remove.length > 0 || + writes.records.upsert.some(([id]) => { + const before = published.records.get(id) + const after = draft.records.get(id) + return ( + !before || + !after || + before.lease.claimStatus !== after.lease.claimStatus || + before.provider !== after.provider || + before.location !== after.location || + before.conversationCommand !== after.conversationCommand + ) + }) + const renamed = writes.records.upsert + .filter( + ([id]) => + published.records.get(id)?.conversationName !== draft.records.get(id)?.conversationName + ) + .map(([id]) => id) + if (topologyChanged) { + const affected = new Set([...writes.records.remove, ...writes.records.upsert.map(([id]) => id)]) + if (writes.sessionTabs) { + for (const id of [ + ...(published.sessionTabs?.sessionIds() ?? []), + ...(draft.sessionTabs?.sessionIds() ?? []) + ]) { + if (published.sessionTabs?.tabIdFor(id) !== draft.sessionTabs?.tabIdFor(id)) { + affected.add(id) + } + } + } + sessionCrewChanges.notify( + Object.assign(new Set(['records'] as const), { + recordIds: [...affected], + trafficAddresses: new Set([...affected].map((id) => `orca_session_id:${id}`)) + }) + ) + } else if (renamed.length) { + sessionCrewChanges.notify(Object.assign(new Set(['names'] as const), { recordIds: renamed })) + } +} diff --git a/src/main/runtime/agent-session/records/agent-session-record-store-slot.ts b/src/main/runtime/agent-session/records/agent-session-record-store-slot.ts index 6ecdfcaa76c..e2d15c7b397 100644 --- a/src/main/runtime/agent-session/records/agent-session-record-store-slot.ts +++ b/src/main/runtime/agent-session/records/agent-session-record-store-slot.ts @@ -61,7 +61,9 @@ export function openAgentSessionRecordStoreOnce( (opened) => { slot.store = opened.store if (opening === slot) { - sessionCrewChanges.notify(new Set(['topology', 'records'])) + sessionCrewChanges.notify( + Object.assign(new Set(['topology', 'records'] as const), { mailGlobal: true as const }) + ) } return opened }, @@ -107,7 +109,9 @@ export async function releaseAgentSessionRecordStore(): Promise null) : null } diff --git a/src/main/runtime/agent-session/records/agent-session-store-transactions.ts b/src/main/runtime/agent-session/records/agent-session-store-transactions.ts index c01cd38dc0f..a5069896ce1 100644 --- a/src/main/runtime/agent-session/records/agent-session-store-transactions.ts +++ b/src/main/runtime/agent-session/records/agent-session-store-transactions.ts @@ -1,4 +1,4 @@ -import { sessionCrewChanges } from '../../orchestration/crew/crew-changes' +import { publishAgentSessionCrewChanges } from './agent-session-crew-invalidation' import type { JournalHostDatabase } from '../../../native-chat/agent-session-journal/database/journal-host-database' import type { JournalOperationReceipt } from '../../../native-chat/agent-session-journal/writing/journal-row-writer' import { journalOpenRefusalError } from '../../../native-chat/agent-session-journal/replay/journal-open-failure' @@ -263,35 +263,7 @@ export class AgentSessionStoreTransactions { if (lineageChanged) { this.lineage = {} } - const topologyChanged = - writes.sessionTabs !== null || - writes.records.remove.length > 0 || - writes.records.upsert.some(([id]) => { - const before = published.records.get(id) - const after = draft.records.get(id) - return ( - !before || - !after || - before.lease.claimStatus !== after.lease.claimStatus || - before.provider !== after.provider || - before.location !== after.location || - before.conversationCommand !== after.conversationCommand - ) - }) - const renamed = writes.records.upsert - .filter( - ([id]) => - published.records.get(id)?.conversationName !== - draft.records.get(id)?.conversationName - ) - .map(([id]) => id) - if (topologyChanged) { - sessionCrewChanges.notify(new Set(['records'])) - } else if (renamed.length) { - sessionCrewChanges.notify( - Object.assign(new Set(['names'] as const), { recordIds: renamed }) - ) - } + publishAgentSessionCrewChanges(published, draft, writes) } } } diff --git a/src/main/runtime/orchestration/crew/crew-changes.ts b/src/main/runtime/orchestration/crew/crew-changes.ts index 79a92adff27..5f2eb9ce8c1 100644 --- a/src/main/runtime/orchestration/crew/crew-changes.ts +++ b/src/main/runtime/orchestration/crew/crew-changes.ts @@ -1,6 +1,8 @@ export type CrewFacet = 'topology' | 'records' | 'status' | 'names' | 'traffic' export type CrewInvalidation = ReadonlySet & { readonly recordIds?: readonly string[] + readonly mailboxOwners?: ReadonlySet + readonly mailGlobal?: true readonly trafficAddresses?: ReadonlySet } diff --git a/src/main/runtime/orchestration/crew/crew-mail-mutation-scope.ts b/src/main/runtime/orchestration/crew/crew-mail-mutation-scope.ts new file mode 100644 index 00000000000..5cdd7fd817b --- /dev/null +++ b/src/main/runtime/orchestration/crew/crew-mail-mutation-scope.ts @@ -0,0 +1,235 @@ +import { isOrcaSessionId } from '../../../../shared/orca-session-address' +import type { OrchestrationDb } from '../db/orchestration-db' + +const dispatchMutations = new Set([ + 'settleRemoteAttachmentInRelayTransaction', + 'completeDispatch', + 'failDispatch', + 'closeQuestionsForDispatch', + 'markWorkerDispatchReady', + 'failWorkerStart', + 'markWorkerStartUnknown', + 'abandonWorkerDispatch', + 'beginWorkerStop', + 'settleWorkerStop', + 'markWorkerStopUnknown', + 'reconcileMissingWorkerTerminal', + 'failRemoteAttachment', + 'markRemoteAttachmentReady', + 'beginRemoteAttachmentStop', + 'settleRemoteAttachmentStop', + 'markRemoteAttachmentStopUnknown', + 'requestRemoteAttachmentTerminalRelease' +]) +const taskMutations = new Set([ + 'completeActiveDispatchesForTask', + 'failActiveDispatchForTask', + 'promoteReadyTasks', + 'updateTaskStatus' +]) + +function fields(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +/** Capture both sides of transfers; nested mutations contribute their own scope before publication. */ +export class CrewMailMutationScope { + readonly trafficAddresses = new Set() + readonly mailboxOwners = new Set() + private readonly visited = new Set() + + get notice() { + return { trafficAddresses: this.trafficAddresses, mailboxOwners: this.mailboxOwners } + } + + constructor(private readonly db: OrchestrationDb) {} + + run(id: string): void { + this.trafficAddresses.add(`run:${id}`) + if (this.seen(`run:${id}`)) { + return + } + this.row(this.db.getRun(id)) + } + + dispatch(id: string): void { + this.trafficAddresses.add(`dispatch:${id}`) + if (this.seen(`dispatch:${id}`)) { + return + } + this.row(this.db.getDispatchContextById(id) ?? this.db.getRemoteDispatchAttachment(id)) + } + + private seen(id: string): boolean { + if (this.visited.has(id)) { + return true + } + this.visited.add(id) + return false + } + + private task(id: string): void { + if (this.seen(`task:${id}`)) { + return + } + this.row(this.db.getTask(id)) + for (const row of this.db.db + .prepare('SELECT id FROM dispatch_contexts WHERE task_id = ?') + .all(id)) { + if (typeof row.id === 'string') { + this.dispatch(row.id) + } + } + } + + row(value: unknown): void { + if (Array.isArray(value)) { + for (const row of value) { + this.row(row) + } + } else if (fields(value)) { + for (const [key, item] of Object.entries(value)) { + if (typeof item === 'string') { + if (['runId', 'run_id', 'local_run_id'].includes(key)) { + this.run(item) + } + if ( + [ + 'dispatchId', + 'dispatch_id', + 'owner_dispatch_id', + 'toDispatchId', + 'creatorDispatchId' + ].includes(key) + ) { + this.dispatch(item) + } + if (['taskId', 'task_id'].includes(key)) { + this.task(item) + } + if ( + [ + 'paneKey', + 'pane_key', + 'fromPaneKey', + 'toPaneKey', + 'assigneePaneKey', + 'assignee_pane_key', + 'coordinatorPaneKey', + 'coordinator_pane_key', + 'creator_pane_key' + ].includes(key) + ) { + this.mailboxOwners.add(item) + } + if ( + [ + 'mailboxHandle', + 'from', + 'to', + 'from_handle', + 'to_handle', + 'assigneeHandle', + 'assignee_handle', + 'coordinatorHandle', + 'coordinator_handle', + 'terminalHandle', + 'terminal_handle' + ].includes(key) + ) { + this.trafficAddresses.add(item) + } + if ( + [ + 'coordinator_orca_session_id', + 'assignee_orca_session_id', + 'creator_orca_session_id' + ].includes(key) + ) { + this.trafficAddresses.add(`orca_session_id:${item}`) + } + if (['messageId', 'message_id'].includes(key)) { + this.row(this.db.getMessageById(item)) + } + if (key === 'resourceId') { + this.row(this.db.getWorkerTerminalResource(item)) + } + } else if (item && typeof item === 'object') { + this.row(item) + } + } + if (typeof value.id === 'string') { + if ('coordinator_pane_key' in value) { + this.trafficAddresses.add(`run:${value.id}`) + } + if ('assignee_pane_key' in value) { + this.trafficAddresses.add(`dispatch:${value.id}`) + } + } + } + } + + capture(name: string, args: unknown[]): void { + const first = args[0] + this.row(args) + if (typeof first === 'string') { + if (dispatchMutations.has(name)) { + this.dispatch(first) + } + if (taskMutations.has(name)) { + this.task(first) + } + if (name === 'resolveGate' || name === 'timeoutGate') { + this.row( + this.db.db.prepare('SELECT run_id, task_id FROM decision_gates WHERE id = ?').get(first) + ) + } + if (name === 'convertLifecycleMessageToRejection') { + this.row(this.db.getMessageById(first)) + } + if (name === 'settleWorkerTerminalRelease') { + this.row(this.db.getWorkerTerminalResource(first)) + } + if (name === 'routeDirectMessagePage') { + this.trafficAddresses.add(first) + if (typeof args[1] === 'string') { + this.run(args[1]) + } + if (typeof args[2] === 'string') { + this.trafficAddresses.add(args[2]) + } + } + if (name === 'routeUnreadDispatchMailboxToRunMailbox') { + this.dispatch(first) + if (typeof args[1] === 'string') { + this.run(args[1]) + } + } + if (name === 'routeForeignDirectMessagesToOwnedMailboxes') { + this.trafficAddresses.add(first) + } + } + if (name === 'unbindOtherRunsForCoordinator' && fields(first)) { + const { paneKey, orcaSessionId, terminalHandle } = first + if ( + (paneKey === null || typeof paneKey === 'string') && + (orcaSessionId === null || + (typeof orcaSessionId === 'string' && isOrcaSessionId(orcaSessionId))) && + (terminalHandle === null || typeof terminalHandle === 'string') + ) { + this.row(this.db.runsBoundToCoordinator({ paneKey, orcaSessionId, terminalHandle })) + } + } + if (name === 'transitionLifecycle' && fields(first) && typeof first.id === 'string') { + if (first.entity === 'run') { + this.run(first.id) + } + if (first.entity === 'dispatch' || first.entity === 'worker') { + this.dispatch(first.id) + } + if (first.entity === 'task') { + this.task(first.id) + } + } + } +} diff --git a/src/main/runtime/orchestration/crew/crew-mutation-notifications.ts b/src/main/runtime/orchestration/crew/crew-mutation-notifications.ts index 4e05720c5ee..72aa63f6892 100644 --- a/src/main/runtime/orchestration/crew/crew-mutation-notifications.ts +++ b/src/main/runtime/orchestration/crew/crew-mutation-notifications.ts @@ -1,3 +1,4 @@ +import { CrewMailMutationScope } from './crew-mail-mutation-scope' import type { CrewInvalidation } from './crew-changes' import type { OrchestrationDb } from '../db/orchestration-db' @@ -89,10 +90,13 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP if (hasDomainMethod(prototype, 'settleWorkerTerminalRelease')) { const release = prototype.settleWorkerTerminalRelease prototype.settleWorkerTerminalRelease = function (this: OrchestrationDb, resourceId) { + const scope = new CrewMailMutationScope(this) + scope.capture('settleWorkerTerminalRelease', [resourceId]) const before = this.getWorkerTerminalResource(resourceId)?.release_state const result = release.call(this, resourceId) if (before !== result.release_state) { - this.crewChanges.notify(new Set(['topology'])) + scope.row(result) + this.crewChanges.notify(Object.assign(new Set(['topology'] as const), scope.notice)) } return result } @@ -100,10 +104,13 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP if (hasDomainMethod(prototype, 'settleDeadWorkerTerminalRelease')) { const release = prototype.settleDeadWorkerTerminalRelease prototype.settleDeadWorkerTerminalRelease = function (this: OrchestrationDb, params) { + const scope = new CrewMailMutationScope(this) + scope.row(params) const before = this.getWorkerTerminalResource(params.resourceId)?.release_state const result = release.call(this, params) if (before !== result.resource.release_state) { - this.crewChanges.notify(new Set(['topology'])) + scope.row(result) + this.crewChanges.notify(Object.assign(new Set(['topology'] as const), scope.notice)) } return result } @@ -111,9 +118,12 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP if (hasDomainMethod(prototype, 'commitLegacyCompatibilityPrincipal')) { const commit = prototype.commitLegacyCompatibilityPrincipal prototype.commitLegacyCompatibilityPrincipal = function (this: OrchestrationDb, params) { + const scope = new CrewMailMutationScope(this) + scope.row(params) const result = commit.call(this, params) if (!result.duplicate) { - this.crewChanges.notify(new Set(['topology'])) + scope.row(result) + this.crewChanges.notify(Object.assign(new Set(['topology'] as const), scope.notice)) } return result } @@ -123,7 +133,9 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP const known = this.localMutationCallerFingerprint const result = sourceIdentity.call(this) if (!known) { - this.crewChanges.notify() + this.crewChanges.notify( + Object.assign(new Set(['topology'] as const), { trafficAddresses: new Set() }) + ) } return result } @@ -133,7 +145,9 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP if (result.type !== 'heartbeat') { const notice: CrewInvalidation = result.delivery_contract === 'legacy_direct' - ? new Set(['topology']) + ? Object.assign(new Set(['topology'] as const), { + trafficAddresses: new Set([result.from_handle, result.to_handle]) + }) : Object.assign(new Set(['traffic'] as const), { trafficAddresses: new Set([result.from_handle, result.to_handle]) }) @@ -150,7 +164,18 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP configurable: true, writable: true, value(this: OrchestrationDb, ...args: unknown[]) { + const before = new CrewMailMutationScope(this) + before.capture(name, args) const result = original.apply(this, args) + const after = new CrewMailMutationScope(this) + after.capture(name, args) + after.row(result) + for (const address of before.trafficAddresses) { + after.trafficAddresses.add(address) + } + for (const owner of before.mailboxOwners) { + after.mailboxOwners.add(owner) + } if ( typeof result === 'object' && result !== null && @@ -160,7 +185,13 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP ) { return result } - this.crewChanges.notify(new Set([trafficMutations.has(name) ? 'traffic' : 'topology'])) + this.crewChanges.notify( + Object.assign( + new Set([trafficMutations.has(name) ? 'traffic' : 'topology'] as const), + after.notice, + name.startsWith('reset') ? { mailGlobal: true as const } : {} + ) + ) return result } }) diff --git a/src/main/runtime/orchestration/db/legacy/legacy-worker-completion.ts b/src/main/runtime/orchestration/db/legacy/legacy-worker-completion.ts index 9cff98e1d15..5eb7146d971 100644 --- a/src/main/runtime/orchestration/db/legacy/legacy-worker-completion.ts +++ b/src/main/runtime/orchestration/db/legacy/legacy-worker-completion.ts @@ -90,10 +90,15 @@ export function setLegacyCompatibilityPrincipalStatus( WHERE id = ? AND status = 'committed'` ) .run(status, id) - if (changed.changes) { - this.crewChanges.notify(new Set(['topology'])) + const principal = this.getLegacyCompatibilityPrincipal(id) + if (changed.changes && principal) { + this.crewChanges.notify( + Object.assign(new Set(['topology'] as const), { + trafficAddresses: new Set([`run:${principal.run_id}`]) + }) + ) } - return this.getLegacyCompatibilityPrincipal(id) + return principal } export function getLegacyOperationReceipt( diff --git a/src/main/runtime/orchestration/db/messages/message-inbox.ts b/src/main/runtime/orchestration/db/messages/message-inbox.ts index 5f177556e85..f0ada2bce33 100644 --- a/src/main/runtime/orchestration/db/messages/message-inbox.ts +++ b/src/main/runtime/orchestration/db/messages/message-inbox.ts @@ -1,3 +1,4 @@ +import { captureMessageMutation } from './message-mutation-invalidation' import { undeliveredInboxQuery, type UndeliveredInboxOptions } from './undelivered-inbox-query' import type { MessageType, MessageRow } from '../orchestration-records' import { exposeMessageTimestamps, exposeMessageListTimestamps } from '../utc-timestamp' @@ -18,6 +19,7 @@ function runBatchedMessageMutation( if (ids.length === 0) { return } + const changed = captureMessageMutation(db, ids) db.db.exec(`SAVEPOINT ${MESSAGE_MUTATION_SAVEPOINT}`) try { for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) { @@ -31,6 +33,7 @@ function runBatchedMessageMutation( db.db.exec(`RELEASE ${MESSAGE_MUTATION_SAVEPOINT}`) throw error } + changed() } export function getUnreadMessages( diff --git a/src/main/runtime/orchestration/db/messages/message-mutation-invalidation.ts b/src/main/runtime/orchestration/db/messages/message-mutation-invalidation.ts new file mode 100644 index 00000000000..5d32b161925 --- /dev/null +++ b/src/main/runtime/orchestration/db/messages/message-mutation-invalidation.ts @@ -0,0 +1,54 @@ +import type { OrchestrationDb } from '../orchestration-db' + +export function captureMessageMutation(db: OrchestrationDb, ids: string[]): () => void { + const read = () => { + const rows = new Map() + for (let offset = 0; offset < ids.length; offset += 500) { + const batch = ids.slice(offset, offset + 500) + for (const row of db.db + .prepare(` + SELECT id, to_handle, read, delivered_at, pointer_enter_pending FROM messages + WHERE id IN (${batch.map(() => '?').join(',')}) + `) + .all(...batch)) { + if (typeof row.id === 'string' && typeof row.to_handle === 'string') { + rows.set(row.id, { address: row.to_handle, state: JSON.stringify(row) }) + } + } + } + return rows + } + const before = read() + return () => { + const publish = () => { + const after = read() + const addresses = new Set() + for (const [id, row] of before) { + if (after.get(id)?.state !== row.state) { + addresses.add(row.address) + const next = after.get(id) + if (next) { + addresses.add(next.address) + } + } + } + if (addresses.size) { + db.crewChanges.notify( + Object.assign(new Set(['traffic'] as const), { trafficAddresses: addresses }) + ) + } + } + // An outer transaction can still roll back; compare committed rows on the next turn. + if (db.db.isTransaction) { + queueMicrotask(() => { + try { + publish() + } catch { + /* The owning database can close after the transaction commits. */ + } + }) + } else { + publish() + } + } +} diff --git a/src/main/runtime/orchestration/db/runs/run-coordinator-mail-routing.ts b/src/main/runtime/orchestration/db/runs/run-coordinator-mail-routing.ts index a29d0302d7c..918569282e9 100644 --- a/src/main/runtime/orchestration/db/runs/run-coordinator-mail-routing.ts +++ b/src/main/runtime/orchestration/db/runs/run-coordinator-mail-routing.ts @@ -67,7 +67,11 @@ export function rememberRunCoordinatorHandle( ) .run(runId, terminalHandle) if (changed.changes) { - this.crewChanges.notify(new Set(['topology'])) + this.crewChanges.notify( + Object.assign(new Set(['topology'] as const), { + trafficAddresses: new Set([`run:${runId}`, terminalHandle]) + }) + ) } } diff --git a/src/main/runtime/orchestration/mailbox/routing/runtime-mail-wake-admission.ts b/src/main/runtime/orchestration/mailbox/routing/runtime-mail-wake-admission.ts index f145c99b677..e2565c6e081 100644 --- a/src/main/runtime/orchestration/mailbox/routing/runtime-mail-wake-admission.ts +++ b/src/main/runtime/orchestration/mailbox/routing/runtime-mail-wake-admission.ts @@ -66,7 +66,7 @@ export abstract class RuntimeMailWakeAdmission extends OrcaRuntimeWithPruneMobil shellInFront: async (ptyId) => (await this.readLaunchedAgentForeground(ptyId, record.agent)) === 'shell' }) - // Host-side input fencing is required before a proven failed invocation may be stopped. + // TEMPORARY: STA-15118 requires host input fencing before automatic surviving-shell replacement. return false } @@ -77,25 +77,32 @@ export abstract class RuntimeMailWakeAdmission extends OrcaRuntimeWithPruneMobil return null } const attempt = record.mailWake?.attempt - const pty = this.getPtyRecordForPaneKey(record.paneKey) - const liveShell = - !!attempt && - pty?.connected === true && - pty.ptyId === attempt.ptyId && - pty.incarnationId === attempt.incarnationId && - pty.launchToken === attempt.launchToken && - pty.launchIncarnationId === attempt.incarnationId && - pty.worktreeId === record.worktreeId && - pty.connectionId === (record.connectionId ?? null) && - this.mailWakeInvocations.ended(record) && - (await this.readLaunchedAgentForeground(attempt.ptyId, record.agent)) === 'shell' && - this.getPtyRecordForPaneKey(record.paneKey) === pty && - this.sleepingMailRecord(record.paneKey) === record + const matchingAttempt = () => { + const pty = this.getPtyRecordForPaneKey(record.paneKey) + return ( + !!attempt && + pty?.connected === true && + pty.ptyId === attempt.ptyId && + pty.incarnationId === attempt.incarnationId && + pty.launchToken === attempt.launchToken && + pty.launchIncarnationId === attempt.incarnationId && + pty.worktreeId === record.worktreeId && + pty.connectionId === (record.connectionId ?? null) && + this.terminalRunFacts.readProcessStartedAt(attempt.ptyId, attempt.incarnationId) !== null + ) + } + const foreground = + attempt && matchingAttempt() + ? await this.readLaunchedAgentForeground(attempt.ptyId, record.agent) + : 'unknown' + const current = matchingAttempt() && this.sleepingMailRecord(record.paneKey) === record + const ended = this.mailWakeInvocations.ended(record) + const liveShell = current && ended && foreground === 'shell' return { kind: 'asleep', waking: mailWakeObligation(record) === 'automatic' && !this.sleptMailboxWake.failedToWake(record), - resuming: mailWakeObligation(record) === 'retained' && !liveShell, + resuming: current && !ended && foreground === 'agent', agentName: TUI_AGENT_DISPLAY_NAMES[record.agent], openTab: record.restoreOnTabOpenOnly === true, failed: this.sleptMailboxWake.failedToWake(record), diff --git a/src/main/runtime/orchestration/mailbox/routing/sleeping-mail-owner-mailboxes.ts b/src/main/runtime/orchestration/mailbox/routing/sleeping-mail-owner-mailboxes.ts new file mode 100644 index 00000000000..8bf4d3bf600 --- /dev/null +++ b/src/main/runtime/orchestration/mailbox/routing/sleeping-mail-owner-mailboxes.ts @@ -0,0 +1,74 @@ +import type { SleptMailboxWake } from './slept-mailbox-wake' +import type { OrchestrationDb } from '../../db/orchestration-db' +import { + RUN_PANE_KEY_MATCH_SUFFIX_SQL, + DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL, + REMOTE_ATTACHMENT_PANE_KEY_MATCH_SUFFIX_SQL, + paneKeyMatchSuffix, + isEquivalentPaneKey +} from '../../db/pane-key-match' +import { potentiallyLiveRemoteAttachmentSql } from '../../db/federation/remote-attachment-liveness' +import { parseOrcaSessionAddress } from '../../../../../shared/orca-session-address' + +/** Owner edits discover mail without enumerating unrelated unread mailboxes. */ +export function sleepingMailOwnerMailboxes( + db: OrchestrationDb, + pane: string | null, + address?: string +): string[] { + const suffix = pane === null ? null : paneKeyMatchSuffix(pane) + const sessionId = parseOrcaSessionAddress(address) + const rows = db.db + .prepare(` + SELECT 'run:' || id AS mailbox, coordinator_pane_key AS owner FROM runs + WHERE coordinator_pane_key IS NOT NULL AND ${RUN_PANE_KEY_MATCH_SUFFIX_SQL} = ? + UNION SELECT 'dispatch:' || id AS mailbox, assignee_pane_key AS owner FROM dispatch_contexts + WHERE assignee_pane_key IS NOT NULL AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ? + AND status IN ('pending', 'dispatched') + UNION SELECT 'dispatch:' || dispatch_id AS mailbox, pane_key AS owner FROM remote_dispatch_attachments + WHERE pane_key IS NOT NULL AND ${REMOTE_ATTACHMENT_PANE_KEY_MATCH_SUFFIX_SQL} = ? + AND ${potentiallyLiveRemoteAttachmentSql()} + UNION SELECT 'run:' || id AS mailbox, NULL AS owner FROM runs WHERE coordinator_orca_session_id = ? + UNION SELECT 'run:' || run_id AS mailbox, NULL AS owner FROM run_coordinator_handles WHERE terminal_handle = ? + UNION SELECT 'dispatch:' || id AS mailbox, NULL AS owner FROM dispatch_contexts + WHERE (assignee_handle = ? OR assignee_orca_session_id = ?) + AND status IN ('pending', 'dispatched') + `) + .all(suffix, suffix, suffix, sessionId, address ?? null, address ?? null, sessionId) + return rows.flatMap((row) => + typeof row.mailbox === 'string' && + (pane === null || + row.owner === null || + (typeof row.owner === 'string' && isEquivalentPaneKey(pane, row.owner))) + ? [row.mailbox] + : [] + ) +} + +export function refreshSleepingMailOwners( + wake: SleptMailboxWake, + db: OrchestrationDb, + owners: Iterable, + addresses: Iterable +): Set { + const affected = new Set(owners) + const mailboxes = new Set(addresses) + for (const pane of affected) { + for (const mailbox of [...wake.mailboxesForPane(pane), ...wake.discover(pane)]) { + mailboxes.add(mailbox) + } + } + for (const address of addresses) { + if (!address.startsWith('run:') && !address.startsWith('dispatch:')) { + for (const mailbox of sleepingMailOwnerMailboxes(db, null, address)) { + mailboxes.add(mailbox) + } + } + } + for (const mailbox of mailboxes) { + for (const pane of wake.refresh(mailbox)) { + affected.add(pane) + } + } + return affected +} diff --git a/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-runtime.ts b/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-runtime.ts index 09e00f02aad..90163ca00ed 100644 --- a/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-runtime.ts +++ b/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-runtime.ts @@ -1,3 +1,5 @@ +import { refreshSleepingMailOwners } from './sleeping-mail-owner-mailboxes' +import { sessionCrewChanges, type CrewInvalidation } from '../../crew/crew-changes' import { replaceMailWakeRecord } from '../../../session-lifecycle/projection/retained-mail-wake-attempt' import { SleepingMailRecordIndex } from './sleeping-mail-record-index' import { @@ -118,13 +120,16 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission { consumesTerminalSideEffects: false } ) this.subscribeAgentStatusRowChanges((paneKey) => this.watchMailWakeInvocations(paneKey)) - this.crewChanges.subscribe((facets) => { + const changed = (facets: CrewInvalidation) => { if (facets.has('traffic') || facets.has('topology') || facets.has('records')) { this.publishSleepingMail( - facets.has('topology') || facets.has('records') ? undefined : facets.trafficAddresses + facets.mailGlobal ? undefined : (facets.trafficAddresses ?? []), + facets.mailboxOwners ) } - }) + } + this.crewChanges.subscribe(changed) + sessionCrewChanges.subscribe(changed) this.store?.onWorkspaceSessionWritten?.((hostId) => { const hosts = hostId === null ? (this.store?.getWorkspaceSessionHostIds?.() ?? ['local']) : [hostId] @@ -132,7 +137,7 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission const changed = this.refreshSleepingRecords(host) this.watchMailWakeInvocations() if (changed) { - queueMicrotask(() => this.retrySleptMailboxWakes()) + queueMicrotask(() => this.publishSleepingMail([])) } else { this.publishSleepingMail([]) } @@ -221,7 +226,10 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission private sleepingMailPublicationQueued = false private sleepingMailPublication = '' - protected publishSleepingMail(mailboxes?: Iterable): void { + protected publishSleepingMail(mailboxes?: Iterable, owners?: Iterable): void { + for (const owner of owners ?? []) { + this.changedSleepingPanes.add(owner) + } if (mailboxes) { for (const mailbox of mailboxes) { this.sleepingMailAddresses.add(mailbox) @@ -240,7 +248,8 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission return } const global = this.globalSleepingMailInvalidation - const affected = new Set(this.changedSleepingPanes) + let affected = new Set(this.changedSleepingPanes) + const addresses = new Set(this.sleepingMailAddresses) this.changedSleepingPanes.clear() if (global && (this.sleepingRecords.size || this.sleptMailboxWake.hasTrackedMail)) { this.sleptMailboxWake.reconcile() @@ -248,11 +257,12 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission affected.add(pane) } } else { - for (const mailbox of this.sleepingMailAddresses) { - for (const pane of this.sleptMailboxWake.refresh(mailbox)) { - affected.add(pane) - } - } + affected = refreshSleepingMailOwners( + this.sleptMailboxWake, + this._orchestrationDb, + affected, + addresses + ) } this.globalSleepingMailInvalidation = false this.sleepingMailAddresses.clear() @@ -262,11 +272,9 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission await this.editMailWakeRecord(record, withoutMailWake(record)) } } - if (global) { - for (const pane of this.sleepingRecords.keys()) { - for (const mailbox of this.sleptMailboxWake.mailboxesForPane(pane)) { - this.sleptMailboxWake.request(mailbox) - } + for (const pane of global ? this.sleepingRecords.keys() : affected) { + for (const mailbox of this.sleptMailboxWake.mailboxesForPane(pane)) { + this.sleptMailboxWake.request(mailbox) } } const settledIds = [...this.settledMailWakeIds] diff --git a/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.test.ts b/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.test.ts index 7c18ebe86e1..9620e0c7612 100644 --- a/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.test.ts +++ b/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.test.ts @@ -1,6 +1,8 @@ import type { ExecutionHostId } from '../../../../../shared/execution-host' import type { WorkspaceSessionState } from '../../../../../shared/workspace-session/workspace-session-state-types' import { OrchestrationMailboxPointerDelivery } from '../terminal-pointer/mailbox-pointer-delivery' +import { createRootDispatch } from '../../db/root-dispatch-test-fixture' +import { ORCHESTRATION_REPLY_METHOD } from '../../../rpc/methods/orchestration/messaging/reply-method' import { MailWakeInvocations } from './mail-wake-invocations' import { admitsMailWakeRetry } from './mail-wake-retry-admission' import { @@ -37,6 +39,24 @@ const record: SleepingAgentSessionRecord = { } const settled = () => new Promise((resolve) => setImmediate(resolve)) +const mailNotifier = ( + resume: RuntimeNotifier['resumeSleepingAgents'], + bell?: RuntimeNotifier['sleepingMailChanged'] +): RuntimeNotifier => ({ + worktreesChanged: vi.fn(), + reposChanged: vi.fn(), + activateWorktree: vi.fn(), + createTerminal: vi.fn(), + splitTerminal: vi.fn(), + renameTerminal: vi.fn(), + focusTerminal: vi.fn(), + closeTerminal: vi.fn(), + terminalFitOverrideChanged: vi.fn(), + terminalDriverChanged: vi.fn(), + resumeSleepingAgents: resume, + sleepingMailChanged: bell +}) + it('wakes only the automatic sleeper owed deliverable mail, coalesces it, and retries a failed wake on availability', async () => { const db = new OrchestrationDb(':memory:') try { @@ -156,20 +176,7 @@ it('holds manual Sleep through the sender receipt and keeps its bell until mail ) const resume = vi.fn(async () => true) const bell = vi.fn() - const notifier: RuntimeNotifier = { - worktreesChanged: vi.fn(), - reposChanged: vi.fn(), - activateWorktree: vi.fn(), - createTerminal: vi.fn(), - splitTerminal: vi.fn(), - renameTerminal: vi.fn(), - focusTerminal: vi.fn(), - closeTerminal: vi.fn(), - terminalFitOverrideChanged: vi.fn(), - terminalDriverChanged: vi.fn(), - resumeSleepingAgents: resume, - sleepingMailChanged: bell - } + const notifier = mailNotifier(resume, bell) try { runtime.setNotifier(notifier) runtime.syncWindowGraph(1, { tabs: [], leaves: [] }) @@ -454,11 +461,16 @@ it('suppresses a definitively confirmed late wake only for its obligation and re delivery.request(mailbox) await settled() expect(wake).toHaveBeenCalledTimes(1) + sleeping = null + delivery.refresh(mailbox) + sleeping = record + delivery.request(mailbox) + await settled() + expect(wake).toHaveBeenCalledTimes(1) db.markAsRead([message.id]) delivery.refresh(mailbox) - expect(Reflect.get(delivery, 'succeeded').size).toBe(0) - expect(Reflect.get(delivery, 'failed').size).toBe(0) - expect(Reflect.get(delivery, 'attempted').size).toBe(0) + expect(delivery.failedToWake(retained)).toBe(false) + expect(delivery.heldWorktrees()).toEqual([]) for (let i = 0; i < 20; i++) { sleeping = record @@ -480,9 +492,8 @@ it('suppresses a definitively confirmed late wake only for its obligation and re db.markAsRead([row.id]) delivery.refresh(handle) } - expect(Reflect.get(delivery, 'succeeded').size).toBe(0) - expect(Reflect.get(delivery, 'failed').size).toBe(0) - expect(Reflect.get(delivery, 'attempted').size).toBe(0) + expect(wake).toHaveBeenCalledTimes(21) + expect(delivery.failedToWake(record)).toBe(false) expect(delivery.heldWorktrees()).toEqual([]) } finally { db.close() @@ -499,7 +510,7 @@ it.each([10, 100])( const records: Record = {} const mailboxes: string[] = [] for (let i = 0; i < count; i++) { - const key = `tab-${i}:11111111-1111-4111-8111-111111111111` + const key = `tab-${i}:11111111-1111-4111-8111-${String(i).padStart(12, '0')}` records[key] = new Proxy( { ...record, @@ -537,6 +548,13 @@ it.each([10, 100])( db.insertMessage({ runId: run.id, from: 'worker', to: mailbox, subject: 'done' }) } } + const task = db.createTask({ spec: 'local completion', runId: mailboxes[0]!.slice(4) }) + const dispatch = createRootDispatch( + db, + task.id, + 'term_local_worker', + 'tab-0:11111111-1111-4111-8111-000000000000' + ) let session = { ...getDefaultWorkspaceSession(), sleepingAgentSessionsByPaneKey: records } const readSession = vi.fn(() => session) const runtime = new OrcaRuntimeService( @@ -564,24 +582,20 @@ it.each([10, 100])( readSession.mockClear() statusChanged('unrelated:11111111-1111-4111-8111-111111111111') expect(visits).toBe(0) - statusChanged('tab-0:11111111-1111-4111-8111-111111111111') + statusChanged('tab-0:11111111-1111-4111-8111-000000000000') // A status event needs only this pane's attempt and PTY identity; no session inventory. expect(visits).toBeLessThanOrEqual(4) expect(readSession).not.toHaveBeenCalled() const inventory = vi.spyOn(db, 'getUndeliveredUnreadMailboxHandles') const fullRows = vi.spyOn(db, 'getUndeliveredUnreadMessages') const newest = vi.spyOn(db, 'getUndeliveredUnreadSequence') - runtime.crewChanges.notify( - Object.assign(new Set(['traffic'] as const), { trafficAddresses: new Set([mailboxes[0]!]) }) - ) + db.completeDispatch(dispatch.id) await settled() expect(inventory).not.toHaveBeenCalled() expect(fullRows).not.toHaveBeenCalled() - // Refresh and the custody check each need one scalar newest-sequence row for the affected mailbox. - expect(newest.mock.calls.length).toBeLessThanOrEqual(3) - expect( - newest.mock.results.filter((result) => result.value !== undefined).length - ).toBeLessThanOrEqual(3) + expect(newest.mock.calls.filter(([mailbox]) => mailbox !== mailboxes[0])).toEqual([]) + // One affected Run: refresh, custody validation and each start-stage revalidation; no N-dependent work. + expect(newest.mock.calls.length).toBeLessThanOrEqual(8) expect(newest).toHaveBeenCalled() session = { ...session, @@ -604,9 +618,18 @@ it.each([10, 100])( await settled() visits = 0 readSession.mockClear() - statusChanged('tab-0:11111111-1111-4111-8111-111111111111') + statusChanged('tab-0:11111111-1111-4111-8111-000000000000') expect(visits).toBe(0) expect(readSession).not.toHaveBeenCalled() + session = { ...session, sleepingAgentSessionsByPaneKey: {} } + sessionWritten('local') + await settled() + inventory.mockClear() + newest.mockClear() + db.completeDispatch(dispatch.id) + await settled() + expect(inventory).not.toHaveBeenCalled() + expect(newest).not.toHaveBeenCalled() } finally { db.close() } @@ -640,19 +663,7 @@ it('holds unknown dispatch custody but retries a positively undispatched wake on const resume = vi.fn( async (_workspace: string, _pane?: string, prepare?: boolean) => prepare === true ) - const notifier: RuntimeNotifier = { - worktreesChanged: vi.fn(), - reposChanged: vi.fn(), - activateWorktree: vi.fn(), - createTerminal: vi.fn(), - splitTerminal: vi.fn(), - renameTerminal: vi.fn(), - focusTerminal: vi.fn(), - closeTerminal: vi.fn(), - terminalFitOverrideChanged: vi.fn(), - terminalDriverChanged: vi.fn(), - resumeSleepingAgents: resume - } + const notifier = mailNotifier(resume) try { const run = db.createRun({ objective: 'refused launch', @@ -666,7 +677,8 @@ it('holds unknown dispatch custody but retries a positively undispatched wake on runtime.syncWindowGraph(1, { tabs: [], leaves: [] }) await settled() expect(await runtime.readSleepingMailHold(mailbox)).toMatchObject({ - resuming: true, + resuming: false, + openTab: true, liveShell: false }) runtime.retrySleptMailboxWakes() @@ -689,3 +701,101 @@ it('holds unknown dispatch custody but retries a positively undispatched wake on db.close() } }) + +it('clears the original workspace hold and custody when another terminal replies to its last mail', async () => { + const db = new OrchestrationDb(':memory:') + let session: WorkspaceSessionState = { + ...getDefaultWorkspaceSession(), + sleepingAgentSessionsByPaneKey: { + [paneKey]: { + ...record, + origin: 'quit', + restoreOnTabOpenOnly: true, + mailWake: { id: 'reply-custody', capturedAt: 1, startedAt: 100 } + } + } + } + let written: (host: ExecutionHostId | null) => void = () => {} + const runtime = new OrcaRuntimeService( + partialRuntimeStore({ + getWorkspaceSession: () => session, + getWorkspaceSessionHostIds: () => ['local'], + setWorkspaceSession: (next) => { + session = next + written('local') + }, + runDurableMutation: async (mutation) => mutation().value, + onWorkspaceSessionWritten: (listener) => { + written = listener + return () => {} + } + }) + ) + const bell = vi.fn() + runtime.setNotifier( + mailNotifier( + vi.fn(async () => false), + bell + ) + ) + try { + const run = db.createRun({ + objective: 'reply consumes mail', + coordinatorHandle: 'term_asleep', + coordinatorPaneKey: paneKey + }) + const mailbox = `run:${run.id}` + const sender = db.createRun({ + objective: 'sender', + coordinatorHandle: 'term_sender', + coordinatorPaneKey: null + }) + const original = db.insertMessage({ + runId: run.id, + from: `run:${sender.id}`, + to: mailbox, + subject: 'question' + }) + runtime.setOrchestrationDb(db) + runtime.retrySleptMailboxWakes() + await settled() + expect(await runtime.readSleepingMailHold(mailbox)).toMatchObject({ + kind: 'asleep', + resuming: false + }) + expect(bell).toHaveBeenLastCalledWith(['folder:workspace']) + const inventory = vi.spyOn(db, 'getUndeliveredUnreadMailboxHandles') + const originalInvalidation = vi.fn() + const unwatch = db.crewChanges.subscribe((event) => { + if (event.trafficAddresses?.has(mailbox)) { + originalInvalidation(db.getMessageById(original.id)?.read) + } + }) + db.db.exec('BEGIN IMMEDIATE') + db.markAsRead([original.id]) + expect(originalInvalidation).not.toHaveBeenCalled() + db.db.exec('ROLLBACK') + await settled() + expect(originalInvalidation).not.toHaveBeenCalled() + expect(await runtime.readSleepingMailHold(mailbox)).not.toBeNull() + await ORCHESTRATION_REPLY_METHOD.handler( + ORCHESTRATION_REPLY_METHOD.params.parse({ + id: original.id, + from: 'term_other', + body: 'answer' + }), + { runtime } + ) + await settled() + unwatch() + expect(originalInvalidation).toHaveBeenCalledWith(1) + expect(db.getMessageById(original.id)?.read).toBe(1) + expect(await runtime.readSleepingMailHold(mailbox)).toBeNull() + expect(session.sleepingAgentSessionsByPaneKey?.[paneKey]?.mailWake).toBeUndefined() + expect(inventory).not.toHaveBeenCalled() + expect(bell).toHaveBeenLastCalledWith([], ['reply-custody']) + } finally { + runtime.setNotifier(null) + db.close() + } +}) diff --git a/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.ts b/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.ts index 24af66733b2..c54298ed22e 100644 --- a/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.ts +++ b/src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.ts @@ -1,3 +1,4 @@ +import { sleepingMailOwnerMailboxes } from './sleeping-mail-owner-mailboxes' import { POTENTIALLY_LIVE_REMOTE_ATTACHMENT_STATES } from '../../db/federation/remote-attachment-liveness' import type { SleepingAgentSessionRecord } from '../../../../../shared/agent-session/lifecycle/agent-session-resume' import { agentProviderSessionsEqual } from '../../../../../shared/agent-session/lifecycle/agent-session-resume' @@ -76,7 +77,11 @@ export class SleptMailboxWake { private selection(mailbox: string, reservedTypes?: ReadonlySet) { const db = this.deps.db() const paneKey = db && mailboxOwnerPane(db, mailbox) - if (!db || !paneKey) { + if ( + !db || + !paneKey || + (!mailWakeObligation(this.deps.record(paneKey)) && !this.succeeded.has(mailbox)) + ) { return null } const sequence = selectOrchestrationPointerSequence({ @@ -90,7 +95,9 @@ export class SleptMailboxWake { refresh(mailbox: string): string[] { const before = this.owed.get(mailbox) - const next = this.selection(mailbox) + const selection = this.selection(mailbox) + const next = + selection && mailWakeObligation(this.deps.record(selection.paneKey)) ? selection : null if (before) { const handles = this.byPane.get(before.paneKey) handles?.delete(mailbox) @@ -106,7 +113,11 @@ export class SleptMailboxWake { this.byPane.set(next.paneKey, handles) } const success = this.succeeded.get(mailbox) - if (!next || success?.paneKey !== next.paneKey || success.sequence !== next.sequence) { + if ( + !selection || + success?.paneKey !== selection.paneKey || + success.sequence !== selection.sequence + ) { this.succeeded.delete(mailbox) } const attempted = this.attempted.get(mailbox) @@ -126,12 +137,12 @@ export class SleptMailboxWake { } get hasTrackedMail(): boolean { - return this.owed.size > 0 || this.attempted.size > 0 + return this.owed.size > 0 || this.attempted.size > 0 || this.succeeded.size > 0 } reconcile(): void { const handles = new Set(this.deps.db()?.getUndeliveredUnreadMailboxHandles() ?? []) - for (const handle of this.owed.keys()) { + for (const handle of [...this.owed.keys(), ...this.succeeded.keys()]) { handles.add(handle) } for (const handle of handles) { @@ -139,6 +150,13 @@ export class SleptMailboxWake { } } + discover(paneKey: string): readonly string[] { + const db = this.deps.db() + return db && mailWakeObligation(this.deps.record(paneKey)) + ? sleepingMailOwnerMailboxes(db, paneKey) + : [] + } + mailboxesForPane(paneKey: string): readonly string[] { return [...(this.byPane.get(paneKey) ?? [])] } diff --git a/src/main/runtime/workspaces/persisted-surfaces/crew-operations.ts b/src/main/runtime/workspaces/persisted-surfaces/crew-operations.ts index bae2ce39d92..e72f7b440fe 100644 --- a/src/main/runtime/workspaces/persisted-surfaces/crew-operations.ts +++ b/src/main/runtime/workspaces/persisted-surfaces/crew-operations.ts @@ -32,7 +32,9 @@ export abstract class OrcaRuntimeWithCrewOperations extends OrcaRuntimeWithOrche this.crewChanges.notify(facets) }) this.orchestrationDbChanges.notify() - this.crewChanges.notify() + this.crewChanges.notify( + Object.assign(new Set(['topology'] as const), { mailGlobal: true as const }) + ) } private readonly crewReadModel = new CrewReadModel(