diff --git a/src/facade.ts b/src/facade.ts index c239f39..10aab14 100644 --- a/src/facade.ts +++ b/src/facade.ts @@ -16,6 +16,7 @@ export interface NativeThreadSnapshot { readonly status: string; readonly activeTurnId: string | null; }; + readonly latestUserMessageId?: string; readonly latestTurn: { readonly turnId: string; readonly status: string; @@ -269,6 +270,20 @@ function toAgentSnapshot(snapshot: NativeThreadSnapshot): AgentSnapshot { }; } +function matchesMessageIdentity( + snapshot: NativeThreadSnapshot, + messageId: string, +): boolean { + const observedMessageIds = [ + snapshot.latestUserMessageId, + snapshot.latestTurn?.userMessageId, + ].filter((candidate): candidate is string => candidate !== undefined); + return ( + observedMessageIds.length > 0 && + observedMessageIds.every((candidate) => candidate === messageId) + ); +} + function matchesSpawnIdentity( snapshot: NativeThreadSnapshot, threadId: string, @@ -278,7 +293,7 @@ function matchesSpawnIdentity( return ( snapshot.threadId === threadId && snapshot.projectId === projectId && - snapshot.latestTurn?.userMessageId === messageId + matchesMessageIdentity(snapshot, messageId) ); } @@ -604,7 +619,7 @@ export function createT3Facade(runtime: NativeRuntime, options: FacadeOptions) { ); if ( snapshot?.threadId === agentId && - snapshot.latestTurn?.userMessageId === messageId + matchesMessageIdentity(snapshot, messageId) ) { return { agentId, @@ -631,7 +646,7 @@ export function createT3Facade(runtime: NativeRuntime, options: FacadeOptions) { ); if ( finalSnapshot?.threadId !== agentId || - finalSnapshot.latestTurn?.userMessageId !== messageId + !matchesMessageIdentity(finalSnapshot, messageId) ) { throw retryError; } diff --git a/src/nativeRuntime.ts b/src/nativeRuntime.ts index 0430760..bcc9e55 100644 --- a/src/nativeRuntime.ts +++ b/src/nativeRuntime.ts @@ -459,6 +459,9 @@ function toNativeSnapshot( shell: OrchestrationThreadShell, ): NativeThreadSnapshot { const latestTurn = detail.latestTurn; + const latestUserMessage = [...detail.messages] + .reverse() + .find((message) => message.role === "user"); const userMessage = latestTurn === null ? undefined @@ -484,6 +487,9 @@ function toNativeSnapshot( status: session?.status ?? "unknown", activeTurnId: session?.activeTurnId ?? null, }, + ...(latestUserMessage === undefined + ? {} + : { latestUserMessageId: latestUserMessage.id }), latestTurn: latestTurn === null ? null diff --git a/test/facade.send.test.ts b/test/facade.send.test.ts index 154ad12..4c840e7 100644 --- a/test/facade.send.test.ts +++ b/test/facade.send.test.ts @@ -265,10 +265,10 @@ describe("send", () => { projectId: "project-1", snapshotSequence: 22, session: { status: "running", activeTurnId: "turn-2" }, + latestUserMessageId: "message-2", latestTurn: { turnId: "turn-2", status: "running", - userMessageId: "message-2", assistantMessage: null, }, pendingApproval: null, @@ -394,12 +394,14 @@ describe("send", () => { queryCount === 3 ? { status: "running", activeTurnId: "turn-2" } : { status: "ready", activeTurnId: null }, + ...(queryCount === 3 + ? { latestUserMessageId: "message-stable" } + : {}), latestTurn: queryCount === 3 ? { turnId: "turn-2", status: "running", - userMessageId: "message-stable", assistantMessage: null, } : null, @@ -433,6 +435,72 @@ describe("send", () => { }); }); + test("fails closed when ambiguous send identity sources conflict", async () => { + let attempts = 0; + let queryCount = 0; + const runtime = { + async listProjects() { + return []; + }, + async createProject() { + return { sequence: 1 }; + }, + async startThread() { + return { sequence: 2 }; + }, + async startTurn() { + attempts += 1; + throw new AmbiguousDispatchError(); + }, + async getThread(threadId: string) { + queryCount += 1; + if (queryCount === 1) { + return { + threadId, + projectId: "project-1", + snapshotSequence: 20, + session: { status: "ready", activeTurnId: null }, + latestTurn: null, + pendingApproval: null, + pendingInput: null, + }; + } + return { + threadId, + projectId: "project-1", + snapshotSequence: 22, + session: { status: "running", activeTurnId: "turn-2" }, + latestUserMessageId: "message-stable", + latestTurn: { + turnId: "turn-2", + status: "running", + userMessageId: "message-stale", + assistantMessage: null, + }, + pendingApproval: null, + pendingInput: null, + }; + }, + async *subscribeThread() { + return; + }, + }; + const ids = ["command-stable", "message-stable"]; + const facade = createT3Facade(runtime, { + ...DISPATCH_MODES, + id: () => ids.shift()!, + now: () => "2026-07-31T00:00:00.000Z", + }); + + await expect(facade.send("thread-1", "follow-up")).rejects.toMatchObject({ + code: "turn_error", + sequence: 22, + }); + expect(attempts).toBe(1); + expect(queryCount).toBe(2); + expect(ids).toHaveLength(0); + }); + test("fails closed without retry when the first ambiguous-send reconciliation returns a different thread", async () => { let attempts = 0; let queryCount = 0; diff --git a/test/facade.spawn.test.ts b/test/facade.spawn.test.ts index 0028083..5d89d6b 100644 --- a/test/facade.spawn.test.ts +++ b/test/facade.spawn.test.ts @@ -662,6 +662,70 @@ describe("spawn", () => { expect(ids).toHaveLength(0); }); + test("accepts an exact initial message before the provider assigns a turn id", async () => { + const runtime = { + async listProjects() { + return [{ projectId: "project-selected", workspaceRoot: "/work/app" }]; + }, + async createProject() { + return { sequence: 1 }; + }, + async startThread() { + return { sequence: 9 }; + }, + async startTurn() { + return { sequence: 10 }; + }, + async getThread() { + return { + threadId: "thread-stable", + projectId: "project-selected", + snapshotSequence: 9, + session: { status: "starting", activeTurnId: null }, + latestUserMessageId: "message-stable", + latestTurn: null, + pendingApproval: null, + pendingInput: null, + }; + }, + async *subscribeThread() { + return; + }, + }; + const ids = ["thread-stable", "command-stable", "message-stable"]; + const facade = createT3Facade(runtime, { + ...DISPATCH_MODES, + id: () => ids.shift()!, + now: () => "2026-07-31T00:00:00.000Z", + }); + + const snapshot = await facade.spawn({ + workspaceRoot: "/work/app", + title: "worker", + message: "run once", + modelSelection: { + instanceId: "codex", + model: "gpt-5.6-sol", + options: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + + expect(snapshot).toMatchObject({ + agentId: "thread-stable", + projectId: "project-selected", + sequence: 9, + native: { + latestUserMessageId: "message-stable", + latestTurn: null, + }, + }); + expect(ids).toHaveLength(0); + }); + for (const mismatch of [ { label: "thread", @@ -681,6 +745,13 @@ describe("spawn", () => { projectId: "project-selected", userMessageId: "message-other", }, + { + label: "inconsistent message identity", + threadId: "thread-stable", + projectId: "project-selected", + latestUserMessageId: "message-stable", + userMessageId: "message-other", + }, ]) { test(`fails closed when an accepted spawn lookup returns a mismatched ${mismatch.label}`, async () => { const runtime = { @@ -704,6 +775,9 @@ describe("spawn", () => { projectId: mismatch.projectId, snapshotSequence: 9, session: { status: "running", activeTurnId: "turn-1" }, + ...("latestUserMessageId" in mismatch + ? { latestUserMessageId: mismatch.latestUserMessageId } + : {}), latestTurn: { turnId: "turn-1", status: "running", diff --git a/test/native-runtime-adapter.test.ts b/test/native-runtime-adapter.test.ts index bb4d251..e09ab26 100644 --- a/test/native-runtime-adapter.test.ts +++ b/test/native-runtime-adapter.test.ts @@ -522,6 +522,85 @@ describe("T3 native runtime adapter", () => { expect(closes).toBe(1); }); + test("preserves initial user-message identity before a provider turn is assigned", async () => { + const unassignedThread = thread(4, { + latestTurn: null, + messages: [ + { + id: "message-pending", + role: "user", + text: "start", + turnId: null, + streaming: false, + createdAt: NOW, + updatedAt: NOW, + }, + ], + session: { + threadId: "thread-1", + status: "starting", + providerName: "codex", + providerInstanceId: "codex", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: NOW, + }, + } as unknown as Partial); + const unassignedShell = { + ...shellThread(4), + latestTurn: null, + session: unassignedThread.session, + } as unknown as OrchestrationThreadShell; + const runtime = createT3NativeRuntime({ + environmentId: "environment-1", + label: "MacBook Pro", + acquireSocketUrl: async () => "ws://127.0.0.1/ephemeral", + sessionFactory: { + async connect() { + return { + async dispatchCommand() { + return { sequence: 1 }; + }, + subscribeShell() { + return stream([ + { + kind: "snapshot", + snapshot: shellSnapshot(4, [unassignedShell]), + }, + { kind: "synchronized" }, + ]); + }, + subscribeThread() { + return stream([ + { + kind: "snapshot", + snapshot: { + snapshotSequence: 4, + thread: unassignedThread, + }, + }, + { kind: "synchronized" }, + ]); + }, + async close() {}, + }; + }, + }, + }); + + const snapshot = await runtime.getThread("thread-1"); + + expect(snapshot).toMatchObject({ + threadId: "thread-1", + projectId: "project-1", + snapshotSequence: 4, + session: { status: "starting", activeTurnId: null }, + latestUserMessageId: "message-pending", + latestTurn: null, + }); + }); + test("reconciles a synchronized detail snapshot ahead of a compatible shell snapshot", async () => { const runtime = createT3NativeRuntime({ environmentId: "environment-1",