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
55 changes: 54 additions & 1 deletion apps/server/src/cli/thread.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,14 @@ import { assert, it } from "@effect/vitest";
import { RunId, ThreadId } from "@t3tools/contracts";
import * as Effect from "effect/Effect";

import { readLatestMessages, type ThreadTools, titleFromMessage, waitForRun } from "./thread.ts";
import {
handoffThread,
ThreadUsageError,
readLatestMessages,
type ThreadTools,
titleFromMessage,
waitForRun,
} from "./thread.ts";

const threadId = ThreadId.make("thread-1");
const runId = RunId.make("run-1");
Expand Down Expand Up @@ -165,3 +172,49 @@ it("titles from a long first line without splitting an emoji", () => {
const title = titleFromMessage(`${"a".repeat(78)}😀 and the rest of a long first line`);
assert.strictEqual(title, `${"a".repeat(78)}…`);
});
it.effect.each(["wait", "interrupt"] as const)(
"handoff asks once before applying the %s policy",
(policy) =>
Effect.gen(function* () {
const blocked = { threadId, attached: true, readOnly: false, blocked: true };
const done = { threadId, attached: false, readOnly: false };
const { tools, calls } = scriptedTools({
t3_thread_attach: policy === "wait" ? [blocked, blocked, done] : [blocked, done],
});
let choices = 0;
const result = yield* handoffThread(
tools,
threadId,
Effect.sync(() => {
choices++;
return policy;
}),
Effect.void,
);
assert.deepStrictEqual(result, done);
assert.strictEqual(choices, 1);
assert.deepStrictEqual(
calls.map((call) => call.args),
policy === "wait"
? [0, 1, 2].map(() => ({ action: "handoff", threadId }))
: [
{ action: "handoff", threadId },
{ action: "handoff", threadId, interrupt: true },
],
);
}),
);
it.effect("cancelled handoff never interrupts the source", () =>
Effect.gen(function* () {
const { tools, calls } = scriptedTools({
t3_thread_attach: [{ threadId, attached: true, readOnly: false, blocked: true }],
});
yield* handoffThread(
tools,
threadId,
Effect.fail(new ThreadUsageError({ detail: "cancelled" })),
Effect.void,
).pipe(Effect.flip);
assert.strictEqual(calls.length, 1);
}),
);
157 changes: 155 additions & 2 deletions apps/server/src/cli/thread.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import {
ClaudeSessionAttachmentResult,
OrchestratorMcpFailure,
OrchestratorMcpThreadListResult,
OrchestratorMcpThreadReadResult,
Expand All @@ -24,7 +25,7 @@ import * as Runtime from "effect/Runtime";
import * as Schema from "effect/Schema";
import * as Stdio from "effect/Stdio";
import * as Stream from "effect/Stream";
import { Argument, Command, Flag, GlobalFlag } from "effect/cli";
import { Argument, Command, Flag, GlobalFlag, Prompt } from "effect/cli";

import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
import * as ServerConfig from "../config.ts";
Expand Down Expand Up @@ -696,7 +697,7 @@ const threadSendCommand = Command.make("send", {
{ threadId, message, mode: "queue" },
OrchestratorMcpThreadSendResult,
);
if (flags.wait) {
if (flags.wait && sent.runId !== null) {
return yield* finishRun(tools, { threadId, runId: sent.runId, json: flags.json });
}
yield* flags.json
Expand Down Expand Up @@ -762,9 +763,161 @@ const threadShowCommand = Command.make("show", {
),
);

type HandoffFlags = { readonly wait: boolean; readonly interrupt: boolean; readonly json: boolean };
const handoffFlags = {
wait: Flag.Boolean("wait").pipe(
Flag.withDescription("Wait for the source CLI and its agents to exit."),
Flag.withDefault(false),
),
interrupt: Flag.Boolean("interrupt").pipe(
Flag.withDescription("Stop the source CLI and its agents, then resume work in T3 (Linux)."),
Flag.withDefault(false),
),
};
const validateHandoffFlags = (flags: HandoffFlags) =>
flags.wait && flags.interrupt
? Effect.fail(new ThreadUsageError({ detail: "Choose --wait or --interrupt, not both." }))
: Effect.void;
const runHandoff = Effect.fn("cli.thread.runHandoff")(function* (
tools: ThreadTools,
threadId: ThreadId,
flags: HandoffFlags,
) {
yield* validateHandoffFlags(flags);
const choose = flags.interrupt
? Effect.succeed("interrupt" as const)
: flags.wait
? Effect.succeed("wait" as const)
: Effect.gen(function* () {
if (flags.json || !process.stdin.isTTY || !process.stdout.isTTY)
return yield* new ThreadUsageError({
detail:
"The source CLI is still running. Use --wait to wait for it and its agents to exit, or --interrupt to stop them and resume the conversation in T3.",
});
const interrupt = yield* Prompt.run(
Prompt.Confirm({
message:
"The source CLI is still running. Interrupt it and its agents, transfer the conversation, and resume work in T3? Choose No to wait for them to exit.",
initial: false,
}),
).pipe(
Effect.mapError(
() =>
new ThreadUsageError({
detail: "Handoff cancelled; the source was not interrupted.",
}),
),
);
return interrupt ? ("interrupt" as const) : ("wait" as const);
});
const result = yield* handoffThread(
tools,
threadId,
choose.pipe(
Effect.tap((policy) =>
flags.json
? Effect.void
: Console.error(
policy === "wait"
? "Waiting for the source CLI and its agents to exit. Ctrl+C cancels this wait."
: "Stopping the source and its agents, then importing history and resuming in T3.",
),
),
),
);
return result;
});

const threadAttachCommand = Command.make("attach", {
...projectLocationFlags,
...handoffFlags,
session: Argument.String("session"),
provider: Flag.String("provider").pipe(Flag.optional),
project: Flag.String("project").pipe(Flag.withDefault(".")),
json: jsonFlag,
}).pipe(
Command.withDescription(
"Move a local Claude conversation into T3, waiting or explicitly interrupting its source CLI.",
),
Command.withHandler((flags) =>
withThreadTools(
flags,
Effect.fn("cli.thread.attach")(function* (tools) {
yield* validateHandoffFlags(flags);
const project = yield* resolveProject(tools, flags.project);
const attached = yield* tools.call(
"t3_thread_attach",
{
action: "attach",
projectId: project.id,
session: flags.session,
...Option.match(flags.provider, {
onNone: () => ({}),
onSome: (providerInstanceId) => ({ providerInstanceId }),
}),
},
ClaudeSessionAttachmentResult,
);
const result = yield* runHandoff(tools, attached.threadId, flags);
yield* flags.json
? printJson(result)
: Console.log(`Handoff complete. Continue in ${result.link ?? result.threadId}.`);
}),
).pipe(reportThreadErrors),
),
);
export const handoffThread = Effect.fn("cli.thread.handoffThread")(function* <R>(
tools: ThreadTools,
threadId: ThreadId,
choose: Effect.Effect<"wait" | "interrupt", ThreadUsageError, R>,
wait = Effect.sleep("1 second"),
) {
let policy: "wait" | "interrupt" | undefined;
while (true) {
const result = yield* tools.call(
"t3_thread_attach",
{ action: "handoff", threadId, ...(policy === "interrupt" ? { interrupt: true } : {}) },
ClaudeSessionAttachmentResult,
);
if (!result.blocked) return result;
if (policy === "interrupt") {
return yield* new ThreadUsageError({
detail: "The source did not stop; handoff was not completed.",
});
}
policy ??= yield* choose;
if (policy === "wait") yield* wait;
}
});

const threadHandoffCommand = Command.make("handoff", {
...projectLocationFlags,
thread: threadArgument,
json: jsonFlag,
...handoffFlags,
}).pipe(
Command.withDescription(
"Continue an attached Claude conversation through T3 after Claude and its agents have exited.",
),
Command.withHandler((flags) =>
withThreadTools(
flags,
Effect.fn("cli.thread.handoff")(function* (tools) {
const threadId = yield* decodeThreadId(flags.thread);
const result = yield* runHandoff(tools, threadId, flags);
yield* flags.json
? printJson(result)
: Console.log(`Handoff complete. Continue in ${result.link ?? result.threadId}.`);
}),
).pipe(reportThreadErrors),
),
);

export const threadCommand = Command.make("thread").pipe(
Command.withDescription("Start, message, and read threads on the running server."),
Command.withSubcommands([
threadAttachCommand,
threadHandoffCommand,
threadListCommand,
threadStartCommand,
threadSendCommand,
Expand Down
44 changes: 44 additions & 0 deletions apps/server/src/mcp/toolkits/project/handlers.test.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import * as ClaudeAttachment from "../../../project/ClaudeSessionAttachment.ts";
import * as ProviderRegistry from "../../../provider/ProviderRegistry.ts";
import { expect, it } from "@effect/vitest";
import {
Expand Down Expand Up @@ -664,3 +665,46 @@ it.effect("deleting a project through MCP also discards its clone record", () =>
expect(discarded).toEqual([harness.projectId]);
}),
);

it.effect("captures the attachment service and links only after handoff", () =>
Effect.gen(function* () {
const { projectId, dependencies } = clientLaunchHarness({
runtimeModeCeiling: "full-access",
launched: [],
});
const threadId = ThreadId.make("attached-thread");
const requests: string[] = [];
const attachment = Layer.mock(ClaudeAttachment.ClaudeSessionAttachment)({
change: (input) =>
Effect.sync(() => {
requests.push(input.action);
return { threadId, attached: input.action === "attach", readOnly: false };
}),
});
const toolkit = yield* ProjectToolkit.pipe(
Effect.provide(
McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe(
Layer.provide(Layer.merge(dependencies, attachment)),
),
),
);
const result = yield* toolkit
.handle("t3_thread_attach", { action: "attach", projectId, session: "repair-session" })
.pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies));
expect(result.at(-1)?.result).not.toHaveProperty("link");
const completed = yield* toolkit
.handle("t3_thread_attach", { action: "handoff", threadId })
.pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies));
expect(completed.at(-1)?.result).toMatchObject({
threadId,
attached: false,
link: expect.any(String),
});
expect(requests).toEqual(["attach", "handoff"]);
expect(result.at(-1)?.result).toMatchObject({
threadId,
attached: true,
readOnly: false,
});
}),
);
31 changes: 31 additions & 0 deletions apps/server/src/mcp/toolkits/project/handlers.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import * as ClaudeAttachment from "../../../project/ClaudeSessionAttachment.ts";
import {
ClaudeSessionAttachmentInput,
CommandId,
MessageId,
ThreadId,
Expand Down Expand Up @@ -81,7 +83,36 @@ const access = Effect.gen(function* () {
yield* readCaller();
return yield* Project.ProjectService;
});
const decodeAttachmentInput = Schema.decodeUnknownEffect(ClaudeSessionAttachmentInput);

export const layer = McpToolAccess.toLayer(ProjectToolkit, {
t3_thread_attach: McpToolAccess.writesEnvironment((input) =>
Effect.gen(function* () {
const service = yield* Effect.serviceOption(ClaudeAttachment.ClaudeSessionAttachment);
if (Option.isNone(service)) return yield* unavailable();
const decoded = yield* decodeAttachmentInput(input).pipe(
Effect.mapError(
() =>
new OrchestratorMcpFailure({
code: "invalid_request",
message: "Attach requires projectId and session; handoff requires threadId.",
}),
),
);
const result = yield* service.value
.change(decoded)
.pipe(
Effect.mapError(
(error) =>
new OrchestratorMcpFailure({ code: "invalid_request", message: error.message }),
),
);
const scope = yield* McpInvocationContext.McpInvocationContext;
return result.attached
? result
: { ...result, link: McpInvocationContext.threadLinkFor(scope, result.threadId) };
}),
),
t3_thread_launch: McpToolAccess.startsThreads(
(input) => input,
(input, { runtimeMode, interactionMode, linkOrigin }) =>
Expand Down
19 changes: 19 additions & 0 deletions apps/server/src/mcp/toolkits/project/tools.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { McpAttachmentInput } from "../attachment/input.ts";
import {
ClaudeSessionAttachmentResult,
NonNegativeInt,
ModelSelection,
TrimmedNonEmptyString,
Expand All @@ -14,6 +15,7 @@ import {
ProjectCreatePayload,
ProjectUpdatePayload,
ProjectId,
ProviderInstanceId,
OrchestratorMcpClientRequestId,
OrchestratorMcpEnvironmentTarget,
OrchestratorMcpFailure,
Expand Down Expand Up @@ -178,7 +180,24 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", {
.annotate(Tool.Destructive, true)
.annotate(Tool.OpenWorld, true);

const ThreadAttachTool = Tool.make("t3_thread_attach", {
...shared,
description:
"Import a local Claude conversation and its subagent history into normal T3 controls. Attach prepares the transfer using an exact session name or ID; handoff completes it after the source CLI exits. A blocked result leaves the source running. Only with explicit user authorization, set interrupt to stop the source and its agents and resume work through T3.",
parameters: Schema.Struct({
Comment on lines +185 to +187

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Run the required MCP tool-use eval for the new descriptor

This introduces a model-facing MCP description and schema for a destructive, two-step transfer API, but the commit's verification notes explicitly state that the MCP model-evaluation case has not been run. The handler and CLI unit tests do not validate a model's tool choices. AGENTS.md requires running the named evals when MCP descriptions or schemas change; run the MCP tool-use evaluation, including external-claude-explicit-handoff, before shipping this descriptor.

AGENTS.md reference: AGENTS.md:L146-L146

Useful? React with 👍 / 👎.

action: Schema.Literals(["attach", "handoff"]),
projectId: Schema.optional(ProjectId),
session: Schema.optional(TrimmedNonEmptyString),
providerInstanceId: Schema.optional(ProviderInstanceId),
threadId: Schema.optional(ThreadId),
interrupt: Schema.optional(Schema.Boolean),
}),
success: ClaudeSessionAttachmentResult,
dependencies: shared.dependencies,
}).annotate(Tool.Destructive, true);

export const ProjectToolkit = Toolkit.make(
ThreadAttachTool,
ThreadLaunchTool,
ProjectListTool,
ProjectReadTool,
Expand Down
Loading
Loading