Skip to content

Commit edbf52c

Browse files
juliusmarmingejuliusmarminge
committed
fix(provider): avoid spans in Codex streaming helpers
1 parent 4eb183e commit edbf52c

1 file changed

Lines changed: 147 additions & 148 deletions

File tree

‎apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts‎

Lines changed: 147 additions & 148 deletions
Original file line numberDiff line numberDiff line change
@@ -2058,7 +2058,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
20582058
);
20592059
});
20602060

2061-
const awaitActiveTurn = Effect.fn("CodexAdapterV2.awaitActiveTurn")(function* (
2061+
// Streaming updates repeatedly await this context, including retries before registration.
2062+
const awaitActiveTurn = Effect.fnUntraced(function* (
20622063
nativeTurnId: string,
20632064
attemptsRemaining = 1_000,
20642065
): Effect.fn.Return<ActiveCodexTurnContext | undefined> {
@@ -3194,81 +3195,80 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
31943195
return inputItems;
31953196
});
31963197

3197-
const buildAgentMessageArtifacts = Effect.fn("CodexAdapterV2.buildAgentMessageArtifacts")(
3198-
function* (
3199-
context: ActiveCodexTurnContext,
3200-
item: { readonly id: string; readonly text: string },
3201-
completed: boolean,
3202-
) {
3203-
const updatedAt = yield* DateTime.now;
3204-
const completedAt = completed ? updatedAt : null;
3205-
const nodeId = idAllocator.derive.nodeFromProviderItem({
3206-
driver: CODEX_PROVIDER,
3207-
nativeItemId: item.id,
3208-
});
3209-
const { ordinal, startedAt } = yield* resolveItemPosition(context, item.id);
3210-
const messageId = idAllocator.derive.messageFromProviderItem({
3211-
driver: CODEX_PROVIDER,
3212-
nativeItemId: item.id,
3213-
});
3214-
const turnItemId = idAllocator.derive.turnItemFromProviderItem({
3215-
driver: CODEX_PROVIDER,
3216-
nativeItemId: item.id,
3217-
});
3218-
const node: OrchestrationV2ExecutionNode = {
3219-
id: nodeId,
3220-
threadId: context.projectionThreadId,
3221-
runId: context.projectionRunId,
3222-
parentNodeId: context.itemParentNodeId,
3223-
rootNodeId: context.rootNodeId,
3224-
kind: "assistant_message",
3225-
status: completed ? "completed" : "running",
3226-
countsForRun: false,
3227-
providerThreadId: context.providerThread.id,
3228-
providerTurnId: context.providerTurnId,
3229-
nativeItemRef: codexNativeItemRef(item.id),
3230-
runtimeRequestId: null,
3231-
checkpointScopeId: null,
3232-
startedAt,
3233-
completedAt,
3234-
};
3235-
const message: OrchestrationV2ConversationMessage = {
3236-
createdBy: "agent",
3237-
creationSource: "provider",
3238-
id: messageId,
3239-
threadId: context.projectionThreadId,
3240-
runId: context.projectionRunId,
3241-
nodeId,
3242-
role: "assistant",
3243-
text: item.text,
3244-
attachments: [],
3245-
streaming: !completed,
3246-
createdAt: startedAt,
3247-
updatedAt,
3248-
};
3249-
const turnItem: OrchestrationV2TurnItem = {
3250-
id: turnItemId,
3251-
threadId: context.projectionThreadId,
3252-
runId: context.projectionRunId,
3253-
nodeId,
3254-
providerThreadId: context.providerThread.id,
3255-
providerTurnId: context.providerTurnId,
3256-
nativeItemRef: codexNativeItemRef(item.id),
3257-
parentItemId: null,
3258-
ordinal,
3259-
status: completed ? "completed" : "running",
3260-
title: null,
3261-
startedAt,
3262-
completedAt,
3263-
updatedAt,
3264-
type: "assistant_message",
3265-
messageId,
3266-
text: item.text,
3267-
streaming: !completed,
3268-
};
3269-
return { node, message, turnItem };
3270-
},
3271-
);
3198+
// Reasoning and assistant-text coalescers build these artifacts on every streamed update.
3199+
const buildAgentMessageArtifacts = Effect.fnUntraced(function* (
3200+
context: ActiveCodexTurnContext,
3201+
item: { readonly id: string; readonly text: string },
3202+
completed: boolean,
3203+
) {
3204+
const updatedAt = yield* DateTime.now;
3205+
const completedAt = completed ? updatedAt : null;
3206+
const nodeId = idAllocator.derive.nodeFromProviderItem({
3207+
driver: CODEX_PROVIDER,
3208+
nativeItemId: item.id,
3209+
});
3210+
const { ordinal, startedAt } = yield* resolveItemPosition(context, item.id);
3211+
const messageId = idAllocator.derive.messageFromProviderItem({
3212+
driver: CODEX_PROVIDER,
3213+
nativeItemId: item.id,
3214+
});
3215+
const turnItemId = idAllocator.derive.turnItemFromProviderItem({
3216+
driver: CODEX_PROVIDER,
3217+
nativeItemId: item.id,
3218+
});
3219+
const node: OrchestrationV2ExecutionNode = {
3220+
id: nodeId,
3221+
threadId: context.projectionThreadId,
3222+
runId: context.projectionRunId,
3223+
parentNodeId: context.itemParentNodeId,
3224+
rootNodeId: context.rootNodeId,
3225+
kind: "assistant_message",
3226+
status: completed ? "completed" : "running",
3227+
countsForRun: false,
3228+
providerThreadId: context.providerThread.id,
3229+
providerTurnId: context.providerTurnId,
3230+
nativeItemRef: codexNativeItemRef(item.id),
3231+
runtimeRequestId: null,
3232+
checkpointScopeId: null,
3233+
startedAt,
3234+
completedAt,
3235+
};
3236+
const message: OrchestrationV2ConversationMessage = {
3237+
createdBy: "agent",
3238+
creationSource: "provider",
3239+
id: messageId,
3240+
threadId: context.projectionThreadId,
3241+
runId: context.projectionRunId,
3242+
nodeId,
3243+
role: "assistant",
3244+
text: item.text,
3245+
attachments: [],
3246+
streaming: !completed,
3247+
createdAt: startedAt,
3248+
updatedAt,
3249+
};
3250+
const turnItem: OrchestrationV2TurnItem = {
3251+
id: turnItemId,
3252+
threadId: context.projectionThreadId,
3253+
runId: context.projectionRunId,
3254+
nodeId,
3255+
providerThreadId: context.providerThread.id,
3256+
providerTurnId: context.providerTurnId,
3257+
nativeItemRef: codexNativeItemRef(item.id),
3258+
parentItemId: null,
3259+
ordinal,
3260+
status: completed ? "completed" : "running",
3261+
title: null,
3262+
startedAt,
3263+
completedAt,
3264+
updatedAt,
3265+
type: "assistant_message",
3266+
messageId,
3267+
text: item.text,
3268+
streaming: !completed,
3269+
};
3270+
return { node, message, turnItem };
3271+
});
32723272

32733273
// Reuse the text coalescer so token bursts do not create one database write per token.
32743274
const reasoningParts = new Map<
@@ -3865,78 +3865,77 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
38653865
},
38663866
);
38673867

3868-
const buildProposedPlanArtifacts = Effect.fn("CodexAdapterV2.buildProposedPlanArtifacts")(
3869-
function* (input: {
3870-
readonly context: ActiveCodexTurnContext;
3871-
readonly nativeItemId: string;
3872-
readonly status: OrchestrationV2PlanArtifact["status"];
3873-
readonly markdown: string;
3874-
readonly completed?: boolean;
3875-
}) {
3876-
const updatedAt = yield* DateTime.now;
3877-
const completedAt = input.completed === true ? updatedAt : null;
3878-
const planId = yield* resolvePlanId(input.context, input.nativeItemId);
3879-
const nodeId = idAllocator.derive.nodeFromProviderItem({
3880-
driver: CODEX_PROVIDER,
3881-
nativeItemId: input.nativeItemId,
3882-
});
3883-
const turnItemId = idAllocator.derive.turnItemFromProviderItem({
3884-
driver: CODEX_PROVIDER,
3885-
nativeItemId: input.nativeItemId,
3886-
});
3887-
const { ordinal, startedAt } = yield* resolveItemPosition(
3888-
input.context,
3889-
input.nativeItemId,
3890-
);
3891-
const plan: OrchestrationV2PlanArtifact = {
3892-
id: planId,
3893-
threadId: input.context.projectionThreadId,
3894-
runId: input.context.projectionRunId,
3895-
nodeId,
3896-
kind: "proposed_plan",
3897-
status: input.status,
3898-
markdown: input.markdown,
3899-
};
3900-
const node: OrchestrationV2ExecutionNode = {
3901-
id: nodeId,
3902-
threadId: input.context.projectionThreadId,
3903-
runId: input.context.projectionRunId,
3904-
parentNodeId: input.context.itemParentNodeId,
3905-
rootNodeId: input.context.rootNodeId,
3906-
kind: "plan",
3907-
status: input.completed === true ? "completed" : "running",
3908-
countsForRun: false,
3909-
providerThreadId: input.context.providerThread.id,
3910-
providerTurnId: input.context.providerTurnId,
3911-
nativeItemRef: codexNativeItemRef(input.nativeItemId),
3912-
runtimeRequestId: null,
3913-
checkpointScopeId: null,
3914-
startedAt,
3915-
completedAt,
3916-
};
3917-
const turnItem: OrchestrationV2TurnItem = {
3918-
id: turnItemId,
3919-
threadId: input.context.projectionThreadId,
3920-
runId: input.context.projectionRunId,
3921-
nodeId,
3922-
providerThreadId: input.context.providerThread.id,
3923-
providerTurnId: input.context.providerTurnId,
3924-
nativeItemRef: codexNativeItemRef(input.nativeItemId),
3925-
parentItemId: null,
3926-
ordinal,
3927-
status: input.completed === true ? "completed" : "running",
3928-
title: null,
3929-
startedAt,
3930-
completedAt,
3931-
updatedAt,
3932-
type: "proposed_plan",
3933-
planId,
3934-
markdown: input.markdown,
3935-
streaming: input.completed !== true,
3936-
};
3937-
return { node, plan, turnItem };
3938-
},
3939-
);
3868+
// The plan coalescer builds these artifacts on every streamed update.
3869+
const buildProposedPlanArtifacts = Effect.fnUntraced(function* (input: {
3870+
readonly context: ActiveCodexTurnContext;
3871+
readonly nativeItemId: string;
3872+
readonly status: OrchestrationV2PlanArtifact["status"];
3873+
readonly markdown: string;
3874+
readonly completed?: boolean;
3875+
}) {
3876+
const updatedAt = yield* DateTime.now;
3877+
const completedAt = input.completed === true ? updatedAt : null;
3878+
const planId = yield* resolvePlanId(input.context, input.nativeItemId);
3879+
const nodeId = idAllocator.derive.nodeFromProviderItem({
3880+
driver: CODEX_PROVIDER,
3881+
nativeItemId: input.nativeItemId,
3882+
});
3883+
const turnItemId = idAllocator.derive.turnItemFromProviderItem({
3884+
driver: CODEX_PROVIDER,
3885+
nativeItemId: input.nativeItemId,
3886+
});
3887+
const { ordinal, startedAt } = yield* resolveItemPosition(
3888+
input.context,
3889+
input.nativeItemId,
3890+
);
3891+
const plan: OrchestrationV2PlanArtifact = {
3892+
id: planId,
3893+
threadId: input.context.projectionThreadId,
3894+
runId: input.context.projectionRunId,
3895+
nodeId,
3896+
kind: "proposed_plan",
3897+
status: input.status,
3898+
markdown: input.markdown,
3899+
};
3900+
const node: OrchestrationV2ExecutionNode = {
3901+
id: nodeId,
3902+
threadId: input.context.projectionThreadId,
3903+
runId: input.context.projectionRunId,
3904+
parentNodeId: input.context.itemParentNodeId,
3905+
rootNodeId: input.context.rootNodeId,
3906+
kind: "plan",
3907+
status: input.completed === true ? "completed" : "running",
3908+
countsForRun: false,
3909+
providerThreadId: input.context.providerThread.id,
3910+
providerTurnId: input.context.providerTurnId,
3911+
nativeItemRef: codexNativeItemRef(input.nativeItemId),
3912+
runtimeRequestId: null,
3913+
checkpointScopeId: null,
3914+
startedAt,
3915+
completedAt,
3916+
};
3917+
const turnItem: OrchestrationV2TurnItem = {
3918+
id: turnItemId,
3919+
threadId: input.context.projectionThreadId,
3920+
runId: input.context.projectionRunId,
3921+
nodeId,
3922+
providerThreadId: input.context.providerThread.id,
3923+
providerTurnId: input.context.providerTurnId,
3924+
nativeItemRef: codexNativeItemRef(input.nativeItemId),
3925+
parentItemId: null,
3926+
ordinal,
3927+
status: input.completed === true ? "completed" : "running",
3928+
title: null,
3929+
startedAt,
3930+
completedAt,
3931+
updatedAt,
3932+
type: "proposed_plan",
3933+
planId,
3934+
markdown: input.markdown,
3935+
streaming: input.completed !== true,
3936+
};
3937+
return { node, plan, turnItem };
3938+
});
39403939

39413940
const planDeltas = yield* makeProviderTextDeltaCoalescer({
39423941
flushIntervalMs: CODEX_PLAN_DELTA_FLUSH_INTERVAL_MS,

0 commit comments

Comments
 (0)