authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-04 19:23:39-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 19:23:39-07:00
logd7fe47fd0957a34fc2e8bb05f12db19bac3403b6
treefbf5f2b5858f3060d460f6ba988b0707b64a6536
parent03c6cd8badb7af4dfd05d370f3180d16f6927591
signature Signed by PGP key B5690EEEBB952194

fix(server): dismiss native questions when their turn ends (#9851)

2 files changed, 235 insertions(+), 1 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+187
......@@ -20,6 +20,7 @@ import {
2020 type OrchestrationCommand,
2121 ProjectId,
2222 ProviderItemId,
23 RuntimeRequestId,
2324 type ServerSettings,
2425 ThreadId,
2526 TurnId,
......@@ -30,6 +31,7 @@ import * as Effect from "effect/Effect";
3031import * as Exit from "effect/Exit";
3132import * as Layer from "effect/Layer";
3233import * as ManagedRuntime from "effect/ManagedRuntime";
34import * as Option from "effect/Option";
3335import * as PubSub from "effect/PubSub";
3436import * as Scope from "effect/Scope";
3537import * as Stream from "effect/Stream";
......@@ -372,6 +374,12 @@ describe("ProviderRuntimeIngestion", () => {
372374 engine,
373375 dispatch,
374376 readModel: () => testRuntime.runPromise(snapshotQuery.getSnapshot()),
377 readThreadShell: () =>
378 testRuntime.runPromise(
379 snapshotQuery
380 .getThreadShellById(asThreadId("thread-1"))
381 .pipe(Effect.map(Option.getOrThrow)),
382 ),
375383 emit: provider.emit,
376384 emitAndDrain,
377385 sqlCount: sqlCounter.count,
......@@ -2346,6 +2354,185 @@ describe("ProviderRuntimeIngestion", () => {
23462354 expect(message?.streaming).toBe(false);
23472355 });
23482356
2357 function userInputEvent(
2358 turnId: string,
2359 requestId: string,
2360 responseMode?: "message",
2361 ): ProviderRuntimeEvent {
2362 return {
2363 type: "user-input.requested",
2364 eventId: asEventId(`requested:${requestId}`),
2365 provider: ProviderDriverKind.make("codex"),
2366 threadId: asThreadId("thread-1"),
2367 turnId: asTurnId(turnId),
2368 requestId: RuntimeRequestId.make(requestId),
2369 createdAt: "2026-01-01T00:00:01.000Z",
2370 payload: {
2371 ...(responseMode ? { responseMode } : {}),
2372 questions: ["first", "second"].map((id) => ({
2373 id,
2374 header: id,
2375 question: `Choose ${id}`,
2376 options: [{ label: "yes", description: "Continue" }],
2377 multiSelect: false,
2378 })),
2379 },
2380 };
2381 }
2382
2383 it.each(["completed", "interrupted", "failed", "aborted"] as const)(
2384 "resolves native questions when their turn is %s",
2385 async (state) => {
2386 const harness = await createHarness();
2387 const request = userInputEvent("question-turn", "question-request");
2388 await harness.emitAndDrain([
2389 {
2390 type: "turn.started",
2391 eventId: asEventId("question-started"),
2392 provider: request.provider,
2393 threadId: request.threadId,
2394 turnId: request.turnId,
2395 createdAt: request.createdAt,
2396 },
2397 request,
2398 ]);
2399 expect((await harness.readThreadShell()).hasPendingUserInput).toBe(true);
2400 if (state === "interrupted") {
2401 await harness.dispatch({
2402 type: "thread.turn.interrupt",
2403 commandId: CommandId.make("question-interrupt"),
2404 threadId: request.threadId,
2405 createdAt: "2026-01-01T00:00:02.000Z",
2406 });
2407 }
2408 const completed: ProviderRuntimeEvent = {
2409 ...(state === "aborted"
2410 ? ({ type: "turn.aborted", payload: { reason: "Interrupted by user." } } as const)
2411 : ({ type: "turn.completed", payload: { state } } as const)),
2412 eventId: asEventId("question-completed"),
2413 provider: request.provider,
2414 threadId: request.threadId,
2415 turnId: request.turnId,
2416 createdAt: "2026-01-01T00:00:03.000Z",
2417 };
2418 await harness.emitAndDrain([completed]);
2419 const thread = (await harness.readModel()).threads[0]!;
2420 expect(thread.session?.activeTurnId).toBeNull();
2421 expect((await harness.readThreadShell()).hasPendingUserInput).toBe(false);
2422 expect(
2423 thread.activities.filter((activity) => activity.kind === "user-input.resolved"),
2424 ).toMatchObject([{ turnId: request.turnId, payload: { requestId: request.requestId } }]);
2425
2426 await harness.emitAndDrain([completed]);
2427 expect(
2428 (await harness.readModel()).threads[0]!.activities.filter(
2429 (activity) => activity.kind === "user-input.resolved",
2430 ),
2431 ).toHaveLength(1);
2432 },
2433 );
2434
2435 it("keeps a stale request dismissed when its turn later completes", async () => {
2436 const harness = await createHarness();
2437 const request = userInputEvent("stale-turn", "stale-question");
2438 await harness.emitAndDrain([request]);
2439 await harness.dispatch({
2440 type: "thread.activity.append",
2441 commandId: CommandId.make("stale-question-response"),
2442 threadId: request.threadId,
2443 activity: {
2444 id: asEventId("stale-question-failed"),
2445 kind: "provider.user-input.respond.failed",
2446 tone: "error",
2447 summary: "User input response failed",
2448 turnId: request.turnId ?? null,
2449 payload: {
2450 requestId: request.requestId,
2451 detail: "Unknown pending user input request",
2452 },
2453 createdAt: "2026-01-01T00:00:02.000Z",
2454 },
2455 createdAt: "2026-01-01T00:00:02.000Z",
2456 });
2457 expect((await harness.readThreadShell()).hasPendingUserInput).toBe(false);
2458 await harness.emitAndDrain([
2459 {
2460 type: "turn.completed",
2461 eventId: asEventId("stale-turn-completed"),
2462 provider: request.provider,
2463 threadId: request.threadId,
2464 turnId: request.turnId,
2465 createdAt: "2026-01-01T00:00:03.000Z",
2466 payload: { state: "completed" },
2467 },
2468 ]);
2469 expect((await harness.readThreadShell()).hasPendingUserInput).toBe(false);
2470 expect(
2471 (await harness.readModel()).threads[0]!.activities.filter(
2472 (activity) => activity.kind === "user-input.resolved",
2473 ),
2474 ).toMatchObject([{ payload: { requestId: request.requestId } }]);
2475 });
2476
2477 it("preserves answered questions and leaves newer, child and async questions pending", async () => {
2478 const harness = await createHarness();
2479 const answered = userInputEvent("old-turn", "answered-question");
2480 const unresolved = userInputEvent("old-turn", "old-question");
2481 const newer = userInputEvent("new-turn", "new-question");
2482 const child = userInputEvent("child-turn", "child-question");
2483 const asynchronous = userInputEvent("old-turn", "async-question", "message");
2484 const answer: ProviderRuntimeEvent = {
2485 type: "user-input.resolved",
2486 eventId: asEventId("normal-answer"),
2487 provider: answered.provider,
2488 threadId: answered.threadId,
2489 turnId: answered.turnId,
2490 requestId: answered.requestId,
2491 createdAt: "2026-01-01T00:00:02.000Z",
2492 payload: { answers: { first: "yes", second: "yes" } },
2493 };
2494 await harness.emitAndDrain([
2495 answered,
2496 unresolved,
2497 newer,
2498 child,
2499 asynchronous,
2500 answer,
2501 {
2502 type: "turn.started",
2503 eventId: asEventId("new-turn-started"),
2504 provider: newer.provider,
2505 threadId: newer.threadId,
2506 turnId: newer.turnId,
2507 createdAt: "2026-01-01T00:00:03.000Z",
2508 },
2509 ]);
2510 expect((await harness.readThreadShell()).hasPendingUserInput).toBe(true);
2511 await harness.emitAndDrain([
2512 {
2513 type: "turn.completed",
2514 eventId: asEventId("old-turn-completed"),
2515 provider: answered.provider,
2516 threadId: answered.threadId,
2517 turnId: answered.turnId,
2518 createdAt: "2026-01-01T00:00:04.000Z",
2519 payload: { state: "interrupted" },
2520 },
2521 ]);
2522 const thread = (await harness.readModel()).threads[0]!;
2523 expect(thread.session?.activeTurnId).toBe(newer.turnId);
2524 expect((await harness.readThreadShell()).hasPendingUserInput).toBe(true);
2525 expect(
2526 thread.activities.filter((activity) => activity.kind === "user-input.resolved"),
2527 ).toMatchObject([
2528 {
2529 id: answer.eventId,
2530 payload: { requestId: answered.requestId, answers: answer.payload.answers },
2531 },
2532 { turnId: unresolved.turnId, payload: { requestId: unresolved.requestId } },
2533 ]);
2534 });
2535
23492536 it("keeps streaming while an async question is pending", async () => {
23502537 const harness = await createHarness({ serverSettings: { enableLegacyTokenStreaming: true } });
23512538 const base = {
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+48-1
......@@ -36,6 +36,8 @@ import { formatTokens } from "@t3tools/shared/usageFormat";
3636import { ProviderService } from "../../provider/Services/ProviderService.ts";
3737import { ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts";
3838import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts";
39import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts";
40import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts";
3941import * as CheckpointStore from "../../checkpointing/CheckpointStore.ts";
4042import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
4143import { ThreadBackgroundLivenessService } from "../ThreadBackgroundLiveness.ts";
......@@ -955,6 +957,7 @@ const make = Effect.gen(function* () {
955957 const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;
956958 const providerService = yield* ProviderService;
957959 const projectionTurnRepository = yield* ProjectionTurnRepository;
960 const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository;
958961 const serverSettingsService = yield* ServerSettingsService;
959962 const checkpointStore = yield* CheckpointStore.CheckpointStore;
960963 const providerCommandId = (event: ProviderRuntimeEvent, tag: string) =>
......@@ -1931,6 +1934,47 @@ const make = Effect.gen(function* () {
19311934 const proposedPlans = detailedThread?.proposedPlans ?? [];
19321935 const turnId = toTurnId(event.turnId);
19331936 if (turnId) {
1937 const userInputActivities =
1938 yield* projectionThreadActivityRepository.listUserInputLifecycleByThreadId({
1939 threadId: thread.id,
1940 });
1941 const pendingRequestIds = new Set<string>();
1942 for (const activity of userInputActivities) {
1943 const payload =
1944 typeof activity.payload === "object" && activity.payload !== null
1945 ? (activity.payload as Record<string, unknown>)
1946 : null;
1947 const requestId = payload?.requestId;
1948 if (typeof requestId !== "string") continue;
1949 if (
1950 activity.kind === "user-input.requested" &&
1951 activity.turnId === turnId &&
1952 payload?.responseMode !== "message"
1953 ) {
1954 pendingRequestIds.add(requestId);
1955 } else if (activity.kind === "user-input.resolved") {
1956 pendingRequestIds.delete(requestId);
1957 }
1958 }
1959 // A terminal turn cannot accept native callback answers. Message-mode
1960 // questions may outlive that turn and still accept a later user message.
1961 for (const requestId of pendingRequestIds) {
1962 yield* orchestrationEngine.dispatch({
1963 type: "thread.activity.append",
1964 commandId: yield* providerCommandId(event, "terminal-user-input-resolved"),
1965 threadId: thread.id,
1966 activity: {
1967 id: EventId.make(`${event.eventId}:user-input-resolved:${requestId}`),
1968 createdAt: now,
1969 tone: "info",
1970 kind: "user-input.resolved",
1971 summary: "User input dismissed",
1972 payload: { requestId },
1973 turnId,
1974 },
1975 createdAt: now,
1976 });
1977 }
19341978 const assistantMessageIds = yield* getAssistantMessageIdsForTurn(thread.id, turnId);
19351979 yield* Effect.forEach(
19361980 assistantMessageIds,
......@@ -2229,4 +2273,7 @@ const make = Effect.gen(function* () {
22292273export const ProviderRuntimeIngestionLive = Layer.effect(
22302274 ProviderRuntimeIngestionService,
22312275 make,
2232).pipe(Layer.provide(ProjectionTurnRepositoryLive));
2276).pipe(
2277 Layer.provide(ProjectionTurnRepositoryLive),
2278 Layer.provide(ProjectionThreadActivityRepositoryLive),
2279);