Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -451,6 +451,7 @@ impl DialogScheduler {
reason: None,
target_turn_id: None,
steering_id: None,
steering_display: None,
};
s.entries.insert(
message.turn_id.clone(),
Expand Down Expand Up @@ -666,6 +667,19 @@ impl DialogScheduler {
e.view.reason = None;
e.view.target_turn_id = Some(target.clone());
e.view.steering_id = Some(steering_id);
e.view.steering_display =
Some(openbitfun_runtime_ports::DialogQueueSteeringDisplay {
target_turn_id: target.clone(),
content: turn
.original_user_input
.clone()
.unwrap_or_else(|| turn.user_input.clone()),
composer_presentation: turn
.user_message_metadata
.as_ref()
.and_then(|metadata| metadata.get("composerPresentation"))
.cloned(),
});
e.held = Some(turn);
}
self.round_injection_buffer.push(session, injection);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -196,13 +196,15 @@ async fn host_queue_cancel_is_idempotent_and_never_cancels_active_turn() {
#[tokio::test]
async fn host_queue_steering_retains_payload_until_formal_turn_handoff_and_rejects_cancel() {
let (scheduler, _, _root, epoch) = fixture().await;
let mut input = message("queued-a");
let display_content = "display text ".repeat(250);
input.display_content = Some(display_content.clone());
input.metadata.insert(
"composerPresentation".into(),
serde_json::json!({"version":1}),
);
scheduler
.manage_host_queue(request(
Some(&epoch),
Action::Submit {
message: message("queued-a"),
},
))
.manage_host_queue(request(Some(&epoch), Action::Submit { message: input }))
.await
.unwrap();
let promote = request(
Expand All @@ -214,7 +216,16 @@ async fn host_queue_steering_retains_payload_until_formal_turn_handoff_and_rejec
},
);
let snapshot = scheduler.manage_host_queue(promote.clone()).await.unwrap();
assert_eq!(snapshot.receipt.unwrap().status, Status::SteeringPending);
let receipt = snapshot.receipt.unwrap();
assert_eq!(receipt.status, Status::SteeringPending);
assert!(receipt.preview_truncated);
let display = receipt.steering_display.unwrap();
assert_eq!(display.target_turn_id, "active-turn");
assert_eq!(display.content, display_content);
assert_eq!(
display.composer_presentation,
Some(serde_json::json!({"version":1}))
);
scheduler.manage_host_queue(promote).await.unwrap();
let injections = scheduler
.round_injection_source
Expand Down Expand Up @@ -254,7 +265,9 @@ async fn host_queue_steering_retains_payload_until_formal_turn_handoff_and_rejec
))
.await
.unwrap();
assert_eq!(result.receipt.unwrap().status, Status::Queued);
let receipt = result.receipt.unwrap();
assert_eq!(receipt.status, Status::Queued);
assert_eq!(receipt.steering_display.unwrap(), display);
assert_eq!(result.used, 1);
assert_eq!(scheduler.queues.depth("host-queue-session"), 1);
assert_eq!(
Expand Down
30 changes: 30 additions & 0 deletions src/crates/contracts/runtime-ports/src/dialog_queue.rs
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,16 @@ impl DialogQueueStatus {
}
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DialogQueueSteeringDisplay {
/// The original target survives the brief queued handoff to a formal Turn.
pub target_turn_id: String,
pub content: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub composer_presentation: Option<serde_json::Value>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DialogQueueItem {
Expand All @@ -83,6 +93,10 @@ pub struct DialogQueueItem {
pub reason: Option<String>,
pub target_turn_id: Option<String>,
pub steering_id: Option<String>,
/// Display-only receipt for accepted steering; never a second history record.
/// Older hosts omit this field and clients retain the composer queue UI.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub steering_display: Option<DialogQueueSteeringDisplay>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
Expand Down Expand Up @@ -132,4 +146,20 @@ mod tests {
let request: DialogQueueRequest = serde_json::from_value(input.clone()).unwrap();
assert_eq!(serde_json::to_value(request).unwrap(), input);
}

#[test]
fn queue_item_legacy_and_display_receipts_round_trip() {
let legacy = serde_json::json!({"turnId":"next", "displayContent":"short",
"previewTruncated":true, "attachmentCount":2, "agentType":"Standard",
"createdAtMs":1, "status":"steering_pending", "reason":null,
"targetTurnId":"active", "steeringId":"steer"});
let item: DialogQueueItem = serde_json::from_value(legacy.clone()).unwrap();
assert!(item.steering_display.is_none());
assert_eq!(serde_json::to_value(item).unwrap(), legacy);
let mut extended = legacy;
extended["steeringDisplay"] = serde_json::json!({"targetTurnId":"active", "content":"complete prompt",
"composerPresentation":{"version":1}});
let item: DialogQueueItem = serde_json::from_value(extended.clone()).unwrap();
assert_eq!(serde_json::to_value(item).unwrap(), extended);
}
}
2 changes: 2 additions & 0 deletions src/shared/dialog-queue/HostDialogQueue.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@ export interface QueueItem {
turnId: string; displayContent: string; previewTruncated: boolean; attachmentCount: number;
agentType: string; createdAtMs: number; status: QueueStatus; reason: string | null;
targetTurnId: string | null; steeringId: string | null;
/** Optional host-authoritative display receipt; absent on older hosts. */
steeringDisplay?: { targetTurnId: string; content: string; composerPresentation?: unknown };
}
export interface QueueSnapshot {
sessionId: string; queueEpoch: string; revision: number; activeTurnId: string | null;
Expand Down
9 changes: 6 additions & 3 deletions src/web-ui/src/flow_chat/components/HostPendingQueuePanel.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,8 @@ import { useI18n } from '@/infrastructure/i18n';
import { Button, Icon, IconButton, OverflowText, Tooltip } from '@openbitfun/ui';
import { RotateCw } from 'lucide-react';
import './HostPendingQueuePanel.scss';
import { HostDialogQueue, observeHostQueue, type QueueOutboxRecord, type QueueItem } from '../../../../shared/dialog-queue/HostDialogQueue';
import { hostQueuePresentation, observePresentedHostQueue } from '../services/hostQueuePresentation';
import { HostDialogQueue, type QueueOutboxRecord, type QueueItem } from '../../../../shared/dialog-queue/HostDialogQueue';
import {
ChatComposerQueue, ChatComposerQueueHeader, ChatComposerQueueTitle,
ChatComposerQueueList, ChatComposerQueueItem, ChatComposerQueueItemContent, ChatComposerQueueItemActions,
Expand All @@ -14,12 +15,14 @@ import {
export function HostPendingQueuePanel({ queue, onRestore }: { queue: HostDialogQueue; onRestore: (item: QueuedMessage) => boolean }) {
const { t } = useI18n('flow-chat');
const view = useSyncExternalStore(queue.subscribe, queue.getSnapshot, queue.getSnapshot);
const presentation = hostQueuePresentation(queue);
const projected = useSyncExternalStore(presentation.subscribe, presentation.getSnapshot, presentation.getSnapshot);
const [expanded, setExpanded] = useState(true);
const [showHelp, setShowHelp] = useState(false);
const listId = useId();
const [busy, setBusy] = useState(false);
const [error, setError] = useState<string | null>(null);
useEffect(() => observeHostQueue(queue), [queue]);
useEffect(() => observePresentedHostQueue(queue), [queue]);
const run = async (operation: () => Promise<unknown>) => {
setBusy(true); setError(null);
try { await operation(); } catch (e) { setError(String(e)); }
Expand Down Expand Up @@ -49,7 +52,7 @@ export function HostPendingQueuePanel({ queue, onRestore }: { queue: HostDialogQ
}
await queue.dismiss(saved);
};
const items = view.snapshot?.items ?? [];
const items = (view.snapshot?.items ?? []).filter(item => item.status === 'blocked' || !projected.has(item.turnId));
const visibleError = error || (view.error?.includes('Session is not loaded') ? null : view.error);
if (!items.length && !view.pending.length && !visibleError) return null;
return <ChatComposerQueue data-openbitfun-product-component="pending-queue-panel" data-openbitfun-product-part="root" className="host-pending-queue" aria-label={t('hostQueue.title')}>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ each missing what the other had.

| Test | Contract it holds |
|---|---|
| `../../timeline/hostQueueProjection.test.ts`, `../../timeline/useHostQueueTimeline.test.tsx`, `../HostPendingQueuePanel.test.tsx` | host-accepted preview, canonical handoff/event ordering, stable keys, blocked recovery, old-host fallback, history-window visibility and device lease cleanup; fixtures only, no live remote transport evidence |
| `TurnFooter.test.tsx`, `modelRoundItemMemo.test.ts` | latest-turn identity, hover/focus across virtual rows, lease cleanup, no projection churn and the existing completion reveal gate; behavior only, not visual proof |
| `../../../shared/ui/Tooltip.test.tsx` | shared metric click opening, immediate scroll dismissal, delayed-open cancellation, viewport-exit observer cleanup and no automatic reopening; DOM/observer behavior only, not native scrolling or visual proof |
| `../../utils/turnUsageMetrics.test.ts` | cache denominator, whole-turn output rate, speed boundaries, invalid and unavailable telemetry |
Expand Down Expand Up @@ -345,6 +346,21 @@ Keyboard focus has its own outline and does not navigate until activation.
11. With reduced motion enabled, the hover fan changes without animation. Touch
navigation must not leave a hover fan behind.

### Accepted steering

1. Send immediately during a long tool/model round in the main transcript and
BTW pane. After host acceptance, verify the message uses the ordinary user-message
appearance without a pending label and leaves the composer queue. Confirm long text and attachment count.
2. Let the formal Turn start. Verify one message remains and
attachments appear normally. Send multiple messages and check ordering.
3. Exercise blocked handoff, history navigation, transcript closure, disconnect,
reconnect, host restart and device switching; recovery controls must remain
available and another device's preview must never appear.
4. Repeat on a remote workspace and Peer Device host with two controllers, one
sending from mobile. Mobile retains its existing queue UI. Check an older
host without display receipts retains the existing desktop queue UI. Detached
dispatch and ACP use their existing paths and have no preview requirement.

### History paging demand

1. Page upward at least three times from a long session's live tail, and page
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,26 @@
# FlowChat Virtualization

## Accepted steering display

The host queue's optional `steeringDisplay` receipt projects an accepted message
after its target Turn, without changing recorded rounds or writing a second
history record. Full display text and composer references come from the host;
pending attachments show their count rather than retransmitting image bytes.
Accepted text messages use the ordinary user-message appearance without a pending
status label; internal admission state still disables edit and rollback actions.
Formal Turn handoff keeps the original queue Turn id and the preview's virtual
key. Canonical history replaces the preview even when events precede the receipt.
Queue disappearance alone is not consumption: the observer queries the receipt
and retains the pending shell until history arrives. Blocked messages show a
not-applied label and restore composer recovery controls.

Only mounted transcript projections lease hidden queue rows. History windows
without the target, older hosts without display receipts, and closed transcripts
retain queue controls. Leases and previews are isolated by queue client, surface
activation and host queue epoch. Restart recovery remains with the existing
outbox; mobile retains its existing queue presentation. No scroll placement or
mount animation is added, and execution order remains host-owned.

## Measured row spacing

`VirtualItemRenderer` establishes a `flow-root` on the measured wrapper so
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -438,6 +438,34 @@ describe('UserMessageItem steering tag', () => {
expect(container.querySelector('.user-message-item__actions')?.parentElement).toBe(meta);
});

it.each(['pending', 'failed'] as const)('shows host queue %s feedback without edit or rollback actions', status => {
act(() => root.render(
<FlowChatContext.Provider value={{ allowUserMessageRollback: true, allowUserMessageEdit: true }}>
<UserMessageItem message={{ id: 'preview', content: 'Accepted instruction', timestamp: 1000 }}
turnId="queued-turn" hostQueueStatus={status} hostQueueAttachmentCount={2} />
</FlowChatContext.Provider>,
));
if (status === 'failed') expect(container.querySelector('.user-message-item__steering-tag')?.textContent)
.toContain('steering.statusNotApplied');
expect(container.querySelector('.user-message-item__steering-tag')?.textContent).toContain('hostQueue.attachments');
expect(container.querySelector('.user-message-item__edit-btn')).toBeNull();
expect(container.querySelector('.user-message-item__rollback-btn')).toBeNull();
expect(container.querySelector('.user-message-item__copy-btn')).not.toBeNull();
});

it('renders an accepted text message without a pending status label', () => {
act(() => root.render(
<FlowChatContext.Provider value={{ allowUserMessageRollback: true, allowUserMessageEdit: true }}>
<UserMessageItem message={{ id: 'preview-text', content: 'Accepted instruction', timestamp: 1000 }}
turnId="queued-turn" hostQueueStatus="pending" />
</FlowChatContext.Provider>,
));
expect(container.querySelector('.user-message-item__steering-tag')).toBeNull();
expect(container.textContent).toContain('Accepted instruction');
expect(container.querySelector('.user-message-item__edit-btn')).toBeNull();
expect(container.querySelector('.user-message-item__rollback-btn')).toBeNull();
});

it('does not render a steering tag after steering is triggered', () => {
act(() => {
root.render(
Expand Down
25 changes: 19 additions & 6 deletions src/web-ui/src/flow_chat/components/modern/UserMessageItem.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,8 @@ interface UserMessageItemProps {
steeringStatus?: FlowUserSteeringItem['status'];
submissionPhase?: 'forming' | 'failed';
submissionError?: string;
hostQueueStatus?: 'pending' | 'failed';
hostQueueAttachmentCount?: number;
}

function buildPresentationRerunPayload(presentation: ComposerPresentation): {
Expand Down Expand Up @@ -108,7 +110,7 @@ function buildPresentationRerunPayload(presentation: ComposerPresentation): {
}

export const UserMessageItem = React.memo<UserMessageItemProps>(
({ message, turnId, absoluteTurnIndex, turnStatus, steeringStatus, submissionPhase, submissionError }) => {
({ message, turnId, absoluteTurnIndex, turnStatus, steeringStatus, submissionPhase, submissionError, hostQueueStatus, hostQueueAttachmentCount }) => {
const { t, formatDate } = useI18n('flow-chat');
const {
sessionId,
Expand Down Expand Up @@ -203,7 +205,7 @@ export const UserMessageItem = React.memo<UserMessageItemProps>(
const isFailed = resolvedTurnStatus === 'error';
const resolvedSessionId = sessionId ?? currentSession?.sessionId;
useSubmittedMessageMotion(shellRef, resolvedSessionId, turnId, message?.id,
isFailed || isEditing || submissionPhase === 'failed', submissionPhase);
isFailed || isEditing || submissionPhase === 'failed' || !!hostQueueStatus, submissionPhase);
const sessionMachine = useSessionStateMachine(resolvedSessionId ?? null);
const sessionExecutionState = sessionMachine && sessionMachine.sessionId === resolvedSessionId
? sessionMachine.currentState
Expand All @@ -227,6 +229,7 @@ export const UserMessageItem = React.memo<UserMessageItemProps>(
);
const canRollback =
!steeringStatus &&
!hostQueueStatus &&
canShowRollbackAction &&
!!resolvedSessionId &&
actionTurnIndex >= 0 &&
Expand All @@ -241,9 +244,9 @@ export const UserMessageItem = React.memo<UserMessageItemProps>(
!isRemoteSession &&
!isThreadGoalSystemMessage &&
!isSystemTriggered &&
!steeringStatus;
!steeringStatus && !hostQueueStatus;
const canEdit = canEditBase && isSessionIdle && !isEditSubmitting && !sessionMutation;
const canShowEditAction = allowUserMessageEdit && !isFailed && !isThreadGoalSystemMessage;
const canShowEditAction = allowUserMessageEdit && !isFailed && !isThreadGoalSystemMessage && !hostQueueStatus;
const editDisabledReason = isDispatchSession
? t('message.editDisabledDispatch')
: isRemoteSession
Expand All @@ -266,7 +269,17 @@ export const UserMessageItem = React.memo<UserMessageItemProps>(
: !isSessionIdle
? t('message.rollbackDisabledBusy')
: t('message.cannotRollback');
const steeringTag = steeringStatus === 'pending'
const hostQueueLabel = hostQueueStatus
? [hostQueueStatus === 'failed' ? t('steering.statusNotApplied') : '',
hostQueueAttachmentCount ? t('hostQueue.attachments', { count: hostQueueAttachmentCount }) : '']
.filter(Boolean).join(' · ')
: '';
const steeringTag = hostQueueLabel
? {
className: 'user-message-item__steering-tag--pending',
label: hostQueueLabel,
}
: steeringStatus === 'pending'
? {
className: 'user-message-item__steering-tag--pending',
label: t('steering.statusPending'),
Expand Down Expand Up @@ -763,7 +776,7 @@ export const UserMessageItem = React.memo<UserMessageItemProps>(
icon={<Icon name="arrow-down" size="sm" />}
/>
</Tooltip>
) : canShowRollbackAction && !steeringStatus ? (
) : canShowRollbackAction && !steeringStatus && !hostQueueStatus ? (
<Tooltip content={rollbackTooltip}>
<IconButton
className="user-message-item__rollback-btn"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,8 @@ export const VirtualItemRenderer = React.memo<VirtualItemRendererProps>(
turnStatus={item.turnStatus}
submissionPhase={item.submissionPhase}
submissionError={item.submissionError}
hostQueueStatus={item.hostQueueStatus}
hostQueueAttachmentCount={item.hostQueueAttachmentCount}
/>
);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ export function getVirtualItemStableKey(item: VirtualItem): string {
if (item.timeline) return `block:${item.timeline.key}`;
switch (item.type) {
case 'user-message':
return item.presentationKey ?? `${item.type}:${item.turnId}:${item.data.id}`;
case 'user-steering-message':
return `${item.type}:${item.turnId}:${item.data.id}`;
case 'model-round':
Expand Down
Loading
Loading