Skip to content

Commit cfdc89b

Browse files
juliusmarmingeclaude
juliusmarminge
and
claude
committed
test(server): idle-resume replays wait for the session release before the next prompt
After an `advance_clock` past the idle timeout, the scenario dispatched the next message right away. The idle release runs on the fiber the clock jump woke, and it yields (background-work probe, child archive) before it removes the live entry. When the next turn's `markBusy` got there first, it interrupted the release, so the turn went to the old process. The Kiro replay agent expected `session/close` and a restart, saw `session/prompt`, and the test hung until Vitest timed out. CPU contention from running four server test files per shard made the losing order show up in CI. Fixtures whose jump releases the session now say so (`releasesProviderSession`), and the scenario awaits the committed `provider-session.updated` stopped event before running later steps. Codex's rollback fixture keeps its shared process across the jump, so it does not opt in. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent 56b6c2a commit cfdc89b

6 files changed

Lines changed: 50 additions & 8 deletions

File tree

‎apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts‎

Lines changed: 37 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import type {
1414
TurnItemId,
1515
} from "@t3tools/contracts";
1616
import * as Clock from "effect/Clock";
17+
import * as Deferred from "effect/Deferred";
1718
import * as Duration from "effect/Duration";
1819
import * as Effect from "effect/Effect";
1920
import * as Fiber from "effect/Fiber";
@@ -37,6 +38,8 @@ export type OrchestratorV2ScenarioStep =
3738
| {
3839
readonly type: "advance_clock";
3940
readonly duration: Duration.Input;
41+
/** Wait until the jump idle-releases this thread's provider session. */
42+
readonly awaitProviderSessionRelease?: ThreadId;
4043
}
4144
| {
4245
readonly type: "await";
@@ -271,9 +274,21 @@ export function runOrchestratorV2Scenario(
271274
const orchestrator = yield* Orchestrator.OrchestratorV2;
272275
const storedEventGroups: Array<ReadonlyArray<OrchestrationV2StoredEvent>> = [];
273276
const observedStoredEvents = yield* Ref.make<Array<OrchestrationV2StoredEvent>>([]);
277+
let releaseWatch:
278+
| { readonly threadId: ThreadId; readonly released: Deferred.Deferred<void> }
279+
| undefined;
274280
yield* orchestrator.streamStoredEvents.pipe(
275-
Stream.runForEach((event) =>
276-
Ref.update(observedStoredEvents, (existing) => [...existing, event]),
281+
Stream.runForEach((stored) =>
282+
Ref.update(observedStoredEvents, (existing) => [...existing, stored]).pipe(
283+
Effect.andThen(
284+
releaseWatch !== undefined &&
285+
stored.event.type === "provider-session.updated" &&
286+
stored.event.threadId === releaseWatch.threadId &&
287+
stored.event.payload.status === "stopped"
288+
? Deferred.succeed(releaseWatch.released, undefined)
289+
: Effect.void,
290+
),
291+
),
277292
),
278293
Effect.forkScoped,
279294
);
@@ -647,9 +662,28 @@ export function runOrchestratorV2Scenario(
647662
);
648663
break;
649664
}
650-
case "advance_clock":
665+
case "advance_clock": {
666+
if (step.awaitProviderSessionRelease === undefined) {
667+
yield* TestClock.adjust(step.duration);
668+
break;
669+
}
670+
// The release runs on a woken fiber; a dispatch before it commits cancels it.
671+
const released = yield* Deferred.make<void>();
672+
releaseWatch = { threadId: step.awaitProviderSessionRelease, released };
651673
yield* TestClock.adjust(step.duration);
674+
const settled = yield* Deferred.await(released).pipe(
675+
Effect.timeoutOption(Duration.millis(SCENARIO_WAIT_DEADLINE_MS)),
676+
Effect.provideService(Clock.Clock, Clock.Clock.defaultValue()),
677+
);
678+
releaseWatch = undefined;
679+
if (Option.isNone(settled)) {
680+
return yield* new OrchestratorV2ScenarioStepError({
681+
scenario: scenario.name,
682+
step: `advance_clock:${step.awaitProviderSessionRelease}:no_provider_session_release`,
683+
});
684+
}
652685
break;
686+
}
653687
case "await":
654688
yield* awaitDispatch(step.key);
655689
break;

‎apps/server/src/orchestration-v2/testkit/fixtures/claude_compact_after_resume_wake/input.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ export function claudeCompactAfterResumeWakeInput(): OrchestratorFixtureInput {
55
return {
66
steps: [
77
{ type: "message", text: CLAUDE_COMPACT_FIRST_PROMPT },
8-
{ type: "advance_clock", duration: "31 minutes" },
8+
{ type: "advance_clock", duration: "31 minutes", releasesProviderSession: true },
99
{ type: "message", text: "/compact" },
1010
],
1111
};

‎apps/server/src/orchestration-v2/testkit/fixtures/claude_idle_resume/input.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ export function claudeIdleResumeInput(): OrchestratorFixtureInput {
77
return {
88
steps: [
99
{ type: "message", text: CLAUDE_IDLE_RESUME_PROMPT_1 },
10-
{ type: "advance_clock", duration: "31 minutes" },
10+
{ type: "advance_clock", duration: "31 minutes", releasesProviderSession: true },
1111
{ type: "message", text: CLAUDE_IDLE_RESUME_PROMPT_2 },
1212
],
1313
};

‎apps/server/src/orchestration-v2/testkit/fixtures/provider_thread_resume/input.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ export function providerThreadResumeInput(): OrchestratorFixtureInput {
99
return {
1010
steps: [
1111
{ type: "message", text: PROVIDER_THREAD_RESUME_FIRST_PROMPT },
12-
{ type: "advance_clock", duration: "31 minutes" },
12+
{ type: "advance_clock", duration: "31 minutes", releasesProviderSession: true },
1313
{ type: "message", text: PROVIDER_THREAD_RESUME_SECOND_PROMPT },
1414
],
1515
};

‎apps/server/src/orchestration-v2/testkit/fixtures/shared.ts‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -268,6 +268,8 @@ export type OrchestratorFixtureInputStep =
268268
| {
269269
readonly type: "advance_clock";
270270
readonly duration: Duration.Input;
271+
/** The jump idle-releases the provider session; later steps wait for that release. */
272+
readonly releasesProviderSession?: boolean;
271273
}
272274
| {
273275
/** Picks another model of the same provider instance; later messages use it. */
@@ -883,7 +885,13 @@ export function materializeFixtureInput(input: {
883885
});
884886
break;
885887
case "advance_clock":
886-
steps.push({ type: "advance_clock", duration: step.duration });
888+
steps.push({
889+
type: "advance_clock",
890+
duration: step.duration,
891+
...(step.releasesProviderSession === true
892+
? { awaitProviderSessionRelease: ids.threadId }
893+
: {}),
894+
});
887895
break;
888896
case "set_model":
889897
modelSelection = { ...input.modelSelection, model: step.model };

‎apps/server/src/orchestration-v2/testkit/fixtures/stop_background_work_after_failed_turn/input.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ export function stopBackgroundWorkAfterReleaseInput(): OrchestratorFixtureInput
1616
return {
1717
steps: [
1818
{ type: "message", text: STOP_BACKGROUND_WORK_AFTER_FAILED_TURN_PROMPT },
19-
{ type: "advance_clock", duration: "31 minutes" },
19+
{ type: "advance_clock", duration: "31 minutes", releasesProviderSession: true },
2020
{ type: "stop_background_work", targetRunIndex: 1 },
2121
],
2222
};

0 commit comments

Comments
 (0)