Skip to content
Open
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
94 changes: 93 additions & 1 deletion apps/server/src/mcp/OrchestratorMcpService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import {
type OrchestrationV2ThreadShell,
type ScheduledTask,
ScheduledTaskId,
type Project,
ProjectId,
ProviderDriverKind,
ProviderInstanceId,
Expand Down Expand Up @@ -1818,7 +1819,7 @@ describe("OrchestratorMcpService provider resolution", () => {
projectId,
title: "MCP parent",
modelSelection: { instanceId: codexInstanceId, model: "gpt-5.4" },
runtimeMode: "full-access",
runtimeMode: "auto",
interactionMode: "default",
branch: null,
worktreePath: cwd,
Expand Down Expand Up @@ -1863,6 +1864,26 @@ describe("OrchestratorMcpService provider resolution", () => {
clientRequestId: "empty",
});

// The user raised this thread above the caller after its creation.
const raised = yield* service.createThreads(scope, {
...request,
clientRequestId: "raised",
});
const raisedThreadId = raised.threads[0]!.threadId;
yield* orchestrator.dispatch({
type: "thread.runtime-mode.set",
commandId: CommandId.make("user-raise"),
threadId: raisedThreadId,
runtimeMode: "full-access",
});
const raisedError = yield* service
.createThreads(scope, { ...withPrompt, clientRequestId: "raised" })
.pipe(Effect.flip);
assert.equal(raisedError.code, "runtime_mode_escalation_denied");
const unprompted = yield* orchestrator.getThreadProjection(raisedThreadId);
assert.equal(unprompted.messages.length, 0);
assert.equal(unprompted.runs.length, 0);

yield* settings.updateSettings({
projectSettingsOverrides: {
[projectId]: {
Expand Down Expand Up @@ -2350,6 +2371,77 @@ describe("OrchestratorMcpService provider resolution", () => {
}),
);

it.effect("refuses a replayed create over a task the user raised", () =>
Effect.gen(function* () {
const saved = yield* Ref.make<ReadonlyArray<ScheduledTask>>([]);
const upserted = yield* Ref.make(0);
const project = {
id: projectId,
title: "Scheduled",
workspaceRoot: "/tmp/scheduled",
defaultModelSelection: task({}).modelSelection,
scripts: [],
createdAt: "2026-10-05T00:00:00.000Z",
updatedAt: "2026-10-05T00:00:00.000Z",
deletedAt: null,
} as Project;
const mcp = yield* OrchestratorMcpService.OrchestratorMcpService.pipe(
Effect.provide(
layerOrchestratorService.pipe(
Layer.provide(
Layer.mergeAll(
NodeServices.layer,
Layer.mock(ThreadManagementService.ThreadManagementService)({
getThreadShell: () => Effect.succeed(null),
}),
ServerSettings.layerTest(),
Layer.mock(ProviderRegistry.ProviderRegistry)({
getProviders: Effect.succeed([]),
}),
Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({
list: () => Effect.succeed([]),
}),
Layer.mock(ProjectService.ProjectService)({
getById: () => Effect.succeedSome(project),
}),
Layer.mock(SecretRequests.SecretRequests)({}),
Layer.mock(RemoteDelegation.RemoteDelegation)({}),
Layer.mock(ScheduledTaskService.ScheduledTaskService)({
list: () => Effect.map(Ref.get(saved), (tasks) => ({ tasks })),
upsert: (input) =>
Effect.gen(function* () {
yield* Ref.update(upserted, (count) => count + 1);
const saving = task({
...(input.id === undefined ? {} : { id: input.id }),
prompt: input.prompt,
runtimeMode: input.runtimeMode,
interactionMode: input.interactionMode,
});
yield* Ref.set(saved, [saving]);
return { task: saving };
}),
}),
),
),
),
),
);
const request = {
projectId,
prompt: "Check in",
schedule: { type: "interval", everyMs: 3_600_000 },
clientRequestId: "schedule-retry",
} as const;
yield* mcp.scheduleTask(supervisedClient, request);
yield* Ref.update(saved, (tasks) =>
tasks.map((task) => ({ ...task, runtimeMode: "full-access" as const })),
);
const error = yield* mcp.scheduleTask(supervisedClient, request).pipe(Effect.flip);
assert.equal(error.code, "runtime_mode_escalation_denied");
assert.equal(yield* Ref.get(upserted), 1);
}),
);

it.effect("reports a saved task even when its bound thread cannot be read", () =>
Effect.gen(function* () {
const upserted = yield* Ref.make(0);
Expand Down
75 changes: 59 additions & 16 deletions apps/server/src/mcp/OrchestratorMcpService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ import {
ProviderRequestKind,
type RuntimeMode,
type ScheduledTask,
ScheduledTaskId,
type ScheduledTaskUpsertInput,
type ServerProvider,
type ServerProviderModel,
Expand Down Expand Up @@ -1823,11 +1824,23 @@ const make = Effect.gen(function* () {
if (task === undefined) {
return yield* failure("task_not_found", `Scheduled task ${scheduledTaskId} was not found.`);
}
yield* assertScheduledTaskWithinLimits(task, limits);
return task;
});

const assertScheduledTaskWithinLimits = Effect.fn(
"OrchestratorMcpService.assertScheduledTaskWithinLimits",
)(function* (
task: ScheduledTask,
limits: {
readonly runtimeMode: RuntimeMode;
readonly interactionMode: ProviderInteractionMode;
},
) {
for (const modes of yield* scheduledTaskRunModes(task)) {
yield* resolveRuntimeMode(limits.runtimeMode, modes.runtimeMode);
yield* resolveInteractionMode(limits.interactionMode, modes.interactionMode);
}
return task;
});

return OrchestratorMcpService.of({
Expand All @@ -1852,6 +1865,33 @@ const make = Effect.gen(function* () {
const derivedTitle = input.prompt.split("\n")[0]?.trim() ?? "";
const title =
input.title ?? (derivedTitle.length > 0 ? derivedTitle.slice(0, 80) : "Scheduled task");
// Scope the idempotency key by provider session so two callers
// reusing the same clientRequestId cannot collide on one task row.
const commandId =
input.clientRequestId === undefined
? undefined
: stableCommandId({
scope,
requestKey: input.clientRequestId,
operation: "schedule-task",
});
const replay =
commandId === undefined
? undefined
: { id: ScheduledTaskId.make(`scheduled-task:${commandId}`), commandId };
if (replay !== undefined) {
// A replay saves over the task it created, which the user may have
// since raised above this caller's modes.
const { tasks } = yield* scheduledTasks
.list()
.pipe(
Effect.mapError((error) =>
failure("orchestration_error", `Could not load scheduled task: ${error.message}`),
),
);
const replayed = tasks.find((task) => task.id === replay.id);
if (replayed !== undefined) yield* assertScheduledTaskWithinLimits(replayed, limits);
}
const upsertInput: ScheduledTaskUpsertInput = {
title,
prompt: input.prompt,
Expand All @@ -1865,17 +1905,7 @@ const make = Effect.gen(function* () {
interactionMode: limits.interactionMode,
createdBy: "agent",
creationSource: "mcp",
// Scope the idempotency key by provider session so two callers
// reusing the same clientRequestId cannot collide on one task row.
...(input.clientRequestId === undefined
? {}
: {
commandId: stableCommandId({
scope,
requestKey: input.clientRequestId,
operation: "schedule-task",
}),
}),
...replay,
};
const { task } = yield* scheduledTasks
.upsert(upsertInput)
Expand Down Expand Up @@ -2710,13 +2740,26 @@ const make = Effect.gen(function* () {
dispatchMode: { type: "start_immediately" },
})
.pipe(
// A replay prompts the thread it created, which the user may have since
// raised above this caller's modes; the orchestrator refuses under the lock.
Effect.provideService(DispatchModeLimit, {
runtimeMode: parent.thread.runtimeMode,
interactionMode: parent.thread.interactionMode,
}),
Effect.mapError((error) =>
error._tag === "ThreadManagementProviderInstanceDisabledError"
? threadManagementFailure(error)
: failure(
"orchestration_error",
`Unable to start thread ${index + 1}: ${errorMessage(error)}`,
),
: error._tag === "OrchestratorThreadAboveModeLimitError"
? failure(
error.mode === "runtime"
? "runtime_mode_escalation_denied"
: "interaction_mode_escalation_denied",
`Thread ${error.threadId} now runs in ${error.runtimeMode}/${error.interactionMode} mode, above this caller's.`,
)
: failure(
"orchestration_error",
`Unable to start thread ${index + 1}: ${errorMessage(error)}`,
),
),
);
}
Expand Down
Loading