From 4f430472ad2e920bb13635ece97768af653e7a6e Mon Sep 17 00:00:00 2001 From: Mnigos Date: Sun, 11 Oct 2026 13:01:53 +0200 Subject: [PATCH 1/4] fix(mcp): t3_thread_read tells agents which thread sent a message Messages sent from another thread already record senderThreadId, and the web and mobile timelines use it, but t3_thread_read items left it out, so the receiving agent could not tell which thread sent a message. Timeline items now carry an optional senderThreadId, included only when the message has one. --- .../OrchestratorMcpService.activity.test.ts | 109 ++++++++++++++++++ apps/server/src/mcp/OrchestratorMcpService.ts | 1 + packages/contracts/src/orchestratorMcp.ts | 2 + 3 files changed, 112 insertions(+) diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index 7c6f9d28c950..41aaa210541b 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -519,3 +519,112 @@ it("readThread and sendToThread reach threads in other projects", async () => { expect(staleDelete.code).toBe("parent_not_active"); }).pipe(Effect.provide(layer), Effect.runPromise); }); + +it("readThread names the thread that sent a message with t3_thread_send", async () => { + const userMessageRow = (position: number, messageId: string, text: string) => ({ + position, + visibility: "local", + sourceThreadId: parentThreadId, + sourceItemId: `item-${messageId}`, + item: { + id: `item-${messageId}`, + type: "user_message", + messageId, + runId: activeRunId, + status: "completed", + title: null, + text, + createdAt: now, + updatedAt: now, + }, + }); + const visibleTurnItems = [ + userMessageRow(0, "message-from-child", "Child reporting in"), + userMessageRow(1, "message-from-user", "Thanks"), + ]; + const projection = { + thread: baseThread({ + threadId: parentThreadId, + title: "Parent", + instanceId: parentInstanceId, + model: "gpt-5.4", + }), + runs: [makeRun({ id: activeRunId, ordinal: 1, status: "running" })], + visibleTurnItems, + runtimeRequests: [], + messages: [ + { + id: "message-from-child", + threadId: parentThreadId, + role: "user", + text: "Child reporting in", + createdBy: "agent", + creationSource: "mcp", + senderThreadId: childThreadId, + }, + { + id: "message-from-user", + threadId: parentThreadId, + role: "user", + text: "Thanks", + createdBy: "user", + creationSource: "web", + }, + ], + contextTransfers: [], + subagents: [], + updatedAt: now, + } as unknown as OrchestrationV2ThreadProjection; + + const layer = OrchestratorMcpService.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getTimelinePage: () => + Effect.succeed({ + items: visibleTurnItems, + totalItems: visibleTurnItems.length, + hasMore: false, + } as never), + getThreadRecords: (threadId) => + threadId === parentThreadId + ? Effect.succeed(projection) + : Effect.die(`unexpected thread ${threadId}`), + getThreadShell: (threadId) => + Effect.succeed( + threadId === parentThreadId + ? (projection.thread as unknown as OrchestrationV2ThreadShell) + : null, + ), + getProjectThreadRecords: (input) => + input.threadId === parentThreadId + ? Effect.succeed(projection) + : Effect.die(`unexpected thread ${input.threadId}`), + } satisfies Partial), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([]), + } satisfies Partial), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({ + list: () => Effect.succeed({ tasks: [] }), + } satisfies Partial), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ + list: () => Effect.succeed([]), + } satisfies Partial), + NodeCrypto.layer, + ), + ), + ); + + await Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const result = yield* service.readThread(makeScope(), { threadId: parentThreadId }); + const [fromChild, fromUser] = result.items; + expect(fromChild?.messageId).toBe("message-from-child"); + expect(fromChild?.senderThreadId).toBe(childThreadId); + expect(fromUser?.messageId).toBe("message-from-user"); + // Messages without a sending thread omit the field rather than returning null. + expect(fromUser).not.toHaveProperty("senderThreadId"); + }).pipe(Effect.provide(layer), Effect.runPromise); +}); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 872163b660d1..4d357818e339 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -855,6 +855,7 @@ function timelineItem(input: { messageId, createdBy: message?.createdBy ?? null, creationSource: message?.creationSource ?? null, + ...(message?.senderThreadId === undefined ? {} : { senderThreadId: message.senderThreadId }), type: input.row.item.type, status: input.row.item.status, title: input.row.item.title, diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index f0ffa07bd401..4e65d874ea9c 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -403,6 +403,8 @@ export const OrchestratorMcpThreadTimelineItem = Schema.Struct({ messageId: Schema.NullOr(MessageId), createdBy: Schema.NullOr(OrchestrationV2Actor), creationSource: Schema.NullOr(OrchestrationV2CreationSource), + /** The agent thread that sent this message to another thread; omitted for other messages. */ + senderThreadId: Schema.optional(ThreadId), type: Schema.String, status: OrchestrationV2TurnItemStatus, title: Schema.NullOr(Schema.String), From d73854e842b1afaad94157e3e66911a7d53dab50 Mon Sep 17 00:00:00 2001 From: Mnigos Date: Sun, 11 Oct 2026 13:15:48 +0200 Subject: [PATCH 2/4] test(mcp): sender test provides the server settings layer The new readThread sender test now provides ServerSettings.layerTest(), which OrchestratorMcpService requires since main added the server settings dependency. --- apps/server/src/mcp/OrchestratorMcpService.activity.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index db8ab1b70d91..9b4f8f8de1e8 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -606,6 +606,7 @@ it("readThread names the thread that sent a message with t3_thread_send", async ? Effect.succeed(projection) : Effect.die(`unexpected thread ${input.threadId}`), } satisfies Partial), + ServerSettings.layerTest(), Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]), } satisfies Partial), From 6887a6eafc360f86ed47e0bc75dee21f5b1cbd7d Mon Sep 17 00:00:00 2001 From: Mnigos Date: Thu, 15 Oct 2026 15:47:02 +0200 Subject: [PATCH 3/4] test(mcp): sender thread test uses main's orchestrator test layer --- apps/server/src/mcp/OrchestratorMcpService.activity.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index 1855fbfe7b5e..bafd33907736 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -588,7 +588,7 @@ it("readThread names the thread that sent a message with t3_thread_send", async updatedAt: now, } as unknown as OrchestrationV2ThreadProjection; - const layer = OrchestratorMcpService.layer.pipe( + const layer = layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ From 6db50a48c5cb209fc8252bbffb2733c1d7d2e531 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 02:30:28 +0000 Subject: [PATCH 4/4] test(mcp): keep sender attribution regression focused --- .../OrchestratorMcpService.activity.test.ts | 124 +++++------------- packages/contracts/src/orchestratorMcp.ts | 1 - 2 files changed, 36 insertions(+), 89 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index 022f16972d9f..82c49485270f 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -28,7 +28,7 @@ import { OrchestratorDispatchError } from "../orchestration-v2/Orchestrator.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import type * as McpInvocationContext from "./McpInvocationContext.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; -import { liveThreadShell } from "./McpToolAccess.testkit.ts"; +import { idleThreadProjection, liveThreadShell } from "./McpToolAccess.testkit.ts"; const layerOrchestratorService = OrchestratorMcpService.layer.pipe( Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), @@ -687,99 +687,48 @@ it("sendToThread reports why the orchestrator refused the message", async () => }).pipe(Effect.provide(layer), Effect.runPromise); }); -it("readThread names the thread that sent a message with t3_thread_send", async () => { - const userMessageRow = (position: number, messageId: string, text: string) => ({ - position, - visibility: "local", - sourceThreadId: parentThreadId, - sourceItemId: `item-${messageId}`, - item: { - id: `item-${messageId}`, - type: "user_message", - messageId, - runId: activeRunId, - status: "completed", - title: null, - text, - createdAt: now, - updatedAt: now, - }, - }); - const visibleTurnItems = [ - userMessageRow(0, "message-from-child", "Child reporting in"), - userMessageRow(1, "message-from-user", "Thanks"), +it("readThread exposes the stored sender thread only for thread-sent messages", async () => { + const shell = liveThreadShell(parentThreadId); + const messages = [ + { id: "message-from-child", text: "Child reporting in", senderThreadId: childThreadId }, + { id: "message-from-user", text: "Thanks" }, ]; const projection = { - thread: baseThread({ - threadId: parentThreadId, - title: "Parent", - instanceId: parentInstanceId, - model: "gpt-5.4", - }), - runs: [makeRun({ id: activeRunId, ordinal: 1, status: "running" })], - visibleTurnItems, - runtimeRequests: [], - messages: [ - { - id: "message-from-child", - threadId: parentThreadId, - role: "user", - text: "Child reporting in", - createdBy: "agent", - creationSource: "mcp", - senderThreadId: childThreadId, - }, - { - id: "message-from-user", - threadId: parentThreadId, - role: "user", - text: "Thanks", - createdBy: "user", - creationSource: "web", + ...idleThreadProjection(shell), + messages, + visibleTurnItems: messages.map((message, position) => ({ + position, + visibility: "local", + sourceThreadId: parentThreadId, + sourceItemId: `item-${message.id}`, + item: { + id: `item-${message.id}`, + type: "user_message", + messageId: message.id, + runId: activeRunId, + status: "completed", + title: null, + text: message.text, + updatedAt: now, }, - ], - contextTransfers: [], - subagents: [], - updatedAt: now, + })), } as unknown as OrchestrationV2ThreadProjection; - const layer = layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadShell: () => Effect.succeed(shell), + getThreadRecords: () => Effect.succeed(projection), + getProjectThreadRecords: () => Effect.succeed(projection), getTimelinePage: () => - Effect.succeed({ - items: visibleTurnItems, - totalItems: visibleTurnItems.length, - hasMore: false, - } as never), - getThreadRecords: (threadId) => - threadId === parentThreadId - ? Effect.succeed(projection) - : Effect.die(`unexpected thread ${threadId}`), - getThreadShell: (threadId) => - Effect.succeed( - threadId === parentThreadId - ? (projection.thread as unknown as OrchestrationV2ThreadShell) - : null, - ), - getProjectThreadRecords: (input) => - input.threadId === parentThreadId - ? Effect.succeed(projection) - : Effect.die(`unexpected thread ${input.threadId}`), - } satisfies Partial), + Effect.succeed({ items: projection.visibleTurnItems, totalItems: 2, hasMore: false }), + }), ServerSettings.layerTest(), - Layer.mock(ProviderRegistry.ProviderRegistry)({ - getProviders: Effect.succeed([]), - } satisfies Partial), + Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ProjectService.ProjectService)({}), Layer.mock(SecretRequests.SecretRequests)({}), - Layer.mock(ScheduledTaskService.ScheduledTaskService)({ - list: () => Effect.succeed({ tasks: [] }), - } satisfies Partial), - Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ - list: () => Effect.succeed([]), - } satisfies Partial), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({}), NodeCrypto.layer, ), ), @@ -788,11 +737,10 @@ it("readThread names the thread that sent a message with t3_thread_send", async await Effect.gen(function* () { const service = yield* OrchestratorMcpService.OrchestratorMcpService; const result = yield* service.readThread(makeScope(), { threadId: parentThreadId }); - const [fromChild, fromUser] = result.items; - expect(fromChild?.messageId).toBe("message-from-child"); - expect(fromChild?.senderThreadId).toBe(childThreadId); - expect(fromUser?.messageId).toBe("message-from-user"); - // Messages without a sending thread omit the field rather than returning null. - expect(fromUser).not.toHaveProperty("senderThreadId"); + expect(result.items).toMatchObject([ + { messageId: "message-from-child", senderThreadId: childThreadId }, + { messageId: "message-from-user" }, + ]); + expect(result.items[1]).not.toHaveProperty("senderThreadId"); }).pipe(Effect.provide(layer), Effect.runPromise); }); diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index d0027b79f2b6..bf41ea299d95 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -421,7 +421,6 @@ export const OrchestratorMcpThreadTimelineItem = Schema.Struct({ messageId: Schema.NullOr(MessageId), createdBy: Schema.NullOr(OrchestrationV2Actor), creationSource: Schema.NullOr(OrchestrationV2CreationSource), - /** The agent thread that sent this message to another thread; omitted for other messages. */ senderThreadId: Schema.optional(ThreadId), type: Schema.String, status: OrchestrationV2TurnItemStatus,