Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
527 changes: 523 additions & 4 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts

Large diffs are not rendered by default.

236 changes: 223 additions & 13 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Medium

const makeDefaultClaudeAdapterV2 = Effect.fn("ClaudeAdapterV2.layer")(function* () {

makeDefaultClaudeAdapterV2 reads ProviderContinuationRequests from the context, but since it is a Context.Reference with a no-op default (offer: () => Effect.void), any caller that provides ClaudeAdapterV2.layer without 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 adding ProviderContinuationRequests to layer'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:
In file @apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts around line 3848:

`makeDefaultClaudeAdapterV2` reads `ProviderContinuationRequests` from the context, but since it is a `Context.Reference` with a no-op default (`offer: () => Effect.void`), any caller that provides `ClaudeAdapterV2.layer` without 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 adding `ProviderContinuationRequests` to `layer`'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.

Large diffs are not rendered by default.

6 changes: 6 additions & 0 deletions apps/server/src/orchestration-v2/ProviderAdapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -474,6 +474,12 @@ export interface ProviderAdapterV2SessionRuntime {
* Adapter runtimes may omit this and expose only their single-consumer event stream.
*/
readonly subscribeEvents?: Effect.Effect<ProviderAdapterV2EventSubscription>;
/**
* Adapters whose native runtime can hold pending work outside an active
* turn (for example Claude background tasks and their wake turns) report it
* here so the session manager defers idle release while it is pending.
*/
readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
readonly ensureThread: (
input: ProviderAdapterV2EnsureThreadInput,
) => Effect.Effect<OrchestrationV2ProviderThread, ProviderAdapterV2Error>;
Expand Down
39 changes: 39 additions & 0 deletions apps/server/src/orchestration-v2/ProviderContinuationRequests.ts
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),
};
}),
);
72 changes: 72 additions & 0 deletions apps/server/src/orchestration-v2/ProviderContinuationService.ts
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({

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Medium orchestration-v2/ProviderContinuationService.ts:42

dispatchContinuation builds the message.dispatch without passing request.providerThreadId or request.driver, so the orchestrator resolves the target provider thread from the thread's current active/model state. If the app thread has switched providers or has no active provider thread when the wake arrives, the continuation run starts on the wrong (or empty) native thread and the buffered wake messages are drained from an unrelated buffer — the wake turn silently vanishes. Consider propagating providerThreadId and driver into the dispatch payload so the run targets the originating native thread.

🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/orchestration-v2/ProviderContinuationService.ts around line 42:

`dispatchContinuation` builds the `message.dispatch` without passing `request.providerThreadId` or `request.driver`, so the orchestrator resolves the target provider thread from the thread's current active/model state. If the app thread has switched providers or has no active provider thread when the wake arrives, the continuation run starts on the wrong (or empty) native thread and the buffered wake messages are drained from an unrelated buffer — the wake turn silently vanishes. Consider propagating `providerThreadId` and `driver` into the dispatch payload so the run targets the originating native thread.

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,
);
}),
);
Loading
Loading