puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-06 16:34:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(app): close canonical submission races

Browse Source
This commit is contained in:
Mohamed Boudra
2026-07-27 15:40:23 +02:00
parent 3063f2868b
commit 3d4e6bb71e
9 changed files with 295 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/composer/submission/model.test.ts
+39
View File
@@ -5,6 +5,7 @@ import {
beginMessageSubmission,
getPendingMessageSubmission,
observeMessageSubmissionRunning,
observeMessageSubmissionCanonical,
rejectMessageSubmission,
type MessageSubmissionState,
} from "./model";
@@ -40,6 +41,44 @@ describe("message submission model", () => {
expect(runningObserved).toEqual([]);
});
it("closes RPC-accepted pending state when canonical history proves acceptance", () => {
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: new Date("2026-07-26T10:00:00.000Z"),
});
const begun = beginMessageSubmission(initialState(), {
message,
submittedAt: message.timestamp,
acceptance: "rpc-and-running",
});
const accepted = acceptMessageSubmission(begun, "client-1");
expect(observeMessageSubmissionCanonical(accepted.submissions, ["client-1"])).toEqual([]);
});
it("remembers canonical acceptance until an in-flight RPC rejection settles", () => {
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: new Date("2026-07-26T10:00:00.000Z"),
});
const begun = beginMessageSubmission(initialState(), {
message,
submittedAt: message.timestamp,
acceptance: "rpc-and-running",
});
const canonicalObserved = observeMessageSubmissionCanonical(begun.submissions, ["client-1"]);
expect(getPendingMessageSubmission(canonicalObserved)).toBeNull();
expect(
rejectMessageSubmission({ ...begun, submissions: canonicalObserved }, "client-1"),
).toEqual({
outcome: "accepted",
state: { ...begun, submissions: [] },
});
});
it("rejects a row when an unrelated running transition precedes its RPC error", () => {
const message = createUserMessage({
clientMessageId: "client-1",
packages/app/src/composer/submission/model.ts
+36 -2
View File
@@ -11,12 +11,13 @@ export type MessageSubmissionRecord = PendingMessageSubmission &
(
| { phase: "waiting-for-rpc"; acceptance: MessageSubmissionAcceptance }
| { phase: "waiting-for-running" }
| { phase: "canonical-observed" }
);
export function getPendingMessageSubmission(
submissions: readonly MessageSubmissionRecord[] | null | undefined,
): PendingMessageSubmission | null {
return submissions?.at(-1) ?? null;
return submissions?.findLast((submission) => submission.phase !== "canonical-observed") ?? null;
}
export interface MessageSubmissionState {
@@ -75,6 +76,11 @@ export function acceptMessageSubmission(
);
if (index === -1) return state;
const submission = state.submissions[index];
if (submission.phase === "canonical-observed") {
const submissions = state.submissions.slice();
submissions.splice(index, 1);
return { ...state, submissions };
}
if (submission.phase === "waiting-for-running") return state;
const nextSubmission: MessageSubmissionRecord | null =
submission.acceptance === "rpc-only" || isAgentRunning
@@ -97,6 +103,34 @@ export function observeMessageSubmissionRunning(
return next.length === submissions.length ? (submissions as MessageSubmissionRecord[]) : next;
}
export function observeMessageSubmissionCanonical(
submissions: readonly MessageSubmissionRecord[],
clientMessageIds: readonly string[],
): MessageSubmissionRecord[] {
if (clientMessageIds.length === 0) return submissions as MessageSubmissionRecord[];
const observed = new Set(clientMessageIds);
let changed = false;
const next = submissions.flatMap((submission): MessageSubmissionRecord[] => {
if (!observed.has(submission.clientMessageId)) return [submission];
if (submission.phase === "waiting-for-running") {
changed = true;
return [];
}
if (submission.phase === "waiting-for-rpc") {
changed = true;
return [
{
clientMessageId: submission.clientMessageId,
submittedAt: submission.submittedAt,
phase: "canonical-observed",
},
];
}
return [submission];
});
return changed ? next : (submissions as MessageSubmissionRecord[]);
}
function removeSubmittedMessage(items: StreamItem[], clientMessageId: string): StreamItem[] {
const next = items.filter(
(item) => item.kind !== "user_message" || item.clientMessageId !== clientMessageId,
@@ -115,7 +149,7 @@ export function rejectMessageSubmission(
const submission = state.submissions[index];
const submissions = state.submissions.slice();
submissions.splice(index, 1);
if (submission.phase === "waiting-for-running") {
if (submission.phase === "waiting-for-running" || submission.phase === "canonical-observed") {
return { outcome: "accepted", state: { ...state, submissions } };
}
return {
packages/app/src/contexts/session-context.tsx
+9 -32
View File
@@ -192,9 +192,7 @@ type WorkspaceSetupProgressPayload = Extract<
type SessionStoreActions = ReturnType<typeof useSessionStore.getState>;
type SetInitializingAgents = SessionStoreActions["setInitializingAgents"];
type SetAgentStreamTail = SessionStoreActions["setAgentStreamTail"];
type SetAgentStreamHead = SessionStoreActions["setAgentStreamHead"];
type ClearAgentStreamHead = SessionStoreActions["clearAgentStreamHead"];
type SetAgentStreamState = SessionStoreActions["setAgentStreamState"];
type SetAgentTimelineCursor = SessionStoreActions["setAgentTimelineCursor"];
type MarkAgentHistorySynchronized = SessionStoreActions["markAgentHistorySynchronized"];
type SetAgentAuthoritativeHistoryApplied =
@@ -237,9 +235,7 @@ function applyTimelineStreamPatches(input: {
serverId: string;
currentTail: StreamItem[];
currentHead: StreamItem[];
setAgentStreamTail: SetAgentStreamTail;
setAgentStreamHead: SetAgentStreamHead;
clearAgentStreamHead: ClearAgentStreamHead;
setAgentStreamState: SetAgentStreamState;
setAgentTimelineCursor: SetAgentTimelineCursor;
}): void {
const {
@@ -248,32 +244,17 @@ function applyTimelineStreamPatches(input: {
serverId,
currentTail,
currentHead,
setAgentStreamTail,
setAgentStreamHead,
clearAgentStreamHead,
setAgentStreamState,
setAgentTimelineCursor,
} = input;
if (result.tail !== currentTail) {
setAgentStreamTail(serverId, (prev) => {
const next = new Map(prev);
next.set(agentId, result.tail);
return next;
if (result.tail !== currentTail || result.head !== currentHead) {
setAgentStreamState(serverId, agentId, {
...(result.tail !== currentTail ? { tail: result.tail } : {}),
...(result.head !== currentHead ? { head: result.head } : {}),
});
}
if (result.head !== currentHead) {
if (result.head.length === 0) {
clearAgentStreamHead(serverId, agentId);
} else {
setAgentStreamHead(serverId, (prev) => {
const next = new Map(prev);
next.set(agentId, result.head);
return next;
});
}
}
if (result.cursorChanged) {
setAgentTimelineCursor(serverId, (prev) => {
const current = prev.get(agentId);
@@ -701,9 +682,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
serverId,
currentTail,
currentHead,
setAgentStreamTail,
setAgentStreamHead,
clearAgentStreamHead,
setAgentStreamState,
setAgentTimelineCursor,
});
@@ -725,13 +704,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
});
},
[
clearAgentStreamHead,
markAgentHistorySynchronized,
recoverTimelineGap,
serverId,
setAgentAuthoritativeHistoryApplied,
setAgentStreamHead,
setAgentStreamTail,
setAgentStreamState,
setAgentTimelineCursor,
setAgentTimelineHasOlder,
setInitializingAgents,
packages/app/src/stores/session-store.ts
+28 -1
View File
@@ -13,6 +13,7 @@ import {
acceptMessageSubmission,
beginMessageSubmission,
observeMessageSubmissionRunning,
observeMessageSubmissionCanonical,
rejectMessageSubmission,
type MessageSubmissionRecord,
type MessageSubmissionRejectionOutcome,
@@ -1105,10 +1106,35 @@ export const useSessionStore = create<SessionStore>()(
}
}
if (!changedTail && !changedHead) {
const canonicalClientMessageIds = [
...(state.tail ?? session.agentStreamTail.get(agentId) ?? []),
...(state.head ?? session.agentStreamHead.get(agentId) ?? []),
].flatMap((item) =>
item.kind === "user_message" && item.clientMessageId && item.messageId
? [item.clientMessageId]
: [],
);
const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
const observedSubmissions = observeMessageSubmissionCanonical(
currentSubmissions,
canonicalClientMessageIds,
);
const changedSubmissions = observedSubmissions !== currentSubmissions;
if (!changedTail && !changedHead && !changedSubmissions) {
return prev;
}
let messageSubmissions = session.messageSubmissions;
if (changedSubmissions) {
messageSubmissions = new Map(session.messageSubmissions);
if (observedSubmissions.length > 0) {
messageSubmissions.set(agentId, observedSubmissions);
} else {
messageSubmissions.delete(agentId);
}
}
return {
...prev,
sessions: {
@@ -1117,6 +1143,7 @@ export const useSessionStore = create<SessionStore>()(
...session,
agentStreamTail: nextTail,
agentStreamHead: nextHead,
messageSubmissions,
},
},
};
packages/app/src/timeline/session-stream-reducers.test.ts
+33
View File
@@ -330,6 +330,39 @@ describe("processTimelineResponse", () => {
expect(result.head).toEqual([{ ...liveAssistant, text: "Live" }, submitted]);
});
it("preserves newer live head items when canonical replacement ends in a tool call", () => {
const liveThought: StreamItem = {
kind: "thought",
id: "live-thought",
text: "newer reasoning",
timestamp: new Date(3000),
status: "loading",
};
const liveAssistant = makeAssistantItem("newer answer", "live-answer");
const result = processTimelineResponse({
...baseTimelineInput,
currentHead: [liveThought, liveAssistant],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
payload: {
...baseTimelineInput.payload,
reset: true,
epoch: "epoch-1",
startCursor: { seq: 1 },
endCursor: { seq: 1 },
entries: [
makeToolCallTimelineEntry(1, "canonical-call", "completed", {
type: "read",
filePath: "/tmp/older.ts",
}),
],
},
});
expect(result.tail.map((item) => item.kind)).toEqual(["tool_call"]);
expect(result.head).toEqual([liveThought, liveAssistant]);
});
it("uses the timeline entry timestamp as canonical", () => {
const result = processTimelineResponse({
...baseTimelineInput,
packages/app/src/timeline/session-stream-reducers.ts
+4
View File
@@ -233,6 +233,7 @@ function applyTimelineReplacePath(args: {
currentHead: StreamItem[];
pendingClientMessageId: string | null;
replacementPolicy: "preserve-local-submissions" | "replace-history";
preserveLiveHead: boolean;
toHydratedEvents: (
units: TimelineUnit[],
) => Array<{ event: AgentStreamEventPayload; timestamp: Date }>;
@@ -245,6 +246,7 @@ function applyTimelineReplacePath(args: {
currentHead,
pendingClientMessageId,
replacementPolicy,
preserveLiveHead,
toHydratedEvents,
} = args;
const hydratedTail = hydrateStreamState(toHydratedEvents(timelineUnits), { source: "canonical" });
@@ -254,6 +256,7 @@ function applyTimelineReplacePath(args: {
previousHead: currentHead,
pendingClientMessageId,
policy: replacementPolicy,
preserveLiveHead,
});
const cursor: TimelineCursor | null =
payload.startCursor && payload.endCursor
@@ -867,6 +870,7 @@ export function processTimelineResponse(
currentHead,
pendingClientMessageId,
replacementPolicy,
preserveLiveHead: currentCursor?.epoch === payload.epoch,
toHydratedEvents,
})
: applyTimelineIncrementalPath({
packages/app/src/types/stream.test.ts
+50 -3
View File
@@ -1486,7 +1486,7 @@ describe("turn lifecycle events", () => {
);
});
it("reconciles a submitted user message that was pending in the streaming head", () => {
it("flushes an interrupted head when its submitted prompt becomes canonical", () => {
const submitted: StreamItem = {
kind: "user_message",
id: "msg_head_submitted",
@@ -1511,13 +1511,60 @@ describe("turn lifecycle events", () => {
source: "live",
});
assert.strictEqual(result.tail.length, 0);
const userMessages = result.head.filter((item) => item.kind === "user_message");
assert.deepStrictEqual(result.head, []);
const userMessages = result.tail.filter((item) => item.kind === "user_message");
assert.strictEqual(userMessages.length, 1);
assert.strictEqual(userMessages[0]?.id, "msg_head_submitted");
assert.strictEqual(userMessages[0]?.messageId, "provider-owned-head");
});
it("keeps a replacement assistant separate after an interrupted prompt is reconciled", () => {
const interruptedAssistant: StreamItem = {
kind: "assistant_message",
id: "interrupted",
text: "old answer",
timestamp: new Date("2025-01-01T15:03:01Z"),
};
const submitted = createUserMessage({
clientMessageId: "msg_interrupt",
text: "replacement prompt",
timestamp: new Date("2025-01-01T15:03:02Z"),
});
const reconciled = applyStreamEvent({
tail: [],
head: [interruptedAssistant, submitted],
event: {
type: "timeline",
provider: "opencode",
item: {
type: "user_message",
text: submitted.text,
messageId: "provider-prompt",
clientMessageId: submitted.clientMessageId,
},
},
timestamp: new Date("2025-01-01T15:03:03Z"),
});
const replacement = applyStreamEvent({
tail: reconciled.tail,
head: reconciled.head,
event: {
type: "timeline",
provider: "opencode",
item: { type: "assistant_message", text: "new answer" },
},
timestamp: new Date("2025-01-01T15:03:04Z"),
});
expect(replacement.tail.map((item) => item.kind)).toEqual([
"assistant_message",
"user_message",
]);
expect(replacement.head).toEqual([
expect.objectContaining({ kind: "assistant_message", text: "new answer" }),
]);
});
it("replaces multiple submitted user messages in FIFO order", () => {
const submittedTimestamp = new Date("2025-01-01T15:04:00Z");
const serverTimestamp = new Date("2025-01-01T15:04:01Z");
packages/app/src/types/stream.ts
+63 -45
View File
@@ -320,6 +320,7 @@ export interface CanonicalStreamReplacementInput {
previousHead: StreamItem[];
pendingClientMessageId: string | null;
policy: "preserve-local-submissions" | "replace-history";
preserveLiveHead: boolean;
}
export interface CanonicalStreamReplacementResult {
@@ -334,15 +335,23 @@ function removeUserMessageAt(items: UserMessageItem[], index: number): UserMessa
function preserveReplacementHead(
tail: StreamItem[],
currentHead: StreamItem[],
preserveLiveHead: boolean,
): CanonicalStreamReplacementResult {
const liveAssistantIndex = currentHead.findLastIndex((item) => item.kind === "assistant_message");
const submittedRows = currentHead.filter(
(item): item is UserMessageItem =>
item.kind === "user_message" && item.clientMessageId !== undefined,
const retainedHead = preserveLiveHead
? currentHead
: currentHead.filter(
(item) => item.kind === "user_message" && item.clientMessageId !== undefined,
);
const tailIds = new Set(tail.map((item) => item.id));
const unreconciledHead = retainedHead.filter(
(item) => item.kind === "assistant_message" || !tailIds.has(item.id),
);
if (liveAssistantIndex < 0) return { tail, head: submittedRows };
const liveAssistantIndex = unreconciledHead.findLastIndex(
(item) => item.kind === "assistant_message",
);
if (liveAssistantIndex < 0) return { tail, head: unreconciledHead };
const liveAssistant = currentHead[liveAssistantIndex];
const liveAssistant = unreconciledHead[liveAssistantIndex];
const tailAssistant = tail.at(-1);
if (
liveAssistant.kind !== "assistant_message" ||
@@ -350,19 +359,14 @@ function preserveReplacementHead(
tailAssistant.kind !== "assistant_message" ||
!liveAssistant.text.startsWith(tailAssistant.text)
) {
return { tail, head: submittedRows };
return { tail, head: unreconciledHead };
}
const head: StreamItem[] = [];
currentHead.forEach((item, index) => {
if (index === liveAssistantIndex) {
head.push({ ...liveAssistant, text: tailAssistant.text });
return;
}
if (item.kind === "user_message" && item.clientMessageId) {
head.push(item);
}
});
const head = [
...unreconciledHead.slice(0, liveAssistantIndex),
{ ...liveAssistant, text: tailAssistant.text },
...unreconciledHead.slice(liveAssistantIndex + 1),
];
return { tail: tail.slice(0, -1), head };
}
@@ -423,7 +427,7 @@ export function replaceWithCanonicalStream(
);
});
return preserveReplacementHead(nextTail, nextHead);
return preserveReplacementHead(nextTail, nextHead, input.preserveLiveHead);
}
export interface AssistantMessageItem {
@@ -1407,6 +1411,45 @@ export interface ApplyStreamEventResult {
changedHead: boolean;
}
function applyCanonicalUserMessageEvent(params: {
tail: StreamItem[];
head: StreamItem[];
event: AgentStreamEventPayload;
timestamp: Date;
}): ApplyStreamEventResult | null {
const { tail, head, event, timestamp } = params;
if (event.type !== "timeline" || event.item.type !== "user_message") return null;
const normalized = normalizeChunk(event.item.text);
if (!normalized.hasContent) {
return { tail, head, changedTail: false, changedHead: false };
}
const flushedTail = head.length > 0 ? flushHeadToTail(tail, head) : tail;
const flushedHead = head.length > 0 ? [] : head;
const canonical = createUserMessage({
id:
event.item.messageId ??
createUniqueTimelineId([...tail, ...head], "user", normalized.chunk.trim(), timestamp),
messageId: event.item.messageId,
clientMessageId: event.item.clientMessageId,
text: normalized.chunk,
timestamp,
});
const reconciled = upsertUserMessageAcrossStream({
tail: flushedTail,
head: flushedHead,
message: canonical,
insert: "tail",
presentation: "existing",
});
return {
tail: reconciled.tail,
head: reconciled.head,
changedTail: flushedTail !== tail || reconciled.changedTail,
changedHead: flushedHead !== head || reconciled.changedHead,
};
}
/**
* Apply a stream event using head/tail model.
*
@@ -1427,38 +1470,13 @@ export function applyStreamEvent(params: {
timelineCursor?: TimelinePosition;
}): ApplyStreamEventResult {
const { tail, head, event, timestamp } = params;
const canonicalUserResult = applyCanonicalUserMessageEvent({ tail, head, event, timestamp });
if (canonicalUserResult) return canonicalUserResult;
const source = params.source ?? "live";
let nextTail = tail;
let nextHead = head;
let changedTail = false;
let changedHead = false;
if (event.type === "timeline" && event.item.type === "user_message") {
const normalized = normalizeChunk(event.item.text);
if (!normalized.hasContent) {
return { tail, head, changedTail: false, changedHead: false };
}
const canonical = createUserMessage({
id:
event.item.messageId ??
createUniqueTimelineId([...tail, ...head], "user", normalized.chunk.trim(), timestamp),
messageId: event.item.messageId,
clientMessageId: event.item.clientMessageId,
text: normalized.chunk,
timestamp,
});
const reconciled = upsertUserMessageAcrossStream({
tail,
head,
message: canonical,
insert: "none",
presentation: "existing",
});
if (reconciled.location) {
return reconciled;
}
}
const flushHead = () => {
if (nextHead.length === 0) {
return;
packages/app/src/utils/agent-directory-sync.test.ts
+33
View File
@@ -134,6 +134,39 @@ describe("message submission lifecycle ordering", () => {
).toBeUndefined();
store.clearSession(serverId);
});
it("clears accepted pending state when canonical history arrives after a missed run", () => {
const serverId = "server-canonical-after-missed-run";
const agentId = "agent-1";
const store = useSessionStore.getState();
store.initializeSession(serverId, null as unknown as DaemonClient);
applyAgentStatus({
serverId,
agentId,
status: "idle",
updatedAt: "2026-07-27T10:00:00.000Z",
});
const clientMessageId = beginPendingSubmission(serverId, agentId);
store.acceptAgentMessageSubmission(serverId, agentId, clientMessageId);
store.setAgentStreamState(serverId, agentId, {
tail: [
createUserMessage({
id: "provider-message",
messageId: "provider-message",
clientMessageId,
text: "Run this",
timestamp: new Date("2026-07-27T10:00:01.000Z"),
}),
],
head: [],
});
expect(
useSessionStore.getState().sessions[serverId]?.messageSubmissions.get(agentId),
).toBeUndefined();
store.clearSession(serverId);
});
});
describe("replaceFetchedAgentDirectory", () => {