fix(server): messages from other threads no longer reset a pending question

apps/server/src/orchestration-v2/EventSink.ts

@@ -348,42 +348,60 @@ const layerBase: Layer.Layer<
         const staleRequests = new Set<RuntimeRequestId>();
         const staleNodes = new Set<NodeId>();
         for (const event of events) {
-          if (
-            event.type !== "runtime-request.updated" ||
-            event.payload.kind !== "user_input" ||
-            event.payload.status !== "cancelled"
-          )
+          if (event.type !== "runtime-request.updated" || event.payload.status !== "cancelled")
             continue;
           const current = yield* projectionStore.getRuntimeRequest(
             event.threadId,
             event.payload.id,
           );
           if (
             current?.status !== "pending" ||
-            current.kind !== "user_input" ||
+            current.kind !== event.payload.kind ||
             current.providerTurnId !== event.payload.providerTurnId ||
             current.responseCapability.type === "message"
           ) {
             staleRequests.add(event.payload.id);
             staleNodes.add(event.payload.nodeId);
           }
         }
-        return events.filter((event) => {
-          switch (event.type) {
-            case "runtime-request.updated":
-              return event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id);
-            case "node.updated":
-              return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id);
-            case "turn-item.updated":
-              return (
-                event.payload.type !== "user_input_request" ||
-                event.payload.status !== "cancelled" ||
-                !staleRequests.has(event.payload.requestId)
-              );
-            default:
-              return true;
-          }
-        });
+        return yield* Effect.filter(
+          events,
+          Effect.fnUntraced(function* (event) {
+            const requestId =
+              event.type === "node.updated" &&
+              (event.payload.kind === "user_input_request" ||
+                event.payload.kind === "approval_request")
+                ? event.payload.runtimeRequestId
+                : event.type === "turn-item.updated" &&
+                    (event.payload.type === "user_input_request" ||
+                      event.payload.type === "approval_request")
+                  ? event.payload.requestId
+                  : null;
+            if (
+              requestId !== null &&
+              (event.type === "node.updated" || event.type === "turn-item.updated") &&
+              event.payload.status === "cancelled"
+            ) {
+              const current = yield* projectionStore.getRuntimeRequest(event.threadId, requestId);
+              if (current?.status === "resolved") return false;
+            }
+            switch (event.type) {
+              case "runtime-request.updated":
+                return event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id);
+              case "node.updated":
+                return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id);
+              case "turn-item.updated":
+                return (
+                  (event.payload.type !== "user_input_request" &&
+                    event.payload.type !== "approval_request") ||
+                  event.payload.status !== "cancelled" ||
+                  !staleRequests.has(event.payload.requestId)
+                );
+              default:
+                return true;
+            }
+          }),
+        );
       });
 
     const normalizeEvents = (events: ReadonlyArray<OrchestrationV2DomainEvent>) => {