diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index d61739f72c21..6dd2f5758193 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1439,6 +1439,62 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.streaming).toBe(false); }); + it("keeps a usage limit through the limit notice and clears it once work resumes after the reset", async () => { + const harness = await createHarness(); + const limitedAt = "2026-01-01T00:00:00.000Z"; + const resetsAt = "2026-01-01T01:00:00.000Z"; + const emitAssistantText = (id: string, createdAt: string) => { + harness.emit({ + type: "content.delta", + eventId: asEventId(`evt-${id}-delta`), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-2"), + itemId: asItemId(id), + payload: { streamKind: "assistant_text", delta: "text" }, + }); + harness.emit({ + type: "item.completed", + eventId: asEventId(`evt-${id}-completed`), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-2"), + itemId: asItemId(id), + payload: { itemType: "assistant_message", status: "completed" }, + }); + }; + + harness.emit({ + type: "turn.usage-limited", + eventId: asEventId("evt-usage-limited"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: limitedAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-2"), + payload: { resetsAt }, + }); + // Claude streams its "You've hit your session limit" notice right after the limit. + emitAssistantText("limit-notice", limitedAt); + const limited = await waitForThread(harness.readModel, (entry) => + entry.messages.some( + (message: ProviderRuntimeTestMessage) => + message.id === "assistant:limit-notice" && !message.streaming, + ), + ); + expect(limited.usageLimit?.resetsAt).toBe(resetsAt); + + emitAssistantText("resumed-work", "2026-01-01T01:00:30.000Z"); + const resumed = await waitForThread(harness.readModel, (entry) => + entry.messages.some( + (message: ProviderRuntimeTestMessage) => + message.id === "assistant:resumed-work" && !message.streaming, + ), + ); + expect(resumed.usageLimit ?? null).toBeNull(); + }); + it("streams reasoning deltas into a finalized reasoning message", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 3542b112f1b1..046135b1d95f 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -1973,13 +1973,18 @@ const make = Effect.gen(function* () { }); } else if ( thread.usageLimit != null && - ((event.type === "content.delta" && event.payload.streamKind === "assistant_text") || + // Claude streams its limit notice as assistant text, so only output + // after the reset shows a parked turn carrying on. + ((event.type === "content.delta" && + event.payload.streamKind === "assistant_text" && + thread.usageLimit.resetsAt !== null && + Date.parse(now) >= Date.parse(thread.usageLimit.resetsAt)) || (event.type === "turn.completed" && shouldApplyThreadLifecycle && normalizeRuntimeTurnState(event.payload.state) === "completed")) ) { - // A parked turn the provider carried on by itself no longer needs a resume, - // and must not be interrupted by one. + // A parked turn the provider carried on by itself no longer needs a + // resume, and must not be interrupted by one. yield* orchestrationEngine.dispatch({ type: "thread.usage-limit.set", commandId: yield* providerCommandId(event, "usage-limit-clear"),