}).queueSteerReceipts.size
- ).toBe(0);
await expect(service.steerQueuedMessage(request)).resolves.toMatchObject({
accepted: false,
@@ -795,7 +1155,7 @@ describe('SessionExecutionService', () => {
completedAt: expect.any(Number),
response: { disposition: 'no-active-turn', error: 'provider refused' },
});
- expect(upsertDocMeta.mock.calls.filter(([, patch]) => patch.latestUserMsgId)).toHaveLength(3);
+ expect(meta.latestUserMsgId).toBe('user:C');
const restartedService = new SessionExecutionService(deps);
await expect(restartedService.steerQueuedMessage(request)).resolves.toMatchObject({
@@ -803,7 +1163,7 @@ describe('SessionExecutionService', () => {
disposition: 'no-active-turn',
error: 'provider refused',
});
- expect(upsertDocMeta.mock.calls.filter(([, patch]) => patch.latestUserMsgId)).toHaveLength(3);
+ expect(meta.latestUserMsgId).toBe('user:C');
});
it.each(['submitting', 'acknowledged', 'applied'] as const)(
@@ -811,7 +1171,8 @@ describe('SessionExecutionService', () => {
async (phase) => {
const sessionId = `session-native-steer-crash-${phase}` as SessionId;
const operation = {
- version: 1 as const,
+ version: 2 as const,
+ queueRevision: queueItemRevision({ $cid: 'C' }),
workspaceId: 'workspace-1',
machineId: 'machine-1',
sessionId,
@@ -883,18 +1244,14 @@ describe('SessionExecutionService', () => {
expect(deps.recordChatFailure).toHaveBeenCalledWith(
sessionDoc,
'agent_disconnected',
- expect.stringContaining(phase === 'applied' ? 'was applied' : 'not replayed')
+ expect.stringContaining('not replayed')
);
}
);
it('recovers a crash after native reservation by dispatching that exact row', async () => {
const sessionId = 'session-native-steer-crash-reserved' as SessionId;
- let queue = [
- { $cid: 'A' },
- { $cid: 'B' },
- { $cid: 'C', isEditing: true, editingStartedAt: Number.MAX_SAFE_INTEGER },
- ];
+ let queue = [{ $cid: 'A' }, { $cid: 'B' }, { $cid: 'C' }];
let history: SessionHistoryInput[] = [
{
id: 'user:C',
@@ -935,6 +1292,7 @@ describe('SessionExecutionService', () => {
workspaceId: 'workspace-1',
machineId: 'machine-1',
sessionId,
+ queueRevision: queueItemRevision({ $cid: 'C' }),
operationKey: 'operation-reserved',
queueItemId: 'C',
expectedTurnId: 'assistant:old',
@@ -953,14 +1311,6 @@ describe('SessionExecutionService', () => {
);
await service.recoverPendingQueueSteers();
- expect(queue.map((item) => item.$cid)).toEqual(['A', 'B', 'C']);
- const editingMarker = await queueSteerOperationStore.read(sessionId);
- expect(editingMarker).toMatchObject({ phase: 'reserved' });
- expect(editingMarker?.completedAt).toBeUndefined();
-
- queue = queue.map((item) => (item.$cid === 'C' ? { ...item, isEditing: false } : item));
- await service.recoverPendingQueueSteers();
-
expect(history).toEqual([expect.objectContaining({ id: 'user:C', status: 'pending' })]);
expect(queue).toEqual([{ $cid: 'A' }, { $cid: 'B' }]);
expect(meta).toMatchObject({ latestUserMsgId: 'user:C' });
@@ -6372,7 +6722,7 @@ describe('SessionExecutionService', () => {
steerApplied.resolve({ release: () => steerReleased.resolve() });
await expect(steering).resolves.toMatchObject({
applied: false,
- disposition: 'stale-turn',
+ disposition: 'error',
});
await steerReleased.promise;
expect(onTurnSettled).not.toHaveBeenCalled();
diff --git a/locales/en.json b/locales/en.json
index aa14124ba..76d99cdda 100644
--- a/locales/en.json
+++ b/locales/en.json
@@ -1795,7 +1795,7 @@
"sessions.messageQueue.saveEdit": "Save changes (Enter)",
"sessions.messageQueue.title": "Queued messages",
"sessions.messageQueue.upNext": "Up next",
- "sessions.messageQueue.updateAgentForLaterSteer": "Update the local agent to steer a later queued message",
+ "sessions.messageQueue.draftDisplaced": "This message left the queue. Your unsaved draft is kept below; copy it before dismissing.",
"sessions.messageStatus.deliverNow": "Deliver now",
"sessions.messageStatus.deliverNowFailed": "Failed to deliver the message - please try again",
"sessions.messageStatus.notDelivered": "Not delivered",
diff --git a/locales/zh_CN.json b/locales/zh_CN.json
index fbfb2e4e5..630f009a2 100644
--- a/locales/zh_CN.json
+++ b/locales/zh_CN.json
@@ -1795,7 +1795,7 @@
"sessions.messageQueue.saveEdit": "保存修改(回车)",
"sessions.messageQueue.title": "排队中的消息",
"sessions.messageQueue.upNext": "接下来",
- "sessions.messageQueue.updateAgentForLaterSteer": "更新本地 Agent 后才能引导后续排队消息",
+ "sessions.messageQueue.draftDisplaced": "这条消息已离开队列。未保存的草稿仍保留在下方,请在关闭前复制。",
"sessions.messageStatus.deliverNow": "重新发送",
"sessions.messageStatus.deliverNowFailed": "送达失败,请重试",
"sessions.messageStatus.notDelivered": "未送达",
diff --git a/packages/components/src/components/sessions/message-queue/AGENTS.md b/packages/components/src/components/sessions/message-queue/AGENTS.md
index 2428e91f1..b444364fd 100644
--- a/packages/components/src/components/sessions/message-queue/AGENTS.md
+++ b/packages/components/src/components/sessions/message-queue/AGENTS.md
@@ -11,9 +11,9 @@ queued-turn list (`message-queue-display.tsx`, `message-queue-row.tsx`,
Exact-item Steer requires the negotiated `queueItemSteer` daemon protocol; missing means
unsupported. The daemon chooses acknowledged native Steer or exact cancel-and-dispatch.
-For older daemons, an authoritative `acknowledgedSteer` capability must retain the legacy
-native path; otherwise only queue-head interrupt stays enabled. Never reorder a later row
-to emulate Steer. A missing, stale, or actively edited exact target must leave the current
+No supported version means no Steer on any row, including the head; no legacy native/cancel path.
+Daemon-reserved rows reject edits/removal/reordering; retain a rejected edit's local draft even
+when the last row disappears. Never reorder to emulate Steer. A missing, stale, or edited target leaves the current
turn running. A row's number and message body are one drag activator; its actions stay out.
The queue intentionally stays OUT of the composer info bar
diff --git a/packages/components/src/components/sessions/message-queue/index.ts b/packages/components/src/components/sessions/message-queue/index.ts
index bc994ea31..078978312 100644
--- a/packages/components/src/components/sessions/message-queue/index.ts
+++ b/packages/components/src/components/sessions/message-queue/index.ts
@@ -4,11 +4,6 @@ export { MessageQueueRow } from './message-queue-row';
export type { MessageQueueRowProps } from './message-queue-row';
export { QueuedImagePreview } from './queued-image-preview';
export type { QueuedImageBlock } from './queued-image-preview';
-export {
- resolveQueuedMessageSteerRoute,
- shouldUseLegacyNativeQueueSteer,
- type QueuedMessageSteerRoute,
-} from './queued-message-steer-compat';
export {
useMessageQueueEditing,
getEditableTaskText,
diff --git a/packages/components/src/components/sessions/message-queue/message-queue-display.tsx b/packages/components/src/components/sessions/message-queue/message-queue-display.tsx
index 2e1255060..94e92a4c4 100644
--- a/packages/components/src/components/sessions/message-queue/message-queue-display.tsx
+++ b/packages/components/src/components/sessions/message-queue/message-queue-display.tsx
@@ -63,6 +63,11 @@ export function MessageQueueDisplay({
const [overflow, setOverflow] = useState(NO_SCROLL_EDGE_OVERFLOW);
const editing = useMessageQueueEditing(items, { onEditStart, onEditCancel, onEditSave });
+ const displacedDraft =
+ editing.editingItem && !items.some((item) => item.$cid === editing.editingCid)
+ ? editing.editingItem
+ : null;
+ const visibleItems = displacedDraft ? [...items, displacedDraft] : items;
const itemIds = useMemo(() => items.map((item) => item.$cid), [items]);
const canReorder = items.length > 1;
@@ -107,7 +112,7 @@ export function MessageQueueDisplay({
[onReorder]
);
- if (items.length === 0) {
+ if (items.length === 0 && !displacedDraft) {
return null;
}
@@ -146,13 +151,21 @@ export function MessageQueueDisplay({
WebkitMaskImage: fadeMask,
}}
>
+ {displacedDraft && (
+
+ {t(
+ 'sessions.messageQueue.draftDisplaced',
+ 'This message left the queue. Your unsaved draft is kept below; copy it before dismissing.'
+ )}
+
+ )}
- {items.map((item, index) => {
+ {visibleItems.map((item, index) => {
const isEditing = editing.editingCid === item.$cid;
return (
0}
steerDisabledReason={steerDisabledReason}
- canReorder={canReorder}
+ canReorder={canReorder && !displacedDraft}
isEditing={isEditing}
editValue={isEditing ? editing.editValue : ''}
isPending={editing.pendingCid === item.$cid}
diff --git a/packages/components/src/components/sessions/message-queue/queued-message-steer-compat.ts b/packages/components/src/components/sessions/message-queue/queued-message-steer-compat.ts
deleted file mode 100644
index a926a0136..000000000
--- a/packages/components/src/components/sessions/message-queue/queued-message-steer-compat.ts
+++ /dev/null
@@ -1,26 +0,0 @@
-import type { AcpCapabilityAuthority, AcpCapabilityCacheEntry } from '@lody/shared';
-
-/**
- * Older daemons have no exact-item queue RPC. Only their acknowledged ACP
- * capability can preserve true in-prompt steering; an absent or provisional
- * cache entry must not be guessed into support.
- */
-export function shouldUseLegacyNativeQueueSteer(
- authority: AcpCapabilityAuthority,
- capability: Pick | undefined
-): boolean {
- return authority === 'authoritative' && capability?.acknowledgedSteer === true;
-}
-
-export type QueuedMessageSteerRoute = 'exact-daemon' | 'legacy-native' | 'legacy-head';
-
-export function resolveQueuedMessageSteerRoute(options: {
- supportsExactDaemonProtocol: boolean;
- authority: AcpCapabilityAuthority;
- capability: Pick | undefined;
-}): QueuedMessageSteerRoute {
- if (options.supportsExactDaemonProtocol) return 'exact-daemon';
- return shouldUseLegacyNativeQueueSteer(options.authority, options.capability)
- ? 'legacy-native'
- : 'legacy-head';
-}
diff --git a/packages/components/src/components/sessions/message-queue/use-message-queue-editing.ts b/packages/components/src/components/sessions/message-queue/use-message-queue-editing.ts
index 90256c4bb..005415e18 100644
--- a/packages/components/src/components/sessions/message-queue/use-message-queue-editing.ts
+++ b/packages/components/src/components/sessions/message-queue/use-message-queue-editing.ts
@@ -19,6 +19,7 @@ export type MessageQueueEditingCallbacks = {
export type MessageQueueEditing = {
editingCid: string | null;
+ editingItem: MessageQueueItem | null;
editValue: string;
pendingCid: string | null;
setEditValue: (value: string) => void;
@@ -33,6 +34,7 @@ export function useMessageQueueEditing(
): MessageQueueEditing {
const { onEditStart, onEditCancel, onEditSave } = callbacks;
const [editingCid, setEditingCid] = useState(null);
+ const [editingItem, setEditingItem] = useState(null);
const [editValue, setEditValue] = useState('');
const [pendingCid, setPendingCid] = useState(null);
@@ -40,6 +42,7 @@ export function useMessageQueueEditing(
const itemsRef = useRef(items);
const editingCidRef = useRef(null);
const onEditCancelRef = useRef(onEditCancel);
+ const dismissedLeaseRef = useRef<{ cid: string; startedAt?: number } | null>(null);
useEffect(() => {
itemsRef.current = items;
@@ -55,31 +58,34 @@ export function useMessageQueueEditing(
useEffect(() => {
return () => {
const cid = editingCidRef.current;
- const item = cid
- ? itemsRef.current.find((candidate) => candidate.$cid === cid)
- : undefined;
+ const item = cid ? itemsRef.current.find((candidate) => candidate.$cid === cid) : undefined;
if (item?.isEditing) {
- void onEditCancelRef.current(item);
+ void Promise.resolve()
+ .then(() => onEditCancelRef.current(item))
+ .catch((error) => {
+ console.error('Failed to release queued message editing lease', error);
+ });
}
};
}, []);
- // Sync local editing state with server-side `isEditing` flag: if the row disappears or another
- // client opens an edit, reflect it.
+ // A displaced row must not discard an unsaved local draft.
useEffect(() => {
if (editingCid) {
- const item = items.find((candidate) => candidate.$cid === editingCid);
- if (!item) {
- setEditingCid(null);
- setEditValue('');
- }
return;
}
- const editingItem = items.find((item) => item.isEditing);
- if (editingItem) {
- setEditingCid(editingItem.$cid);
- setEditValue(getEditableTaskText(editingItem));
+ const sharedEditor = items.find((item) => item.isEditing);
+ if (sharedEditor) {
+ const dismissed = dismissedLeaseRef.current;
+ if (
+ dismissed?.cid === sharedEditor.$cid &&
+ dismissed.startedAt === sharedEditor.editingStartedAt
+ )
+ return;
+ setEditingCid(sharedEditor.$cid);
+ setEditingItem(sharedEditor);
+ setEditValue(getEditableTaskText(sharedEditor));
}
}, [editingCid, items]);
@@ -95,7 +101,9 @@ export function useMessageQueueEditing(
await onEditCancel(previous);
}
await onEditStart(item);
+ dismissedLeaseRef.current = null;
setEditingCid(item.$cid);
+ setEditingItem(item);
setEditValue(getEditableTaskText(item));
} catch (error) {
console.error('Failed to start queued message edit', error);
@@ -110,8 +118,11 @@ export function useMessageQueueEditing(
async (item: MessageQueueItem) => {
setPendingCid(item.$cid);
try {
- await onEditCancel(item);
+ if (itemsRef.current.some((candidate) => candidate.$cid === item.$cid))
+ await onEditCancel(item);
+ dismissedLeaseRef.current = { cid: item.$cid, startedAt: item.editingStartedAt };
setEditingCid(null);
+ setEditingItem(null);
setEditValue('');
} catch (error) {
console.error('Failed to cancel queued message edit', error);
@@ -127,7 +138,10 @@ export function useMessageQueueEditing(
setPendingCid(item.$cid);
try {
await onEditSave(item, editValue.trim());
+ // The daemon ACK can precede its CRDT delta; do not reopen the stale shared lease.
+ dismissedLeaseRef.current = { cid: item.$cid, startedAt: item.editingStartedAt };
setEditingCid(null);
+ setEditingItem(null);
setEditValue('');
} catch (error) {
console.error('Failed to save queued message edit', error);
@@ -140,6 +154,7 @@ export function useMessageQueueEditing(
return {
editingCid,
+ editingItem,
editValue,
pendingCid,
setEditValue,
diff --git a/packages/components/src/components/sessions/session-chat-interface.tsx b/packages/components/src/components/sessions/session-chat-interface.tsx
index 522ef450c..c436e3cb6 100644
--- a/packages/components/src/components/sessions/session-chat-interface.tsx
+++ b/packages/components/src/components/sessions/session-chat-interface.tsx
@@ -68,7 +68,7 @@ import {
type SessionTurnAgentRoleSelection,
} from './session-chat-input-area';
import { useSessionMcpSelection } from '@/hooks/use-session-mcp-selection';
-import { MessageQueueDisplay, resolveQueuedMessageSteerRoute } from './message-queue';
+import { MessageQueueDisplay } from './message-queue';
import { useTranslation } from 'react-i18next';
import { useRouter } from '@tanstack/react-router';
import { toast } from 'sonner';
@@ -5104,19 +5104,11 @@ export const SessionChatInterface = memo(
[captureSessionEvent, reorderMessageQueueItem, t]
);
- const queueSteerCapability = session.agentConfigId
- ? sessionMachine?.acpCapabilities?.[getAcpCapabilityCacheKey(session.agentConfigId)]
- : undefined;
- const queuedMessageSteerRoute = resolveQueuedMessageSteerRoute({
- supportsExactDaemonProtocol: machineSupportsQueueItemSteerProtocol(sessionMachine),
- authority: capabilityAuthority,
- capability: queueSteerCapability,
- });
- const legacyQueueSteerRequiresHead = queuedMessageSteerRoute === 'legacy-head';
+ const supportsQueueItemSteer = machineSupportsQueueItemSteerProtocol(sessionMachine);
- const handleExactQueueItemSteer = useCallback(
+ const handleSteerQueuedMessage = useCallback(
async (item: MessageQueueItem) => {
- if (isExternalHistoryRefreshing || !activeAssistantTurnId) {
+ if (isExternalHistoryRefreshing || !activeAssistantTurnId || !supportsQueueItemSteer) {
return;
}
if (steeringQueueItemIdsRef.current.has(item.$cid)) return;
@@ -5167,141 +5159,13 @@ export const SessionChatInterface = memo(
captureSessionEvent,
isExternalHistoryRefreshing,
requestSessionQueueSteer,
+ supportsQueueItemSteer,
session.id,
session.machineId,
t,
]
);
- const handleLegacyNativeQueueSteer = useCallback(
- async (item: MessageQueueItem) => {
- if (isExternalHistoryRefreshing || !activeAssistantTurnId) return;
- if (steeringQueueItemIdsRef.current.has(item.$cid)) return;
- steeringQueueItemIdsRef.current.add(item.$cid);
- try {
- const inputConfig = normalizeSessionTurnInputConfig(item.acpSessionConfig);
- const userId = item.userId?.trim() || currentUser?.id;
- if (!inputConfig || !userId) {
- throw new Error('Queued message input is invalid');
- }
- const inputBlocks = normalizeSessionInputBlocks(
- inputConfig.inputBlocks,
- inputConfig.prompt ?? item.task
- );
- const pendingHistoryEntry = buildPendingUserHistoryEntry({
- userId,
- inputBlocks,
- timestamp: item.timestamp,
- inputConfig,
- status: 'pending_apply',
- });
- if (!pendingHistoryEntry) throw new Error('Queued message is empty');
-
- const queuedUserTurnId = item.userTurnId?.trim() || `queued-${item.$cid}`;
- const { entry } = await addSessionHistory({
- ...pendingHistoryEntry,
- id: queuedUserTurnId,
- });
- await removeMessageQueueItem(item.$cid);
- trackMessageSend(entry.id);
- void touchSessionActivity(session.id).catch((error: unknown) => {
- console.warn('Failed to update session activity for steer', error);
- });
- const applied = await guideHistoryEntry(entry.id, activeAssistantTurnId);
- captureSessionEvent('session/queue_legacy_native_steer_result', {
- queue_item_id: item.$cid,
- active_assistant_turn_id: activeAssistantTurnId,
- applied,
- });
- } catch (error) {
- console.error('Failed to steer queued message through legacy native path', error);
- captureSessionEvent('session/queue_legacy_native_steer_failed', {
- queue_item_id: item.$cid,
- active_assistant_turn_id: activeAssistantTurnId,
- error_name: error instanceof Error ? error.name : typeof error,
- error_message: getErrorMessage(error),
- });
- toast.error(t('sessions.sendError'), { description: getErrorMessage(error) });
- } finally {
- steeringQueueItemIdsRef.current.delete(item.$cid);
- }
- },
- [
- activeAssistantTurnId,
- addSessionHistory,
- captureSessionEvent,
- currentUser?.id,
- guideHistoryEntry,
- isExternalHistoryRefreshing,
- removeMessageQueueItem,
- session.id,
- t,
- touchSessionActivity,
- trackMessageSend,
- ]
- );
-
- const handleLegacyHeadQueueSteer = useCallback(
- async (item: MessageQueueItem) => {
- if (
- isExternalHistoryRefreshing ||
- !activeAssistantTurnId ||
- messageQueue[0]?.$cid !== item.$cid
- ) {
- return;
- }
- setInputActionState('ready');
- pendingUserInterruptRef.current = true;
- try {
- await requestSessionCancel(session.id, activeAssistantTurnId);
- captureSessionEvent('session/queue_legacy_head_steer_succeeded', {
- queue_item_id: item.$cid,
- active_assistant_turn_id: activeAssistantTurnId,
- });
- } catch (error) {
- pendingUserInterruptRef.current = false;
- captureSessionEvent('session/queue_legacy_head_steer_failed', {
- queue_item_id: item.$cid,
- active_assistant_turn_id: activeAssistantTurnId,
- error_name: error instanceof Error ? error.name : typeof error,
- error_message: getErrorMessage(error),
- });
- toast.error(t('sessions.interruptFailed', 'Failed to interrupt current task'), {
- description: getErrorMessage(error),
- });
- }
- },
- [
- activeAssistantTurnId,
- captureSessionEvent,
- isExternalHistoryRefreshing,
- messageQueue,
- requestSessionCancel,
- session.id,
- t,
- ]
- );
-
- const handleSteerQueuedMessage = useCallback(
- async (item: MessageQueueItem) => {
- if (queuedMessageSteerRoute === 'exact-daemon') {
- await handleExactQueueItemSteer(item);
- return;
- }
- if (queuedMessageSteerRoute === 'legacy-native') {
- await handleLegacyNativeQueueSteer(item);
- return;
- }
- await handleLegacyHeadQueueSteer(item);
- },
- [
- handleExactQueueItemSteer,
- handleLegacyHeadQueueSteer,
- handleLegacyNativeQueueSteer,
- queuedMessageSteerRoute,
- ]
- );
-
const handleStartQueueItemEdit = useCallback(
async (item: MessageQueueItem) => {
const isFirstItem = messageQueue[0]?.$cid === item.$cid;
@@ -6201,28 +6065,24 @@ export const SessionChatInterface = memo(
commandsEnabled={isVisible}
freeTurnLimitNotice={freeSessionTurnNotice}
queueDisplay={
- messageQueue.length > 0 ? (
-
- ) : null
+
}
mcp={mcpSelection.menu}
skipNextViewportResizeAutoScrollRef={skipNextViewportResizeAutoScrollRef}
diff --git a/packages/components/src/hooks/use-session-doc.ts b/packages/components/src/hooks/use-session-doc.ts
index 313341a56..12b2f9f5f 100644
--- a/packages/components/src/hooks/use-session-doc.ts
+++ b/packages/components/src/hooks/use-session-doc.ts
@@ -1,3 +1,4 @@
+import { queueItemRevision } from '@lody/shared';
import { useCallback, useEffect, useMemo, useState } from 'react';
import { arrayMove } from '@dnd-kit/sortable';
import {
@@ -335,7 +336,9 @@ export function useSessionDoc(
const mq = await withStore((store) => (store.getState().mq ?? []) as MessageQueueItem[]);
const idx = mq.findIndex((item) => item.$cid === cid);
if (idx < 0) {
- return;
+ throw new Error(
+ 'This message is no longer in the editable queue. Your draft was not saved.'
+ );
}
const current = mq[idx] as MessageQueueItem;
const next = updater(current);
@@ -348,7 +351,8 @@ export function useSessionDoc(
await runtime.writer.updateSessionMessage(
sessionId,
cid,
- next as unknown as Record
+ next as unknown as Record,
+ queueItemRevision(current)
);
},
[withStore, runtime, sessionId]
@@ -371,7 +375,11 @@ export function useSessionDoc(
// Full-order is idempotent and robust across the intent wire; compute the
// resulting `$cid` order renderer-side.
const orderedItemIds = arrayMove(mq, fromIndex, toIndex).map((item) => item.$cid);
- await runtime.writer.reorderSessionMessages(sessionId, orderedItemIds);
+ await runtime.writer.reorderSessionMessages(
+ sessionId,
+ orderedItemIds,
+ mq.map((item) => item.$cid)
+ );
},
[withStore, runtime, sessionId]
);
diff --git a/packages/components/src/providers/AGENTS.md b/packages/components/src/providers/AGENTS.md
index 9aaa61d70..de3d4c568 100644
--- a/packages/components/src/providers/AGENTS.md
+++ b/packages/components/src/providers/AGENTS.md
@@ -53,6 +53,9 @@ and update only decision fields through HistoryWriter; never replace a rendered
- `create-workspace-runtime.ts` maintains one Repo view. `WorkspaceTargetRouter` owns
target ownership and transport selection; do not restore a second writer or a
proxy-authoring/write-intent mirror.
+- Queue edit/remove/reorder on queueItemSteer v2 use the narrow daemon domain RPC,
+ with revision checks against reservation ownership. Never direct-write after RPC failure.
+ Enqueue and other user writes retain their renderer authority.
- Repo storage, durable Streams cursors, and eager-sync high-water state must use the
same per-renderer cache namespace. A checkpoint must never be shared by independently
persisted Repo views.
diff --git a/packages/components/src/providers/create-workspace-runtime.ts b/packages/components/src/providers/create-workspace-runtime.ts
index 5e24f3fdf..cba1ad310 100644
--- a/packages/components/src/providers/create-workspace-runtime.ts
+++ b/packages/components/src/providers/create-workspace-runtime.ts
@@ -1,3 +1,4 @@
+import { machineSupportsQueueItemSteerProtocol, type SessionMeta } from '@lody/shared';
import { jotaiStore } from '@/lib/utils';
import { desktopWindowId } from '@/lib/desktop-window';
import { navigationSidebarHiddenAtom } from '@/atoms/layout-state';
@@ -1689,6 +1690,7 @@ export async function createWorkspaceRuntime(deps: RuntimeDeps): Promise repo.unloadDoc(getTaskRoomId(taskId)),
});
- // Dual-author: every client direct-authors its own durable writes and uploads
- // them over its own cloud connection; local targets additionally converge with
- // the CLI over the local plane (specs/local-first-two-plane.md 作者规则).
+ // Queue controls share the daemon's reservation authority; other user writes remain local.
const workspaceWriter = createDirectWorkspaceWriter({
repo,
+ mutateQueue: async (request) => {
+ const session = await repo.getDocMeta(getSessionRoomId(request.sessionId));
+ const machineId = (session?.meta as Partial | undefined)?.machineId;
+ if (!machineId) throw new Error('Queue owner is unavailable.');
+ const machine = await repo.getDocMeta(getMachineRoomId(machineId));
+ if (!machine) throw new Error('Queue owner capabilities are unavailable.');
+ const protocolCapabilities = (machine.meta as Partial).protocolCapabilities;
+ if (!machineSupportsQueueItemSteerProtocol({ protocolCapabilities })) return false;
+ const response = await requestSessionQueueMutation(machineId, request);
+ if (!response?.success)
+ throw new Error(response?.error ?? 'Queue change could not be confirmed.');
+ return true;
+ },
acquireSessionStore: sessionStoreCache.acquire,
releaseSessionStoreRef: sessionStoreCache.releaseRef,
acquirePreviewVisualCommentStore: previewVisualCommentStoreCache.acquire,
diff --git a/packages/components/src/providers/workspace-machine-rpc-facade.ts b/packages/components/src/providers/workspace-machine-rpc-facade.ts
index f7138862c..3d2fda48b 100644
--- a/packages/components/src/providers/workspace-machine-rpc-facade.ts
+++ b/packages/components/src/providers/workspace-machine-rpc-facade.ts
@@ -1,3 +1,4 @@
+import type { SessionQueueMutation, SessionQueueMutationResponse } from '@lody/shared';
import type { LocalFilePreviewResource } from '@lody/shared/local-file-preview';
import type {
LoroStreamsMachineRpcClient,
@@ -770,6 +771,40 @@ export function createWorkspaceMachineRpcFacade(deps: WorkspaceMachineRpcFacadeD
}
};
+ const requestSessionQueueMutation = async (
+ machineId: MachineId,
+ args: SessionQueueMutation
+ ): Promise => {
+ try {
+ const protocolCapabilities = await deps.getMachineProtocolCapabilities(machineId);
+ if (!machineSupportsQueueItemSteerProtocol({ protocolCapabilities })) {
+ throw new Error('This daemon does not support queue ownership controls.');
+ }
+ if (await canUseLocalMachineRpc(machineId)) {
+ const response = await getLocalMachineRpcSender()?.({
+ machineId,
+ workspaceId,
+ method: 'session/queue-mutate',
+ params: args,
+ timeoutMs: 5_000,
+ });
+ if (!response) throw new Error('Local queue control is unavailable.');
+ if (!response.ok) throw new Error(response.error);
+ return response.result as SessionQueueMutationResponse;
+ }
+ if (!deps.getAuthorizedMachineIds?.()?.has(machineId)) {
+ throw new Error('Source authorization for this machine is unavailable or denied.');
+ }
+ return await (await getMachineRpcClient(machineId)).requestSessionQueueMutation(args);
+ } catch (error) {
+ return {
+ type: 'session/queue-mutate_response',
+ success: false,
+ error: error instanceof Error ? error.message : String(error),
+ };
+ }
+ };
+
const requestSessionQueueSteer = async (
machineId: MachineId,
args: {
@@ -810,6 +845,7 @@ export function createWorkspaceMachineRpcFacade(deps: WorkspaceMachineRpcFacadeD
};
}
if (response?.ok) return response.result as SessionQueueSteerResponse;
+ throw new Error('Local queue control is unavailable.');
}
const authorizedMachineIds = deps.getAuthorizedMachineIds?.() ?? null;
if (!authorizedMachineIds?.has(machineId)) {
@@ -1247,6 +1283,7 @@ export function createWorkspaceMachineRpcFacade(deps: WorkspaceMachineRpcFacadeD
return {
requestSessionCancel,
+ requestSessionQueueMutation,
requestSessionQueueSteer,
requestSessionSteer,
requestSessionGoal,
diff --git a/packages/components/src/providers/workspace-writer-impl.ts b/packages/components/src/providers/workspace-writer-impl.ts
index 88bec2a1d..5f50f2980 100644
--- a/packages/components/src/providers/workspace-writer-impl.ts
+++ b/packages/components/src/providers/workspace-writer-impl.ts
@@ -2,6 +2,8 @@ import {
applyPreviewVisualCommentMutation,
getServerNow,
getSessionRoomId,
+ queueItemRevision,
+ type SessionQueueMutation,
type MessageQueueItem,
type PreviewVisualCommentDocInput,
} from '@lody/shared';
@@ -16,24 +18,21 @@ import type { WorkspaceWriter } from './workspace-writer';
// # WorkspaceWriter implementation
//
-// Dual-author: every client authors the mutation against its own repo / session
-// stores (identical for Web, Mobile, and Electron; see `workspace-writer.ts`).
-// A pure factory with injected deps so hooks stay agnostic to the runtime.
+// Renderer authorship with a narrow daemon-owned queue-control exception.
+// Injected deps keep transport and capability policy out of hooks.
/** Deps the writer needs from the runtime (repo + session stores). */
export type DirectWorkspaceWriterDeps = {
repo: LoroRepo;
+ /** False means a confirmed legacy target; rejection never permits a direct write. */
+ mutateQueue?: (request: SessionQueueMutation) => Promise;
acquireSessionStore: (sessionId: SessionId) => Promise;
releaseSessionStoreRef: (sessionId: SessionId) => void;
acquirePreviewVisualCommentStore: (sessionId: SessionId) => Promise;
releasePreviewVisualCommentStoreRef: (sessionId: SessionId) => void;
};
-/**
- * Direct-mode writer (web/cloud): applies each mutation to the renderer's own
- * repo / session stores. This is exactly what the hooks did before the seam, so
- * there is zero behavior change in cloud mode.
- */
+/** Ordinary writes stay local; supported queue controls await authoritative daemon acceptance. */
export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): WorkspaceWriter {
const withSessionStore = async (
sessionId: string,
@@ -182,17 +181,47 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo
},
async removeSessionMessage(sessionId, itemId) {
- await withSessionStore(sessionId, (store) => {
+ const remote = await withSessionStore(sessionId, async (store) => {
+ const row = store.getState().mq?.find((item) => item.$cid === itemId);
+ if (!row) throw new Error('This message is no longer in the editable queue.');
+ if (
+ await deps.mutateQueue?.({
+ sessionId: sessionId as SessionId,
+ mutation: {
+ kind: 'remove',
+ queueItemId: itemId,
+ expectedRevision: queueItemRevision(row),
+ },
+ })
+ )
+ return true;
store.setState((draft: SessionDocDraft) => {
const mq = (draft.mq ?? []) as MessageQueueItem[];
draft.mq = mq.filter((item) => item.$cid !== itemId);
});
});
- await bumpMessageQueueWatermark(sessionId);
+ if (!remote) await bumpMessageQueueWatermark(sessionId);
},
- async updateSessionMessage(sessionId, itemId, patch) {
- await withSessionStore(sessionId, (store) => {
+ async updateSessionMessage(sessionId, itemId, patch, expectedRevision) {
+ const remote = await withSessionStore(sessionId, async (store) => {
+ const row = store.getState().mq?.find((item) => item.$cid === itemId);
+ if (!row)
+ throw new Error(
+ 'This message is no longer in the editable queue. Your draft was not saved.'
+ );
+ if (
+ await deps.mutateQueue?.({
+ sessionId: sessionId as SessionId,
+ mutation: {
+ kind: 'update',
+ queueItemId: itemId,
+ expectedRevision: expectedRevision ?? queueItemRevision(row),
+ patch,
+ },
+ })
+ )
+ return true;
store.setState((draft: SessionDocDraft) => {
const mq = (draft.mq ?? []) as MessageQueueItem[];
draft.mq = mq.map((item) =>
@@ -202,11 +231,21 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo
);
});
});
- await bumpMessageQueueWatermark(sessionId);
+ if (!remote) await bumpMessageQueueWatermark(sessionId);
},
- async reorderSessionMessages(sessionId, orderedItemIds) {
- await withSessionStore(sessionId, (store) => {
+ async reorderSessionMessages(sessionId, orderedItemIds, expectedIds) {
+ const remote = await withSessionStore(sessionId, async (store) => {
+ const expectedItemIds = expectedIds
+ ? [...expectedIds]
+ : (store.getState().mq ?? []).map((item) => item.$cid!);
+ if (
+ await deps.mutateQueue?.({
+ sessionId: sessionId as SessionId,
+ mutation: { kind: 'reorder', orderedItemIds: [...orderedItemIds], expectedItemIds },
+ })
+ )
+ return true;
store.setState((draft: SessionDocDraft) => {
const mq = (draft.mq ?? []) as MessageQueueItem[];
const byCid = new Map(mq.map((item) => [item.$cid, item] as const));
@@ -226,7 +265,7 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo
draft.mq = ordered;
});
});
- await bumpMessageQueueWatermark(sessionId);
+ if (!remote) await bumpMessageQueueWatermark(sessionId);
},
async mutatePreviewVisualComments(sessionId, mutation) {
diff --git a/packages/components/src/providers/workspace-writer.ts b/packages/components/src/providers/workspace-writer.ts
index 029433be0..4bf78c43b 100644
--- a/packages/components/src/providers/workspace-writer.ts
+++ b/packages/components/src/providers/workspace-writer.ts
@@ -15,6 +15,8 @@ import type {
// uploads them over its own cloud connection; for local targets the local data
// plane converges the same ops with the CLI. The seam stays so hooks depend on
// one narrow mutation surface rather than raw repo/store handles.
+// Queue edit/remove/reorder on queueItemSteer v2 instead require the daemon's
+// revision-checked domain operation; failures never authorize a local fallback.
export interface WorkspaceWriter {
/** `repo.upsertDocMeta(roomId, patch)` — session/machine doc-meta write. */
upsertDocMeta(roomId: string, patch: Record): Promise;
@@ -116,9 +118,14 @@ export interface WorkspaceWriter {
updateSessionMessage(
sessionId: string,
itemId: string,
- patch: Record
+ patch: Record,
+ expectedRevision?: string
+ ): Promise;
+ reorderSessionMessages(
+ sessionId: string,
+ orderedItemIds: readonly string[],
+ expectedItemIds?: readonly string[]
): Promise;
- reorderSessionMessages(sessionId: string, orderedItemIds: readonly string[]): Promise;
/** Mutate the dedicated preview-comment doc (renderer-authored user data). */
mutatePreviewVisualComments(
diff --git a/packages/components/tests/message-queue-row-editing.test.tsx b/packages/components/tests/message-queue-row-editing.test.tsx
index d9d19d2c8..acf4e0fc0 100644
--- a/packages/components/tests/message-queue-row-editing.test.tsx
+++ b/packages/components/tests/message-queue-row-editing.test.tsx
@@ -133,6 +133,56 @@ describe('queued message editing commits', () => {
expect(saved).toEqual([{ cid: 'cid-0', task: 'Rewrite the queue instead' }]);
});
+ it('retains an unsaved draft when another client reserves the last queue row', async () => {
+ const item = makeItem();
+ const render = async (items: MessageQueueItem[]) => {
+ await act(async () => {
+ root?.render(
+ createElement(MessageQueueDisplay, {
+ sessionId: 'session-test' as SessionId,
+ items,
+ onRemove: () => {},
+ onReorder: () => {},
+ onSteer: () => {},
+ onEditStart: () => {},
+ onEditCancel: () => {},
+ onEditSave: async () => {
+ throw new Error('Message is operation-owned');
+ },
+ })
+ );
+ });
+ };
+ await render([item]);
+ const textarea = await startEditing(container!);
+ await act(async () => setTextareaValue(textarea, 'Roll back instead'));
+ await render([]);
+ expect(container?.querySelector('textarea')?.value).toBe('Roll back instead');
+ expect(container?.querySelector('[role="alert"]')?.textContent).toContain('unsaved draft');
+ await pressEnter(container!.querySelector('textarea')!);
+ expect(container?.querySelector('textarea')?.value).toBe('Roll back instead');
+ });
+
+ it('does not reopen an old editing lease when the save ACK arrives before replication', async () => {
+ const item = { ...makeItem(), isEditing: true, editingStartedAt: 1 };
+ await act(async () => {
+ root?.render(
+ createElement(MessageQueueDisplay, {
+ sessionId: 'session-test' as SessionId,
+ items: [item],
+ onRemove: () => {},
+ onReorder: () => {},
+ onSteer: () => {},
+ onEditStart: () => {},
+ onEditCancel: () => {},
+ onEditSave: () => {},
+ })
+ );
+ });
+ await pressEnter(container!.querySelector('textarea')!);
+ expect(container?.querySelector('textarea')).toBeNull();
+ });
+
it('focuses the editor when the editing flag arrives before the start write completes', async () => {
let finishStart!: () => void;
function QueueWithEarlyUpdate() {
diff --git a/packages/components/tests/queued-message-steer.test.ts b/packages/components/tests/queued-message-steer.test.ts
deleted file mode 100644
index 48ededdae..000000000
--- a/packages/components/tests/queued-message-steer.test.ts
+++ /dev/null
@@ -1,49 +0,0 @@
-import { describe, expect, it } from 'vitest';
-import {
- resolveQueuedMessageSteerRoute,
- shouldUseLegacyNativeQueueSteer,
-} from '../src/components/sessions/message-queue';
-
-describe('legacy queued-message steering', () => {
- it('uses true native steering only for an authoritative acknowledged capability', () => {
- expect(shouldUseLegacyNativeQueueSteer('authoritative', { acknowledgedSteer: true })).toBe(
- true
- );
- expect(shouldUseLegacyNativeQueueSteer('authoritative', { acknowledgedSteer: false })).toBe(
- false
- );
- expect(shouldUseLegacyNativeQueueSteer('provisional', { acknowledgedSteer: true })).toBe(false);
- expect(shouldUseLegacyNativeQueueSteer('unavailable', undefined)).toBe(false);
- });
-
- it('routes new renderers across new and old daemon capability combinations', () => {
- expect(
- resolveQueuedMessageSteerRoute({
- supportsExactDaemonProtocol: true,
- authority: 'authoritative',
- capability: { acknowledgedSteer: true },
- })
- ).toBe('exact-daemon');
- expect(
- resolveQueuedMessageSteerRoute({
- supportsExactDaemonProtocol: false,
- authority: 'authoritative',
- capability: { acknowledgedSteer: true },
- })
- ).toBe('legacy-native');
- expect(
- resolveQueuedMessageSteerRoute({
- supportsExactDaemonProtocol: false,
- authority: 'authoritative',
- capability: { acknowledgedSteer: false },
- })
- ).toBe('legacy-head');
- expect(
- resolveQueuedMessageSteerRoute({
- supportsExactDaemonProtocol: false,
- authority: 'provisional',
- capability: { acknowledgedSteer: true },
- })
- ).toBe('legacy-head');
- });
-});
diff --git a/packages/components/tests/workspace-machine-rpc-facade.test.ts b/packages/components/tests/workspace-machine-rpc-facade.test.ts
index c4c459eb4..7af58e6ca 100644
--- a/packages/components/tests/workspace-machine-rpc-facade.test.ts
+++ b/packages/components/tests/workspace-machine-rpc-facade.test.ts
@@ -17,6 +17,64 @@ afterEach(() => {
});
describe('createWorkspaceMachineRpcFacade', () => {
+ it.each([undefined, { queueItemSteer: 1 }])(
+ 'disables every old-daemon queue row for %j',
+ async (capabilities) => {
+ const facade = createWorkspaceMachineRpcFacade({
+ workspaceId,
+ getMachineProtocolCapabilities: async () => capabilities,
+ targetRouter: {
+ getPlaneForMachine: () => 'remote',
+ resolvePlaneForMachine: async () => 'remote',
+ },
+ getMachineRpcClient: async () => {
+ throw new Error('Legacy delivery must not run');
+ },
+ });
+ for (const queueItemId of ['A', 'B', 'C']) {
+ await expect(
+ facade.requestSessionQueueSteer(remoteMachineId, {
+ sessionId,
+ expectedTurnId: 'active',
+ queueItemId,
+ })
+ ).resolves.toMatchObject({ accepted: false, disposition: 'unsupported', queueItemId });
+ }
+ }
+ );
+
+ it('returns a queue mutation failure from local IPC without trying remote delivery', async () => {
+ vi.stubGlobal('window', {
+ __LODY_ELECTRON__: true,
+ ipc: {
+ invoke: async () => ({
+ ok: true,
+ result: {
+ type: 'session/queue-mutate_response',
+ success: false,
+ error: 'Row is reserved',
+ },
+ }),
+ },
+ });
+ const facade = createWorkspaceMachineRpcFacade({
+ workspaceId,
+ getMachineProtocolCapabilities: async () => CURRENT_MACHINE_PROTOCOL_CAPABILITIES,
+ targetRouter: {
+ getPlaneForMachine: () => 'local',
+ resolvePlaneForMachine: async () => 'local',
+ },
+ getMachineRpcClient: async () => {
+ throw new Error('Must not use remote');
+ },
+ });
+ await expect(
+ facade.requestSessionQueueMutation(localMachineId, {
+ sessionId,
+ mutation: { kind: 'remove', queueItemId: 'C', expectedRevision: '{}' },
+ })
+ ).resolves.toMatchObject({ success: false, error: 'Row is reserved' });
+ });
it('sends queued Steer as one exact local daemon operation', async () => {
const invoke = vi.fn(async () => ({
ok: true as const,
diff --git a/packages/components/tests/workspace-writer.test.ts b/packages/components/tests/workspace-writer.test.ts
index f8f360283..33f673815 100644
--- a/packages/components/tests/workspace-writer.test.ts
+++ b/packages/components/tests/workspace-writer.test.ts
@@ -46,6 +46,43 @@ const anchor: MinimalVisualAnnotationAnchor = {
};
describe('createDirectWorkspaceWriter', () => {
+ it.each(['update', 'remove', 'reorder'] as const)(
+ 'does not direct-write after daemon rejects queue %s',
+ async (kind) => {
+ const row = { $cid: 'C', task: 'Repair database' };
+ const state = { mq: [row] };
+ let references = 0;
+ const writer = createDirectWorkspaceWriter({
+ repo: { upsertDocMeta: async () => {} } as never,
+ acquireSessionStore: async () => {
+ references++;
+ return {
+ getState: () => state,
+ setState: (update: (draft: typeof state) => void) => update(state),
+ } as never;
+ },
+ releaseSessionStoreRef: () => {
+ references--;
+ },
+ acquirePreviewVisualCommentStore: async () => {
+ throw new Error('unused');
+ },
+ releasePreviewVisualCommentStoreRef: () => {},
+ mutateQueue: async () => {
+ throw new Error('Queue item is reserved');
+ },
+ });
+ const request =
+ kind === 'update'
+ ? writer.updateSessionMessage('session', 'C', { task: 'Rollback' })
+ : kind === 'remove'
+ ? writer.removeSessionMessage('session', 'C')
+ : writer.reorderSessionMessages('session', ['C']);
+ await expect(request).rejects.toThrow('reserved');
+ expect(state).toEqual({ mq: [{ $cid: 'C', task: 'Repair database' }] });
+ expect(references).toBe(0);
+ }
+ );
it.each(['unchanged', 'edited', 'deleted', 'cancelled', 'other-owner', 'write-failure'] as const)(
'reconciles the durable role without overwriting intervening changes: %s',
async (scenario) => {
diff --git a/packages/loro-streams-rpc/AGENTS.md b/packages/loro-streams-rpc/AGENTS.md
index 83114c45d..910dc680a 100644
--- a/packages/loro-streams-rpc/AGENTS.md
+++ b/packages/loro-streams-rpc/AGENTS.md
@@ -46,7 +46,7 @@ and the `file/preview` namespace are in
needs read-check-write atomicity serializes in its own service layer (Code Collab
`save-text` per absolute path in `code-collab-v2-service.ts`), not in the request loop.
- Control-plane methods (`machine/status`, `machine/ping`, `session/cancel`,
- `session/live-status`, `session/queue-steer`, `session/steer`, `session/terminate`, `machine/restart`,
+ `session/live-status`, `session/queue-steer`, `session/queue-mutate`, `session/steer`, `session/terminate`, `machine/restart`,
`machine/upgrade`, `session/dispatch-turn`)
bypass the shared semaphore and run on a small dedicated lane
(`CONTROL_METHODS` in `machine-rpc-server.ts`) so saturated code-collab
diff --git a/packages/loro-streams-rpc/src/machine-rpc-server.ts b/packages/loro-streams-rpc/src/machine-rpc-server.ts
index 11d904b96..9e10e8e71 100644
--- a/packages/loro-streams-rpc/src/machine-rpc-server.ts
+++ b/packages/loro-streams-rpc/src/machine-rpc-server.ts
@@ -1,3 +1,4 @@
+import type { SessionQueueMutation, SessionQueueMutationResponse } from '@lody/shared';
import type {
AgentConfigId,
CodeCollabV2InitDirectoryOk,
@@ -119,6 +120,7 @@ const CONTROL_METHODS: ReadonlySet = new Set([
'machine/acp-capabilities-refresh-cancel',
'session/cancel',
'session/live-status',
+ 'session/queue-mutate',
'session/queue-steer',
'session/steer',
'session/goal',
@@ -352,6 +354,7 @@ type RpcServerDeps = {
getSessionLiveStatus?: (args: {
sessionId: SessionId;
}) => Promise;
+ mutateQueuedMessage?: (args: SessionQueueMutation) => Promise;
steerQueuedMessage?: (args: {
sessionId: SessionId;
expectedTurnId: string;
@@ -1114,6 +1117,17 @@ export class LoroStreamsMachineRpcServer {
await this.appendResultResponse(request.replyTo, request.id, request.method, response);
return;
}
+ case 'session/queue-mutate': {
+ const response: SessionQueueMutationResponse = this.deps.mutateQueuedMessage
+ ? await this.deps.mutateQueuedMessage(request.params)
+ : {
+ type: 'session/queue-mutate_response',
+ success: false,
+ error: 'Queue mutation is unavailable on this machine.',
+ };
+ await this.appendResultResponse(request.replyTo, request.id, request.method, response);
+ return;
+ }
case 'session/queue-steer': {
if (!this.deps.steerQueuedMessage) {
await this.appendErrorResponse(request.replyTo, request.id, request.method, {
@@ -1638,6 +1652,7 @@ export class LoroStreamsMachineRpcServer {
| MachineBugReportResponse
| SessionCancelResponse
| LoroSessionLiveStatusRpcResponse
+ | SessionQueueMutationResponse
| SessionQueueSteerResponse
| SessionSteerResponse
| SessionGoalResponse
diff --git a/packages/loro-streams-rpc/src/rpc.ts b/packages/loro-streams-rpc/src/rpc.ts
index 6e44f05a0..5b0fc8aef 100644
--- a/packages/loro-streams-rpc/src/rpc.ts
+++ b/packages/loro-streams-rpc/src/rpc.ts
@@ -1,3 +1,9 @@
+import {
+ SessionQueueMutationSchema,
+ SessionQueueMutationResponseSchema,
+ type SessionQueueMutation,
+ type SessionQueueMutationResponse,
+} from '@lody/shared';
import { z } from 'zod';
import {
StreamsClient,
@@ -194,6 +200,7 @@ export const LoroStreamsRpcMethodSchema = z.enum([
'file/preview',
'session/cancel',
'session/live-status',
+ 'session/queue-mutate',
'session/queue-steer',
'session/steer',
'session/goal',
@@ -475,6 +482,11 @@ export const LoroSessionSteerRpcRequestSchema = BaseRpcRequestSchema.extend({
.strict(),
}).strict();
+export const LoroSessionQueueMutationRpcRequestSchema = BaseRpcRequestSchema.extend({
+ method: z.literal('session/queue-mutate'),
+ params: SessionQueueMutationSchema,
+}).strict();
+
export const LoroSessionQueueSteerRpcRequestSchema = BaseRpcRequestSchema.extend({
method: z.literal('session/queue-steer'),
params: z
@@ -613,6 +625,7 @@ export const LoroStreamsRpcRequestSchema = z.discriminatedUnion('method', [
LoroFilePreviewRpcRequestSchema,
LoroSessionCancelRpcRequestSchema,
LoroSessionLiveStatusRpcRequestSchema,
+ LoroSessionQueueMutationRpcRequestSchema,
LoroSessionQueueSteerRpcRequestSchema,
LoroSessionSteerRpcRequestSchema,
LoroSessionGoalRpcRequestSchema,
@@ -1457,6 +1470,7 @@ export type LoroMachineRpcResult =
| MachineBugReportResponse
| SessionCancelResponse
| LoroSessionLiveStatusRpcResponse
+ | SessionQueueMutationResponse
| SessionQueueSteerResponse
| SessionSteerResponse
| SessionGoalResponse
@@ -1663,6 +1677,9 @@ const toLegacyRpcErrorResponse = (
};
}
+ if (method === 'session/queue-mutate') {
+ return { type: 'session/queue-mutate_response', success: false, error: error.message };
+ }
if (method === 'session/queue-steer') {
return {
type: 'session/queue-steer_response',
@@ -1848,6 +1865,10 @@ const parseRpcSuccessResult = async (
const parsed = SessionSteerResponseSchema.safeParse(response.result);
return parsed.success ? (parsed.data as SessionSteerResponse) : null;
}
+ if (response.method === 'session/queue-mutate') {
+ const parsed = SessionQueueMutationResponseSchema.safeParse(response.result);
+ return parsed.success ? parsed.data : null;
+ }
if (response.method === 'session/queue-steer') {
const parsed = SessionQueueSteerResponseSchema.safeParse(response.result);
return parsed.success ? (parsed.data as SessionQueueSteerResponse) : null;
@@ -2727,6 +2748,16 @@ export class LoroStreamsMachineRpcClient {
})) as SessionSteerResponse | null;
}
+ async requestSessionQueueMutation(
+ options: SessionQueueMutation & { timeoutMs?: number }
+ ): Promise {
+ return (await this.sendRequest({
+ method: 'session/queue-mutate',
+ timeoutMs: options.timeoutMs ?? 5000,
+ params: { sessionId: options.sessionId, mutation: options.mutation },
+ })) as SessionQueueMutationResponse | null;
+ }
+
async requestSessionQueueSteer(options: {
sessionId: SessionId;
expectedTurnId: string;
@@ -3221,6 +3252,11 @@ export class LoroStreamsMachineRpcClient {
sessionId: string;
};
}
+ | {
+ method: 'session/queue-mutate';
+ timeoutMs: number;
+ params: SessionQueueMutation;
+ }
| {
method: 'session/queue-steer';
timeoutMs: number;
@@ -3577,6 +3613,9 @@ export class LoroStreamsMachineRpcClient {
case 'session/live-status':
request = { ...envelope, method: args.method, params: args.params };
break;
+ case 'session/queue-mutate':
+ request = { ...envelope, method: args.method, params: args.params };
+ break;
case 'session/queue-steer':
request = { ...envelope, method: args.method, params: args.params };
break;
diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts
index 5840015a4..b16356f28 100644
--- a/packages/shared/src/index.ts
+++ b/packages/shared/src/index.ts
@@ -474,3 +474,4 @@ export interface Attachment {
uploadedAt: Date;
uploadedBy: User;
}
+export * from './session-queue-mutation';
diff --git a/packages/shared/src/local-machine-rpc.ts b/packages/shared/src/local-machine-rpc.ts
index 8b2c232df..b5b3a2580 100644
--- a/packages/shared/src/local-machine-rpc.ts
+++ b/packages/shared/src/local-machine-rpc.ts
@@ -1,3 +1,7 @@
+import {
+ SessionQueueMutationSchema,
+ SessionQueueMutationResponseSchema,
+} from './session-queue-mutation';
import { LocalFileResolutionSchema } from './local-file-preview';
import { z } from 'zod';
import { SESSION_GOAL_ACTIONS } from './goal';
@@ -194,6 +198,10 @@ export const LocalMachineRpcRequestSchema = z.discriminatedUnion('method', [
method: z.literal('session/prepare-cancel'),
params: SessionPreparationCancelSpecSchema,
}).strict(),
+ BaseLocalMachineRpcRequestSchema.extend({
+ method: z.literal('session/queue-mutate'),
+ params: SessionQueueMutationSchema,
+ }).strict(),
BaseLocalMachineRpcRequestSchema.extend({
method: z.literal('session/queue-steer'),
params: z
@@ -261,6 +269,7 @@ export type LocalMachineRpcRequest = z.infer;
+export type SessionQueueMutationResponse = z.infer;
+
+/** Canonical JSON identity for compare-and-update across independent replicas. */
+export function queueItemRevision(value: unknown): string {
+ return JSON.stringify(value, (_key, item: unknown) => {
+ if (item && typeof item === 'object' && !Array.isArray(item)) {
+ return Object.fromEntries(
+ Object.entries(item).sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))
+ );
+ }
+ return item;
+ });
+}
diff --git a/packages/shared/tests/local-machine-rpc.test.ts b/packages/shared/tests/local-machine-rpc.test.ts
index 5333191b7..67b398b5a 100644
--- a/packages/shared/tests/local-machine-rpc.test.ts
+++ b/packages/shared/tests/local-machine-rpc.test.ts
@@ -6,6 +6,32 @@ import {
describe('local Machine RPC', () => {
it.each([
+ {
+ method: 'session/queue-mutate',
+ params: {
+ sessionId: 'session-1',
+ mutation: {
+ kind: 'update',
+ queueItemId: 'queue-C',
+ expectedRevision: '{}',
+ patch: { task: 'rollback' },
+ },
+ },
+ },
+ {
+ method: 'session/queue-mutate',
+ params: {
+ sessionId: 'session-1',
+ mutation: { kind: 'remove', queueItemId: 'queue-C', expectedRevision: '{}' },
+ },
+ },
+ {
+ method: 'session/queue-mutate',
+ params: {
+ sessionId: 'session-1',
+ mutation: { kind: 'reorder', expectedItemIds: ['A', 'B'], orderedItemIds: ['B', 'A'] },
+ },
+ },
{
method: 'session/get-active-invocation-context',
params: { sessionId: 'session-1' },
diff --git a/packages/shared/tests/machine-protocol-capabilities.test.ts b/packages/shared/tests/machine-protocol-capabilities.test.ts
index b6235ab4c..8684699f7 100644
--- a/packages/shared/tests/machine-protocol-capabilities.test.ts
+++ b/packages/shared/tests/machine-protocol-capabilities.test.ts
@@ -50,6 +50,9 @@ it('requires an advertised local file resource protocol, independent of release
it('requires an advertised exact queue-item steer protocol', () => {
expect(machineSupportsQueueItemSteerProtocol(undefined)).toBe(false);
+ expect(
+ machineSupportsQueueItemSteerProtocol({ protocolCapabilities: { queueItemSteer: 1 } })
+ ).toBe(false);
expect(
machineSupportsQueueItemSteerProtocol({
protocolCapabilities: { [MACHINE_PROTOCOL_CAPABILITIES.queueItemSteer]: 0 },
diff --git a/specs/message-queue-interactions.md b/specs/message-queue-interactions.md
index e42a58612..7d61c9d76 100644
--- a/specs/message-queue-interactions.md
+++ b/specs/message-queue-interactions.md
@@ -65,6 +65,12 @@ the queue by hand.
A rejected edit retains the user's draft. Neither optimistic UI success nor a client-local
editing lease proves that an edit was accepted by that authority.
+ `queueItemSteer` v2 binds this guarantee to `session/queue-mutate`: renderer edit/remove
+ carry the observed row revision, and reorder carries the observed ID sequence. The owning
+ daemon compares them under the reservation boundary and persists accepted changes before
+ replying. A failed RPC never falls back to a direct CRDT write. Enqueue and ordinary sends
+ retain their renderer authorship; this is not a generic write-intent protocol.
+
The native submission order is:
1. Validate the selected identity, content, editing ownership, and expected turn; establish
exclusive reservation ownership against concurrent ordinary queue mutations.
@@ -129,14 +135,11 @@ reported, never silently replayed.
## Implementation evidence
-The service/port split, reservation ownership transfer, pre-submission durable queue removal,
-and capability-only availability above are intended changes, not completed implementation.
-The current native path retains the editable row through provider handoff; ordinary row updates
-do not enforce operation ownership, and missing-row updates can silently succeed. This permits
-an accepted edit to disappear when handoff removes the row. The renderer also still contains
-old-daemon native/head paths and the execution service owns queue orchestration. Implementation
-and two-client mutation/crash regression verification remain
-pending under the [Effect boundary proposal](../.agents/notes/proposed/architecture/2026-09-14-queue-steer-effect-boundary.md).
+The [Effect boundary decision](../.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md)
+records the implemented service/port split, v2 mutation authority, durable pre-submission removal,
+capability-only availability, and verification limits. Deterministic tests cover reservation versus
+second-client mutations, displaced drafts, persistence failures and marker-based recovery.
+Real provider, installed-app and process-kill end-to-end verification remain outstanding.
The new availability policy applies to queued-row Steer, not composer submission routing.
- `packages/components/src/components/sessions/session-message-submit-route.ts`
@@ -144,6 +147,7 @@ The new availability policy applies to queued-row Steer, not composer submission
- `packages/components/src/components/sessions/message-queue/`
- `packages/components/tests/{session-message-submit-route,session-chat-input-submission,message-queue-row-editing}.test.*`
- `apps/cli/{tests/session-execution-service.test.ts,src/lib/loro/doc-user-turn.test.ts,src/session/session-queue-steer-operation-store.ts}`
+- `apps/cli/src/session/{queue-steer-service,active-turn-steer-port}.ts`
- [Decision record](../.agents/notes/implemented/feature/2026-09-13-queue-steer-controls.md)
This is a draft for human review. Implementation and passing tests do not approve it.
diff --git a/specs/message-queue-interactions.zh.md b/specs/message-queue-interactions.zh.md
index 904403dd2..1a2db8870 100644
--- a/specs/message-queue-interactions.zh.md
+++ b/specs/message-queue-interactions.zh.md
@@ -55,6 +55,11 @@ Translation: current
后续修改须明确拒绝。被拒绝的编辑保留用户草稿。乐观 UI 成功或客户端本地 editing lease
都不证明权威写入端已接受编辑。
+ `queueItemSteer` v2 通过 `session/queue-mutate` 建立该保证:renderer edit/remove
+ 携带观察到的 row revision,reorder 携带观察到的 ID 顺序。所属 daemon 在 reservation
+ 边界内比较,并在已接受修改持久化后回复。RPC 失败绝不退回直接写 CRDT。
+ enqueue 和普通发送仍由 renderer 写入;这不是通用 write-intent 协议。
+
Native 提交顺序为:
1. 验证目标标识、内容、编辑归属和 expected turn,并针对并发普通队列修改取得排他 reservation
ownership。
@@ -110,19 +115,18 @@ Translation: current
## 实现证据
-上述 service/port 拆分、reservation ownership 转移、提交前持久化删除队列项及仅按
-capability 开放能力均是待实现的意图变更。当前 native 路径仍保留可编辑 row 直到 provider
-handoff;普通 row 更新没有 operation ownership 检查,更新缺失 row 也可能静默成功。
-因此已接受的编辑可能在 handoff 删除 row 时丢失。renderer 也仍包含旧 daemon 的
-native/head 路径,execution service 仍拥有队列编排。实现及双客户端修改/崩溃回归验证
-待按 [Effect 边界提案](../.agents/notes/proposed/architecture/2026-09-14-queue-steer-effect-boundary.zh.md)
-完成。新的可用性策略只适用于队列行“引导”,不改变 composer 提交路由。
+[Effect 边界决策](../.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md)
+记录已实现的 service/port 拆分、v2 修改权威、提交前持久化删除、仅按 capability 开放能力及验证限制。
+确定性测试覆盖 reservation 与第二客户端修改、草稿保留、持久化失败和 marker 恢复。
+真实 provider、已安装应用及进程 kill 的端到端验证尚未完成。
+新的可用性策略只适用于队列行“引导”,不改变 composer 提交路由。
- `packages/components/src/components/sessions/session-message-submit-route.ts`
- `packages/components/src/components/sessions/session-chat-input-area.tsx`
- `packages/components/src/components/sessions/message-queue/`
- `packages/components/tests/{session-message-submit-route,session-chat-input-submission,message-queue-row-editing}.test.*`
- `apps/cli/{tests/session-execution-service.test.ts,src/lib/loro/doc-user-turn.test.ts,src/session/session-queue-steer-operation-store.ts}`
+- `apps/cli/src/session/{queue-steer-service,active-turn-steer-port}.ts`
- [决策记录](../.agents/notes/implemented/feature/2026-09-13-queue-steer-controls.zh.md)
这是供人工审阅的草稿;实现和测试通过不代表 Spec 已获批准。
From 446cd45c62172199d4a7c460a07039aeae99cf2a Mon Sep 17 00:00:00 2001
From: wibus-wee <62133302+wibus-wee@users.noreply.github.com>
Date: Mon, 14 Sep 2026 17:42:44 +0800
Subject: [PATCH 12/15] fix: secure queue session controls and pre-history
retries
Share source-side session authorization across queue controls, preserve routing planes on transport failure, and allow safe pre-history recovery retries.
Model: gpt-6
---
.agents/docs/rpc-loro-streams-rpc.md | 9 +-
.../2026-09-14-queue-steer-effect-boundary.md | 21 +-
...26-09-14-queue-steer-effect-boundary.zh.md | 20 +-
apps/cli/src/session/AGENTS.md | 6 +-
apps/cli/src/session/queue-steer-service.ts | 64 ++---
.../tests/session-execution-service.test.ts | 232 ++++++++++++++----
packages/components/src/providers/AGENTS.md | 3 +
.../src/providers/create-workspace-runtime.ts | 8 +-
.../src/providers/runtime-provider.tsx | 37 ++-
.../session-control-authorization.ts | 30 +++
.../providers/workspace-machine-rpc-facade.ts | 37 ++-
.../workspace-machine-rpc-facade.test.ts | 83 ++++++-
specs/message-queue-interactions.md | 21 +-
specs/message-queue-interactions.zh.md | 17 +-
14 files changed, 472 insertions(+), 116 deletions(-)
create mode 100644 packages/components/src/providers/session-control-authorization.ts
diff --git a/.agents/docs/rpc-loro-streams-rpc.md b/.agents/docs/rpc-loro-streams-rpc.md
index fb2625695..14564aa05 100644
--- a/.agents/docs/rpc-loro-streams-rpc.md
+++ b/.agents/docs/rpc-loro-streams-rpc.md
@@ -30,8 +30,13 @@ editing lease is inactive, then preserves native ACP Steer when acknowledged or
ordinary follow-up before cancelling the expected turn. The ordinary path retains the queue
row until history and its activation pointer are durable. Missing, editing, and stale targets
do neither.
-Before a remote renderer writes this control request, it fails closed against its authenticated
-`machines:listVisibleMachines` snapshot. The retained request contains no requester identity:
+Before either remote queue control is written, the shared
+[source authorizer](../../packages/components/src/providers/session-control-authorization.ts)
+matches session metadata to the target machine and applies the existing session visibility
+policy to authenticated machine/project snapshots and current user. Incomplete snapshots fail
+closed; shared machine access does not expose another user's private project. Local routing
+with an unavailable sender fails locally without creating a Streams client.
+The retained request contains no requester identity:
the target daemon cannot authenticate such a claim and must not grant its owner fast path from it.
Native queue Steer instead inherits the frozen requester from the authenticated active invocation;
it fails before consumption when that identity is unavailable and never trusts the queue row.
diff --git a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md
index 581a71e53..75766889e 100644
--- a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md
+++ b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md
@@ -51,9 +51,17 @@ were read as supplementary evidence, not run or assumed identical to the pinned
The queue service never reads runtime maps, agentClient, promptInFlight, invocation or successor,
and receives no onSubmitting/onAcknowledged/onApplied/onUndelivered callbacks.
+It defines its own narrow dependency contract, without importing SessionExecutionServiceDeps.
Provider requester identity comes from the active invocation, not the shared queue author.
Ordinary turn and composer routing remain unchanged.
+The source facade shares [session authorization](../../../../packages/components/src/providers/session-control-authorization.ts)
+between queue Steer and mutation, using the existing session visibility predicate and complete
+authenticated machine/project snapshots. A machine-only check would expose private-project
+sessions. RuntimeProvider supplies a workspace-fenced snapshot; missing metadata or authorization
+fails closed. Routing plane is decided independently of sender availability: local failure cannot
+fall through to Streams. The target daemon cannot authenticate a caller identity from this RPC.
+
## Resources and failures
Rewrite/ownership guards and the adapter's local ACK gate belong to Scope through
@@ -86,7 +94,11 @@ Recovery uses marker and frozen history, without requiring a surviving row. Exis
remain readable. A legacy row without a revision may be removed only when its frozen content
is provably unchanged and it is not being edited; otherwise preserve it for reconciliation.
-Reserved/fallback recover the same ordinary turn; submitting/acknowledged never replay;
+Reserved without history clears its marker and leaves the queue untouched, with no terminal
+receipt. Same C/T retries revalidate and reserve anew, either after startup recovery or within
+the current request; a failed clear prevents proceeding. A cached error here would incorrectly
+make a proven-unsubmitted operation permanently unsteerable for the remaining active turn.
+Reserved with history/fallback recover the same ordinary turn; submitting/acknowledged never replay;
applied recovers an accepted receipt. Ordinary cancel-and-dispatch retains its history/activation
publication order and in-memory receipts. The [Spec](../../../../specs/message-queue-interactions.md)
owns the full contract and remains draft. This supersedes the
@@ -106,6 +118,13 @@ reservation/history/removal/submission-marker persistence failure, marker-plus-h
and local guard release. Components verify displaced-draft retention; shared negotiation rejects v1.
No race assertion depends on sleeps or real network scheduling.
+The execution suite connects the real source facade, Streams client/server, LoroDoc and execution
+service over an in-memory transport. A visible machine with a denied private project rejects
+both controls with zero appends, no marker, and unchanged queue/history/active turn. Local routing
+with no sender also rejects without constructing the remote client. A positive project-access
+control reaches the daemon and applies C. Restart and in-request pre-history recovery both allow
+the same C/T to proceed. These are deterministic synthetic traces, not production-user traces.
+
Direct CLI/components typechecks and targeted tests were run. Root pnpm check / pnpm format
cannot start because corepack is absent; installed pnpm runs targeted checks and Prettier instead.
No real-provider, full desktop end-to-end, process kill/restart, or arbitrary mixed-old-client
diff --git a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md
index b02559755..17952eb74 100644
--- a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md
+++ b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md
@@ -49,7 +49,14 @@ CLI 规则引用的 `context/cli-effect-ts.md` 在当前 checkout 缺失。
Queue 服务不读取 runtime map、agentClient、promptInFlight、invocation 或 successor,
也不使用 onSubmitting/onAcknowledged/onApplied/onUndelivered 回调。Provider requester
-来自活动 invocation,不相信队列作者。既有普通 turn 与 composer 路由不变。
+来自活动 invocation,不相信队列作者。服务定义自己的窄依赖契约,不导入
+SessionExecutionServiceDeps。既有普通 turn 与 composer 路由不变。
+
+Source facade 为队列 Steer 和 mutation 共用 [session 授权](../../../../packages/components/src/providers/session-control-authorization.ts),
+使用已有 session visibility 判断和完整的已认证 machine/project 快照。Machine-only 检查会暴露
+私有项目 session。RuntimeProvider 提供按 workspace 隔离的快照;metadata 或授权缺失时
+fail closed。Routing plane 独立于 sender 可用性,local 失败不能落入 Streams。
+目标 daemon 无法从此 RPC 认证调用方身份。
## 资源和失败
@@ -80,7 +87,10 @@ Native 顺序是 reservation marker → pending_apply history durable → queue
恢复使用 marker 和冻结 history;不要求原 row 存在。旧 marker 仍可读;若残留 row 无 revision,
只有可证明与冻结内容一致且不在编辑中才删除,否则保留并等待对账。不能通过恢复丢掉已接受编辑。
-`reserved`/`fallback` 可恢复同一个普通 turn;`submitting`/`acknowledged` 不重放;
+没有 history 的 `reserved` 清除 marker、保留 queue,不生成终态 receipt。相同 C/T 的重试
+重新验证并建立 reservation,既支持启动恢复后重试,也支持当前请求继续;clear 失败则不能继续。
+此处缓存错误会让确定未提交的操作在当前 active turn 剩余期间永久无法再次 Steer。
+有 history 的 `reserved`/`fallback` 可恢复同一个普通 turn;`submitting`/`acknowledged` 不重放;
`applied` 恢复 accepted 回执。普通 cancel-and-dispatch 保留既有 history/activation
发布顺序与内存回执。完整契约由 [Spec](../../../../specs/message-queue-interactions.zh.md) 拥有;
它仍是 draft。本决策替代[原记录](../feature/2026-09-13-queue-steer-controls.zh.md)的
@@ -97,6 +107,12 @@ receipt/restart,以及真实 LoroDoc 上的 reservation 与 edit/remove/reorde
持久化失败时零提交、marker 加 history 的恢复和本地 guard 释放。组件覆盖 row 消失后的草稿保留,
共享协议拒绝 v1。测试不使用睡眠或真实网络来决定竞态。
+Execution suite 用内存传输连接真实 source facade、Streams client/server、LoroDoc 和
+execution service。Machine 可见但私有项目不可见时,两种 control 均拒绝:零 append、无 marker,
+queue/history/active turn 不变。Local 缺 sender 同样拒绝且不创建远程 client。开放项目权限的
+正向对照可到达 daemon 并应用 C。重启恢复和请求内 pre-history 恢复都允许相同 C/T 继续。
+这些是确定性合成数据 trace,不是生产用户 trace。
+
直接 CLI、components 类型检查与目标测试已运行;根级 pnpm check / pnpm format
因缺少 corepack 无法启动,改用已安装 pnpm 运行目标检查和 Prettier。
没有真实 provider、完整桌面端到端、进程 kill/restart 或任意旧客户端混跑的验证。
diff --git a/apps/cli/src/session/AGENTS.md b/apps/cli/src/session/AGENTS.md
index c52767ca2..c0e2908c3 100644
--- a/apps/cli/src/session/AGENTS.md
+++ b/apps/cli/src/session/AGENTS.md
@@ -56,9 +56,9 @@ Contract: specs/session-orchestration.md.
tombstone; CLI dispatch producers keep their own marker policy.
- Ordinary turn execution writes only `processingUserMsgId` and `lastHandledUserMsgId`; no start
or terminal path may read-await-rewrite the other slots.
-- Never steer after Stop; late ACK is indeterminate. Reserve against queue mutations/promotion;
- persist marker, frozen history and removal BEFORE native submission. Scope owns local guards,
- not provider effects. Only proven non-delivery may requeue; unknown delivery never replays.
+- No Steer after Stop; late ACK never replays. Exclude queue mutations/promotion; persist
+ marker/history/removal before submission. Scope owns guards only. Requeue only proven
+ non-delivery. Reserved without history clears marker, not row: retryable, no receipt.
- Resume must REOPEN the in-progress assistant entry, clearing
`finished`/`endedAt`/`permissionWaitMs` there only; never write `finished=false` from teardown.
- Keep JSON-RPC/transport matching in `acp-error-classification.ts`: disposed/stale `-32603` is
diff --git a/apps/cli/src/session/queue-steer-service.ts b/apps/cli/src/session/queue-steer-service.ts
index c22c1c7c9..6b928c8ac 100644
--- a/apps/cli/src/session/queue-steer-service.ts
+++ b/apps/cli/src/session/queue-steer-service.ts
@@ -5,33 +5,45 @@ import {
queueItemRevision,
resolveSessionHistoryStatus,
type SessionId,
+ type WorkspaceId,
+ type MachineId,
+ type ChatFailedReason,
+ type ChatFailedCode,
type SessionQueueSteerResponse,
type SessionQueueMutation,
type SessionQueueMutationResponse,
} from '@lody/shared';
import { readSessionHistory } from '@lody/shared/session-data';
-import { hasActiveMessageQueueEditingLease, type SessionDocument } from '@/lib/loro/doc';
+import {
+ hasActiveMessageQueueEditingLease,
+ type SessionDocument,
+ type LoroDocumentManager,
+} from '@/lib/loro/doc';
+import type { Logger } from '@/utils/logger';
import { formatErrorMessage } from '@/utils/format-error';
import { ActiveTurnSteerPort, PersistenceFailure } from './active-turn-steer-port';
import { buildQueuedMessageUserTurn } from './queued-message-turn';
import {
isQueueSteerMarkerOwnedBy,
type QueueSteerOperationMarker,
+ type QueueSteerOperationStore,
} from './session-queue-steer-operation-store';
-import type { SessionExecutionServiceDeps } from './session-execution-service';
type Request = { sessionId: SessionId; expectedTurnId: string; queueItemId: string };
type Marker = QueueSteerOperationMarker;
-type Recovery = SessionQueueSteerResponse | 'deferred' | null;
-type Deps = Pick<
- SessionExecutionServiceDeps,
- | 'workspaceId'
- | 'machineId'
- | 'workspaceDocument'
- | 'queueSteerOperationStore'
- | 'recordChatFailure'
- | 'logger'
-> & {
+type Recovery = SessionQueueSteerResponse | 'deferred' | 'retryable' | null;
+export type QueueSteerServiceDeps = {
+ workspaceId: WorkspaceId;
+ machineId: MachineId;
+ workspaceDocument: Pick;
+ queueSteerOperationStore: QueueSteerOperationStore;
+ logger: Logger;
+ recordChatFailure(
+ sessionDoc: SessionDocument,
+ reason: ChatFailedReason,
+ message?: string,
+ code?: ChatFailedCode
+ ): Promise;
requeue(
sessionId: SessionId,
userTurnId: string
@@ -70,11 +82,11 @@ export class QueueSteerService extends Context.Tag('lody/QueueSteerService')<
mutate(request: SessionQueueMutation): Effect.Effect;
}
>() {
- static layer(deps: Deps) {
+ static layer(deps: QueueSteerServiceDeps) {
return Layer.effect(QueueSteerService, QueueSteerService.make(deps));
}
- static make(deps: Deps) {
+ static make(deps: QueueSteerServiceDeps) {
return Effect.gen(function* () {
const active = yield* ActiveTurnSteerPort;
const receipts = new Map();
@@ -218,14 +230,8 @@ export class QueueSteerService extends Context.Tag('lody/QueueSteerService')<
const request = { sessionId, queueItemId: marker.queueItemId };
if (!entry) {
if (marker.phase !== 'reserved') return 'deferred';
- return yield* complete(
- sessionId,
- marker,
- respond(request, 'error', {
- userTurnId: marker.userTurnId,
- error: 'Reservation ended before history was persisted. The message remains queued.',
- })
- );
+ yield* persist(() => deps.queueSteerOperationStore.clear(sessionId));
+ return 'retryable';
}
if (marker.phase === 'reserved' || marker.phase === 'fallback') {
return yield* fallback(
@@ -306,13 +312,13 @@ export class QueueSteerService extends Context.Tag('lody/QueueSteerService')<
if (previous.operationKey === operationKey) {
if (previousReceipt) return remember(operationKey, previousReceipt);
const recovered = yield* recoverMarker(sessionId, doc, previous);
- return recovered && recovered !== 'deferred'
- ? recovered
- : respond(request, 'error', {
- error: 'The previous delivery is still indeterminate.',
- });
- }
- if (!previousReceipt)
+ if (recovered !== 'retryable')
+ return recovered && recovered !== 'deferred'
+ ? recovered
+ : respond(request, 'error', {
+ error: 'The previous delivery is still indeterminate.',
+ });
+ } else if (!previousReceipt)
return respond(request, 'busy', {
error: 'Another queue operation is being recovered.',
});
diff --git a/apps/cli/tests/session-execution-service.test.ts b/apps/cli/tests/session-execution-service.test.ts
index d249e5522..a8237713d 100644
--- a/apps/cli/tests/session-execution-service.test.ts
+++ b/apps/cli/tests/session-execution-service.test.ts
@@ -1,4 +1,12 @@
import { queueItemRevision } from '@lody/shared';
+import { CURRENT_MACHINE_PROTOCOL_CAPABILITIES } from '@lody/shared';
+import { createWorkspaceMachineRpcFacade } from '../../../packages/components/src/providers/workspace-machine-rpc-facade';
+import {
+ LoroStreamsMachineRpcClient,
+ LoroStreamsMachineRpcServer,
+ type LoroStreamsJsonStreamClient,
+ type LoroJsonLiveBatchHandler,
+} from '@lody/loro-streams-rpc';
import { SessionDocument } from '../src/lib/loro/doc';
import { composeTestSessionDoc } from './session-doc-fixture';
import { withHistoryPort } from './history-port-fixture';
@@ -331,6 +339,189 @@ describe('SessionExecutionService', () => {
};
};
+ it.each(['startup', 'same-request', 'clear-failure'] as const)(
+ 'retries C/T after pre-history recovery via %s',
+ async (mode) => {
+ const h = await ownedQueue();
+ await h.deps.queueSteerOperationStore.record({
+ version: 2,
+ workspaceId: 'workspace-1',
+ machineId: 'machine-1',
+ ...h.request,
+ operationKey: JSON.stringify([
+ h.request.sessionId,
+ h.request.expectedTurnId,
+ h.request.queueItemId,
+ ]),
+ userTurnId: 'user:C',
+ phase: 'reserved',
+ updatedAt: 1,
+ });
+ const restarted = new SessionExecutionService(h.deps);
+ if (mode === 'startup') {
+ await restarted.recoverPendingQueueSteers();
+ expect(await h.deps.queueSteerOperationStore.read(h.request.sessionId)).toBeNull();
+ expect(await h.doc.getMessageQueue()).toEqual(h.rows);
+ expect(await h.doc.sessionData.history.readAll()).toEqual([]);
+ }
+ (
+ restarted as unknown as { turnRuntimeBySession: Map }
+ ).turnRuntimeBySession.set(h.request.sessionId, h.runtime);
+ h.steerPrompt.mockImplementation(() => ({
+ applied: Promise.resolve({ release: () => {} }),
+ completion: Promise.resolve(),
+ }));
+ if (mode === 'clear-failure') {
+ vi.spyOn(h.deps.queueSteerOperationStore, 'clear').mockRejectedValueOnce(
+ new Error('marker clear failed')
+ );
+ expect(await restarted.steerQueuedMessage(h.request)).toMatchObject({
+ accepted: false,
+ error: 'marker clear failed',
+ });
+ expect(await h.deps.queueSteerOperationStore.read(h.request.sessionId)).toMatchObject({
+ phase: 'reserved',
+ });
+ expect(await h.doc.getMessageQueue()).toEqual(h.rows);
+ expect(await h.doc.sessionData.history.readAll()).toEqual([]);
+ expect(h.runtime.userTurnId).toBe('user:active');
+ }
+ expect(await restarted.steerQueuedMessage(h.request)).toMatchObject({ accepted: true });
+ expect((await h.doc.getMessageQueue()).map((row) => row.task)).toEqual(['A', 'B']);
+ expect(h.runtime.userTurnId).toBe('user:C');
+ expect(await h.deps.queueSteerOperationStore.read(h.request.sessionId)).toMatchObject({
+ phase: 'applied',
+ response: { disposition: 'accepted' },
+ });
+ }
+ );
+
+ it.each(['private-project', 'local-sender-missing'] as const)(
+ 'traces %s rejection before Streams and daemon state changes',
+ async (scenario) => {
+ const h = await ownedQueue();
+ const machineId = 'machine-1' as MachineId;
+ const meta = await h.doc.getMetaState();
+ vi.spyOn(h.doc, 'getMetaState').mockResolvedValue({
+ ...meta,
+ project: { kind: 'local', localProjectId: 'private-P' },
+ } as SessionMeta);
+ const appended: unknown[] = [];
+ const readers = new Map();
+ const streamClient: LoroStreamsJsonStreamClient = {
+ ensureJsonStream: async () => {},
+ appendJson: async (streamId, value) => {
+ appended.push(value);
+ const reader = readers.get(streamId);
+ if (!reader) throw new Error('Missing test stream reader');
+ await reader({ messages: [value], nextOffset: String(appended.length), upToDate: true });
+ return String(appended.length);
+ },
+ readJsonLive: async (streamId, _state, onBatch, options) => {
+ readers.set(streamId, onBatch);
+ await new Promise((resolve) => {
+ if (options?.signal?.aborted) resolve();
+ else options?.signal?.addEventListener('abort', () => resolve(), { once: true });
+ });
+ readers.delete(streamId);
+ },
+ };
+ const server = new LoroStreamsMachineRpcServer({
+ workspaceId: h.deps.workspaceId,
+ machineId,
+ logger: createSilentLogger(),
+ streamClient,
+ getMachineStatus: vi.fn(),
+ refreshMachineAcpCapabilities: vi.fn(),
+ steerQueuedMessage: (args) => h.service.steerQueuedMessage(args),
+ mutateQueuedMessage: (args) => h.service.mutateQueuedMessage(args),
+ });
+ const client = new LoroStreamsMachineRpcClient({
+ workspaceId: h.deps.workspaceId,
+ machineId,
+ streamClient,
+ });
+ let projectVisible = false;
+ let plane: 'local' | 'cloud' = scenario === 'private-project' ? 'cloud' : 'local';
+ const getMachineRpcClient = vi.fn(async () => client);
+ const facade = createWorkspaceMachineRpcFacade({
+ workspaceId: h.deps.workspaceId,
+ targetRouter: {
+ getPlaneForMachine: () => plane,
+ resolvePlaneForMachine: async () => plane,
+ },
+ getMachineProtocolCapabilities: async () => CURRENT_MACHINE_PROTOCOL_CAPABILITIES,
+ getSessionMeta: async () => h.doc.getMetaState(),
+ getSessionControlAuthorization: () => ({
+ visibleMachineIds: new Set([machineId]),
+ visibleLocalProjectKeys: new Set(projectVisible ? [machineId + ':private-P'] : []),
+ currentUserId: 'user-U',
+ }),
+ getMachineRpcClient,
+ });
+ await server.start();
+ try {
+ const error =
+ scenario === 'private-project'
+ ? 'Source authorization for this session is unavailable or denied.'
+ : 'Local queue control is unavailable.';
+ expect(await facade.requestSessionQueueSteer(machineId, h.request)).toMatchObject({
+ accepted: false,
+ error,
+ });
+ expect(
+ await facade.requestSessionQueueMutation(machineId, {
+ sessionId: h.request.sessionId,
+ mutation: {
+ kind: 'remove',
+ queueItemId: h.request.queueItemId,
+ expectedRevision: queueItemRevision(h.rows[2]),
+ },
+ })
+ ).toMatchObject({ success: false, error });
+ expect(getMachineRpcClient).not.toHaveBeenCalled();
+ expect(appended).toEqual([]);
+ expect(await h.deps.queueSteerOperationStore.read(h.request.sessionId)).toBeNull();
+ expect(await h.doc.getMessageQueue()).toEqual(h.rows);
+ expect(await h.doc.sessionData.history.readAll()).toEqual([]);
+ expect(h.runtime.userTurnId).toBe('user:active');
+ expect(h.runtime.promptInFlight).toBe(true);
+ // Positive control traverses the same real RPC client/server and execution service.
+ projectVisible = true;
+ plane = 'cloud';
+ h.steerPrompt.mockImplementation(() => ({
+ applied: Promise.resolve({ release: () => {} }),
+ completion: Promise.resolve(),
+ }));
+ expect(await facade.requestSessionQueueSteer(machineId, h.request)).toMatchObject({
+ accepted: true,
+ });
+ expect(appended.length).toBeGreaterThan(0);
+ expect((await h.doc.getMessageQueue()).map((row) => row.task)).toEqual(['A', 'B']);
+ expect(h.runtime.userTurnId).toBe('user:C');
+ expect(await h.deps.queueSteerOperationStore.read(h.request.sessionId)).toMatchObject({
+ phase: 'applied',
+ });
+ const secondRow = h.rows[1];
+ if (!secondRow) throw new Error('Missing B fixture');
+ expect(
+ await facade.requestSessionQueueMutation(machineId, {
+ sessionId: h.request.sessionId,
+ mutation: {
+ kind: 'remove',
+ queueItemId: secondRow.$cid,
+ expectedRevision: queueItemRevision(secondRow),
+ },
+ })
+ ).toMatchObject({ success: true });
+ expect((await h.doc.getMessageQueue()).map((row) => row.task)).toEqual(['A']);
+ } finally {
+ client.stop();
+ server.stop();
+ }
+ }
+ );
+
it.each(['update', 'remove', 'reorder'] as const)(
'rejects a stale second-client %s after reservation, with removal durable before submission',
async (kind) => {
@@ -1346,47 +1537,6 @@ describe('SessionExecutionService', () => {
).resolves.toMatchObject({ machineId: 'machine-2', phase: 'reserved' });
});
- it('leaves the row queued when a crash precedes the reserved history write', async () => {
- const sessionId = 'session-native-steer-pre-history-crash' as SessionId;
- const queueSteerOperationStore = createMemoryQueueSteerOperationStore();
- await queueSteerOperationStore.record({
- version: 1,
- workspaceId: 'workspace-1',
- machineId: 'machine-1',
- sessionId,
- operationKey: 'pre-history-operation',
- queueItemId: 'C',
- expectedTurnId: 'assistant:old',
- userTurnId: 'user:C',
- phase: 'reserved',
- updatedAt: 1,
- });
- const removeMessageQueueItem = vi.fn(async () => {});
- const sessionDoc = withHistoryPort({
- waitUntilSynced: vi.fn(async () => {}),
- getHistory: () => [],
- getMessageQueue: vi.fn(async () => [{ $cid: 'C', task: 'task C' }]),
- removeMessageQueueItem,
- });
- const deps = createBaseDeps({
- queueSteerOperationStore,
- workspaceDocument: {
- repo: { upsertDocMeta: vi.fn(async () => {}) },
- getOrCreateSessionDoc: vi.fn(async () => sessionDoc),
- } as unknown as LoroDocumentManager,
- });
- const service = new SessionExecutionService(deps);
-
- await service.recoverPendingQueueSteers();
-
- expect(removeMessageQueueItem).not.toHaveBeenCalled();
- await expect(queueSteerOperationStore.read(sessionId)).resolves.toMatchObject({
- phase: 'reserved',
- completedAt: expect.any(Number),
- response: { disposition: 'error' },
- });
- });
-
it('advances one session owner through consecutive prompt handoffs', async () => {
const steerPrompt = vi.fn(() => ({
completion: new Promise(() => {}),
diff --git a/packages/components/src/providers/AGENTS.md b/packages/components/src/providers/AGENTS.md
index de3d4c568..e05454a43 100644
--- a/packages/components/src/providers/AGENTS.md
+++ b/packages/components/src/providers/AGENTS.md
@@ -56,6 +56,9 @@ and update only decision fields through HistoryWriter; never replace a rendered
- Queue edit/remove/reorder on queueItemSteer v2 use the narrow daemon domain RPC,
with revision checks against reservation ownership. Never direct-write after RPC failure.
Enqueue and other user writes retain their renderer authority.
+- Remote queue Steer and mutation share source-side session visibility authorization:
+ verify target metadata against authenticated machine/project visibility and current user.
+ Missing metadata or incomplete snapshots fail closed; machine access alone is insufficient.
- Repo storage, durable Streams cursors, and eager-sync high-water state must use the
same per-renderer cache namespace. A checkpoint must never be shared by independently
persisted Repo views.
diff --git a/packages/components/src/providers/create-workspace-runtime.ts b/packages/components/src/providers/create-workspace-runtime.ts
index cba1ad310..4b270e7ce 100644
--- a/packages/components/src/providers/create-workspace-runtime.ts
+++ b/packages/components/src/providers/create-workspace-runtime.ts
@@ -127,6 +127,7 @@ import { META_REMOTE_CURSOR_BYPASS_STORAGE_KEY_PREFIX } from '@/lib/clear-local-
import { runStartupAcpCapabilitiesRefresh } from './startup-acp-capabilities-refresh';
import { createLocalLoroDataPlaneConnection } from './local-loro-data-plane-connection';
import { createWorkspaceMachineRpcFacade } from './workspace-machine-rpc-facade';
+import type { SessionControlAuthorization } from './session-control-authorization';
import { resyncMachineFlockRows } from '@/hooks/use-machine-flock-rows';
import { createCodeCollabFileIndexCache } from '@/lib/code-collab-file-index-cache';
import { getIpcServices, onIpcEvent, sendLocalSessionControl } from '@/lib/electron-ipc-client';
@@ -203,6 +204,7 @@ type RuntimeDeps = {
* authorization is not ready, so optional startup capability refresh is skipped.
*/
getAuthorizedMachineIds?: () => ReadonlySet | null;
+ getSessionControlAuthorization?: () => SessionControlAuthorization | null;
eagerSyncSurface?: EagerSyncSurface;
};
@@ -1722,7 +1724,11 @@ export async function createWorkspaceRuntime(deps: RuntimeDeps): Promise | undefined)?.protocolCapabilities;
},
- getAuthorizedMachineIds: deps.getAuthorizedMachineIds,
+ getSessionControlAuthorization: deps.getSessionControlAuthorization,
+ getSessionMeta: async (sessionId) => {
+ const entry = await repo.getDocMeta(getSessionRoomId(sessionId));
+ return entry?.meta as SessionMeta | undefined;
+ },
workspaceId,
targetRouter,
getMachineRpcClient,
diff --git a/packages/components/src/providers/runtime-provider.tsx b/packages/components/src/providers/runtime-provider.tsx
index 8f7f92760..687603f0f 100644
--- a/packages/components/src/providers/runtime-provider.tsx
+++ b/packages/components/src/providers/runtime-provider.tsx
@@ -2,7 +2,7 @@ import { useEffect, useRef, type ReactNode } from 'react';
import { useAtomValue, useSetAtom } from 'jotai';
import { LODY_PRESENCE_HEARTBEAT_MS, type MachineId, type WorkspaceId } from '@lody/shared';
import { authTokenAtom, runtimeAtom } from '@/atoms/runtime';
-import { currentWorkspaceIdAtom, currentWorkspaceSlugAtom } from '@/atoms';
+import { currentWorkspaceIdAtom, currentWorkspaceSlugAtom, userAtom } from '@/atoms';
import { clearDocMetaCacheAtom, docMetaSubscriptionAtom } from '@/atoms/doc-meta';
import {
clearLodyPresenceStatesAtom,
@@ -35,6 +35,9 @@ import { isElectronRenderer } from '@/lib/electron';
import { isNativeAppShell } from '@/lib/native-platform';
import { usePlatform } from '@lody/platform/react';
import { useVisibleMachineMetas } from '@/hooks/use-visible-machine-metas';
+import { useVisibleLocalProjectsFromMachineIndex } from '@/hooks/use-visible-local-projects';
+import { useAuthenticatedConvex } from '@/hooks/use-authenticated-convex';
+import type { SessionControlAuthorization } from './session-control-authorization';
const isExpectedRuntimeShutdownError = (error: unknown): boolean => {
if (!(error instanceof Error)) {
@@ -87,6 +90,29 @@ export function RuntimeProvider({ children }: { children: ReactNode }) {
includeMachineFlock: false,
syncMachineFlock: false,
});
+ const visibleProjectIndex = useVisibleLocalProjectsFromMachineIndex(visibleMachineIndex);
+ const currentUserId = useAtomValue(userAtom)?.id;
+ const authentication = useAuthenticatedConvex();
+ const sessionAuthorizationRef = useRef<{
+ workspaceId: WorkspaceId;
+ authorization: SessionControlAuthorization;
+ } | null>(null);
+ sessionAuthorizationRef.current =
+ !workspaceId ||
+ !currentUserId ||
+ !authentication.isAuthenticated ||
+ authentication.isLoading ||
+ visibleMachineIndex.isLoading ||
+ visibleProjectIndex.isLoading
+ ? null
+ : {
+ workspaceId,
+ authorization: {
+ visibleMachineIds: new Set(visibleMachineIndex.convexAuthorizedMachineIds),
+ visibleLocalProjectKeys: new Set(visibleProjectIndex.accessByProjectKey.keys()),
+ currentUserId,
+ },
+ };
const authorizedMachineIdsRef = useRef<{
machineIds: ReadonlySet;
workspaceId: WorkspaceId;
@@ -264,9 +290,16 @@ export function RuntimeProvider({ children }: { children: ReactNode }) {
const snapshot = authorizedMachineIdsRef.current;
return snapshot?.workspaceId === effectiveWorkspaceId ? snapshot.machineIds : null;
},
+ getSessionControlAuthorization: () => {
+ const snapshot = sessionAuthorizationRef.current;
+ return snapshot?.workspaceId === effectiveWorkspaceId ? snapshot.authorization : null;
+ },
...(telemetryEnabled
? {
- onAnalyticsEvent: (event: { name: string; properties?: Record }) => {
+ onAnalyticsEvent: (event: {
+ name: string;
+ properties?: Record;
+ }) => {
capturePostHogEvent(postHogRef.current, event.name, event.properties);
},
}
diff --git a/packages/components/src/providers/session-control-authorization.ts b/packages/components/src/providers/session-control-authorization.ts
new file mode 100644
index 000000000..5538ff145
--- /dev/null
+++ b/packages/components/src/providers/session-control-authorization.ts
@@ -0,0 +1,30 @@
+import type { MachineId, SessionId } from '@lody/shared';
+import { isSessionVisibleToUser, type SessionMachineRecord } from '../lib/session-visibility';
+
+export type SessionControlAuthorization = {
+ visibleMachineIds: ReadonlySet;
+ visibleLocalProjectKeys: ReadonlySet;
+ currentUserId?: string;
+};
+
+export type SessionControlAuthorizationDeps = {
+ getSessionMeta?: (sessionId: SessionId) => Promise;
+ /** Null until the authenticated machine and project snapshots are both ready. */
+ getSessionControlAuthorization?: () => SessionControlAuthorization | null;
+};
+
+export async function authorizeSessionControl(
+ deps: SessionControlAuthorizationDeps,
+ sessionId: SessionId,
+ machineId: MachineId
+): Promise {
+ const meta = await deps.getSessionMeta?.(sessionId);
+ const authorization = deps.getSessionControlAuthorization?.();
+ if (!meta || meta.machineId !== machineId || !authorization) return false;
+ return isSessionVisibleToUser(
+ meta,
+ authorization.visibleMachineIds,
+ authorization.visibleLocalProjectKeys,
+ authorization.currentUserId
+ );
+}
diff --git a/packages/components/src/providers/workspace-machine-rpc-facade.ts b/packages/components/src/providers/workspace-machine-rpc-facade.ts
index 3d2fda48b..f4e74a59c 100644
--- a/packages/components/src/providers/workspace-machine-rpc-facade.ts
+++ b/packages/components/src/providers/workspace-machine-rpc-facade.ts
@@ -67,9 +67,13 @@ import {
sessionForkFailure,
sessionEditAndResendFailure,
} from '@lody/shared';
-import { createAsyncConcurrencyGate } from '@/lib/async-concurrency-gate';
-import { getIpcServices } from '@/lib/electron-ipc-client';
+import { createAsyncConcurrencyGate } from '../lib/async-concurrency-gate';
+import { getIpcServices } from '../lib/electron-ipc-client';
import type { WorkspaceTargetRouter } from './workspace-target-router';
+import {
+ authorizeSessionControl,
+ type SessionControlAuthorizationDeps,
+} from './session-control-authorization';
const LOCAL_MACHINE_ID_READY_TIMEOUT_MS = 2_000;
const CODE_COLLAB_DIFF_RPC_CONCURRENCY_LIMIT = 4;
@@ -90,14 +94,12 @@ type LspRequest = {
readonly character?: number;
};
-export type WorkspaceMachineRpcFacadeDeps = {
+export type WorkspaceMachineRpcFacadeDeps = SessionControlAuthorizationDeps & {
workspaceId: WorkspaceId;
targetRouter: Pick;
getMachineProtocolCapabilities: (
machineId: MachineId
) => Promise;
- /** Authenticated source snapshot; null means cloud authorization is not ready. */
- getAuthorizedMachineIds?: () => ReadonlySet | null;
getMachineRpcClient: (machineId: MachineId) => Promise;
};
@@ -137,6 +139,18 @@ export function createWorkspaceMachineRpcFacade(deps: WorkspaceMachineRpcFacadeD
);
};
+ const resolveSessionControlPlane = async (machineId: MachineId) => {
+ const plane =
+ targetRouter.getPlaneForMachine(machineId) ??
+ (await targetRouter.resolvePlaneForMachine(machineId, {
+ timeoutMs: LOCAL_MACHINE_ID_READY_TIMEOUT_MS,
+ }));
+ if (plane !== 'local' && plane !== 'cloud') {
+ throw new Error('Session control routing is unavailable.');
+ }
+ return plane;
+ };
+
const sendLocalMachineRpcRequest = async (
request: LocalMachineRpcRequest
): Promise => {
@@ -780,7 +794,7 @@ export function createWorkspaceMachineRpcFacade(deps: WorkspaceMachineRpcFacadeD
if (!machineSupportsQueueItemSteerProtocol({ protocolCapabilities })) {
throw new Error('This daemon does not support queue ownership controls.');
}
- if (await canUseLocalMachineRpc(machineId)) {
+ if ((await resolveSessionControlPlane(machineId)) === 'local') {
const response = await getLocalMachineRpcSender()?.({
machineId,
workspaceId,
@@ -792,8 +806,8 @@ export function createWorkspaceMachineRpcFacade(deps: WorkspaceMachineRpcFacadeD
if (!response.ok) throw new Error(response.error);
return response.result as SessionQueueMutationResponse;
}
- if (!deps.getAuthorizedMachineIds?.()?.has(machineId)) {
- throw new Error('Source authorization for this machine is unavailable or denied.');
+ if (!(await authorizeSessionControl(deps, args.sessionId, machineId))) {
+ throw new Error('Source authorization for this session is unavailable or denied.');
}
return await (await getMachineRpcClient(machineId)).requestSessionQueueMutation(args);
} catch (error) {
@@ -826,7 +840,7 @@ export function createWorkspaceMachineRpcFacade(deps: WorkspaceMachineRpcFacadeD
error: 'This machine does not support exact queued-message steering.',
};
}
- if (await canUseLocalMachineRpc(machineId)) {
+ if ((await resolveSessionControlPlane(machineId)) === 'local') {
const response = await getLocalMachineRpcSender()?.({
machineId,
workspaceId,
@@ -847,15 +861,14 @@ export function createWorkspaceMachineRpcFacade(deps: WorkspaceMachineRpcFacadeD
if (response?.ok) return response.result as SessionQueueSteerResponse;
throw new Error('Local queue control is unavailable.');
}
- const authorizedMachineIds = deps.getAuthorizedMachineIds?.() ?? null;
- if (!authorizedMachineIds?.has(machineId)) {
+ if (!(await authorizeSessionControl(deps, args.sessionId, machineId))) {
return {
type: 'session/queue-steer_response',
sessionId: args.sessionId,
queueItemId: args.queueItemId,
accepted: false,
disposition: 'error',
- error: 'Source authorization for this machine is unavailable or denied.',
+ error: 'Source authorization for this session is unavailable or denied.',
};
}
return await (
diff --git a/packages/components/tests/workspace-machine-rpc-facade.test.ts b/packages/components/tests/workspace-machine-rpc-facade.test.ts
index 7af58e6ca..6be044a47 100644
--- a/packages/components/tests/workspace-machine-rpc-facade.test.ts
+++ b/packages/components/tests/workspace-machine-rpc-facade.test.ts
@@ -17,6 +17,61 @@ afterEach(() => {
});
describe('createWorkspaceMachineRpcFacade', () => {
+ it.each([
+ 'snapshot-missing',
+ 'meta-missing',
+ 'wrong-machine',
+ 'revoked-during-meta-read',
+ 'route-missing',
+ ] as const)('fails both session controls closed for %s', async (scenario) => {
+ let authorization =
+ scenario === 'snapshot-missing'
+ ? null
+ : {
+ visibleMachineIds: new Set([remoteMachineId]),
+ visibleLocalProjectKeys: new Set(),
+ currentUserId: 'user-U',
+ };
+ const getMachineRpcClient = vi.fn(async () => {
+ throw new Error('Remote client must not be created');
+ });
+ const facade = createWorkspaceMachineRpcFacade({
+ workspaceId,
+ getMachineProtocolCapabilities: async () => CURRENT_MACHINE_PROTOCOL_CAPABILITIES,
+ getSessionControlAuthorization: () => authorization,
+ getSessionMeta: async () => {
+ if (scenario === 'revoked-during-meta-read') authorization = null;
+ return scenario === 'meta-missing'
+ ? undefined
+ : {
+ machineId: scenario === 'wrong-machine' ? localMachineId : remoteMachineId,
+ };
+ },
+ targetRouter: {
+ getPlaneForMachine: () => (scenario === 'route-missing' ? null : 'cloud'),
+ resolvePlaneForMachine: async () => null,
+ },
+ getMachineRpcClient,
+ });
+ const error =
+ scenario === 'route-missing'
+ ? 'Session control routing is unavailable.'
+ : 'Source authorization for this session is unavailable or denied.';
+ expect(
+ await facade.requestSessionQueueSteer(remoteMachineId, {
+ sessionId,
+ queueItemId: 'C',
+ expectedTurnId: 'T',
+ })
+ ).toMatchObject({ accepted: false, error });
+ expect(
+ await facade.requestSessionQueueMutation(remoteMachineId, {
+ sessionId,
+ mutation: { kind: 'remove', queueItemId: 'C', expectedRevision: '{}' },
+ })
+ ).toMatchObject({ success: false, error });
+ expect(getMachineRpcClient).not.toHaveBeenCalled();
+ });
it.each([undefined, { queueItemSteer: 1 }])(
'disables every old-daemon queue row for %j',
async (capabilities) => {
@@ -24,8 +79,8 @@ describe('createWorkspaceMachineRpcFacade', () => {
workspaceId,
getMachineProtocolCapabilities: async () => capabilities,
targetRouter: {
- getPlaneForMachine: () => 'remote',
- resolvePlaneForMachine: async () => 'remote',
+ getPlaneForMachine: () => 'cloud',
+ resolvePlaneForMachine: async () => 'cloud',
},
getMachineRpcClient: async () => {
throw new Error('Legacy delivery must not run');
@@ -133,8 +188,8 @@ describe('createWorkspaceMachineRpcFacade', () => {
workspaceId,
getMachineProtocolCapabilities: async () => undefined,
targetRouter: {
- getPlaneForMachine: () => 'remote',
- resolvePlaneForMachine: async () => 'remote',
+ getPlaneForMachine: () => 'cloud',
+ resolvePlaneForMachine: async () => 'cloud',
},
getMachineRpcClient,
});
@@ -155,10 +210,10 @@ describe('createWorkspaceMachineRpcFacade', () => {
const facade = createWorkspaceMachineRpcFacade({
workspaceId,
getMachineProtocolCapabilities: async () => CURRENT_MACHINE_PROTOCOL_CAPABILITIES,
- getAuthorizedMachineIds: () => null,
+ getSessionControlAuthorization: () => null,
targetRouter: {
- getPlaneForMachine: () => 'remote',
- resolvePlaneForMachine: async () => 'remote',
+ getPlaneForMachine: () => 'cloud',
+ resolvePlaneForMachine: async () => 'cloud',
},
getMachineRpcClient,
});
@@ -185,10 +240,14 @@ describe('createWorkspaceMachineRpcFacade', () => {
const facade = createWorkspaceMachineRpcFacade({
workspaceId,
getMachineProtocolCapabilities: async () => CURRENT_MACHINE_PROTOCOL_CAPABILITIES,
- getAuthorizedMachineIds: () => new Set([remoteMachineId]),
+ getSessionMeta: async () => ({ machineId: remoteMachineId }),
+ getSessionControlAuthorization: () => ({
+ visibleMachineIds: new Set([remoteMachineId]),
+ visibleLocalProjectKeys: new Set(),
+ }),
targetRouter: {
- getPlaneForMachine: () => 'remote',
- resolvePlaneForMachine: async () => 'remote',
+ getPlaneForMachine: () => 'cloud',
+ resolvePlaneForMachine: async () => 'cloud',
},
getMachineRpcClient: async () => ({ requestSessionQueueSteer }) as never,
});
@@ -213,8 +272,8 @@ describe('createWorkspaceMachineRpcFacade', () => {
workspaceId,
getMachineProtocolCapabilities: async () => undefined,
targetRouter: {
- getPlaneForMachine: () => 'remote',
- resolvePlaneForMachine: async () => 'remote',
+ getPlaneForMachine: () => 'cloud',
+ resolvePlaneForMachine: async () => 'cloud',
},
getMachineRpcClient: async () => {
throw new Error('Unexpected RPC');
diff --git a/specs/message-queue-interactions.md b/specs/message-queue-interactions.md
index 7d61c9d76..b08a7ab2b 100644
--- a/specs/message-queue-interactions.md
+++ b/specs/message-queue-interactions.md
@@ -46,11 +46,18 @@ the queue by hand.
preconditions. Indeterminate delivery must propagate without replay; local failures recover
from durable evidence and never imply non-delivery by themselves.
- A remote renderer must authorize the target before writing the request, using its
- authenticated authoritative machine-access snapshot and failing closed if that snapshot is
- unavailable. The request carries no requester identity: workspace Machine RPC cannot
+ Both `session/queue-steer` and `session/queue-mutate` must pass one source-side session
+ authorization before a remote request is written. The session metadata must match the target
+ machine; access follows `isSessionVisibleToUser` using authenticated machine visibility,
+ local-project visibility and the current user. Visible machine access alone does not grant
+ access to another user's private local-project session. Missing metadata or an incomplete
+ authorization snapshot fails closed, with no machine-only fallback.
+ The request carries no requester identity: workspace Machine RPC cannot
authenticate a caller-supplied member ID, and the target daemon must not use one for an owner
fast path. Same-host local IPC is already the trusted local control boundary.
+ Routing is resolved before transport availability: local with no sender returns a local
+ transport error, and unresolved routing returns an error. Neither can create a Streams
+ client or append a remote request; transport failure cannot change the selected plane.
Native exact Steer must derive its requester identity from the authenticated active
invocation, never from the shared queue row. If that frozen identity is unavailable, the
@@ -83,9 +90,11 @@ the queue by hand.
Any failure to persist the reservation, history, or removal forbids provider submission.
Removing the row before the network call is necessary but does not replace the ownership
boundary during steps 1–4. Native recovery uses the machine-local marker plus frozen history,
- not an editable queue row as a retry token. A pre-history reservation interrupted by a crash
- must be reconciled as unsubmitted before its surviving row becomes editable again; it must not
- report a stale edit as saved or construct a turn from later queue content. Once history is
+ not an editable queue row as a retry token. A `reserved` marker with no history is unsubmitted:
+ recovery clears the marker, leaves the surviving row untouched and creates no terminal receipt.
+ The same queue item/expected turn may retry immediately, including within the current request,
+ after rereading and validating the queue and live turn as a new reservation. Failed marker
+ cleanup blocks that retry. Once history is
durable, recovery can finish removal and recover the same turn without requiring the row to
exist. A new missing-row request still fails; replay of the same reserved operation resolves
through its marker or receipt.
diff --git a/specs/message-queue-interactions.zh.md b/specs/message-queue-interactions.zh.md
index 1a2db8870..556316294 100644
--- a/specs/message-queue-interactions.zh.md
+++ b/specs/message-queue-interactions.zh.md
@@ -40,10 +40,16 @@ Translation: current
交付不确定须向上传递,禁止 replay;本地失败依据 durable evidence 恢复,不能仅凭本地
失败推断未交付。
- 远程 Renderer 在写入请求前,必须使用其已认证的权威 machine-access 快照验证目标;快照
- 不可用时应 fail closed。请求不得携带请求者身份:workspace Machine RPC 无法认证调用方
+ `session/queue-steer` 与 `session/queue-mutate` 在写入远程请求前,必须共用 source-side
+ session 授权。Session metadata 必须匹配目标 machine;权限通过 `isSessionVisibleToUser`
+ 根据已认证的 machine visibility、local-project visibility 和 current user 判断。
+ Machine 可见不意味着其他用户的私有 local-project session 可见。Metadata 缺失或授权快照
+ 不完整时 fail closed,绝不降级成 machine-only 检查。
+ 请求不得携带请求者身份:workspace Machine RPC 无法认证调用方
声称的成员 ID,目标 daemon 也不得用它命中 owner fast path。同机 local IPC 已是可信的
本地控制边界。
+ 先确定 routing plane,再检查 transport:local 缺少 sender 返回本地传输错误,路由未确定
+ 则返回路由错误。两者都不得创建 Streams client 或追加远程请求;传输失败不能改变已选 plane。
Native 精确“引导”的 requester identity 必须来自已认证的活动 invocation,绝不来自共享
queue row。若该冻结身份不可用,daemon 必须在消费 row、提交 provider 或停止 turn 之前失败。
@@ -71,9 +77,10 @@ Translation: current
reservation、history 或删除任一持久化失败,都禁止提交 provider。网络调用前删除 row
是必要条件,但不能替代步骤 1–4 期间的 ownership 边界。Native 恢复依赖 machine-local
- marker 与冻结 history,不再拿可编辑 queue row 当 retry token。若在 history 持久化前
- 崩溃,须先将 reservation 对账为未提交,剩余 row 才能恢复可编辑;不得把旧客户端编辑
- 报成保存成功,也不得从后来的 queue 内容构建 turn。history 持久化后,恢复可完成删除并
+ marker 与冻结 history,不再拿可编辑 queue row 当 retry token。`reserved` marker 没有
+ history 时属于未提交:恢复清除 marker,保留剩余 row,不生成终态 receipt。同一个 queue
+ item/expected turn 可立即重试,包括继续当前请求,但必须重新读取并验证 queue 和 live turn,
+ 建立新的 reservation。Marker 清理失败时不得继续重试。history 持久化后,恢复可完成删除并
恢复同一个 turn,无需 row 仍存在。新的缺失 row 请求仍失败;同一 reserved operation
的重试通过 marker 或 receipt 解析。
From 204e0d3d18cf3248f2dd06a4c1ebbc6ee4a18ce2 Mon Sep 17 00:00:00 2001
From: wibus-wee <62133302+wibus-wee@users.noreply.github.com>
Date: Mon, 14 Sep 2026 17:54:26 +0800
Subject: [PATCH 13/15] fix: satisfy queue steer static checks
Make local/remote mutation outcomes and successful Effect guard completion explicit, and align the machine registration assertion with queueItemSteer v2.\n\nModel: gpt-5
---
apps/cli/src/session/queue-steer-service.ts | 1 +
apps/cli/src/session/session-execution-service.ts | 1 +
apps/cli/tests/message-handler-machine-registration.test.ts | 2 +-
packages/components/src/providers/workspace-writer-impl.ts | 3 +++
4 files changed, 6 insertions(+), 1 deletion(-)
diff --git a/apps/cli/src/session/queue-steer-service.ts b/apps/cli/src/session/queue-steer-service.ts
index 6b928c8ac..ea03e624a 100644
--- a/apps/cli/src/session/queue-steer-service.ts
+++ b/apps/cli/src/session/queue-steer-service.ts
@@ -178,6 +178,7 @@ export class QueueSteerService extends Context.Tag('lody/QueueSteerService')<
yield* persist(() => doc.removeMessageQueueItem(marker.queueItemId));
}
yield* flush();
+ return undefined;
});
const fallback = Effect.fn('QueueSteer.fallback')(function* (
diff --git a/apps/cli/src/session/session-execution-service.ts b/apps/cli/src/session/session-execution-service.ts
index acac76ab7..22a4891bb 100644
--- a/apps/cli/src/session/session-execution-service.ts
+++ b/apps/cli/src/session/session-execution-service.ts
@@ -1624,6 +1624,7 @@ export class SessionExecutionService {
message: 'The active turn is stopping.',
});
}
+ return undefined;
});
const sessionDoc = yield* prepare(() =>
self.deps.workspaceDocument.getOrCreateSessionDoc(options.sessionId)
diff --git a/apps/cli/tests/message-handler-machine-registration.test.ts b/apps/cli/tests/message-handler-machine-registration.test.ts
index 90c428cfe..39fb95301 100644
--- a/apps/cli/tests/message-handler-machine-registration.test.ts
+++ b/apps/cli/tests/message-handler-machine-registration.test.ts
@@ -187,7 +187,7 @@ describe('MessageHandler machine registration', () => {
localFileResources: 1,
providerSetup: 1,
acpProtocolAuthentication: 2,
- queueItemSteer: 1,
+ queueItemSteer: 2,
subagentCancellation: 1,
});
diff --git a/packages/components/src/providers/workspace-writer-impl.ts b/packages/components/src/providers/workspace-writer-impl.ts
index 5f50f2980..fd708735e 100644
--- a/packages/components/src/providers/workspace-writer-impl.ts
+++ b/packages/components/src/providers/workspace-writer-impl.ts
@@ -199,6 +199,7 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo
const mq = (draft.mq ?? []) as MessageQueueItem[];
draft.mq = mq.filter((item) => item.$cid !== itemId);
});
+ return false;
});
if (!remote) await bumpMessageQueueWatermark(sessionId);
},
@@ -230,6 +231,7 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo
: item
);
});
+ return false;
});
if (!remote) await bumpMessageQueueWatermark(sessionId);
},
@@ -264,6 +266,7 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo
}
draft.mq = ordered;
});
+ return false;
});
if (!remote) await bumpMessageQueueWatermark(sessionId);
},
From 64fc2f1d1d14cb2f2c9dac0de84df41806b711b5 Mon Sep 17 00:00:00 2001
From: wibus-wee <62133302+wibus-wee@users.noreply.github.com>
Date: Mon, 14 Sep 2026 18:15:33 +0800
Subject: [PATCH 14/15] fix: separate session control authorization from UI
visibility
Require current machine and local-project access for queue controls, remove the session-owner authorization fallback, and cover revoked-owner requests with end-to-end in-memory traces.
Model: gpt-6
---
.agents/docs/rpc-loro-streams-rpc.md | 7 +-
.../2026-09-14-queue-steer-effect-boundary.md | 13 +-
...26-09-14-queue-steer-effect-boundary.zh.md | 9 +-
.../tests/session-execution-service.test.ts | 255 +++++++++---------
packages/components/src/providers/AGENTS.md | 6 +-
.../src/providers/runtime-provider.tsx | 1 -
.../session-control-authorization.ts | 28 +-
.../workspace-machine-rpc-facade.test.ts | 1 -
specs/message-queue-interactions.md | 7 +-
specs/message-queue-interactions.zh.md | 7 +-
10 files changed, 181 insertions(+), 153 deletions(-)
diff --git a/.agents/docs/rpc-loro-streams-rpc.md b/.agents/docs/rpc-loro-streams-rpc.md
index 14564aa05..9eeca24bb 100644
--- a/.agents/docs/rpc-loro-streams-rpc.md
+++ b/.agents/docs/rpc-loro-streams-rpc.md
@@ -32,9 +32,10 @@ row until history and its activation pointer are durable. Missing, editing, and
do neither.
Before either remote queue control is written, the shared
[source authorizer](../../packages/components/src/providers/session-control-authorization.ts)
-matches session metadata to the target machine and applies the existing session visibility
-policy to authenticated machine/project snapshots and current user. Incomplete snapshots fail
-closed; shared machine access does not expose another user's private project. Local routing
+matches session metadata to the target machine and requires current machine access plus
+matching project access for local-project sessions. Unlike UI visibility, control has no
+session-owner fallback: authorship cannot override revoked access. Incomplete snapshots fail
+closed. Local routing
with an unavailable sender fails locally without creating a Streams client.
The retained request contains no requester identity:
the target daemon cannot authenticate such a claim and must not grant its owner fast path from it.
diff --git a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md
index 75766889e..ba98b9707 100644
--- a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md
+++ b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md
@@ -56,9 +56,12 @@ Provider requester identity comes from the active invocation, not the shared que
Ordinary turn and composer routing remain unchanged.
The source facade shares [session authorization](../../../../packages/components/src/providers/session-control-authorization.ts)
-between queue Steer and mutation, using the existing session visibility predicate and complete
-authenticated machine/project snapshots. A machine-only check would expose private-project
-sessions. RuntimeProvider supplies a workspace-fenced snapshot; missing metadata or authorization
+between queue Steer and mutation, using complete authenticated machine/project snapshots.
+Control requires machine access plus matching project access for local-project sessions.
+Reusing UI visibility was incorrect: its intentional session-owner fallback permits display
+after machine access is revoked, but must not grant control. The control contract therefore
+contains no currentUserId or session-owner input and does not call the visibility predicate.
+RuntimeProvider supplies a workspace-fenced snapshot; missing metadata or authorization
fails closed. Routing plane is decided independently of sender availability: local failure cannot
fall through to Streams. The target daemon cannot authenticate a caller identity from this RPC.
@@ -120,7 +123,9 @@ No race assertion depends on sleeps or real network scheduling.
The execution suite connects the real source facade, Streams client/server, LoroDoc and execution
service over an in-memory transport. A visible machine with a denied private project rejects
-both controls with zero appends, no marker, and unchanged queue/history/active turn. Local routing
+both controls with zero appends, no marker, and unchanged queue/history/active turn. The same
+trace covers session creators with revoked machine or denied project access, even while UI
+visibility remains true; a retained project grant cannot override machine revocation. Local routing
with no sender also rejects without constructing the remote client. A positive project-access
control reaches the daemon and applies C. Restart and in-request pre-history recovery both allow
the same C/T to proceed. These are deterministic synthetic traces, not production-user traces.
diff --git a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md
index 17952eb74..954a97fe1 100644
--- a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md
+++ b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md
@@ -53,8 +53,10 @@ Queue 服务不读取 runtime map、agentClient、promptInFlight、invocation
SessionExecutionServiceDeps。既有普通 turn 与 composer 路由不变。
Source facade 为队列 Steer 和 mutation 共用 [session 授权](../../../../packages/components/src/providers/session-control-authorization.ts),
-使用已有 session visibility 判断和完整的已认证 machine/project 快照。Machine-only 检查会暴露
-私有项目 session。RuntimeProvider 提供按 workspace 隔离的快照;metadata 或授权缺失时
+使用完整的已认证 machine/project 快照。控制需要 machine 权限,local-project session 还需
+对应项目权限。复用 UI visibility 是错误的:它刻意允许 session 创建者在 machine 权限撤销后
+仍看到 session,但不能据此授予控制权。因此控制契约不包含 currentUserId 或 session owner,
+也不调用 visibility predicate。RuntimeProvider 提供按 workspace 隔离的快照;metadata 或授权缺失时
fail closed。Routing plane 独立于 sender 可用性,local 失败不能落入 Streams。
目标 daemon 无法从此 RPC 认证调用方身份。
@@ -109,7 +111,8 @@ receipt/restart,以及真实 LoroDoc 上的 reservation 与 edit/remove/reorde
Execution suite 用内存传输连接真实 source facade、Streams client/server、LoroDoc 和
execution service。Machine 可见但私有项目不可见时,两种 control 均拒绝:零 append、无 marker,
-queue/history/active turn 不变。Local 缺 sender 同样拒绝且不创建远程 client。开放项目权限的
+queue/history/active turn 不变。同一 trace 覆盖 session 创建者被撤销 machine 或拒绝项目权限,
+即使 UI visibility 仍为 true;残留项目权限不能绕过 machine 撤权。Local 缺 sender 同样拒绝且不创建远程 client。开放项目权限的
正向对照可到达 daemon 并应用 C。重启恢复和请求内 pre-history 恢复都允许相同 C/T 继续。
这些是确定性合成数据 trace,不是生产用户 trace。
diff --git a/apps/cli/tests/session-execution-service.test.ts b/apps/cli/tests/session-execution-service.test.ts
index a8237713d..ca87c85d9 100644
--- a/apps/cli/tests/session-execution-service.test.ts
+++ b/apps/cli/tests/session-execution-service.test.ts
@@ -1,6 +1,7 @@
import { queueItemRevision } from '@lody/shared';
import { CURRENT_MACHINE_PROTOCOL_CAPABILITIES } from '@lody/shared';
import { createWorkspaceMachineRpcFacade } from '../../../packages/components/src/providers/workspace-machine-rpc-facade';
+import { isSessionVisibleToUser } from '../../../packages/components/src/lib/session-visibility';
import {
LoroStreamsMachineRpcClient,
LoroStreamsMachineRpcServer,
@@ -396,131 +397,143 @@ describe('SessionExecutionService', () => {
}
);
- it.each(['private-project', 'local-sender-missing'] as const)(
- 'traces %s rejection before Streams and daemon state changes',
- async (scenario) => {
- const h = await ownedQueue();
- const machineId = 'machine-1' as MachineId;
- const meta = await h.doc.getMetaState();
- vi.spyOn(h.doc, 'getMetaState').mockResolvedValue({
- ...meta,
- project: { kind: 'local', localProjectId: 'private-P' },
- } as SessionMeta);
- const appended: unknown[] = [];
- const readers = new Map();
- const streamClient: LoroStreamsJsonStreamClient = {
- ensureJsonStream: async () => {},
- appendJson: async (streamId, value) => {
- appended.push(value);
- const reader = readers.get(streamId);
- if (!reader) throw new Error('Missing test stream reader');
- await reader({ messages: [value], nextOffset: String(appended.length), upToDate: true });
- return String(appended.length);
- },
- readJsonLive: async (streamId, _state, onBatch, options) => {
- readers.set(streamId, onBatch);
- await new Promise((resolve) => {
- if (options?.signal?.aborted) resolve();
- else options?.signal?.addEventListener('abort', () => resolve(), { once: true });
- });
- readers.delete(streamId);
- },
- };
- const server = new LoroStreamsMachineRpcServer({
- workspaceId: h.deps.workspaceId,
- machineId,
- logger: createSilentLogger(),
- streamClient,
- getMachineStatus: vi.fn(),
- refreshMachineAcpCapabilities: vi.fn(),
- steerQueuedMessage: (args) => h.service.steerQueuedMessage(args),
- mutateQueuedMessage: (args) => h.service.mutateQueuedMessage(args),
+ it.each([
+ 'private-project',
+ 'local-sender-missing',
+ 'revoked-owner-machine',
+ 'revoked-owner-project',
+ 'owner-private-project',
+ ] as const)('traces %s rejection before Streams and daemon state changes', async (scenario) => {
+ const h = await ownedQueue();
+ const machineId = 'machine-1' as MachineId;
+ const meta = await h.doc.getMetaState();
+ const sessionMeta = {
+ ...meta,
+ userId: scenario.includes('owner') ? 'user-U' : meta?.userId,
+ project:
+ scenario === 'revoked-owner-machine'
+ ? undefined
+ : { kind: 'local', localProjectId: 'private-P' },
+ } as SessionMeta;
+ vi.spyOn(h.doc, 'getMetaState').mockResolvedValue(sessionMeta);
+ const appended: unknown[] = [];
+ const readers = new Map();
+ const streamClient: LoroStreamsJsonStreamClient = {
+ ensureJsonStream: async () => {},
+ appendJson: async (streamId, value) => {
+ appended.push(value);
+ const reader = readers.get(streamId);
+ if (!reader) throw new Error('Missing test stream reader');
+ await reader({ messages: [value], nextOffset: String(appended.length), upToDate: true });
+ return String(appended.length);
+ },
+ readJsonLive: async (streamId, _state, onBatch, options) => {
+ readers.set(streamId, onBatch);
+ await new Promise((resolve) => {
+ if (options?.signal?.aborted) resolve();
+ else options?.signal?.addEventListener('abort', () => resolve(), { once: true });
+ });
+ readers.delete(streamId);
+ },
+ };
+ const server = new LoroStreamsMachineRpcServer({
+ workspaceId: h.deps.workspaceId,
+ machineId,
+ logger: createSilentLogger(),
+ streamClient,
+ getMachineStatus: vi.fn(),
+ refreshMachineAcpCapabilities: vi.fn(),
+ steerQueuedMessage: (args) => h.service.steerQueuedMessage(args),
+ mutateQueuedMessage: (args) => h.service.mutateQueuedMessage(args),
+ });
+ const client = new LoroStreamsMachineRpcClient({
+ workspaceId: h.deps.workspaceId,
+ machineId,
+ streamClient,
+ });
+ let projectVisible = scenario.startsWith('revoked');
+ let machineVisible = !scenario.startsWith('revoked');
+ let plane: 'local' | 'cloud' = scenario === 'local-sender-missing' ? 'local' : 'cloud';
+ const getMachineRpcClient = vi.fn(async () => client);
+ const facade = createWorkspaceMachineRpcFacade({
+ workspaceId: h.deps.workspaceId,
+ targetRouter: {
+ getPlaneForMachine: () => plane,
+ resolvePlaneForMachine: async () => plane,
+ },
+ getMachineProtocolCapabilities: async () => CURRENT_MACHINE_PROTOCOL_CAPABILITIES,
+ getSessionMeta: async () => h.doc.getMetaState(),
+ getSessionControlAuthorization: () => ({
+ visibleMachineIds: new Set(machineVisible ? [machineId] : []),
+ visibleLocalProjectKeys: new Set(projectVisible ? [machineId + ':private-P'] : []),
+ }),
+ getMachineRpcClient,
+ });
+ await server.start();
+ try {
+ const error =
+ scenario === 'local-sender-missing'
+ ? 'Local queue control is unavailable.'
+ : 'Source authorization for this session is unavailable or denied.';
+ if (scenario.includes('owner')) {
+ expect(isSessionVisibleToUser(sessionMeta, new Set(), new Set(), 'user-U')).toBe(true);
+ }
+ expect(await facade.requestSessionQueueSteer(machineId, h.request)).toMatchObject({
+ accepted: false,
+ error,
});
- const client = new LoroStreamsMachineRpcClient({
- workspaceId: h.deps.workspaceId,
- machineId,
- streamClient,
+ expect(
+ await facade.requestSessionQueueMutation(machineId, {
+ sessionId: h.request.sessionId,
+ mutation: {
+ kind: 'remove',
+ queueItemId: h.request.queueItemId,
+ expectedRevision: queueItemRevision(h.rows[2]),
+ },
+ })
+ ).toMatchObject({ success: false, error });
+ expect(getMachineRpcClient).not.toHaveBeenCalled();
+ expect(appended).toEqual([]);
+ expect(await h.deps.queueSteerOperationStore.read(h.request.sessionId)).toBeNull();
+ expect(await h.doc.getMessageQueue()).toEqual(h.rows);
+ expect(await h.doc.sessionData.history.readAll()).toEqual([]);
+ expect(h.runtime.userTurnId).toBe('user:active');
+ expect(h.runtime.promptInFlight).toBe(true);
+ // Positive control traverses the same real RPC client/server and execution service.
+ projectVisible = true;
+ machineVisible = true;
+ plane = 'cloud';
+ h.steerPrompt.mockImplementation(() => ({
+ applied: Promise.resolve({ release: () => {} }),
+ completion: Promise.resolve(),
+ }));
+ expect(await facade.requestSessionQueueSteer(machineId, h.request)).toMatchObject({
+ accepted: true,
});
- let projectVisible = false;
- let plane: 'local' | 'cloud' = scenario === 'private-project' ? 'cloud' : 'local';
- const getMachineRpcClient = vi.fn(async () => client);
- const facade = createWorkspaceMachineRpcFacade({
- workspaceId: h.deps.workspaceId,
- targetRouter: {
- getPlaneForMachine: () => plane,
- resolvePlaneForMachine: async () => plane,
- },
- getMachineProtocolCapabilities: async () => CURRENT_MACHINE_PROTOCOL_CAPABILITIES,
- getSessionMeta: async () => h.doc.getMetaState(),
- getSessionControlAuthorization: () => ({
- visibleMachineIds: new Set([machineId]),
- visibleLocalProjectKeys: new Set(projectVisible ? [machineId + ':private-P'] : []),
- currentUserId: 'user-U',
- }),
- getMachineRpcClient,
+ expect(appended.length).toBeGreaterThan(0);
+ expect((await h.doc.getMessageQueue()).map((row) => row.task)).toEqual(['A', 'B']);
+ expect(h.runtime.userTurnId).toBe('user:C');
+ expect(await h.deps.queueSteerOperationStore.read(h.request.sessionId)).toMatchObject({
+ phase: 'applied',
});
- await server.start();
- try {
- const error =
- scenario === 'private-project'
- ? 'Source authorization for this session is unavailable or denied.'
- : 'Local queue control is unavailable.';
- expect(await facade.requestSessionQueueSteer(machineId, h.request)).toMatchObject({
- accepted: false,
- error,
- });
- expect(
- await facade.requestSessionQueueMutation(machineId, {
- sessionId: h.request.sessionId,
- mutation: {
- kind: 'remove',
- queueItemId: h.request.queueItemId,
- expectedRevision: queueItemRevision(h.rows[2]),
- },
- })
- ).toMatchObject({ success: false, error });
- expect(getMachineRpcClient).not.toHaveBeenCalled();
- expect(appended).toEqual([]);
- expect(await h.deps.queueSteerOperationStore.read(h.request.sessionId)).toBeNull();
- expect(await h.doc.getMessageQueue()).toEqual(h.rows);
- expect(await h.doc.sessionData.history.readAll()).toEqual([]);
- expect(h.runtime.userTurnId).toBe('user:active');
- expect(h.runtime.promptInFlight).toBe(true);
- // Positive control traverses the same real RPC client/server and execution service.
- projectVisible = true;
- plane = 'cloud';
- h.steerPrompt.mockImplementation(() => ({
- applied: Promise.resolve({ release: () => {} }),
- completion: Promise.resolve(),
- }));
- expect(await facade.requestSessionQueueSteer(machineId, h.request)).toMatchObject({
- accepted: true,
- });
- expect(appended.length).toBeGreaterThan(0);
- expect((await h.doc.getMessageQueue()).map((row) => row.task)).toEqual(['A', 'B']);
- expect(h.runtime.userTurnId).toBe('user:C');
- expect(await h.deps.queueSteerOperationStore.read(h.request.sessionId)).toMatchObject({
- phase: 'applied',
- });
- const secondRow = h.rows[1];
- if (!secondRow) throw new Error('Missing B fixture');
- expect(
- await facade.requestSessionQueueMutation(machineId, {
- sessionId: h.request.sessionId,
- mutation: {
- kind: 'remove',
- queueItemId: secondRow.$cid,
- expectedRevision: queueItemRevision(secondRow),
- },
- })
- ).toMatchObject({ success: true });
- expect((await h.doc.getMessageQueue()).map((row) => row.task)).toEqual(['A']);
- } finally {
- client.stop();
- server.stop();
- }
+ const secondRow = h.rows[1];
+ if (!secondRow) throw new Error('Missing B fixture');
+ expect(
+ await facade.requestSessionQueueMutation(machineId, {
+ sessionId: h.request.sessionId,
+ mutation: {
+ kind: 'remove',
+ queueItemId: secondRow.$cid,
+ expectedRevision: queueItemRevision(secondRow),
+ },
+ })
+ ).toMatchObject({ success: true });
+ expect((await h.doc.getMessageQueue()).map((row) => row.task)).toEqual(['A']);
+ } finally {
+ client.stop();
+ server.stop();
}
- );
+ });
it.each(['update', 'remove', 'reorder'] as const)(
'rejects a stale second-client %s after reservation, with removal durable before submission',
diff --git a/packages/components/src/providers/AGENTS.md b/packages/components/src/providers/AGENTS.md
index e05454a43..236c9890c 100644
--- a/packages/components/src/providers/AGENTS.md
+++ b/packages/components/src/providers/AGENTS.md
@@ -56,9 +56,9 @@ and update only decision fields through HistoryWriter; never replace a rendered
- Queue edit/remove/reorder on queueItemSteer v2 use the narrow daemon domain RPC,
with revision checks against reservation ownership. Never direct-write after RPC failure.
Enqueue and other user writes retain their renderer authority.
-- Remote queue Steer and mutation share source-side session visibility authorization:
- verify target metadata against authenticated machine/project visibility and current user.
- Missing metadata or incomplete snapshots fail closed; machine access alone is insufficient.
+- Remote queue Steer and mutation share source-side control authorization, not UI visibility:
+ require target machine access and, for local-project sessions, matching project access.
+ Session authorship never grants control. Missing metadata or incomplete snapshots fail closed.
- Repo storage, durable Streams cursors, and eager-sync high-water state must use the
same per-renderer cache namespace. A checkpoint must never be shared by independently
persisted Repo views.
diff --git a/packages/components/src/providers/runtime-provider.tsx b/packages/components/src/providers/runtime-provider.tsx
index 687603f0f..554ea2912 100644
--- a/packages/components/src/providers/runtime-provider.tsx
+++ b/packages/components/src/providers/runtime-provider.tsx
@@ -110,7 +110,6 @@ export function RuntimeProvider({ children }: { children: ReactNode }) {
authorization: {
visibleMachineIds: new Set(visibleMachineIndex.convexAuthorizedMachineIds),
visibleLocalProjectKeys: new Set(visibleProjectIndex.accessByProjectKey.keys()),
- currentUserId,
},
};
const authorizedMachineIdsRef = useRef<{
diff --git a/packages/components/src/providers/session-control-authorization.ts b/packages/components/src/providers/session-control-authorization.ts
index 5538ff145..d2dfea88c 100644
--- a/packages/components/src/providers/session-control-authorization.ts
+++ b/packages/components/src/providers/session-control-authorization.ts
@@ -1,14 +1,15 @@
-import type { MachineId, SessionId } from '@lody/shared';
-import { isSessionVisibleToUser, type SessionMachineRecord } from '../lib/session-visibility';
+import type { MachineId, SessionId, SessionMeta } from '@lody/shared';
+import { getLocalProjectVisibilityKey } from '../lib/visible-local-project-index';
export type SessionControlAuthorization = {
visibleMachineIds: ReadonlySet;
visibleLocalProjectKeys: ReadonlySet;
- currentUserId?: string;
};
export type SessionControlAuthorizationDeps = {
- getSessionMeta?: (sessionId: SessionId) => Promise;
+ getSessionMeta?: (
+ sessionId: SessionId
+ ) => Promise | undefined>;
/** Null until the authenticated machine and project snapshots are both ready. */
getSessionControlAuthorization?: () => SessionControlAuthorization | null;
};
@@ -20,11 +21,16 @@ export async function authorizeSessionControl(
): Promise {
const meta = await deps.getSessionMeta?.(sessionId);
const authorization = deps.getSessionControlAuthorization?.();
- if (!meta || meta.machineId !== machineId || !authorization) return false;
- return isSessionVisibleToUser(
- meta,
- authorization.visibleMachineIds,
- authorization.visibleLocalProjectKeys,
- authorization.currentUserId
- );
+ if (!meta || meta.machineId !== machineId || !authorization?.visibleMachineIds.has(machineId))
+ return false;
+ // Display ownership is not a control grant; revoked machine access always denies control.
+ if (meta.project?.kind === 'local') {
+ const projectId = meta.project.localProjectId;
+ return (
+ typeof projectId === 'string' &&
+ projectId.length > 0 &&
+ authorization.visibleLocalProjectKeys.has(getLocalProjectVisibilityKey(machineId, projectId))
+ );
+ }
+ return true;
}
diff --git a/packages/components/tests/workspace-machine-rpc-facade.test.ts b/packages/components/tests/workspace-machine-rpc-facade.test.ts
index 6be044a47..0aca3d6e5 100644
--- a/packages/components/tests/workspace-machine-rpc-facade.test.ts
+++ b/packages/components/tests/workspace-machine-rpc-facade.test.ts
@@ -30,7 +30,6 @@ describe('createWorkspaceMachineRpcFacade', () => {
: {
visibleMachineIds: new Set([remoteMachineId]),
visibleLocalProjectKeys: new Set(),
- currentUserId: 'user-U',
};
const getMachineRpcClient = vi.fn(async () => {
throw new Error('Remote client must not be created');
diff --git a/specs/message-queue-interactions.md b/specs/message-queue-interactions.md
index b08a7ab2b..cde2ef07e 100644
--- a/specs/message-queue-interactions.md
+++ b/specs/message-queue-interactions.md
@@ -48,9 +48,10 @@ the queue by hand.
Both `session/queue-steer` and `session/queue-mutate` must pass one source-side session
authorization before a remote request is written. The session metadata must match the target
- machine; access follows `isSessionVisibleToUser` using authenticated machine visibility,
- local-project visibility and the current user. Visible machine access alone does not grant
- access to another user's private local-project session. Missing metadata or an incomplete
+ machine; control requires current authenticated machine access and, for local-project sessions,
+ matching project access. Control policy is separate from UI visibility: session authorship
+ never grants control or overrides revoked access, even if the UI still displays the session.
+ Missing metadata or an incomplete
authorization snapshot fails closed, with no machine-only fallback.
The request carries no requester identity: workspace Machine RPC cannot
authenticate a caller-supplied member ID, and the target daemon must not use one for an owner
diff --git a/specs/message-queue-interactions.zh.md b/specs/message-queue-interactions.zh.md
index 556316294..787546696 100644
--- a/specs/message-queue-interactions.zh.md
+++ b/specs/message-queue-interactions.zh.md
@@ -41,9 +41,10 @@ Translation: current
失败推断未交付。
`session/queue-steer` 与 `session/queue-mutate` 在写入远程请求前,必须共用 source-side
- session 授权。Session metadata 必须匹配目标 machine;权限通过 `isSessionVisibleToUser`
- 根据已认证的 machine visibility、local-project visibility 和 current user 判断。
- Machine 可见不意味着其他用户的私有 local-project session 可见。Metadata 缺失或授权快照
+ session 控制授权。Session metadata 必须匹配目标 machine;控制必须具备当前已认证的
+ machine 访问权限,local-project session 还必须具备对应项目权限。控制策略独立于 UI
+ visibility:创建 session 不授予控制权,也不能绕过权限撤销,即使 UI 仍显示该 session。
+ Metadata 缺失或授权快照
不完整时 fail closed,绝不降级成 machine-only 检查。
请求不得携带请求者身份:workspace Machine RPC 无法认证调用方
声称的成员 ID,目标 daemon 也不得用它命中 owner fast path。同机 local IPC 已是可信的
From 48e21a9b21cb4faa0474939fcb7b393eb8d5ac2f Mon Sep 17 00:00:00 2001
From: wibus-wee <62133302+wibus-wee@users.noreply.github.com>
Date: Mon, 14 Sep 2026 18:51:35 +0800
Subject: [PATCH 15/15] fix: separate native steer preparation from submission
Complete preparation before journaling submission, recover preparation failures as ordinary dispatch, and prevent replay after indeterminate provider failures.
Model: gpt-6
---
.../2026-09-14-queue-steer-effect-boundary.md | 29 +-
...26-09-14-queue-steer-effect-boundary.zh.md | 26 +-
apps/cli/src/session/AGENTS.md | 4 +-
.../cli/src/session/active-turn-steer-port.ts | 20 +-
apps/cli/src/session/queue-steer-service.ts | 88 +++--
.../src/session/session-execution-service.ts | 338 +++++++++++-------
.../tests/session-execution-service.test.ts | 92 ++++-
specs/message-queue-interactions.md | 13 +-
specs/message-queue-interactions.zh.md | 12 +-
9 files changed, 418 insertions(+), 204 deletions(-)
diff --git a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md
index ba98b9707..7e022ed0c 100644
--- a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md
+++ b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.md
@@ -72,14 +72,22 @@ Effect.acquireRelease. Interruption is masked between submission and ACK cleanup
so local cleanup ownership cannot be abandoned. Stop still cancels ACP, not its owner fiber.
Scope neither reverses external submission nor runs after process death.
-| Category | Behavior |
-| ------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| ProviderRejected, pre-submission StaleTurn | Only proven non-delivery and eligible preconditions allow ordinary dispatch recovery. Missing, editing or stale selection before reservation remains a no-op. |
-| ProviderDeliveryUnknown | Return the error without fallback or replay, including ownership loss or handoff failure after submission. |
-| PersistenceFailure | Local preparation/persistence failure; retain durable evidence, never infer non-delivery from a local failure. |
-
-A crash after write-ahead submission evidence but before local preparation finishes remains
-conservatively indeterminate. Do not introduce a phase per Effect step or retry the whole operation.
+| Category | Behavior |
+| ------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| SteerPreparationFailure, ProviderRejected, pre-submission StaleTurn | Only proven non-delivery and eligible preconditions allow ordinary dispatch recovery. Missing, editing or stale selection before reservation remains a no-op. |
+| ProviderDeliveryUnknown | Return the error without fallback or replay, including ownership loss or handoff failure after submission. |
+| PersistenceFailure | Journal/history persistence failure; retain durable evidence, never infer non-delivery from the error tag alone. |
+
+`prepareSteer` completes prompt building, configuration application and ownership validation
+before the queue service writes `submitting`. Its failures are handled only at the preparation
+call, not through a blanket PersistenceFailure fallback. `submitSteer` consumes an opaque,
+single-use handle; the execution service privately retains the lazy submission effect in a
+WeakMap, exposing neither runtime objects nor callbacks. The port rechecks ownership after
+journal persistence without asynchronous preparation before the provider call. Generic throws
+from that call are ProviderDeliveryUnknown, whether synchronous or asynchronous; explicit
+AgentSteerNotDeliveredError remains proven rejection. An invalid/reused handle cannot replay.
+Crashes between write-ahead evidence and the provider call remain indeterminate. No new phase
+or whole-operation retry is introduced.
## Authoritative mutation and recovery
@@ -92,7 +100,7 @@ Enqueue, ordinary sends and other UI data remain renderer-authored, without a ge
write-intent mirror. Old daemons do not expose Queue Steer.
Native order is reservation marker → pending_apply history durable → queue removal durable
-→ submitting marker → native port → durable receipt. Every pre-submit barrier gates provider calls.
+→ prepareSteer → submitting marker → submitSteer → durable receipt. Every pre-submit barrier gates provider calls.
Recovery uses marker and frozen history, without requiring a surviving row. Existing markers
remain readable. A legacy row without a revision may be removed only when its frozen content
is provably unchanged and it is not being edited; otherwise preserve it for reconciliation.
@@ -120,6 +128,9 @@ Explicit promise gates prove durable removal before submission, zero submissions
reservation/history/removal/submission-marker persistence failure, marker-plus-history recovery,
and local guard release. Components verify displaced-draft retention; shared negotiation rejects v1.
No race assertion depends on sleeps or real network scheduling.
+Prompt-build and mode/model failures leave C dispatchable with a fallback marker and no provider
+call. Synchronous provider throws and rejected ACKs retain submission evidence without fallback
+or replay. Stop during submission-marker persistence is checked before any provider call.
The execution suite connects the real source facade, Streams client/server, LoroDoc and execution
service over an in-memory transport. A visible machine with a denied private project rejects
diff --git a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md
index 954a97fe1..36e505dea 100644
--- a/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md
+++ b/.agents/notes/implemented/architecture/2026-09-14-queue-steer-effect-boundary.zh.md
@@ -66,14 +66,19 @@ rewrite/ownership guard 和 adapter 本地 ACK gate 用 Effect.acquireRelease
提交到 ACK cleanup 注册之间屏蔽 fiber interruption,避免释放本地 handle 的责任丢失;
Stop 仍取消 ACP,而非其 owner fiber。Scope 既不能撤销外部提交,也不能跨进程死亡执行。
-| 类别 | 行为 |
-| ---------------------------------- | ----------------------------------------------------------------------------------------- |
-| ProviderRejected、提交前 StaleTurn | 只有确定未交付且前提允许才恢复普通 dispatch;reservation 前缺失、编辑中或过期仍无副作用。 |
-| ProviderDeliveryUnknown | 向上返回错误,不进入 fallback、不重放。包括提交后失去 ownership 或 handoff 失败。 |
-| PersistenceFailure | 本地准备/持久化失败;保留 durable evidence,不能从本地失败推导未交付。 |
-
-Native submission 写前证据之后、本地准备尚未完成的 crash 仍保守地是不确定交付。
-不为每个 Effect 步骤再加 phase,也不重试整个 operation。
+| 类别 | 行为 |
+| ----------------------------------------------------------- | ----------------------------------------------------------------------------------------- |
+| SteerPreparationFailure、ProviderRejected、提交前 StaleTurn | 只有确定未交付且前提允许才恢复普通 dispatch;reservation 前缺失、编辑中或过期仍无副作用。 |
+| ProviderDeliveryUnknown | 向上返回错误,不进入 fallback、不重放。包括提交后失去 ownership 或 handoff 失败。 |
+| PersistenceFailure | Journal/history 持久化失败;保留 durable evidence,不能仅凭错误 tag 推导未交付。 |
+
+`prepareSteer` 完成 prompt 构建、配置应用和 ownership 验证后,queue service 才写
+`submitting`。只在 preparation 调用处处理其失败,不 blanket catch PersistenceFailure
+来 fallback。`submitSteer` 消费不透明的单次 handle;execution service 用私有 WeakMap
+保留惰性 submission effect,不暴露 runtime 对象或 callback。Journal 持久化后,port 再次
+检查 ownership,provider call 前不再等待异步准备。该调用的普通同步/异步异常均为
+ProviderDeliveryUnknown;明确的 AgentSteerNotDeliveredError 仍证明拒绝交付。无效或已用
+handle 不能重放。写前证据与 provider call 之间的 crash 仍不确定;不新增 phase 或重试整个 operation。
## 权威写入与恢复
@@ -85,7 +90,7 @@ RPC 失败绝不退回直接写入。enqueue、普通发送和其他 UI 数据
没有恢复通用 write-intent mirror。旧 daemon 不开放队列 Steer。
Native 顺序是 reservation marker → pending_apply history durable → queue removal durable
-→ submitting marker → native port → durable receipt。任一提交前 barrier 失败禁止 provider 调用。
+→ prepareSteer → submitting marker → submitSteer → durable receipt。任一提交前 barrier 失败禁止 provider 调用。
恢复使用 marker 和冻结 history;不要求原 row 存在。旧 marker 仍可读;若残留 row 无 revision,
只有可证明与冻结内容一致且不在编辑中才删除,否则保留并等待对账。不能通过恢复丢掉已接受编辑。
@@ -108,6 +113,9 @@ receipt/restart,以及真实 LoroDoc 上的 reservation 与 edit/remove/reorde
显式 promise gate 验证删除持久化先于 provider,以及 reservation/history/removal/submission-marker
持久化失败时零提交、marker 加 history 的恢复和本地 guard 释放。组件覆盖 row 消失后的草稿保留,
共享协议拒绝 v1。测试不使用睡眠或真实网络来决定竞态。
+Prompt 构建或 mode/model 失败时,C 可普通 dispatch、marker 为 fallback,provider 零调用。
+Provider 同步抛错或 ACK 拒绝保留 submission 证据,不 fallback、不重放。
+Submission marker 持久化期间的 Stop 也会在 provider call 前被检查。
Execution suite 用内存传输连接真实 source facade、Streams client/server、LoroDoc 和
execution service。Machine 可见但私有项目不可见时,两种 control 均拒绝:零 append、无 marker,
diff --git a/apps/cli/src/session/AGENTS.md b/apps/cli/src/session/AGENTS.md
index c0e2908c3..4f47b6f54 100644
--- a/apps/cli/src/session/AGENTS.md
+++ b/apps/cli/src/session/AGENTS.md
@@ -25,8 +25,8 @@ Contract: specs/session-orchestration.md.
## Dispatch
-- QueueSteerService owns selection/recovery; ActiveTurnSteerPort owns execution.
- No runtime handles or phase callbacks. Missing/stale targets never stop.
+- QueueSteerService selects/recovers; ActiveTurnSteerPort prepares/submits.
+ Prepare before submitting evidence. No runtime handles/phase callbacks; stale/missing never stop.
- Absent session meta is "unknown", not foreign: hold the TTL-bounded RPC stash until meta lands;
drop it only on a definitive verdict.
- Subscribe to RPC offers BEFORE awaiting Doc Room join/sync and never dispatch from the RPC
diff --git a/apps/cli/src/session/active-turn-steer-port.ts b/apps/cli/src/session/active-turn-steer-port.ts
index 09f7d470f..a27189d58 100644
--- a/apps/cli/src/session/active-turn-steer-port.ts
+++ b/apps/cli/src/session/active-turn-steer-port.ts
@@ -29,11 +29,16 @@ export class PersistenceFailure extends Data.TaggedError('PersistenceFailure')<{
cause: unknown;
}> {}
-export type NativeSteerFailure =
- | StaleTurn
- | ProviderRejected
- | ProviderDeliveryUnknown
- | PersistenceFailure;
+/** Preparation failed before steer submission; the frozen turn may become ordinary dispatch. */
+export class SteerPreparationFailure extends Data.TaggedError('SteerPreparationFailure')<{
+ message: string;
+ cause: unknown;
+}> {}
+
+/** Opaque, single-use handle; execution state stays with the live-turn owner. */
+export class PreparedSteer extends Data.TaggedClass('PreparedSteer')<{}> {}
+
+export type NativeSteerFailure = StaleTurn | ProviderRejected | ProviderDeliveryUnknown;
export type NativeSteerInput = {
sessionId: SessionId;
expectedTurnId: string;
@@ -49,7 +54,10 @@ export class ActiveTurnSteerPort extends Context.Tag('lody/ActiveTurnSteerPort')
sessionId: SessionId,
expectedTurnId: string
): Effect.Effect<{ native: boolean; requesterUserId: string }, StaleTurn | ProviderRejected>;
- steer(input: NativeSteerInput): Effect.Effect<{ userTurnId: string }, NativeSteerFailure>;
+ prepareSteer(
+ input: NativeSteerInput
+ ): Effect.Effect;
+ submitSteer(prepared: PreparedSteer): Effect.Effect<{ userTurnId: string }, NativeSteerFailure>;
cancel(sessionId: SessionId, expectedTurnId: string): Effect.Effect;
ownsPrompt(sessionId: SessionId, expectedTurnId: string): boolean;
activeUserTurnId(sessionId: SessionId): string | undefined;
diff --git a/apps/cli/src/session/queue-steer-service.ts b/apps/cli/src/session/queue-steer-service.ts
index ea03e624a..6c586c5be 100644
--- a/apps/cli/src/session/queue-steer-service.ts
+++ b/apps/cli/src/session/queue-steer-service.ts
@@ -397,9 +397,8 @@ export class QueueSteerService extends Context.Tag('lody/QueueSteerService')<
}
yield* flush();
yield* removeReservedRow(doc, marker);
- const submitting = yield* write(sessionId, { ...marker, phase: 'submitting' });
- return yield* active
- .steer({
+ const preparation = yield* Effect.either(
+ active.prepareSteer({
sessionId,
expectedTurnId,
turn: {
@@ -409,43 +408,58 @@ export class QueueSteerService extends Context.Tag('lody/QueueSteerService')<
inputConfig,
},
})
- .pipe(
- Effect.flatMap(() =>
- flush().pipe(
- Effect.flatMap(() => write(sessionId, { ...marker, phase: 'applied' })),
- Effect.flatMap((applied) =>
- complete(
- sessionId,
- applied,
- respond(request, 'accepted', { userTurnId: entry.id })
- )
+ );
+ if (preparation._tag === 'Left') {
+ const error = preparation.left;
+ return yield* fallback(
+ sessionId,
+ doc,
+ marker,
+ respond(
+ request,
+ error._tag === 'SteerPreparationFailure' ? 'error' : error.disposition,
+ { userTurnId: entry.id, error: error.message }
+ )
+ );
+ }
+ const submitting = yield* write(sessionId, { ...marker, phase: 'submitting' });
+ return yield* active.submitSteer(preparation.right).pipe(
+ Effect.flatMap(() =>
+ flush().pipe(
+ Effect.flatMap(() => write(sessionId, { ...marker, phase: 'applied' })),
+ Effect.flatMap((applied) =>
+ complete(
+ sessionId,
+ applied,
+ respond(request, 'accepted', { userTurnId: entry.id })
)
)
- ),
- Effect.catchTag('ProviderRejected', (error) =>
- fallback(
- sessionId,
- doc,
- submitting,
- respond(request, error.disposition, {
- userTurnId: entry.id,
- error: error.message,
- })
- )
- ),
- // Pre-submission ownership loss never cancels a newer active turn.
- Effect.catchTag('StaleTurn', (error) =>
- fallback(
- sessionId,
- doc,
- submitting,
- respond(request, error.disposition, {
- userTurnId: entry.id,
- error: error.message,
- })
- )
)
- );
+ ),
+ Effect.catchTag('ProviderRejected', (error) =>
+ fallback(
+ sessionId,
+ doc,
+ submitting,
+ respond(request, error.disposition, {
+ userTurnId: entry.id,
+ error: error.message,
+ })
+ )
+ ),
+ // Pre-submission ownership loss never cancels a newer active turn.
+ Effect.catchTag('StaleTurn', (error) =>
+ fallback(
+ sessionId,
+ doc,
+ submitting,
+ respond(request, error.disposition, {
+ userTurnId: entry.id,
+ error: error.message,
+ })
+ )
+ )
+ );
}).pipe(
Effect.catchTags({
StaleTurn: (error) =>
diff --git a/apps/cli/src/session/session-execution-service.ts b/apps/cli/src/session/session-execution-service.ts
index 22a4891bb..ef1ea0051 100644
--- a/apps/cli/src/session/session-execution-service.ts
+++ b/apps/cli/src/session/session-execution-service.ts
@@ -67,7 +67,8 @@ import type { ModelInfo } from '@lody/shared';
import { Cause, Context, Data, Effect, Exit, Fiber, Layer, type Scope } from 'effect';
import {
ActiveTurnSteerPort,
- PersistenceFailure,
+ SteerPreparationFailure,
+ PreparedSteer,
ProviderRejected,
ProviderDeliveryUnknown,
StaleTurn,
@@ -756,6 +757,10 @@ export class SessionExecutionService {
// this is pure per-session serialization, matching the old hand-rolled lock.
private readonly steerMutationQueue = new ConcurrentQueue