| ... | ... | @@ -3094,6 +3094,180 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { |
| 3094 | 3094 | }), |
| 3095 | 3095 | ); |
| 3096 | 3096 | |
| 3097 | it.effect("looks up a thread's last provider error through its own session bindings", () => |
| 3098 | Effect.gen(function* () { |
| 3099 | const projectionStore = yield* ProjectionStore.ProjectionStoreV2; |
| 3100 | const sql = yield* SqlClient.SqlClient; |
| 3101 | const now = yield* DateTime.now; |
| 3102 | const later = DateTime.add(now, { minutes: 1 }); |
| 3103 | const projectId = ProjectId.make("project:projection-session-error-lookup"); |
| 3104 | const attachErroredSession = Effect.fnUntraced(function* ( |
| 3105 | name: string, |
| 3106 | lastError: string, |
| 3107 | updatedAt: DateTime.Utc, |
| 3108 | ) { |
| 3109 | const threadId = ThreadId.make(`thread:projection-session-error-lookup:${name}`); |
| 3110 | yield* projectionStore.apply({ |
| 3111 | id: EventId.make(`event:projection-session-error-lookup:${name}:thread`), |
| 3112 | type: "thread.created", |
| 3113 | threadId, |
| 3114 | occurredAt: now, |
| 3115 | payload: { |
| 3116 | createdBy: "user", |
| 3117 | creationSource: "web", |
| 3118 | id: threadId, |
| 3119 | projectId, |
| 3120 | title: name, |
| 3121 | providerInstanceId, |
| 3122 | modelSelection, |
| 3123 | runtimeMode: "full-access", |
| 3124 | interactionMode: "default", |
| 3125 | branch: null, |
| 3126 | worktreePath: null, |
| 3127 | activeProviderThreadId: null, |
| 3128 | lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, |
| 3129 | forkedFrom: null, |
| 3130 | createdAt: now, |
| 3131 | updatedAt: now, |
| 3132 | archivedAt: null, |
| 3133 | settledOverride: null, |
| 3134 | settledAt: null, |
| 3135 | lastVisitedAt: null, |
| 3136 | deletedAt: null, |
| 3137 | }, |
| 3138 | }); |
| 3139 | yield* projectionStore.apply({ |
| 3140 | id: EventId.make(`event:projection-session-error-lookup:${name}:session`), |
| 3141 | type: "provider-session.attached", |
| 3142 | threadId, |
| 3143 | driver, |
| 3144 | providerInstanceId, |
| 3145 | occurredAt: updatedAt, |
| 3146 | payload: { |
| 3147 | id: ProviderSessionId.make(`provider-session:projection-session-error-lookup:${name}`), |
| 3148 | driver, |
| 3149 | providerInstanceId, |
| 3150 | status: "error", |
| 3151 | cwd: "/workspace", |
| 3152 | model: modelSelection.model, |
| 3153 | capabilities: CodexProviderCapabilitiesV2, |
| 3154 | createdAt: updatedAt, |
| 3155 | updatedAt, |
| 3156 | lastError, |
| 3157 | }, |
| 3158 | }); |
| 3159 | return threadId; |
| 3160 | }); |
| 3161 | |
| 3162 | const failure = { |
| 3163 | class: "usage_limit" as const, |
| 3164 | message: "Plan limit reached.", |
| 3165 | resetAt: "2099-01-01T00:00:00.000Z", |
| 3166 | code: "usageLimitExceeded", |
| 3167 | retryable: null, |
| 3168 | }; |
| 3169 | const threadId = yield* attachErroredSession("own", failure.message, now); |
| 3170 | // The own session repeats the root turn's limit, so the thread stays a |
| 3171 | // recovery candidate only while the lookup reads that session. |
| 3172 | const runId = RunId.make("run:projection-session-error-lookup:own"); |
| 3173 | const rootNodeId = NodeId.make("node:projection-session-error-lookup:own"); |
| 3174 | yield* projectionStore.apply({ |
| 3175 | id: EventId.make("event:projection-session-error-lookup:own:run"), |
| 3176 | type: "run.created", |
| 3177 | threadId, |
| 3178 | runId, |
| 3179 | nodeId: rootNodeId, |
| 3180 | driver, |
| 3181 | providerInstanceId, |
| 3182 | occurredAt: now, |
| 3183 | payload: { |
| 3184 | id: runId, |
| 3185 | threadId, |
| 3186 | ordinal: 1, |
| 3187 | providerInstanceId, |
| 3188 | modelSelection, |
| 3189 | providerThreadId: null, |
| 3190 | userMessageId: MessageId.make("message:projection-session-error-lookup:own"), |
| 3191 | rootNodeId, |
| 3192 | activeAttemptId: null, |
| 3193 | status: "failed", |
| 3194 | requestedAt: now, |
| 3195 | startedAt: now, |
| 3196 | completedAt: now, |
| 3197 | checkpointId: null, |
| 3198 | contextHandoffId: null, |
| 3199 | }, |
| 3200 | }); |
| 3201 | yield* projectionStore.apply({ |
| 3202 | id: EventId.make("event:projection-session-error-lookup:own:error"), |
| 3203 | type: "turn-item.updated", |
| 3204 | threadId, |
| 3205 | occurredAt: now, |
| 3206 | payload: { |
| 3207 | id: TurnItemId.make("projection-session-error-lookup:own:error"), |
| 3208 | threadId, |
| 3209 | runId, |
| 3210 | nodeId: rootNodeId, |
| 3211 | providerThreadId: null, |
| 3212 | providerTurnId: null, |
| 3213 | nativeItemRef: null, |
| 3214 | parentItemId: null, |
| 3215 | ordinal: 1, |
| 3216 | status: "failed", |
| 3217 | title: "Usage limit reached", |
| 3218 | startedAt: now, |
| 3219 | completedAt: now, |
| 3220 | updatedAt: now, |
| 3221 | type: "error", |
| 3222 | failure, |
| 3223 | }, |
| 3224 | }); |
| 3225 | // Every thread on the provider instance has its own session, and these |
| 3226 | // are newer. They must neither leak into the thread nor be walked for it. |
| 3227 | for (let index = 0; index < 24; index++) { |
| 3228 | yield* attachErroredSession(`other-${index}`, "other session failed", later); |
| 3229 | } |
| 3230 | |
| 3231 | const { statements, tracer } = traceSqlStatements(); |
| 3232 | const shell = yield* projectionStore.getShellSnapshot().pipe(Effect.withTracer(tracer)); |
| 3233 | assert.equal( |
| 3234 | shell.threads.find((thread) => thread.id === threadId)?.lastError, |
| 3235 | failure.message, |
| 3236 | ); |
| 3237 | const candidates = yield* projectionStore |
| 3238 | .getLimitRecoveryCandidates({ now, autoResume: true, snooze: false }) |
| 3239 | .pipe(Effect.withTracer(tracer)); |
| 3240 | // Another thread's newer error would replace the limit and drop the candidate. |
| 3241 | const candidate = candidates.find((row) => row.id === threadId); |
| 3242 | assert.isDefined(candidate); |
| 3243 | assert.include(candidate, { |
| 3244 | status: "failed", |
| 3245 | lastErrorClass: "usage_limit", |
| 3246 | usageLimitResetAt: failure.resetAt, |
| 3247 | latestRunId: runId, |
| 3248 | }); |
| 3249 | |
| 3250 | // One provider instance usually holds most sessions. Driving the lookup |
| 3251 | // from sessions walks all of them for every thread in the snapshot. |
| 3252 | for (const marker of ["AS blocking_failure_payload_json", "AS failure_payload_json"]) { |
| 3253 | const statement = statements.find((query) => query.includes(marker)); |
| 3254 | assert.isDefined(statement, marker); |
| 3255 | const plan = (yield* sql.unsafe<{ readonly detail: string }>( |
| 3256 | `EXPLAIN QUERY PLAN ${statement}`, |
| 3257 | )).map((row) => row.detail); |
| 3258 | const bindingLookup = plan.findIndex((detail) => detail.startsWith("SEARCH binding ")); |
| 3259 | const sessionLookup = plan.findIndex((detail) => detail.startsWith("SEARCH session ")); |
| 3260 | assert.isAtLeast(bindingLookup, 0, marker); |
| 3261 | assert.include( |
| 3262 | plan[bindingLookup], |
| 3263 | "orchestration_v2_projection_provider_session_bindings_thread_idx (thread_id=?)", |
| 3264 | ); |
| 3265 | assert.isAbove(sessionLookup, bindingLookup, marker); |
| 3266 | assert.include(plan[sessionLookup], "(provider_session_id=?)"); |
| 3267 | } |
| 3268 | }), |
| 3269 | ); |
| 3270 | |
| 3097 | 3271 | it.effect( |
| 3098 | 3272 | "reads checkpoint context without decoding transcript or checkpoint file payloads", |
| 3099 | 3273 | () => |
only21mil <thevictorvogel@gmail.com> 2026-10-10 20:39:53-05:00
GitHub <noreply@github.com> 2026-10-10 18:39:53-07:00