1import { NodeHttpServer } from "@effect/platform-node";
2import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
3import * as NodeServices from "@effect/platform-node/NodeServices";
4import {
5 CommandId,
6 EventId,
7 MessageId,
8 NodeId,
9 OrchestratorMcpFailure,
10 ProjectId,
11 ProviderDriverKind,
12 ProviderInstanceId,
13 ProviderThreadId,
14 type RuntimeMode,
15 RunId,
16 ThreadId,
17 TurnItemId,
18} from "@t3tools/contracts";
19import { expect, it } from "@effect/vitest";
20import * as Clock from "effect/Clock";
21import * as Context from "effect/Context";
22import * as DateTime from "effect/DateTime";
23import * as Deferred from "effect/Deferred";
24import * as Duration from "effect/Duration";
25import * as Effect from "effect/Effect";
26import * as Fiber from "effect/Fiber";
27import * as HttpServer from "effect/http/HttpServer";
28import * as HttpServerResponse from "effect/http/HttpServerResponse";
29import * as Layer from "effect/Layer";
30import * as Option from "effect/Option";
31import * as Schema from "effect/Schema";
32import * as SqlClient from "effect/sql/SqlClient";
33import { McpSchema, McpServer, Tool, Toolkit } from "effect/ai";
34
35import * as ServerConfig from "../config.ts";
36import * as McpHttpServer from "../mcp/McpHttpServer.ts";
37import * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
38import * as McpToolAccess from "../mcp/McpToolAccess.ts";
39import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts";
40import * as EventSink from "../orchestration-v2/EventSink.ts";
41import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
42import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
43import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
44import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
45import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
46import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
47import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts";
48import * as SqlitePersistence from "../persistence/Sqlite.ts";
49import * as ProjectService from "../project/ProjectService.ts";
50import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
51import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
52import * as SecretRequests from "../secrets/SecretRequests.ts";
53import * as ServerSettings from "../serverSettings.ts";
54import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts";
55import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts";
56import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
57import * as PeerForwarding from "./PeerForwarding.ts";
58import * as PeerLinkRequests from "./PeerLinkRequests.ts";
59import * as PeerLinks from "./PeerLinks.ts";
60import {
61 descriptorOf,
62 layerLinkingEnvironment,
63 linkTo,
64 servePeer,
65 type ServedPeer,
66} from "./PeerLinks.testkit.ts";
67import * as RemoteDelegation from "./RemoteDelegation.ts";
68
69// The laptop's agent asks the user to link the box. The laptop is its real
70// orchestrator and toolkit; the box is its real descriptor and MCP OAuth on a
71// socket. The user's answer goes through PeerLinkRequests, as the card's does.
72
73const laptop = descriptorOf("environment-laptop", "Laptop");
74const box = descriptorOf("environment-box", "Box");
75const instanceId = ProviderInstanceId.make("codex");
76const driver = ProviderDriverKind.make("codex");
77const modelSelection = { instanceId, model: "gpt-5.4" };
78const threadId = ThreadId.make("thread:laptop");
79const runId = RunId.make("run:laptop");
80const rootNode = NodeId.make("node:laptop:root");
81
82/** The box needs no tools of its own: linking only signs in to its /mcp. */
83const NoTools = Toolkit.make(
84 Tool.make("noop", { success: Schema.Struct({}), failure: OrchestratorMcpFailure }),
85);
86const serveBox = servePeer(
87 box,
88 McpHttpServer.toolkitRegistration(
89 NoTools,
90 McpToolAccess.toLayer(NoTools, { noop: McpToolAccess.reads(() => Effect.succeed({})) }),
91 ),
92);
93
94const adapter = {
95 instanceId,
96 driver,
97 getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
98 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
99 openSession: () => Effect.die("No provider process needed"),
100} as ProviderAdapter.ProviderAdapterV2["Service"];
101
102const mcpClient = McpSchema.McpServerClient.of({
103 clientId: 1,
104 protocolVersion: "2025-06-18",
105 clientCapabilities: {},
106 clientInfo: { name: "peer-link-requests", version: "1" },
107 initializePayload: {
108 protocolVersion: "2025-06-18",
109 capabilities: {},
110 clientInfo: { name: "peer-link-requests", version: "1" },
111 },
112 getClient: Effect.die("unused"),
113});
114
115const scope: McpInvocationContext.McpInvocationScope = {
116 environmentId: laptop.environmentId,
117 requestNamespace: "provider-session:laptop",
118 thread: {
119 threadId,
120 providerSessionId: "provider-session:laptop",
121 providerInstanceId: instanceId,
122 },
123 client: undefined,
124 capabilities: new Set(["orchestration"]),
125 issuedAt: 1,
126};
127
128/** The laptop: a real orchestrator, its orchestrator toolkit, and its links. */
129const makeLaptop = Effect.gen(function* () {
130 const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build);
131 // The orchestrator's own database, apart from the one links are kept in.
132 const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build);
133 const layerDatabase = Layer.succeedContext(database);
134 const layerOrchestrator = Layer.mergeAll(
135 layerDatabase,
136 ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
137 ProviderReplayHarness.layerWithRegistry(
138 { name: "peer-link-requests" },
139 ProviderAdapterRegistry.layerFromAdapters([adapter]),
140 { databaseLayer: layerDatabase, runEffectWorker: false },
141 ),
142 ).pipe(
143 Layer.provide(
144 Layer.succeed(ProviderContinuationRequests.ProviderContinuationRequests, {
145 offer: () => Effect.void,
146 take: Effect.never,
147 }),
148 ),
149 );
150 const layerThreads = ThreadManagement.layer.pipe(
151 Layer.provide(CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase))),
152 Layer.provideMerge(Layer.merge(layerOrchestrator, ServerSettings.layerTest())),
153 );
154 const layerHere = McpHttpServer.layerOrchestratorToolkit.pipe(
155 Layer.provideMerge(McpHttpServer.layerInProcessMcpServer),
156 Layer.provideMerge(PeerLinkRequests.layer),
157 Layer.provide(Layer.mock(PeerForwarding.PeerForwarding)({})),
158 Layer.provide(Layer.mock(RemoteDelegation.RemoteDelegation)({})),
159 Layer.provide(Layer.succeedContext(linking)),
160 Layer.provide(NodeCrypto.layer),
161 Layer.provideMerge(layerThreads),
162 Layer.provide(Layer.mock(ProviderRegistry.ProviderRegistry)({})),
163 Layer.provide(Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({})),
164 Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})),
165 Layer.provide(Layer.mock(ProjectService.ProjectService)({})),
166 Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
167 Layer.provide(Layer.mock(ThreadLaunch.ThreadLaunchService)({})),
168 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
169 Layer.provide(
170 ServerConfig.layerTest(process.cwd(), { prefix: "t3-peer-link-requests-" }).pipe(
171 Layer.provide(NodeServices.layer),
172 ),
173 ),
174 Layer.provide(NodeServices.layer),
175 Layer.fresh,
176 );
177 const here = yield* Layer.build(layerHere);
178 const server = Context.get(here, McpServer.McpServer);
179 return {
180 links: Context.get(linking, PeerLinks.PeerLinks),
181 linksSql: Context.get(linking, SqlClient.SqlClient),
182 requests: Context.get(here, PeerLinkRequests.PeerLinkRequests),
183 orchestrator: Context.get(here, Orchestrator.OrchestratorV2),
184 sink: Context.get(here, EventSink.EventSinkV2),
185 projections: Context.get(here, ProjectionStore.ProjectionStoreV2),
186 sql: Context.get(database, SqlClient.SqlClient),
187 call: (args: Record<string, unknown>, callScope = scope) =>
188 server
189 .callTool({ name: "t3_environment_link", arguments: args })
190 .pipe(
191 Effect.provideService(McpInvocationContext.McpInvocationContext, callScope),
192 Effect.provideService(McpSchema.McpServerClient, mcpClient),
193 ),
194 };
195});
196type Laptop = Effect.Success<typeof makeLaptop>;
197
198/** The laptop's thread, mid-turn when its agent asks, in `runtimeMode`. */
199const seedThread = (
200 a: Laptop,
201 runtimeMode: RuntimeMode = "full-access",
202 linkOrigin?: { readonly sessionId: string; readonly label: string },
203) =>
204 Effect.gen(function* () {
205 const now = yield* DateTime.now;
206 const providerThreadId = ProviderThreadId.make("provider-thread:laptop");
207 yield* a.orchestrator.dispatch({
208 type: "thread.create",
209 createdBy: "user",
210 creationSource: "web",
211 commandId: CommandId.make("command:create:laptop"),
212 threadId,
213 projectId: ProjectId.make("project:laptop"),
214 title: "Laptop",
215 modelSelection,
216 runtimeMode,
217 interactionMode: "default",
218 branch: null,
219 worktreePath: null,
220 ...(linkOrigin === undefined ? {} : { linkOrigin }),
221 });
222 yield* a.sink.write({
223 commandId: CommandId.make("command:seed:laptop"),
224 events: [
225 {
226 id: EventId.make("event:seed-provider-thread"),
227 type: "provider-thread.updated",
228 threadId,
229 driver,
230 providerInstanceId: instanceId,
231 occurredAt: now,
232 payload: {
233 id: providerThreadId,
234 driver,
235 providerInstanceId: instanceId,
236 providerSessionId: null,
237 appThreadId: threadId,
238 ownerNodeId: rootNode,
239 nativeThreadRef: { driver, nativeId: "native:laptop", strength: "strong" },
240 nativeConversationHeadRef: null,
241 status: "active",
242 firstRunOrdinal: 1,
243 lastRunOrdinal: 1,
244 handoffIds: [],
245 forkedFrom: null,
246 createdAt: now,
247 updatedAt: now,
248 },
249 },
250 {
251 id: EventId.make("event:seed-node"),
252 type: "node.updated",
253 threadId,
254 runId,
255 nodeId: rootNode,
256 driver,
257 providerInstanceId: instanceId,
258 occurredAt: now,
259 payload: {
260 id: rootNode,
261 threadId,
262 runId,
263 parentNodeId: null,
264 rootNodeId: rootNode,
265 kind: "root_turn",
266 status: "running",
267 countsForRun: true,
268 providerThreadId,
269 providerTurnId: null,
270 nativeItemRef: null,
271 runtimeRequestId: null,
272 checkpointScopeId: null,
273 startedAt: now,
274 completedAt: null,
275 },
276 },
277 {
278 id: EventId.make("event:seed-run"),
279 type: "run.updated",
280 threadId,
281 runId,
282 nodeId: rootNode,
283 providerInstanceId: instanceId,
284 occurredAt: now,
285 payload: {
286 id: runId,
287 threadId,
288 ordinal: 1,
289 providerInstanceId: instanceId,
290 modelSelection,
291 providerThreadId,
292 userMessageId: MessageId.make("message:seed-user"),
293 rootNodeId: rootNode,
294 activeAttemptId: null,
295 status: "running",
296 requestedAt: now,
297 startedAt: now,
298 completedAt: null,
299 checkpointId: null,
300 contextHandoffId: null,
301 },
302 },
303 ],
304 });
305 });
306
307const cardOf = (a: Laptop) =>
308 a.orchestrator
309 .getThreadProjection(threadId)
310 .pipe(
311 Effect.map((projection) => projection.turnItems.find((item) => item.type === "link_request")),
312 );
313
314/** The open card the agent's call left, as the user's client sees it. */
315const pendingCard = (a: Laptop) =>
316 Effect.gen(function* () {
317 const card = yield* cardOf(a);
318 expect(card).toMatchObject({ status: "waiting", linkStatus: "pending" });
319 return card as Extract<NonNullable<typeof card>, { readonly type: "link_request" }>;
320 });
321
322/** The outcome notifications the thread received: what the agent reads and the user sees. */
323const outcomeMessages = (a: Laptop) =>
324 a.orchestrator.getThreadProjection(threadId).pipe(
325 Effect.map((projection) => ({
326 messages: projection.messages.filter((message) => message.notification !== undefined),
327 items: projection.turnItems.filter((item) => item.type === "notification"),
328 })),
329 );
330
331/** The agent's turn ends; the card outlives it. */
332const endRun = (a: Laptop) =>
333 Effect.gen(function* () {
334 const projection = yield* a.orchestrator.getThreadProjection(threadId);
335 const run = projection.runs.find((candidate) => candidate.id === runId)!;
336 const now = yield* DateTime.now;
337 yield* a.sink.write({
338 commandId: CommandId.make("command:end-run:laptop"),
339 events: [
340 {
341 id: EventId.make("event:end-run"),
342 type: "run.updated",
343 threadId,
344 runId,
345 nodeId: rootNode,
346 providerInstanceId: instanceId,
347 occurredAt: now,
348 payload: { ...run, status: "completed", completedAt: now },
349 },
350 ],
351 });
352 });
353
354/** A steer restarts the agent's turn: its run gets a new root node. */
355const restartRun = (a: Laptop) =>
356 Effect.gen(function* () {
357 const projection = yield* a.orchestrator.getThreadProjection(threadId);
358 const run = projection.runs.find((candidate) => candidate.id === runId)!;
359 const node = projection.nodes.find((candidate) => candidate.id === rootNode)!;
360 const restarted = NodeId.make("node:laptop:root:2");
361 const now = yield* DateTime.now;
362 yield* a.sink.write({
363 commandId: CommandId.make("command:restart-run:laptop"),
364 events: [
365 {
366 id: EventId.make("event:restart-node"),
367 type: "node.updated",
368 threadId,
369 runId,
370 nodeId: restarted,
371 driver,
372 providerInstanceId: instanceId,
373 occurredAt: now,
374 payload: { ...node, id: restarted, rootNodeId: restarted, startedAt: now },
375 },
376 {
377 id: EventId.make("event:restart-run"),
378 type: "run.updated",
379 threadId,
380 runId,
381 nodeId: restarted,
382 providerInstanceId: instanceId,
383 occurredAt: now,
384 payload: { ...run, rootNodeId: restarted },
385 },
386 ],
387 });
388 });
389
390/** What `request` records for a new card, as if it read the run before anything changed. */
391const recordPendingCard = (a: Laptop, key: string) =>
392 a.orchestrator.dispatch({
393 type: "link_request.record",
394 commandId: CommandId.make(`command:pending:${key}`),
395 threadId,
396 runId,
397 nodeId: rootNode,
398 turnItemId: TurnItemId.make(`turn-item:link-request:${key}`),
399 reason: "",
400 linkStatus: "pending",
401 });
402
403/**
404 * Every row of every table in the laptop's database, as text, to scan for
405 * anything leaked: events, projections, receipts and outbox alike.
406 */
407const storedText = (a: Laptop) =>
408 Effect.gen(function* () {
409 const tables = yield* a.sql<{ name: string }>`
410 SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'
411 `;
412 const rows = yield* Effect.forEach(tables, ({ name }) =>
413 a.sql.unsafe<Record<string, unknown>>(`SELECT * FROM "${name}"`),
414 );
415 const text = rows
416 .flat()
417 .map((row) => JSON.stringify(row))
418 .join("\n");
419 // The scan saw the card itself, so it read what the thread recorded.
420 expect(text).toContain("link_request");
421 return text;
422 });
423
424const errorOf = (result: McpSchema.CallToolResult) => {
425 const text = result.content[0];
426 return result.isError === true && text?.type === "text" ? JSON.parse(text.text) : undefined;
427};
428
429/** Runs `effect` a year from now, when every link made today has expired. */
430const afterLinksExpire = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
431 Effect.gen(function* () {
432 const clock = yield* Clock.Clock;
433 const ahead = Duration.days(365);
434 const later = () => clock.currentTimeMillisUnsafe() + Duration.toMillis(ahead);
435 const laterNanos = () => clock.currentTimeNanosUnsafe() + Duration.toNanosUnsafe(ahead);
436 return yield* effect.pipe(
437 Effect.provideService(Clock.Clock, {
438 currentTimeMillisUnsafe: later,
439 currentTimeMillis: Effect.sync(later),
440 currentTimeNanosUnsafe: laterNanos,
441 currentTimeNanos: Effect.sync(laterNanos),
442 monotonicTimeNanosUnsafe: () => clock.monotonicTimeNanosUnsafe(),
443 monotonicTimeNanos: clock.monotonicTimeNanos,
444 sleep: (duration) => clock.sleep(duration),
445 }),
446 );
447 });
448
449const withBoth = <A, E, R>(body: (a: Laptop, b: ServedPeer) => Effect.Effect<A, E, R>) =>
450 Effect.scoped(
451 Effect.gen(function* () {
452 const b = yield* serveBox;
453 const a = yield* makeLaptop;
454 return yield* body(a, b);
455 }),
456 ).pipe(Effect.provide(NodeHttpServer.layerTest));
457
458it.live("the agent asks without waiting, and the user links the box after its turn ended", () =>
459 withBoth((a, b) =>
460 Effect.gen(function* () {
461 yield* seedThread(a);
462 const asked = yield* a.call({
463 url: b.url,
464 reason: "Run the suite on the box.",
465 requestedAccess: "approval-required",
466 clientRequestId: "link-box",
467 });
468 const card = yield* pendingCard(a);
469 // The call returns at once: no tool call outlives a provider's timeout.
470 expect(asked.isError).toBe(false);
471 expect(asked.structuredContent).toEqual({ status: "pending", turnItemId: card.id });
472 expect(card).toMatchObject({
473 url: b.url,
474 reason: "Run the suite on the box.",
475 requestedAccess: "approval-required",
476 });
477 // Anything can answer at an address as the box, so the card names no machine.
478 expect(card.environmentId).toBeUndefined();
479 expect(card.label).toBeUndefined();
480 // The agent waits on the user, so the thread asks for input.
481 expect((yield* a.orchestrator.getThreadShell(threadId))?.pendingRuntimeRequest).toMatchObject(
482 { kind: "user_input" },
483 );
484
485 for (const secretStatus of ["pending", "saved"] as const) {
486 yield* a.orchestrator.dispatch({
487 type: "secret_request.record",
488 commandId: CommandId.make("command:secret-" + secretStatus),
489 threadId,
490 runId,
491 nodeId: rootNode,
492 turnItemId: TurnItemId.make("turn-item:snoozed-secret"),
493 label: "Credential",
494 reason: "Connect",
495 secretStatus,
496 });
497 if (secretStatus === "pending") {
498 yield* a.orchestrator.dispatch({
499 type: "thread.snooze",
500 commandId: CommandId.make("command:snooze-secret"),
501 threadId,
502 snoozedUntil: "2099-01-01T00:00:00.000Z",
503 });
504 expect((yield* a.projections.getThread(threadId)).snoozedUntil).not.toBeNull();
505 } else {
506 expect((yield* a.projections.getThread(threadId)).snoozedUntil).toBeNull();
507 }
508 }
509
510 // The agent ends its turn; the card stays answerable, and the thread still asks.
511 yield* endRun(a);
512 expect(yield* cardOf(a)).toMatchObject({ status: "waiting", linkStatus: "pending" });
513 expect((yield* a.orchestrator.getThreadShell(threadId))?.pendingRuntimeRequest).toMatchObject(
514 { kind: "user_input" },
515 );
516 expect(
517 ProjectionStore.threadShellFromProjection(
518 yield* a.orchestrator.getThreadProjection(threadId),
519 ).pendingRuntimeRequest,
520 ).toMatchObject({ kind: "user_input" });
521 expect(yield* a.projections.getSettlementCandidates(threadId)).toEqual([]);
522
523 // Once the user writes in the thread again, the card stops asking but stays answerable.
524 const wroteAt = DateTime.add(card.startedAt!, { seconds: 1 });
525 yield* a.sink.write({
526 commandId: CommandId.make("command:user-writes:laptop"),
527 events: [
528 {
529 id: EventId.make("event:user-writes"),
530 type: "message.updated",
531 threadId,
532 occurredAt: wroteAt,
533 payload: {
534 createdBy: "user",
535 creationSource: "web",
536 id: MessageId.make("message:user-writes"),
537 threadId,
538 runId: null,
539 nodeId: null,
540 role: "user",
541 text: "Something else",
542 attachments: [],
543 streaming: false,
544 createdAt: wroteAt,
545 updatedAt: wroteAt,
546 },
547 },
548 ],
549 });
550 expect((yield* a.orchestrator.getThreadShell(threadId))?.pendingRuntimeRequest).toBeNull();
551 expect(
552 ProjectionStore.threadShellFromProjection(
553 yield* a.orchestrator.getThreadProjection(threadId),
554 ).pendingRuntimeRequest,
555 ).toBeNull();
556 expect(
557 (yield* a.projections.getSettlementCandidates(threadId)).map((thread) => thread.id),
558 ).toEqual([threadId]);
559 expect(yield* cardOf(a)).toMatchObject({ status: "waiting", linkStatus: "pending" });
560
561 yield* a.orchestrator.dispatch({
562 type: "thread.snooze",
563 commandId: CommandId.make("command:snooze-link"),
564 threadId,
565 snoozedUntil: "2099-01-01T00:00:00.000Z",
566 });
567 expect((yield* a.projections.getThread(threadId)).snoozedUntil).not.toBeNull();
568
569 // The user's client mints a code on the box and picks more than the agent suggested.
570 const pairing = yield* b.auth.issuePairingCredential();
571 yield* a.requests.answer({
572 threadId,
573 turnItemId: card.id,
574 answer: { type: "link", urls: [b.url], access: "auto", pairingCode: pairing.credential },
575 });
576
577 expect((yield* a.projections.getThread(threadId)).snoozedUntil).toBeNull();
578
579 const link = yield* a.links.get(box.environmentId);
580 expect(Option.getOrUndefined(link)?.access).toBe("auto");
581 expect(yield* cardOf(a)).toMatchObject({
582 status: "completed",
583 linkStatus: "linked",
584 linkedEnvironmentId: box.environmentId,
585 linkedLabel: "Box",
586 linkedAccess: "auto",
587 });
588 expect((yield* b.linkedSessions).length).toBe(1);
589
590 // The agent hears it as a notification in its thread, not a user message.
591 const { messages, items } = yield* outcomeMessages(a);
592 expect(messages).toHaveLength(1);
593 expect(messages[0]).toMatchObject({
594 role: "user",
595 createdBy: "agent",
596 creationSource: "server",
597 notification: { outcome: "completed", summary: "Linked Box" },
598 });
599 expect(messages[0]!.text).toContain(box.environmentId);
600 expect(messages[0]!.text).toContain("auto");
601 expect(items).toHaveLength(1);
602 expect(items[0]).toMatchObject({ type: "notification", summary: "Linked Box" });
603
604 // The code appears in nothing the thread recorded or the agent received.
605 expect(yield* storedText(a)).not.toContain(pairing.credential);
606
607 // A card is answered once.
608 const again = yield* a.requests
609 .answer({ threadId, turnItemId: card.id, answer: { type: "decline" } })
610 .pipe(Effect.flip);
611 expect(again.reason).toBe("already_answered");
612
613 // A retry that lost the first result gets the card's outcome, and records nothing.
614 const before = (yield* a.orchestrator.getThreadProjection(threadId)).turnItems.length;
615 const retried = yield* a.call({ url: b.url, clientRequestId: "link-box" });
616 expect(retried.structuredContent).toEqual({
617 status: "linked",
618 environmentId: box.environmentId,
619 label: "Box",
620 access: "auto",
621 });
622 expect((yield* a.orchestrator.getThreadProjection(threadId)).turnItems).toHaveLength(before);
623 expect((yield* outcomeMessages(a)).messages).toHaveLength(1);
624 }),
625 ),
626);
627
628it.live("a retry of a pending request returns pending and shows no second card", () =>
629 withBoth((a, b) =>
630 Effect.gen(function* () {
631 yield* seedThread(a);
632 const first = yield* a.call({ url: b.url, clientRequestId: "link-box" });
633 const retried = yield* a.call({ url: b.url, clientRequestId: "link-box" });
634 expect(retried.structuredContent).toEqual(first.structuredContent);
635 const cards = (yield* a.orchestrator.getThreadProjection(threadId)).turnItems.filter(
636 (item) => item.type === "link_request",
637 );
638 expect(cards).toHaveLength(1);
639 }),
640 ),
641);
642
643it.live("a new MCP credential reusing a clientRequestId gets a card of its own", () =>
644 withBoth((a, b) =>
645 Effect.gen(function* () {
646 yield* seedThread(a);
647 const first = yield* a.call({ url: b.url, clientRequestId: "link" });
648 const later = yield* a.call(
649 { environmentId: box.environmentId, clientRequestId: "link" },
650 {
651 ...scope,
652 requestNamespace: "provider-session:laptop-2",
653 thread: { ...scope.thread!, providerSessionId: "provider-session:laptop-2" },
654 },
655 );
656 expect(later.structuredContent).toMatchObject({ status: "pending" });
657 expect(later.structuredContent).not.toEqual(first.structuredContent);
658 const cards = (yield* a.orchestrator.getThreadProjection(threadId)).turnItems.filter(
659 (item) => item.type === "link_request",
660 );
661 expect(cards).toHaveLength(2);
662 }),
663 ),
664);
665
666it.live("a declined request links nothing and tells the agent", () =>
667 withBoth((a, b) =>
668 Effect.gen(function* () {
669 yield* seedThread(a);
670 yield* a.call({ url: b.url });
671 const card = yield* pendingCard(a);
672 yield* a.requests.answer({ threadId, turnItemId: card.id, answer: { type: "decline" } });
673 expect(Option.isNone(yield* a.links.get(box.environmentId))).toBe(true);
674 expect(yield* cardOf(a)).toMatchObject({ status: "cancelled", linkStatus: "declined" });
675 const { messages } = yield* outcomeMessages(a);
676 expect(messages).toHaveLength(1);
677 expect(messages[0]).toMatchObject({
678 notification: { outcome: "cancelled", summary: `Declined to link ${b.url}` },
679 });
680 expect(messages[0]!.text).toContain("declined");
681 }),
682 ),
683);
684
685it.live("stopping the thread closes its open card, which then takes no answer", () =>
686 withBoth((a, b) =>
687 Effect.gen(function* () {
688 yield* seedThread(a);
689 yield* a.call({ url: b.url });
690 const card = yield* pendingCard(a);
691 yield* a.orchestrator.dispatch({
692 type: "thread.stop",
693 commandId: CommandId.make("command:stop:laptop"),
694 threadId,
695 });
696 expect(yield* cardOf(a)).toMatchObject({ status: "cancelled", linkStatus: "cancelled" });
697 const late = yield* a.requests
698 .answer({
699 threadId,
700 turnItemId: card.id,
701 answer: { type: "link", urls: [b.url], access: "auto", pairingCode: "late-code" },
702 })
703 .pipe(Effect.flip);
704 expect(late.reason).toBe("already_answered");
705 expect(Option.isNone(yield* a.links.get(box.environmentId))).toBe(true);
706 // An answer already linking when Stop landed keeps the card closed and wakes nobody.
707 yield* a.orchestrator.dispatch({
708 type: "link_request.record",
709 commandId: CommandId.make("command:late-answer:laptop"),
710 threadId,
711 runId,
712 nodeId: rootNode,
713 turnItemId: card.id,
714 url: card.url,
715 reason: card.reason,
716 linkStatus: "declined",
717 wake: {
718 messageId: MessageId.make("message:late-answer"),
719 text: "The user declined.",
720 notification: {
721 source: { kind: "background_task" },
722 outcome: "cancelled",
723 summary: "Declined",
724 },
725 },
726 });
727 expect(yield* cardOf(a)).toMatchObject({ status: "cancelled", linkStatus: "cancelled" });
728 // The user stopped the thread: nothing wakes the agent.
729 expect((yield* outcomeMessages(a)).messages).toHaveLength(0);
730 }),
731 ),
732);
733
734it.live("a request that read the run before Stop shows no card once Stop lands", () =>
735 withBoth((a, b) =>
736 Effect.gen(function* () {
737 yield* seedThread(a);
738 yield* a.call({ url: b.url });
739 const open = yield* pendingCard(a);
740 yield* a.orchestrator.dispatch({
741 type: "thread.stop",
742 commandId: CommandId.make("command:stop:laptop"),
743 threadId,
744 });
745 const late = yield* recordPendingCard(a, "after-stop").pipe(Effect.flip);
746 expect(String(late.cause)).toContain("not active");
747 const cards = (yield* a.orchestrator.getThreadProjection(threadId)).turnItems.filter(
748 (item) => item.type === "link_request",
749 );
750 expect(cards.map((card) => card.id)).toEqual([open.id]);
751 }),
752 ),
753);
754
755it.live("a card shows up only while its run is going, and takes an answer after", () =>
756 withBoth((a, b) =>
757 Effect.gen(function* () {
758 yield* seedThread(a);
759 yield* a.call({ url: b.url });
760 const card = yield* pendingCard(a);
761 yield* endRun(a);
762 const late = yield* recordPendingCard(a, "after-end").pipe(Effect.flip);
763 expect(String(late.cause)).toContain("not active");
764 yield* a.requests.answer({ threadId, turnItemId: card.id, answer: { type: "decline" } });
765 expect(yield* cardOf(a)).toMatchObject({ linkStatus: "declined" });
766 }),
767 ),
768);
769
770it.live("a steer that restarts the turn leaves its open card answerable", () =>
771 withBoth((a, b) =>
772 Effect.gen(function* () {
773 yield* seedThread(a);
774 yield* a.call({ url: b.url });
775 const card = yield* pendingCard(a);
776 yield* restartRun(a);
777 const pairing = yield* b.auth.issuePairingCredential();
778 yield* a.requests.answer({
779 threadId,
780 turnItemId: card.id,
781 answer: { type: "link", urls: [b.url], access: "auto", pairingCode: pairing.credential },
782 });
783 // The card keeps the node it was asked from, and the agent hears the answer.
784 expect(yield* cardOf(a)).toMatchObject({
785 linkStatus: "linked",
786 nodeId: rootNode,
787 linkedEnvironmentId: box.environmentId,
788 });
789 expect((yield* outcomeMessages(a)).messages).toHaveLength(1);
790 }),
791 ),
792);
793
794it.live("a slow machine holds up only its own card's answer", () =>
795 withBoth((a, b) =>
796 Effect.scoped(
797 Effect.gen(function* () {
798 // A machine that takes the descriptor request and answers when the test says.
799 const reached = yield* Deferred.make<void>();
800 const release = yield* Deferred.make<void>();
801 const slow = yield* Layer.build(Layer.fresh(NodeHttpServer.layerTest));
802 const server = Context.get(slow, HttpServer.HttpServer);
803 yield* server.serve(
804 Deferred.succeed(reached, undefined).pipe(
805 Effect.andThen(Deferred.await(release)),
806 Effect.as(HttpServerResponse.empty({ status: 404 })),
807 ),
808 );
809 if (server.address._tag === "UnixPathAddress") return yield* Effect.die("expected TCP");
810 const slowUrl = `http://127.0.0.1:${server.address.port}`;
811
812 yield* seedThread(a);
813 // Recorded directly: asking would probe the slow machine before the answer does.
814 const slowCard = { id: TurnItemId.make("turn-item:link-request:slow") };
815 yield* a.orchestrator.dispatch({
816 type: "link_request.record",
817 commandId: CommandId.make("command:pending:slow"),
818 threadId,
819 runId,
820 nodeId: rootNode,
821 turnItemId: slowCard.id,
822 url: slowUrl,
823 reason: "",
824 linkStatus: "pending",
825 });
826 yield* a.call({ url: b.url, clientRequestId: "box" });
827 const boxCard = (yield* a.orchestrator.getThreadProjection(threadId)).turnItems.find(
828 (item) => item.type === "link_request" && item.url === b.url,
829 )!;
830
831 const slowAnswer = yield* a.requests
832 .answer({
833 threadId,
834 turnItemId: slowCard.id,
835 answer: { type: "link", urls: [slowUrl], access: "auto", pairingCode: "code" },
836 })
837 .pipe(Effect.forkChild);
838 yield* Deferred.await(reached);
839
840 // While the slow machine holds its card, the user answers the other one.
841 yield* a.requests.answer({ threadId, turnItemId: boxCard.id, answer: { type: "decline" } });
842 const items = (yield* a.orchestrator.getThreadProjection(threadId)).turnItems;
843 expect(items.find((item) => item.id === boxCard.id)).toMatchObject({
844 linkStatus: "declined",
845 });
846 expect(items.find((item) => item.id === slowCard.id)).toMatchObject({
847 linkStatus: "pending",
848 });
849
850 yield* Deferred.succeed(release, undefined);
851 yield* Fiber.join(slowAnswer);
852 const failed = (yield* a.orchestrator.getThreadProjection(threadId)).turnItems.find(
853 (item) => item.id === slowCard.id,
854 );
855 expect(failed).toMatchObject({ linkStatus: "failed" });
856 }),
857 ),
858 ),
859);
860
861it.live("archiving the thread closes its open card", () =>
862 withBoth((a, b) =>
863 Effect.gen(function* () {
864 yield* seedThread(a);
865 yield* a.call({ url: b.url });
866 yield* pendingCard(a);
867 yield* endRun(a);
868 yield* a.orchestrator.dispatch({
869 type: "thread.archive",
870 commandId: CommandId.make("command:archive:laptop"),
871 threadId,
872 });
873 expect(yield* cardOf(a)).toMatchObject({ status: "cancelled", linkStatus: "cancelled" });
874 }),
875 ),
876);
877
878it.live("a code the box refuses fails the request with its reason, and links nothing", () =>
879 withBoth((a, b) =>
880 Effect.gen(function* () {
881 yield* seedThread(a);
882 yield* a.call({ url: b.url });
883 const card = yield* pendingCard(a);
884 yield* a.requests.answer({
885 threadId,
886 turnItemId: card.id,
887 answer: {
888 type: "link",
889 urls: [b.url],
890 access: "auto",
891 pairingCode: "not-a-real-code",
892 },
893 });
894 const failed = yield* cardOf(a);
895 expect(failed).toMatchObject({ status: "failed", linkStatus: "failed" });
896 const reason = failed?.type === "link_request" ? failed.failure : undefined;
897 expect(reason?.length).toBeGreaterThan(0);
898 const { messages } = yield* outcomeMessages(a);
899 expect(messages[0]).toMatchObject({ notification: { outcome: "failed" } });
900 expect(messages[0]!.text).toContain(reason);
901 expect(Option.isNone(yield* a.links.get(box.environmentId))).toBe(true);
902 expect(yield* storedText(a)).not.toContain("not-a-real-code");
903 }),
904 ),
905);
906
907it.live("a card an older server left naming a machine beside its address links no machine", () =>
908 withBoth((a, b) =>
909 Effect.gen(function* () {
910 yield* seedThread(a);
911 yield* a.orchestrator.dispatch({
912 type: "link_request.record",
913 commandId: CommandId.make("command:pending:older"),
914 threadId,
915 runId,
916 nodeId: rootNode,
917 turnItemId: TurnItemId.make("turn-item:link-request:older"),
918 url: b.url,
919 environmentId: box.environmentId,
920 label: "Box",
921 reason: "",
922 linkStatus: "pending",
923 });
924 const card = yield* pendingCard(a);
925 // An older client takes the machine from the card and mints a code on it.
926 const pairing = yield* b.auth.issuePairingCredential();
927 yield* a.requests.answer({
928 threadId,
929 turnItemId: card.id,
930 answer: {
931 type: "link",
932 environmentId: box.environmentId,
933 label: "Box",
934 urls: [b.url],
935 access: "auto",
936 pairingCode: pairing.credential,
937 },
938 });
939 expect(yield* cardOf(a)).toMatchObject({ linkStatus: "failed" });
940 expect(Option.isNone(yield* a.links.get(box.environmentId))).toBe(true);
941 expect(yield* b.linkedSessions).toEqual([]);
942 }),
943 ),
944);
945
946it.live("an agent below full access may not ask, and no card is shown", () =>
947 withBoth((a, b) =>
948 Effect.gen(function* () {
949 yield* seedThread(a, "auto");
950 const refused = yield* a.call({ url: b.url });
951 expect(errorOf(refused)).toMatchObject({ code: "capability_denied" });
952 expect(yield* cardOf(a)).toBeUndefined();
953 }),
954 ),
955);
956
957it.live("work another environment's link started may not ask, and no card is shown", () =>
958 withBoth((a, b) =>
959 Effect.gen(function* () {
960 yield* seedThread(a, "full-access", { sessionId: "session:desk", label: "T3 Code · Desk" });
961 const refused = yield* a.call({ url: b.url });
962 expect(errorOf(refused)).toMatchObject({ code: "capability_denied" });
963 expect(errorOf(refused).message).toContain("T3 Code · Desk");
964 expect(yield* cardOf(a)).toBeUndefined();
965 }),
966 ),
967);
968
969it.live(
970 "an agent that names no machine leaves the pick to the user, who links one by its addresses",
971 () =>
972 withBoth((a, b) =>
973 Effect.gen(function* () {
974 yield* seedThread(a);
975 const asked = yield* a.call({ reason: "Delegate the build to the vps.", hint: "vps" });
976 expect(asked.structuredContent).toMatchObject({ status: "pending" });
977 const card = yield* pendingCard(a);
978 expect(card).toMatchObject({ reason: "Delegate the build to the vps.", hint: "vps" });
979 expect(card.url).toBeUndefined();
980 expect(card.environmentId).toBeUndefined();
981
982 // The user picks the box; their client knows a stale address for it first.
983 const pairing = yield* b.auth.issuePairingCredential();
984 yield* a.requests.answer({
985 threadId,
986 turnItemId: card.id,
987 answer: {
988 type: "link",
989 environmentId: box.environmentId,
990 label: "Box",
991 urls: ["http://127.0.0.1:9", b.url],
992 access: "auto",
993 pairingCode: pairing.credential,
994 },
995 });
996 expect(Option.getOrUndefined(yield* a.links.get(box.environmentId))).toMatchObject({
997 urls: [b.url],
998 access: "auto",
999 });
1000 expect(yield* cardOf(a)).toMatchObject({
1001 linkStatus: "linked",
1002 linkedEnvironmentId: box.environmentId,
1003 });
1004 expect((yield* outcomeMessages(a)).messages[0]!.text).toContain("linked Box");
1005 }),
1006 ),
1007);
1008
1009it.live("naming a machine that is already linked returns the link at once, with no card", () =>
1010 withBoth((a, b) =>
1011 Effect.gen(function* () {
1012 yield* seedThread(a);
1013 const pairing = yield* b.auth.issuePairingCredential();
1014 yield* a.links.link({ url: b.url, pairingCode: pairing.credential, access: "auto" });
1015 const asked = yield* a.call({ environmentId: box.environmentId });
1016 expect(asked.structuredContent).toEqual({
1017 status: "linked",
1018 environmentId: box.environmentId,
1019 label: "Box",
1020 access: "auto",
1021 });
1022 expect(yield* cardOf(a)).toBeUndefined();
1023
1024 // Not both: the address could name another machine than the id.
1025 const both = yield* a.call({ environmentId: box.environmentId, url: b.url });
1026 expect(errorOf(both)).toMatchObject({ code: "invalid_request" });
1027 }),
1028 ),
1029);
1030
1031it.live("a retry naming a machine linked meanwhile closes the card it left as linked", () =>
1032 withBoth((a, b) =>
1033 Effect.gen(function* () {
1034 yield* seedThread(a);
1035 yield* a.call({ url: b.url, clientRequestId: "link-box" });
1036 yield* pendingCard(a);
1037 // Linked from Settings meanwhile, not through the card.
1038 const pairing = yield* b.auth.issuePairingCredential();
1039 yield* a.links.link({ url: b.url, pairingCode: pairing.credential, access: "auto" });
1040
1041 const retried = yield* a.call({
1042 environmentId: box.environmentId,
1043 clientRequestId: "link-box",
1044 });
1045 expect(retried.structuredContent).toEqual({
1046 status: "linked",
1047 environmentId: box.environmentId,
1048 label: "Box",
1049 access: "auto",
1050 });
1051 expect(yield* cardOf(a)).toMatchObject({
1052 linkStatus: "linked",
1053 linkedEnvironmentId: box.environmentId,
1054 });
1055 expect((yield* a.orchestrator.getThreadShell(threadId))?.pendingRuntimeRequest).toBeNull();
1056 }),
1057 ),
1058);
1059
1060it.live("a link that expired goes through the card, so the user can link it again", () =>
1061 withBoth((a, b) =>
1062 Effect.gen(function* () {
1063 yield* seedThread(a);
1064 yield* linkTo(a.links, b, "read-only");
1065 const asked = yield* a.call({ environmentId: box.environmentId }).pipe(afterLinksExpire);
1066 expect(asked.structuredContent).toMatchObject({ status: "pending" });
1067 const card = yield* pendingCard(a);
1068
1069 // Keeping the expired link is no answer: the card fails and says why.
1070 yield* a.requests
1071 .answer({
1072 threadId,
1073 turnItemId: card.id,
1074 answer: { type: "use-existing", environmentId: box.environmentId },
1075 })
1076 .pipe(afterLinksExpire);
1077 expect(yield* cardOf(a)).toMatchObject({ linkStatus: "failed" });
1078 expect((yield* outcomeMessages(a)).messages[0]!.text).toContain("expired");
1079 }),
1080 ),
1081);
1082
1083it.live("a link made before links were fenced goes through the card, as an expired one", () =>
1084 withBoth((a, b) =>
1085 Effect.gen(function* () {
1086 yield* seedThread(a);
1087 yield* linkTo(a.links, b);
1088 yield* a.linksSql`UPDATE peer_environment_links SET fenced = 0`;
1089 const asked = yield* a.call({ environmentId: box.environmentId });
1090 expect(asked.structuredContent).toMatchObject({ status: "pending" });
1091 yield* pendingCard(a);
1092 }),
1093 ),
1094);
1095
1096it.live(
1097 "a card for a machine named by id keeps it, and the user may keep a link it already has",
1098 () =>
1099 withBoth((a, b) =>
1100 Effect.gen(function* () {
1101 yield* seedThread(a);
1102 yield* a.call({ environmentId: box.environmentId });
1103 const card = yield* pendingCard(a);
1104 expect(card.environmentId).toBe(box.environmentId);
1105 // Linked from Settings meanwhile; the card answers with that link.
1106 const pairing = yield* b.auth.issuePairingCredential();
1107 yield* a.links.link({ url: b.url, pairingCode: pairing.credential, access: "read-only" });
1108 yield* a.requests.answer({
1109 threadId,
1110 turnItemId: card.id,
1111 answer: { type: "use-existing", environmentId: box.environmentId },
1112 });
1113 expect(yield* cardOf(a)).toMatchObject({
1114 linkStatus: "linked",
1115 linkedAccess: "read-only",
1116 });
1117 expect((yield* b.linkedSessions).length).toBe(1);
1118 }),
1119 ),
1120);