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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion apps/web/e2e/channel-archive.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ test("channel branch list shows real branches without default branch chrome", as
const quickSplitRootThreadId = isRecord(quickSplitBranch) && typeof quickSplitBranch.rootThreadId === "string" ? quickSplitBranch.rootThreadId : "";
await channelComposerTextbox(page).fill("Branch-scoped message");
await page.locator(".channel-composer").getByRole("button", { name: "Send" }).click();
await expect.poll(() => state.lastChannelMessageRequest?.threadId).toBe(quickSplitRootThreadId);
await expect.poll(() => state.lastChannelTurnRequest?.surface).toEqual({ kind: "thread", threadId: quickSplitRootThreadId });

await openArchivedItemsSettings(page);
const archivedSettings = page.locator(".archived-items-settings-panel");
Expand Down
526 changes: 461 additions & 65 deletions apps/web/e2e/channel.spec.ts

Large diffs are not rendered by default.

596 changes: 524 additions & 72 deletions apps/web/e2e/support/mock-daemon.ts

Large diffs are not rendered by default.

4 changes: 2 additions & 2 deletions apps/web/e2e/ui2-channel-error-ownership.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ test("T13-P.9D a late Engineering send failure cannot leak into Research", async
const requestStarted = createDeferred<void>();
const releaseFailure = createDeferred<void>();
const failureMessage = "Late Engineering message failed.";
await page.route(/\/channels\/channel_engineering\/messages\/stream(?:\?.*)?$/u, async (route) => {
await page.route(/\/channels\/channel_engineering\/turns\/stream(?:\?.*)?$/u, async (route) => {
if (route.request().method() !== "POST") {
await route.fallback();
return;
Expand Down Expand Up @@ -95,7 +95,7 @@ test("T13-P.9D a late Engineering send failure cannot leak into Research", async

const failedResponse = page.waitForResponse((response) =>
response.status() === 503 &&
/\/channels\/channel_engineering\/messages\/stream(?:\?.*)?$/u.test(response.url())
/\/channels\/channel_engineering\/turns\/stream(?:\?.*)?$/u.test(response.url())
);
releaseFailure.resolve();
await failedResponse;
Expand Down
2 changes: 1 addition & 1 deletion apps/web/e2e/ui2-channel-header-search.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ test("channel header refresh and search keep local state and overlays independen
await expect(searchInput).toHaveCount(0);
await expect(page.getByLabel("Channel message feed")).toContainText("Refresh source message.");

await page.route("**/channels/channel_engineering/messages?*", async (route) => {
await page.route("**/channels/channel_engineering/items?*", async (route) => {
await new Promise((resolve) => setTimeout(resolve, 450));
await route.fallback();
});
Expand Down
5 changes: 4 additions & 1 deletion apps/web/e2e/ui2-channel-utility-input-migration.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,10 @@ test("dark reduced-motion Channel terminal and topic reply keep command and keyb
await expectNoOverflow(topicPanel);
await captureSurface(page, "1280x800-ling-dark-channel-topic-reply-input.png");
await replyInput.press("Enter");
await expect.poll(() => state.lastChannelMessageRequest?.threadId).toBe("message_shared_topic_root");
await expect.poll(() => state.lastChannelTurnRequest?.surface).toEqual({
kind: "thread",
threadId: "thread_topic_channel_engineering_message_shared_topic_root",
});
await expect(topicPanel).toContainText("First reply line");
await expect(topicPanel).toContainText("Second reply line");
await topicPanel.getByRole("button", { name: "Close topic" }).click();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ test("Channel WorkCell keeps artifacts, changed files, references, and hover pre
const composer = page.getByRole("textbox", { name: "Message this channel or @ an AI teammate", exact: true });
await composer.fill("Please inspect apps/web/src/features/workspace/WorkspaceClientView.tsx");
await page.locator(".channel-composer").getByRole("button", { name: "Send" }).click();
await expect.poll(() => state.lastChannelMessageRequest?.fileRefs?.[0]?.path)
await expect.poll(() => state.lastChannelTurnRequest?.fileRefs?.[0]?.path)
.toBe("apps/web/src/features/workspace/WorkspaceClientView.tsx");
const references = feed.getByLabel("References").last();
const fileReference = references.getByRole("button", { name: /WorkspaceClientView\.tsx/u });
Expand Down
49 changes: 35 additions & 14 deletions apps/web/src/api/channels.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,11 @@ import type {
ChannelLinkPreviewListResponse,
ChannelLinkPreviewMutationResponse,
ChannelListResponse,
ChannelMessageListResponse,
ChannelConversationContextInspectResponse,
ChannelConversationItemPageResponse,
ChannelConversationTurnResponse,
ChannelIntentDecisionListResponse,
ChannelMessageRouteDecisionListResponse,
ChannelMessageMutationResponse,
ChannelMutationResponse,
ChannelPlaybookCandidateListResponse,
ChannelPlaybookCandidateMutationResponse,
Expand Down Expand Up @@ -63,13 +64,14 @@ import type {
CreateChannelExecutionProposalRequest,
CreateChannelHandoffRequest,
CreateChannelLinkPreviewRequest,
CreateChannelMessageRequest,
CreateChannelTurnRequest,
CreateTaskRequest,
CreateChannelWorkflowFromHandoffRequest,
CreateChannelWorkflowRequest,
CreateChannelWorkflowStepRequest,
CreateChannelWorkflowTemplateCompileRequest,
StartChannelWorkflowStepRequest,
ListChannelConversationItemsOptions,
Task,
TaskEditFormState,
TaskListResponse,
Expand All @@ -87,6 +89,8 @@ import type {
UpdateChannelWorkflowRequest,
UpdateChannelWorkflowStepRequest,
UpdateChannelRequest,
CorrectChannelTurnContextRequest,
CorrectChannelTurnContextResponse,
} from "../types.js";
import { channelFormToPayload, channelFormToUpdatePayload, taskEditFormToPayload } from "./channel-payloads.js";
import { daemonClient } from "./client.js";
Expand Down Expand Up @@ -280,28 +284,45 @@ export async function openChannelWorkspace(cwd: string, channelId: string): Prom
return daemonClient.openChannelWorkspace(cwd, channelId);
}

export async function listChannelMessages(
export async function listChannelConversationItems(
cwd: string,
channelId: string,
options: { readonly limit?: number; readonly beforeMessageId?: string; readonly threadId?: string } = {},
): Promise<ChannelMessageListResponse> {
return daemonClient.listChannelMessages(cwd, channelId, options);
options: ListChannelConversationItemsOptions = {},
): Promise<ChannelConversationItemPageResponse> {
return daemonClient.listChannelConversationItems(cwd, channelId, options);
}

export async function createChannelMessage(
export async function createChannelTurn(
cwd: string,
channelId: string,
input: CreateChannelMessageRequest,
): Promise<ChannelMessageMutationResponse> {
return daemonClient.createChannelMessage(cwd, channelId, input);
input: CreateChannelTurnRequest,
): Promise<ChannelConversationTurnResponse> {
return daemonClient.createChannelTurn(cwd, channelId, input);
}

export async function streamChannelMessage(
export async function streamChannelTurn(
cwd: string,
channelId: string,
input: CreateChannelMessageRequest,
input: CreateChannelTurnRequest,
): Promise<Response> {
return daemonClient.streamChannelMessage(cwd, channelId, input);
return daemonClient.streamChannelTurn(cwd, channelId, input);
}

export async function getChannelTurnContext(
cwd: string,
channelId: string,
turnId: string,
): Promise<ChannelConversationContextInspectResponse> {
return daemonClient.getChannelTurnContext(cwd, channelId, turnId);
}

export async function correctChannelTurnContext(
cwd: string,
channelId: string,
turnId: string,
input: CorrectChannelTurnContextRequest,
): Promise<CorrectChannelTurnContextResponse> {
return daemonClient.correctChannelTurnContext(cwd, channelId, turnId, input);
}

export async function listChannelMessageRouteDecisions(
Expand Down
29 changes: 23 additions & 6 deletions apps/web/src/app/channel-message-stream-projection.ts
Original file line number Diff line number Diff line change
@@ -1,14 +1,31 @@
import type {
ChannelConversationTurnResponse,
ChannelMessage,
ChannelMessageStreamBlockedEvent,
ChannelMessageStreamDoneEvent,
ChannelMessageStreamErrorEvent,
ChannelMessageStreamReadyEvent,
ChannelReplyDispatch,
} from "../types.js";

export interface ChannelPresentationStreamReadyEvent {
readonly message: ChannelMessage;
readonly aiReply?: ChannelMessage;
readonly aiReplies?: readonly ChannelMessage[];
}

export interface ChannelPresentationStreamTerminalEvent {
readonly message?: ChannelMessage;
readonly aiReply?: ChannelMessage;
readonly aiReplies?: readonly ChannelMessage[];
readonly replyDispatch?: ChannelReplyDispatch;
readonly autoStartedWorkflow?: ChannelConversationTurnResponse["autoStartedWorkflow"];
readonly error?: string;
}

export interface ChannelPresentationStreamErrorEvent extends ChannelPresentationStreamTerminalEvent {
readonly error: string;
}

export function projectChannelStreamReady(
current: readonly ChannelMessage[],
event: ChannelMessageStreamReadyEvent,
event: ChannelPresentationStreamReadyEvent,
): readonly ChannelMessage[] {
const replies = channelStreamReplies(event);
return replies.length > 0
Expand All @@ -18,7 +35,7 @@ export function projectChannelStreamReady(

export function projectChannelStreamTerminal(
current: readonly ChannelMessage[],
event: ChannelMessageStreamDoneEvent | ChannelMessageStreamBlockedEvent | ChannelMessageStreamErrorEvent,
event: ChannelPresentationStreamTerminalEvent,
): readonly ChannelMessage[] {
const replies = channelStreamReplies(event);
if (event.replyDispatch?.disposition === "followup_queued") {
Expand Down
80 changes: 59 additions & 21 deletions apps/web/src/app/use-channel-actions-controller.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import {
createChannelWorkflowFromHandoff,
createTask,
deleteChannelAttachment,
streamChannelMessage,
streamChannelTurn,
updateChannel,
updateChannelBrief,
updateChannelCollaborationProtocol,
Expand Down Expand Up @@ -48,13 +48,14 @@ import type {
ChannelReplyDisposition,
ConversationActiveReplyTarget,
ChannelMessageFileRef,
ChannelMessageStreamBlockedEvent,
ChannelMessageStreamDeltaEvent,
ChannelMessageStreamDoneEvent,
ChannelMessageStreamErrorEvent,
ChannelMessageStreamReplyPendingEvent,
ChannelMessageStreamReadyEvent,
ChannelMessageStreamRunLinkedEvent,
ChannelConversationStreamBlockedEvent,
ChannelConversationStreamDeltaEvent,
ChannelConversationStreamDoneEvent,
ChannelConversationStreamErrorEvent,
ChannelReplyPendingEvent,
ChannelConversationStreamReplyPendingEvent,
ChannelConversationStreamReadyEvent,
ChannelConversationStreamRunLinkedEvent,
ChannelWorkflow,
ChannelWorkflowStep,
ProviderPreset,
Expand All @@ -80,7 +81,13 @@ import {
import {
projectChannelStreamReady,
projectChannelStreamTerminal,
type ChannelPresentationStreamErrorEvent,
} from "./channel-message-stream-projection.js";
import {
projectChannelConversationTurn,
projectMessageItem,
projectMessageItems,
} from "../lib/channel-conversation-items.js";

export interface ChannelActionsController {
readonly saveChannel: () => Promise<void>;
Expand Down Expand Up @@ -134,7 +141,7 @@ interface SendChannelMessageInput {
readonly requestAiReply: boolean;
readonly replyDisposition?: ChannelReplyDisposition;
readonly onMessageAccepted?: (message: ChannelMessage) => void;
readonly onReplyPending?: (pending: ChannelMessageStreamReplyPendingEvent) => void;
readonly onReplyPending?: (pending: ChannelReplyPendingEvent) => void;
readonly onReplyLinked?: (target: ConversationActiveReplyTarget) => void;
readonly onReplyDispatch?: (dispatch: ChannelReplyDispatch) => void;
}
Expand Down Expand Up @@ -488,8 +495,10 @@ export function useChannelActionsController(input: {
let messageAccepted = false;
let autoStartedWorkflow = false;
let streamErrorMessage = "";
const response = await streamChannelMessage(input.form.cwd.trim(), channelId, {
threadId: messageInput.threadId,
const response = await streamChannelTurn(input.form.cwd.trim(), channelId, {
surface: messageInput.threadId
? { kind: "thread", threadId: messageInput.threadId }
: { kind: "channel_root" },
body: messageInput.body,
mentions: messageInput.mentions,
attachmentIds: messageInput.attachmentIds,
Expand All @@ -501,7 +510,8 @@ export function useChannelActionsController(input: {
});
await readSseResponse(response, (event) => {
if (event.event === "ready") {
const data = JSON.parse(event.data) as ChannelMessageStreamReadyEvent;
const raw = JSON.parse(event.data) as ChannelConversationStreamReadyEvent;
const data = projectChannelConversationTurn(raw);
const aiReplies = data.aiReplies?.length ? data.aiReplies : data.aiReply ? [data.aiReply] : [];
receivedAiReply = receivedAiReply || aiReplies.length > 0;
input.setChannelMessages((current) => projectChannelStreamReady(current, data));
Expand All @@ -512,19 +522,23 @@ export function useChannelActionsController(input: {
return;
}
if (event.event === "reply_pending") {
const data = JSON.parse(event.data) as ChannelMessageStreamReplyPendingEvent;
messageInput.onReplyPending?.(data);
const data = JSON.parse(event.data) as ChannelConversationStreamReplyPendingEvent;
messageInput.onReplyPending?.({
sourceMessageId: data.sourceItemId,
targetTeammateId: data.targetTeammateId,
startedAt: data.startedAt,
});
return;
}
if (event.event === "delta") {
const data = JSON.parse(event.data) as ChannelMessageStreamDeltaEvent;
input.setChannelMessages((current) => current.map((message) => message.id === data.messageId ? { ...message, body: data.body, status: "pending" } : message));
const data = JSON.parse(event.data) as ChannelConversationStreamDeltaEvent;
input.setChannelMessages((current) => current.map((message) => message.id === data.itemId ? { ...message, body: data.body, status: "pending" } : message));
return;
}
if (event.event === "run_linked") {
const data = JSON.parse(event.data) as ChannelMessageStreamRunLinkedEvent;
const data = JSON.parse(event.data) as ChannelConversationStreamRunLinkedEvent;
input.setChannelMessages((current) => current.map((message) =>
message.id === data.messageId ? { ...message, linkedRunId: data.linkedRunId } : message
message.id === data.itemId ? { ...message, linkedRunId: data.linkedRunId } : message
));
if (data.activeReply) {
messageInput.onReplyLinked?.(data.activeReply);
Expand All @@ -537,7 +551,8 @@ export function useChannelActionsController(input: {
return;
}
if (event.event === "done") {
const data = JSON.parse(event.data) as ChannelMessageStreamDoneEvent;
const raw = JSON.parse(event.data) as ChannelConversationStreamDoneEvent;
const data = projectChannelConversationTurn(raw);
const aiReplies = data.aiReplies?.length ? data.aiReplies : data.aiReply ? [data.aiReply] : [];
receivedAiReply = receivedAiReply || aiReplies.length > 0;
autoStartedWorkflow = Boolean(data.autoStartedWorkflow);
Expand All @@ -546,7 +561,8 @@ export function useChannelActionsController(input: {
return;
}
if (event.event === "blocked") {
const data = JSON.parse(event.data) as ChannelMessageStreamBlockedEvent;
const raw = JSON.parse(event.data) as ChannelConversationStreamBlockedEvent;
const data = { ...projectChannelConversationTurn(raw), ...(raw.error ? { error: raw.error } : {}) };
const aiReplies = data.aiReplies?.length ? data.aiReplies : data.aiReply ? [data.aiReply] : [];
receivedAiReply = receivedAiReply || aiReplies.length > 0;
input.setChannelMessages((current) => projectChannelStreamTerminal(current, data));
Expand All @@ -557,7 +573,29 @@ export function useChannelActionsController(input: {
return;
}
if (event.event === "error") {
const data = JSON.parse(event.data) as ChannelMessageStreamErrorEvent;
const raw = JSON.parse(event.data) as ChannelConversationStreamErrorEvent;
const data: ChannelPresentationStreamErrorEvent = raw.userItem && raw.turn && raw.thread
? {
...projectChannelConversationTurn({
thread: raw.thread,
turn: raw.turn,
userItem: raw.userItem,
...(raw.assistantItems ? { assistantItems: raw.assistantItems } : {}),
decorations: raw.decorations ?? [],
...(raw.replyDispatch ? { replyDispatch: raw.replyDispatch } : {}),
}),
error: raw.error,
}
: {
...(raw.userItem
? { message: projectMessageItem(raw.userItem, raw.decorations ?? []) }
: {}),
...(raw.assistantItems
? { aiReplies: projectMessageItems(raw.assistantItems, raw.decorations ?? []) }
: {}),
...(raw.replyDispatch ? { replyDispatch: raw.replyDispatch } : {}),
error: raw.error,
};
const messages = [data.message, ...(data.aiReplies?.length ? data.aiReplies : data.aiReply ? [data.aiReply] : [])]
.filter((message): message is ChannelMessage => Boolean(message));
receivedAiReply = receivedAiReply || messages.some((message) => message.kind === "ai_reply");
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import {
createChannelDecision,
createChannelHandoff,
createChannelLinkPreview,
createChannelMessage,
createChannelTurn,
createChannelWorkflowFromHandoff,
createChannelWorkflowFromTemplate,
createTask,
Expand All @@ -21,7 +21,7 @@ import {
patchChannel,
retryChannelWorkflowStep,
startChannelWorkflowStep,
streamChannelMessage,
streamChannelTurn,
updateChannel,
updateChannelBranch,
updateChannelBrief,
Expand All @@ -48,7 +48,7 @@ export function useWorkspaceCommandController() {
createChannelDecision,
createChannelHandoff,
createChannelLinkPreview,
createChannelMessage,
createChannelTurn,
createChannelWorkflowFromHandoff,
createChannelWorkflowFromTemplate,
createTask,
Expand All @@ -60,7 +60,7 @@ export function useWorkspaceCommandController() {
patchChannel,
retryChannelWorkflowStep,
startChannelWorkflowStep,
streamChannelMessage,
streamChannelTurn,
updateChannel,
updateChannelBranch,
updateChannelBrief,
Expand Down
Loading
Loading