diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index b290ad124c0f2..7f8ac313e6a7c 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -1,6 +1,8 @@ import { EnvironmentId, + MessageId, NodeId, + OrchestratorMcpThreadTimelineItem, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadShell, ProjectId, @@ -9,10 +11,12 @@ import { RunId, ScheduledTaskId, ThreadId, + TurnItemId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { expect, it } from "vite-plus/test"; @@ -45,6 +49,7 @@ const cancelledRunId = RunId.make("run-mcp-cancelled"); const childRunId = RunId.make("run-mcp-child"); const taskId = NodeId.make("node-mcp-task-1"); const now = DateTime.makeUnsafe("2026-08-04T12:00:00.000Z"); +const decodeTimelineItem = Schema.decodeUnknownEffect(OrchestratorMcpThreadTimelineItem); const codexDriver = ProviderDriverKind.make("codex"); const customCodexInstanceId = ProviderInstanceId.make("codex-custom-workspace"); @@ -119,6 +124,13 @@ function makeRun(input: { it("readThread preserves caller access errors and prefers activity-run status over a newer cancelled queued run", async () => { let callerUnavailable = false; + const messageId = MessageId.make("message:old-run"); + let scopedRunId: RunId | null = activeRunId; + let scopedMatches = 1; + const runs = [ + makeRun({ id: activeRunId, ordinal: 1, status: "running" }), + makeRun({ id: cancelledRunId, ordinal: 2, status: "cancelled" }), + ]; const projection = { thread: baseThread({ threadId: parentThreadId, @@ -126,10 +138,7 @@ it("readThread preserves caller access errors and prefers activity-run status ov instanceId: parentInstanceId, model: "gpt-5.4", }), - runs: [ - makeRun({ id: activeRunId, ordinal: 1, status: "running" }), - makeRun({ id: cancelledRunId, ordinal: 2, status: "cancelled" }), - ], + runs, visibleTurnItems: [], runtimeRequests: [], messages: [], @@ -142,7 +151,22 @@ it("readThread preserves caller access errors and prefers activity-run status ov Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ - getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }), + getTimelinePage: (_threadId, options) => + Effect.succeed({ + items: [], + totalItems: 300, + hasMore: false, + ...(options.forMessageId === undefined + ? {} + : { + messageScope: { + messageId: options.forMessageId, + inputMatches: scopedMatches, + runId: scopedRunId, + itemCount: 0, + }, + }), + }), getThreadRecords: (threadId) => callerUnavailable ? Effect.fail( @@ -191,10 +215,65 @@ it("readThread preserves caller access errors and prefers activity-run status ov expect(result.thread.status).toBe("running"); expect(result.thread.latestRunId).toBe(cancelledRunId); expect(result.thread.activeRunId).toBe(activeRunId); + expect(result).not.toHaveProperty("messageScope"); + runs.push( + ...Array.from({ length: 60 }, (_, index) => + makeRun({ id: RunId.make(`run:newer:${index}`), ordinal: index + 3, status: "completed" }), + ), + ); + const scoped = yield* service.readThread(makeScope(), { + threadId: parentThreadId, + view: "activity", + forMessageId: messageId, + runLimit: 1, + }); + expect(scoped.recentRuns.map((run) => run.runId)).toEqual([activeRunId]); + expect(scoped.messageScope).toEqual({ + messageId, + inputMatches: 1, + runId: activeRunId, + itemCount: 0, + }); + for (const [matches, runId] of [ + [0, null], + [2, null], + [1, null], + [1, RunId.make("run:missing")], + ] as const) { + scopedMatches = matches; + scopedRunId = runId; + const unresolved = yield* service.readThread(makeScope(), { + threadId: parentThreadId, + view: "activity", + forMessageId: messageId, + }); + expect(unresolved.recentRuns).toEqual([]); + } + for (const incompatible of [ + {}, + { view: "messages" as const }, + { view: "activity" as const, itemId: TurnItemId.make("item:conflict") }, + ]) { + const invalid = yield* service + .readThread(makeScope(), { + threadId: parentThreadId, + forMessageId: messageId, + ...incompatible, + }) + .pipe(Effect.flip); + expect(invalid.code).toBe("invalid_argument"); + } callerUnavailable = true; const denied = yield* service - .readThread({ ...makeScope(), capabilities: new Set() }, { threadId: parentThreadId }) + .readThread( + { ...makeScope(), capabilities: new Set() }, + { + threadId: parentThreadId, + forMessageId: messageId, + view: "activity", + }, + ) .pipe(Effect.flip); expect(denied.code).toBe("capability_denied"); const missingCaller = yield* service @@ -206,6 +285,7 @@ it("readThread preserves caller access errors and prefers activity-run status ov }); it("taskStatus returns task.providerInstanceId rather than the driver kind", async () => { + let evidenceLookup = false; const parentProjection = { thread: baseThread({ threadId: parentThreadId, @@ -260,12 +340,15 @@ it("taskStatus returns task.providerInstanceId rather than the driver kind", asy createdBy: "agent", }, runs: [ - makeRun({ - id: childRunId, - ordinal: 1, - status: "running", - instanceId: customCodexInstanceId, - }), + { + ...makeRun({ + id: childRunId, + ordinal: 1, + status: "running", + instanceId: customCodexInstanceId, + }), + delegatedTaskId: taskId, + }, ], visibleTurnItems: [], runtimeRequests: [], @@ -287,8 +370,50 @@ it("taskStatus returns task.providerInstanceId rather than the driver kind", asy Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ - getTimelinePage: () => Effect.succeed({ items: [], totalItems: 0, hasMore: false }), - getThreadRecords: (threadId) => { + getTimelinePage: (_threadId, options) => + Effect.succeed({ + items: [ + { + position: 280, + sourceThreadId: childThreadId, + sourceItemId: TurnItemId.make("item:evidence-activity"), + visibility: "local" as const, + item: { + id: TurnItemId.make("item:evidence-activity"), + threadId: childThreadId, + runId: childRunId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed" as const, + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "command_execution" as const, + input: "inspect fixture", + }, + }, + ], + totalItems: 281, + hasMore: false, + messageScope: { + messageId: options.forMessageId!, + inputMatches: 1, + runId: childRunId, + itemCount: 2, + }, + }), + getThreadShell: () => + Effect.succeed(childProjection.thread as unknown as OrchestrationV2ThreadShell), + getProjectThreadRecords: () => Effect.succeed(childProjection), + getThreadRecords: (threadId, fields) => { + if (evidenceLookup && fields?.some((field) => field === "subagents")) { + return Effect.die("Evidence reads must not enter task-result delivery"); + } if (threadId === parentThreadId) return Effect.succeed(parentProjection); if (threadId === childThreadId) return Effect.succeed(childProjection); return Effect.die(`unexpected thread ${threadId}`); @@ -320,6 +445,15 @@ it("taskStatus returns task.providerInstanceId rather than the driver kind", asy expect(result.status).toBe("running"); expect(result.taskId).toBe(taskId); expect(result.childThreadId).toBe(childThreadId); + evidenceLookup = true; + const evidence = yield* service.readThread(makeScope(), { + threadId: childThreadId, + view: "activity", + forMessageId: MessageId.make("message:evidence"), + afterPosition: 279, + }); + expect(evidence.items[0]?.text).toBe("$ inspect fixture"); + expect(evidence.recentRuns.map((run) => run.runId)).toEqual([childRunId]); }).pipe(Effect.provide(layer), Effect.runPromise); }); @@ -359,24 +493,26 @@ it("readThread and sendToThread reach threads in other projects", async () => { projectId: foreignProjectId, }, runs: [], - visibleTurnItems: [ - { - position: 0, + visibleTurnItems: (["final_answer", "commentary", undefined, "final_answer"] as const).map( + (phase, position) => ({ + position, visibility: "inherited", sourceThreadId: threadId, - sourceItemId: "item-1", + sourceItemId: `item-${position}`, item: { - id: "item-1", - type: "assistant_message", - messageId: "assistant-1", + id: `item-${position}`, + type: position === 3 ? "user_message" : "assistant_message", + messageId: `message-${position}`, + runId: activeRunId, + ...(phase === undefined ? {} : { phase }), status: "completed", title: null, text: "Foreign thread said hello", createdAt: now, updatedAt: now, }, - }, - ], + }), + ), runtimeRequests: [], messages: [], contextTransfers: [], @@ -401,12 +537,28 @@ it("readThread and sendToThread reach threads in other projects", async () => { ? parentShell : null, ), - getTimelinePage: (threadId) => - Effect.succeed({ - items: foreignProjection(threadId).visibleTurnItems, - totalItems: 1, - hasMore: false, - }), + getTimelinePage: (threadId, options) => { + const rows = foreignProjection(threadId).visibleTurnItems.filter((row) => + options.itemId === undefined + ? row.position > (options.afterPosition ?? -1) + : row.sourceItemId === options.itemId, + ); + return Effect.succeed({ + items: rows.slice(0, options.limit), + totalItems: 4, + hasMore: rows.length > options.limit, + ...(options.forMessageId === undefined + ? {} + : { + messageScope: { + messageId: options.forMessageId, + inputMatches: 1, + runId: activeRunId, + itemCount: 4, + }, + }), + }); + }, getProjectThreadRecords: (input) => input.threadId === foreignThreadId && input.projectId === foreignProjectId ? Effect.succeed(foreignProjection(input.threadId)) @@ -455,7 +607,55 @@ it("readThread and sendToThread reach threads in other projects", async () => { const foreign = yield* service.readThread(makeScope(), { threadId: foreignThreadId }); expect(foreign.thread.threadId).toBe(foreignThreadId); expect(foreign.thread.projectId).toBe(foreignProjectId); - expect(foreign.items.map((item) => item.text)).toEqual(["Foreign thread said hello"]); + expect(foreign.items.map((item) => item.text)).toEqual( + Array.from({ length: 4 }, () => "Foreign thread said hello"), + ); + for (const selection of [ + { view: "messages" }, + { view: "activity" }, + { view: "activity", forMessageId: MessageId.make("message-0") }, + ] as const) { + for (let position = 0; position < 4; position += 1) { + const page = yield* service.readThread(makeScope(), { + threadId: foreignThreadId, + ...selection, + ...(position === 0 ? {} : { afterPosition: position - 1 }), + limit: 1, + maxCharsPerItem: 7, + }); + expect(page.items).toHaveLength(1); + const item = yield* decodeTimelineItem(page.items[0]); + expect(item).toMatchObject({ + itemId: `item-${position}`, + messageId: `message-${position}`, + sourceThreadId: foreignThreadId, + runId: activeRunId, + text: "Foreign", + textTruncated: true, + nextTextOffset: 7, + }); + if (position < 2) expect(item.phase).toBe(position === 0 ? "final_answer" : "commentary"); + else expect(item).not.toHaveProperty("phase"); + if (position === 0) { + const continuation = yield* service.readThread(makeScope(), { + threadId: foreignThreadId, + view: selection.view, + itemId: item.itemId, + textOffset: 7, + maxCharsPerItem: 50, + }); + expect(continuation.items[0]).toMatchObject({ + itemId: item.itemId, + phase: "final_answer", + text: " thread said hello", + textTruncated: false, + nextTextOffset: null, + }); + } + expect(page.nextPosition).toBe(position); + expect(page.hasMore).toBe(position < 3); + } + } const sent = yield* service.sendToThread(makeScope(), { threadId: foreignThreadId, diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index b124a03ea2c40..220cf037d0046 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -977,6 +977,9 @@ function timelineItem(input: { creationSource: message?.creationSource ?? null, ...(message?.senderThreadId === undefined ? {} : { senderThreadId: message.senderThreadId }), type: input.row.item.type, + ...(input.row.item.type === "assistant_message" && input.row.item.phase !== undefined + ? { phase: input.row.item.phase } + : {}), status: input.row.item.status, title: input.row.item.title, text: text === null ? null : text.slice(offset, end), @@ -2944,6 +2947,15 @@ const make = Effect.gen(function* () { }), readThread: Effect.fn("OrchestratorMcpService.readThread")(function* (scope, input) { const { parent, target, shell } = yield* loadReadableThread(scope, input.threadId); + if ( + input.forMessageId !== undefined && + (input.itemId !== undefined || input.view !== "activity") + ) { + return yield* failure( + "invalid_argument", + "forMessageId requires activity view and cannot be combined with itemId.", + ); + } const view = input.view ?? "messages"; const afterPosition = input.afterPosition ?? -1; const limit = input.limit ?? DEFAULT_THREAD_READ_LIMIT; @@ -2954,6 +2966,7 @@ const make = Effect.gen(function* () { limit, view, ...(input.itemId === undefined ? {} : { itemId: input.itemId }), + ...(input.forMessageId === undefined ? {} : { forMessageId: input.forMessageId }), }) .pipe(Effect.mapError(threadManagementFailure)); const page = timeline.items; @@ -2978,7 +2991,7 @@ const make = Effect.gen(function* () { page.filter((row) => row.sourceThreadId === target.thread.id).map((row) => row.item.runId), ); const tasks = - parent === undefined + parent === undefined || input.forMessageId !== undefined ? [] : directAppOwnedChildTasks( { @@ -3036,7 +3049,13 @@ const make = Effect.gen(function* () { shell, yield* Clock.currentTimeMillis, ), - recentRuns: target.runs + recentRuns: (input.forMessageId === undefined + ? target.runs + : target.runs.filter( + (run) => + timeline.messageScope?.inputMatches === 1 && run.id === timeline.messageScope.runId, + ) + ) .toSorted((left, right) => right.ordinal - left.ordinal) .slice(0, input.runLimit ?? DEFAULT_THREAD_RUN_LIMIT) .map(threadRun), @@ -3050,6 +3069,7 @@ const make = Effect.gen(function* () { ), nextPosition: page.at(-1)?.position ?? null, hasMore: timeline.hasMore, + ...(timeline.messageScope === undefined ? {} : { messageScope: timeline.messageScope }), } satisfies OrchestratorMcpThreadReadResult; }), sendToThread: Effect.fn("OrchestratorMcpService.sendToThread")(function* (scope, input) { diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 5774f0e4c4f65..71b56e7316834 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -915,6 +915,39 @@ it.effect("discovers workflow stop targets and requires orchestration capability expect(read.isError).toBe(false); const result = yield* decodeThreadReadResult(read.structuredContent); expect(result.items[0]?.subagentId).toBe(subagentId); + const scopedRead = (arguments_: Record, capabilities = scope.capabilities) => + server + .callTool({ + name: "t3_thread_read", + arguments: { + threadId, + forMessageId: "message:missing", + ...arguments_, + }, + }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, { + ...scope, + capabilities, + }), + Effect.provideService(McpSchema.McpServerClient, client), + ); + const scoped = yield* scopedRead({ view: "activity" }); + expect(scoped.isError).toBe(false); + const scopedResult = yield* decodeThreadReadResult(scoped.structuredContent); + expect(scopedResult.items).toEqual([]); + expect(scopedResult.messageScope).toEqual({ + messageId: "message:missing", + inputMatches: 0, + runId: null, + itemCount: 0, + }); + for (const selectors of [{}, { view: "messages" }, { view: "activity", itemId }]) { + expect(declaredFailure(yield* scopedRead(selectors)).code).toBe("invalid_argument"); + } + expect(declaredFailure(yield* scopedRead({ view: "activity" }, new Set())).code).toBe( + "capability_denied", + ); const call = (invocation: McpInvocationContext.McpInvocationScope) => server .callTool({ @@ -941,7 +974,22 @@ it.effect("discovers workflow stop targets and requires orchestration capability getThreadShell: (id) => Effect.succeed(McpToolAccessTestkit.liveThreadShell(id)), getProjectThreadRecords: () => Effect.succeed(McpToolAccessTestkit.idleThreadProjection(shell)), - getTimelinePage: () => Effect.succeed({ items: [row], totalItems: 1, hasMore: false }), + getTimelinePage: (_id, options) => + Effect.succeed({ + items: options.forMessageId === undefined ? [row] : [], + totalItems: 1, + hasMore: false, + ...(options.forMessageId === undefined + ? {} + : { + messageScope: { + messageId: options.forMessageId, + inputMatches: 0, + runId: null, + itemCount: 0, + }, + }), + }), stopWorkflow: (input) => Effect.sync(() => { stopped.push(input); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 9bd3566e559a8..ffa5033fd061b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -750,6 +750,150 @@ const addOrphanedRecoveryCandidate = Effect.fn("addOrphanedRecoveryCandidate")(f return threadId; }); +it.effect("message scopes select the complete source-run interval before sparse pagination", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const threadId = yield* addOrphanedRecoveryCandidate("message-scope"); + const otherThreadId = yield* addOrphanedRecoveryCandidate("message-scope-other"); + const now = yield* DateTime.now; + const runId = RunId.make("run:message-scope:target"); + const messageId = MessageId.make("message:message-scope:input"); + const addItem = Effect.fnUntraced(function* ( + owner: ThreadId, + ordinal: number, + itemRunId: RunId | null, + inputMessageId: MessageId, + type: "user_message" | "assistant_message" | "command_execution" = "user_message", + ) { + yield* store.apply({ + id: EventId.make(`event:${owner}:${ordinal}`), + type: "turn-item.updated", + threadId: owner, + occurredAt: now, + payload: { + id: TurnItemId.make(`item:${owner}:${ordinal}`), + threadId: owner, + runId: itemRunId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + ...(type === "user_message" + ? { + type, + messageId: inputMessageId, + text: "input", + attachments: [], + createdBy: "user", + creationSource: "mcp", + inputIntent: "turn_start", + } + : type === "assistant_message" + ? { type, messageId: inputMessageId, text: "answer", streaming: false } + : { type, input: "inspect fixture" }), + }, + }); + }); + yield* addItem(threadId, 2, runId, messageId); + for (let ordinal = 3; ordinal < 264; ordinal += 1) { + yield* addItem( + threadId, + ordinal, + ordinal === 3 ? runId : RunId.make(`run:unrelated:${ordinal}`), + MessageId.make(`message:unrelated:${ordinal}`), + ordinal === 3 ? "command_execution" : "user_message", + ); + } + const answerId = MessageId.make("message:message-scope:answer"); + yield* addItem(threadId, 264, runId, answerId, "assistant_message"); + yield* addItem(otherThreadId, 2, runId, MessageId.make("message:other-source")); + const options = { view: "activity" as const, forMessageId: messageId, limit: 1 }; + const first = yield* store.getTimelinePage(threadId, options); + assert.deepEqual(first.messageScope, { messageId, inputMatches: 1, runId, itemCount: 3 }); + assert.deepEqual( + first.items.map((row) => row.position), + [1], + ); + assert.isTrue(first.hasMore); + assert.equal(first.totalItems, 264); + const second = yield* store.getTimelinePage(threadId, { ...options, afterPosition: 1 }); + assert.deepEqual( + second.items.map((row) => row.position), + [2], + ); + assert.equal(second.items[0]?.item.type, "command_execution"); + assert.deepEqual(second.messageScope, first.messageScope); + assert.isTrue(second.hasMore); + const third = yield* store.getTimelinePage(threadId, { ...options, afterPosition: 2 }); + assert.deepEqual( + third.items.map((row) => row.position), + [263], + ); + assert.deepEqual(third.messageScope, first.messageScope); + assert.isFalse(third.hasMore); + const exhausted = yield* store.getTimelinePage(threadId, { ...options, afterPosition: 263 }); + assert.deepEqual(exhausted.items, []); + assert.deepEqual(exhausted.messageScope, first.messageScope); + assert.isFalse(exhausted.hasMore); + for (const absentId of [MessageId.make("absent"), MessageId.make("message:other-source")]) { + const absent = yield* store.getTimelinePage(threadId, { ...options, forMessageId: absentId }); + assert.deepEqual(absent.items, []); + assert.deepEqual(absent.messageScope, { + messageId: absentId, + inputMatches: 0, + runId: null, + itemCount: 0, + }); + } + const wrongType = yield* store.getTimelinePage(threadId, { + ...options, + forMessageId: answerId, + limit: 100, + }); + assert.deepEqual(wrongType.messageScope, { + messageId: answerId, + inputMatches: 1, + runId, + itemCount: 3, + }); + assert.equal(wrongType.items[2]?.item.type, "assistant_message"); + const nullRunId = MessageId.make("message:null-run"); + yield* addItem(threadId, 265, null, nullRunId); + const nullRun = yield* store.getTimelinePage(threadId, { ...options, forMessageId: nullRunId }); + assert.deepEqual(nullRun.messageScope, { + messageId: nullRunId, + inputMatches: 1, + runId: null, + itemCount: 1, + }); + yield* addItem(threadId, 266, runId, messageId, "assistant_message"); + const duplicate = yield* store.getTimelinePage(threadId, { ...options, limit: 100 }); + assert.deepEqual(duplicate.messageScope, { + messageId, + inputMatches: 2, + runId: null, + itemCount: 2, + }); + assert.deepEqual( + duplicate.items.map((row) => row.position), + [1, 265], + ); + const duplicatePage = yield* store.getTimelinePage(threadId, { ...options, afterPosition: 1 }); + assert.deepEqual(duplicatePage.messageScope, duplicate.messageScope); + assert.deepEqual( + duplicatePage.items.map((row) => row.position), + [265], + ); + }).pipe(Effect.provide(layerTest)), +); + const restartCancelledWorkSurvivesStaleRunUpdate = Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const threadId = yield* addRolledBackRecoveryCandidate("restart-cancelled-work-stale-update"); @@ -6156,6 +6300,25 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }); assert.deepEqual(followingPage.items, targetAfterRollback.visibleTurnItems.slice(2)); assert.isFalse(followingPage.hasMore); + const inheritedMessageId = MessageId.make("message:projection-fork-source-rollback:user:2"); + const inheritedScope = yield* projectionStore.getTimelinePage(targetThreadId, { + view: "activity", + forMessageId: inheritedMessageId, + limit: 10, + }); + assert.deepEqual(inheritedScope.messageScope, { + messageId: inheritedMessageId, + inputMatches: 1, + runId: sourceRun2Id, + itemCount: 2, + }); + assert.deepEqual( + inheritedScope.items.map((row) => [row.position, row.sourceThreadId]), + [ + [2, sourceThreadId], + [3, sourceThreadId], + ], + ); assert.deepEqual( targetAfterRollback.visibleTurnItems.map((row) => [ diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 95c0f7a4bae53..066290c05a4b2 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -382,6 +382,7 @@ export type ProjectionRecords = Pick< export interface ProjectionTimelinePageOptions { readonly afterPosition?: number; readonly itemId?: TurnItemId; + readonly forMessageId?: MessageId; readonly view?: "messages" | "activity"; readonly limit: number; } @@ -389,6 +390,12 @@ export interface ProjectionTimelinePage { readonly items: ReadonlyArray; readonly totalItems: number; readonly hasMore: boolean; + readonly messageScope?: { + readonly messageId: MessageId; + readonly inputMatches: number; + readonly runId: RunId | null; + readonly itemCount: number; + }; } export interface ShellSnapshotOptions { @@ -5732,17 +5739,29 @@ export const layer: Layer.Layer = .withTransaction( Effect.gen(function* () { const index = yield* readTimelineIndex(threadId, new Set()); - const matching = index.visible - .map((row, position) => ({ ...row, position })) - .filter((row) => - options.itemId === undefined - ? row.position > (options.afterPosition ?? -1) && - (options.view === "activity" || - ["user_message", "assistant_message", "proposed_plan"].includes( - row.item.type, - )) - : row.sourceItemId === options.itemId, - ); + const positioned = index.visible.map((row, position) => ({ ...row, position })); + const inputs = + options.forMessageId === undefined + ? undefined + : positioned.filter((row) => row.item.messageId === options.forMessageId); + const unique = inputs?.length === 1 ? inputs[0] : undefined; + const scoped = + inputs === undefined + ? positioned + : unique === undefined + ? inputs + : positioned.filter( + (row) => + row.sourceThreadId === unique.sourceThreadId && + row.item.runId === unique.item.runId, + ); + const matching = scoped.filter((row) => + options.itemId === undefined + ? row.position > (options.afterPosition ?? -1) && + (options.view === "activity" || + ["user_message", "assistant_message", "proposed_plan"].includes(row.item.type)) + : row.sourceItemId === options.itemId, + ); const page = matching.slice(0, options.limit); const items = yield* Effect.forEach(page, (row) => Effect.gen(function* () { @@ -5780,6 +5799,16 @@ export const layer: Layer.Layer = items, totalItems: index.visible.length, hasMore: matching.length > page.length, + ...(options.forMessageId === undefined + ? {} + : { + messageScope: { + messageId: options.forMessageId, + inputMatches: inputs?.length ?? 0, + runId: unique?.item.runId ?? null, + itemCount: scoped.length, + }, + }), }; }), ) diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 438a3fe2def98..0d71c74abdc47 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -407,6 +407,10 @@ export const OrchestratorMcpThreadReadInput = Schema.Struct({ environmentId: OrchestratorMcpEnvironmentTarget, threadId: ThreadId, itemId: Schema.optional(TurnItemId), + forMessageId: Schema.optional(MessageId).annotate({ + description: + "Exact visible message ID. Requires activity view and no itemId; pages only its source/run activity without acknowledging task delivery.", + }), textOffset: Schema.optional(NonNegativeInt), view: Schema.optional(Schema.Literals(["messages", "activity"])), afterPosition: Schema.optional(NonNegativeInt), @@ -480,6 +484,7 @@ export const OrchestratorMcpThreadTimelineItem = Schema.Struct({ creationSource: Schema.NullOr(OrchestrationV2CreationSource), senderThreadId: Schema.optional(ThreadId), type: Schema.String, + phase: Schema.optional(Schema.Literals(["final_answer", "commentary"])), status: OrchestrationV2TurnItemStatus, title: Schema.NullOr(Schema.String), text: Schema.NullOr(Schema.String), @@ -495,6 +500,14 @@ export const OrchestratorMcpThreadReadResult = Schema.Struct({ items: Schema.Array(OrchestratorMcpThreadTimelineItem), nextPosition: Schema.NullOr(NonNegativeInt), hasMore: Schema.Boolean, + messageScope: Schema.optional( + Schema.Struct({ + messageId: MessageId, + inputMatches: NonNegativeInt, + runId: Schema.NullOr(RunId), + itemCount: NonNegativeInt, + }), + ), }); export type OrchestratorMcpThreadReadResult = typeof OrchestratorMcpThreadReadResult.Type; @@ -868,6 +881,7 @@ export class OrchestratorMcpFailure extends Schema.TaggedError