From e98cd24d67de738481581f6dcab767025ef8bd1b Mon Sep 17 00:00:00 2001 From: Adolanium <94890352+Adolanium@users.noreply.github.com> Date: Thu, 27 Aug 2026 20:31:18 +0300 Subject: [PATCH] fix(server): drop pending approvals when a thread reverts Revert already dropped later-turn messages and activities. Pending approval rows stayed. The shell count still included them, so the user could answer a request the provider no longer had. On thread.reverted, drop pending approvals whose turn was discarded. Approvals with no turn stay, matching activities. The shell pending count then follows. --- .../Layers/ProjectionPipeline.test.ts | 204 ++++++++++++++++++ .../Layers/ProjectionPipeline.ts | 59 ++++- 2 files changed, 262 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index d27caed4a8e2..153e8ca3f5d6 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -2626,6 +2626,210 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { ]); }), ); + + it.effect("drops later-turn pending approvals when a thread is reverted", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); + + yield* appendAndProject({ + type: "project.created", + eventId: EventId.make("evt-revert-approval-1"), + aggregateKind: "project", + aggregateId: ProjectId.make("project-revert-approval"), + occurredAt: "2026-02-26T13:00:00.000Z", + commandId: CommandId.make("cmd-revert-approval-1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-approval-1"), + metadata: {}, + payload: { + projectId: ProjectId.make("project-revert-approval"), + title: "Project Revert Approval", + workspaceRoot: "/tmp/project-revert-approval", + defaultModelSelection: null, + scripts: [], + createdAt: "2026-02-26T13:00:00.000Z", + updatedAt: "2026-02-26T13:00:00.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.make("evt-revert-approval-2"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-revert-approval"), + occurredAt: "2026-02-26T13:00:01.000Z", + commandId: CommandId.make("cmd-revert-approval-2"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-approval-2"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-revert-approval"), + projectId: ProjectId.make("project-revert-approval"), + title: "Thread Revert Approval", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "approval-required", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: "2026-02-26T13:00:01.000Z", + updatedAt: "2026-02-26T13:00:01.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.turn-diff-completed", + eventId: EventId.make("evt-revert-approval-3"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-revert-approval"), + occurredAt: "2026-02-26T13:00:02.000Z", + commandId: CommandId.make("cmd-revert-approval-3"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-approval-3"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-revert-approval"), + turnId: TurnId.make("turn-keep"), + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-revert-approval/turn/1"), + status: "ready", + files: [], + assistantMessageId: MessageId.make("assistant-keep-approval"), + completedAt: "2026-02-26T13:00:02.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-revert-approval-4"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-revert-approval"), + occurredAt: "2026-02-26T13:00:02.100Z", + commandId: CommandId.make("cmd-revert-approval-4"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-approval-4"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-revert-approval"), + activity: { + id: EventId.make("activity-keep-approval"), + tone: "approval", + kind: "approval.requested", + summary: "Keep this approval", + payload: { + requestId: "approval-request-keep", + requestKind: "command", + }, + turnId: TurnId.make("turn-keep"), + createdAt: "2026-02-26T13:00:02.100Z", + }, + }, + }); + + yield* appendAndProject({ + type: "thread.turn-diff-completed", + eventId: EventId.make("evt-revert-approval-5"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-revert-approval"), + occurredAt: "2026-02-26T13:00:03.000Z", + commandId: CommandId.make("cmd-revert-approval-5"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-approval-5"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-revert-approval"), + turnId: TurnId.make("turn-drop"), + checkpointTurnCount: 2, + checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-revert-approval/turn/2"), + status: "ready", + files: [], + assistantMessageId: MessageId.make("assistant-drop-approval"), + completedAt: "2026-02-26T13:00:03.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-revert-approval-6"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-revert-approval"), + occurredAt: "2026-02-26T13:00:03.100Z", + commandId: CommandId.make("cmd-revert-approval-6"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-approval-6"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-revert-approval"), + activity: { + id: EventId.make("activity-drop-approval"), + tone: "approval", + kind: "approval.requested", + summary: "Drop this approval", + payload: { + requestId: "approval-request-drop", + requestKind: "command", + }, + turnId: TurnId.make("turn-drop"), + createdAt: "2026-02-26T13:00:03.100Z", + }, + }, + }); + + yield* appendAndProject({ + type: "thread.reverted", + eventId: EventId.make("evt-revert-approval-7"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-revert-approval"), + occurredAt: "2026-02-26T13:00:04.000Z", + commandId: CommandId.make("cmd-revert-approval-7"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-revert-approval-7"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-revert-approval"), + turnCount: 1, + }, + }); + + const approvalRows = yield* sql<{ + readonly requestId: string; + readonly status: string; + readonly turnId: string | null; + }>` + SELECT + request_id AS "requestId", + status, + turn_id AS "turnId" + FROM projection_pending_approvals + WHERE thread_id = 'thread-revert-approval' + ORDER BY request_id ASC + `; + assert.deepEqual(approvalRows, [ + { + requestId: "approval-request-keep", + status: "pending", + turnId: "turn-keep", + }, + ]); + + const threadRows = yield* sql<{ + readonly pendingApprovalCount: number; + }>` + SELECT pending_approval_count AS "pendingApprovalCount" + FROM projection_threads + WHERE thread_id = 'thread-revert-approval' + `; + assert.deepEqual(threadRows, [{ pendingApprovalCount: 1 }]); + }), + ); }); it.layer(makeProjectionPipelinePrefixedTestLayer("t3-pending-turn-terminal-test-"))( diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 348615b1033b..f1d0e9ed267b 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -15,7 +15,10 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../persistence/Errors.ts"; import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts"; -import { ProjectionPendingApprovalRepository } from "../../persistence/Services/ProjectionPendingApprovals.ts"; +import { + type ProjectionPendingApproval, + ProjectionPendingApprovalRepository, +} from "../../persistence/Services/ProjectionPendingApprovals.ts"; import { ProjectionProjectRepository } from "../../persistence/Services/ProjectionProjects.ts"; import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts"; import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts"; @@ -350,6 +353,26 @@ function retainProjectionProposedPlansAfterRevert( ); } +function retainProjectionPendingApprovalsAfterRevert( + approvals: ReadonlyArray, + turns: ReadonlyArray, + turnCount: number, +): ReadonlyArray { + const retainedTurnIds = new Set( + turns + .filter( + (turn) => + turn.turnId !== null && + turn.checkpointTurnCount !== null && + turn.checkpointTurnCount <= turnCount, + ) + .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])), + ); + return approvals.filter( + (approval) => approval.turnId === null || retainedTurnIds.has(approval.turnId), + ); +} + function collectThreadAttachmentRelativePaths( threadId: string, messages: ReadonlyArray, @@ -1640,6 +1663,40 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + case "thread.reverted": { + const existingRows = yield* projectionPendingApprovalRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + if (existingRows.length === 0) { + return; + } + + const existingTurns = yield* projectionTurnRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + const keptRows = retainProjectionPendingApprovalsAfterRevert( + existingRows, + existingTurns, + event.payload.turnCount, + ); + if (keptRows.length === existingRows.length) { + return; + } + + const keptRequestIds = new Set(keptRows.map((row) => row.requestId)); + yield* Effect.forEach( + existingRows, + (row) => + keptRequestIds.has(row.requestId) + ? Effect.void + : projectionPendingApprovalRepository.deleteByRequestId({ + requestId: row.requestId, + }), + { concurrency: 1 }, + ).pipe(Effect.asVoid); + return; + } + default: return; }