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
8 changes: 4 additions & 4 deletions apps/mobile/src/features/threads/McpAppWebView.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import {
mcpAppAllowAttribute,
mcpAppFileName,
mcpAppReferencesEqual,
mcpAppToolCall,
type McpAppReference,
} from "@t3tools/shared/mcpApp";
import * as Predicate from "effect/Predicate";
Expand Down Expand Up @@ -95,10 +96,9 @@ function storedToolCall(
item: { readonly type: string; readonly input?: unknown; readonly output?: unknown } | null,
) {
if (item?.type !== "dynamic_tool") return undefined;
const output = item.output as { readonly result?: unknown } | undefined;
return {
arguments: item.input,
result: output?.result as McpAppCallToolResult | undefined,
return mcpAppToolCall({ input: item.input, output: item.output }) as {
readonly arguments: unknown;
readonly result: McpAppCallToolResult | undefined;
};
}

Expand Down
82 changes: 81 additions & 1 deletion apps/server/src/mcpApps/McpAppRequests.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import {
type OrchestrationV2ServerCommand,
ProviderSessionId,
ProviderThreadId,
ThreadId,
Expand All @@ -7,6 +8,7 @@ import {
type OrchestrationV2ProviderThread,
type OrchestrationV2TurnItem,
} from "@t3tools/contracts";
import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
import { assert, describe, it } from "@effect/vitest";
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
Expand Down Expand Up @@ -68,10 +70,13 @@ function makeLayer(input: {
readonly item: OrchestrationV2TurnItem | null;
readonly mcpApps?: ProviderAdapter.ProviderAdapterV2McpApps;
readonly live?: boolean;
readonly recordFails?: boolean;
readonly dispatched?: Array<OrchestrationV2ServerCommand>;
}) {
return McpAppRequests.layer.pipe(
Layer.provide(
Layer.mergeAll(
NodeCrypto.layer,
Layer.mock(McpAppModelContext.McpAppModelContext)({
set: (entry) =>
Effect.sync(() => {
Expand All @@ -84,6 +89,13 @@ function makeLayer(input: {
getTurnItem: () => Effect.succeed(input.item),
}),
Layer.mock(ThreadManagementService.ThreadManagementService)({
dispatch: (command) =>
input.recordFails === true
? Effect.fail(new Orchestrator.OrchestratorProjectionError({ threadId }))
: Effect.sync(() => {
input.dispatched?.push(command);
return {} as never;
}),
getThreadRecords: (id: ThreadId) =>
Effect.succeed({
thread: {
Expand Down Expand Up @@ -139,11 +151,15 @@ describe("McpAppRequests", () => {
Effect.succeed([
{ name: "refresh", annotations: { readOnlyHint: true } },
{ name: "model_only", _meta: { ui: { visibility: ["model"] } } },
{ name: "get_weather" },
{ name: "set_alert" },
]),
callTool: (input) =>
Effect.sync(() => {
calls.push({ server: input.server, tool: input.tool });
return { content: [{ type: "text", text: "ok" }] };
return input.arguments.fail === true
? { content: [{ type: "text", text: "no" }], isError: true }
: { content: [{ type: "text", text: "ok" }] };
}),
readResource: () => Effect.succeed({ contents: [] }),
};
Expand All @@ -167,6 +183,70 @@ describe("McpAppRequests", () => {
}).pipe(Effect.provide(makeLayer({ item: appItem({ t3McpApp: app }), mcpApps }))),
);

it.effect("keeps the app's writes through its own tool on the tool call", () => {
const dispatched: Array<OrchestrationV2ServerCommand> = [];
return Effect.gen(function* () {
const requests = yield* McpAppRequests.McpAppRequests;
const call = (name: string, args: Record<string, unknown>) =>
requests.callTool({ threadId, itemId, name, arguments: args });
yield* call("get_weather", { city: "Paris", _ui_submitted: true });
// Other tools and failed writes leave the call as it was.
yield* call("refresh", {});
yield* call("set_alert", {});
yield* call("get_weather", { fail: true });
assert.deepEqual(
dispatched.map((command) =>
command.type === "mcp_app.call.record"
? {
threadId: command.threadId,
turnItemId: command.turnItemId,
arguments: command.arguments,
result: command.result,
}
: command.type,
),
[
{
threadId,
turnItemId: itemId,
arguments: { city: "Paris", _ui_submitted: true },
result: { content: [{ type: "text", text: "ok" }] },
},
],
);
}).pipe(Effect.provide(makeLayer({ item: appItem({ t3McpApp: app }), mcpApps, dispatched })));
});

it.effect("doesn't keep reads through the app's own tool", () => {
const dispatched: Array<OrchestrationV2ServerCommand> = [];
const readOnlyApp: ProviderAdapter.ProviderAdapterV2McpApps = {
...mcpApps,
listTools: () =>
Effect.succeed([{ name: "get_weather", annotations: { readOnlyHint: true } }]),
};
return Effect.gen(function* () {
const requests = yield* McpAppRequests.McpAppRequests;
yield* requests.callTool({ threadId, itemId, name: "get_weather", arguments: {} });
assert.deepEqual(dispatched, []);
}).pipe(
Effect.provide(
makeLayer({ item: appItem({ t3McpApp: app }), mcpApps: readOnlyApp, dispatched }),
),
);
});

it.effect("returns a write's result even when it can't be recorded", () =>
Effect.gen(function* () {
const requests = yield* McpAppRequests.McpAppRequests;
assert.deepEqual(
yield* requests.callTool({ threadId, itemId, name: "get_weather", arguments: {} }),
{ content: [{ type: "text", text: "ok" }] },
);
}).pipe(
Effect.provide(makeLayer({ item: appItem({ t3McpApp: app }), mcpApps, recordFails: true })),
),
);

it.effect("refuses tools hidden from apps or missing from the server", () =>
Effect.gen(function* () {
const requests = yield* McpAppRequests.McpAppRequests;
Expand Down
31 changes: 30 additions & 1 deletion apps/server/src/mcpApps/McpAppRequests.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import {
CommandId,
McpAppRequestError,
type McpAppCallToolInput,
type McpAppCallToolResult,
Expand All @@ -14,6 +15,7 @@ import {
import { mcpAppToolCallableByApp, type McpAppReference } from "@t3tools/shared/mcpApp";
import { mcpAppFromToolItem } from "@t3tools/shared/toolOutput";
import * as Context from "effect/Context";
import * as Crypto from "effect/Crypto";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
Expand Down Expand Up @@ -61,6 +63,7 @@ const make = Effect.gen(function* () {
const threadManagement = yield* ThreadManagementService.ThreadManagementService;
const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
const modelContext = yield* McpAppModelContext.McpAppModelContext;
const crypto = yield* Crypto.Crypto;

const fail = (threadId: ThreadId, reason: McpAppRequestError["reason"], cause?: unknown) =>
new McpAppRequestError({ threadId, reason, ...(cause === undefined ? {} : { cause }) });
Expand Down Expand Up @@ -135,14 +138,40 @@ const make = Effect.gen(function* () {
if (tool === undefined || !mcpAppToolCallableByApp(tool._meta)) {
return yield* fail(input.threadId, "tool-not-callable");
}
return yield* resolved.mcpApps
const result = yield* resolved.mcpApps
.callTool({
providerThread: resolved.providerThread,
server: resolved.app.server,
tool: input.name,
arguments: input.arguments,
})
.pipe(Effect.mapError((cause) => fail(input.threadId, "request-failed", cause)));
// A write through the app's own tool, such as a deferred form's submit, is
// kept so the app opens on its outcome. Reads are left out: an app can
// repeat those unprompted.
if (input.name === resolved.app.tool && !readOnlyHint(tool) && result.isError !== true) {
const uuid = yield* crypto.randomUUIDv4().pipe(Effect.orDie);
yield* threadManagement
.dispatch({
type: "mcp_app.call.record",
commandId: CommandId.make(`server:mcp-app-call:${input.itemId}:${uuid}`),
threadId: input.threadId,
turnItemId: input.itemId,
arguments: input.arguments,
result,
})
.pipe(
// The write already happened; failing the call would offer it again.
Effect.catch((error) =>
Effect.logWarning("Could not record an MCP App's tool call", {
threadId: input.threadId,
itemId: input.itemId,
error: error._tag,
}),
),
);
}
return result;
});

const readResource = Effect.fn("McpAppRequests.readResource")(function* (
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5142,6 +5142,8 @@ describe("CodexAdapterV2 post-settle continuation", () => {
content: [{ type: "text", text: "An interactive form is being shown to the user." }],
structuredContent: { status: "awaiting_user_submission" },
};
// Codex reports a tool's `isError` result, which a write that defers to its
// form returns, as a failed call.
const mcpAppDeferredTranscript = makeCodexReplayTranscript({
scenario: MCP_APP_SCENARIO,
entries: mcpAppTranscript.entries.map((entry) =>
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3743,6 +3743,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
// Only reads and storage are cancellable; remove partial writes. Once stored, attach the app or delete it if the item settled.
Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
// Codex reports an `isError` result as failed. Apps still render
// for it: a write tool that defers to its form returns one.
const reference =
item.result != null && item.error == null
? yield* restore(
Expand Down
41 changes: 38 additions & 3 deletions apps/server/src/orchestration-v2/EventSink.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import {
type OrchestrationV2Run,
OrchestrationV2DomainEvent,
type OrchestrationV2ProviderThread,
type OrchestrationV2TurnItem,
OrchestrationV2StoredEvent,
ProviderThreadId,
RunAttemptId,
Expand All @@ -17,15 +18,17 @@ import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Predicate from "effect/Predicate";
import * as PubSub from "effect/PubSub";
import * as Semaphore from "effect/Semaphore";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
import * as SqlClient from "effect/sql/SqlClient";

import { MCP_APP_CALL_OUTPUT_KEY, MCP_APP_OUTPUT_KEY } from "@t3tools/shared/mcpApp";
import { replayAndBufferProjectedLiveEvents } from "./LiveStreamBudget.ts";
import type { UnsequencedProjectEvent } from "../persistence/OrchestrationEventStore.ts";
import { projectDomainEventForWire } from "./WireProjection.ts";
import { projectDomainEventForWire, toolOutputFitsDetail } from "./WireProjection.ts";
import { makeRunOutcomeRecorder, type RunOutcome } from "./RunMetrics.ts";

import * as CommandReceiptStore from "./CommandReceiptStore.ts";
Expand Down Expand Up @@ -329,7 +332,35 @@ const layerBase: Layer.Layer<
});
});

const normalizeEvents = (events: ReadonlyArray<OrchestrationV2DomainEvent>) => {
// Merge against the latest item inside the write transaction; either writer may hold a stale copy.
const mergeMcpAppCall =
/* This helper runs for individual orchestration events; the enclosing operation owns the span. */
Effect.fnUntraced(function* (item: OrchestrationV2TurnItem, recorded: boolean) {
if (item.type !== "dynamic_tool") return item;
const output = Predicate.isObject(item.output) ? item.output : undefined;
if (output === undefined || !(MCP_APP_OUTPUT_KEY in output)) return item;
const stored = yield* projectionStore.getTurnItem({
threadId: item.threadId,
itemId: item.id,
});
if (stored?.type !== "dynamic_tool" || !Predicate.isObject(stored.output)) return item;
const call = (recorded ? output : stored.output)[MCP_APP_CALL_OUTPUT_KEY];
if (call === undefined) return item;
const base = recorded ? { ...stored, updatedAt: item.updatedAt } : item;
const merged = {
...(Predicate.isObject(base.output) ? base.output : {}),
[MCP_APP_CALL_OUTPUT_KEY]: call,
};
// Past the detail limit clients would read both results as truncated
// text, so the provider's output wins over the call.
if (!toolOutputFitsDetail(merged)) return recorded ? stored : item;
return { ...base, output: merged };
});

const normalizeEvents = (
events: ReadonlyArray<OrchestrationV2DomainEvent>,
recorded = false,
) => {
const runOrdinals = new Map(
events.flatMap((event) =>
event.type === "run.created" || event.type === "run.updated"
Expand All @@ -346,7 +377,10 @@ const layerBase: Layer.Layer<
event.payload,
event.payload.runId === null ? undefined : runOrdinals.get(event.payload.runId),
)
.pipe(Effect.map((payload) => ({ ...event, payload })))
.pipe(
Effect.flatMap((item) => mergeMcpAppCall(item, recorded)),
Effect.map((payload) => ({ ...event, payload })),
)
: Effect.succeed(event),
{ concurrency: 1 },
);
Expand Down Expand Up @@ -656,6 +690,7 @@ const layerBase: Layer.Layer<
runId === undefined
? input.events
: yield* updateDelegatedCompletionOnly(input.events, (id) => id === runId),
input.commandType === "mcp_app.call.record",
);
const storedEvents = yield* eventStore.append({
commandId: input.commandId,
Expand Down
Loading
Loading