feat: add queue-first collaboration request protocol

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -347,6 +347,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           });
           yield* projectionThreadRepository.upsert({
             nudging: event.payload.nudging,
+            collaborationRequests: [],
             threadId: event.payload.threadId,
             projectId: event.payload.projectId,
             parentThreadId: event.payload.parentThreadId ?? null,
@@ -667,6 +668,42 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           return;
         }
 
+        case "thread.collaboration-request-updated": {
+          const existingRow = yield* projectionThreadRepository.getById({
+            threadId: event.payload.threadId,
+          });
+          if (Option.isNone(existingRow)) {
+            return;
+          }
+          const requests = [
+            ...(existingRow.value.collaborationRequests ?? []).filter(
+              (request) => request.requestId !== event.payload.request.requestId,
+            ),
+            event.payload.request,
+          ].toSorted((left, right) => left.createdAt.localeCompare(right.createdAt));
+          yield* projectionThreadRepository.upsert({
+            ...existingRow.value,
+            collaborationRequests: requests,
+            updatedAt: event.payload.updatedAt,
+          });
+          return;
+        }
+
+        case "thread.collaboration-state-cleared": {
+          const existingRow = yield* projectionThreadRepository.getById({
+            threadId: event.payload.threadId,
+          });
+          if (Option.isNone(existingRow)) {
+            return;
+          }
+          yield* projectionThreadRepository.upsert({
+            ...existingRow.value,
+            collaborationRequests: [],
+            updatedAt: event.payload.clearedAt,
+          });
+          return;
+        }
+
         case "thread.pull-request-linked": {
           const existingRow = yield* projectionThreadRepository.getById({
             threadId: event.payload.threadId,

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts

@@ -1,5 +1,6 @@
 import {
   ChatAttachment,
+  CollaborationRequest,
   EventId,
   IsoDateTime,
   MessageId,
@@ -135,6 +136,7 @@ const WorkspaceBindingDbSchema = Schema.NullOr(
 const ProjectionThreadDbRowSchema = Schema.Struct({
   ...ProjectionThread.fields,
   nudging: Schema.fromJsonString(ThreadNudging),
+  collaborationRequests: Schema.fromJsonString(Schema.Array(CollaborationRequest)),
   modelSelection: Schema.fromJsonString(ModelSelection),
   pullRequest: Schema.NullOr(Schema.fromJsonString(Schema.NullOr(GitPullRequestAssociation))),
   reviewSnapshot: Schema.NullOr(Schema.fromJsonString(Schema.NullOr(ReviewSnapshot))),
@@ -158,6 +160,7 @@ const ProjectionChatArchiveThreadDbRowSchema = Schema.Struct({
 const ProjectionThreadWithProjectTitleDbRowSchema = Schema.Struct({
   ...ProjectionThread.fields,
   nudging: Schema.fromJsonString(ThreadNudging),
+  collaborationRequests: Schema.fromJsonString(Schema.Array(CollaborationRequest)),
   modelSelection: Schema.fromJsonString(ModelSelection),
   pullRequest: Schema.NullOr(Schema.fromJsonString(Schema.NullOr(GitPullRequestAssociation))),
   reviewSnapshot: Schema.NullOr(Schema.fromJsonString(Schema.NullOr(ReviewSnapshot))),
@@ -533,6 +536,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
     thread_id AS "threadId",
     project_id AS "projectId",
     parent_thread_id AS "parentThreadId",
+    collaboration_requests_json AS "collaborationRequests",
     title,
     model_selection_json AS "modelSelection",
     runtime_mode AS "runtimeMode",
@@ -1188,6 +1192,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           threads.thread_id AS "threadId",
           threads.project_id AS "projectId",
           threads.parent_thread_id AS "parentThreadId",
+          threads.collaboration_requests_json AS "collaborationRequests",
           threads.title,
           threads.model_selection_json AS "modelSelection",
           threads.runtime_mode AS "runtimeMode",
@@ -2049,6 +2054,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
                   pinOrderKey: row.pinOrderKey,
                   titleRegeneration: mapTitleRegeneration(row),
                   nudging: row.nudging,
+                  ...(row.collaborationRequests.length > 0
+                    ? { collaborationRequests: row.collaborationRequests }
+                    : {}),
                   deletedAt: row.deletedAt,
                   messages: messagesByThread.get(row.threadId) ?? [],
                   proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],

apps/server/src/orchestration/Layers/QueuedTurnReactor.ts

@@ -132,6 +132,11 @@ const makeQueuedTurnReactor = Effect.gen(function* () {
         }
         if (!followUp.reason) eligibleTurns.push(turn);
       }
+      eligibleTurns.sort((left, right) => {
+        const leftUser = left.origin === undefined ? 0 : 1;
+        const rightUser = right.origin === undefined ? 0 : 1;
+        return leftUser - rightUser || left.createdAt.localeCompare(right.createdAt);
+      });
       let nextQueuedTurn = eligibleTurns[0];
       if (eligibleTurns.some((turn) => turn.origin?.kind === "pull-request-monitor")) {
         const settings = yield* serverSettings.getSettings;
@@ -169,6 +174,18 @@ const makeQueuedTurnReactor = Effect.gen(function* () {
         return;
       }
 
+      const blockedByCollaborationWait = (thread.collaborationRequests ?? []).some(
+        (request) =>
+          request.senderThreadId === thread.id && request.blocking && request.status === "waiting",
+      );
+      if (
+        blockedByCollaborationWait &&
+        nextQueuedTurn.origin?.kind !== "collaboration-response" &&
+        nextQueuedTurn.origin !== undefined
+      ) {
+        return;
+      }
+
       const origin = nextQueuedTurn.origin;
       if (origin?.kind === "pull-request-monitor" && origin.headSha !== undefined) {
         const observedHeadSha = origin.headSha;

apps/server/src/orchestration/Schemas.ts

@@ -5,6 +5,8 @@ import {
   ThreadCreatedPayload as ContractsThreadCreatedPayloadSchema,
   ThreadArchivedPayload as ContractsThreadArchivedPayloadSchema,
   ThreadMetaUpdatedPayload as ContractsThreadMetaUpdatedPayloadSchema,
+  ThreadCollaborationRequestUpdatedPayload as ContractsThreadCollaborationRequestUpdatedPayloadSchema,
+  ThreadCollaborationStateClearedPayload as ContractsThreadCollaborationStateClearedPayloadSchema,
   ThreadPullRequestLinkedPayload as ContractsThreadPullRequestLinkedPayloadSchema,
   ThreadPullRequestRekeyedPayload as ContractsThreadPullRequestRekeyedPayloadSchema,
   ThreadPullRequestUnlinkedPayload as ContractsThreadPullRequestUnlinkedPayloadSchema,
@@ -56,6 +58,10 @@ export const ProjectDeletedPayload = ContractsProjectDeletedPayloadSchema;
 export const ThreadCreatedPayload = ContractsThreadCreatedPayloadSchema;
 export const ThreadArchivedPayload = ContractsThreadArchivedPayloadSchema;
 export const ThreadMetaUpdatedPayload = ContractsThreadMetaUpdatedPayloadSchema;
+export const ThreadCollaborationRequestUpdatedPayload =
+  ContractsThreadCollaborationRequestUpdatedPayloadSchema;
+export const ThreadCollaborationStateClearedPayload =
+  ContractsThreadCollaborationStateClearedPayloadSchema;
 export const ThreadPullRequestLinkedPayload = ContractsThreadPullRequestLinkedPayloadSchema;
 export const ThreadPullRequestRekeyedPayload = ContractsThreadPullRequestRekeyedPayloadSchema;
 export const ThreadPullRequestUnlinkedPayload = ContractsThreadPullRequestUnlinkedPayloadSchema;