Skip to content

Commit 4ce0070

Browse files
rynfarrynfar
authored
Merge pull request #192 from pylon-code/fix/provider-start-reconciliation
fix(server): reconcile stalled provider starts
2 parents e62ad6a + ccbd0e9 commit 4ce0070

51 files changed

Lines changed: 7368 additions & 843 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎apps/server/integration/TestProviderAdapter.integration.ts‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -265,6 +265,9 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
265265
...(input.providerInstanceId !== undefined
266266
? { providerInstanceId: input.providerInstanceId }
267267
: {}),
268+
...(input.sessionIncarnationId !== undefined
269+
? { sessionIncarnationId: input.sessionIncarnationId }
270+
: {}),
268271
status: "ready",
269272
runtimeMode: input.runtimeMode,
270273
threadId,
@@ -316,6 +319,12 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
316319
eventId: nextEventId(input.threadId),
317320
provider,
318321
sessionId: RuntimeSessionId.make(String(input.threadId)),
322+
...(state.session.sessionIncarnationId !== undefined
323+
? { sessionIncarnationId: state.session.sessionIncarnationId }
324+
: {}),
325+
...(input.admissionRequestId !== undefined
326+
? { admissionRequestId: input.admissionRequestId }
327+
: {}),
319328
};
320329
rawEvent.threadId = state.snapshot.threadId;
321330
if (Object.hasOwn(rawEvent, "turnId")) {
@@ -375,6 +384,12 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
375384
createdAt: nowIso(),
376385
threadId: state.snapshot.threadId,
377386
turnId,
387+
...(state.session.sessionIncarnationId !== undefined
388+
? { sessionIncarnationId: state.session.sessionIncarnationId }
389+
: {}),
390+
...(input.admissionRequestId !== undefined
391+
? { admissionRequestId: input.admissionRequestId }
392+
: {}),
378393
payload: {
379394
state: "completed",
380395
},

‎apps/server/integration/orchestrationEngine.integration.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1267,7 +1267,7 @@ it.live("forwards thread.turn.interrupt to claudeAgent provider sessions", () =>
12671267

12681268
const thread = yield* harness.waitForThread(
12691269
THREAD_ID,
1270-
(entry) => entry.session?.threadId === "thread-1",
1270+
(entry) => entry.session?.status === "ready" && entry.session.threadId === "thread-1",
12711271
);
12721272
assert.equal(thread.session?.threadId, "thread-1");
12731273

‎apps/server/integration/providerService.integration.test.ts‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import type { ProviderRuntimeEvent } from "@t3tools/contracts";
2-
import { ProviderDriverKind, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
2+
import { CommandId, ProviderDriverKind, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
33
import { DEFAULT_SERVER_SETTINGS } from "@t3tools/contracts/settings";
44
import * as NodeServices from "@effect/platform-node/NodeServices";
55
import { it, assert } from "@effect/vitest";
@@ -136,6 +136,7 @@ const runTurn = (input: {
136136
readonly threadId: ThreadId;
137137
readonly userText: string;
138138
readonly response: TestTurnResponse;
139+
readonly admissionRequestId?: CommandId;
139140
}) =>
140141
Effect.gen(function* () {
141142
yield* input.harness.queueTurnResponse(input.threadId, input.response);
@@ -146,6 +147,9 @@ const runTurn = (input: {
146147
threadId: input.threadId,
147148
input: input.userText,
148149
attachments: [],
150+
...(input.admissionRequestId !== undefined
151+
? { admissionRequestId: input.admissionRequestId }
152+
: {}),
149153
}),
150154
);
151155
});
@@ -164,13 +168,15 @@ it.live("replays typed runtime fixture events", () =>
164168
runtimeMode: "full-access",
165169
});
166170
assert.equal((session.threadId ?? "").length > 0, true);
171+
const admissionRequestId = CommandId.make("cmd-integration-typed");
167172

168173
const observedEvents = yield* runTurn({
169174
provider,
170175
harness: fixture.harness,
171176
threadId: session.threadId,
172177
userText: "hello",
173178
response: { events: codexTurnTextFixture },
179+
admissionRequestId,
174180
});
175181

176182
assert.deepEqual(
@@ -181,6 +187,14 @@ it.live("replays typed runtime fixture events", () =>
181187
observedEvents.map((event) => event.providerInstanceId),
182188
codexTurnTextFixture.map(() => codexInstanceId),
183189
);
190+
assert.deepEqual(
191+
observedEvents.map((event) => event.admissionRequestId),
192+
codexTurnTextFixture.map(() => admissionRequestId),
193+
);
194+
assert.deepEqual(
195+
observedEvents.map((event) => event.sessionIncarnationId),
196+
codexTurnTextFixture.map(() => session.sessionIncarnationId),
197+
);
184198
}).pipe(Effect.provide(fixture.layer));
185199
}).pipe(Effect.provide(NodeServices.layer)),
186200
);

‎apps/server/scripts/acp-mock-agent.ts‎

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ const emitGenericToolPlaceholders = process.env.T3_ACP_EMIT_GENERIC_TOOL_PLACEHO
2525
const streamCommandChunks = Number(process.env.T3_ACP_STREAM_COMMAND_CHUNKS ?? "0");
2626
const streamCommandChunkChars = Number(process.env.T3_ACP_STREAM_COMMAND_CHUNK_CHARS ?? "64");
2727
const emitAskQuestion = process.env.T3_ACP_EMIT_ASK_QUESTION === "1";
28+
const emitCursorStartupPlan = process.env.T3_ACP_EMIT_CURSOR_STARTUP_PLAN === "1";
2829
const emitXAiAskUserQuestion = process.env.T3_ACP_EMIT_XAI_ASK_USER_QUESTION === "1";
2930
const emitXAiExitPlanMode = process.env.T3_ACP_EMIT_XAI_EXIT_PLAN_MODE === "1";
3031
const emitXAiPlanMdWrite = process.env.T3_ACP_EMIT_XAI_PLAN_MD_WRITE === "1";
@@ -357,11 +358,21 @@ const program = Effect.gen(function* () {
357358
yield* agent.handleAuthenticate(() => Effect.succeed({}));
358359

359360
yield* agent.handleCreateSession(() =>
360-
Effect.succeed({
361-
sessionId,
362-
modes: modeState(),
363-
models: modelState(),
364-
configOptions: configOptions(),
361+
Effect.gen(function* () {
362+
if (emitCursorStartupPlan) {
363+
yield* agent.client.extRequest("cursor/create_plan", {
364+
toolCallId: "startup-plan-tool-call",
365+
name: "Startup plan",
366+
plan: "# Startup plan",
367+
todos: [],
368+
});
369+
}
370+
return {
371+
sessionId,
372+
modes: modeState(),
373+
models: modelState(),
374+
configOptions: configOptions(),
375+
};
365376
}),
366377
);
367378

‎apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts‎

Lines changed: 105 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import {
44
DEFAULT_PROVIDER_INTERACTION_MODE,
55
MessageId,
66
ProjectId,
7+
RuntimeSessionId,
78
ThreadId,
89
TurnId,
910
type OrchestrationEvent,
@@ -243,6 +244,95 @@ describe("OrchestrationEngine", () => {
243244
await runtime.dispose();
244245
});
245246

247+
it("accepts a stale session lifecycle CAS without producing an event", async () => {
248+
const system = await createOrchestrationSystem();
249+
const { engine } = system;
250+
const createdAt = now();
251+
const projectId = asProjectId("project-stale-session-lifecycle");
252+
const threadId = ThreadId.make("thread-stale-session-lifecycle");
253+
const providerInstanceId = ProviderInstanceId.make("codex");
254+
const sessionIncarnationId = RuntimeSessionId.make("session-stale-lifecycle-current");
255+
256+
await system.run(
257+
engine.dispatch({
258+
type: "project.create",
259+
commandId: CommandId.make("cmd-project-stale-session-lifecycle-create"),
260+
projectId,
261+
title: "Stale session lifecycle",
262+
workspaceRoot: "/tmp/project-stale-session-lifecycle",
263+
defaultModelSelection: {
264+
instanceId: providerInstanceId,
265+
model: "gpt-5-codex",
266+
},
267+
createdAt,
268+
}),
269+
);
270+
await system.run(
271+
engine.dispatch({
272+
type: "thread.create",
273+
commandId: CommandId.make("cmd-thread-stale-session-lifecycle-create"),
274+
threadId,
275+
projectId,
276+
title: "Stale lifecycle",
277+
modelSelection: {
278+
instanceId: providerInstanceId,
279+
model: "gpt-5-codex",
280+
},
281+
interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
282+
runtimeMode: "full-access",
283+
branch: null,
284+
worktreePath: null,
285+
createdAt,
286+
}),
287+
);
288+
const currentSession = {
289+
threadId,
290+
status: "ready" as const,
291+
providerName: "codex",
292+
providerInstanceId,
293+
runtimeMode: "full-access" as const,
294+
sessionIncarnationId,
295+
activeTurnId: null,
296+
lastError: null,
297+
updatedAt: createdAt,
298+
};
299+
await system.run(
300+
engine.dispatch({
301+
type: "thread.session.set",
302+
commandId: CommandId.make("cmd-session-stale-lifecycle-set"),
303+
threadId,
304+
session: currentSession,
305+
createdAt,
306+
}),
307+
);
308+
const sequenceBeforeStaleLifecycle = await system.run(engine.latestSequence);
309+
310+
const result = await system.run(
311+
engine.dispatch({
312+
type: "thread.session.apply-lifecycle",
313+
commandId: CommandId.make("cmd-session-stale-lifecycle-apply"),
314+
threadId,
315+
expectedStatus: "running",
316+
expectedProviderInstanceId: providerInstanceId,
317+
expectedSessionIncarnationId: sessionIncarnationId,
318+
expectedPendingTurnRequestId: null,
319+
expectedPendingTurnSessionId: null,
320+
expectedActiveTurnRequestId: null,
321+
expectedActiveTurnId: null,
322+
expectedFailedTurnRequestId: null,
323+
session: currentSession,
324+
createdAt,
325+
}),
326+
);
327+
328+
expect(result).toEqual({ sequence: sequenceBeforeStaleLifecycle, eventCount: 0 });
329+
expect(await system.run(engine.latestSequence)).toBe(sequenceBeforeStaleLifecycle);
330+
expect(
331+
(await system.readModel()).threads.find((thread) => thread.id === threadId)?.session,
332+
).toEqual(currentSession);
333+
await system.dispose();
334+
});
335+
246336
it("persists deterministic read models for repeated snapshot reads", async () => {
247337
const createdAt = now();
248338
const system = await createOrchestrationSystem();
@@ -1006,7 +1096,7 @@ describe("OrchestrationEngine", () => {
10061096
]);
10071097

10081098
const retryResult = await runtime.runPromise(engine.dispatch(turnStartCommand));
1009-
expect(retryResult.sequence).toBe(4);
1099+
expect(retryResult.sequence).toBe(5);
10101100

10111101
const eventsAfterRetry = await runtime.runPromise(
10121102
Stream.runCollect(engine.readEvents(0)).pipe(
@@ -1017,11 +1107,12 @@ describe("OrchestrationEngine", () => {
10171107
"project.created",
10181108
"thread.created",
10191109
"thread.message-sent",
1110+
"thread.session-set",
10201111
"thread.turn-start-requested",
10211112
]);
10221113
expect(
10231114
eventsAfterRetry.filter((event) => event.commandId === turnStartCommand.commandId),
1024-
).toHaveLength(2);
1115+
).toHaveLength(3);
10251116

10261117
await runtime.dispose();
10271118
});
@@ -1288,6 +1379,18 @@ describe("OrchestrationEngine", () => {
12881379
} as const;
12891380

12901381
const first = await system.run(engine.dispatch(turnStart));
1382+
await expect(
1383+
system.run(
1384+
engine.dispatch({
1385+
...turnStart,
1386+
commandId: CommandId.make("cmd-retry-turn-start-conflicting"),
1387+
message: {
1388+
...turnStart.message,
1389+
messageId: asMessageId("msg-retry-conflicting"),
1390+
},
1391+
}),
1392+
),
1393+
).rejects.toThrow("already has pending turn admission");
12911394
const second = await system.run(engine.dispatch(turnStart));
12921395
expect(second.sequence).toBe(first.sequence);
12931396

‎apps/server/src/orchestration/Layers/OrchestrationEngine.ts‎

Lines changed: 17 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -208,26 +208,32 @@ const makeOrchestrationEngine = Effect.gen(function* () {
208208
}
209209

210210
const lastSavedEvent = committedEvents.at(-1) ?? null;
211-
if (lastSavedEvent === null) {
211+
const acceptsStaleNoEvent =
212+
envelope.command.type === "thread.turn.admission.accept" ||
213+
envelope.command.type === "thread.turn.admission.fail" ||
214+
envelope.command.type === "thread.session.bind-pending" ||
215+
envelope.command.type === "thread.session.apply-lifecycle";
216+
if (lastSavedEvent === null && !acceptsStaleNoEvent) {
212217
return yield* new OrchestrationCommandInvariantError({
213218
commandType: envelope.command.type,
214219
detail: "Command produced no events.",
215220
});
216221
}
217-
222+
const resultSequence =
223+
lastSavedEvent?.sequence ?? nextCommandReadModel.snapshotSequence;
218224
yield* commandReceiptRepository.upsert({
219225
commandId: envelope.command.commandId,
220-
aggregateKind: lastSavedEvent.aggregateKind,
221-
aggregateId: lastSavedEvent.aggregateId,
222-
acceptedAt: lastSavedEvent.occurredAt,
223-
resultSequence: lastSavedEvent.sequence,
226+
aggregateKind: lastSavedEvent?.aggregateKind ?? aggregateRef.aggregateKind,
227+
aggregateId: lastSavedEvent?.aggregateId ?? aggregateRef.aggregateId,
228+
acceptedAt: lastSavedEvent?.occurredAt ?? (yield* nowIso),
229+
resultSequence,
224230
status: "accepted",
225231
error: null,
226232
});
227233

228234
return {
229235
committedEvents,
230-
lastSequence: lastSavedEvent.sequence,
236+
lastSequence: resultSequence,
231237
nextCommandReadModel,
232238
} as const;
233239
}),
@@ -256,7 +262,10 @@ const makeOrchestrationEngine = Effect.gen(function* () {
256262
);
257263
}
258264
}
259-
return { sequence: committedCommand.lastSequence };
265+
return {
266+
sequence: committedCommand.lastSequence,
267+
eventCount: committedCommand.committedEvents.length,
268+
};
260269
}).pipe(Effect.withSpan(`orchestration.command.${envelope.command.type}`)),
261270
).pipe(
262271
Effect.flatMap((exit) =>

0 commit comments

Comments
 (0)