-
Notifications
You must be signed in to change notification settings - Fork 2.8k
[orchestrator-v2] fix(orchestrator): Surface Claude background wake turns as continuation runs #3752
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: t3code/codex-turn-mapping
Are you sure you want to change the base?
[orchestrator-v2] fix(orchestrator): Surface Claude background wake turns as continuation runs #3752
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Large diffs are not rendered by default.
Large diffs are not rendered by default.
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| import { ProviderDriverKind, ProviderThreadId, ThreadId } from "@t3tools/contracts"; | ||
| import * as Context from "effect/Context"; | ||
| import * as Effect from "effect/Effect"; | ||
| import * as Layer from "effect/Layer"; | ||
| import * as Queue from "effect/Queue"; | ||
|
|
||
| export interface ProviderContinuationRequest { | ||
| readonly threadId: ThreadId; | ||
| readonly providerThreadId: ProviderThreadId; | ||
| readonly driver: ProviderDriverKind; | ||
| readonly detail: string | null; | ||
| } | ||
|
|
||
| /** | ||
| * Adapters offer a continuation request when provider-native work completes | ||
| * outside an active turn (for example a Claude background task wake turn) so | ||
| * the orchestrator can start a run that ingests it. The default reference | ||
| * drops requests, keeping adapter construction dependency-free in tests; the | ||
| * live layer must be shared with the ProviderContinuationService worker that | ||
| * drains it. | ||
| */ | ||
| export class ProviderContinuationRequests extends Context.Reference<{ | ||
| readonly offer: (request: ProviderContinuationRequest) => Effect.Effect<void>; | ||
| readonly take: Effect.Effect<ProviderContinuationRequest>; | ||
| }>("t3/orchestration-v2/ProviderContinuationRequests", { | ||
| defaultValue: () => ({ offer: () => Effect.void, take: Effect.never }), | ||
| }) {} | ||
|
|
||
| export const layer = Layer.effect( | ||
| ProviderContinuationRequests, | ||
| Effect.gen(function* () { | ||
| const queue = yield* Queue.unbounded<ProviderContinuationRequest>(); | ||
| return { | ||
| offer: (request: ProviderContinuationRequest) => | ||
| Queue.offer(queue, request).pipe(Effect.asVoid), | ||
| take: Queue.take(queue), | ||
| }; | ||
| }), | ||
| ); |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| import { CommandId } from "@t3tools/contracts"; | ||
| import * as Effect from "effect/Effect"; | ||
| import * as Layer from "effect/Layer"; | ||
|
|
||
| import { IdAllocatorV2 } from "./IdAllocator.ts"; | ||
| import { | ||
| type ProviderContinuationRequest, | ||
| ProviderContinuationRequests, | ||
| } from "./ProviderContinuationRequests.ts"; | ||
| import { ThreadManagementService } from "./ThreadManagementService.ts"; | ||
|
|
||
| const CONTINUATION_MESSAGE_TEXT = "Background task completed."; | ||
|
|
||
| /** | ||
| * Drains ProviderContinuationRequests and dispatches an internal | ||
| * message.dispatch per request so the wake turn buffered by the adapter is | ||
| * ingested as a normal run. Dispatches queue_after_active, so a continuation | ||
| * racing a user run simply queues behind it and drains the wake buffer once | ||
| * that run finishes. | ||
| */ | ||
| export const workerLive = Layer.effectDiscard( | ||
| Effect.gen(function* () { | ||
| const ids = yield* IdAllocatorV2; | ||
| const requests = yield* ProviderContinuationRequests; | ||
| const threads = yield* ThreadManagementService; | ||
|
|
||
| const dispatchContinuation = Effect.fn("ProviderContinuationService.dispatchContinuation")( | ||
| function* (request: ProviderContinuationRequest) { | ||
| const projection = yield* threads.getThreadProjection(request.threadId); | ||
| if (projection.thread.archivedAt !== null) { | ||
| yield* Effect.logInfo("orchestration-v2.provider-continuation.thread-archived", { | ||
| threadId: request.threadId, | ||
| providerThreadId: request.providerThreadId, | ||
| }); | ||
| return; | ||
| } | ||
| const messageId = yield* ids.allocate.message({ | ||
| threadId: request.threadId, | ||
| ordinal: projection.messages.length + 1, | ||
| }); | ||
| const commandId = CommandId.make(`provider-continuation:${messageId}`); | ||
| yield* threads.dispatch({ | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 Medium
🤖 Copy this AI Prompt to have your agent fix this: |
||
| type: "message.dispatch", | ||
| commandId, | ||
| threadId: request.threadId, | ||
| messageId, | ||
| text: request.detail ?? CONTINUATION_MESSAGE_TEXT, | ||
| attachments: [], | ||
| dispatchMode: { type: "queue_after_active" }, | ||
| createdBy: "agent", | ||
| creationSource: "provider", | ||
| }); | ||
| }, | ||
| ); | ||
|
|
||
| yield* requests.take.pipe( | ||
| Effect.flatMap((request) => | ||
| dispatchContinuation(request).pipe( | ||
| Effect.catchCause((cause) => | ||
| Effect.logWarning("orchestration-v2.provider-continuation.dispatch-failed", { | ||
| threadId: request.threadId, | ||
| providerThreadId: request.providerThreadId, | ||
| cause, | ||
| }), | ||
| ), | ||
| ), | ||
| ), | ||
| Effect.forever, | ||
| Effect.forkScoped, | ||
| ); | ||
| }), | ||
| ); | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟡 Medium
t3code/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Line 3848 in 6054588
makeDefaultClaudeAdapterV2readsProviderContinuationRequestsfrom the context, but since it is aContext.Referencewith a no-op default (offer: () => Effect.void), any caller that providesClaudeAdapterV2.layerwithout also supplying the live continuation queue captures the no-op. In that configuration Claude background wake turns are silently dropped, defeating the continuation behavior this PR introduces. Consider addingProviderContinuationRequeststolayer's environment type so the live queue must be provided, or falling back to the live implementation inside the adapter when the default is detected.🤖 Copy this AI Prompt to have your agent fix this: