Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down
11 changes: 8 additions & 3 deletions apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"),
Expand Down
Loading