You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/09/16 11:41:56 UTC
(maka) 01/01: fix(workhub): preserve follow-up messages during streamed replies
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch fix/workhub-followup-visibility
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 37bfe84c0cc83bdeee69d5f5e641c81a516eaf4a
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Wed Sep 16 19:41:44 2026 +0800
fix(workhub): preserve follow-up messages during streamed replies
Transfer queued message content into its admitted Turn atomically and retire the transient only after transcript publication. Cover edited and restored queues, publication ordering, repeated admission, and withdrawal.
Generated-by: OpenAI Codex
---
.../main/__tests__/workhub-send-visibility.test.ts | 48 ++++++++++++++
.../workhub/controller/use-workhub-controller.ts | 77 +++++++++++++++-------
2 files changed, 101 insertions(+), 24 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/workhub-send-visibility.test.ts b/apps/desktop/src/main/__tests__/workhub-send-visibility.test.ts
index 885afa2c35..381986c074 100644
--- a/apps/desktop/src/main/__tests__/workhub-send-visibility.test.ts
+++ b/apps/desktop/src/main/__tests__/workhub-send-visibility.test.ts
@@ -658,12 +658,60 @@ test('WorkHub defaults to follow-up and moves each message into its admitted suc
steering: [], followup: entries.map((entry) => entry.content.text), followupEntries: entries }));
await act(() => h.emit({ type: 'message_admission', id: 'first-admission', turnId: 'successor', messageId: first, ts: 2, outcome: 'admitted' }));
assert.deepEqual(h.controller.messageQueue.entries.map((entry) => entry.messageId), [second]);
+ await act(() => {
+ h.admit('successor');
+ h.emit({ type: 'text_delta', id: 'successor-output', turnId: 'successor', messageId: 'successor-answer', ts: 3, text: 'Responding to first follow-up' });
+ });
+ assert.equal(h.controller.liveTurn?.steps[0]?.text?.text, 'Responding to first follow-up');
+ assert.deepEqual(h.controller.transientMessages.map(({ id, text, attachments, hostTurnId, transientPlacement, pendingSteering }) =>
+ ({ id, text, attachments, hostTurnId, transientPlacement, pendingSteering })), [{
+ id: first, text: 'first follow-up', attachments, hostTurnId: 'successor', transientPlacement: 'current_turn', pendingSteering: false,
+ }], 'the admitted prompt must accompany its live answer before transcript publication');
+ await act(() => h.emit({ type: 'queue_update', id: 'remaining', turnId: 'successor', ts: 3,
+ steering: [], followup: ['second follow-up'], followupEntries: entries.slice(1) }));
+ assert.equal(h.controller.transientMessages[0]?.id, first, 'later queue snapshots cannot retire an admitted prompt');
await act(() => h.publish([{ type: 'user', id: first, turnId: 'successor', text: 'first follow-up', attachments, ts: 2 }]));
assert.deepEqual(h.controller.transientMessages, []);
assert.deepEqual(h.controller.messageQueue.entries.map((entry) => entry.messageId), [second]);
h.latestRead.resolve();
});
+test('restored follow-ups transfer edited Host content once, regardless of transcript arrival order', async () => {
+ for (const transcriptFirst of [false, true]) {
+ const h = await mountController();
+ const attachments: AttachmentRef[] = [{ kind: 'doc', name: 'brief.txt', mimeType: 'text/plain', bytes: 4, ref: { kind: 'workspace_file', relativePath: 'brief.txt' } }];
+ const entry = { entryId: 'restored-entry', messageId: 'restored-message', placement: 'next_turn' as const, state: 'queued' as const,
+ content: { text: 'model-facing envelope', displayText: 'edited follow-up', attachments } };
+ await act(() => {
+ h.emit({ type: 'queue_update', id: 'restored', turnId: 'predecessor', ts: 1, steering: [], followup: ['old follow-up'],
+ followupEntries: [{ ...entry, content: { text: 'old follow-up' } }] });
+ h.emit({ type: 'queue_update', id: 'edited', turnId: 'predecessor', ts: 2, steering: [], followup: [entry.content.text], followupEntries: [entry] });
+ });
+ const publish = () => h.publish([{ type: 'user', id: entry.messageId, turnId: 'successor', ts: 3, ...entry.content }]);
+ const admit = () => h.emit({ type: 'message_admission', id: 'admitted', turnId: 'successor', messageId: entry.messageId, ts: 3, outcome: 'admitted' });
+ if (transcriptFirst) await act(publish);
+ await act(() => { admit(); admit(); });
+ assert.deepEqual(h.controller.messageQueue.entries, []);
+ assert.deepEqual(h.controller.transientMessages.map(({ id, text, attachments, hostTurnId }) => ({ id, text, attachments, hostTurnId })),
+ transcriptFirst ? [] : [{ id: entry.messageId, text: 'edited follow-up', attachments, hostTurnId: 'successor' }]);
+ if (!transcriptFirst) await act(publish);
+ await act(admit);
+ assert.deepEqual(h.controller.transientMessages, [], 'late admission cannot recreate a published user row');
+ h.latestRead.resolve();
+ cleanupFakeDom();
+ }
+});
+
+test('withdrawing a queued follow-up never transfers it into the conversation', async () => {
+ const h = await mountController();
+ const entry = { entryId: 'queued-entry', messageId: 'queued-message', placement: 'next_turn' as const, state: 'queued' as const, content: { text: 'withdraw me' } };
+ await act(() => h.emit({ type: 'queue_update', id: 'queued', turnId: 'active', ts: 1, steering: [], followup: ['withdraw me'], followupEntries: [entry] }));
+ await act(() => h.emit({ type: 'message_admission', id: 'retracted', turnId: 'active', messageId: entry.messageId, ts: 2, outcome: 'retracted' }));
+ assert.deepEqual(h.controller.messageQueue.entries, []);
+ assert.deepEqual(h.controller.transientMessages, []);
+ h.latestRead.resolve();
+});
+
test('a queued follow-up returns the window to the tail so its own retry guard can clear', async () => {
const h = await mountController();
await act(() => { h.admit('active-turn'); h.emit({ type: 'text_delta', id: 'live', turnId: 'active-turn', messageId: 'answer', ts: 1, text: 'Working' }); });
diff --git a/apps/desktop/src/renderer/features/workhub/controller/use-workhub-controller.ts b/apps/desktop/src/renderer/features/workhub/controller/use-workhub-controller.ts
index f35824f9f7..0d42a80c97 100644
--- a/apps/desktop/src/renderer/features/workhub/controller/use-workhub-controller.ts
+++ b/apps/desktop/src/renderer/features/workhub/controller/use-workhub-controller.ts
@@ -33,7 +33,7 @@ import {
type TransientUserMessageProjection,
} from '@maka/ui';
import type { WorkHubAnswerInput, WorkHubAnswerResult } from '../../../../shared/workhub-conversation.js';
-import type { AttachmentRef, FollowUpMode, MessageQueuePlacement } from '@maka/core/events';
+import type { AttachmentRef, FollowUpMode, MessageQueueEntryProjection, MessageQueuePlacement } from '@maka/core/events';
import type { ThinkingLevel } from '@maka/core/model-thinking';
import type { ChatModelChoice } from '@maka/core/chat-model-choice';
import { startWorkHubCoordinationLifecycle } from '../../../application/contracts/workhub-workspace/coordination-lifecycle.js';
@@ -54,6 +54,10 @@ interface SendAttempt {
reconciling?: boolean;
stop?: 'requested' | 'sending' | 'resend';
}
+interface MessagePresentation {
+ transientMessages: TransientUserMessageProjection[];
+ messageQueue: { entries: MessageQueueEntryProjection[]; revision?: number };
+}
export function useWorkHubController(onSubmit?: () => void) {
const services = useWorkHubServices();
const locale = useUiLocale();
@@ -72,8 +76,9 @@ export function useWorkHubController(onSubmit?: () => void) {
// Renderer completion is a one-shot signal; publication may arrive later.
const settledBeforePublication = useRef(new Set<string>());
const [viewportNavigation] = useState(createTranscriptViewportNavigation);
- const [transientMessages, setTransientMessages] = useState<TransientUserMessageProjection[]>([]);
- const [messageQueue, setMessageQueue] = useState<{ entries: import('@maka/core/events').MessageQueueEntryProjection[]; revision?: number }>({ entries: [] });
+ const [{ transientMessages, messageQueue }, setMessagePresentation] = useState<MessagePresentation>({
+ transientMessages: [], messageQueue: { entries: [] },
+ });
const [execution, setExecution] = useState<SessionExecutionProjection>();
const [liveTurns, setLiveTurns] = useState<LiveTurnBuffer>();
const liveTurn = liveTurns?.find((turn) => turn.turnId === execution?.rootTurn?.turnId) ?? liveTurns?.at(-1);
@@ -104,8 +109,10 @@ export function useWorkHubController(onSubmit?: () => void) {
if (!retracted) return false;
if (currentSessionId.current === target && retracted.length > 0) {
const ids = new Set(retracted);
- setTransientMessages((messages) => messages.filter((message) => !ids.has(message.id)));
- setMessageQueue((queue) => ({ ...queue, entries: queue.entries.filter((entry) => !ids.has(entry.messageId)) }));
+ setMessagePresentation((previous) => ({
+ transientMessages: previous.transientMessages.filter((message) => !ids.has(message.id)),
+ messageQueue: { ...previous.messageQueue, entries: previous.messageQueue.entries.filter((entry) => !ids.has(entry.messageId)) },
+ }));
if (pendingQueued.current?.sessionId === target && ids.has(pendingQueued.current.messageId)) pendingQueued.current = undefined;
}
return true;
@@ -261,10 +268,10 @@ export function useWorkHubController(onSubmit?: () => void) {
setExecution(undefined);
setLiveTurns(pending ? [armLiveTurn(pending.input.turnId)] : undefined);
setStopPending(Boolean(pending?.stop));
- setTransientMessages(pending ? [{
+ setMessagePresentation({ messageQueue: { entries: [] }, transientMessages: pending ? [{
id: pending.input.turnId, hostTurnId: pending.input.turnId, text: pending.input.text,
attachments: pending.input.attachments, ts: Date.now(), transientPlacement: 'current_turn',
- }] : []);
+ }] : [] });
}, [sessionId]);
useEffect(() => {
@@ -290,7 +297,7 @@ export function useWorkHubController(onSubmit?: () => void) {
useEffect(() => {
if (!sessionId) return;
- setMessageQueue({ entries: [] });
+ setMessagePresentation((previous) => ({ ...previous, messageQueue: { entries: [] } }));
let disposed = false;
let handle: WorkHubTranscript | undefined;
let observationPhase: 'pending' | 'ready' = 'pending';
@@ -320,31 +327,53 @@ export function useWorkHubController(onSubmit?: () => void) {
if (terminal) setTurnStates((current) => Object.fromEntries([...Object.entries(current), [event.turnId, event.type === 'complete' ? 'completed' : event.type === 'abort' ? 'aborted' : 'failed']].slice(-64)));
if (event.type === 'queue_update') {
const entries = [...(event.steeringEntries ?? []), ...(event.followupEntries ?? [])];
- setMessageQueue({ entries: entries.filter((entry) => entry.state === 'queued'), revision: event.queueRevision });
// Host evidence retires local submission placeholders. Queue state
// lives only in messageQueue, including after reconnect or withdrawal.
const ids = new Set(entries.map((entry) => entry.messageId));
if (pendingQueued.current && ids.has(pendingQueued.current.messageId)) pendingQueued.current.observed = true;
- setTransientMessages((previous) => previous.filter((message) => !ids.has(message.id)));
+ setMessagePresentation((previous) => ({
+ messageQueue: { entries: entries.filter((entry) => entry.state === 'queued'), revision: event.queueRevision },
+ transientMessages: previous.transientMessages.filter((message) => !ids.has(message.id)),
+ }));
}
if (event.type === 'message_admission') {
if (event.outcome === 'admitted') {
if (pendingQueued.current?.messageId === event.messageId) pendingQueued.current.observed = true;
- setMessageQueue((previous) => ({ ...previous, entries: previous.entries.filter((entry) => entry.messageId !== event.messageId) }));
- setTransientMessages((previous) => previous.map((message) => message.id === event.messageId
- ? { ...message, hostTurnId: event.turnId, transientPlacement: 'current_turn', pendingSteering: false }
- : message));
+ // Admission transfers the queued content to its Host-named Turn.
+ // Neither the queue receipt nor admission proves the transcript
+ // has displayed that message yet; only publication retires it.
+ setMessagePresentation((previous) => {
+ const queued = previous.messageQueue.entries.find((entry) => entry.messageId === event.messageId);
+ const local = previous.transientMessages.find((message) => message.id === event.messageId);
+ const content = queued ? {
+ ...queued.content, text: queued.content.displayText ?? queued.content.text,
+ id: queued.messageId, ts: local?.ts ?? event.ts,
+ } : local;
+ const messages = previous.transientMessages.filter((message) => message.id !== event.messageId);
+ if (content && !transcriptRef.current.messages.some((message) => message.type === 'user' && message.id === event.messageId)) {
+ messages.push({ ...content, hostTurnId: event.turnId, transientPlacement: 'current_turn', pendingSteering: false });
+ }
+ return {
+ messageQueue: { ...previous.messageQueue, entries: previous.messageQueue.entries.filter((entry) => entry.messageId !== event.messageId) },
+ transientMessages: messages,
+ };
+ });
} else {
if (pendingQueued.current?.messageId === event.messageId) pendingQueued.current = undefined;
- setTransientMessages((previous) => previous.filter((message) => message.id !== event.messageId));
+ setMessagePresentation((previous) => ({
+ messageQueue: { ...previous.messageQueue, entries: previous.messageQueue.entries.filter((entry) => entry.messageId !== event.messageId) },
+ transientMessages: previous.transientMessages.filter((message) => message.id !== event.messageId),
+ }));
}
}
if (event.type === 'steering_message') {
- setMessageQueue((previous) => ({ ...previous, entries: previous.entries.filter((entry) => entry.messageId !== event.messageId) }));
if (pendingQueued.current?.messageId === event.messageId) pendingQueued.current.observed = true;
// The live Turn now owns this row, before the durable transcript
// necessarily catches up. Retire its admission placeholder.
- setTransientMessages((previous) => previous.filter((message) => message.id !== event.messageId));
+ setMessagePresentation((previous) => ({
+ messageQueue: { ...previous.messageQueue, entries: previous.messageQueue.entries.filter((entry) => entry.messageId !== event.messageId) },
+ transientMessages: previous.transientMessages.filter((message) => message.id !== event.messageId),
+ }));
}
reconcileAdmission(sessionId, event.turnId, event.type === 'abort' || event.type === 'error' || event.type === 'complete');
setLiveTurns((previous) => {
@@ -381,10 +410,10 @@ export function useWorkHubController(onSubmit?: () => void) {
transcriptRef.current = snapshot;
setTranscript(snapshot);
if (snapshot.ready && observationPhase === 'ready') setReadError(undefined);
- setTransientMessages((previous) => previous.filter((pending) =>
+ setMessagePresentation((previous) => ({ ...previous, transientMessages: previous.transientMessages.filter((pending) =>
!snapshot.messages.some((message) => message.type === 'user' &&
(message.id === pending.id || (pending.id === pending.hostTurnId && message.turnId === pending.hostTurnId))),
- ));
+ ) }));
const settled = snapshot.messages.filter((message) =>
message.type === 'assistant' && settledBeforePublication.current.delete(message.id));
setLiveTurns((previous) => {
@@ -441,10 +470,10 @@ export function useWorkHubController(onSubmit?: () => void) {
if (queuedTurnId) {
const attempt = sameQueued ?? { sessionId: target, turnId: queuedTurnId, messageId: crypto.randomUUID(), text, attachments: [...attachments], placement, observed: false };
pendingQueued.current = attempt;
- if (!attempt.observed) setTransientMessages((messages) => [...messages.filter((message) => message.id !== attempt.messageId), {
+ if (!attempt.observed) setMessagePresentation((previous) => ({ ...previous, transientMessages: [...previous.transientMessages.filter((message) => message.id !== attempt.messageId), {
id: attempt.messageId, hostTurnId: queuedTurnId, text, attachments: [...attachments],
ts: Date.now(), transientPlacement: attempt.placement, pendingSteering: attempt.placement === 'current_turn',
- }]);
+ }] }));
viewportNavigation.followLatest(target);
// A queued message becomes visible only where the tail is, and its own
// retry guard waits on seeing it. Issue the read before admission so an
@@ -455,7 +484,7 @@ export function useWorkHubController(onSubmit?: () => void) {
const result = await services.enqueueMessage(target, attempt.messageId, text, attachments, attempt.placement);
if (result === 'rejected' && pendingQueued.current === attempt) {
pendingQueued.current = undefined;
- setTransientMessages((messages) => messages.filter((message) => message.id !== attempt.messageId));
+ setMessagePresentation((previous) => ({ ...previous, transientMessages: previous.transientMessages.filter((message) => message.id !== attempt.messageId) }));
}
if (result !== 'admitted' && !attempt.observed) throw new Error(workHubLiveCopy[localeRef.current][result === 'unknown' ? 'sendUnknown' : 'sendNotAdmitted']);
if (pendingQueued.current === attempt) pendingQueued.current = undefined;
@@ -472,10 +501,10 @@ export function useWorkHubController(onSubmit?: () => void) {
pendingSend.current = attempt;
setTurnStates((states) => ({ ...states, [attempt.input.turnId]: 'running' }));
setLiveTurns((previous) => retainLiveTurn(previous, armLiveTurn(attempt.input.turnId)));
- setTransientMessages((previous) => [...previous.filter((message) => message.hostTurnId !== attempt.input.turnId && message.hostTurnId !== pendingRejectedTurnId), {
+ setMessagePresentation((previous) => ({ ...previous, transientMessages: [...previous.transientMessages.filter((message) => message.hostTurnId !== attempt.input.turnId && message.hostTurnId !== pendingRejectedTurnId), {
id: attempt.input.turnId, hostTurnId: attempt.input.turnId, text, ts: Date.now(),
attachments: [...attachments], transientPlacement: 'current_turn',
- }]);
+ }] }));
viewportNavigation.followLatest(target);
// Return a historical range to the tail without delaying message admission.
void range.current?.loadLatest().catch((reason: unknown) => {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org