Skip to content
Closed
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
25 changes: 24 additions & 1 deletion apps/server/src/orchestration-v2/ProjectionStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3115,7 +3115,30 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
assert.include(recoveryThreadIds, childThreadId);
assert.notInclude(recoveryThreadIds, settledThreadId);
assert.notInclude(recoveryThreadIds, rolledBackThreadId);
const memberRecovery = yield* projectionStore.getRuntimeRecoveryProjection(memberThreadId);
const { statements, tracer } = traceSqlStatements();
const memberRecovery = yield* projectionStore
.getRuntimeRecoveryProjection(memberThreadId)
.pipe(Effect.withTracer(tracer));
// Completed history grows without bound, so the provider-thread read's
// unfinished-item lookups must stay on a partial index.
const providerThreadRead = statements.find((query) =>
query.includes("SELECT provider_thread.payload_json"),
);
assert.isDefined(providerThreadRead);
const sql = yield* SqlClient.SqlClient;
const itemLookups = (yield* sql.unsafe<{ readonly detail: string }>(
`EXPLAIN QUERY PLAN ${providerThreadRead}`,
))
.map((row) => row.detail)
.filter((detail) => /^(SEARCH|SCAN) item /.test(detail));
assert.isNotEmpty(itemLookups);
for (const detail of itemLookups) {
const indexName = /INDEX (\S+)/.exec(detail)?.[1] ?? "";
const [index] = yield* sql<{ readonly sql: string }>`
SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ${indexName}
`;
assert.match(index?.sql ?? "", /WHERE[\s\S]*status/, detail);
}
const childRecovery = yield* projectionStore.getRuntimeRecoveryProjection(childThreadId);
assert.deepEqual(
memberRecovery.subagents.map((subagent) => subagent.id),
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/persistence/Migrations.ts
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ import Migration0064 from "./Migrations/064_CanonicalizeThreadPullRequestLinks.t
import Migration0065 from "./Migrations/065_CodexUsageNativeTurnIndex.ts";
import Migration0066 from "./Migrations/066_RepairClientSettlementTimestamps.ts";
import Migration0067 from "./Migrations/067_WebhookDedupe.ts";
import Migration0068 from "./Migrations/068_UnfinishedTurnItemsIndex.ts";

const migrationEntries = [
[1, "OrchestrationEvents", Migration0001],
Expand Down Expand Up @@ -140,6 +141,7 @@ const migrationEntries = [
[65, "CodexUsageNativeTurnIndex", Migration0065],
[66, "RepairClientSettlementTimestamps", Migration0066],
[67, "WebhookDedupe", Migration0067],
[68, "UnfinishedTurnItemsIndex", Migration0068],
] as const;

export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
import * as Effect from "effect/Effect";
import * as SqlClient from "effect/sql/SqlClient";

export default Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;

// Recovery's provider-thread read looks up unfinished items of any type, which
// the type-filtered recovery index cannot serve.
yield* sql`
CREATE INDEX orchestration_v2_projection_turn_items_unfinished_idx
ON orchestration_v2_projection_turn_items(thread_id)
WHERE status IN ('pending', 'running', 'waiting')
`;
});
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ describe("V2 preview upgrade", () => {
[65, "CodexUsageNativeTurnIndex"],
[66, "RepairClientSettlementTimestamps"],
[67, "WebhookDedupe"],
[68, "UnfinishedTurnItemsIndex"],
]);
assert.deepStrictEqual(yield* runMigrations(), []);
assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);
Expand Down Expand Up @@ -137,6 +138,7 @@ describe("V2 preview upgrade", () => {
[65, "CodexUsageNativeTurnIndex"],
[66, "RepairClientSettlementTimestamps"],
[67, "WebhookDedupe"],
[68, "UnfinishedTurnItemsIndex"],
]);
}).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
);
Expand Down
Loading