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
12 changes: 11 additions & 1 deletion apps/server/src/mcp/BridgeStatusService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import * as Bridge from "./BridgeStatusService.ts";
import * as Invocation from "./McpInvocationContext.ts";
import * as Events from "./McpEvents.ts";
import * as Metrics from "./McpBridgeMetrics.ts";
import * as AutoReview from "../orchestration-v2/ApprovalAutoReviewStatus.ts";

it.effect(
"reports fresh peer probe results and preserves bridge status when links are unavailable",
Expand All @@ -23,6 +24,7 @@ it.effect(
issuedAt: 0,
};
const dependencies = Layer.mergeAll(
AutoReview.layer,
Metrics.layer,
Layer.mock(Events.McpEvents)({
bridgeStatus: () =>
Expand Down Expand Up @@ -67,6 +69,7 @@ it.effect(
yield* Effect.gen(function* () {
const service = yield* Bridge.BridgeStatusService;
const status = yield* service.status();
expect(status.autoReview).toEqual({ status: "healthy", failures: 0, lastError: null });
expect(status.linkedEnvironments).toMatchObject({
checkedAt: "1970-01-01T00:00:00.000Z",
links: [
Expand All @@ -76,12 +79,19 @@ it.effect(
],
});
yield* Ref.set(unavailable, true);
const autoReview = yield* AutoReview.ApprovalAutoReviewStatus;
yield* autoReview.failed("Subscription overflow; resubscribing");
expect(yield* service.status()).toMatchObject({
linkedEnvironments: null,
daysUntilSessionExpires: 1,
autoReview: {
status: "degraded",
failures: 1,
lastError: "Subscription overflow; resubscribing",
},
});
}).pipe(
Effect.provide(Bridge.layer.pipe(Layer.provide(dependencies))),
Effect.provide(Bridge.layer.pipe(Layer.provideMerge(dependencies))),
Effect.provideService(Invocation.McpInvocationContext, scope),
);
}),
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/mcp/BridgeStatusService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,9 @@ import * as PeerForwarding from "../peer/PeerForwarding.ts";
import * as Invocation from "./McpInvocationContext.ts";
import * as McpEvents from "./McpEvents.ts";
import * as Metrics from "./McpBridgeMetrics.ts";
import * as AutoReview from "../orchestration-v2/ApprovalAutoReviewStatus.ts";
export const BridgeStatusResult = Schema.Struct({
autoReview: AutoReview.Snapshot,
serviceStartedAt: Schema.String,
tunnel: Schema.NullOr(EnvironmentConnectReachability),
daysUntilSessionExpires: Schema.Number,
Expand Down Expand Up @@ -64,6 +66,7 @@ export const layer = Layer.effect(
const events = yield* McpEvents.McpEvents;
const metrics = yield* Metrics.McpBridgeMetrics;
const forwarding = yield* PeerForwarding.PeerForwarding;
const autoReview = yield* AutoReview.ApprovalAutoReviewStatus;
return BridgeStatusService.of({
status: Effect.fn("BridgeStatusService.status")(function* () {
const scope = yield* Invocation.McpInvocationContext;
Expand Down Expand Up @@ -95,6 +98,7 @@ export const layer = Layer.effect(
);
return {
...status,
autoReview: yield* autoReview.read,
linkedEnvironments,
toolCalls: yield* metrics.read(scope.client!.sessionId),
};
Expand Down
20 changes: 19 additions & 1 deletion apps/server/src/mcp/McpApprovalAnswerContext.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,22 @@
import type { EnvironmentId } from "@t3tools/contracts";
import {
type EnvironmentId,
ModelSelection,
OrchestrationV2McpApprovalAnswer,
RuntimeRequestId,
ThreadId,
} from "@t3tools/contracts";
import * as Context from "effect/Context";
import * as Schema from "effect/Schema";

export const ApprovalReviewEscalation = Schema.Struct({
threadId: ThreadId,
requestId: RuntimeRequestId,
decision: Schema.Literal("escalate"),
requestKind: Schema.optional(Schema.String),
prompt: Schema.optional(Schema.String),
answeredAt: OrchestrationV2McpApprovalAnswer.fields.answeredAt,
reviewer: Schema.Struct({ modelSelection: ModelSelection, reason: Schema.String }),
});

export class McpApprovalAnswerContext extends Context.Service<
McpApprovalAnswerContext,
Expand All @@ -8,5 +25,6 @@ export class McpApprovalAnswerContext extends Context.Service<
readonly clientLabel: string;
readonly originEnvironmentId?: EnvironmentId;
readonly note?: string;
readonly reviewer?: { readonly modelSelection: ModelSelection; readonly reason: string };
}
>()("t3/mcp/McpApprovalAnswerContext") {}
3 changes: 2 additions & 1 deletion apps/server/src/mcp/McpHttpServer.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import * as JarvisPlaybook from "./JarvisPlaybook.ts";
import * as BridgeStatusService from "./BridgeStatusService.ts";
import * as ApprovalAutoReviewStatus from "../orchestration-v2/ApprovalAutoReviewStatus.ts";
import { BridgeToolkit } from "./toolkits/bridge/tools.ts";
import * as BridgeHandlers from "./toolkits/bridge/handlers.ts";
import * as McpRequestLog from "../persistence/McpRequestLog.ts";
Expand Down Expand Up @@ -1485,7 +1486,7 @@ export const layerTools = Layer.mergeAll(
content: JarvisPlaybook.read(),
}).pipe(Layer.provide(provideRegistrationServices<FileSystem.FileSystem | Path.Path>().layer)),
toolkitRegistration(BridgeToolkit, BridgeHandlers.layer).pipe(
Layer.provide(BridgeStatusService.layer),
Layer.provide(BridgeStatusService.layer.pipe(Layer.provide(ApprovalAutoReviewStatus.layer))),
),
layerLedgerToolkit,
layerPreviewToolkit,
Expand Down
8 changes: 6 additions & 2 deletions apps/server/src/mcp/PendingApprovalMcpService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ import * as Schema from "effect/Schema";
import * as SqlClient from "effect/sql/SqlClient";

import * as Invocation from "./McpInvocationContext.ts";
import { McpApprovalAnswerContext } from "./McpApprovalAnswerContext.ts";
import { ApprovalReviewEscalation, McpApprovalAnswerContext } from "./McpApprovalAnswerContext.ts";
import {
assertFullAccess,
dispatchFailure,
Expand All @@ -34,7 +34,11 @@ const ApprovalRefusal = Schema.Struct({
attemptedDecision: OrchestrationV2McpApprovalAnswer.fields.decision,
reason: Schema.String,
});
const ApprovalRecord = Schema.Union([OrchestrationV2McpApprovalAnswer, ApprovalRefusal]);
const ApprovalRecord = Schema.Union([
OrchestrationV2McpApprovalAnswer,
ApprovalRefusal,
ApprovalReviewEscalation,
]);
export const ApprovalLogResult = Schema.Struct({
records: Schema.Array(ApprovalRecord),
nextCursor: Schema.NullOr(Schema.Number),
Expand Down
67 changes: 64 additions & 3 deletions apps/server/src/mcp/toolkits/core.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ import * as McpInvocationContext from "../McpInvocationContext.ts";
import * as McpToolAccessTestkit from "../McpToolAccess.testkit.ts";
import * as AnalyticsService from "../../telemetry/AnalyticsService.ts";
import { dispatchFailure } from "../threadAccess.ts";
import { McpApprovalAnswerContext } from "../McpApprovalAnswerContext.ts";
import { ApprovalReviewEscalation, McpApprovalAnswerContext } from "../McpApprovalAnswerContext.ts";
import * as EnvironmentHandlers from "./environment/handlers.ts";
import * as AttachmentHandlers from "./attachment/handlers.ts";
import { ThreadToolkit } from "./thread/tools.ts";
Expand Down Expand Up @@ -103,12 +103,14 @@ const decodeMcpAttachmentInput = Schema.decodeUnknownEffect(McpAttachmentInput);
const encodeApprovalAnswer = Schema.encodeSync(
Schema.fromJsonString(OrchestrationV2McpApprovalAnswer),
);
const encodeEscalation = Schema.encodeEffect(Schema.fromJsonString(ApprovalReviewEscalation));
const decodeThreadReadResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadReadResult);
const decodeThreadSearchResult = Schema.decodeUnknownEffect(
ThreadToolkit.tools.t3_thread_search.successSchema,
);

const layerThreadToolkit = McpHttpServer.layerThreadToolkit.pipe(
Layer.provide(ServerSettings.layerTest()),
Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})),
Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
Layer.provide(Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({})),
Expand Down Expand Up @@ -420,10 +422,24 @@ it.effect("only outside full-access sign-ins answer approvals, with attribution"
answeredAt: "2026-11-12T00:00:00.000Z",
note: "Chris said yes.",
};
const reviewer = {
modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "reviewer" },
reason: "Requested work",
};
yield* sql`INSERT INTO mcp_approval_answers (thread_id, request_id, answer_json)
VALUES (${shell.id}, ${requestId}, ${encodeApprovalAnswer(record)})`;
VALUES (${shell.id}, ${requestId}, ${encodeApprovalAnswer({ ...record, reviewer })})`;
const escalation = {
threadId: shell.id,
requestId,
answeredAt: record.answeredAt,
decision: "escalate" as const,
reviewer: { ...reviewer, reason: "Possible spending" },
};
const escalationJson = yield* encodeEscalation(escalation);
yield* sql`INSERT INTO mcp_approval_answers (thread_id, request_id, outcome, answer_json)
VALUES (${shell.id}, ${requestId}, 'refused', ${escalationJson})`;
expect((yield* log(outside("full-access"))).structuredContent).toEqual({
records: [...refused.records, record],
records: [...refused.records, { ...record, reviewer }, escalation],
nextCursor: null,
});
}).pipe(
Expand Down Expand Up @@ -884,6 +900,7 @@ it.effect("requires a projectId to file a scratch thread and returns the moved s
}).pipe(
Effect.provide(
McpHttpServer.layerThreadToolkit.pipe(
Layer.provide(ServerSettings.layerTest()),
Layer.provide(Sqlite.layerMemory),
Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})),
Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
Expand Down Expand Up @@ -1164,6 +1181,7 @@ it.effect("search names the calling thread, so an agent can tell its own match a
Layer.provide(Layer.mock(PeerLinkRequests.PeerLinkRequests)({})),
Layer.provide(Layer.mock(PeerLinks.PeerLinks)({})),
Layer.provide(Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({})),
Layer.provide(ServerSettings.layerTest()),
Layer.provide(Sqlite.layerMemory),
Layer.provide(NodeServices.layer),
),
Expand All @@ -1190,6 +1208,48 @@ it.effect("a read-only client reads threads and is refused every write before it
});
expect(configuration.isError).toBe(false);
expect(configuration.structuredContent).toMatchObject({ runtimeMode: "auto" });
expect(configuration.structuredContent).toMatchObject({
autoReview: { override: "inherit", effective: null },
});
for (const access of ["read-only", "approval-required", "auto"] as const) {
const denied = yield* server
.callTool({
name: "t3_thread_configure",
arguments: {
threadId: "other-project-thread",
autoReview: { instanceId: "codex", model: "reviewer" },
},
})
.pipe(
Effect.provideService(McpInvocationContext.McpInvocationContext, clientScope(access)),
Effect.provideService(McpSchema.McpServerClient, client),
);
expect(declaredFailure(denied)).toMatchObject({ code: "capability_denied" });
}
for (const override of [{ instanceId: "codex", model: "reviewer" }, null, "inherit"] as const) {
const changed = yield* server
.callTool({
name: "t3_thread_configure",
arguments: {
threadId: "other-project-thread",
autoReview: override,
},
})
.pipe(
Effect.provideService(
McpInvocationContext.McpInvocationContext,
clientScope("full-access"),
),
Effect.provideService(McpSchema.McpServerClient, client),
);
expect(changed.isError).toBe(false);
expect(changed.structuredContent).toMatchObject({
autoReview: {
override,
effective: override === "inherit" ? null : override,
},
});
}

const pinned = yield* call("t3_thread_organize", {
action: "pin",
Expand Down Expand Up @@ -1412,6 +1472,7 @@ it.effect.each([
}).pipe(
Effect.provide(
McpHttpServer.layerThreadToolkit.pipe(
Layer.provide(ServerSettings.layerTest()),
Layer.provide(Sqlite.layerMemory),
Layer.provideMerge(McpHttpServer.layerInProcessMcpServer),
Layer.provide(NodeCrypto.layer),
Expand Down
27 changes: 26 additions & 1 deletion apps/server/src/mcp/toolkits/thread/handlers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,10 @@ import { modelSelectionCommandType } from "@t3tools/shared/model";
import * as McpInvocationContext from "../../McpInvocationContext.ts";
import * as McpToolAccess from "../../McpToolAccess.ts";
import * as PendingApproval from "../../PendingApprovalMcpService.ts";
import * as AutoReviewConfig from "../../../orchestration-v2/ApprovalAutoReviewConfig.ts";
import {
dispatchFailure,
assertFullAccess,
newCommandId,
readCaller,
readThread,
Expand Down Expand Up @@ -224,7 +226,10 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, {
),
),
t3_thread_configure: McpToolAccess.either(
(input) => input.modelSelection === undefined && input.runtimeMode === undefined,
(input) =>
input.modelSelection === undefined &&
input.runtimeMode === undefined &&
input.autoReview === undefined,
McpToolAccess.reads((input) =>
Effect.gen(function* () {
const {
Expand All @@ -235,6 +240,7 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, {
modelSelection: thread.modelSelection,
runtimeMode: thread.runtimeMode,
interactionMode: thread.interactionMode,
autoReview: yield* AutoReviewConfig.read(thread.id).pipe(Effect.mapError(unavailable)),
};
}),
),
Expand All @@ -244,6 +250,25 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, {
threads,
projection: { thread },
} = yield* readThread(input.threadId);
if (input.autoReview !== undefined) {
yield* assertFullAccess(yield* readCaller(), "configure approval auto-review");
if (input.modelSelection !== undefined || input.runtimeMode !== undefined) {
return yield* new OrchestratorMcpFailure({
code: "invalid_request",
message: "Set autoReview separately from model and runtimeMode.",
});
}
const autoReview = yield* AutoReviewConfig.write(thread.id, input.autoReview).pipe(
Effect.mapError(unavailable),
);
return {
threadId: thread.id,
modelSelection: thread.modelSelection,
runtimeMode: thread.runtimeMode,
interactionMode: thread.interactionMode,
autoReview,
};
}
if (input.runtimeMode !== undefined) {
if (input.modelSelection !== undefined) {
return yield* new OrchestratorMcpFailure({
Expand Down
15 changes: 14 additions & 1 deletion apps/server/src/mcp/toolkits/thread/tools.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ import {
} from "@t3tools/contracts";
import * as SecureRandom from "effect/crypto/SecureRandom";
import * as Schema from "effect/Schema";
import * as SqlClient from "effect/sql/SqlClient";
import * as ServerSettings from "../../../serverSettings.ts";
import { Tool, Toolkit } from "effect/ai";

import * as ScratchThreadPromotion from "../../../project/ScratchThreadPromotion.ts";
Expand All @@ -32,6 +34,7 @@ import * as ScheduledTaskService from "../../../scheduledTasks/ScheduledTaskServ
import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts";
import * as McpInvocationContext from "../../McpInvocationContext.ts";
import * as PendingApproval from "../../PendingApprovalMcpService.ts";
import { AutoReviewOverride } from "../../../orchestration-v2/ApprovalAutoReviewConfig.ts";

const ThreadOrganizeTool = Tool.make("t3_thread_organize", {
description:
Expand Down Expand Up @@ -202,20 +205,30 @@ const ApprovalLogTool = Tool.make("t3_approval_log", {
.annotate(Tool.Destructive, false);

const threadConfiguration = Schema.Struct({
autoReview: Schema.Struct({
override: AutoReviewOverride,
effective: Schema.NullOr(ModelSelection),
}),
threadId: ThreadId,
modelSelection: ModelSelection,
runtimeMode: RuntimeMode,
interactionMode: ProviderInteractionMode,
});
const ThreadConfigureTool = Tool.make("t3_thread_configure", {
...commandTool,
dependencies: [
...commandTool.dependencies,
SqlClient.SqlClient,
ServerSettings.ServerSettingsService,
],
description:
"Read a thread's provider, model, and modes, or set its provider, model, and options with `modelSelection`. Omit `threadId` for this thread. Set `runtimeMode` alone to lower permissions on an idle thread; raises and active runs are refused. Values from `orchestrator_capabilities`.",
"Read or set a thread's model and modes. Set provider, model and options with modelSelection. Omit threadId for this thread. Set runtimeMode alone to lower permissions on an idle thread; raises and active runs are refused. autoReview sets a reviewer model, null disables it, or 'inherit' uses the server default. Values from orchestrator_capabilities.",
parameters: Schema.Struct({
environmentId: OrchestratorMcpEnvironmentTarget,
threadId: Schema.optional(ThreadId),
modelSelection: Schema.optional(ModelSelection),
runtimeMode: Schema.optional(RuntimeMode),
autoReview: Schema.optional(AutoReviewOverride),
}),
success: Schema.Union([threadConfiguration, OrchestrationV2DispatchCommandResult]),
}).annotate(Tool.Destructive, true);
Expand Down
Loading
Loading