puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-26 03:03:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(timeline): make submitted prompts canonical

Browse Source 
Submission settlement previously depended on status updates that could arrive after the optimistic row. Persist every accepted prompt in the canonical timeline and keep the footer active until RPC and canonical acknowledgement converge.
This commit is contained in:
Mohamed Boudra
2026-07-31 00:43:22 +02:00
parent b98df29b19
commit a1ff4773e5
33 changed files with 551 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/timeline-sync.md
+19 -5
View File
@@ -126,11 +126,25 @@ back a prompt already observed canonically; it does not own whether the local pr
visible.
The daemon's accepted response already waits for the correlated run start, but its response and the
directory update reach client state separately. An accepted transaction remains active until the
directory observes that run or canonical ingestion acknowledges the prompt, bridging those ordered
authorities without inspecting timeline snapshots. Either signal clears only an RPC-accepted
transaction, regardless of which arrived first; it cannot settle a fresh send.
Overlapping sends settle independently rather than collapsing to one newest pending message.
canonical submitted row reach client state independently and may arrive in either order. The
transaction therefore records RPC acceptance and provider acknowledgement as separate phases. RPC
acceptance marks the record accepted unless provider acknowledgement already arrived; canonical
acknowledgement marks the record acknowledged unless RPC acceptance already arrived. Only the second
authority removes the transaction. Directory status never settles a submission. Overlapping sends
settle independently rather than collapsing to one newest pending message.
Daemons advertising `server_info.features.canonicalSubmittedPrompts` guarantee that every accepted
prompt carrying a client message id is recorded and streamed as a canonical `user_message` with that
same id. This includes daemon-handled commands that do not allocate a foreground turn; their submitted
row is recorded before handler output. The app creates two-phase submission records only for hosts with
this capability. Older hosts keep the shipped untracked optimistic-row behavior and roll that row back
on RPC rejection.
Turn liveness is the union of an unsettled submission and an open stream turn. `turn_started` opens the
turn; ordered `turn_completed`, `turn_failed`, and `turn_canceled` events close it. A running directory
transition may seed or raise an open turn during hydration or after a selective stream subscription
lapses, but directory status never closes one or settles a submission. Disconnect and replica removal
are the other destructive close boundaries.
Canonical submitted user rows carry the provider's `messageId` and Paseo's optional
`clientMessageId`. The user-message producer reconciles them by `clientMessageId`, adds provider
packages/app/e2e/agent-consecutive-turns.spec.ts
-6
View File
@@ -533,8 +533,6 @@ async function recordDelayedRunningTransition(
const prompt = "Second prompt keeps streaming.";
gate.holdNextAgentUpdate(agent.agentId, "running");
gate.setMessageSubmissionDispositionStripped(true);
gate.setAgentStreamEventSuppressed("turn_started", true);
gate.setAgentStreamItemSuppressed("assistant_message", true);
await recordTurnFrames(page, prompt);
await submitMessage(page, prompt);
@@ -553,14 +551,10 @@ async function recordDelayedRunningTransition(
released = true;
await waitForRecordedFrames(page, "turn-running", 3);
const frames = await stopTurnFrameRecording(page);
gate.setMessageSubmissionDispositionStripped(false);
gate.setAgentStreamEventSuppressed("turn_started", false);
gate.setAgentStreamItemSuppressed("assistant_message", false);
return frames;
} finally {
if (!released) gate.releaseHeldServerMessage();
gate.setMessageSubmissionDispositionStripped(false);
gate.setAgentStreamEventSuppressed("turn_started", false);
gate.setAgentStreamItemSuppressed("assistant_message", false);
}
}
packages/app/e2e/agent-message-submission.spec.ts
+94
View File
@@ -11,6 +11,7 @@ import {
fillComposerDraft,
sendDraftToQueue,
startRunningMockAgent,
submitMessage,
} from "./helpers/composer";
import { openAgentRoute, seedMockAgentWorkspace } from "./helpers/mock-agent";
import { seedWorkspace } from "./helpers/seed-client";
@@ -517,6 +518,89 @@ async function expectCanonicalOrderWinsAcrossOverlappingClients(
}
}
async function expectDaemonHandledSubmissionSurvivesReload(
page: Page,
testInfo: { workerIndex: number },
): Promise<void> {
const agent = await seedMockAgentWorkspace({
repoPrefix: `daemon-handled-submission-${testInfo.workerIndex}-`,
title: "Daemon-handled submission",
});
const prompt = "/mock handled-command";
try {
await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId });
await expectComposerVisible(page);
await expectAgentIdle(page);
await submitMessage(page, prompt);
const command = page.getByTestId("user-message").filter({ hasText: prompt });
await expect(command).toHaveAttribute("aria-busy", "false");
await expect(page.getByText("Mock command handled", { exact: true })).toBeVisible();
await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0);
await page.reload();
await expectComposerVisible(page);
await expect(command).toBeVisible();
await expect(command).toHaveAttribute("aria-busy", "false");
await expect(page.getByText("Mock command handled", { exact: true })).toBeVisible();
await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0);
} finally {
await agent.cleanup();
}
}
async function expectOldHostSubmissionBehavior(
page: Page,
testInfo: { workerIndex: number },
): Promise<void> {
const gate = await installDaemonWebSocketGate(page);
gate.setCanonicalSubmittedPromptsStripped(true);
const agent = await seedMockAgentWorkspace({
repoPrefix: `old-host-submission-${testInfo.workerIndex}-`,
title: "Old host submission",
});
const prompt = "/mock handled-command";
try {
await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId });
await expectComposerVisible(page);
await expectAgentIdle(page);
gate.holdNextClientRequest("send_agent_message_request");
const composer = page.getByRole("textbox", { name: "Message agent..." }).first();
await composer.fill(prompt);
await composer.press("Enter");
const nextFrame = await composer.evaluate(
(_, submittedPrompt) =>
new Promise<{ rowPresent: boolean; ariaBusy: string | null; workingPresent: boolean }>(
(resolve) => {
requestAnimationFrame(() => {
const row = Array.from(
document.querySelectorAll('[data-testid="user-message"]'),
).find((candidate) => candidate.textContent?.includes(submittedPrompt));
resolve({
rowPresent: Boolean(row),
ariaBusy: row?.getAttribute("aria-busy") ?? null,
workingPresent: Boolean(
document.querySelector('[data-testid="turn-working-indicator"]'),
),
});
});
},
),
prompt,
);
expect(nextFrame).toEqual({ rowPresent: true, ariaBusy: "false", workingPresent: false });
await gate.waitForHeldClientRequest();
gate.releaseHeldClientRequest();
const command = page.getByTestId("user-message").filter({ hasText: prompt });
await expect(command).toHaveAttribute("aria-busy", "false");
await expect(page.getByText("Mock command handled", { exact: true })).toBeVisible();
await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0);
} finally {
gate.restore();
await agent.cleanup();
}
}
async function expectRenderedBefore(first: Locator, second: Locator): Promise<void> {
const secondElement = await second.elementHandle();
if (!secondElement) throw new Error("Expected the second timeline item to be rendered");
@@ -711,4 +795,14 @@ test.describe("Agent message submission", () => {
}, testInfo) => {
await expectCanonicalOrderWinsAcrossOverlappingClients(page, testInfo);
});
test("keeps a daemon-handled submitted row after authoritative reload", async ({
page,
}, testInfo) => {
await expectDaemonHandledSubmissionSurvivesReload(page, testInfo);
});
test("uses untracked optimistic rows for an old host", async ({ page }, testInfo) => {
await expectOldHostSubmissionBehavior(page, testInfo);
});
});
packages/app/e2e/helpers/daemon-websocket-gate.ts
+11 -13
View File
@@ -73,21 +73,19 @@ function stripAssistantMessageId(
return JSON.stringify(envelope);
}
function stripMessageSubmissionDisposition(
function stripCanonicalSubmittedPrompts(
message: string | Buffer,
enabled: boolean,
messageType: unknown,
): string | Buffer {
if (!enabled || messageType !== "send_agent_message_response" || typeof message !== "string") {
return message;
}
if (!enabled || messageType !== "status" || typeof message !== "string") return message;
const envelope = JSON.parse(message) as {
message?: { payload?: Record<string, unknown> };
payload?: Record<string, unknown>;
message?: { payload?: { status?: unknown; features?: Record<string, unknown> } };
payload?: { status?: unknown; features?: Record<string, unknown> };
};
const payload = envelope.message?.payload ?? envelope.payload;
if (!payload) return message;
delete payload.outOfBand;
if (payload?.status !== "server_info" || !payload.features) return message;
delete payload.features.canonicalSubmittedPrompts;
return JSON.stringify(envelope);
}
@@ -164,7 +162,7 @@ export async function installDaemonWebSocketGate(page: Page) {
let suppressAgentStream = false;
let forceTimelineEpochReset = false;
let stripAssistantMessageIds = false;
let stripSubmissionDisposition = false;
let stripCanonicalSubmittedPromptsFeature = false;
let heldClientRequestType: string | null = null;
let heldClientRequest: { server: WebSocketRoute; message: string | Buffer } | null = null;
let resolveHeldClientRequest: (() => void) | null = null;
@@ -240,9 +238,9 @@ export async function installDaemonWebSocketGate(page: Page) {
stripAssistantMessageIds,
serverMessage?.type,
);
outboundMessage = stripMessageSubmissionDisposition(
outboundMessage = stripCanonicalSubmittedPrompts(
outboundMessage,
stripSubmissionDisposition,
stripCanonicalSubmittedPromptsFeature,
serverMessage?.type,
);
const shouldForceTimelineReset =
@@ -428,8 +426,8 @@ export async function installDaemonWebSocketGate(page: Page) {
setAssistantMessageIdsStripped(stripped: boolean): void {
stripAssistantMessageIds = stripped;
},
setMessageSubmissionDispositionStripped(stripped: boolean): void {
stripSubmissionDisposition = stripped;
setCanonicalSubmittedPromptsStripped(stripped: boolean): void {
stripCanonicalSubmittedPromptsFeature = stripped;
},
setAgentStreamSuppressed(suppressed: boolean): void {
suppressAgentStream = suppressed;
packages/app/e2e/out-of-band-command.codex.real.spec.ts
-23
View File
@@ -4,7 +4,6 @@ import path from "node:path";
import { expect, test } from "./fixtures";
import { submitMessage } from "./helpers/composer";
import { cleanupRewindFlow, launchAgent, type AgentHandle } from "./helpers/rewind-flow";
import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate";
test.describe("Codex out-of-band commands", () => {
test.setTimeout(300_000);
@@ -27,26 +26,4 @@ test.describe("Codex out-of-band commands", () => {
await cleanupRewindFlow({ handle, cwd });
}
});
test("settles the submitted row when an older daemon omits submission disposition", async ({
page,
}) => {
const gate = await installDaemonWebSocketGate(page);
gate.setMessageSubmissionDispositionStripped(true);
const cwd = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-codex-command-compat-")));
let handle: AgentHandle | undefined;
try {
handle = await launchAgent({ page, provider: "codex", cwd, mode: "full-access" });
await submitMessage(page, "/goal clear");
const command = page.getByTestId("user-message").filter({ hasText: "/goal clear" });
await expect(command).toBeVisible();
await expect(command).toHaveAttribute("aria-busy", "false", { timeout: 30_000 });
await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0);
} finally {
gate.restore();
await cleanupRewindFlow({ handle, cwd });
}
});
});
packages/app/src/agent-stream/view.tsx
+1 -5
View File
@@ -565,11 +565,7 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
}
const effectiveStreamItems = isActive ? streamItems : frozenStreamItemsRef.current;
const effectiveStreamHead = isActive ? streamHead : frozenStreamHeadRef.current;
const isTurnActive = [
context.status === "running",
pendingMessageSubmissions.length > 0,
hasActiveTurn,
].some(Boolean);
const isTurnActive = pendingMessageSubmissions.length > 0 || hasActiveTurn;
// Keep retained history outside the 48ms live-head flush path.
const preparedToolCallHistory = useMemo(
() => prepareToolCallHistory(toolCallDetailLevel, effectiveStreamItems),
packages/app/src/composer/actions.test.ts
+1 -1
View File
@@ -223,7 +223,7 @@ function createFakeStream(initialHead: Map<string, StreamItem[]> = new Map()): F
const current = readSubmission(fake, agentId);
writeSubmission(fake, agentId, {
...current,
submissions: acceptMessageSubmission(current.submissions, clientMessageId, true, false),
submissions: acceptMessageSubmission(current.submissions, clientMessageId),
});
},
reject: (agentId, clientMessageId) => {
packages/app/src/composer/actions.ts
+4 -4
View File
@@ -46,7 +46,7 @@ export interface ComposerSendClient {
images: Array<{ data: string; mimeType: string }>;
attachments: ReturnType<typeof splitComposerAttachmentsForSubmit>["attachments"];
},
) => Promise<void | { outOfBand?: boolean }>;
) => Promise<void>;
uploadFile: (input: { fileName: string; mimeType: string; bytes: Uint8Array }) => Promise<{
requestId: string;
file: {
@@ -67,7 +67,7 @@ export interface ComposerCancelClient {
export interface MessageSubmissionWriter {
begin: (agentId: string, message: UserMessageItem) => void;
accept: (agentId: string, clientMessageId: string, outOfBand: boolean | undefined) => void;
accept: (agentId: string, clientMessageId: string) => void;
reject: (agentId: string, clientMessageId: string) => MessageSubmissionRejectionOutcome;
}
@@ -183,12 +183,12 @@ export async function dispatchComposerAgentMessage(
input.submission.begin(input.agentId, userMessage);
try {
const imagesData = await input.encodeImages(wirePayload.images);
const result = await input.client.sendAgentMessage(input.agentId, input.text, {
await input.client.sendAgentMessage(input.agentId, input.text, {
messageId: clientMessageId,
images: imagesData ?? [],
attachments: wirePayload.attachments,
});
input.submission.accept(input.agentId, clientMessageId, result?.outOfBand);
input.submission.accept(input.agentId, clientMessageId);
} catch (error) {
const outcome = input.submission.reject(input.agentId, clientMessageId);
if (outcome === "accepted") return;
packages/app/src/composer/draft/create-flow.ts
+2 -6
View File
@@ -6,6 +6,7 @@ import {
splitComposerAttachmentsForSubmit,
} from "@/composer/attachments/submit";
import { useCreateFlowStore } from "@/stores/create-flow-store";
import { handoffCreatedAgentMessageSubmission } from "@/composer/submission/writer";
import { useSessionStore } from "@/stores/session-store";
import {
createUserMessage,
@@ -127,10 +128,6 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
const updatePendingAgentId = useCreateFlowStore((state) => state.updateAgentId);
const markPendingCreateLifecycle = useCreateFlowStore((state) => state.markLifecycle);
const clearPendingCreateAttempt = useCreateFlowStore((state) => state.clear);
const handoffCreatedAgentUserMessage = useSessionStore(
(state) => state.handoffCreatedAgentUserMessage,
);
const formErrorMessage = machine.tag === "draft" ? machine.errorMessage : "";
const isSubmitting = machine.tag === "creating";
@@ -202,7 +199,7 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
if (createResult.agentId) {
updatePendingAgentId({ draftId, agentId: createResult.agentId });
handoffCreatedAgentUserMessage(
handoffCreatedAgentMessageSubmission(
pendingServerId,
createResult.agentId,
createUserMessage({
@@ -228,7 +225,6 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
}
},
[
handoffCreatedAgentUserMessage,
clearPendingCreateAttempt,
createRequest,
draftId,
packages/app/src/composer/submission/model.test.ts
+19 -19
View File
@@ -4,7 +4,6 @@ import {
beginMessageSubmission,
getActiveMessageSubmissions,
getSendingClientMessageIds,
observeAcceptedMessageSubmissionsRunning,
observeMessageSubmissionCanonical,
rejectMessageSubmission,
} from "./model";
@@ -26,13 +25,14 @@ describe("message submission transactions", () => {
expect(getSendingClientMessageIds(both)).toEqual(["client-1", "client-2"]);
});
it("removes only the RPC-accepted transaction", () => {
it("removes only the provider-acknowledged transaction on RPC acceptance", () => {
const both = beginMessageSubmission(
beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }),
{ clientMessageId: "client-2", submittedAt },
);
const acknowledged = observeMessageSubmissionCanonical(both, ["client-1"]);
expect(acceptMessageSubmission(both, "client-1", true, false)).toEqual([
expect(acceptMessageSubmission(acknowledged, "client-1")).toEqual([
{
clientMessageId: "client-2",
submittedAt,
@@ -42,34 +42,34 @@ describe("message submission transactions", () => {
]);
});
it("bridges an accepted RPC until the correlated running state is observed", () => {
it("keeps an accepted RPC active until canonical acknowledgement", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
const accepted = acceptMessageSubmission(sending, "client-1", false, false);
const accepted = acceptMessageSubmission(sending, "client-1");
expect(getActiveMessageSubmissions(accepted)).toHaveLength(1);
expect(accepted[0].rpcAccepted).toBe(true);
expect(observeAcceptedMessageSubmissionsRunning(accepted)).toEqual([]);
});
it("does not settle RPC acceptance from directory running status", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
expect(acceptMessageSubmission(sending, "client-1")).toEqual([
{
clientMessageId: "client-1",
submittedAt,
rpcAccepted: true,
providerAcknowledged: false,
},
]);
});
it("settles an accepted RPC when provider acknowledgement arrives after running was missed", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
const accepted = acceptMessageSubmission(sending, "client-1", false, false);
const accepted = acceptMessageSubmission(sending, "client-1");
expect(observeMessageSubmissionCanonical(accepted, ["client-1"])).toEqual([]);
});
it("settles an explicitly out-of-band acceptance without lifecycle inference", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
expect(acceptMessageSubmission(sending, "client-1", false, true)).toEqual([]);
});
it("settles an idle acceptance from a daemon without submission disposition", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
expect(acceptMessageSubmission(sending, "client-1", false, undefined)).toEqual([]);
});
it("records provider acknowledgement without settling another transaction", () => {
const both = beginMessageSubmission(
beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }),
packages/app/src/composer/submission/model.ts
+1 -19
View File
@@ -44,23 +44,12 @@ export function beginMessageSubmission(
export function acceptMessageSubmission(
submissions: readonly MessageSubmissionRecord[],
clientMessageId: string,
isAgentRunning: boolean,
outOfBand: boolean | undefined,
): MessageSubmissionRecord[] {
const index = submissions.findIndex(
(submission) => submission.clientMessageId === clientMessageId,
);
if (index < 0) return submissions as MessageSubmissionRecord[];
// COMPAT(messageSubmissionDisposition): daemons before v0.2.3 omitted outOfBand.
// An idle response cannot distinguish a normal send that has not reported running yet
// from an out-of-band send; the separate active-turn lease bridges that gap. Remove after 2027-01-27.
const legacyOutOfBand = outOfBand === undefined && !isAgentRunning;
if (
outOfBand === true ||
legacyOutOfBand ||
isAgentRunning ||
submissions[index].providerAcknowledged
) {
if (submissions[index].providerAcknowledged) {
return submissions.filter((_, submissionIndex) => submissionIndex !== index);
}
if (submissions[index].rpcAccepted) return submissions as MessageSubmissionRecord[];
@@ -69,13 +58,6 @@ export function acceptMessageSubmission(
return next;
}
export function observeAcceptedMessageSubmissionsRunning(
submissions: readonly MessageSubmissionRecord[],
): MessageSubmissionRecord[] {
const next = submissions.filter((submission) => !submission.rpcAccepted);
return next.length === submissions.length ? (submissions as MessageSubmissionRecord[]) : next;
}
export function observeMessageSubmissionCanonical(
submissions: readonly MessageSubmissionRecord[],
clientMessageIds: readonly string[],
packages/app/src/composer/submission/writer.ts
+67 -7
View File
@@ -1,20 +1,80 @@
import type { MessageSubmissionWriter } from "@/composer/actions";
import { useSessionStore } from "@/stores/session-store";
import {
appendSubmittedUserMessage,
removeSubmittedUserMessage,
type UserMessageItem,
} from "@/types/stream";
function appendUntrackedSubmission(serverId: string, agentId: string, message: UserMessageItem) {
const session = useSessionStore.getState().sessions[serverId];
if (!session) return;
const tail = session.agentStreamTail.get(agentId) ?? [];
const head = session.agentStreamHead.get(agentId) ?? [];
useSessionStore
.getState()
.setAgentStreamState(serverId, agentId, appendSubmittedUserMessage({ tail, head, message }));
}
function removeUntrackedSubmission(
serverId: string,
agentId: string,
clientMessageId: string,
): void {
const session = useSessionStore.getState().sessions[serverId];
if (!session) return;
const tail = session.agentStreamTail.get(agentId) ?? [];
const head = session.agentStreamHead.get(agentId) ?? [];
useSessionStore
.getState()
.setAgentStreamState(
serverId,
agentId,
removeSubmittedUserMessage({ tail, head, clientMessageId }),
);
}
function createUntrackedMessageSubmissionWriter(serverId: string): MessageSubmissionWriter {
return {
begin: (agentId, message) => appendUntrackedSubmission(serverId, agentId, message),
accept: () => undefined,
reject: (agentId, clientMessageId) => {
removeUntrackedSubmission(serverId, agentId, clientMessageId);
return "rejected";
},
};
}
/**
* Binds the submission lifecycle to a host session. Every path that sends a message to an
* agent — composer send, queued send-now, automatic queue drain — goes through this so a
* submitted row and its pending state are always created together.
* Binds message submission presentation to a host session. Capable hosts use the tracked
* two-phase lifecycle; older hosts preserve the untracked optimistic-row behavior.
*/
export function createMessageSubmissionWriter(serverId: string): MessageSubmissionWriter {
const supportsCanonicalSubmittedPrompts =
useSessionStore.getState().sessions[serverId]?.serverInfo?.features
?.canonicalSubmittedPrompts === true;
if (!supportsCanonicalSubmittedPrompts) {
return createUntrackedMessageSubmissionWriter(serverId);
}
return {
begin: (agentId, message) =>
useSessionStore.getState().beginAgentMessageSubmission(serverId, agentId, message),
accept: (agentId, clientMessageId, outOfBand) =>
useSessionStore
.getState()
.acceptAgentMessageSubmission(serverId, agentId, clientMessageId, outOfBand),
accept: (agentId, clientMessageId) =>
useSessionStore.getState().acceptAgentMessageSubmission(serverId, agentId, clientMessageId),
reject: (agentId, clientMessageId) =>
useSessionStore.getState().rejectAgentMessageSubmission(serverId, agentId, clientMessageId),
};
}
export function handoffCreatedAgentMessageSubmission(
serverId: string,
agentId: string,
message: UserMessageItem,
): boolean {
const supportsCanonicalSubmittedPrompts =
useSessionStore.getState().sessions[serverId]?.serverInfo?.features
?.canonicalSubmittedPrompts === true;
return useSessionStore
.getState()
.handoffCreatedAgentUserMessage(serverId, agentId, message, supportsCanonicalSubmittedPrompts);
}
packages/app/src/runtime/host-runtime.test.ts
+6 -1
View File
@@ -89,7 +89,6 @@ class FakeDaemonClient {
if (response) await response;
const failure = this.sendAgentMessageFailures.shift();
if (failure) throw failure;
return {};
}
async waitForSentMessages(count: number): Promise<void> {
@@ -2280,6 +2279,12 @@ describe("HostRuntimeStore", () => {
});
const sessionStore = useSessionStore.getState();
sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
sessionStore.updateSessionServerInfo(host.serverId, {
serverId: host.serverId,
hostname: null,
version: null,
features: { canonicalSubmittedPrompts: true },
});
sessionStore.setQueuedMessages(
host.serverId,
new Map([
packages/app/src/stores/session-store.test.ts
+52
View File
@@ -7,6 +7,7 @@ import {
normalizeWorkspaceDescriptor,
selectAgentTimelineState,
useSessionStore,
type Agent,
type WorkspaceDescriptor,
} from "./session-store";
import type { StreamItem } from "../types/stream";
@@ -41,6 +42,37 @@ function initializeTestSession(): void {
useSessionStore.getState().initializeSession("test-server", null as unknown as DaemonClient);
}
function createAgent(status: Agent["status"]): Agent {
const timestamp = new Date("2026-07-30T10:00:00.000Z");
return {
serverId: "test-server",
id: "agent-1",
provider: "codex",
status,
createdAt: timestamp,
updatedAt: timestamp,
lastUserMessageAt: null,
lastActivityAt: timestamp,
capabilities: {
supportsStreaming: true,
supportsSessionPersistence: true,
supportsDynamicModes: true,
supportsMcpServers: true,
supportsReasoningStream: true,
supportsToolInvocations: true,
},
currentModeId: null,
availableModes: [],
pendingPermissions: [],
persistence: null,
title: "Agent",
cwd: "/repo",
model: null,
parentAgentId: null,
labels: {},
};
}
function getTestSessionReferences() {
const state = useSessionStore.getState();
const session = state.sessions["test-server"];
@@ -101,6 +133,26 @@ describe("agent timeline state", () => {
selectAgentTimelineState(useSessionStore.getState().sessions["test-server"], "agent-1"),
).toEqual({ status: "synced", items: [], range: null, older: "none" });
});
it("lets a later running directory transition raise but never lower an open turn", () => {
initializeTestSession();
const store = useSessionStore.getState();
store.setAgents("test-server", new Map([["agent-1", createAgent("idle")]]));
store.applyAgentTimelineResponseState("test-server", "agent-1", {
items: [],
head: [],
range: null,
older: "none",
synchronized: true,
acknowledgedClientMessageIds: [],
});
store.setAgents("test-server", new Map([["agent-1", createAgent("running")]]));
expect(store.getSession("test-server")?.activeAgentTurns.has("agent-1")).toBe(true);
store.setAgents("test-server", new Map([["agent-1", createAgent("idle")]]));
expect(store.getSession("test-server")?.activeAgentTurns.has("agent-1")).toBe(true);
});
});
describe("normalizeWorkspaceDescriptor", () => {
packages/app/src/stores/session-store.ts
+43 -65
View File
@@ -7,6 +7,7 @@ import type { AgentDirectoryEntry } from "@/types/agent-directory";
import {
appendSubmittedUserMessage,
handoffCreatedAgentUserMessageToStream,
isUnreconciledLocalUserMessage,
removeSubmittedUserMessage,
type StreamItem,
type UserMessageItem,
@@ -14,7 +15,6 @@ import {
import {
acceptMessageSubmission,
beginMessageSubmission,
observeAcceptedMessageSubmissionsRunning,
observeMessageSubmissionCanonical,
rejectMessageSubmission,
type MessageSubmissionRecord,
@@ -513,7 +513,6 @@ interface SessionStoreActions {
serverId: string,
agentId: string,
clientMessageId: string,
outOfBand: boolean | undefined,
) => void;
rejectAgentMessageSubmission: (
serverId: string,
@@ -524,6 +523,7 @@ interface SessionStoreActions {
serverId: string,
agentId: string,
message: UserMessageItem,
trackSubmission: boolean,
) => boolean;
clearAgentStreamHead: (serverId: string, agentId: string) => void;
setAgentTimelineCursor: (
@@ -638,27 +638,6 @@ type SessionStore = SessionStoreState & SessionStoreActions;
const agentLastActivityCoalescer = createAgentLastActivityCoalescer();
function applyRunningAgentsToAcceptedSubmissions(input: {
previousAgents: Map<string, Agent>;
nextAgents: Map<string, Agent>;
submissions: Map<string, MessageSubmissionRecord[]>;
}): Map<string, MessageSubmissionRecord[]> {
let nextSubmissions = input.submissions;
for (const [agentId, submissions] of input.submissions) {
const previousAgent = input.previousAgents.get(agentId);
const nextAgent = input.nextAgents.get(agentId);
if (!nextAgent || previousAgent?.status === "running" || nextAgent.status !== "running") {
continue;
}
const remaining = observeAcceptedMessageSubmissionsRunning(submissions);
if (remaining === submissions) continue;
if (nextSubmissions === input.submissions) nextSubmissions = new Map(input.submissions);
if (remaining.length > 0) nextSubmissions.set(agentId, remaining);
else nextSubmissions.delete(agentId);
}
return nextSubmissions;
}
function updateActiveAgentTurn(
activeAgentTurns: ReadonlySet<string>,
agentId: string,
@@ -1222,7 +1201,6 @@ export const useSessionStore = create<SessionStore>()(
stream.head === currentHead
? session.agentStreamHead
: new Map(session.agentStreamHead).set(agentId, stream.head),
activeAgentTurns: updateActiveAgentTurn(session.activeAgentTurns, agentId, true),
messageSubmissions,
},
},
@@ -1230,30 +1208,13 @@ export const useSessionStore = create<SessionStore>()(
});
},
acceptAgentMessageSubmission: (serverId, agentId, clientMessageId, outOfBand) => {
acceptAgentMessageSubmission: (serverId, agentId, clientMessageId) => {
set((prev) => {
const session = prev.sessions[serverId];
if (!session) return prev;
const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
const isAgentRunning = session.agents.get(agentId)?.status === "running";
const submissions = acceptMessageSubmission(
currentSubmissions,
clientMessageId,
isAgentRunning,
outOfBand,
);
let activeAgentTurns = session.activeAgentTurns;
if (
submissions !== currentSubmissions &&
!isAgentRunning &&
outOfBand === true &&
submissions.length === 0
) {
activeAgentTurns = updateActiveAgentTurn(activeAgentTurns, agentId, false);
}
if (submissions === currentSubmissions && activeAgentTurns === session.activeAgentTurns) {
return prev;
}
const submissions = acceptMessageSubmission(currentSubmissions, clientMessageId);
if (submissions === currentSubmissions) return prev;
const messageSubmissions = new Map(session.messageSubmissions);
if (submissions.length > 0) {
messageSubmissions.set(agentId, submissions);
@@ -1264,7 +1225,7 @@ export const useSessionStore = create<SessionStore>()(
...prev,
sessions: {
...prev.sessions,
[serverId]: { ...session, activeAgentTurns, messageSubmissions },
[serverId]: { ...session, messageSubmissions },
},
};
});
@@ -1295,12 +1256,6 @@ export const useSessionStore = create<SessionStore>()(
} else {
messageSubmissions.delete(agentId);
}
const activeAgentTurns =
outcome === "rejected" &&
result.submissions.length === 0 &&
session.agents.get(agentId)?.status !== "running"
? updateActiveAgentTurn(session.activeAgentTurns, agentId, false)
: session.activeAgentTurns;
return {
...prev,
sessions: {
@@ -1315,7 +1270,6 @@ export const useSessionStore = create<SessionStore>()(
stream.head === currentHead
? session.agentStreamHead
: new Map(session.agentStreamHead).set(agentId, stream.head),
activeAgentTurns,
messageSubmissions,
},
},
@@ -1324,7 +1278,7 @@ export const useSessionStore = create<SessionStore>()(
return outcome;
},
handoffCreatedAgentUserMessage: (serverId, agentId, message) => {
handoffCreatedAgentUserMessage: (serverId, agentId, message, trackSubmission) => {
let didHandoff = false;
set((prev) => {
const session = prev.sessions[serverId];
@@ -1349,6 +1303,28 @@ export const useSessionStore = create<SessionStore>()(
const nextHead = result.changedHead
? new Map(session.agentStreamHead).set(agentId, result.head)
: session.agentStreamHead;
const clientMessageId = message.clientMessageId;
let messageSubmissions = session.messageSubmissions;
if (
trackSubmission &&
clientMessageId &&
[...result.tail, ...result.head].some(
(item) =>
item.kind === "user_message" &&
item.clientMessageId === clientMessageId &&
isUnreconciledLocalUserMessage(item),
)
) {
const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
const accepted = acceptMessageSubmission(
beginMessageSubmission(currentSubmissions, {
clientMessageId,
submittedAt: message.timestamp,
}),
clientMessageId,
);
messageSubmissions = new Map(session.messageSubmissions).set(agentId, accepted);
}
didHandoff = true;
return {
@@ -1359,7 +1335,7 @@ export const useSessionStore = create<SessionStore>()(
...session,
agentStreamTail: nextTail,
agentStreamHead: nextHead,
activeAgentTurns: updateActiveAgentTurn(session.activeAgentTurns, agentId, true),
messageSubmissions,
},
},
};
@@ -1613,12 +1589,14 @@ export const useSessionStore = create<SessionStore>()(
return prev;
}
const nextAgents = typeof agents === "function" ? agents(session.agents) : agents;
const messageSubmissions = applyRunningAgentsToAcceptedSubmissions({
previousAgents: session.agents,
nextAgents,
submissions: session.messageSubmissions,
});
if (session.agents === nextAgents && session.messageSubmissions === messageSubmissions) {
let activeAgentTurns = session.activeAgentTurns;
for (const [agentId, agent] of nextAgents) {
const previousAgent = session.agents.get(agentId);
if (previousAgent?.status !== "running" && agent.status === "running") {
activeAgentTurns = updateActiveAgentTurn(activeAgentTurns, agentId, true);
}
}
if (session.agents === nextAgents && activeAgentTurns === session.activeAgentTurns) {
return prev;
}
return {
@@ -1628,11 +1606,11 @@ export const useSessionStore = create<SessionStore>()(
[serverId]: {
...session,
agents: nextAgents,
messageSubmissions,
workspaceAgentActivity:
nextAgents === session.agents
? session.workspaceAgentActivity
: buildWorkspaceAgentActivityIndex(nextAgents, session.workspaceAgentActivity),
activeAgentTurns,
workspaceAgentActivity: buildWorkspaceAgentActivityIndex(
nextAgents,
session.workspaceAgentActivity,
),
},
},
};
packages/app/src/timeline/session-stream-reducers.test.ts
+16 -4
View File
@@ -149,10 +149,22 @@ describe("deriveAgentStreamTurnActivity", () => {
}
});
it("leaves activity unchanged when a batch has no lifecycle event", () => {
expect(
deriveAgentStreamTurnActivity([makeStreamReducerEvent(makeAssistantTimelineEvent("hi"), 1)]),
).toBeUndefined();
it("keeps an open turn without stream lifecycle and closes it on an ordered terminal event", () => {
let hasOpenTurn = true;
hasOpenTurn =
deriveAgentStreamTurnActivity([
makeStreamReducerEvent(makeAssistantTimelineEvent("still working"), 1),
]) ?? hasOpenTurn;
expect(hasOpenTurn).toBe(true);
hasOpenTurn =
deriveAgentStreamTurnActivity([
makeStreamReducerEvent(
{ type: "turn_completed", provider: "claude" } as AgentStreamEventPayload,
2,
),
]) ?? hasOpenTurn;
expect(hasOpenTurn).toBe(false);
});
});
packages/app/src/types/stream.ts
+4 -1
View File
@@ -161,7 +161,10 @@ export function removeSubmittedUserMessage(input: {
}): { tail: StreamItem[]; head: StreamItem[] } {
const remove = (items: StreamItem[]) => {
const next = items.filter(
(item) => item.kind !== "user_message" || item.clientMessageId !== input.clientMessageId,
(item) =>
item.kind !== "user_message" ||
item.clientMessageId !== input.clientMessageId ||
!isUnreconciledLocalUserMessage(item),
);
return next.length === items.length ? items : next;
};
packages/client/src/daemon-client.ts
+1 -7
View File
@@ -324,11 +324,6 @@ export interface SendMessageOptions {
attachments?: SendAgentMessageRequest["attachments"];
}
export interface SendMessageResult {
/** Undefined when connected to a daemon predating message submission disposition. */
outOfBand?: boolean;
}
export interface AgentAttentionRequiredNotification {
agentId: string;
reason: "finished" | "error" | "permission";
@@ -2864,7 +2859,7 @@ export class DaemonClient {
agentId: string,
text: string,
options?: SendMessageOptions,
): Promise<SendMessageResult> {
): Promise<void> {
const requestId = this.createRequestId();
const messageId = options?.messageId ?? crypto.randomUUID();
const message = SessionInboundMessageSchema.parse({
@@ -2893,7 +2888,6 @@ export class DaemonClient {
if (!payload.accepted) {
throw new Error(payload.error ?? "sendAgentMessage rejected");
}
return payload.outOfBand === undefined ? {} : { outOfBand: payload.outOfBand };
}
async sendMessage(agentId: string, text: string, options?: SendMessageOptions): Promise<void> {
packages/protocol/src/messages.ts
+2 -2
View File
@@ -2828,6 +2828,8 @@ export const ServerInfoStatusPayloadSchema = z
forgeProviders: z.boolean().optional(),
// COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12.
selectiveAgentTimeline: z.boolean().optional(),
// COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30.
canonicalSubmittedPrompts: z.boolean().optional(),
// COMPAT(stableProjectIdentity): added in v0.1.109, remove gate after 2027-01-15.
stableProjectIdentity: z.boolean().optional(),
// COMPAT(workspaceScriptManagement): added in v0.1.105, remove gate after 2027-01-10.
@@ -3710,8 +3712,6 @@ export const SendAgentMessageResponseMessageSchema = z.object({
agentId: z.string(),
accepted: z.boolean(),
error: z.string().nullable(),
// COMPAT(messageSubmissionDisposition): added in v0.2.3, remove optional parsing after 2027-01-27.
outOfBand: z.boolean().optional(),
}),
});
packages/server/src/server/agent/agent-manager.test.ts
+62 -1
View File
@@ -16,7 +16,7 @@ import {
import { AgentStorage } from "./agent-storage.js";
import { toAgentPayload } from "./agent-projections.js";
import { PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels";
import { formatSystemNotificationPrompt } from "./agent-prompt.js";
import { formatSystemNotificationPrompt, startAgentRun } from "./agent-prompt.js";
import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent-loading.js";
import type { StoredAgentRecord } from "./agent-storage.js";
import type {
@@ -8001,6 +8001,67 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
}
});
test("authoritative timeline records a daemon-handled submitted prompt before its output", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-daemon-handled-prompt-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
const commandCompleted = deferred<void>();
class DaemonHandledPromptSession extends TestAgentSession {
override tryHandleOutOfBand(prompt: AgentPromptInput) {
if (prompt !== "/handled") return null;
return {
run: async ({ emit }: { emit: (event: AgentStreamEvent) => void }) => {
emit({
type: "timeline",
provider: this.provider,
item: { type: "assistant_message", text: "Handled by the daemon" },
});
commandCompleted.resolve();
},
};
}
}
class DaemonHandledPromptClient extends TestAgentClient {
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new DaemonHandledPromptSession(config);
}
}
const manager = new AgentManager({
clients: { codex: new DaemonHandledPromptClient() },
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000403",
});
try {
const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
await startAgentRun(manager, snapshot.id, "/handled", logger, {
runOptions: { clientMessageId: "msg-client-daemon-handled" },
});
await commandCompleted.promise;
const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows;
expect(timeline.map((row) => row.item)).toEqual([
{
type: "user_message",
text: "/handled",
clientMessageId: "msg-client-daemon-handled",
},
{ type: "assistant_message", text: "Handled by the daemon" },
]);
} finally {
await manager.flush().catch(() => undefined);
await storage.flush().catch(() => undefined);
rmSync(workdir, { recursive: true, force: true });
}
});
test("replaceAgentRun succeeds when foreground turn terminal event is never delivered", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-stale-fg-"));
const storagePath = join(workdir, "agents");
packages/server/src/server/agent/agent-manager.ts
+16 -1
View File
@@ -1868,12 +1868,27 @@ export class AgentManager {
* emitted by the handler flow through dispatchStream so they persist and
* broadcast like normal timeline events.
*/
tryRunOutOfBand(agentId: string, prompt: AgentPromptInput): boolean {
tryRunOutOfBand(agentId: string, prompt: AgentPromptInput, options?: AgentRunOptions): boolean {
const agent = this.requireSessionAgent(agentId);
const handler = agent.session.tryHandleOutOfBand?.(prompt);
if (!handler) {
return false;
}
if (options?.clientMessageId) {
if (typeof prompt !== "string") {
throw new Error("Daemon-handled submitted prompts must be text");
}
this.touchUpdatedAt(agent);
this.recordAndDispatchTimelineItem(
agent.id,
{
type: "user_message",
text: prompt,
clientMessageId: options.clientMessageId,
},
agent.provider,
);
}
const dispatch = (event: AgentStreamEvent): void => {
// Persist timeline items so they show up in fetchAgentTimeline; broadcast
// for live subscribers. Other event types are broadcast only.
packages/server/src/server/agent/agent-prompt.ts
+1 -1
View File
@@ -41,7 +41,7 @@ export async function startAgentRun(
// Out-of-band commands (e.g. /goal pause) must run WITHOUT canceling an
// in-flight turn — replaceAgentRun would interrupt the running turn. The
// intercept lives at this layer so it covers every prompt entrypoint.
if (agentManager.tryRunOutOfBand(agentId, prompt)) {
if (agentManager.tryRunOutOfBand(agentId, prompt, options?.runOptions)) {
return { outOfBand: true };
}
const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId));
packages/server/src/server/agent/providers/acp-agent.test.ts
+9
View File
@@ -2345,6 +2345,15 @@ describe("ACPAgentSession", () => {
messageId: "msg-client-1",
prompt: [{ type: "text", text: "hello" }],
});
expect(
events
.filter(
(event) =>
event.type === "turn_started" ||
(event.type === "timeline" && event.item.type === "user_message"),
)
.map((event) => event.type),
).toEqual(["turn_started", "timeline"]);
expect(
events.filter((event) => event.type === "timeline" && event.item.type === "user_message"),
).toEqual([
packages/server/src/server/agent/providers/claude/agent.test.ts
+16
View File
@@ -1507,6 +1507,22 @@ describe("ClaudeAgentSession context window usage", () => {
};
}
test("emits turn_started before the submitted user message", async () => {
const session = await createSessionForTurns([[]]);
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.startTurn("turn", { clientMessageId: "client-message-1" });
await new Promise((resolve) => setTimeout(resolve, 0));
expect(events.slice(0, 2).map((event) => event.type)).toEqual(["turn_started", "timeline"]);
expect(events[1]).toMatchObject({
type: "timeline",
item: { type: "user_message", clientMessageId: "client-message-1" },
});
await session.close();
});
test("passes persistSession through to the Claude SDK query options", async () => {
const createResultTurn = (sessionId: string) => [
{
packages/server/src/server/agent/providers/codex-app-server-agent.test.ts
+7
View File
@@ -1119,7 +1119,13 @@ describe("Codex app-server provider", () => {
async () => appServer.child,
);
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.startTurn("remember this", { clientMessageId: "client-message" });
asInternals(session).handleNotification("turn/started", {
threadId: "thread-1",
turn: { id: "turn-1" },
});
const userMessage = waitForNextTimelineItem(session, "user_message");
emitCodexUserMessage(appServer, { id: "codex-message", text: "remember this" });
@@ -1130,6 +1136,7 @@ describe("Codex app-server provider", () => {
clientMessageId: "client-message",
},
});
expect(events.slice(0, 2).map((event) => event.type)).toEqual(["turn_started", "timeline"]);
appServer.completeTurn();
await session.close();
});
packages/server/src/server/agent/providers/mock-load-test-agent.test.ts
+26
View File
@@ -230,6 +230,32 @@ describe("MockLoadTestAgentClient", () => {
});
});
test("emits turn_started before the submitted user message", async () => {
vi.useFakeTimers();
const client = new MockLoadTestAgentClient();
const session = await client.createSession({
provider: "mock",
cwd: process.cwd(),
model: "ten-second-stream",
});
const events: AgentStreamEvent[] = [];
const unsubscribe = session.subscribe((event) => events.push(event));
await session.startTurn("Order the submitted prompt.", {
clientMessageId: "client-message-1",
});
await vi.advanceTimersByTimeAsync(0);
unsubscribe();
expect(
events
.slice(0, 2)
.map((event) =>
event.type === "timeline" ? `${event.type}:${event.item.type}` : event.type,
),
).toEqual(["turn_started", "timeline:user_message"]);
});
test("emits the free-write question scenario selected by prompt", async () => {
vi.useFakeTimers();
const client = new MockLoadTestAgentClient();
packages/server/src/server/agent/providers/mock-load-test-agent.ts
+36 -38
View File
@@ -33,6 +33,7 @@ import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest
export const MOCK_LOAD_TEST_PROVIDER_ID = "mock";
export const MOCK_LOAD_TEST_DEFAULT_MODEL_ID = "five-minute-stream";
export const MOCK_LOAD_TEST_HANDLED_COMMAND = "/mock handled-command";
const MOCK_LOAD_TEST_MODE_ID = "load-test";
const MOCK_LOAD_TEST_DURATION_MS = 5 * 60 * 1000;
const MOCK_LOAD_TEST_INTERVAL_MS = 40;
@@ -692,6 +693,7 @@ export class MockLoadTestAgentSession implements AgentSession {
if (this.activeTurn?.turnId !== turnId) {
return;
}
this.emitTurnStarted(turn);
this.emit({
type: "timeline",
provider: this.provider,
@@ -710,6 +712,21 @@ export class MockLoadTestAgentSession implements AgentSession {
return { turnId };
}
tryHandleOutOfBand(
prompt: AgentPromptInput,
): { run(ctx: { emit: (event: AgentStreamEvent) => void }): Promise<void> } | null {
if (prompt !== MOCK_LOAD_TEST_HANDLED_COMMAND) return null;
return {
run: async ({ emit }) => {
emit({
type: "timeline",
provider: this.provider,
item: { type: "assistant_message", text: "Mock command handled" },
});
},
};
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.listeners.add(callback);
return () => {
@@ -842,6 +859,18 @@ export class MockLoadTestAgentSession implements AgentSession {
turn.timer.unref?.();
}
private emitTurnStarted(turn: ActiveTurn): void {
if (turn.turnStarted) {
return;
}
turn.turnStarted = true;
this.emit({
type: "turn_started",
provider: this.provider,
turnId: turn.turnId,
});
}
private failConfiguredRewind(): void {
if (this.rewindError) {
throw new Error(this.rewindError);
@@ -864,11 +893,7 @@ export class MockLoadTestAgentSession implements AgentSession {
return;
}
this.clearTurnTimer(turn);
this.emit({
type: "turn_started",
provider: this.provider,
turnId: turn.turnId,
});
this.emitTurnStarted(turn);
this.activeTurn = null;
this.emit({
type: "turn_failed",
@@ -923,11 +948,7 @@ export class MockLoadTestAgentSession implements AgentSession {
}
this.clearTurnTimer(turn);
this.emit({
type: "turn_started",
provider: this.provider,
turnId: turn.turnId,
});
this.emitTurnStarted(turn);
this.emitTimeline(turn.turnId, {
type: "assistant_message",
@@ -960,11 +981,7 @@ export class MockLoadTestAgentSession implements AgentSession {
}
this.clearTurnTimer(turn);
this.emit({
type: "turn_started",
provider: this.provider,
turnId: turn.turnId,
});
this.emitTurnStarted(turn);
const request: AgentPermissionRequest = {
id: `mock-plan-${turn.turnId}`,
@@ -1015,11 +1032,7 @@ export class MockLoadTestAgentSession implements AgentSession {
}
this.clearTurnTimer(turn);
this.emit({
type: "turn_started",
provider: this.provider,
turnId: turn.turnId,
});
this.emitTurnStarted(turn);
const request: AgentPermissionRequest = {
id: `mock-questions-${turn.turnId}`,
@@ -1050,11 +1063,7 @@ export class MockLoadTestAgentSession implements AgentSession {
}
this.clearTurnTimer(turn);
this.emit({
type: "turn_started",
provider: this.provider,
turnId: turn.turnId,
});
this.emitTurnStarted(turn);
for (let index = 0; index < stress.count; index += 1) {
this.emitTimeline(
@@ -1103,11 +1112,7 @@ export class MockLoadTestAgentSession implements AgentSession {
}
this.clearTurnTimer(turn);
this.emit({
type: "turn_started",
provider: this.provider,
turnId: turn.turnId,
});
this.emitTurnStarted(turn);
const payload = buildRepeatedPayload(largePayload.bytes, largePayload.kind);
if (largePayload.kind === "diff") {
@@ -1179,14 +1184,7 @@ export class MockLoadTestAgentSession implements AgentSession {
}
this.clearTurnTimer(turn);
if (!turn.turnStarted) {
turn.turnStarted = true;
this.emit({
type: "turn_started",
provider: this.provider,
turnId: turn.turnId,
});
}
this.emitTurnStarted(turn);
const elapsedMs = Date.now() - turn.startedAt;
if (elapsedMs >= turn.durationMs) {
packages/server/src/server/agent/providers/omp/agent.test.ts
+1
View File
@@ -147,6 +147,7 @@ describe("OMP agent client and session", () => {
{ type: "user_message", text: "hello OMP", messageId: "user-1" },
{ type: "assistant_message", text: "hello from OMP", messageId: "omp-assistant-1" },
]);
expect(omp.eventTypes().slice(0, 2)).toEqual(["turn_started", "timeline"]);
expect(omp.completedTurnCount()).toBe(1);
});
packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts
+4
View File
@@ -391,6 +391,10 @@ export class OmpHarness {
return this.events.flatMap((event) => (event.type === "timeline" ? [event.item] : []));
}
eventTypes(): AgentStreamEvent["type"][] {
return this.events.map((event) => event.type);
}
async history(): Promise<AgentTimelineItem[]> {
const items: AgentTimelineItem[] = [];
for await (const event of this.requireSession().streamHistory()) {
packages/server/src/server/agent/providers/opencode-agent.test.ts
+22 -1
View File
@@ -1168,6 +1168,19 @@ describe("OpenCode adapter startTurn error handling", () => {
properties: {},
},
},
{
directory: "/tmp/test",
payload: {
type: "message.updated",
properties: {
info: {
id: "msg_user",
sessionID: "ses_unit_test",
role: "user",
},
},
},
},
{
directory: "/tmp/test",
payload: {
@@ -1229,17 +1242,25 @@ describe("OpenCode adapter startTurn error handling", () => {
createTestLogger(),
);
const turn = await collectTurnEvents(streamSession(session, "hello"));
const turn = await collectTurnEvents(
streamSession(session, "hello", { clientMessageId: "client-message-1" }),
);
expect(turn.events.map((event) => event.type)).toEqual([
"turn_started",
"timeline",
"timeline",
"turn_completed",
]);
expect(turn.events[1]).toMatchObject({
type: "timeline",
item: { type: "user_message", clientMessageId: "client-message-1" },
});
expect(turn.events.map((event) => ("turnId" in event ? event.turnId : undefined))).toEqual([
"opencode-turn-0",
"opencode-turn-0",
"opencode-turn-0",
"opencode-turn-0",
]);
});
packages/server/src/server/agent/providers/pi/agent.test.ts
+6
View File
@@ -181,6 +181,10 @@ class SessionEvents {
});
}
eventTypes(): AgentStreamEvent["type"][] {
return this.events.map((event) => event.type);
}
turnCompletedEvents() {
return this.events.filter(
(event): event is Extract<AgentStreamEvent, { type: "turn_completed" }> =>
@@ -608,6 +612,7 @@ describe("PiRpcAgentSession", () => {
const fakeSession = pi.latestSession();
await session.startTurn("hello");
fakeSession.emit({ type: "turn_start" });
fakeSession.finishSubmittedUserMessage({
id: "entry-user-1",
parentId: null,
@@ -619,6 +624,7 @@ describe("PiRpcAgentSession", () => {
expect(events.timelineItems()).toEqual([
{ type: "user_message", text: "hello", messageId: "entry-user-1" },
]);
expect(events.eventTypes().slice(0, 2)).toEqual(["turn_started", "timeline"]);
});
test("uses the Pi entry attached to a submitted prompt after resuming old history", async () => {
packages/server/src/server/session.ts
-2
View File
@@ -6345,7 +6345,6 @@ export class Session {
agentId,
accepted: true,
error: null,
outOfBand: true,
},
});
return;
@@ -6373,7 +6372,6 @@ export class Session {
agentId,
accepted: true,
error: null,
outOfBand: false,
},
});
} catch (error) {
packages/server/src/server/websocket-server.ts
+2
View File
@@ -1574,6 +1574,8 @@ export class VoiceAssistantWebSocketServer {
forgeProviders: true,
// COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12.
selectiveAgentTimeline: true,
// COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30.
canonicalSubmittedPrompts: true,
// COMPAT(stableProjectIdentity): added in v0.1.109, remove gate after 2027-01-15.
stableProjectIdentity: true,
// COMPAT(workspaceScriptManagement): added in v0.1.105, remove gate after 2027-01-10.