jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Keep agent timelines stable from submission through resume (#2718)

Browse Source 
* Reapply: Stop completed turns from appearing stuck (#2484)

* Reapply: Keep older chat history and image previews stable (#2490)

* Lock timeline layout stability with user-driven specs

Two reds reproduce the reported jank:
- new agent, first prompt: the row paints at top=100 then jumps to 144 through
  authoritative hydration, because the draft tab and the agent panel each render
  their own AgentStreamView for the same logical row.
- long loaded timeline, submit: scrollHeight collapses 10793 -> 9111 mid-turn,
  because splitOrderedTail reclassifies measured rows into the virtualized segment
  where the virtualizer falls back to estimateStreamItemHeight.

Also restores centered timeline rows: the pagination anchor wrapper introduced in
#2490 was a plain block div, which broke alignSelf: center.

* Measure the turn footer blink from when the footer appears

Bounding the window at the prompt row conflated two things: the footer arriving a
frame after the row, and the footer leaving the layout and coming back. Bounded from
the footer's first frame, the blink still reproduces, so the footer-ownership race is
real rather than a sampling artifact.

* Rename the shadowed prompt row binding in the frame recorder

* fix(desktop): allow concurrent worktree dev instances

Desktop terminals can inherit another instance's profile path and debugging flags. Scope the profile to the current dev root and select an available debugging port so instances do not share global runtime state.

* fix(desktop): expose development providers in dev builds

Desktop-managed daemons inherited no Paseo runtime mode, so dev builds constructed the production provider registry. Mark unpackaged internal launches as development while continuing to force packaged launches to production.

* fix(timeline): stabilize optimistic turn layout

Keep initial history loading dormant until user intent and preserve active-turn layout across delayed transport updates. Add painted-frame coverage for optimistic timeline transitions.

* fix(timeline): preserve optimistic prompts through hydration

Canonical replacement relied on a transport submission record that can settle before a provider echoes the user row. Keep presentation continuity owned by the unreconciled row while destructive history replacement remains governed by active transactions.

* fix(timeline): make submitted prompts canonical

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.

* fix(timeline): stop older history pages repeating entries

Backward pagination counted source sequence rows, so page boundaries inside a merged projected item could return that item repeatedly. Count projected entries and advance from their display anchors instead, with server and browser regressions covering the contract.

* fix(timeline): resume agents from the latest history

Use a bounded tail probe on resume so long background runs appear atomically instead of replaying every missed page. Preserve unchanged views, reconcile local/live rows, and keep skipped history reachable through backward pagination.

* fix(timeline): reconcile turn state after missed updates

Turn completion events can be missed while an agent timeline is unsubscribed, leaving footer activity stale even after the directory reports idle. Reconcile the single-owned turn lifecycle from authoritative resume snapshots while preserving newer live evidence.

* test(timeline): cover running state across refresh

Protect the hydration boundary where directory state must restore consistent tab, composer, and footer activity. Expose the running indicator semantically so the regression checks user-visible state instead of spinner implementation details.

* fix(timeline): keep history loading geometry stable

Reserve a fixed history-start gutter regardless of pagination state so showing or hiding the loader cannot shift timeline content.

* fix(timeline): keep turn activity coherent

Derive footer, tab, and composer activity from one lifecycle and submission projection. Idle directory transitions now request an authoritative resume so missed terminal events converge without treating directory delivery order as turn authority.

* fix(tests): align checks with runtime boundaries

* fix(desktop): stop dev app surviving its terminal

Electron was detached from the desktop runner, allowing it to outlive a stopped workspace terminal with broken output pipes and retain the single-instance lock. Keep Electron owned by the runner and preserve explicit process-group cleanup for detached child trees.

* fix(timeline): preserve activity through turn handoff

Correlate lifecycle events and snapshots with daemon-owned turn identity, and keep optimistic submissions active until their canonical row arrives. This prevents stale updates and RPC timing from exposing a false idle state.

* refactor(timeline): remove submission timing state

* fix(timeline): restore prompt identity on resume

* fix(timeline): exclude attachments from canonical prompt text

Structured provider prompts can include hidden history attachments before the user's text. Persist only plain prompt blocks so authoritative user messages match what the composer displayed while provider input remains unchanged.

* refactor(agent): remove unused pending-run message identity

* fix(desktop): let Chromium allocate the debugging port

* refactor(agent): remove impossible finalized-turn tracking

* fix(timeline): load history after keyboard scrolling

Treat observed upward displacement as user intent so keyboard and scrollbar navigation can wake dormant pagination.

* test(server): preserve accepted turn stream order

The manager-owned accepted turn boundary intentionally precedes coalesced output. Assert that lifecycle event explicitly while retaining the timeline-before-terminal ordering contract.

* test(app): scope sidebar fixtures by identity

Shared E2E daemon state may contain unrelated projects. Keep stale-route coverage tied to its named missing project and drag only the two rows seeded by the reorder spec.

* fix(timeline): gate prompt revisions by client capability

* fix(timeline): finalize canonical prompt delivery

Treat provider identity as a single stored-row transition so legacy clients receive one deliverable canonical prompt. Release identity-pending rows before later foreground output or the current terminal while preserving stale-turn isolation.

* fix(server): preserve legacy prompt pagination

* fix(timeline): keep assistant block identity stable

Assign canonical block IDs during live promotion so terminal finalization cannot remount image-bearing rows.

* fix(timeline): apply covered revisions during resume append

Run the exact canonical revision pass before overlapping resume pages append newer rows, while leaving cursor advancement to the incremental path.

* fix(server): preserve queued agent transitions

Project each queued live snapshot before awaiting stored metadata so later lifecycle state cannot overwrite an earlier transition.

* fix(server): snapshot queued agent updates

Capture each live agent payload before it enters the per-agent queue so shared nested state cannot change while an earlier storage read is pending.

* docs(app): date agent turn compatibility cleanup

* refactor(timeline): mint submitted prompt identity at acceptance

Canonical prompt rows now keep one immutable wire identity while provider identity remains internal for rewind translation. This removes the unreleased same-sequence revision and legacy projection contract.

* refactor(app): remove prompt revision reconciliation

Submitted prompt rows are immutable on the wire, so resume and live reducers no longer need to apply already-covered same-sequence updates.

* test(app): cover submitted prompt rewind roundtrip

* docs(timeline): describe immutable submitted prompt identity

Provider echoes now enrich daemon-internal identity only; wire rows keep their acceptance-time identity and rewind performs the provider translation.

* test(server): validate mock rewind targets

Reject unknown provider message IDs so app rewind coverage exercises wire-to-provider identity translation.

* fix(server): validate rewind before interrupting

* fix(server): omit rewind identity for daemon prompts

* test(app): guard rewind eligibility and rejection

* fix(server): preserve prompt identity across hydration

* refactor(timeline): remove stale revision remnants

* Revert "fix(server): preserve prompt identity across hydration"

This reverts commit 6adcf7e5b0bc8601761af59b461a746cb5e395fa.

* fix(timeline): settle missed submission acknowledgements
This commit is contained in:
Mohamed Boudra
2026-08-01 14:13:47 +02:00
committed by GitHub
parent 8b87299ce1
commit 8021d2b6c9
161 changed files with 13757 additions and 2523 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/composer/actions.test.ts
+136 -41
View File
@@ -6,7 +6,17 @@ import type {
UserComposerAttachment,
WorkspaceComposerAttachment,
} from "@/attachments/types";
import type { StreamItem } from "@/types/stream";
import {
appendSubmittedUserMessage,
removeSubmittedUserMessage,
type StreamItem,
} from "@/types/stream";
import {
acceptMessageSubmission,
beginMessageSubmission,
rejectMessageSubmission,
type MessageSubmissionRecord,
} from "@/composer/submission/model";
import {
cancelComposerAgent,
dispatchComposerAgentMessage,
@@ -20,7 +30,7 @@ import {
sendQueuedComposerMessageNow,
toggleGithubAttachment,
toggleGithubAttachmentFromPicker,
type AgentStreamWriter,
type MessageSubmissionWriter,
type AttachmentPersister,
type ComposerCancelClient,
type ComposerSendClient,
@@ -168,14 +178,16 @@ interface FakeSendCall {
}
function createFakeSendClient(
options: { rejection?: Error } = {},
options: { rejection?: Error; beforeRejection?: (call: FakeSendCall) => void } = {},
): ComposerSendClient & { calls: FakeSendCall[] } {
const calls: FakeSendCall[] = [];
return {
calls,
sendAgentMessage: async (agentId, text, opts) => {
calls.push({ agentId, text, options: opts });
const call = { agentId, text, options: opts };
calls.push(call);
if (options.rejection) {
options.beforeRejection?.(call);
throw options.rejection;
}
},
@@ -183,7 +195,7 @@ function createFakeSendClient(
};
}
interface FakeStream extends AgentStreamWriter {
interface FakeStream extends MessageSubmissionWriter {
head: Map<string, StreamItem[]>;
tail: Map<string, StreamItem[]>;
}
@@ -192,18 +204,69 @@ function createFakeStream(initialHead: Map<string, StreamItem[]> = new Map()): F
const fake: FakeStream = {
head: new Map(initialHead),
tail: new Map(),
getTail: (agentId) => fake.tail.get(agentId),
getHead: (agentId) => fake.head.get(agentId),
setHead: (updater) => {
fake.head = updater(fake.head);
begin: (agentId, message) => {
const current = readSubmission(fake, agentId);
const stream = appendSubmittedUserMessage({
tail: current.tail,
head: current.head,
message,
});
writeSubmission(fake, agentId, {
...stream,
submissions: beginMessageSubmission(current.submissions, {
clientMessageId: message.clientMessageId!,
}),
});
},
setTail: (updater) => {
fake.tail = updater(fake.tail);
accept: (agentId, clientMessageId) => {
const current = readSubmission(fake, agentId);
writeSubmission(fake, agentId, {
...current,
submissions: acceptMessageSubmission(current.submissions, clientMessageId),
});
},
reject: (agentId, clientMessageId) => {
const current = readSubmission(fake, agentId);
const result = rejectMessageSubmission(current.submissions, clientMessageId);
const stream =
result.outcome === "rejected"
? removeSubmittedUserMessage({
tail: current.tail,
head: current.head,
clientMessageId,
})
: current;
writeSubmission(fake, agentId, { ...stream, submissions: result.submissions });
return result.outcome;
},
};
return fake;
}
const submissionsByFakeStream = new WeakMap<FakeStream, Map<string, MessageSubmissionRecord[]>>();
interface FakeSubmissionState {
tail: StreamItem[];
head: StreamItem[];
submissions: MessageSubmissionRecord[];
}
function readSubmission(fake: FakeStream, agentId: string): FakeSubmissionState {
return {
tail: fake.tail.get(agentId) ?? [],
head: fake.head.get(agentId) ?? [],
submissions: submissionsByFakeStream.get(fake)?.get(agentId) ?? [],
};
}
function writeSubmission(fake: FakeStream, agentId: string, state: FakeSubmissionState): void {
fake.tail = new Map(fake.tail).set(agentId, state.tail);
fake.head = new Map(fake.head).set(agentId, state.head);
const submissions = submissionsByFakeStream.get(fake) ?? new Map();
submissions.set(agentId, state.submissions);
submissionsByFakeStream.set(fake, submissions);
}
function createFakeQueue(
initial: Map<string, QueuedComposerMessage[]> = new Map(),
): QueueWriter & { state: Map<string, QueuedComposerMessage[]> } {
@@ -227,7 +290,6 @@ describe("cancelComposerAgent", () => {
isAgentRunning: boolean;
isCancellingAgent: boolean;
isConnected: boolean;
onCancelFailed: (error: unknown) => void;
} {
const canceledIds: string[] = [];
return {
@@ -241,53 +303,45 @@ describe("cancelComposerAgent", () => {
isAgentRunning: true,
isCancellingAgent: false,
isConnected: true,
onCancelFailed: () => undefined,
};
}
it("issues a cancel and reports true when the agent is running, connected, and not already canceling", () => {
it("returns the cancel request when the agent is running, connected, and not already canceling", async () => {
const input = baseInput();
const result = cancelComposerAgent(input);
expect(result).toBe(true);
expect(result).not.toBeNull();
await result;
expect(input.client.canceledIds).toEqual(["agent"]);
});
it("reports a rejected cancel so the composer can leave its canceling state", async () => {
it("returns a rejected cancel request to the composer", async () => {
const cancellationError = new Error("Provider rejected the interrupt");
const failures: unknown[] = [];
const input = baseInput();
input.client.cancelAgent = async () => {
throw cancellationError;
};
const result = cancelComposerAgent({
...input,
onCancelFailed: (error: unknown) => failures.push(error),
});
await new Promise((resolve) => setTimeout(resolve, 0));
expect(result).toBe(true);
expect(failures).toEqual([cancellationError]);
await expect(cancelComposerAgent(input)).rejects.toBe(cancellationError);
});
it("does nothing when the agent is not running", () => {
const input = baseInput();
const result = cancelComposerAgent({ ...input, isAgentRunning: false });
expect(result).toBe(false);
expect(result).toBeNull();
expect(input.client.canceledIds).toEqual([]);
});
it("does nothing when the agent is already being canceled", () => {
const input = baseInput();
const result = cancelComposerAgent({ ...input, isCancellingAgent: true });
expect(result).toBe(false);
expect(result).toBeNull();
expect(input.client.canceledIds).toEqual([]);
});
it("does nothing when disconnected or the client is null", () => {
const input = baseInput();
expect(cancelComposerAgent({ ...input, isConnected: false })).toBe(false);
expect(cancelComposerAgent({ ...input, client: null })).toBe(false);
expect(cancelComposerAgent({ ...input, isConnected: false })).toBeNull();
expect(cancelComposerAgent({ ...input, client: null })).toBeNull();
expect(input.client.canceledIds).toEqual([]);
});
});
@@ -337,7 +391,7 @@ describe("pickAndPersistImages", () => {
});
describe("dispatchComposerAgentMessage", () => {
it("removes the optimistic prompt when the host rejects it", async () => {
it("removes the submitted prompt when the host rejects it", async () => {
const rejection = new Error("Host rejected prompt");
const client = createFakeSendClient({ rejection });
const stream = createFakeStream();
@@ -349,14 +403,54 @@ describe("dispatchComposerAgentMessage", () => {
text: "rejected prompt",
attachments: [],
encodeImages: passthroughEncodeImages,
stream,
submission: stream,
}),
).rejects.toBe(rejection);
expect(stream.head.get("agent")).toBeUndefined();
expect(stream.head.get("agent")).toEqual([]);
expect(stream.tail.get("agent") ?? []).toEqual([]);
});
it("rolls back an already-running force send when its RPC fails", async () => {
const stream = createFakeStream();
const transportError = new Error("Force send failed while the prior turn was running");
const client = createFakeSendClient({ rejection: transportError });
await expect(
dispatchComposerAgentMessage({
client,
agentId: "agent",
text: "force send",
attachments: [],
encodeImages: passthroughEncodeImages,
submission: stream,
}),
).rejects.toBe(transportError);
expect(stream.tail.get("agent") ?? []).toEqual([]);
});
it("does not swallow a transport error when submission state is missing", async () => {
const transportError = new Error("Connection lost with unknown submission state");
const client = createFakeSendClient({ rejection: transportError });
const submission: MessageSubmissionWriter = {
begin: () => {},
accept: () => {},
reject: () => "unknown",
};
await expect(
dispatchComposerAgentMessage({
client,
agentId: "agent",
text: "unknown state",
attachments: [],
encodeImages: passthroughEncodeImages,
submission,
}),
).rejects.toBe(transportError);
});
it("sends text + image data + structured attachments and appends user_message to the tail when head is empty", async () => {
const client = createFakeSendClient();
const stream = createFakeStream();
@@ -371,7 +465,7 @@ describe("dispatchComposerAgentMessage", () => {
{ kind: "github_pr", item: prItem },
],
encodeImages: passthroughEncodeImages,
stream,
submission: stream,
});
expect(client.calls).toHaveLength(1);
@@ -393,7 +487,7 @@ describe("dispatchComposerAgentMessage", () => {
},
]);
expect(stream.head.get("agent")).toBeUndefined();
expect(stream.head.get("agent")).toEqual([]);
const tail = stream.tail.get("agent");
expect(tail).toHaveLength(1);
const userMessage = tail?.[0] as Extract<StreamItem, { kind: "user_message" }>;
@@ -402,7 +496,8 @@ describe("dispatchComposerAgentMessage", () => {
expect(userMessage.images).toEqual([image]);
expect(userMessage.attachments).toEqual(call.options.attachments);
expect(userMessage.id).toBe(call.options.messageId);
expect(userMessage.optimistic).toBe(true);
expect(userMessage.clientMessageId).toBe(call.options.messageId);
expect(userMessage.messageId).toBeUndefined();
});
it("can send legacy GitHub attachment payloads for old daemons", async () => {
@@ -416,7 +511,7 @@ describe("dispatchComposerAgentMessage", () => {
attachments: [{ kind: "forge_change_request", item: prItem }],
attachmentSubmitFormat: "legacy-github",
encodeImages: passthroughEncodeImages,
stream,
submission: stream,
});
expect(client.calls[0].options.attachments).toEqual([
@@ -449,11 +544,11 @@ describe("dispatchComposerAgentMessage", () => {
text: "next message",
attachments: [],
encodeImages: passthroughEncodeImages,
stream,
submission: stream,
});
expect(stream.head.get("agent")).toHaveLength(2);
expect(stream.tail.get("agent")).toBeUndefined();
expect(stream.tail.get("agent")).toEqual([]);
});
it("submits empty wire arrays when no attachments are provided", async () => {
@@ -466,7 +561,7 @@ describe("dispatchComposerAgentMessage", () => {
text: "plain message",
attachments: [],
encodeImages: passthroughEncodeImages,
stream,
submission: stream,
});
expect(client.calls[0]?.options).toMatchObject({
@@ -486,7 +581,7 @@ describe("dispatchComposerAgentMessage", () => {
text: "review this",
attachments: [review],
encodeImages: passthroughEncodeImages,
stream,
submission: stream,
});
expect(client.calls[0]?.options.attachments).toEqual([review.attachment]);
@@ -504,7 +599,7 @@ describe("dispatchComposerAgentMessage", () => {
text: "inspect element",
attachments: [browserElement],
encodeImages: passthroughEncodeImages,
stream,
submission: stream,
});
expect(client.calls[0]?.options.attachments).toEqual([
packages/app/src/composer/actions.ts
+20 -59
View File
@@ -12,13 +12,8 @@ import {
splitComposerAttachmentsForSubmit,
type ComposerAttachmentSubmitFormat,
} from "@/composer/attachments/submit";
import {
appendOptimisticUserMessageToStream,
buildOptimisticUserMessage,
generateMessageId,
type StreamItem,
type UserMessageItem,
} from "@/types/stream";
import { createUserMessage, generateMessageId, type UserMessageItem } from "@/types/stream";
import type { MessageSubmissionRejectionOutcome } from "@/composer/submission/model";
import type { PickedImageAttachmentInput } from "@/hooks/image-attachment-picker";
import { i18n } from "@/i18n/i18next";
@@ -70,11 +65,10 @@ export interface ComposerCancelClient {
cancelAgent: (agentId: string) => Promise<void> | void;
}
export interface AgentStreamWriter {
getTail: (agentId: string) => StreamItem[] | undefined;
getHead: (agentId: string) => StreamItem[] | undefined;
setHead: (updater: (prev: Map<string, StreamItem[]>) => Map<string, StreamItem[]>) => void;
setTail: (updater: (prev: Map<string, StreamItem[]>) => Map<string, StreamItem[]>) => void;
export interface MessageSubmissionWriter {
begin: (agentId: string, message: UserMessageItem) => void;
accept: (agentId: string, clientMessageId: string) => void;
reject: (agentId: string, clientMessageId: string) => MessageSubmissionRejectionOutcome;
}
export interface QueueWriter {
@@ -145,19 +139,16 @@ export interface CancelComposerAgentInput {
isAgentRunning: boolean;
isCancellingAgent: boolean;
isConnected: boolean;
onCancelFailed: (error: unknown) => void;
}
export function cancelComposerAgent(input: CancelComposerAgentInput): boolean {
if (!input.isAgentRunning || input.isCancellingAgent) return false;
if (!input.isConnected || !input.client) return false;
export function cancelComposerAgent(input: CancelComposerAgentInput): Promise<void> | null {
if (!input.isAgentRunning || input.isCancellingAgent) return null;
if (!input.isConnected || !input.client) return null;
try {
void Promise.resolve(input.client.cancelAgent(input.agentId)).catch(input.onCancelFailed);
return Promise.resolve(input.client.cancelAgent(input.agentId));
} catch (error) {
input.onCancelFailed(error);
return false;
return Promise.reject(error);
}
return true;
}
export interface DispatchComposerAgentMessageInput {
@@ -169,7 +160,7 @@ export interface DispatchComposerAgentMessageInput {
encodeImages: (
images: AttachmentMetadata[],
) => Promise<Array<{ data: string; mimeType: string }> | undefined>;
stream: AgentStreamWriter;
submission: MessageSubmissionWriter;
}
export async function dispatchComposerAgentMessage(
@@ -178,60 +169,30 @@ export async function dispatchComposerAgentMessage(
const wirePayload = splitComposerAttachmentsForSubmit(input.attachments, {
format: input.attachmentSubmitFormat,
});
const messageId = generateMessageId();
const userMessage = buildOptimisticUserMessage({
id: messageId,
const clientMessageId = generateMessageId();
const userMessage = createUserMessage({
clientMessageId,
text: input.text,
timestamp: new Date(),
images: wirePayload.images,
attachments: wirePayload.attachments,
});
const rollbackOptimisticMessage = appendUserMessageToStream(
input.agentId,
userMessage,
input.stream,
);
input.submission.begin(input.agentId, userMessage);
try {
const imagesData = await input.encodeImages(wirePayload.images);
await input.client.sendAgentMessage(input.agentId, input.text, {
messageId,
messageId: clientMessageId,
images: imagesData ?? [],
attachments: wirePayload.attachments,
});
input.submission.accept(input.agentId, clientMessageId);
} catch (error) {
rollbackOptimisticMessage();
const outcome = input.submission.reject(input.agentId, clientMessageId);
if (outcome === "accepted") return;
throw error;
}
}
function appendUserMessageToStream(
agentId: string,
userMessage: UserMessageItem,
stream: AgentStreamWriter,
): () => void {
const result = appendOptimisticUserMessageToStream({
tail: stream.getTail(agentId) ?? [],
head: stream.getHead(agentId) ?? [],
message: userMessage,
placement: "active-head",
});
const write = result.changedHead ? stream.setHead : stream.setTail;
const items = result.changedHead ? result.head : result.tail;
write((prev) => new Map(prev).set(agentId, items));
return () => {
write((prev) => {
const current = prev.get(agentId);
if (!current) return prev;
const nextItems = current.filter(
(item) => item.id !== userMessage.id || item.kind !== "user_message" || !item.optimistic,
);
if (nextItems.length === current.length) return prev;
return new Map(prev).set(agentId, nextItems);
});
};
}
export interface QueueComposerMessageInput {
agentId: string;
text: string;
packages/app/src/composer/draft/create-flow.test.ts
+3 -3
View File
@@ -13,7 +13,7 @@ describe("useDraftAgentCreateFlow", () => {
useCreateFlowStore.setState({ pendingByDraftId: {} });
});
it("renders a prepared new-workspace create attempt as optimistic chat before continuing it", async () => {
it("renders a prepared new-workspace submission before continuing it", async () => {
const image: UserMessageImageAttachment = {
id: "image-1",
mimeType: "image/png",
@@ -60,13 +60,13 @@ describe("useDraftAgentCreateFlow", () => {
expect(result.current.isSubmitting).toBe(true);
expect(result.current.draftAgent).toEqual({ currentAttempt: attempt });
expect(result.current.optimisticStreamItems).toEqual([
expect(result.current.submittedStreamItems).toEqual([
{
kind: "user_message",
id: "msg-prepared",
clientMessageId: "msg-prepared",
text: "build this",
timestamp: attempt.timestamp,
optimistic: true,
images: [image],
attachments: [attachment],
},
packages/app/src/composer/draft/create-flow.ts
+19 -13
View File
@@ -6,14 +6,16 @@ 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 {
buildOptimisticUserMessage,
createUserMessage,
generateMessageId,
type StreamItem,
type UserMessageImageAttachment,
} from "@/types/stream";
import type { AgentAttachment } from "@getpaseo/protocol/messages";
import type { PendingMessageSubmission } from "@/composer/submission/model";
const EMPTY_STREAM_ITEMS: StreamItem[] = [];
@@ -126,14 +128,10 @@ 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";
const optimisticStreamItems = useMemo<StreamItem[]>(() => {
const submittedStreamItems = useMemo<StreamItem[]>(() => {
if (machine.tag !== "creating") {
return EMPTY_STREAM_ITEMS;
}
@@ -147,8 +145,8 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
}
return [
buildOptimisticUserMessage({
id: machine.attempt.clientMessageId,
createUserMessage({
clientMessageId: machine.attempt.clientMessageId,
text: machine.attempt.text,
timestamp: machine.attempt.timestamp,
images: machine.attempt.images,
@@ -156,6 +154,14 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
}),
];
}, [machine]);
const pendingMessageSubmissions = useMemo<readonly PendingMessageSubmission[]>(() => {
if (machine.tag !== "creating") return [];
return [
{
clientMessageId: machine.attempt.clientMessageId,
},
];
}, [machine]);
const draftAgent = useMemo<TDraftAgent | null>(() => {
if (machine.tag !== "creating") {
@@ -192,11 +198,11 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
if (createResult.agentId) {
updatePendingAgentId({ draftId, agentId: createResult.agentId });
handoffCreatedAgentUserMessage(
handoffCreatedAgentMessageSubmission(
pendingServerId,
createResult.agentId,
buildOptimisticUserMessage({
id: attempt.clientMessageId,
createUserMessage({
clientMessageId: attempt.clientMessageId,
text: attempt.text,
timestamp: attempt.timestamp,
images: attempt.images,
@@ -218,7 +224,6 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
}
},
[
handoffCreatedAgentUserMessage,
clearPendingCreateAttempt,
createRequest,
draftId,
@@ -326,7 +331,8 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
machine,
formErrorMessage,
isSubmitting,
optimisticStreamItems,
submittedStreamItems,
pendingMessageSubmissions,
draftAgent,
handleCreateFromInput,
continueCreateFromAttempt,
packages/app/src/composer/draft/workspace-tab.tsx
+11 -2
View File
@@ -15,6 +15,7 @@ import { composerWorkspaceAttachment } from "@/composer/attachments/workspace";
import { useAgentInputDraft } from "@/composer/draft/input-draft";
import type { CreateAgentInitialValues } from "@/hooks/use-agent-form-state";
import { useDraftAgentCreateFlow, type DraftCreateAttempt } from "@/composer/draft/create-flow";
import { resolveTurnPresentation, TURN_LIVENESS_IDLE } from "@/timeline/turn-liveness";
import { useHostRuntimeClient, useHostRuntimeIsConnected } from "@/runtime/host-runtime";
import { buildWorkspaceDraftAgentConfig } from "@/screens/workspace/workspace-draft-agent-config";
import { buildDraftStoreKey } from "@/stores/draft-keys";
@@ -243,6 +244,7 @@ function buildDraftAgentSnapshot(input: {
id: tabId,
provider,
status: "running",
activeTurn: null,
createdAt: now,
updatedAt: now,
lastUserMessageAt: now,
@@ -467,7 +469,8 @@ export function WorkspaceDraftAgentTab({
const {
formErrorMessage,
isSubmitting,
optimisticStreamItems,
submittedStreamItems,
pendingMessageSubmissions,
draftAgent,
handleCreateFromInput,
continueCreateFromAttempt,
@@ -529,6 +532,10 @@ export function WorkspaceDraftAgentTab({
onCreated(result);
},
});
const turnPresentation = useMemo(
() => resolveTurnPresentation(TURN_LIVENESS_IDLE, pendingMessageSubmissions.length > 0),
[pendingMessageSubmissions],
);
useAgentControlCommandCenterActions({
sourceId: `draft:${serverId}:${tabId}`,
enabled: isPaneFocused && !isSubmitting,
@@ -654,7 +661,9 @@ export function WorkspaceDraftAgentTab({
agentId={tabId}
serverId={serverId}
context={draftAgent}
streamItems={optimisticStreamItems}
streamItems={submittedStreamItems}
pendingMessageSubmissions={pendingMessageSubmissions}
turnPresentation={turnPresentation}
pendingPermissions={EMPTY_PENDING_PERMISSIONS}
onOpenWorkspaceFile={onOpenWorkspaceFile}
/>
packages/app/src/composer/index.tsx
+40 -75
View File
@@ -41,7 +41,7 @@ import {
} from "@/composer/agent-controls";
import { ContextWindowMeter } from "@/components/context-window-meter";
import { useImageAttachmentPicker } from "@/hooks/use-image-attachment-picker";
import { useSessionStore } from "@/stores/session-store";
import { selectAgentTurnPresentation, useSessionStore } from "@/stores/session-store";
import { useFilePicker } from "@/hooks/use-file-picker";
import { useFileDrop } from "@/components/file-drop/use-file-drop";
import type { DroppedItem } from "@/components/file-drop/types";
@@ -64,7 +64,6 @@ import {
sendQueuedComposerMessageNow,
toggleGithubAttachmentFromPicker,
uploadFileAttachments,
type AgentStreamWriter,
type QueueWriter,
type QueuedComposerMessage,
} from "@/composer/actions";
@@ -91,6 +90,7 @@ import { useKeyboardActionHandler } from "@/hooks/use-keyboard-action-handler";
import type { KeyboardActionDefinition } from "@/keyboard/keyboard-action-dispatcher";
import type { MessageInputKeyboardActionKind } from "@/keyboard/actions";
import { submitAgentInput } from "@/composer/submit";
import { createMessageSubmissionWriter } from "@/composer/submission/writer";
import { ComposerKeyboardScopeProvider } from "@/composer/keyboard-scope";
import { useAppSettings } from "@/hooks/use-settings";
import { isWeb, isNative } from "@/constants/platform";
@@ -910,22 +910,6 @@ function ComposerCancelButton({
);
}
interface ComposerCancelButtonSlotProps extends ComposerCancelButtonProps {
isAgentRunning: boolean;
hasSendableContent: boolean;
isProcessing: boolean;
}
function ComposerCancelButtonSlot({
isAgentRunning,
hasSendableContent,
isProcessing,
...rest
}: ComposerCancelButtonSlotProps) {
if (!isAgentRunning || hasSendableContent || isProcessing) return null;
return <ComposerCancelButton {...rest} />;
}
interface ComposerVoiceModeButtonProps {
buttonIconSize: number;
handleToggleRealtimeVoice: () => void;
@@ -943,9 +927,7 @@ interface ComposerRightControlsSlotProps extends ComposerVoiceModeButtonProps {
hasAgent: boolean;
isAgentRunning: boolean;
hasSendableContent: boolean;
isProcessing: boolean;
isCompact: boolean;
cancelButton: ReactElement;
}
function ComposerRightControlsSlot({
@@ -953,20 +935,16 @@ function ComposerRightControlsSlot({
hasAgent,
isAgentRunning,
hasSendableContent,
isProcessing,
isCompact,
cancelButton,
...voiceProps
}: ComposerRightControlsSlotProps) {
const hideVoiceForCompactInput = isCompact && hasSendableContent;
const showVoiceModeButton =
!isVoiceModeForAgent && hasAgent && !isAgentRunning && !hideVoiceForCompactInput;
const shouldShowCancelButton = isAgentRunning && !hasSendableContent && !isProcessing;
if (!showVoiceModeButton && !shouldShowCancelButton) return null;
if (!showVoiceModeButton) return null;
return (
<View style={styles.rightControls}>
{showVoiceModeButton ? <ComposerVoiceModeButton {...voiceProps} /> : null}
{cancelButton}
<ComposerVoiceModeButton {...voiceProps} />
</View>
);
}
@@ -1079,8 +1057,6 @@ export function Composer({
const queuedMessages = queuedMessagesRaw ?? EMPTY_ARRAY;
const setQueuedMessages = useSessionStore((state) => state.setQueuedMessages);
const setAgentStreamTail = useSessionStore((state) => state.setAgentStreamTail);
const setAgentStreamHead = useSessionStore((state) => state.setAgentStreamHead);
const isCompactFormFactor = useIsCompactFormFactor();
const isCompactLayout = resolveCompactLayout(isCompactLayoutOverride, isCompactFormFactor);
@@ -1125,7 +1101,6 @@ export function Composer({
const [cursorIndex, setCursorIndex] = useState(0);
const [isProcessing, setIsProcessing] = useState(false);
const [isUploadingFile, setIsUploadingFile] = useState(false);
const [isCancellingAgent, setIsCancellingAgent] = useState(false);
const [sendError, setSendError] = useState<string | null>(null);
const [isMessageInputFocused, setIsMessageInputFocused] = useState(false);
const [isGithubPickerOpen, setIsGithubPickerOpen] = useState(false);
@@ -1283,12 +1258,6 @@ export function Composer({
if (!client) {
throw new Error(t("workspace.terminal.hostDisconnected"));
}
const stream: AgentStreamWriter = {
getTail: (id) => useSessionStore.getState().sessions[serverId]?.agentStreamTail?.get(id),
getHead: (id) => useSessionStore.getState().sessions[serverId]?.agentStreamHead?.get(id),
setHead: (updater) => setAgentStreamHead(serverId, updater),
setTail: (updater) => setAgentStreamTail(serverId, updater),
};
await dispatchComposerAgentMessage({
client,
agentId: targetAgentId,
@@ -1298,25 +1267,25 @@ export function Composer({
supportsForgeAttachments: supportsForgeSearch,
}),
encodeImages,
stream,
submission: createMessageSubmissionWriter(serverId),
});
onAttentionPromptSend?.();
};
}, [
client,
onAttentionPromptSend,
serverId,
setAgentStreamTail,
setAgentStreamHead,
supportsForgeSearch,
t,
]);
}, [client, onAttentionPromptSend, serverId, supportsForgeSearch, t]);
useEffect(() => {
onSubmitMessageRef.current = onSubmitMessage;
}, [onSubmitMessage]);
const isAgentRunning = agentState.status === "running";
const hasActiveTurn = useSessionStore(
(state) => selectAgentTurnPresentation(state.sessions[serverId], agentId).isActive,
);
const isCancellingAgent = useSessionStore(
(state) => selectAgentTurnPresentation(state.sessions[serverId], agentId).isCancelling,
);
const beginAgentCancellation = useSessionStore((state) => state.beginAgentCancellation);
const settleAgentCancellation = useSessionStore((state) => state.settleAgentCancellation);
const isAgentRunning = hasActiveTurn;
const hasAgent = agentState.status !== null;
const queueWriter = useMemo<QueueWriter>(
@@ -1550,31 +1519,37 @@ export function Composer({
[openAttachment],
);
useEffect(() => {
if (!isAgentRunning || !isConnected) {
setIsCancellingAgent(false);
}
}, [isAgentRunning, isConnected]);
const handleCancelAgent = useCallback(() => {
const didCancel = cancelComposerAgent({
const targetAgentId = agentIdRef.current;
const cancellation = cancelComposerAgent({
client,
agentId: agentIdRef.current,
agentId: targetAgentId,
isAgentRunning,
isCancellingAgent,
isConnected,
onCancelFailed: (error) => {
setIsCancellingAgent(false);
});
if (!cancellation) return;
const requestId = beginAgentCancellation(serverId, targetAgentId);
void cancellation
.catch((error) => {
const message = resolveErrorMessage(error);
if (message && message.trim().length > 0) {
toastErrorRef.current(message);
}
},
});
if (!didCancel) return;
setIsCancellingAgent(true);
})
.finally(() => {
settleAgentCancellation(serverId, targetAgentId, requestId);
});
messageInputRef.current?.focus();
}, [client, isAgentRunning, isCancellingAgent, isConnected]);
}, [
beginAgentCancellation,
client,
isAgentRunning,
isCancellingAgent,
isConnected,
serverId,
settleAgentCancellation,
]);
const focusMessageInputForKeyboardAction = useCallback(() => {
focusMessageInputWithPlatformStrategy(messageInputRef);
@@ -1708,12 +1683,9 @@ export function Composer({
[isCompactLayout, voiceButtonDisabled],
);
const cancelButton = useMemo(
const activeActionContent = useMemo(
() => (
<ComposerCancelButtonSlot
isAgentRunning={isAgentRunning}
hasSendableContent={hasSendableContent}
isProcessing={isProcessing}
<ComposerCancelButton
buttonIconSize={buttonIconSize}
cancelButtonStyle={cancelButtonStyle}
handleCancelAgent={handleCancelAgent}
@@ -1728,11 +1700,8 @@ export function Composer({
buttonIconSize,
cancelButtonStyle,
handleCancelAgent,
hasSendableContent,
isAgentRunning,
isCancellingAgent,
isConnected,
isProcessing,
t,
],
);
@@ -1744,7 +1713,6 @@ export function Composer({
hasAgent={hasAgent}
isAgentRunning={isAgentRunning}
hasSendableContent={hasSendableContent}
isProcessing={isProcessing}
isCompact={isCompactLayout}
buttonIconSize={buttonIconSize}
handleToggleRealtimeVoice={handleToggleRealtimeVoice}
@@ -1753,19 +1721,16 @@ export function Composer({
realtimeVoiceButtonStyle={realtimeVoiceButtonStyle}
voiceToggleKeys={voiceToggleKeys}
t={t}
cancelButton={cancelButton}
/>
),
[
buttonIconSize,
cancelButton,
handleToggleRealtimeVoice,
hasAgent,
hasSendableContent,
isAgentRunning,
isConnected,
isCompactLayout,
isProcessing,
isVoiceModeForAgent,
isVoiceSwitching,
realtimeVoiceButtonStyle,
@@ -1779,8 +1744,7 @@ export function Composer({
agentState.contextWindowUsedTokens,
);
const contextWindowPending =
agentState.status === "initializing" || agentState.status === "running";
const contextWindowPending = agentState.status === "initializing" || isAgentRunning;
const contextWindowMeterGlyphSize = isCompactLayout ? ICON_SIZE.md : buttonIconSize;
const contextWindowMeter = useMemo(
@@ -2102,6 +2066,7 @@ export function Composer({
leftContent={leftContent}
beforeVoiceContent={beforeVoiceContent}
rightContent={rightContent}
activeActionContent={activeActionContent}
voiceServerId={serverId}
voiceAgentId={agentId}
isAgentRunning={isAgentRunning}
packages/app/src/composer/input/input.tsx
+51 -31
View File
@@ -121,8 +121,10 @@ export interface MessageInputProps {
leftContent?: React.ReactNode;
/** Content to render on the right side before the voice button (e.g., context window meter) */
beforeVoiceContent?: React.ReactNode;
/** Content to render on the right side after voice button (e.g., realtime button, cancel button) */
/** Auxiliary content to render on the right side after the voice button. */
rightContent?: React.ReactNode;
/** Primary action to render when the agent is active and the composer has no sendable content. */
activeActionContent?: React.ReactNode;
voiceServerId?: string;
voiceAgentId?: string;
/** When true and there's sendable content, calls onQueue instead of onSubmit */
@@ -779,6 +781,40 @@ function SendButtonTooltip({
);
}
type PrimaryActionKind = "send" | "active" | "none";
function hasSendableComposerContent(input: {
value: string;
attachments: readonly ComposerAttachment[];
hasExternalContent: boolean;
}): boolean {
return input.value.trim().length > 0 || input.attachments.length > 0 || input.hasExternalContent;
}
function resolvePrimaryActionKind(input: {
hasSendableContent: boolean;
allowEmptySubmit: boolean;
isAgentRunning: boolean;
isSubmitLoading: boolean;
}): PrimaryActionKind {
if (input.hasSendableContent || input.allowEmptySubmit) return "send";
if (input.isAgentRunning) return "active";
if (input.isSubmitLoading) return "send";
return "none";
}
function PrimaryAction({
kind,
activeActionContent,
...sendButtonProps
}: {
kind: PrimaryActionKind;
activeActionContent: React.ReactNode;
} & React.ComponentProps<typeof SendButtonTooltip>) {
if (kind === "active") return activeActionContent;
if (kind === "send") return <SendButtonTooltip {...sendButtonProps} />;
return null;
}
interface ToggleRealtimeVoiceContext {
voice:
| {
@@ -928,30 +964,6 @@ function computeShouldShowDictationOverlay(
return isDictating || isDictationProcessing || dictationStatus === "failed";
}
interface SendableContentInput {
value: string;
attachments: ComposerAttachment[];
hasExternalContent: boolean;
allowEmptySubmit: boolean;
isSubmitLoading: boolean;
}
interface SendableContentOutput {
hasAttachments: boolean;
hasRealContent: boolean;
hasSendableContent: boolean;
shouldShowSendButton: boolean;
}
function computeSendableContent(input: SendableContentInput): SendableContentOutput {
const hasAttachments = input.attachments.length > 0;
const hasRealContent = input.value.trim().length > 0 || hasAttachments;
const hasSendableContent = hasRealContent || input.hasExternalContent;
const shouldShowSendButton =
hasSendableContent || input.allowEmptySubmit || input.isSubmitLoading;
return { hasAttachments, hasRealContent, hasSendableContent, shouldShowSendButton };
}
function computeIsDictationStartEnabled(
isReadyForDictation: boolean | undefined,
isConnected: boolean,
@@ -1047,6 +1059,7 @@ interface ResolvedMessageInputProps {
leftContent: React.ReactNode;
beforeVoiceContent: React.ReactNode;
rightContent: React.ReactNode;
activeActionContent: React.ReactNode;
voiceServerId: string | undefined;
voiceAgentId: string | undefined;
isAgentRunning: boolean;
@@ -1089,6 +1102,7 @@ function resolveMessageInputProps(props: MessageInputProps): ResolvedMessageInpu
leftContent: props.leftContent,
beforeVoiceContent: props.beforeVoiceContent,
rightContent: props.rightContent,
activeActionContent: props.activeActionContent,
voiceServerId: props.voiceServerId,
voiceAgentId: props.voiceAgentId,
isAgentRunning: props.isAgentRunning ?? false,
@@ -1139,6 +1153,7 @@ export const MessageInput = forwardRef<MessageInputRef, MessageInputProps>(
leftContent,
beforeVoiceContent,
rightContent,
activeActionContent,
voiceServerId,
voiceAgentId,
isAgentRunning,
@@ -1553,11 +1568,14 @@ export const MessageInput = forwardRef<MessageInputRef, MessageInputProps>(
});
}
const { shouldShowSendButton } = computeSendableContent({
value,
attachments,
hasExternalContent,
const primaryActionKind = resolvePrimaryActionKind({
hasSendableContent: hasSendableComposerContent({
value,
attachments,
hasExternalContent,
}),
allowEmptySubmit,
isAgentRunning,
isSubmitLoading,
});
const { canPressLoadingButton, isSendButtonDisabled, defaultActionQueues } =
@@ -1756,8 +1774,10 @@ export const MessageInput = forwardRef<MessageInputRef, MessageInputProps>(
dictationToggleKeys={dictationToggleKeys}
/>
{rightContent}
<SendButtonTooltip
shouldShow={shouldShowSendButton}
<PrimaryAction
kind={primaryActionKind}
activeActionContent={activeActionContent}
shouldShow
canPressLoadingButton={canPressLoadingButton}
onSubmitLoadingPress={onSubmitLoadingPress}
onDefaultSendAction={handleDefaultSendAction}
packages/app/src/composer/submission/model.test.ts
+104
View File
@@ -0,0 +1,104 @@
import { describe, expect, it } from "vitest";
import {
acceptMessageSubmission,
beginMessageSubmission,
getActiveMessageSubmissions,
getSendingClientMessageIds,
observeMessageSubmissionCanonical,
rejectMessageSubmission,
} from "./model";
describe("message submission transactions", () => {
it("tracks every in-flight submission independently", () => {
const first = beginMessageSubmission([], { clientMessageId: "client-1" });
const both = beginMessageSubmission(first, { clientMessageId: "client-2" });
expect(getActiveMessageSubmissions(both).map((item) => item.clientMessageId)).toEqual([
"client-1",
"client-2",
]);
expect(getSendingClientMessageIds(both)).toEqual(["client-1", "client-2"]);
});
it("settles only the accepted transaction on RPC acceptance", () => {
const both = beginMessageSubmission(
beginMessageSubmission([], { clientMessageId: "client-1" }),
{ clientMessageId: "client-2" },
);
const acknowledged = observeMessageSubmissionCanonical(both, ["client-1"]);
expect(acceptMessageSubmission(acknowledged, "client-1")).toEqual([
{
clientMessageId: "client-2",
providerAcknowledged: false,
rpcSettled: false,
},
]);
});
it("keeps activity until an accepted RPC's canonical row is observed", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1" });
const accepted = acceptMessageSubmission(sending, "client-1");
expect(accepted).toEqual([
{
clientMessageId: "client-1",
providerAcknowledged: false,
rpcSettled: true,
},
]);
expect(getActiveMessageSubmissions(accepted).map((item) => item.clientMessageId)).toEqual([
"client-1",
]);
expect(observeMessageSubmissionCanonical(accepted, ["client-1"])).toEqual([]);
});
it("records provider acknowledgement without settling another transaction", () => {
const both = beginMessageSubmission(
beginMessageSubmission([], { clientMessageId: "client-1" }),
{ clientMessageId: "client-2" },
);
const observed = observeMessageSubmissionCanonical(both, ["client-1"]);
expect(observed).toEqual([
{
clientMessageId: "client-1",
providerAcknowledged: true,
rpcSettled: false,
},
{
clientMessageId: "client-2",
providerAcknowledged: false,
rpcSettled: false,
},
]);
expect(getSendingClientMessageIds(observed)).toEqual(["client-2"]);
});
it("does not roll back a provider-acknowledged prompt on a later transport error", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1" });
const observed = observeMessageSubmissionCanonical(sending, ["client-1"]);
expect(rejectMessageSubmission(observed, "client-1")).toEqual({
outcome: "accepted",
submissions: [],
});
});
it("rejects an unacknowledged transaction", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1" });
expect(rejectMessageSubmission(sending, "client-1")).toEqual({
outcome: "rejected",
submissions: [],
});
});
it("does not create duplicate transaction identity", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1" });
expect(() => beginMessageSubmission(sending, { clientMessageId: "client-1" })).toThrow(
"Message submission already exists",
);
});
});
packages/app/src/composer/submission/model.ts
+92
View File
@@ -0,0 +1,92 @@
export interface PendingMessageSubmission {
clientMessageId: string;
}
export type MessageSubmissionRecord = PendingMessageSubmission & {
providerAcknowledged: boolean;
rpcSettled: boolean;
};
const EMPTY_MESSAGE_SUBMISSIONS: readonly MessageSubmissionRecord[] = [];
export function getActiveMessageSubmissions(
submissions: readonly MessageSubmissionRecord[] | null | undefined,
): readonly PendingMessageSubmission[] {
return (submissions ?? EMPTY_MESSAGE_SUBMISSIONS).filter(
(submission) => !submission.providerAcknowledged,
);
}
export function getSendingClientMessageIds(
submissions: readonly MessageSubmissionRecord[] | null | undefined,
): string[] {
return (submissions ?? [])
.filter((submission) => !submission.providerAcknowledged)
.map((submission) => submission.clientMessageId);
}
export type MessageSubmissionRejectionOutcome = "rejected" | "accepted" | "unknown";
export interface MessageSubmissionRejectionResult {
submissions: MessageSubmissionRecord[];
outcome: MessageSubmissionRejectionOutcome;
}
export function beginMessageSubmission(
submissions: readonly MessageSubmissionRecord[],
input: PendingMessageSubmission,
): MessageSubmissionRecord[] {
if (submissions.some((submission) => submission.clientMessageId === input.clientMessageId)) {
throw new Error(`Message submission already exists: ${input.clientMessageId}`);
}
return [...submissions, { ...input, providerAcknowledged: false, rpcSettled: false }];
}
export function acceptMessageSubmission(
submissions: readonly MessageSubmissionRecord[],
clientMessageId: string,
): MessageSubmissionRecord[] {
const index = submissions.findIndex(
(submission) => submission.clientMessageId === clientMessageId,
);
if (index < 0) return submissions as MessageSubmissionRecord[];
const submission = submissions[index];
if (submission.providerAcknowledged) {
return submissions.filter((_, submissionIndex) => submissionIndex !== index);
}
if (submission.rpcSettled) return submissions as MessageSubmissionRecord[];
return submissions.map((item, submissionIndex) =>
submissionIndex === index ? { ...item, rpcSettled: true } : item,
);
}
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 (submission.providerAcknowledged || !observed.has(submission.clientMessageId)) {
return [submission];
}
changed = true;
return submission.rpcSettled ? [] : [{ ...submission, providerAcknowledged: true }];
});
return changed ? next : (submissions as MessageSubmissionRecord[]);
}
export function rejectMessageSubmission(
submissions: readonly MessageSubmissionRecord[],
clientMessageId: string,
): MessageSubmissionRejectionResult {
const submission = submissions.find((item) => item.clientMessageId === clientMessageId);
if (!submission) {
return { outcome: "unknown", submissions: submissions as MessageSubmissionRecord[] };
}
return {
outcome: submission.providerAcknowledged ? "accepted" : "rejected",
submissions: submissions.filter((item) => item.clientMessageId !== clientMessageId),
};
}
packages/app/src/composer/submission/writer.test.ts
+49
View File
@@ -0,0 +1,49 @@
import { afterEach, expect, test } from "vitest";
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
import { useSessionStore } from "@/stores/session-store";
import type { UserMessageItem } from "@/types/stream";
import { createMessageSubmissionWriter } from "./writer";
const CANONICAL_ONLY_SERVER_ID = "canonical-only";
function submittedMessage(clientMessageId: string): UserMessageItem {
return {
kind: "user_message",
id: clientMessageId,
clientMessageId,
text: "hello",
timestamp: new Date("2026-07-31T10:00:00.000Z"),
};
}
function initializeSession(serverId: string, features: Record<string, boolean>): void {
const store = useSessionStore.getState();
store.initializeSession(serverId, null as unknown as DaemonClient);
store.updateSessionServerInfo(serverId, {
serverId,
hostname: null,
version: "0.2.6",
features,
});
}
function submissionIds(serverId: string): string[] {
return (
useSessionStore.getState().sessions[serverId]?.messageSubmissions.get("agent-1") ?? []
).map((submission) => submission.clientMessageId);
}
afterEach(() => {
useSessionStore.getState().clearSession(CANONICAL_ONLY_SERVER_ID);
});
test("tracks submissions when canonical prompts are supported", () => {
initializeSession(CANONICAL_ONLY_SERVER_ID, { canonicalSubmittedPrompts: true });
createMessageSubmissionWriter(CANONICAL_ONLY_SERVER_ID).begin(
"agent-1",
submittedMessage("canonical-only-message"),
);
expect(submissionIds(CANONICAL_ONLY_SERVER_ID)).toEqual(["canonical-only-message"]);
});
packages/app/src/composer/submission/writer.ts
+72
View File
@@ -0,0 +1,72 @@
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";
},
};
}
export function createMessageSubmissionWriter(serverId: string): MessageSubmissionWriter {
const supportsTrackedMessageSubmissions =
useSessionStore.getState().sessions[serverId]?.serverInfo?.features
?.canonicalSubmittedPrompts === true;
if (!supportsTrackedMessageSubmissions) {
// COMPAT(canonicalSubmittedPrompts): added in v0.2.6; remove the gate after 2027-01-31 once daemon floor >= v0.2.6.
return createUntrackedMessageSubmissionWriter(serverId);
}
return {
begin: (agentId, message) =>
useSessionStore.getState().beginAgentMessageSubmission(serverId, agentId, message),
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 {
return useSessionStore.getState().handoffCreatedAgentUserMessage(serverId, agentId, message);
}
packages/app/src/composer/submit.ts
+1 -1
View File
@@ -51,7 +51,7 @@ export async function submitAgentInput<TAttachment>(
return "queued";
}
// Clear immediately so optimistic stream updates and composer state stay in sync.
// Clear immediately so the submitted timeline row and composer state stay in sync.
if (shouldClearOnSubmit) {
input.setUserInput("");
input.setAttachments([]);