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
80 changes: 71 additions & 9 deletions apps/server/src/peer/PeerMcpClient.test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { NodeHttpServer } from "@effect/platform-node";
import { expect, it } from "@effect/vitest";
import { OrchestratorMcpFailure } from "@t3tools/contracts";
import * as Context from "effect/Context";
import * as DateTime from "effect/DateTime";
import * as Deferred from "effect/Deferred";
Expand Down Expand Up @@ -30,8 +31,19 @@ const toolkit = Toolkit.make(
success: Result,
dependencies: [McpInvocationContext.McpInvocationContext],
}),
Tool.make("refuse", {
description: "Refuses the call.",
success: Result,
failure: OrchestratorMcpFailure,
failureMode: "return",
}),
);

const refusal = new OrchestratorMcpFailure({
code: "thread_not_found",
message: "Thread thread-gone was not found.",
});

it.effect.each([false, true])(
"reuses an initialized session before its first call finishes (cancelled: %s)",
(cancelFirst) =>
Expand All @@ -47,6 +59,7 @@ it.effect.each([false, true])(
);
const handlers = McpToolAccess.toLayer(toolkit, {
probe: McpToolAccess.reads(() => modes),
refuse: McpToolAccess.reads(() => Effect.fail(refusal)),
hold: McpToolAccess.reads(() =>
Deferred.succeed(entered, undefined).pipe(
Effect.andThen(Deferred.await(release)),
Expand Down Expand Up @@ -91,10 +104,45 @@ it.effect.each([false, true])(
).pipe(Effect.provide(NodeHttpServer.layerTest)),
);

it.effect("keeps a link healthy when the peer refuses a call", () =>
Effect.scoped(
Effect.gen(function* () {
const handlers = McpToolAccess.toLayer(toolkit, {
probe: McpToolAccess.reads(() => Effect.succeed({ modes: "" })),
hold: McpToolAccess.reads(() => Effect.succeed({ modes: "" })),
refuse: McpToolAccess.reads(() => Effect.fail(refusal)),
});
const box = descriptorOf("environment-box", "Box");
const b = yield* servePeer(box, McpHttpServer.toolkitRegistration(toolkit, handlers));
const context = yield* Layer.build(
layerLinkingEnvironment(descriptorOf("environment-a", "A")),
);
const links = Context.get(context, PeerLinks.PeerLinks);
const client = Context.get(context, PeerMcpClient.PeerMcpClient);
yield* linkTo(links, b);
yield* links.recordOutcome(box.environmentId, { error: "earlier outage" });
const refused = yield* client
.call({
environmentId: box.environmentId,
tool: "refuse",
arguments: {},
limits: { runtimeMode: "full-access", interactionMode: "default" },
success: Result,
})
.pipe(Effect.flip);
// The agent hears the peer's own refusal; the link answered, so it is not failing.
expect(refused).toMatchObject({ code: refusal.code, message: refusal.message });
const link = yield* links.get(box.environmentId);
expect(link._tag === "Some" ? link.value.lastError : "missing").toBeNull();
}),
).pipe(Effect.provide(NodeHttpServer.layerTest)),
);

it.effect("renews a missing peer session once and reuses its replacement", () =>
Effect.gen(function* () {
const sessionHeaders: Array<string | undefined> = [];
const statuses = [200, 202, 404, 200, 202, 200, 200];
const statuses = [200, 202, 404, 200, 202, 200, 200, 200];
const outcomes: Array<string | null> = [];
const http = HttpClient.make((request) =>
Effect.sync(() => {
const index = sessionHeaders.length;
Expand All @@ -104,13 +152,15 @@ it.effect("renews a missing peer session once and reuses its replacement", () =>
return HttpClientResponse.fromWeb(
request,
new Response(
index >= 5
? JSON.stringify({
result: {
content: [{ type: "text", text: '{"modes":"approval-required/plan"}' }],
},
})
: "",
index === 7
? "{}"
: index >= 5
? JSON.stringify({
result: {
content: [{ type: "text", text: '{"modes":"approval-required/plan"}' }],
},
})
: "",
{
status,
headers:
Expand Down Expand Up @@ -146,7 +196,7 @@ it.effect("renews a missing peer session once and reuses its replacement", () =>
url: "https://peer.example",
token: "fixture",
}),
recordOutcome: () => Effect.void,
recordOutcome: (_, outcome) => Effect.sync(() => outcomes.push(outcome.error)),
}),
),
),
Expand All @@ -164,6 +214,17 @@ it.effect("renews a missing peer session once and reuses its replacement", () =>
}),
).toEqual({ modes: "approval-required/plan" });
}
// An answer with neither a result nor an error is the link's problem, not the call's.
const empty = yield* client
.call({
environmentId: box.environmentId,
tool: "probe",
arguments: {},
limits: { runtimeMode: "approval-required", interactionMode: "plan" },
success: Result,
})
.pipe(Effect.flip);
expect(outcomes).toEqual([null, null, empty.message]);
expect(sessionHeaders).toEqual([
undefined,
"old",
Expand All @@ -172,6 +233,7 @@ it.effect("renews a missing peer session once and reuses its replacement", () =>
"renewed",
"renewed",
"renewed",
"renewed",
]);
}).pipe(Effect.scoped),
);
30 changes: 22 additions & 8 deletions apps/server/src/peer/PeerMcpClient.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import * as Context from "effect/Context";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Ref from "effect/Ref";
import * as Result from "effect/Result";
import * as Schema from "effect/Schema";
import * as HttpBody from "effect/http/HttpBody";
import * as HttpClient from "effect/http/HttpClient";
Expand Down Expand Up @@ -210,25 +211,36 @@ const make = Effect.gen(function* () {
}),
),
);
if (body.error !== undefined || body.result === undefined) {
if (body.error !== undefined) {
return Result.fail(
new OrchestratorMcpFailure({ code: "invalid_request", message: body.error.message }),
);
}
if (body.result === undefined) {
return yield* new OrchestratorMcpFailure({
code: "invalid_request",
message: body.error?.message ?? "The linked environment returned no result.",
message: "The linked environment returned no result.",
});
}
const text = body.result.content.map((part) => part.text ?? "").join("");
if (body.result.isError === true) {
// A refusal keeps the peer's own code, so the agent sees why it was refused.
const refused = decodeFailure(text);
return yield* refused._tag === "Some"
? new OrchestratorMcpFailure({ code: refused.value.code, message: refused.value.message })
: new OrchestratorMcpFailure({
code: "orchestration_error",
message: text || "The linked environment refused the call.",
});
return Result.fail(
refused._tag === "Some"
? new OrchestratorMcpFailure({
code: refused.value.code,
message: refused.value.message,
})
: new OrchestratorMcpFailure({
code: "orchestration_error",
message: text || "The linked environment refused the call.",
}),
);
}
const success = Schema.fromJsonString(input.success);
return yield* Schema.decodeEffect(success)(text).pipe(
Effect.map(Result.succeed),
Effect.mapError(
() =>
new OrchestratorMcpFailure({
Expand All @@ -244,6 +256,7 @@ const make = Effect.gen(function* () {
Effect.tapError((error) =>
links.recordOutcome(input.environmentId, { error: error.message }),
),
// A refusal is still an answer: the link works, so only this call fails.
Effect.tap(() => links.recordOutcome(input.environmentId, { error: null })),
// Recorded above with the transport's own message; the agent hears it plainly.
Effect.catchTags({
Expand All @@ -255,6 +268,7 @@ const make = Effect.gen(function* () {
}),
),
}),
Effect.flatMap(Effect.fromResult),
),
);

Expand Down
Loading