From dddc6cc45279541259a5d667b220c0feb7e9e883 Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Fri, 10 Jul 2026 21:23:59 -0700 Subject: [PATCH] Fix orchestration lifecycle sender resolution and peek/check compat hand - Lifecycle sends (worker_done/heartbeat) now use ORCA_TERMINAL_HANDLE verbatim, skipping the liveness probe and pane remint that could block delivery during restarts or mismatch stale-runtime assignee handles. - --peek now round-trips as {peek:true, unread:false} so older runtimes that strip unknown params degrade to non-destructive "all" instead of mark-read, with client-side filtering to restore peek semantics and a clear error when --peek --wait can't be honored. - Reject combined read-mode flags (--unread/--peek/--all) before calling the runtime. - Distinguish suppressed (already-consumed) lifecycle messages from ignored ones so send doesn't wake --wait waiters for stale heartbeats. - Fix task summary truncation to avoid splitting UTF-16 surrogate pairs and to not misreport whitespace normalization as truncation. --- src/cli/handlers/orchestration.test.ts | 139 ++++++++++++------ src/cli/handlers/orchestration.ts | 66 +++++++-- src/cli/orchestration-task-summary.test.ts | 16 ++ src/cli/orchestration-task-summary.ts | 19 ++- src/cli/specs/core.ts | 2 +- src/cli/specs/orchestration.ts | 2 +- .../lifecycle-reconciliation.test.ts | 2 +- .../orchestration/lifecycle-reconciliation.ts | 6 +- .../runtime/rpc/methods/orchestration.test.ts | 55 +++++++ src/main/runtime/rpc/methods/orchestration.ts | 14 +- 10 files changed, 246 insertions(+), 75 deletions(-) diff --git a/src/cli/handlers/orchestration.test.ts b/src/cli/handlers/orchestration.test.ts index 9d8e40378b3..e855ef802c9 100644 --- a/src/cli/handlers/orchestration.test.ts +++ b/src/cli/handlers/orchestration.test.ts @@ -214,7 +214,7 @@ describe('orchestration send structured payload flags', () => { }) }) - it('uses a live ORCA_TERMINAL_HANDLE as worker lifecycle sender metadata', async () => { + it('sends lifecycle messages from ORCA_TERMINAL_HANDLE without a liveness probe', async () => { process.env.ORCA_TERMINAL_HANDLE = 'term_worker_env' await invokeSend( @@ -225,11 +225,8 @@ describe('orchestration send structured payload flags', () => { ]) ) - expect(callMock).toHaveBeenCalledTimes(2) - expect(callMock).toHaveBeenNthCalledWith(1, 'terminal.show', { - terminal: 'term_worker_env' - }) - expect(callMock).toHaveBeenNthCalledWith(2, 'orchestration.send', { + expect(callMock).toHaveBeenCalledTimes(1) + expect(callMock).toHaveBeenCalledWith('orchestration.send', { from: 'term_worker_env', to: 'term_coord', subject: 'done', @@ -242,48 +239,31 @@ describe('orchestration send structured payload flags', () => { }) }) - it('remints a stale lifecycle sender from its pane key', async () => { - process.env.ORCA_TERMINAL_HANDLE = 'term_worker_stale' - process.env.ORCA_PANE_KEY = 'tab_worker:leaf_worker' - stubStaleHandleRemint('term_worker_live', { result: { message: { id: 'msg_1' } } }) + it.each(['worker_done', 'heartbeat'] as const)( + 'never probes or remints a %s sender even when a pane key is set', + async (type) => { + process.env.ORCA_TERMINAL_HANDLE = 'term_worker_env' + process.env.ORCA_PANE_KEY = 'tab_worker:leaf_worker' - await invokeSend( - new Map([ - ['to', 'term_coord'], - ['subject', 'done'], - ['type', 'worker_done'] - ]) - ) + await invokeSend( + new Map([ + ['to', 'term_coord'], + ['subject', 'update'], + ['type', type] + ]) + ) - expect(callMock).toHaveBeenNthCalledWith(2, 'terminal.resolvePane', { - paneKey: 'tab_worker:leaf_worker' - }) - expect(callMock).toHaveBeenNthCalledWith( - 3, - 'orchestration.send', - expect.objectContaining({ from: 'term_worker_live' }) - ) - }) - - it('keeps a stale lifecycle sender when no pane key is available', async () => { - process.env.ORCA_TERMINAL_HANDLE = 'term_worker_stale' - callMock.mockRejectedValueOnce(staleHandleError()) - - await invokeSend( - new Map([ - ['to', 'term_coord'], - ['subject', 'done'], - ['type', 'worker_done'] - ]) - ) - - expect(callMock).toHaveBeenCalledTimes(2) - expect(callMock).toHaveBeenNthCalledWith( - 2, - 'orchestration.send', - expect.objectContaining({ from: 'term_worker_stale' }) - ) - }) + // Why: pre-payload-authority runtimes only complete a worker_done whose + // sender equals the recorded (equally stale) assignee handle, and + // coordinator replies route to the sender row the worker's env-handle + // `check` actually reads โ€” so lifecycle sends must stay env-verbatim. + expect(callMock).toHaveBeenCalledTimes(1) + expect(callMock).toHaveBeenCalledWith( + 'orchestration.send', + expect.objectContaining({ from: 'term_worker_env' }) + ) + } + ) it('reports sender resolution failure instead of raw no_active_terminal', async () => { getTerminalHandleMock.mockRejectedValue( @@ -690,9 +670,11 @@ describe('orchestration timeout flag validation', () => { ]) ) + // Why: --peek rides with unread:false so pre-peek runtimes fall back to + // the non-consuming all mode instead of the destructive mark-read default. expect(callMock).toHaveBeenCalledWith('orchestration.check', { terminal: 'term_worker', - unread: undefined, + unread: false, peek: true, all: undefined, types: undefined, @@ -702,6 +684,69 @@ describe('orchestration timeout flag validation', () => { }) }) + it('filters already-read rows from a peek response for pre-peek runtimes', async () => { + process.env.ORCA_TERMINAL_HANDLE = 'term_worker' + callMock.mockResolvedValue({ + result: { + messages: [ + { id: 'msg_old', from_handle: 'a', subject: 'seen', read: 1 }, + { id: 'msg_new', from_handle: 'a', subject: 'fresh', read: 0 } + ], + count: 2, + formatted: 'banners built from all rows' + } + }) + vi.mocked(printResult).mockClear() + + await invokeCheck(new Map([['peek', true]])) + + const response = vi.mocked(printResult).mock.calls[0]?.[0] as { + result: { messages: { id: string }[]; count: number; formatted?: string } + } + expect(response.result.messages.map((m) => m.id)).toEqual(['msg_new']) + expect(response.result.count).toBe(1) + // Why: the pre-peek runtime built `formatted` from all rows, including + // the read one the filter just removed. + expect(response.result.formatted).toBeUndefined() + }) + + it('rejects combined read modes before calling the runtime', async () => { + process.env.ORCA_TERMINAL_HANDLE = 'term_worker' + callMock.mockClear() + + await expect( + invokeCheck( + new Map([ + ['unread', true], + ['peek', true] + ]) + ) + ).rejects.toMatchObject({ + code: 'invalid_argument', + message: expect.stringContaining('read mode') + }) + expect(callMock).not.toHaveBeenCalled() + }) + + it('fails --peek --wait against a runtime that returned only read rows', async () => { + process.env.ORCA_TERMINAL_HANDLE = 'term_worker' + callMock.mockResolvedValue({ + result: { + messages: [{ id: 'msg_old', from_handle: 'a', subject: 'seen', read: 1 }], + count: 1 + } + }) + + await expect( + invokeCheck( + new Map([ + ['peek', true], + ['wait', true] + ]) + ) + ).rejects.toMatchObject({ code: 'peek_wait_unsupported' }) + }) + it.each(invalidTimeoutValues)('rejects invalid ask --timeout-ms: %s', async (_label, value) => { const flags = new Map([ ['to', 'term_coord'], diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index 166a1c3ece1..b874c4bdddf 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -77,6 +77,7 @@ type MessageSummary = { type?: string body?: string payload?: string | null + read?: number } function getOptionalStructuredMessagePayload( @@ -132,7 +133,7 @@ async function resolveOrchestrationTerminalHandle( cwd: string, client: Parameters[0]['client'], flagName: 'from' | 'terminal', - options: { validateEnvHandle?: boolean; allowStaleEnvHandle?: boolean } = {} + options: { validateEnvHandle?: boolean } = {} ): Promise { const explicit = getOptionalStringFlag(flags, flagName) if (explicit) { @@ -146,15 +147,10 @@ async function resolveOrchestrationTerminalHandle( // coordinator preambles. const live = await isLiveTerminalHandle(envHandle, client) if (!live) { - const reminted = await resolveOrchestrationPaneTerminalHandle(client, { - optional: options.allowStaleEnvHandle - }) + const reminted = await resolveOrchestrationPaneTerminalHandle(client) if (reminted) { return reminted } - if (options.allowStaleEnvHandle) { - return envHandle - } throwNoActiveSenderTerminal() } } @@ -351,12 +347,13 @@ export const ORCHESTRATION_HANDLERS: Record = { const type = getOptionalStringFlag(flags, 'type') rejectLifecycleGroupRecipient(type, to) - const from = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'from', { - validateEnvHandle: type === 'worker_done' || type === 'heartbeat', - // Why: payload IDs remain authoritative when an older/SSH session lacks - // a pane key, so handle repair must not prevent lifecycle delivery. - allowStaleEnvHandle: type === 'worker_done' || type === 'heartbeat' - }) + // Why: lifecycle senders keep ORCA_TERMINAL_HANDLE verbatim โ€” no liveness + // probe (terminal.show throws runtime_unavailable in the exact mid-restart + // window worker_done must survive) and no pane remint (pre-payload- + // authority runtimes require from === the equally stale assignee_handle, + // and coordinator replies route to the sender row while the worker's own + // `check` reads its env-handle inbox). + const from = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'from') const result = await client.call< { message: { id: string } } | { messages: { id: string }[]; recipients: number } >('orchestration.send', { @@ -380,6 +377,16 @@ export const ORCHESTRATION_HANDLERS: Record = { 'orchestration check': async ({ flags, client, cwd, json }) => { const wait = flags.has('wait') + const peek = flags.has('peek') + // Why: enforce mode exclusivity client-side too โ€” an older runtime strips + // the unknown `peek` param and would otherwise execute --unread --peek as + // a destructive mark-read. + if ([flags.has('unread'), peek, flags.has('all')].filter(Boolean).length > 1) { + throw new RuntimeClientError( + 'invalid_argument', + 'Choose at most one message read mode: --unread, --peek, or --all.' + ) + } const timeoutMs = getOptionalPositiveIntegerValueFlag(flags, 'timeout-ms') const terminal = await resolveOrchestrationTerminalHandle(flags, cwd, client, 'terminal') @@ -400,8 +407,12 @@ export const ORCHESTRATION_HANDLERS: Record = { try { result = await client.call('orchestration.check', { terminal, - unread: flags.has('unread') ? true : undefined, - peek: flags.has('peek') ? true : undefined, + // Why: --peek also sends unread:false so runtimes that predate the + // peek param (which their non-strict schema strips) degrade to the + // non-consuming all-messages mode instead of the destructive + // mark-read default; the read filter below restores peek semantics. + unread: flags.has('unread') ? true : peek ? false : undefined, + peek: peek ? true : undefined, all: flags.has('all') ? true : undefined, types: getOptionalStringFlag(flags, 'types'), inject: flags.has('inject') ? true : undefined, @@ -411,6 +422,31 @@ export const ORCHESTRATION_HANDLERS: Record = { } finally { stopKeepalive?.() } + if (peek) { + const unreadOnly = result.result.messages.filter((m) => m.read !== 1) + const removedReadRows = unreadOnly.length !== result.result.messages.length + // Why: read rows in a peek response are the pre-peek-runtime signature + // (its schema stripped `peek` and it ran the all mode). Such a runtime + // returned instead of blocking, so honoring --wait is impossible โ€” + // failing beats silently returning empty before the deadline. + if (wait && removedReadRows && unreadOnly.length === 0) { + throw new RuntimeClientError( + 'peek_wait_unsupported', + 'The connected runtime does not support --peek with --wait; upgrade the runtime or use --wait without --peek.' + ) + } + result = { + ...result, + result: { + ...result.result, + // Why: a pre-peek runtime builds `formatted` from all rows; drop it + // when the read filter removed any so output matches the peek set. + ...(removedReadRows ? { formatted: undefined } : {}), + messages: unreadOnly, + count: unreadOnly.length + } + } + } printResult(result, json, (r) => { if (r.formatted) { return r.formatted diff --git a/src/cli/orchestration-task-summary.test.ts b/src/cli/orchestration-task-summary.test.ts index 69c7a19131f..75fa6e68374 100644 --- a/src/cli/orchestration-task-summary.test.ts +++ b/src/cli/orchestration-task-summary.test.ts @@ -19,4 +19,20 @@ describe('abbreviateOrchestrationTasks', () => { expect(task).toEqual({ spec: 'Short task', spec_truncated: false }) }) + + it('does not report whitespace normalization as truncation', () => { + const [task] = abbreviateOrchestrationTasks([{ spec: 'Short\n\n task' }]) + + expect(task).toEqual({ spec: 'Short task', spec_truncated: false }) + }) + + it('does not split a surrogate pair at the truncation boundary', () => { + // 158 chars + an astral emoji spanning UTF-16 units 158-159: a naive + // slice(0, 159) would cut the pair and leave a lone high surrogate. + const [task] = abbreviateOrchestrationTasks([{ spec: `${'a'.repeat(158)}๐Ÿ˜€${'b'.repeat(40)}` }]) + + expect(task.spec_truncated).toBe(true) + expect(task.spec.isWellFormed()).toBe(true) + expect(task.spec.endsWith('โ€ฆ')).toBe(true) + }) }) diff --git a/src/cli/orchestration-task-summary.ts b/src/cli/orchestration-task-summary.ts index 930041893b7..5c2cd46e40f 100644 --- a/src/cli/orchestration-task-summary.ts +++ b/src/cli/orchestration-task-summary.ts @@ -5,14 +5,21 @@ export function abbreviateOrchestrationTasks( ): (T & { spec_truncated: boolean })[] { return tasks.map((task) => { const spec = task.spec.replace(/\s+/g, ' ').trim() - const abbreviated = - spec.length > TASK_SPEC_BRIEF_LENGTH - ? `${spec.slice(0, TASK_SPEC_BRIEF_LENGTH - 1).trimEnd()}โ€ฆ` - : spec + const truncated = spec.length > TASK_SPEC_BRIEF_LENGTH return { ...task, - spec: abbreviated, - spec_truncated: abbreviated !== task.spec + spec: truncated ? `${truncateAtCodePoint(spec).trimEnd()}โ€ฆ` : spec, + // Why: whitespace normalization alone is not truncation; flagging it + // would make agents re-fetch full specs that --brief already shows. + spec_truncated: truncated } }) } + +function truncateAtCodePoint(spec: string): string { + const sliced = spec.slice(0, TASK_SPEC_BRIEF_LENGTH - 1) + // Why: a cut through a surrogate pair leaves a lone high surrogate that + // strict JSON consumers reject; drop it rather than emit malformed UTF-16. + const lastUnit = sliced.charCodeAt(sliced.length - 1) + return lastUnit >= 0xd800 && lastUnit <= 0xdbff ? sliced.slice(0, -1) : sliced +} diff --git a/src/cli/specs/core.ts b/src/cli/specs/core.ts index 73c677a0f9b..740f4e5eaa9 100644 --- a/src/cli/specs/core.ts +++ b/src/cli/specs/core.ts @@ -116,7 +116,7 @@ export const CORE_COMMAND_SPECS: CommandSpec[] = [ '--no-parent only affects Orca lineage; omit --base-branch to use the repo default base, or pass the default base ref explicitly for independent top-level work.', 'By default this creates the worktree and its first terminal without switching the active Orca view.', 'Pass --agent to launch an agent in the first terminal; --prompt sends initial work to that agent.', - 'With --agent --json, read the new agent handle from result.agentTerminalHandle (or result.startupTerminal.handle on older runtimes).', + 'With --agent --json, read the new agent handle from result.agentTerminalHandle; older runtimes return only result.startupTerminal.handle, and may return neither for folder-based repos.', 'Repo-defined setup hooks follow the repository setup policy; pass --setup run to force them.', 'Pass --activate when the CLI caller intentionally wants to reveal the new worktree in the app.', 'Passing --run-hooks is kept as a legacy alias for --setup run and reveals the worktree.' diff --git a/src/cli/specs/orchestration.ts b/src/cli/specs/orchestration.ts index 86971bfd8dc..41468b434fc 100644 --- a/src/cli/specs/orchestration.ts +++ b/src/cli/specs/orchestration.ts @@ -26,7 +26,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [ notes: [ 'On Windows PowerShell, quote group addresses such as --to "@all" or --to "@worktree:".', 'worker_done and heartbeat must target a concrete coordinator terminal handle; use status for broadcast updates.', - 'A worker_done with the active task/dispatch IDs completes that task automatically; handles are routing metadata.', + 'A worker_done with the active task/dispatch IDs completes that task automatically; on older runtimes the sender must still match the dispatch assignee handle, so avoid overriding --from.', 'Prefer --task-id/--dispatch-id/etc. over raw --payload JSON in worker commands; PowerShell strips JSON quotes easily.' ] }, diff --git a/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts b/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts index c7e47beab0c..c15cfd116b4 100644 --- a/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts +++ b/src/main/runtime/orchestration/lifecycle-reconciliation.test.ts @@ -68,7 +68,7 @@ describe('lifecycle reconciliation', () => { expect(archived?.delivered_at).not.toBeNull() const lateHeartbeat = insertHeartbeat(dispatch.id, 'term_worker') - expect(reconcileLifecycleMessage(db, lateHeartbeat)).toEqual({ action: 'ignored' }) + expect(reconcileLifecycleMessage(db, lateHeartbeat)).toEqual({ action: 'suppressed' }) expect(db.getMessageById(lateHeartbeat.id)).toMatchObject({ read: 1 }) expect(db.getMessageById(lateHeartbeat.id)?.delivered_at).not.toBeNull() }) diff --git a/src/main/runtime/orchestration/lifecycle-reconciliation.ts b/src/main/runtime/orchestration/lifecycle-reconciliation.ts index 12f828b5d5c..58f39a73445 100644 --- a/src/main/runtime/orchestration/lifecycle-reconciliation.ts +++ b/src/main/runtime/orchestration/lifecycle-reconciliation.ts @@ -3,6 +3,10 @@ import type { MessageRow } from './types' export type LifecycleReconciliationResult = | { action: 'ignored' } + // Why: `suppressed` means the message was consumed at reconcile time (marked + // read); senders must not wake waiters for it, unlike `ignored` rows that + // stay unread and still need delivery. + | { action: 'suppressed' } | { action: 'completed'; taskId: string; dispatchId: string } | { action: 'heartbeat_recorded'; dispatchId: string } @@ -69,7 +73,7 @@ function reconcileHeartbeatMessage( // audit history without surfacing obsolete liveness to the coordinator. db.markAsReadAndDelivered([msg.id]) onLog(`Heartbeat for inactive dispatch ${dispatchId} suppressed`) - return { action: 'ignored' } + return { action: 'suppressed' } } // Why: dispatchId-specific writes let the DB ignore late heartbeats for diff --git a/src/main/runtime/rpc/methods/orchestration.test.ts b/src/main/runtime/rpc/methods/orchestration.test.ts index 0f0c237b00d..3c8f44ca884 100644 --- a/src/main/runtime/rpc/methods/orchestration.test.ts +++ b/src/main/runtime/rpc/methods/orchestration.test.ts @@ -90,6 +90,44 @@ describe('orchestration RPC methods', () => { expect(runtime.deliverPendingMessagesForHandle).toHaveBeenCalledWith('term_b') }) + it('does not wake waiters for a heartbeat suppressed at send time', async () => { + setup() + const task = db.createTask({ spec: 'work' }) + const dispatch = db.createDispatchContext(task.id, 'term_worker') + db.updateTaskStatus(task.id, 'completed') + vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {}) + const notify = vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {}) + + const result = (await call('orchestration.send', { + from: 'term_worker', + to: 'term_coord', + subject: 'alive', + type: 'heartbeat', + payload: JSON.stringify({ dispatchId: dispatch.id }) + })) as { message: { id: string } } + + expect(notify).not.toHaveBeenCalled() + expect(db.getMessageById(result.message.id)).toMatchObject({ read: 1 }) + }) + + it('still wakes waiters for a heartbeat on an active dispatch', async () => { + setup() + const task = db.createTask({ spec: 'work' }) + const dispatch = db.createDispatchContext(task.id, 'term_worker') + vi.spyOn(runtime, 'deliverPendingMessagesForHandle').mockImplementation(() => {}) + const notify = vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {}) + + await call('orchestration.send', { + from: 'term_worker', + to: 'term_coord', + subject: 'alive', + type: 'heartbeat', + payload: JSON.stringify({ dispatchId: dispatch.id }) + }) + + expect(notify).toHaveBeenCalledWith('term_coord', 'heartbeat') + }) + it('rejects missing --to', () => { const method = findMethod('orchestration.send') expect(() => method.params!.parse({ subject: 'hi' })).toThrow() @@ -704,6 +742,23 @@ describe('orchestration RPC methods', () => { expect(db.getUnreadMessages('b')).toHaveLength(1) }) + it("treats the CLI's {peek, unread:false} compat pair as peek, not all", async () => { + setup() + const seen = db.insertMessage({ from: 'a', to: 'b', subject: 'seen' }) + db.markAsRead([seen.id]) + db.insertMessage({ from: 'a', to: 'b', subject: 'fresh' }) + + const result = (await call('orchestration.check', { + terminal: 'b', + peek: true, + unread: false + })) as { messages: { subject: string }[]; count: number } + + expect(result.count).toBe(1) + expect(result.messages[0]?.subject).toBe('fresh') + expect(db.getUnreadMessages('b')).toHaveLength(1) + }) + it('--all returns every message for the handle without marking read', async () => { setup() db.insertMessage({ from: 'a', to: 'b', subject: 'one' }) diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index d33f003e24f..37bbfd3a322 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -87,10 +87,13 @@ const CheckParams = z timeoutMs: OptionalFiniteNumber }) .superRefine((params, ctx) => { + // Why: the CLI encodes --peek as {peek:true, unread:false} so pre-peek + // runtimes degrade to the non-consuming all mode; that pair is one mode, + // not a conflict. const modes = [ params.unread === true, params.peek === true, - params.all === true || params.unread === false + params.all === true || (params.unread === false && params.peek !== true) ].filter(Boolean) if (modes.length > 1) { ctx.addIssue({ @@ -217,7 +220,12 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ // immediately dispatch to the same terminal, which fails if the lock // is still held. if (msg.type === 'worker_done' || msg.type === 'heartbeat') { - reconcileLifecycleMessage(db, msg) + const reconciled = reconcileLifecycleMessage(db, msg) + // Why: a suppressed message is already read; waking a `check --wait` + // waiter for it would return an empty result before the deadline. + if (reconciled.action === 'suppressed') { + return { message: msg } + } } runtime.deliverPendingMessagesForHandle(params.to) runtime.notifyMessageArrived(params.to, msg.type) @@ -279,7 +287,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ // Explicit `unread: false` is also honored for one release as a compat // shim so in-flight callers don't break (see design doc ยง5). Otherwise // today's behavior is preserved: default is unread-only + mark-read. - const showAll = params.all === true || params.unread === false + const showAll = params.all === true || (params.unread === false && params.peek !== true) const consumeUnread = !showAll && params.peek !== true const readAndReturn = () => {