Skip to content

fix(server): preserve usage-limit recovery across empty wake runs #22775

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
juliusmarminge wants to merge 8 commits into
base: main
Choose a base branch
from
+1,108 −36
Open

fix(server): preserve usage-limit recovery across empty wake runs #22775

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8f92940
test(server): takeover tests for usage-limit recovery after backgroun…
juliusmarminge Oct 21, 2026
34a6a96
fix(server): preserve limit recovery after completed background wakes
juliusmarminge Oct 21, 2026
80d27e9
fix: align usage-limit recovery controls after background wakes
juliusmarminge Oct 21, 2026
9fd419a
test(server): add synthetic usage-limit evidence capture
juliusmarminge Oct 21, 2026
fdbd4b4
fix(server): restore usage-limit recovery CI checks
juliusmarminge Oct 21, 2026
866e273
Merge origin/main into fix/issue-16815
juliusmarminge Oct 21, 2026
e8fbc9b
Merge branch 'main' into fix/issue-16815
juliusmarminge Oct 22, 2026
824b0ba
test(server): add usage-limit recovery web evidence scripts
juliusmarminge Oct 22, 2026
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
28 changes: 17 additions & 11 deletions apps/server/src/orchestration-v2/Orchestrator.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1041,7 +1041,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
"contextTransfers",
],
{
turnItemTypes: ["user_message", "error"],
turnItemTypes: ["user_message", "error", "notification"],
messageRoles: ["user"],
omitSettledLeafNodes: true,
},
Expand Down Expand Up @@ -2961,7 +2961,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
const projection = yield* loadProjectionForCommand(
command,
["runs", "runtimeRequests", "turnItems"],
{ turnItemTypes: ["error"] },
{ turnItemTypes: ["user_message", "error", "notification"] },
);
const run = usageLimitBlockedRun(projection.runs, projection.turnItems, null);
const failure = latestRootProviderFailure(run, projection.turnItems);
Expand Down Expand Up @@ -4814,7 +4814,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
source === undefined ||
(source.status !== "interrupted" &&
!(source.status === "failed" && limited?.class === "usage_limit")) ||
latestExecutedRun(projection.runs)?.id !== source.id ||
(source.status === "failed"
? usageLimitBlockedRun(projection.runs, projection.turnItems, null)
: latestExecutedRun(projection.runs)
)?.id !== source.id ||
projection.thread.archivedAt !== null ||
projection.thread.deletedAt !== null ||
projection.runtimeRequests.some((request) => request.status === "pending")
Expand Down Expand Up @@ -9141,14 +9144,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
const { runs, subagents } = input.projection;
let { thread } = input.projection;
let recovery = thread.limitRecovery;
const latest = latestExecutedRun(runs);
const { turnItems } = yield* projectionStore
.getThreadRecords(thread.id, ["turnItems"], {
turnItemTypes: ["user_message", "notification", "error"],
})
.pipe(mapDispatchError(input.command));
const latest = usageLimitBlockedRun(runs, turnItems, null) ?? latestExecutedRun(runs);
if (latest?.status === "failed") {
const { turnItems } = yield* projectionStore
.getThreadRecords(thread.id, ["turnItems"], {
turnItemTypes: ["error"],
turnItemRunIds: [latest.id],
})
.pipe(mapDispatchError(input.command));
// Stop must also reach the limited source hidden behind completed wakes.
if (latest.id !== latestExecutedRun(runs)?.id) {
yield* markStoppedRun({ ...input, thread, run: latest });
}
const failure = latestRootProviderFailure(latest, turnItems);
if (
failure?.class === "usage_limit" &&
Expand Down Expand Up @@ -11444,7 +11450,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
const projection = yield* loadProjectionForCommand(
command,
["runs", "turnItems", "providerSessions", "messages"],
{ turnItemTypes: ["error"], messageRoles: ["user"] },
{ turnItemTypes: ["user_message", "error", "notification"], messageRoles: ["user"] },
);
if (projection.thread.archivedAt !== null || projection.thread.deletedAt !== null) {
return yield* new OrchestratorDispatchError({
Expand Down
3 changes: 2 additions & 1 deletion apps/server/src/orchestration-v2/ProjectionStore.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3803,7 +3803,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
(detail) =>
/^SEARCH (?:item|error) /.test(detail) && !detail.includes("(turn_item_id=?)"),
);
assert.lengthOf(errorLookups, marker.includes("blocking") ? 2 : 1, marker);
// Recovery selection also reads the latest failed source before skipping notification wakes.
assert.lengthOf(errorLookups, marker.includes("blocking") ? 3 : 2, marker);
for (const errorLookup of errorLookups) {
assert.include(errorLookup, "turn_items_failed_error_idx");
assert.include(errorLookup, "(thread_id=? AND run_id=? AND node_id=?)");
Expand Down
72 changes: 70 additions & 2 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3783,6 +3783,23 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
LIMIT 1
) AS last_error
FROM orchestration_v2_projection_threads t
LEFT JOIN orchestration_v2_projection_runs latest_failure ON latest_failure.run_id = (
SELECT failed.run_id FROM orchestration_v2_projection_runs failed
WHERE failed.thread_id = t.thread_id AND failed.status = 'failed'
ORDER BY failed.completed_at IS NULL DESC, failed.completed_at DESC,
failed.ordinal DESC, failed.run_id DESC
LIMIT 1
)
LEFT JOIN orchestration_v2_projection_turn_items latest_failure_error
ON latest_failure_error.turn_item_id = (
SELECT error.turn_item_id FROM orchestration_v2_projection_turn_items error
WHERE error.thread_id = t.thread_id
AND error.run_id = latest_failure.run_id
AND error.node_id IS json_extract(latest_failure.payload_json, '$.rootNodeId')
AND error.type = 'error' AND error.status = 'failed'
ORDER BY error.updated_at DESC, error.ordinal DESC, error.turn_item_id DESC
LIMIT 1
)
INNER JOIN orchestration_v2_projection_runs r ON r.run_id = (
SELECT latest.run_id FROM orchestration_v2_projection_runs latest
WHERE latest.thread_id = t.thread_id
Expand All @@ -3791,7 +3808,24 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
latest.status = 'cancelled'
AND json_extract(latest.payload_json, '$.startedAt') IS NULL
)
-- latestExecutedRun: the run that ended last (runRanAfter).
AND NOT (
latest.status = 'completed'
AND COALESCE(
json_extract(latest_failure_error.payload_json, '$.failure.class') = 'usage_limit'
AND julianday(latest.completed_at) <= julianday(
json_extract(latest_failure_error.payload_json, '$.failure.resetAt')
), 0
) AND (
SELECT wake.type FROM orchestration_v2_projection_turn_items wake
WHERE wake.run_id = latest.run_id
AND wake.node_id IS NOT NULL
AND wake.node_id IS json_extract(latest.payload_json, '$.rootNodeId')
AND wake.type IN ('user_message', 'notification')
ORDER BY wake.ordinal
LIMIT 1
) IS 'notification'
)
-- usageLimitBlockedRun: only notification wakes completed before reset preserve the limit.
ORDER BY latest.completed_at IS NULL DESC, latest.completed_at DESC,
latest.ordinal DESC, latest.run_id DESC
LIMIT 1
Expand Down Expand Up @@ -5803,6 +5837,23 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
ORDER BY candidate.ordinal DESC, candidate.run_id DESC
LIMIT 1
)
LEFT JOIN orchestration_v2_projection_runs latest_failure ON latest_failure.run_id = (
SELECT failed.run_id FROM orchestration_v2_projection_runs failed
WHERE failed.thread_id = t.thread_id AND failed.status = 'failed'
ORDER BY failed.completed_at IS NULL DESC, failed.completed_at DESC,
failed.ordinal DESC, failed.run_id DESC
LIMIT 1
)
LEFT JOIN orchestration_v2_projection_turn_items latest_failure_error
ON latest_failure_error.turn_item_id = (
SELECT error.turn_item_id FROM orchestration_v2_projection_turn_items error
WHERE error.thread_id = t.thread_id
AND error.run_id = latest_failure.run_id
AND error.node_id IS json_extract(latest_failure.payload_json, '$.rootNodeId')
AND error.type = 'error' AND error.status = 'failed'
ORDER BY error.updated_at DESC, error.ordinal DESC, error.turn_item_id DESC
LIMIT 1
)
LEFT JOIN orchestration_v2_projection_runs blocked ON blocked.run_id = (
SELECT candidate.run_id
FROM orchestration_v2_projection_runs candidate
Expand All @@ -5812,7 +5863,24 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
candidate.status = 'cancelled'
AND json_extract(candidate.payload_json, '$.startedAt') IS NULL
)
-- latestExecutedRun: the run that ended last (runRanAfter).
AND NOT (
candidate.status = 'completed'
AND COALESCE(
json_extract(latest_failure_error.payload_json, '$.failure.class') = 'usage_limit'
AND julianday(candidate.completed_at) <= julianday(
json_extract(latest_failure_error.payload_json, '$.failure.resetAt')
), 0
) AND (
SELECT wake.type FROM orchestration_v2_projection_turn_items wake
WHERE wake.run_id = candidate.run_id
AND wake.node_id IS NOT NULL
AND wake.node_id IS json_extract(candidate.payload_json, '$.rootNodeId')
AND wake.type IN ('user_message', 'notification')
ORDER BY wake.ordinal
LIMIT 1
) IS 'notification'
)
-- usageLimitBlockedRun: only notification wakes completed before reset preserve the limit.
ORDER BY candidate.completed_at IS NULL DESC, candidate.completed_at DESC,
candidate.ordinal DESC, candidate.run_id DESC
LIMIT 1
Expand Down
Loading
Loading