Skip to content
Merged
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
32 changes: 30 additions & 2 deletions apps/server/src/mcp/OrchestratorMcpService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -451,7 +451,7 @@ describe("OrchestratorMcpService provider resolution", () => {
reason: "Driver 'forkOnly' is not registered in this build.",
checkedAt: "2026-09-13T00:00:00.000Z",
});
const providers: ReadonlyArray<ServerProvider> = [
let providers: ReadonlyArray<ServerProvider> = [
providerSnapshot({
instanceId: codexInstanceId,
driver: ProviderDriverKind.make("codex"),
Expand Down Expand Up @@ -495,7 +495,7 @@ describe("OrchestratorMcpService provider resolution", () => {
Layer.mock(ThreadManagementService)({
getThreadRecords: () => Effect.succeed(parentProjection([])),
}),
Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed(providers) }),
Layer.mock(ProviderRegistry)({ getProviders: Effect.sync(() => providers) }),
adapterRegistryLayer([
codexInstanceId,
ProviderInstanceId.make("claudeAgent"),
Expand All @@ -513,6 +513,34 @@ describe("OrchestratorMcpService provider resolution", () => {
const byId = new Map(
capabilities.providers.map((provider) => [provider.providerInstanceId, provider]),
);
for (const provider of providers) {
assert.deepEqual(
byId.get(provider.instanceId)?.models.map((model) => model.id),
provider.models.map((model) => model.slug),
);
}

providers = providers.map((provider) => ({
...provider,
models: [
...provider.models,
{
slug: `${provider.driver}/custom-model-after-refresh`,
name: "Custom model",
isCustom: true,
capabilities: null,
},
],
}));
const refreshed = yield* service.capabilities(scope);
for (const provider of providers) {
assert.deepEqual(
refreshed.providers
.find((entry) => entry.providerInstanceId === provider.instanceId)
?.models.map((model) => model.id),
provider.models.map((model) => model.slug),
);
}

for (const instanceId of [
codexInstanceId,
Expand Down
6 changes: 3 additions & 3 deletions apps/server/src/mcp/toolkits/orchestrator/tools.ts
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ const threadMetadataDependencies = [

const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", {
description:
"List the V2 provider instances, models, inherited runtime settings, and app-owned orchestration features available to this T3 thread. For a separate top-level thread in a new or existing worktree, use t3_thread_launch with workspaceStrategy.",
"List the V2 provider instances and their current models from the same live catalog as the composer, including configured custom models, inherited runtime settings, and app-owned orchestration features available to this T3 thread. For a separate top-level thread in a new or existing worktree, use t3_thread_launch with workspaceStrategy.",
success: OrchestratorMcpCapabilitiesResult,
failure: OrchestratorMcpFailure,
failureMode: "return",
Expand All @@ -54,7 +54,7 @@ const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", {

export const DelegateTaskTool = Tool.make("delegate_task", {
description:
"Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Prefer native subagent tools for same-provider work when available. Use this for cross-provider work, when native delegation is unavailable, or when the user explicitly requests T3-owned child tasks. The childThreadId is backing storage, not an ordinary top-level thread. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. Prefer mode='async' for long work; mode='wait' blocks until completion or timeout. timeoutMs on mode=wait is only the parent's wait budget and does not cancel the child. waitTimedOut on that wait call means the timeout fired; keep that taskId and read status on later task_status. An async child's completion wakes this thread through a notification, steered into active turns where supported or queued otherwise, so end the turn instead of polling or spawning watchers; use task_status only when the result is needed mid-turn.",
"Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Choose providers and models from orchestrator_capabilities, which uses the same live catalog as the composer. Prefer native subagent tools for same-provider work only when they support the chosen model. Use this for any model missing from the native tool, including same-provider work, for cross-provider work, or for explicitly T3-owned child tasks. The childThreadId is backing storage, not an ordinary top-level thread. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. Prefer mode='async' for long work; mode='wait' blocks until completion or timeout. timeoutMs on mode=wait is only the parent's wait budget and does not cancel the child. waitTimedOut on that wait call means the timeout fired; keep that taskId and read status on later task_status. An async child's completion wakes this thread through a notification, steered into active turns where supported or queued otherwise, so end the turn instead of polling or spawning watchers; use task_status only when the result is needed mid-turn.",
parameters: OrchestratorMcpDelegateTaskInput,
success: OrchestratorMcpDelegateTaskResult,
failure: OrchestratorMcpFailure,
Expand Down Expand Up @@ -143,7 +143,7 @@ const DeleteScheduledTaskTool = Tool.make("delete_scheduled_task", {

export const CreateThreadsTool = Tool.make("create_threads", {
description:
"Create one or more ORDINARY TOP-LEVEL T3 conversations. This is not delegation and does not create child agents/subagents. For delegated work, prefer native subagents within the current provider; call delegate_task for cross-provider or explicitly T3-owned child tasks. Use create_threads for a batch of separate top-level threads sharing this checkout. Prefer t3_thread_launch for a single thread. Both require the user to request separate/new/top-level threads or conversations. Each entry may override provider, model, options, runtime mode, and interaction mode; omitted settings inherit. Project, branch, and worktree always inherit and cannot be overridden here. For independent implementation or a PR stack in its own worktree, use t3_thread_launch with workspaceStrategy instead of asking the agent to create a worktree in its prompt.",
"Create one or more ORDINARY TOP-LEVEL T3 conversations. This is not delegation and does not create child agents/subagents. For delegated work, choose models from orchestrator_capabilities. Prefer native subagents only when they support the chosen model; otherwise call delegate_task, including for same-provider work. Use create_threads for a batch of separate top-level threads sharing this checkout. Prefer t3_thread_launch for a single thread. Both require the user to request separate/new/top-level threads or conversations. Each entry may override provider, model, options, runtime mode, and interaction mode; omitted settings inherit. Project, branch, and worktree always inherit and cannot be overridden here. For independent implementation or a PR stack in its own worktree, use t3_thread_launch with workspaceStrategy instead of asking the agent to create a worktree in its prompt.",
parameters: OrchestratorMcpCreateThreadsInput,
success: OrchestratorMcpCreateThreadsResult,
failure: OrchestratorMcpFailure,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4480,7 +4480,7 @@ describe("ClaudeAdapterV2 background wake turns", () => {
),
);

it.effect.each(["requested", "observed-before", "observed-after", "inherit"] as const)(
it.effect.each(["requested", "observed-before", "observed-after", "inherit", "unknown"] as const)(
"records the subagent model from %s without inheriting the parent override",
(source) =>
Effect.scoped(
Expand Down Expand Up @@ -4524,7 +4524,9 @@ describe("ClaudeAdapterV2 background wake turns", () => {
input: {
description: "Haiku puzzle",
subagent_type: "general-purpose",
model: source === "inherit" ? "inherit" : "haiku",
...(source === "unknown"
? {}
: { model: source === "inherit" ? "inherit" : "haiku" }),
prompt: "Solve the puzzle.",
},
},
Expand Down Expand Up @@ -4562,14 +4564,16 @@ describe("ClaudeAdapterV2 background wake turns", () => {
? observedModel
: source === "inherit"
? parentModel
: "haiku";
: source === "unknown"
? null
: "haiku";
assert.equal(subagents[0]?.subagent.model, initialModel);
assert.equal(
subagents.at(-1)?.subagent.model,
source.startsWith("observed") ? observedModel : initialModel,
);
const child = harness.events.find((event) => event.type === "app_thread.created");
assert.equal(child?.appThread.modelSelection?.model, initialModel);
assert.equal(child?.appThread.modelSelection?.model, initialModel ?? parentModel);
}).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))),
),
);
Expand Down
9 changes: 6 additions & 3 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2431,12 +2431,15 @@ function rememberClaudeSubagentRequestedModel(
const model = firstStringInputField(input, ["model"]);
if (
model === undefined ||
model === "inherit" ||
context.subagentsByToolUseId.has(toolUseId) ||
context.pendingSubagentModelsByToolUseId.has(toolUseId)
)
return;
rememberPendingClaudeSubagentModel(context.pendingSubagentModelsByToolUseId, toolUseId, model);
rememberPendingClaudeSubagentModel(
context.pendingSubagentModelsByToolUseId,
toolUseId,
model === "inherit" ? context.input.modelSelection.model : model,
);
}

type PendingClaudeRuntimeRequest =
Expand Down Expand Up @@ -3386,7 +3389,7 @@ export function makeClaudeAdapterV2(
},
prompt: input.prompt ?? "",
title: input.title ?? null,
model: input.model ?? input.context.input.modelSelection.model,
model: input.model ?? null,
result: null,
startedAt: now,
}),
Expand Down
129 changes: 129 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ import {
import {
makeReplayServerConfig,
makeCodexProviderAdapterRegistryReplayLayer,
withCodexReplayChildMetadata,
} from "./CodexAdapterV2.testkit.ts";

const encodeUnknownJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
Expand Down Expand Up @@ -1490,6 +1491,7 @@ describe("CodexAdapterV2 post-settle continuation", () => {
transcript: CodexReplay.CodexAppServerReplayTranscript,
onEvent: (event: ProviderAdapterV2Event) => Effect.Effect<unknown> = () => Effect.void,
onRequest: (method: string) => Effect.Effect<void> = () => Effect.void,
readChildMetadata?: (threadId: string) => Effect.Effect<unknown>,
) =>
Effect.gen(function* () {
const fileSystem = yield* FileSystem.FileSystem;
Expand All @@ -1509,6 +1511,9 @@ describe("CodexAdapterV2 post-settle continuation", () => {
),
Effect.flatMap((context) =>
Effect.service(CodexClient.CodexAppServerClient).pipe(
Effect.map((client) =>
withCodexReplayChildMetadata(client, transcript, readChildMetadata),
),
Effect.map(
(client) =>
({
Expand Down Expand Up @@ -5622,6 +5627,130 @@ describe("CodexAdapterV2 post-settle continuation", () => {
],
});

it.effect.each([
{ name: "Sol", model: "gpt-5.6-sol" },
{ name: "Fable", model: "gpt-5.6-fable" },
{ name: "Astra", model: "gpt-6-astra" },
{ name: "missing", model: null },
{ name: "invalid", model: null },
{ name: "wrong child", model: null },
])("reads $name child metadata without using the parent model", ({ name, model }) =>
Effect.scoped(
Effect.gen(function* () {
const metadataRead = yield* Deferred.make<void>();
const modelReported = yield* Deferred.make<void>();
const harness = yield* makeCodexReplayHarness(
resumeSubagentTranscript,
(event) =>
event.type === "subagent.updated" && event.subagent.model === model
? Deferred.succeed(modelReported, undefined)
: Effect.void,
undefined,
(threadId) => {
assert.equal(threadId, RESUME_CHILD_THREAD);
return Deferred.succeed(metadataRead, undefined).pipe(
Effect.as(
name === "invalid"
? {}
: {
thread: { id: name === "wrong child" ? "other-child" : threadId },
model: name === "wrong child" ? "gpt-5.6-sol" : model,
},
),
);
},
);
yield* harness.runtime.startTurn(
makeCodexTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now: yield* DateTime.now,
attemptId: RunAttemptId.make("attempt-child-model"),
text: RESUME_PROMPT,
}),
);
yield* Deferred.await(metadataRead);
yield* Deferred.await(modelReported);
yield* TestClock.adjust("100 millis");
yield* harness.firstTerminal;
assert.equal(harness.subagentUpdates().at(-1)?.subagent.model, model);
}).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))),
),
);

it.effect.each(["thread/settings/updated", "model/rerouted"] as const)(
"keeps %s child metadata when an older lookup finishes later",
(method) =>
Effect.scoped(
Effect.gen(function* () {
const releaseMetadata = yield* Deferred.make<void>();
const observed = yield* Deferred.make<void>();
const model = "gpt-5.6-sol";
const notification: CodexReplay.CodexAppServerReplayEntry = {
type: "emit_inbound",
frame: {
method,
params:
method === "model/rerouted"
? {
threadId: RESUME_CHILD_THREAD,
turnId: RESUME_CHILD_TURN_1,
fromModel: "gpt-6-astra",
toModel: model,
reason: "highRiskCyberActivity",
}
: {
threadId: RESUME_CHILD_THREAD,
threadSettings: {
model,
modelProvider: "openai",
cwd: "/workspace",
approvalPolicy: "never",
approvalsReviewer: "auto_review",
collaborationMode: { mode: "default", settings: { model } },
sandboxPolicy: { type: "dangerFullAccess" },
},
},
},
};
const harness = yield* makeCodexReplayHarness(
{
...resumeSubagentTranscript,
entries: resumeSubagentTranscript.entries.flatMap((entry) =>
entry.type === "emit_inbound" && entry.label === "turn/completed/root"
? [entry, notification]
: [entry],
),
},
(event) =>
event.type === "subagent.updated" && event.subagent.model === model
? Deferred.succeed(observed, undefined)
: Effect.void,
undefined,
(threadId) =>
Deferred.await(releaseMetadata).pipe(
Effect.as({ thread: { id: threadId }, model: "gpt-6-astra" }),
),
);
yield* harness.runtime.startTurn(
makeCodexTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now: yield* DateTime.now,
attemptId: RunAttemptId.make("attempt-child-model-update"),
text: RESUME_PROMPT,
}),
);
yield* TestClock.adjust("100 millis");
yield* Deferred.await(observed);
assert.equal(harness.subagentUpdates().at(-1)?.subagent.status, "completed");
yield* Deferred.succeed(releaseMetadata, undefined);
yield* TestClock.adjust("30 seconds");
assert.equal(harness.subagentUpdates().at(-1)?.subagent.model, model);
}).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))),
),
);

it.effect("preserves a subagent result across a trailing empty final and resume", () =>
Effect.scoped(
Effect.gen(function* () {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Path from "effect/Path";
import * as PlatformError from "effect/PlatformError";
import * as Predicate from "effect/Predicate";
import * as Schema from "effect/Schema";

import { ServerConfig } from "../../config.ts";
Expand Down Expand Up @@ -45,6 +46,42 @@ export const CodexOrchestratorReplayHarnessError = Schema.Union([
]);
export type CodexOrchestratorReplayHarnessError = typeof CodexOrchestratorReplayHarnessError.Type;

export function withCodexReplayChildMetadata(
client: CodexClient.CodexAppServerClient["Service"],
transcript: CodexReplay.CodexAppServerReplayTranscript,
readMetadata: (threadId: string) => Effect.Effect<unknown> = (threadId) =>
Effect.succeed({ thread: { id: threadId }, model: null }),
): CodexClient.CodexAppServerClient["Service"] {
const childThreadIds = new Set(
transcript.entries.flatMap((entry) => {
if (entry.type !== "emit_inbound" || !Predicate.isObject(entry.frame)) return [];
const params = entry.frame.params;
if (!Predicate.isObject(params) || !Predicate.isObject(params.item)) return [];
const item = params.item;
if (item.type === "subAgentActivity" && typeof item.agentThreadId === "string") {
return [item.agentThreadId];
}
return item.type === "collabAgentToolCall" && Array.isArray(item.receiverThreadIds)
? item.receiverThreadIds.filter(Predicate.isString)
: [];
}),
);
return {
...client,
raw: {
...client.raw,
request: (method, params) =>
method === "thread/resume" &&
Predicate.isObject(params) &&
params.excludeTurns === true &&
typeof params.threadId === "string" &&
childThreadIds.has(params.threadId)
? readMetadata(params.threadId)
: client.raw.request(method, params),
},
};
}

function metadataFromTranscript(transcript: ProviderReplayTranscript): {
readonly provider?: string;
readonly protocol?: string;
Expand Down Expand Up @@ -166,6 +203,7 @@ export function makeCodexProviderAdapterRegistryReplayLayer(input: {
),
);
return yield* Effect.service(CodexClient.CodexAppServerClient).pipe(
Effect.map((client) => withCodexReplayChildMetadata(client, input.transcript)),
Effect.provide(context),
);
}),
Expand Down
Loading
Loading