Skip to content

Commit 70e2260

Browse files
rafael-francadevin-ai-integration[bot]
rafael-franca
and
devin-ai-integration[bot]
committed
fix(server): remove thread lock around Devin notification handler
Wrapping the ACP notification stream in withThreadLock caused a deadlock: prepared.acp.drainEvents emits an EventStreamBarrier and waits for its acknowledgement while holding the per-thread permit, but the consumer acknowledges the barrier only after acquiring the same permit. Grok and Cursor already ack barriers outside the lock. Keep the baseline/delta fixes from the previous commit (no lastWrittenAcpUsage rebase on UsageUpdated, per-field deltas, no baseline advance when no delta is written) and restore the notification handler to run without the lock. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
1 parent 5034747 commit 70e2260

1 file changed

Lines changed: 103 additions & 106 deletions

File tree

‎apps/server/src/provider/Layers/DevinAdapter.ts‎

Lines changed: 103 additions & 106 deletions
Original file line numberDiff line numberDiff line change
@@ -1011,122 +1011,119 @@ export const makeDevinAdapter = Effect.fn("makeDevinAdapter")(function* (
10111011

10121012
const nf = yield* Stream.runDrain(
10131013
Stream.mapEffect(acp.getEvents(), (event) =>
1014-
withThreadLock(
1015-
ctx.threadId,
1016-
Effect.gen(function* () {
1017-
if (event._tag === "EventStreamBarrier") {
1018-
yield* Deferred.succeed(event.acknowledge, undefined);
1019-
return;
1020-
}
1021-
if (
1022-
event._tag === "PlanUpdated" ||
1023-
event._tag === "ToolCallUpdated" ||
1024-
event._tag === "ContentDelta"
1025-
) {
1026-
yield* logNative(ctx.threadId, "session/update", event.rawPayload);
1027-
}
1014+
Effect.gen(function* () {
1015+
if (event._tag === "EventStreamBarrier") {
1016+
yield* Deferred.succeed(event.acknowledge, undefined);
1017+
return;
1018+
}
1019+
if (
1020+
event._tag === "PlanUpdated" ||
1021+
event._tag === "ToolCallUpdated" ||
1022+
event._tag === "ContentDelta"
1023+
) {
1024+
yield* logNative(ctx.threadId, "session/update", event.rawPayload);
1025+
}
10281026

1029-
if (event._tag === "ModeChanged") {
1030-
return;
1031-
}
1027+
if (event._tag === "ModeChanged") {
1028+
return;
1029+
}
10321030

1033-
const notificationTurnId = resolveNotificationTurnId(ctx);
1034-
if (
1035-
notificationTurnId === undefined ||
1036-
ctx.interruptedTurnIds.has(notificationTurnId)
1037-
) {
1038-
return;
1039-
}
1040-
const stamp = yield* makeEventStamp();
1031+
const notificationTurnId = resolveNotificationTurnId(ctx);
1032+
if (
1033+
notificationTurnId === undefined ||
1034+
ctx.interruptedTurnIds.has(notificationTurnId)
1035+
) {
1036+
return;
1037+
}
1038+
const stamp = yield* makeEventStamp();
10411039

1042-
switch (event._tag) {
1043-
case "AssistantItemStarted":
1044-
yield* offerRuntimeEvent(
1045-
makeAcpAssistantItemEvent({
1046-
stamp,
1047-
provider: PROVIDER,
1048-
threadId: ctx.threadId,
1049-
turnId: notificationTurnId,
1050-
itemId: event.itemId,
1051-
lifecycle: "item.started",
1052-
}),
1053-
);
1054-
return;
1055-
case "AssistantItemCompleted":
1056-
yield* offerRuntimeEvent(
1057-
makeAcpAssistantItemEvent({
1058-
stamp,
1059-
provider: PROVIDER,
1060-
threadId: ctx.threadId,
1061-
turnId: notificationTurnId,
1062-
itemId: event.itemId,
1063-
lifecycle: "item.completed",
1064-
}),
1065-
);
1066-
return;
1067-
case "PlanUpdated":
1068-
yield* emitPlanUpdate(
1069-
ctx,
1070-
notificationTurnId,
1040+
switch (event._tag) {
1041+
case "AssistantItemStarted":
1042+
yield* offerRuntimeEvent(
1043+
makeAcpAssistantItemEvent({
10711044
stamp,
1072-
event.payload,
1073-
event.rawPayload,
1074-
"session/update",
1075-
);
1076-
return;
1077-
case "ToolCallUpdated":
1078-
yield* offerRuntimeEvent(
1079-
makeAcpToolCallEvent({
1080-
stamp,
1081-
provider: PROVIDER,
1082-
threadId: ctx.threadId,
1083-
turnId: notificationTurnId,
1084-
toolCall: event.toolCall,
1085-
rawPayload: event.rawPayload,
1086-
}),
1087-
);
1088-
return;
1089-
case "ContentDelta":
1090-
yield* offerRuntimeEvent(
1091-
makeAcpContentDeltaEvent({
1092-
stamp,
1093-
provider: PROVIDER,
1094-
threadId: ctx.threadId,
1095-
turnId: notificationTurnId,
1096-
...(event.itemId ? { itemId: event.itemId } : {}),
1097-
text: event.text,
1098-
rawPayload: event.rawPayload,
1099-
}),
1100-
);
1101-
return;
1102-
case "UsageUpdated": {
1103-
const tokenUsage = makeDevinTokenUsageSnapshotFromUsageUpdate(
1104-
event,
1105-
ctx.lastKnownTokenUsage,
1106-
);
1107-
if (!tokenUsage) {
1108-
return;
1109-
}
1110-
ctx.lastKnownTokenUsage = tokenUsage;
1111-
1112-
const acpUsage = usageFromUsageUpdate(event);
1113-
if (isAcpUsageGreaterOrNew(ctx.lastAcpUsage, acpUsage)) {
1114-
ctx.lastAcpUsage = acpUsage;
1115-
}
1116-
1117-
yield* offerRuntimeEvent({
1118-
type: "thread.token-usage.updated",
1119-
...stamp,
11201045
provider: PROVIDER,
11211046
threadId: ctx.threadId,
11221047
turnId: notificationTurnId,
1123-
payload: { usage: tokenUsage },
1124-
});
1048+
itemId: event.itemId,
1049+
lifecycle: "item.started",
1050+
}),
1051+
);
1052+
return;
1053+
case "AssistantItemCompleted":
1054+
yield* offerRuntimeEvent(
1055+
makeAcpAssistantItemEvent({
1056+
stamp,
1057+
provider: PROVIDER,
1058+
threadId: ctx.threadId,
1059+
turnId: notificationTurnId,
1060+
itemId: event.itemId,
1061+
lifecycle: "item.completed",
1062+
}),
1063+
);
1064+
return;
1065+
case "PlanUpdated":
1066+
yield* emitPlanUpdate(
1067+
ctx,
1068+
notificationTurnId,
1069+
stamp,
1070+
event.payload,
1071+
event.rawPayload,
1072+
"session/update",
1073+
);
1074+
return;
1075+
case "ToolCallUpdated":
1076+
yield* offerRuntimeEvent(
1077+
makeAcpToolCallEvent({
1078+
stamp,
1079+
provider: PROVIDER,
1080+
threadId: ctx.threadId,
1081+
turnId: notificationTurnId,
1082+
toolCall: event.toolCall,
1083+
rawPayload: event.rawPayload,
1084+
}),
1085+
);
1086+
return;
1087+
case "ContentDelta":
1088+
yield* offerRuntimeEvent(
1089+
makeAcpContentDeltaEvent({
1090+
stamp,
1091+
provider: PROVIDER,
1092+
threadId: ctx.threadId,
1093+
turnId: notificationTurnId,
1094+
...(event.itemId ? { itemId: event.itemId } : {}),
1095+
text: event.text,
1096+
rawPayload: event.rawPayload,
1097+
}),
1098+
);
1099+
return;
1100+
case "UsageUpdated": {
1101+
const tokenUsage = makeDevinTokenUsageSnapshotFromUsageUpdate(
1102+
event,
1103+
ctx.lastKnownTokenUsage,
1104+
);
1105+
if (!tokenUsage) {
11251106
return;
11261107
}
1108+
ctx.lastKnownTokenUsage = tokenUsage;
1109+
1110+
const acpUsage = usageFromUsageUpdate(event);
1111+
if (isAcpUsageGreaterOrNew(ctx.lastAcpUsage, acpUsage)) {
1112+
ctx.lastAcpUsage = acpUsage;
1113+
}
1114+
1115+
yield* offerRuntimeEvent({
1116+
type: "thread.token-usage.updated",
1117+
...stamp,
1118+
provider: PROVIDER,
1119+
threadId: ctx.threadId,
1120+
turnId: notificationTurnId,
1121+
payload: { usage: tokenUsage },
1122+
});
1123+
return;
11271124
}
1128-
}),
1129-
),
1125+
}
1126+
}),
11301127
),
11311128
).pipe(
11321129
Effect.catch((cause) =>

0 commit comments

Comments
 (0)