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
21 changes: 10 additions & 11 deletions apps/server/src/mcp/OrchestratorMcpService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -91,15 +91,6 @@ describe("OrchestratorMcpService", () => {
const parentProjection = {
thread: { id: parentThreadId },
runs: [],
contextTransfers: [
{
id: resultTransferId,
type: "subagent_result",
sourceThreadId: childThreadId,
targetThreadId: parentThreadId,
sourcePoint: { runId: childRunId },
},
],
subagents: [
{
id: taskId,
Expand Down Expand Up @@ -131,7 +122,15 @@ describe("OrchestratorMcpService", () => {
completedAt: DateTime.makeUnsafe("2026-10-03T10:05:00Z"),
},
],
contextTransfers: [],
contextTransfers: [
{
id: resultTransferId,
type: "subagent_result",
sourceThreadId: childThreadId,
targetThreadId: parentThreadId,
sourcePoint: { runId: childRunId },
},
],
messages: [
{
id: MessageId.make("message:mcp-ack-result"),
Expand Down Expand Up @@ -160,6 +159,7 @@ describe("OrchestratorMcpService", () => {
return hasNestedWork
? {
...childProjection,
contextTransfers: [],
subagents: [
{ ...parentProjection.subagents[0]!, status: "running" as const },
],
Expand All @@ -169,7 +169,6 @@ describe("OrchestratorMcpService", () => {
const snapshot = hasNestedWork
? {
...parentProjection,
contextTransfers: [],
subagents: parentProjection.subagents.map((task) => ({
...task,
result: null,
Expand Down
143 changes: 43 additions & 100 deletions apps/server/src/mcp/OrchestratorMcpService.ts

Large diffs are not rendered by default.

85 changes: 81 additions & 4 deletions apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -216,13 +216,34 @@ describe("orchestrator MCP toolkit", () => {
},
CodexOrchestratorReplayHarness,
);
// SQL spans carry no bound parameters, so this records which thread each tool read.
const threadRecordReads: Array<{
readonly threadId: ThreadId;
readonly fields: ReadonlyArray<string>;
}> = [];
const layerOrchestration = Layer.merge(
layerOrchestrator,
ThreadManagementService.layer.pipe(
Layer.provide(layerOrchestrator),
Layer.provide(ServerSettings.layerTest()),
Layer.effect(
ThreadManagementService.ThreadManagementService,
Effect.gen(function* () {
const service = yield* ThreadManagementService.ThreadManagementService;
return ThreadManagementService.ThreadManagementService.of({
...service,
getThreadRecords: (threadId, fields, filter) =>
Effect.sync(() => threadRecordReads.push({ threadId, fields })).pipe(
Effect.andThen(service.getThreadRecords(threadId, fields, filter)),
),
});
}),
).pipe(
Layer.provide(
CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
ThreadManagementService.layer.pipe(
Layer.provide(layerOrchestrator),
Layer.provide(ServerSettings.layerTest()),
Layer.provide(
CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
),
),
),
),
);
Expand Down Expand Up @@ -603,6 +624,9 @@ describe("orchestrator MCP toolkit", () => {
);
const transferReads: Array<string> = [];
const messageReads: Array<string> = [];
const runReads: Array<string> = [];
const projectionTableReads: Array<string> = [];
const acknowledgementReads: Array<string> = [];
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options);
Expand All @@ -611,11 +635,29 @@ describe("orchestrator MCP toolkit", () => {
end(endTime, exit);
const query = span.attributes.get("db.query.text");
if (typeof query !== "string") return;
let parent = Option.getOrUndefined(span.parent);
while (parent?._tag === "Span") {
if (
parent.name === "Orchestrator.dispatchDelegatedTaskCompletionDeliveryResolution"
) {
acknowledgementReads.push(query);
break;
}
parent = Option.getOrUndefined(parent.parent);
}
for (const [, table] of query.matchAll(
/FROM orchestration_v2_projection_(\w+)/gi,
)) {
projectionTableReads.push(table!);
}
if (
/SELECT[\s\S]*FROM orchestration_v2_projection_context_transfers/i.test(query)
) {
transferReads.push(query);
}
if (/SELECT\s+payload_json\s+FROM orchestration_v2_projection_runs/i.test(query)) {
runReads.push(query);
}
if (
/SELECT\s+payload_json\s+FROM orchestration_v2_projection_messages/i.test(query)
) {
Expand Down Expand Up @@ -687,18 +729,53 @@ describe("orchestrator MCP toolkit", () => {
expect(read.isError).not.toBe(true);
expect(transferReads.length).toBeGreaterThan(0);
expect(read.structuredContent).toEqual(parentlessRead.structuredContent);
// The acknowledgement reads the task and its spawning run by id, not the parent's history.
expect(acknowledgementReads.length).toBeGreaterThan(0);
expect(
acknowledgementReads.filter(
(query) =>
/FROM orchestration_v2_projection_(nodes|run_attempts)\b/.test(query) ||
(/FROM orchestration_v2_projection_runs\b/.test(query) &&
!query.includes("AND run_id IN")) ||
(/FROM orchestration_v2_projection_subagents\b/.test(query) &&
!query.includes("AND subagent_id IN")),
),
).toEqual([]);
const afterRead = yield* orchestrator.getThreadProjection(parentThreadId);
expect(
afterRead.subagents.find((task) => task.id === queuedFollowup.taskId)
?.completionDelivery?.state,
).toBe("acknowledged");

const messageReadsBeforeCapabilities = messageReads.length;
const tableReadsBeforeCapabilities = projectionTableReads.length;
const capabilities = yield* invoke("orchestrator_capabilities", {}).pipe(
Effect.withTracer(tracer),
);
expect(capabilities.isError).not.toBe(true);
expect(messageReads).toHaveLength(messageReadsBeforeCapabilities);
// The caller check reads only the thread row, never its history.
expect(new Set(projectionTableReads.slice(tableReadsBeforeCapabilities))).toEqual(
new Set(["threads"]),
);
const runReadsBeforeStatus = runReads.length;
const transferReadsBeforeStatus = transferReads.length;
const threadRecordReadsBeforeStatus = threadRecordReads.length;
const statusAgain = yield* invoke("task_status", { taskId: delegated.taskId }).pipe(
Effect.withTracer(tracer),
);
expect(statusAgain.isError).not.toBe(true);
// A task wait reads result transfers and runs from the child; the parent's stay unread.
expect(runReads).toHaveLength(runReadsBeforeStatus + 1);
expect(transferReads).toHaveLength(transferReadsBeforeStatus + 1);
expect(
threadRecordReads
.slice(threadRecordReadsBeforeStatus)
.filter(
({ fields }) => fields.includes("runs") || fields.includes("contextTransfers"),
)
.map(({ threadId }) => threadId),
).toEqual([delegated.childThreadId]);
}).pipe(Effect.provide(layerTest));
}),
),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1055,6 +1055,73 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => {
}),
);

it.effect("acknowledging a queued wake's results rewrites it, then cancels it", () =>
Effect.gen(function* () {
const orchestrator = yield* Orchestrator.OrchestratorV2;
const now = yield* DateTime.now;
const threadId = ThreadId.make("queued-wake-ack-parent");
const runId = RunId.make("queued-wake-ack-origin");
const taskId = NodeId.make("queued-wake-ack-first");
const siblingId = NodeId.make("queued-wake-ack-sibling");
yield* seedParentWithTerminalTask({
threadId,
projectId: ProjectId.make("queued-wake-ack-project"),
runId,
rootNodeId: NodeId.make("queued-wake-ack-root"),
taskId,
deliveryState: "claimed",
deliveryTaskIds: [taskId, siblingId],
siblingTasks: [{ id: siblingId, deliveryState: "claimed" }],
completionWake: "always",
now,
});
const delivery = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!
.delegatedCompletion!.delivery!;
yield* orchestrator.dispatch({
type: "message.dispatch",
createdBy: "agent",
creationSource: "server",
commandId: CommandId.make("queued-wake-ack-wake"),
threadId,
messageId: delivery.messageId,
text: "Delegated tasks completed",
attachments: [],
modelSelection,
dispatchMode: { type: "queue_after_active" },
delegatedCompletion: {
parentRunId: runId,
generation: delivery.generation,
taskIds: delivery.taskIds,
},
});
const acknowledge = (id: NodeId) =>
orchestrator.dispatch({
type: "delegated_task.completion-delivery.acknowledge",
commandId: CommandId.make(`queued-wake-ack:${id}`),
parentThreadId: threadId,
taskId: id,
observedByRunId: runId,
});

yield* acknowledge(taskId);
const rewritten = yield* orchestrator.getThreadProjection(threadId);
const wake = rewritten.runs.find((run) => run.userMessageId === delivery.messageId)!;
assert.equal(wake.status, "queued");
const message = rewritten.messages.find((row) => row.id === delivery.messageId);
assert.deepEqual(message?.delegatedCompletion?.taskIds, [siblingId]);
assert.equal(message?.text, `Delegated task ${siblingId} [completed]:\nchild finished`);

yield* acknowledge(siblingId);
const cancelled = yield* orchestrator.getThreadProjection(threadId);
assert.equal(cancelled.runs.find((run) => run.id === wake.id)?.status, "cancelled");
assert.equal(
cancelled.nodes.find((node) => node.id === wake.rootNodeId)?.status,
"cancelled",
);
assert.isNull(cancelled.runs.find((run) => run.id === runId)?.delegatedCompletion?.delivery);
}),
);

it.effect.each([
{ command: "acknowledge", before: "running", after: "failed", capture: false },
{ command: "acknowledge", before: "waiting", after: "completed", capture: true },
Expand Down
Loading
Loading