diff --git a/apps/mobile/app.config.ts b/apps/mobile/app.config.ts index ef80e4b8212..79f46461f93 100644 --- a/apps/mobile/app.config.ts +++ b/apps/mobile/app.config.ts @@ -125,7 +125,7 @@ const config: ExpoConfig = { userInterfaceStyle: "automatic", updates: { enabled: true, - url: "https://u.expo.dev/d763fcb8-d37c-41ea-a773-b54a0ab4a454", + url: "https://u.expo.dev/c65ac46d-6488-49af-b61e-ab9bef78f96e", checkAutomatically: "ON_LOAD", fallbackToCacheTimeout: 0, }, @@ -285,10 +285,10 @@ const config: ExpoConfig = { tracesToken: repoEnv.EXPO_PUBLIC_OTLP_TRACES_TOKEN ?? null, }, eas: { - projectId: "d763fcb8-d37c-41ea-a773-b54a0ab4a454", + projectId: "c65ac46d-6488-49af-b61e-ab9bef78f96e", }, }, - owner: "pingdotgg", + owner: "quicksaver", }; export default config; diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 8aeacd870cc..6bc3bfd335c 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -1,12 +1,14 @@ import * as NodeAssert from "node:assert/strict"; -import { it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Schema from "effect/Schema"; -import { describe } from "vite-plus/test"; -import { ThreadId } from "@t3tools/contracts"; +import * as TestClock from "effect/testing/TestClock"; +import { describe, it } from "@effect/vitest"; +import { ThreadId, TurnId } from "@t3tools/contracts"; import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexRpc from "effect-codex-app-server/rpc"; +import type * as EffectCodexSchema from "effect-codex-app-server/schema"; import { CODEX_DEFAULT_MODE_DEVELOPER_INSTRUCTIONS, @@ -14,9 +16,12 @@ import { } from "../CodexDeveloperInstructions.ts"; import { buildTurnStartParams, + findActiveCodexTurnId, hasConfiguredMcpServer, isRecoverableThreadResumeError, openCodexThread, + resolveCodexInterruptTurnId, + shouldPreferActiveCodexTurnCandidate, } from "./CodexSessionRuntime.ts"; const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); @@ -60,6 +65,27 @@ function makeThreadOpenResponse( } as unknown as CodexRpc.ClientRequestResponsesByMethod["thread/start"]; } +function makeThreadReadResponse( + turns: EffectCodexSchema.V2ThreadReadResponse["thread"]["turns"], +): EffectCodexSchema.V2ThreadReadResponse { + return { + thread: { + cliVersion: "0.0.0-test", + createdAt: 1, + cwd: "/tmp/project", + ephemeral: false, + id: "provider-thread-1", + modelProvider: "openai", + preview: "test thread", + sessionId: "session-1", + source: "appServer", + status: { type: "active", activeFlags: [] }, + turns, + updatedAt: 2, + }, + }; +} + describe("buildTurnStartParams", () => { it("keeps invalid turn values only in the schema cause", () => { const secret = "codex-turn-input-secret-sentinel"; @@ -194,6 +220,142 @@ describe("buildTurnStartParams", () => { }); }); +describe("findActiveCodexTurnId", () => { + it("selects the most recently started in-progress turn", () => { + const snapshot = makeThreadReadResponse([ + { id: "turn-active-new", status: "inProgress", startedAt: 30, items: [] }, + { id: "turn-completed", status: "completed", startedAt: 20, items: [] }, + { id: "turn-active-old", status: "inProgress", startedAt: 10, items: [] }, + ]); + + NodeAssert.equal(findActiveCodexTurnId(snapshot), "turn-active-new"); + }); + + it("selects a later in-progress turn without a start timestamp", () => { + const snapshot = makeThreadReadResponse([ + { id: "turn-active-old", status: "inProgress", startedAt: 10, items: [] }, + { id: "turn-active-new", status: "inProgress", items: [] }, + ]); + + NodeAssert.equal(findActiveCodexTurnId(snapshot), "turn-active-new"); + }); + + it("selects a later timestamped turn after one without a timestamp", () => { + const snapshot = makeThreadReadResponse([ + { id: "turn-active-old", status: "inProgress", items: [] }, + { id: "turn-active-new", status: "inProgress", startedAt: 10, items: [] }, + ]); + + NodeAssert.equal(findActiveCodexTurnId(snapshot), "turn-active-new"); + }); + + it("returns undefined when no turn is active", () => { + const response = makeThreadReadResponse([]); + NodeAssert.equal(findActiveCodexTurnId(response), undefined); + }); + + it.effect("requests turns when resolving an interrupt without a projected turn id", () => { + let requestedParams: CodexRpc.ClientRequestParamsByMethod["thread/read"] | undefined; + + return Effect.gen(function* () { + const turnId = yield* resolveCodexInterruptTurnId({ + providerThreadId: "provider-thread-1", + requestedTurnId: undefined, + sessionActiveTurnId: undefined, + readThread: (params) => { + requestedParams = params; + return Effect.succeed( + makeThreadReadResponse([ + { id: "turn-active", status: "inProgress", startedAt: 10, items: [] }, + ]), + ); + }, + }); + + NodeAssert.deepStrictEqual(requestedParams, { + threadId: "provider-thread-1", + includeTurns: true, + }); + NodeAssert.equal(turnId, "turn-active"); + }); + }); + + it.effect("does not revive a stale projected turn after a successful empty read", () => + Effect.gen(function* () { + const turnId = yield* resolveCodexInterruptTurnId({ + providerThreadId: "provider-thread-1", + requestedTurnId: undefined, + sessionActiveTurnId: TurnId.make("turn-stale"), + readThread: () => Effect.succeed(makeThreadReadResponse([])), + }); + + NodeAssert.equal(turnId, undefined); + }), + ); + + it.effect("falls back to the projected turn when the live lookup fails", () => + Effect.gen(function* () { + const projectedTurnId = TurnId.make("turn-projected"); + const turnId = yield* resolveCodexInterruptTurnId({ + providerThreadId: "provider-thread-1", + requestedTurnId: undefined, + sessionActiveTurnId: projectedTurnId, + readThread: () => Effect.fail("lookup failed"), + }); + + NodeAssert.equal(turnId, projectedTurnId); + }), + ); + + it.effect("bounds the live lookup and falls back to the projected turn on timeout", () => + Effect.gen(function* () { + const projectedTurnId = TurnId.make("turn-projected"); + const resolution = yield* resolveCodexInterruptTurnId({ + providerThreadId: "provider-thread-1", + requestedTurnId: undefined, + sessionActiveTurnId: projectedTurnId, + readThread: () => Effect.never, + }).pipe(Effect.forkScoped); + + yield* Effect.yieldNow; + yield* TestClock.adjust("2 seconds"); + NodeAssert.equal(yield* Fiber.join(resolution), projectedTurnId); + }), + ); +}); + +describe("shouldPreferActiveCodexTurnCandidate", () => { + it("selects the first candidate", () => { + NodeAssert.equal(shouldPreferActiveCodexTurnCandidate({ startedAt: 10 }, undefined), true); + }); + + it("orders timestamped turns by start time and lets a later equal entry win", () => { + NodeAssert.equal( + shouldPreferActiveCodexTurnCandidate({ startedAt: 20 }, { startedAt: 10 }), + true, + ); + NodeAssert.equal( + shouldPreferActiveCodexTurnCandidate({ startedAt: 10 }, { startedAt: 20 }), + false, + ); + NodeAssert.equal( + shouldPreferActiveCodexTurnCandidate({ startedAt: 10 }, { startedAt: 10 }), + true, + ); + }); + + it("lets the later provider entry win when either timestamp is absent", () => { + for (const [candidate, selected] of [ + [{}, { startedAt: 10 }], + [{ startedAt: null }, { startedAt: 10 }], + [{ startedAt: 10 }, {}], + [{ startedAt: 10 }, { startedAt: null }], + ] as const) { + NodeAssert.equal(shouldPreferActiveCodexTurnCandidate(candidate, selected), true); + } + }); +}); + describe("T3 browser developer instructions", () => { it("prefers the product-native preview tools in both collaboration modes", () => { for (const instructions of [ diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 99ac498f0c3..d46f063386d 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -54,6 +54,7 @@ const BENIGN_ERROR_LOG_SNIPPETS = [ "state db record_discrepancy: find_thread_path_by_id_str_in_subdir, falling_back", ]; const CODEX_APP_SERVER_FORCE_KILL_AFTER = "2 seconds" as const; +const CODEX_INTERRUPT_THREAD_READ_TIMEOUT = "2 seconds" as const; const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [ "not found", "missing thread", @@ -692,6 +693,75 @@ function parseThreadSnapshot( }; } +type CodexTurnOrderingCandidate = Pick< + EffectCodexSchema.V2ThreadReadResponse["thread"]["turns"][number], + "startedAt" +>; + +export function shouldPreferActiveCodexTurnCandidate( + candidate: CodexTurnOrderingCandidate, + selected: CodexTurnOrderingCandidate | undefined, +): boolean { + if (selected === undefined) { + return true; + } + + // When either timestamp is absent, provider response order is authoritative. + // The caller scans in response order, so the later candidate replaces the selection. + if (candidate.startedAt == null || selected.startedAt == null) { + return true; + } + + return candidate.startedAt >= selected.startedAt; +} + +export function findActiveCodexTurnId( + response: EffectCodexSchema.V2ThreadReadResponse, +): TurnId | undefined { + let activeTurn: EffectCodexSchema.V2ThreadReadResponse["thread"]["turns"][number] | undefined; + for (const turn of response.thread.turns) { + if (turn.status !== "inProgress") { + continue; + } + if (shouldPreferActiveCodexTurnCandidate(turn, activeTurn)) { + activeTurn = turn; + } + } + return activeTurn === undefined ? undefined : TurnId.make(activeTurn.id); +} + +export function resolveCodexInterruptTurnId(input: { + readonly providerThreadId: string; + readonly requestedTurnId: TurnId | undefined; + readonly sessionActiveTurnId: TurnId | undefined; + readonly readThread: ( + params: CodexRpc.ClientRequestParamsByMethod["thread/read"], + ) => Effect.Effect; +}): Effect.Effect { + if (input.requestedTurnId !== undefined) { + return Effect.succeed(input.requestedTurnId); + } + + return input + .readThread({ + threadId: input.providerThreadId, + includeTurns: true, + }) + .pipe( + Effect.timeout(CODEX_INTERRUPT_THREAD_READ_TIMEOUT), + Effect.map(findActiveCodexTurnId), + Effect.tapError((cause) => + Effect.logWarning("Failed to resolve active Codex turn before interrupt.", { + providerThreadId: input.providerThreadId, + cause, + }), + ), + // A failed lookup can still use the locally projected id. A successful + // lookup with no active turn must not revive a stale local id. + Effect.orElseSucceed(() => input.sessionActiveTurnId), + ); +} + export const makeCodexSessionRuntime = ( options: CodexSessionRuntimeOptions, ): Effect.Effect< @@ -1316,7 +1386,12 @@ export const makeCodexSessionRuntime = ( Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; const session = yield* Ref.get(sessionRef); - const effectiveTurnId = turnId ?? session.activeTurnId; + const effectiveTurnId = yield* resolveCodexInterruptTurnId({ + providerThreadId, + requestedTurnId: turnId, + sessionActiveTurnId: session.activeTurnId, + readThread: (params) => client.request("thread/read", params), + }); if (!effectiveTurnId) { return; } diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index 0a0103df183..bd55403036a 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -1,4 +1,11 @@ -import { EnvironmentId, ProjectId, ProviderInstanceId, ThreadId, TurnId } from "@t3tools/contracts"; +import { + EnvironmentId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, +} from "@t3tools/contracts"; import { describe, expect, it } from "vite-plus/test"; import type { Thread } from "../types"; @@ -360,6 +367,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { phase: "ready", latestTurn: completedTurn, session: readySession, + latestUserMessageId: null, hasPendingApproval: false, hasPendingUserInput: false, threadError: null, @@ -385,6 +393,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { phase: "ready", latestTurn: newerTurn, session: { ...readySession, updatedAt: newerTurn.completedAt }, + latestUserMessageId: null, hasPendingApproval: false, hasPendingUserInput: false, threadError: null, @@ -415,6 +424,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { status: "running", activeTurnId: TurnId.make("turn-other"), }, + latestUserMessageId: null, hasPendingApproval: false, hasPendingUserInput: false, threadError: null, @@ -430,6 +440,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { status: "running", activeTurnId: runningTurn.turnId, }, + latestUserMessageId: null, hasPendingApproval: false, hasPendingUserInput: false, threadError: null, @@ -444,6 +455,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { phase: "ready" as const, latestTurn: null, session: null, + latestUserMessageId: null, hasPendingApproval: false, hasPendingUserInput: false, threadError: null, @@ -453,4 +465,94 @@ describe("hasServerAcknowledgedLocalDispatch", () => { expect(hasServerAcknowledgedLocalDispatch({ ...common, hasPendingUserInput: true })).toBe(true); expect(hasServerAcknowledgedLocalDispatch({ ...common, threadError: "failed" })).toBe(true); }); + + it("acknowledges a steer when its user message is projected onto the running thread", () => { + const initialMessageId = MessageId.make("message-initial"); + const steerMessageId = MessageId.make("message-steer"); + const runningTurn = { + ...completedTurn, + state: "running" as const, + completedAt: null, + }; + const runningSession = { + ...readySession, + status: "running" as const, + activeTurnId: runningTurn.turnId, + }; + const localDispatch = createLocalDispatchSnapshot( + makeThread({ + messages: [ + { + id: initialMessageId, + role: "user", + text: "start", + turnId: runningTurn.turnId, + streaming: false, + createdAt: now, + updatedAt: now, + }, + ], + latestTurn: runningTurn, + session: runningSession, + }), + { expectedUserMessageId: steerMessageId }, + ); + + expect( + hasServerAcknowledgedLocalDispatch({ + localDispatch, + phase: "running", + latestTurn: runningTurn, + session: runningSession, + latestUserMessageId: initialMessageId, + hasPendingApproval: false, + hasPendingUserInput: false, + threadError: null, + }), + ).toBe(false); + + expect( + hasServerAcknowledgedLocalDispatch({ + localDispatch, + phase: "running", + latestTurn: runningTurn, + session: runningSession, + latestUserMessageId: steerMessageId, + hasPendingApproval: false, + hasPendingUserInput: false, + threadError: null, + }), + ).toBe(true); + + const alreadyProjected = createLocalDispatchSnapshot( + makeThread({ + messages: [ + { + id: steerMessageId, + role: "user", + text: "steer", + turnId: runningTurn.turnId, + streaming: false, + createdAt: now, + updatedAt: now, + }, + ], + latestTurn: runningTurn, + session: runningSession, + }), + { expectedUserMessageId: MessageId.make("message-next-steer") }, + ); + expect( + hasServerAcknowledgedLocalDispatch({ + localDispatch: alreadyProjected, + phase: "running", + latestTurn: runningTurn, + session: runningSession, + latestUserMessageId: steerMessageId, + hasPendingApproval: false, + hasPendingUserInput: false, + threadError: null, + }), + ).toBe(false); + }); }); diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 705793ec77e..2a7c051ddb6 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -1,6 +1,7 @@ import { type EnvironmentId, isProviderDriverKind, + type MessageId, ProjectId, type ModelSelection, type ProviderDriverKind, @@ -387,6 +388,7 @@ export async function waitForStartedServerThread( export interface LocalDispatchSnapshot { startedAt: string; preparingWorktree: boolean; + expectedUserMessageId: MessageId | null; latestTurnTurnId: TurnId | null; latestTurnRequestedAt: string | null; latestTurnStartedAt: string | null; @@ -395,15 +397,28 @@ export interface LocalDispatchSnapshot { sessionUpdatedAt: string | null; } +export function getLatestUserMessageId( + messages: ReadonlyArray>, +): MessageId | null { + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]; + if (message?.role === "user") { + return message.id; + } + } + return null; +} + export function createLocalDispatchSnapshot( activeThread: Thread | undefined, - options?: { preparingWorktree?: boolean }, + options?: { preparingWorktree?: boolean; expectedUserMessageId?: MessageId }, ): LocalDispatchSnapshot { const latestTurn = activeThread?.latestTurn ?? null; const session = activeThread?.session ?? null; return { startedAt: new Date().toISOString(), preparingWorktree: Boolean(options?.preparingWorktree), + expectedUserMessageId: options?.expectedUserMessageId ?? null, latestTurnTurnId: latestTurn?.turnId ?? null, latestTurnRequestedAt: latestTurn?.requestedAt ?? null, latestTurnStartedAt: latestTurn?.startedAt ?? null, @@ -418,6 +433,7 @@ export function hasServerAcknowledgedLocalDispatch(input: { phase: SessionPhase; latestTurn: Thread["latestTurn"] | null; session: Thread["session"] | null; + latestUserMessageId: MessageId | null; hasPendingApproval: boolean; hasPendingUserInput: boolean; threadError: string | null | undefined; @@ -429,6 +445,18 @@ export function hasServerAcknowledgedLocalDispatch(input: { return true; } + // A prompt sent while a turn is already running is a steer. Providers can + // apply that prompt to the existing turn without opening a new one, so the + // latest turn/session fields may remain unchanged until the agent finishes. + // The projected user message is the server acknowledgement in that case. + if ( + input.localDispatch.sessionStatus === "running" && + input.localDispatch.expectedUserMessageId !== null && + input.localDispatch.expectedUserMessageId === input.latestUserMessageId + ) { + return true; + } + const latestTurn = input.latestTurn ?? null; const session = input.session ?? null; const latestTurnChanged = diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index f5ea5bb1eba..5eca7f50aa6 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -220,6 +220,7 @@ import { collectUserMessageBlobPreviewUrls, createLocalDispatchSnapshot, deriveComposerSendState, + getLatestUserMessageId, hasServerAcknowledgedLocalDispatch, getStartedThreadModelChangeBlockReason, LAST_INVOKED_SCRIPT_BY_PROJECT_KEY, @@ -368,6 +369,10 @@ function useLocalDispatchState(input: { threadError: string | null | undefined; }) { const [localDispatch, setLocalDispatch] = useState(null); + const latestUserMessageId = useMemo( + () => getLatestUserMessageId(input.activeThread?.messages ?? []), + [input.activeThread?.messages], + ); const resetLocalDispatch = useCallback(() => { setLocalDispatch(null); @@ -380,6 +385,7 @@ function useLocalDispatchState(input: { phase: input.phase, latestTurn: input.activeLatestTurn, session: input.activeThread?.session ?? null, + latestUserMessageId, hasPendingApproval: input.activePendingApproval !== null, hasPendingUserInput: input.activePendingUserInput !== null, threadError: input.threadError, @@ -391,12 +397,13 @@ function useLocalDispatchState(input: { input.activeThread?.session, input.phase, input.threadError, + latestUserMessageId, localDispatch, ], ); const activeLocalDispatch = serverAcknowledgedLocalDispatch ? null : localDispatch; const beginLocalDispatch = useCallback( - (options?: { preparingWorktree?: boolean }) => { + (options?: { preparingWorktree?: boolean; expectedUserMessageId?: MessageId }) => { const preparingWorktree = Boolean(options?.preparingWorktree); setLocalDispatch((current) => { const active = serverAcknowledgedLocalDispatch ? null : current; @@ -3980,9 +3987,6 @@ function ChatViewContent(props: ChatViewProps) { return; } - sendInFlightRef.current = true; - beginLocalDispatch({ preparingWorktree: Boolean(baseBranchForWorktree) }); - const composerImagesSnapshot = [...composerImages]; const composerTerminalContextsSnapshot = [...sendableComposerTerminalContexts]; const composerElementContextsSnapshot = [...composerElementContexts]; @@ -4001,6 +4005,11 @@ function ChatViewContent(props: ChatViewProps) { composerReviewCommentsSnapshot, ); const messageIdForSend = newMessageId(); + sendInFlightRef.current = true; + beginLocalDispatch({ + preparingWorktree: Boolean(baseBranchForWorktree), + expectedUserMessageId: messageIdForSend, + }); const messageCreatedAt = new Date().toISOString(); const outgoingMessageText = formatOutgoingPrompt({ provider: ctxSelectedProvider, @@ -4162,7 +4171,7 @@ function ChatViewContent(props: ChatViewProps) { : {}), } : undefined; - beginLocalDispatch({ preparingWorktree: false }); + beginLocalDispatch({ preparingWorktree: false, expectedUserMessageId: messageIdForSend }); const startResult = await startThreadTurn({ environmentId, input: { @@ -4462,7 +4471,7 @@ function ChatViewContent(props: ChatViewProps) { }); sendInFlightRef.current = true; - beginLocalDispatch({ preparingWorktree: false }); + beginLocalDispatch({ preparingWorktree: false, expectedUserMessageId: messageIdForSend }); setThreadError(threadIdForSend, null); // Position this sent row once LegendList has measured the anchored tail.