Skip to content
Closed
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
9 changes: 8 additions & 1 deletion apps/mobile/src/features/threads/ThreadQueueControl.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -211,7 +211,7 @@ export function ThreadQueueSheet({ route }: StaticScreenProps<QueueTarget>) {
No messages waiting in this queue.
</Text>
) : null}
{queuedRuns.map(({ run, text, attachments }, index) => {
{queuedRuns.map(({ run, text, attachments, steerRejectionReason }, index) => {
const layout = dragRows?.find((row) => row.id === run.id);
const offset =
sourceLayout?.y !== undefined &&
Expand All @@ -228,6 +228,7 @@ export function ThreadQueueSheet({ route }: StaticScreenProps<QueueTarget>) {
index,
isEditing: editing?.runId === run.id,
queuedCount: queuedRuns.length,
steerRejectionReason,
text,
});
const title =
Expand Down Expand Up @@ -334,6 +335,9 @@ export function ThreadQueueSheet({ route }: StaticScreenProps<QueueTarget>) {
{
id: "steer",
title: workflow?.canSendNow ? "Send now" : "Steer now",
...(!workflow?.canSendNow && steerRejectionReason
? { subtitle: steerRejectionReason }
: {}),
attributes: { disabled: !controls.canSend },
image: Platform.OS === "ios" ? "arrow.turn.left.up" : "arrow_upward",
},
Expand Down Expand Up @@ -396,6 +400,9 @@ export function ThreadQueueSheet({ route }: StaticScreenProps<QueueTarget>) {
? `Send message ${index + 1} now`
: `Steer with message ${index + 1} now`
}
{...(!workflow?.canSendNow && steerRejectionReason
? { accessibilityHint: steerRejectionReason }
: {})}
disabled={!controls.canSend}
onPress={() => void act(run.id, "steer")}
className="h-8 shrink-0 justify-center rounded-full bg-primary px-3 active:opacity-70 disabled:opacity-40"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,28 @@ describe("threadQueueControlPresentation", () => {
canReorder: false,
index: 0,
queuedCount: 1,
text: "Queued message",
steerRejectionReason: "Goal commands must run as a separate turn.",
text: "/goal finish the refactor",
});

expect(controls.canSend).toBe(expected);
},
);

it("rejects steering a goal into an active turn", () => {
const controls = resolveThreadQueueRowControls({
busy: false,
canPromoteToSteer: true,
canSendNow: false,
canReorder: true,
index: 0,
queuedCount: 1,
steerRejectionReason: "Goal commands must run as a separate turn.",
text: "/goal finish the refactor",
});
expect(controls.canSend).toBe(false);
});

it("shows an assistant quote as its quoted text", () => {
const href = formatAssistantCitationHref({
version: 1,
Expand All @@ -48,6 +63,7 @@ describe("threadQueueControlPresentation", () => {
canReorder: true,
index: 0,
queuedCount: 1,
steerRejectionReason: null,
text: `[Assistant quote](${href}) Can we add a backoff here?`,
});

Expand All @@ -65,6 +81,7 @@ describe("threadQueueControlPresentation", () => {
index: 1,
isEditing: true,
queuedCount: 3,
steerRejectionReason: null,
text: "Being edited",
});

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ export function resolveThreadQueueRowControls(input: {
readonly index: number;
readonly isEditing?: boolean;
readonly queuedCount: number;
readonly steerRejectionReason: string | null;
readonly text: string;
}): ThreadQueueRowControls {
const mutationEnabled = !input.busy;
Expand All @@ -34,7 +35,10 @@ export function resolveThreadQueueRowControls(input: {
canEdit: mutationEnabled && !isEditing,
canMoveDown: mutationEnabled && input.canReorder && input.index < input.queuedCount - 1,
canMoveUp: mutationEnabled && input.canReorder && input.index > 0,
canSend: mutationEnabled && (input.canPromoteToSteer || input.canSendNow) && !isEditing,
canSend:
mutationEnabled &&
!isEditing &&
(input.canSendNow || (input.canPromoteToSteer && input.steerRejectionReason === null)),
dismissAccessibilityLabel: REMOVE_QUEUED_MESSAGE_ACCESSIBILITY_LABEL,
// Editing and sending keep the original text so quoted citations stay linked.
displayText: assistantCitationsToPlainText(input.text),
Expand Down
13 changes: 1 addition & 12 deletions apps/server/src/orchestration-v2/CommandPolicy.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import {
type ChatAttachment,
CommandId,
isNativeMaintenanceCommand,
ModelSelection,
type OrchestrationV2Command,
OrchestrationV2ProviderCapabilities,
Expand Down Expand Up @@ -117,17 +117,6 @@ type MessageDispatchMode = Extract<
{ readonly type: "message.dispatch" }
>["dispatchMode"];

export function isNativeMaintenanceCommand(message: {
readonly text: string;
readonly attachments: ReadonlyArray<ChatAttachment>;
readonly context?: import("@t3tools/contracts").OrchestrationMessageContext | undefined;
}): boolean {
return (
message.attachments.length === 0 &&
["/compact", "/logout"].includes(message.text.trim().toLowerCase())
);
}

export function resolveMessageDispatchIntent(
projection: OrchestrationV2ThreadProjection,
requestedMode: MessageDispatchMode,
Expand Down
31 changes: 6 additions & 25 deletions apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import {
ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE,
type ChatAttachment,
CommandId,
isNativeMaintenanceCommand,
isOrchestrationV2WorkActive,
isProviderNativeSubagentThread,
MessageId,
Expand Down Expand Up @@ -53,6 +54,7 @@ import {
type OrchestrationV2TurnItem,
latestProviderTurnForAttempt,
orchestrationV2RunWorkStartedAt,
orchestrationV2SteerRejectionReason,
ProviderInstanceId,
ProviderInteractionMode,
resolveProviderInstanceEnabled,
Expand Down Expand Up @@ -87,11 +89,7 @@ import {
SHARED_WORKSPACE_RESTORE_MESSAGE,
} from "./CheckpointRestoreSafety.ts";
import { CheckpointServiceV2 } from "./CheckpointService.ts";
import {
CommandPolicyV2,
isNativeMaintenanceCommand,
resolveMessageDispatchIntent,
} from "./CommandPolicy.ts";
import { CommandPolicyV2, resolveMessageDispatchIntent } from "./CommandPolicy.ts";
import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts";
import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts";
import { notificationTurnItem } from "@t3tools/provider-core/server/notification";
Expand Down Expand Up @@ -434,13 +432,6 @@ function wakeWorkStartedAt(
return previous === undefined ? {} : { workStartedAt: orchestrationV2RunWorkStartedAt(previous) };
}

function isGoalCommand(message: {
readonly text: string;
readonly attachments: ReadonlyArray<ChatAttachment>;
}): boolean {
return message.attachments.length === 0 && /^\/goal(?:\s|$)/u.test(message.text.trim());
}

const threadPullRequestLinksEqual = Schema.toEquivalence(Schema.NullOr(ThreadLinkedPullRequest));

function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId {
Expand Down Expand Up @@ -4160,22 +4151,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
cause: `Target run ${input.targetRunId} was not found.`,
});
}
if (isNativeMaintenanceCommand(input)) {
const steerRejectionReason = orchestrationV2SteerRejectionReason(input);
if (steerRejectionReason !== null) {
return yield* new OrchestratorDispatchError({
commandId: input.command.commandId,
commandType: input.command.type,
cause:
input.text.trim().toLowerCase() === "/compact"
? "Context compaction must run as a separate turn. Queue it or wait for the active turn to finish."
: "Signing out must run as a separate turn. Queue it or wait for the active turn to finish.",
});
}
if (isGoalCommand(input)) {
return yield* new OrchestratorDispatchError({
commandId: input.command.commandId,
commandType: input.command.type,
cause:
"Goal commands must run as a separate turn. Queue it or wait for the active turn to finish.",
cause: `${steerRejectionReason} Queue it or wait for the active turn to finish.`,
});
}
const targetMessage = input.projection.messages.find(
Expand Down
2 changes: 1 addition & 1 deletion apps/server/src/orchestration-v2/RestartContinuation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
import {
CommandId,
isNativeMaintenanceCommand,
latestProviderTurnForAttempt,
MessageId,
type OrchestrationV2Run,
Expand All @@ -13,7 +14,6 @@ import * as Effect from "effect/Effect";
import type { ProjectionRuntimeRecoveryState } from "./ProjectionStore.ts";

import * as ServerSettings from "../serverSettings.ts";
import { isNativeMaintenanceCommand } from "./CommandPolicy.ts";
import * as ThreadManagementService from "./ThreadManagementService.ts";
import {
awaitingProviderThreadId,
Expand Down
45 changes: 42 additions & 3 deletions apps/web/src/components/chat/QueuedRunsControl.browser.test.tsx
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
import { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
import { serializeAssistantCitation } from "@t3tools/shared/assistantCitations";
import type { ComponentProps } from "react";
import { createRef, type ComponentProps } from "react";
import { expect, it, vi } from "vite-plus/test";
import { page } from "vite-plus/test/browser";

const state = vi.hoisted(() => ({
projection: null as unknown,
workflow: null as unknown,
runCommand: vi.fn(async (_value: unknown) => undefined),
}));

vi.mock(import("@t3tools/client-runtime/state/thread-workflows"), async (importOriginal) => ({
Expand All @@ -21,7 +22,7 @@ vi.mock(import("../../state/entities"), async (importOriginal) => ({

vi.mock(import("../../state/use-atom-command"), async (importOriginal) => ({
...(await importOriginal()),
useAtomCommand: (() => async () => undefined) as never,
useAtomCommand: (() => state.runCommand) as never,
}));

vi.mock(import("../../assets/assetUrls"), async (importOriginal) => ({
Expand All @@ -31,11 +32,12 @@ vi.mock(import("../../assets/assetUrls"), async (importOriginal) => ({
}));

import { render as renderUi } from "../../../test/render";
import { QueuedRunsControl } from "./QueuedRunsControl";
import { QueuedRunsControl, type QueuedRunsControlHandle } from "./QueuedRunsControl";

function setup() {
state.projection = null;
state.workflow = null;
state.runCommand.mockClear();
}

const baseProps = {
Expand Down Expand Up @@ -97,3 +99,40 @@ it("previews assistant quotes as readable text and preserves the citation when e
expect.objectContaining({ text: quoteAndText }),
);
});

it("skips queued messages the server refuses to steer", async () => {
setup();
state.projection = { projection: { messages: [] } };
state.workflow = {
activeRun: { id: "run:active" },
canPromoteToSteer: true,
canReorder: true,
queuedRuns: [
{
run: { id: "run:goal", userMessageId: "message:goal" },
text: "/goal finish the refactor",
attachments: [],
steerRejectionReason: "Goal commands must run as a separate turn.",
},
{
run: { id: "run:plain", userMessageId: "message:plain" },
text: "Also update the docs",
attachments: [],
steerRejectionReason: null,
},
],
};
const ref = createRef<QueuedRunsControlHandle>();
await using _screen = await renderQueue({ ref });

const steerButtons = page.getByRole("button", { name: "Steer", exact: true });
await expect.element(steerButtons.nth(0)).toBeDisabled();
await expect.element(steerButtons.nth(1)).toBeEnabled();

expect(ref.current?.steerNext(false)).toBe(true);
expect(state.runCommand).toHaveBeenCalledExactlyOnceWith(
expect.objectContaining({
input: expect.objectContaining({ queuedRunId: "run:plain", targetRunId: "run:active" }),
}),
);
});
23 changes: 17 additions & 6 deletions apps/web/src/components/chat/QueuedRunsControl.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,9 @@ export function QueuedRunsControl({
const activeRun = workflow?.activeRun ?? null;
const canReorder = workflow?.canReorder === true;
const canSend = workflow?.canSendNow === true || workflow?.canPromoteToSteer === true;
const nextSteerable = queued.find(
({ steerRejectionReason }) => workflow?.canSendNow || steerRejectionReason === null,
);
const queuedImageUrlById = new Map(
queuedImageAttachmentUrls.flatMap((url, index) => {
const resource = queuedImageAttachmentResources[index];
Expand All @@ -119,7 +122,7 @@ export function QueuedRunsControl({
(message) => message.inputIntent === "queued_turn" && !acknowledgedMessageIds.has(message.id),
);
const items = [
...queued.map(({ run, text, attachments }, serverIndex) => ({
...queued.map(({ run, text, attachments, steerRejectionReason }, serverIndex) => ({
key: run.id,
runId: run.id,
messageId: run.userMessageId,
Expand All @@ -133,6 +136,7 @@ export function QueuedRunsControl({
name: attachment.name,
url: queuedImageUrlById.get(attachment.id) ?? null,
})),
steerRejectionReason,
pending: false,
})),
...optimisticQueued.map((message) => ({
Expand All @@ -149,6 +153,7 @@ export function QueuedRunsControl({
name: attachment.name,
url: attachment.previewUrl ?? null,
})),
steerRejectionReason: null,
pending: true,
})),
];
Expand Down Expand Up @@ -193,9 +198,8 @@ export function QueuedRunsControl({

useImperativeHandle(ref, () => ({
steerNext(repeat) {
const next = queued[0];
if (!next || !canSend) return false;
if (!repeat && busyRunId === null) void sendQueuedMessage(next.run.id);
if (!nextSteerable || !canSend) return false;
if (!repeat && busyRunId === null) void sendQueuedMessage(nextSteerable.run.id);
return true;
},
// Declines while a queued message is already being edited so the key keeps
Expand Down Expand Up @@ -433,7 +437,12 @@ export function QueuedRunsControl({
<Button
size="xs"
variant="ghost-muted"
disabled={item.runId === null || busyRunId !== null || !canSend}
disabled={
item.runId === null ||
busyRunId !== null ||
!canSend ||
(!workflow?.canSendNow && item.steerRejectionReason !== null)
}
onClick={() => {
if (item.runId !== null) {
void sendQueuedMessage(item.runId);
Expand All @@ -445,7 +454,9 @@ export function QueuedRunsControl({
</Button>
</TooltipTrigger>
<TooltipPopup>
{`${workflow?.canSendNow ? "Start a turn with this message" : "Send as a steer instead"}${item.serverIndex === 0 && props.steerShortcutLabel ? ` (${props.steerShortcutLabel})` : ""}`}
{!workflow?.canSendNow && item.steerRejectionReason
? item.steerRejectionReason
: `${workflow?.canSendNow ? "Start a turn with this message" : "Send as a steer instead"}${item.runId === nextSteerable?.run.id && props.steerShortcutLabel ? ` (${props.steerShortcutLabel})` : ""}`}
</TooltipPopup>
</Tooltip>
<Tooltip>
Expand Down
23 changes: 23 additions & 0 deletions packages/client-runtime/src/state/threadWorkflows.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,29 @@ describe("thread workflows", () => {
expect(state.canSendNow).toBe(false);
});

it("marks queued runs the server refuses to steer", () => {
const state = deriveThreadQueueWorkflowState({
thread: { id: "thread", activeProviderThreadId: null },
runs: [
{ id: "goal", status: "queued", userMessageId: "message-goal", ordinal: 1 },
{ id: "plain", status: "queued", userMessageId: "message-plain", ordinal: 2 },
],
messages: [
{ id: "message-goal", text: "/goal finish the refactor", attachments: [] },
{ id: "message-plain", text: "Also update the docs", attachments: [] },
],
providerTurns: [],
providerThreads: [],
providerSessions: [],
} as never);

expect(
state.queuedRuns.map(({ run, steerRejectionReason }) => [run.id, steerRejectionReason]),
).toEqual([
["goal", "Goal commands must run as a separate turn."],
["plain", null],
]);
});
it.each(["completed", "interrupted", "cancelled", "failed"])(
"allows sending a held message after a %s turn without a live session",
(status) => {
Expand Down
Loading
Loading