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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 8 additions & 12 deletions apps/desktop/e2e/streaming-remount.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,18 +22,18 @@ import {
FAKE_HOLD_OPEN_REWRITE_PROMPT,
FAKE_WAIT_FOR_STEERING_LARGE_RESPONSE_PROMPT,
} from '@maka/runtime/test-only/fake-backend';
import type { Locator, Page } from '@playwright/test';
import type { Locator } from '@playwright/test';
import { expect, COMPOSER_INPUT, test } from './fixtures';

function sessionRow(sidebar: Locator, sessionId: string): Locator {
return sidebar.locator(`[data-session-id=${JSON.stringify(sessionId)}]`);
}

async function selectSteerMode(page: Page): Promise<void> {
const steerMode = page.getByRole('radio', { name: '插入消息' });
await expect(steerMode).toBeVisible();
await steerMode.click();
await expect(steerMode).toBeChecked();
async function steerActiveTurn(composer: Locator, text: string): Promise<void> {
// Mid-turn steering is Shift+Enter: the one Send stays Send, and the shifted
// submit hands the draft to the active Turn once.
await composer.fill(text);
await composer.press('Shift+Enter');
}

test('remounting a live surface leaves accumulated output settled', async ({
Expand Down Expand Up @@ -87,9 +87,7 @@ test('remounting a live surface leaves accumulated output settled', async ({
});

const steering = 'trigger rewrite after returning to this conversation';
await selectSteerMode(page);
await composer.fill(steering);
await composer.press('Enter');
await steerActiveTurn(composer, steering);
const finalText = 'prefix <redacted> NEW streamed after the remount';
await expect(liveBubble).toContainText(finalText);

Expand Down Expand Up @@ -164,9 +162,7 @@ test('keeps a completed reply after an interrupted turn and conversation remount
});
const steering = 'use the detailed response';
const completedReply = 'Large response complete.';
await selectSteerMode(page);
await composer.fill(steering);
await composer.press('Enter');
await steerActiveTurn(composer, steering);
await expect(page.getByRole('log')).toContainText(completedReply);
await expect(page.getByRole('button', { name: '停止' })).toHaveCount(0, {
timeout: 20_000,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,15 +49,13 @@ describe('follow-up submit routing', () => {
it('routes burst input through the selected follow-up lane', () => {
assert.equal(
resolveFollowUpModeAtSubmit({
defaultMode: 'queue',
hasActiveTurn: true,
}),
'queue',
);
assert.equal(
resolveFollowUpModeAtSubmit({
requestedMode: 'steer',
defaultMode: 'queue',
hasActiveTurn: true,
}),
'steer',
Expand All @@ -67,7 +65,6 @@ describe('follow-up submit routing', () => {
it('starts a normal turn only when no active-turn witness exists', () => {
assert.equal(
resolveFollowUpModeAtSubmit({
defaultMode: 'queue',
hasActiveTurn: false,
}),
undefined,
Expand Down
18 changes: 7 additions & 11 deletions apps/desktop/src/main/__tests__/message-queue-ui-state.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -62,17 +62,13 @@ test('queue_update events drive the independent desktop queue projection', () =>
}],
});

assert.deepEqual(controller.getState().messageQueueBySession['session-1'], {
queueRevision: 3,
steering: [steeringEntry],
followup: [{
entryId: 'entry-next',
messageId: 'message-next',
content: { text: 'do this next' },
placement: 'next_turn',
state: 'queued',
}],
});
assert.deepEqual(controller.getState().messageQueueBySession['session-1'], [{
entryId: 'entry-next',
messageId: 'message-next',
content: { text: 'do this next' },
placement: 'next_turn',
state: 'queued',
}]);

handlers.handleEvent('session-1', {
type: 'queue_update',
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -393,7 +393,16 @@ test('treats empty configuration patches as read-only lookups', async () => {
test('binds message controls to the current Host Epoch', async () => {
const { client, requests } = clientWithResponses([
{ disposition: 'steering', queueRevision: 2 },
{ queueRevision: 3, retracted: [] },
{
queueRevision: 3,
retracted: {
entryId: 'entry-1',
messageId: 'message-1',
content: { text: 'Steer it' },
placement: 'current_turn',
state: 'retracted',
},
},
{
queueRevision: 4,
retracted: [],
Expand All @@ -414,7 +423,11 @@ test('binds message controls to the current Host Epoch', async () => {
content: { text: 'Steer it' },
placement: 'current_turn',
});
await client.retractQueue({ sessionId: 'session-1', retractId: 'retract-1' });
await client.retractQueueEntry({
sessionId: 'session-1',
entryId: 'entry-1',
retractId: 'retract-1',
});
await client.interruptTurn({
sessionId: 'session-1',
interruptId: 'interrupt-1',
Expand All @@ -434,9 +447,10 @@ test('binds message controls to the current Host Epoch', async () => {
},
},
{
operation: 'queue.retract',
operation: 'queue.entry.retract',
input: {
sessionId: 'session-1',
entryId: 'entry-1',
retractId: 'retract-1',
originHostEpoch: 'host-current',
},
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -679,7 +679,7 @@ test("keeps the busy failure for a Skill send instead of degrading it to steerin
assert.deepEqual(submits, []);
});

test("queues explicit Desktop follow-ups and retracts their full content", async () => {
test("queues explicit Desktop follow-ups", async () => {
const submits: unknown[] = [];
let sequence = 0;
const ipc = ipcHarness();
Expand All @@ -691,28 +691,6 @@ test("queues explicit Desktop follow-ups and retracts their full content", async
submits.push(input);
return { disposition: "followup", queueRevision: 4 };
},
retractQueue: async (input) => ({
queueRevision: 5,
retracted: [
{
entryId: "entry-1",
messageId: "message-1",
content: {
text: "first",
quotes: [{ text: "context" }],
},
placement: "next_turn",
state: "retracted",
},
{
entryId: "entry-2",
messageId: "message-2",
content: { text: "second" },
placement: "next_turn",
state: "retracted",
},
],
}),
}),
observer: unusedObserver(),
attachmentApprovals: createAttachmentApprovalRegistry(),
Expand Down Expand Up @@ -786,12 +764,83 @@ test("queues explicit Desktop follow-ups and retracts their full content", async
placement: "next_turn",
},
]);
});

test("routes per-entry queue mutations to the Runtime Host", async () => {
const calls: unknown[] = [];
let sequence = 0;
const ipc = ipcHarness();
registerExecutionIpc(
{
client: executionClient({
retractQueueEntry: async (input) => {
calls.push({ operation: "retract", ...input });
return {
queueRevision: 3,
retracted: {
entryId: input.entryId,
messageId: "message-1",
content: { text: "queued text" },
placement: "next_turn",
state: "retracted",
},
};
},
promoteQueueEntry: async (input) => {
calls.push({ operation: "promote", ...input });
return { queueRevision: 4 };
},
reorderQueueEntries: async (input) => {
calls.push({ operation: "reorder", ...input });
return { queueRevision: 5 };
},
}),
observer: unusedObserver(),
attachmentApprovals: createAttachmentApprovalRegistry(),
emitSessionsChanged() {},
stat: async () => ({ size: 0 }),
resizeImage: async (bytes) => bytes,
beforeStop() {},
newId: () => `id-${++sequence}`,
},
ipc,
);

assert.deepEqual(
await ipc.invoke("sessions:retractQueue", "session-1"),
await ipc.invoke("sessions:retractQueueEntry", "session-1", "entry-1"),
{ text: "queued text" },
);
await ipc.invoke("sessions:promoteQueueEntry", "session-1", "entry-2");
await ipc.invoke("sessions:reorderQueueEntries", "session-1", ["entry-3", "entry-2"]);

assert.deepEqual(calls, [
{
text: "first\n\nsecond",
quotes: [{ text: "context" }],
operation: "retract",
sessionId: "session-1",
entryId: "entry-1",
retractId: "id-1",
},
{
operation: "promote",
sessionId: "session-1",
entryId: "entry-2",
promoteId: "id-2",
},
{
operation: "reorder",
sessionId: "session-1",
reorderId: "id-3",
entryIds: ["entry-3", "entry-2"],
},
]);

await assert.rejects(
() => ipc.invoke("sessions:promoteQueueEntry", "session-1", 42),
/Invalid queue entry identity/,
);
await assert.rejects(
() => ipc.invoke("sessions:reorderQueueEntries", "session-1", ["entry-1", 42]),
/Invalid queue entry order/,
);
});

Expand Down Expand Up @@ -886,7 +935,9 @@ function executionClient(overrides: Partial<ExecutionClient>): ExecutionClient {
queryTurnResume: unavailable,
readExecutionBoundary: unavailable,
regenerateTurn: unavailable,
retractQueue: unavailable,
retractQueueEntry: unavailable,
promoteQueueEntry: unavailable,
reorderQueueEntries: unavailable,
setSessionReadMarker: unavailable,
startTurn: unavailable,
startTurnResume: unavailable,
Expand Down
34 changes: 28 additions & 6 deletions apps/desktop/src/main/runtime-host-client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -89,8 +89,12 @@ import {
PROJECT_DIRECTORY_MAX_ENTRIES,
type ProjectDirectoryEntry,
type ProjectDirectoryRoot,
type QueueRetractInput,
type QueueRetractResult,
type QueueEntriesReorderInput,
type QueueEntriesReorderResult,
type QueueEntryPromoteInput,
type QueueEntryPromoteResult,
type QueueEntryRetractInput,
type QueueEntryRetractResult,
SESSION_TRANSCRIPT_BOOTSTRAP_MAX_BYTES,
type SessionCatalogChangedFrame,
type ScheduledTaskChangedFrame,
Expand Down Expand Up @@ -1056,10 +1060,28 @@ export class DesktopRuntimeHostClient {
});
}

retractQueue(
input: Omit<QueueRetractInput, "originHostEpoch">,
): Promise<QueueRetractResult> {
return this.request("queue.retract", {
retractQueueEntry(
input: Omit<QueueEntryRetractInput, "originHostEpoch">,
): Promise<QueueEntryRetractResult> {
return this.request("queue.entry.retract", {
...input,
originHostEpoch: this.connection.hostEpoch,
});
}

promoteQueueEntry(
input: Omit<QueueEntryPromoteInput, "originHostEpoch">,
): Promise<QueueEntryPromoteResult> {
return this.request("queue.entry.promote", {
...input,
originHostEpoch: this.connection.hostEpoch,
});
}

reorderQueueEntries(
input: Omit<QueueEntriesReorderInput, "originHostEpoch">,
): Promise<QueueEntriesReorderResult> {
return this.request("queue.entries.reorder", {
...input,
originHostEpoch: this.connection.hostEpoch,
});
Expand Down
55 changes: 46 additions & 9 deletions apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ import type { IpcMainInvokeEvent } from "electron";
import { MAX_ATTACHMENT_COUNT } from '@maka/core/attachments';
import { RuntimeHostOperationError } from '@maka/runtime-host/client';
import { SKILL_INVOCATION_TOKEN_SOURCE } from '@maka/core/skill-invocation-token';
import { aggregateMessageContents } from '@maka/core/events';
import {
type SessionChangedEvent,
type SessionChangedReason,
Expand Down Expand Up @@ -73,7 +72,9 @@ type RuntimeHostSessionExecutionClient = Pick<
| "queryTurnResume"
| "readExecutionBoundary"
| "regenerateTurn"
| "retractQueue"
| "retractQueueEntry"
| "promoteQueueEntry"
| "reorderQueueEntries"
| "setSessionReadMarker"
| "startTurn"
| "startTurnResume"
Expand Down Expand Up @@ -439,13 +440,49 @@ export function registerRuntimeHostSessionExecutionIpc(
};
},
);
ipcMain.handle("sessions:retractQueue", async (_event, sessionId: string) => {
const result = await deps.client.retractQueue({
sessionId,
retractId: newId(),
});
return aggregateMessageContents(result.retracted.map((entry) => entry.content));
});
ipcMain.handle(
"sessions:retractQueueEntry",
async (_event, sessionId: string, entryId: unknown) => {
if (typeof entryId !== "string") {
throw new TypeError("Invalid queue entry identity");
}
const result = await deps.client.retractQueueEntry({
sessionId,
entryId,
retractId: newId(),
});
return result.retracted.content;
},
);
ipcMain.handle(
"sessions:promoteQueueEntry",
async (_event, sessionId: string, entryId: unknown) => {
if (typeof entryId !== "string") {
throw new TypeError("Invalid queue entry identity");
}
await deps.client.promoteQueueEntry({
sessionId,
entryId,
promoteId: newId(),
});
},
);
ipcMain.handle(
"sessions:reorderQueueEntries",
async (_event, sessionId: string, entryIds: unknown) => {
if (
!Array.isArray(entryIds) ||
entryIds.some((entryId) => typeof entryId !== "string")
) {
throw new TypeError("Invalid queue entry order");
}
await deps.client.reorderQueueEntries({
sessionId,
reorderId: newId(),
entryIds,
});
},
);
ipcMain.handle("sessions:stop", async (_event, sessionId: string) =>
stopSession(sessionId),
);
Expand Down
Loading