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
50 changes: 39 additions & 11 deletions apps/server/src/mcp/OrchestratorMcpService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -72,8 +72,13 @@ const layerOrchestratorService = OrchestratorMcpService.layer.pipe(
);

describe("OrchestratorMcpService", () => {
it.effect("retries terminal acknowledgement with a fresh command id", () =>
const acknowledgementTest =
"retries terminal acknowledgement with a fresh command id (follow-up: %s)";
const cases = [false, true, "never-started", "queued"] as const;
it.effect.each(cases)(acknowledgementTest, (followup) =>
Effect.gen(function* () {
const failedBeforeStart = followup === "never-started";
const queued = followup === "queued";
const parentThreadId = ThreadId.make("thread:mcp-ack-parent");
const childThreadId = ThreadId.make("thread:mcp-ack-child");
const childRunId = RunId.make("run:mcp-ack-child");
Expand All @@ -92,7 +97,7 @@ describe("OrchestratorMcpService", () => {
childThreadId,
driver: "codex",
model: "gpt-5.6-terra",
result: "terminal result",
result: failedBeforeStart || queued ? null : "terminal result",
completionDelivery: { state: "pending" },
},
],
Expand All @@ -102,9 +107,13 @@ describe("OrchestratorMcpService", () => {
runs: [
{
id: childRunId,
...(followup ? { delegatedTaskId: taskId } : {}),
ordinal: 1,
status: "completed",
startedAt: DateTime.makeUnsafe("2026-10-03T10:00:00Z"),
...(failedBeforeStart || queued
? { ordinal: 3, status: failedBeforeStart ? "failed" : "queued", startedAt: null }
: {}),
completedAt: DateTime.makeUnsafe("2026-10-03T10:10:00Z"),
},
{
Expand All @@ -117,14 +126,16 @@ describe("OrchestratorMcpService", () => {
],
contextTransfers: [],
messages: [],
turnItems: [],
subagents: [],
providerThreads: [],
runtimeRequests: [],
} as unknown as OrchestrationV2ThreadProjection;
let hasNestedWork = true;
let hasNestedWork = !failedBeforeStart && !queued;
const layerDependencies = Layer.mergeAll(
NodeServices.layer,
Layer.mock(ThreadManagementService.ThreadManagementService)({
delegatedTaskResultPending: () => Effect.succeed(false),
getThreadRecords: (threadId) =>
Effect.succeed(
threadId === parentThreadId
Expand All @@ -142,7 +153,11 @@ describe("OrchestratorMcpService", () => {
? {
...childProjection,
subagents: [
{ ...parentProjection.subagents[0]!, status: "running" as const },
{
...parentProjection.subagents[0]!,
runId: childRunId,
status: "running" as const,
},
],
}
: childProjection,
Expand All @@ -154,7 +169,7 @@ describe("OrchestratorMcpService", () => {
]).pipe(
Effect.andThen(Ref.updateAndGet(acknowledgementAttempts, (count) => count + 1)),
Effect.flatMap((attempt) =>
attempt === 1
attempt === 1 && !hasNestedWork
? Effect.fail(new Error("simulated acknowledgement failure") as never)
: Effect.succeed({} as never),
),
Expand Down Expand Up @@ -185,16 +200,29 @@ describe("OrchestratorMcpService", () => {

yield* Effect.gen(function* () {
const service = yield* OrchestratorMcpService.OrchestratorMcpService;
const pending = yield* service.taskStatus(scope, taskId);
assert.equal(pending.status, "running");
assert.equal(pending.workState, "waiting_for_children");
assert.isNull(pending.summary);
assert.equal(yield* Ref.get(acknowledgementAttempts), 0);
hasNestedWork = false;
if (queued) {
const pending = yield* service.taskStatus(scope, taskId);
assert.equal(pending.status, "queued");
assert.equal(pending.workState, "working");
return;
}
if (!failedBeforeStart) {
const pending = yield* service.taskStatus(scope, taskId);
assert.equal(pending.status, "running");
assert.equal(pending.workState, "waiting_for_children");
assert.isNull(pending.summary);
assert.equal(yield* Ref.get(acknowledgementAttempts), 0);
hasNestedWork = false;
}
const error = yield* service.taskStatus(scope, taskId).pipe(Effect.flip);
assert.equal(error.code, "orchestration_error");

const result = yield* service.taskStatus(scope, taskId);
if (failedBeforeStart) {
assert.equal(result.status, "failed");
assert.equal(result.workState, "result_available");
return;
}
assert.equal(result.status, "completed");
assert.equal(result.summary, "terminal result");
assert.equal(result.latestTerminalRunId, childRunId);
Expand Down
15 changes: 12 additions & 3 deletions apps/server/src/mcp/OrchestratorMcpService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ import {
DispatchModeLimit,
type DispatchModeRefusal,
} from "../orchestration-v2/DispatchModeLimit.ts";
import { runOutcome } from "../orchestration-v2/RestartBackgroundNote.ts";
import { neverStarted, runOutcome } from "../orchestration-v2/RestartBackgroundNote.ts";
import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
Expand Down Expand Up @@ -1441,11 +1441,20 @@ const make = Effect.gen(function* () {
.filter((run) => run.delegatedTaskId === task.id)
.toSorted((a, b) => b.ordinal - a.ordinal)[0]
: progress.resultRun;
const taskRuns = childControls.runs.filter((run) => run.delegatedTaskId === task.id);
const taskWorkState =
childRun?.delegatedTaskId === task.id
? isTerminalTaskStatus(taskStatusForRun(taskResultRun))
? taskResultRun !== undefined &&
isTerminalTaskStatus(taskStatusForRun(taskResultRun)) &&
taskRuns.every(neverStarted)
? "result_available"
: "working"
: delegatedTaskProgress({
...childControls,
runs: taskRuns,
subagents: childControls.subagents.filter((subagent) =>
taskRuns.some((run) => run.id === subagent.runId),
),
}).state
: progress.state;
const resultRunIds = [
...new Set(
Expand Down
Loading
Loading