Skip to content

fix: preserve pre-turn spawn identity #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
EtanHey merged 2 commits into from
Jul 31, 2026
Merged

fix: preserve pre-turn spawn identity #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
524dec3
fix(t3layer): preserve pre-turn spawn identity
EtanHey Jul 31, 2026
1adf16e
fix(t3layer): recover ambiguous native sends
EtanHey Jul 31, 2026
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
21 changes: 18 additions & 3 deletions src/facade.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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)
);
}
Comment on lines +273 to +285

@macroscopeapp macroscopeapp Bot Jul 31, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High src/facade.ts:273

matchesMessageIdentity requires latestUserMessageId and latestTurn.userMessageId to be equal, but during ambiguous send recovery the new user message can be durable while latestTurn still reflects the previous completed turn. In that projection window latestUserMessageId matches the retried command but latestTurn.userMessageId identifies the prior turn, so the .every() check returns false and rejects a dispatch that actually succeeded, surfacing the ambiguous error instead of a recovered receipt. Consider not treating the prior turn's userMessageId as conflicting evidence for send — for example, matching when any candidate equals messageId rather than requiring all to agree.

Suggested change
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 matchesMessageIdentity(
snapshot: NativeThreadSnapshot,
messageId: string,
): boolean {
const observedMessageIds = [
snapshot.latestUserMessageId,
snapshot.latestTurn?.userMessageId,
].filter((candidate): candidate is string => candidate !== undefined);
return observedMessageIds.some((candidate) => candidate === messageId);
}
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: 
In file @src/facade.ts around lines 273-285:

`matchesMessageIdentity` requires `latestUserMessageId` and `latestTurn.userMessageId` to be equal, but during ambiguous `send` recovery the new user message can be durable while `latestTurn` still reflects the previous completed turn. In that projection window `latestUserMessageId` matches the retried command but `latestTurn.userMessageId` identifies the prior turn, so the `.every()` check returns `false` and rejects a dispatch that actually succeeded, surfacing the ambiguous error instead of a recovered receipt. Consider not treating the prior turn's `userMessageId` as conflicting evidence for `send` — for example, matching when any candidate equals `messageId` rather than requiring all to agree.


function matchesSpawnIdentity(
snapshot: NativeThreadSnapshot,
threadId: string,
Expand All @@ -278,7 +293,7 @@ function matchesSpawnIdentity(
return (
snapshot.threadId === threadId &&
snapshot.projectId === projectId &&
snapshot.latestTurn?.userMessageId === messageId
matchesMessageIdentity(snapshot, messageId)
);
}

Expand Down Expand Up @@ -604,7 +619,7 @@ export function createT3Facade(runtime: NativeRuntime, options: FacadeOptions) {
);
if (
snapshot?.threadId === agentId &&
snapshot.latestTurn?.userMessageId === messageId
matchesMessageIdentity(snapshot, messageId)
) {
return {
agentId,
Expand All @@ -631,7 +646,7 @@ export function createT3Facade(runtime: NativeRuntime, options: FacadeOptions) {
);
if (
finalSnapshot?.threadId !== agentId ||
finalSnapshot.latestTurn?.userMessageId !== messageId
!matchesMessageIdentity(finalSnapshot, messageId)
) {
throw retryError;
}
Expand Down
6 changes: 6 additions & 0 deletions src/nativeRuntime.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -484,6 +487,9 @@ function toNativeSnapshot(
status: session?.status ?? "unknown",
activeTurnId: session?.activeTurnId ?? null,
},
...(latestUserMessage === undefined
? {}
: { latestUserMessageId: latestUserMessage.id }),
latestTurn:
latestTurn === null
? null
Expand Down
72 changes: 70 additions & 2 deletions test/facade.send.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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;
Expand Down
74 changes: 74 additions & 0 deletions test/facade.spawn.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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 = {
Expand All @@ -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",
Expand Down
79 changes: 79 additions & 0 deletions test/native-runtime-adapter.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<OrchestrationThread>);
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",
Expand Down