diff --git a/apps/server/src/application/bootstrap/server-bootstrap.ts b/apps/server/src/application/bootstrap/server-bootstrap.ts index 2ef812c3a..0e054e668 100644 --- a/apps/server/src/application/bootstrap/server-bootstrap.ts +++ b/apps/server/src/application/bootstrap/server-bootstrap.ts @@ -93,6 +93,7 @@ import { RepositoryGitMutationLock } from "../../features/projects/git/repositor import { SNAPSHOT_STORE_ID, type StoreId } from "../../features/projects/diffs/snapshots/snapshot-store-identity.js"; import { AgentRuntimeCommandPort } from "../../features/agents/orchestration/agent-turn-command-port.js"; import { DelegatedTaskService } from "../../features/thread-control/delegation/delegated-task-service.js"; +import { ThreadLaunchService } from "../../features/thread-control/launch/thread-launch-service.js"; import { ProviderTurnEventApplication } from "../../features/agents/turns/provider-turn-event-application.js"; import { TURN_FINALIZER, TurnFinalizer } from "../../features/agents/turns/turn-finalizer.js"; import { @@ -337,6 +338,7 @@ const turnRecoveryService = container.resolve(TurnRecoveryService); const turnResumeScheduler = container.resolve(TurnResumeScheduler); const threadControlService = container.resolve(ThreadControlService); const delegatedTaskService = container.resolve(DelegatedTaskService); +const threadLaunchService = container.resolve(ThreadLaunchService); const threadStartupService = container.resolve(ThreadStartupService); const stopObservingDelegatedRosters = subagentRosterService.bindDelegatedTasks(delegatedTaskService); const stopObservingDelegatedStartups = threadStartupService.subscribe((startup) => delegatedTaskService.taskChanged(startup.startupId)); @@ -361,6 +363,7 @@ const mcodeMcpServer = new McodeMcpServer({ threadStop: threadControlService.threadStop.bind(threadControlService), threadWait: threadControlService.threadWait.bind(threadControlService), delegate: delegatedTaskService.delegate.bind(delegatedTaskService), + launch: threadLaunchService.launch.bind(threadLaunchService), status: delegatedTaskService.status.bind(delegatedTaskService), cancel: delegatedTaskService.cancel.bind(delegatedTaskService), }, diff --git a/apps/server/src/features/agents/turns/turn-admission-dispatch-coordinator.ts b/apps/server/src/features/agents/turns/turn-admission-dispatch-coordinator.ts index bf71e7d27..86278bc6d 100644 --- a/apps/server/src/features/agents/turns/turn-admission-dispatch-coordinator.ts +++ b/apps/server/src/features/agents/turns/turn-admission-dispatch-coordinator.ts @@ -71,6 +71,14 @@ export type { WorkspaceEnvironmentQueuedTurnSubmission }; /** Injection token for the complete first-turn admission coordinator. */ export const TURN_ADMISSION_DISPATCH_COORDINATOR = "TurnAdmissionDispatchCoordinator"; +/** An existing-worktree choice rejected before a thread is persisted. */ +export class ExistingWorktreePlacementError extends Error { + constructor(message: string) { + super(message); + this.name = "ExistingWorktreePlacementError"; + } +} + /** Why a turn's runtime slot was released; absent for a turn that ran to its terminal state. */ export type RuntimeReleaseCause = "admission-failed"; @@ -406,7 +414,7 @@ export class TurnAdmissionDispatchCoordinator { if (!workspace) throw new Error(`Workspace not found: ${params.workspaceId}`); const known = await this.worktrees.listWorktrees(params.workspaceId); const matched = known.find((worktree) => this.sameWorktreePath(worktree.path, params.existingWorktreePath)); - if (!matched) throw new Error("Path is not a recognized worktree"); + if (!matched) throw new ExistingWorktreePlacementError("Path is not a recognized worktree"); const detached = matched.branch === "(detached)"; const branch = detached ? params.baseBranch : matched.branch; const baseBranch = this.requireAttachedBranch(branch, detached); @@ -423,8 +431,8 @@ export class TurnAdmissionDispatchCoordinator { } private requireAttachedBranch(branch: string | undefined, detached: boolean): string { - if (!branch) throw new Error("Base branch is required when attaching a detached worktree"); - if (detached && branch === "HEAD") throw new Error("Base branch cannot be HEAD when attaching a detached worktree"); + if (!branch) throw new ExistingWorktreePlacementError("Base branch is required when attaching a detached worktree"); + if (detached && branch === "HEAD") throw new ExistingWorktreePlacementError("Base branch cannot be HEAD when attaching a detached worktree"); validateBranchName(branch); return branch; } diff --git a/apps/server/src/features/mcode-mcp/__tests__/mcode-mcp-server.test.ts b/apps/server/src/features/mcode-mcp/__tests__/mcode-mcp-server.test.ts index 0cac08bd7..71f6ceb4e 100644 --- a/apps/server/src/features/mcode-mcp/__tests__/mcode-mcp-server.test.ts +++ b/apps/server/src/features/mcode-mcp/__tests__/mcode-mcp-server.test.ts @@ -81,7 +81,10 @@ function createThreadService() { return { workspaceSearch: vi.fn(), worktreeList: vi.fn(), - threadCreateBatch: vi.fn(), + delegate: vi.fn(), + launch: vi.fn(), + status: vi.fn(), + cancel: vi.fn(), threadTargetList: vi.fn(), threadSearch: vi.fn(), threadGet: vi.fn(), diff --git a/apps/server/src/features/mcode-mcp/__tests__/thread-control-toolkit.test.ts b/apps/server/src/features/mcode-mcp/__tests__/thread-control-toolkit.test.ts index 587b998ad..7be4d28a1 100644 --- a/apps/server/src/features/mcode-mcp/__tests__/thread-control-toolkit.test.ts +++ b/apps/server/src/features/mcode-mcp/__tests__/thread-control-toolkit.test.ts @@ -46,6 +46,7 @@ function createService() { workspaceSearch: vi.fn(), worktreeList: vi.fn(), delegate: vi.fn(), + launch: vi.fn(), status: vi.fn(), cancel: vi.fn(), threadTargetList: vi.fn(), @@ -126,6 +127,7 @@ describe("thread-control toolkit", () => { "workspace_search", "worktree_list", "delegate_task", + "thread_launch", "task_status", "task_cancel", "thread_target_list", @@ -143,6 +145,18 @@ describe("thread-control toolkit", () => { expect(setup({ threadControl: false }).bound).toBeNull(); }); + it("lists thread_launch for an internal lease and marks a rejected launch as an error", async () => { + const { bound, call, service } = setup(); + expect(bound?.tools.filter((tool) => tool.listed).map((tool) => tool.name)).toContain("thread_launch"); + service.launch.mockResolvedValueOnce({ status: "rejected", error: { + code: "invalid_placement", message: "Unknown worktree", retryable: false, + } }); + expect(await call("thread_launch", { title: "Independent work", prompt: "Implement it" })).toMatchObject({ + isError: true, + structuredContent: { status: "rejected", error: { code: "invalid_placement", retryable: false } }, + }); + }); + it.each(["none", "ineligible"] as const)("refuses every call when the turn lease is %s", async (lease) => { const { call, service } = setup({ lease }); diff --git a/apps/server/src/features/mcode-mcp/toolkits/thread-control-toolkit.ts b/apps/server/src/features/mcode-mcp/toolkits/thread-control-toolkit.ts index 916f9833d..b3ac2e030 100644 --- a/apps/server/src/features/mcode-mcp/toolkits/thread-control-toolkit.ts +++ b/apps/server/src/features/mcode-mcp/toolkits/thread-control-toolkit.ts @@ -1,6 +1,8 @@ import { DelegateTaskInputSchema, DelegateTaskResultSchema, + ThreadLaunchInputSchema, + ThreadLaunchResultSchema, TaskStatusInputSchema, TaskStatusResultSchema, TaskCancelInputSchema, @@ -34,6 +36,7 @@ import { toJsonSchemaCompat } from "@modelcontextprotocol/sdk/server/zod-json-sc import { ErrorCode, McpError, ToolSchema, type CallToolResult } from "@modelcontextprotocol/sdk/types.js"; import type { ThreadControlService } from "../../thread-control/index.js"; import type { DelegatedTaskService } from "../../thread-control/delegation/delegated-task-service.js"; +import type { ThreadLaunchService } from "../../thread-control/launch/thread-launch-service.js"; import type { McodeSession, McodeSessionAuthority } from "../mcode-session-authority.js"; import { defineToolkit, @@ -53,7 +56,7 @@ export type ThreadControlToolService = Pick< | "threadSend" | "threadStop" | "threadWait" -> & Pick; +> & Pick & Pick; /** Dependencies for the thread-control toolkit. */ export interface ThreadControlToolkitOptions { @@ -191,6 +194,12 @@ function createTaskTools(context: ToolContext): McodeTool[] { input: DelegateTaskInputSchema(), result: DelegateTaskResultSchema(), run: (service, authority, input) => service.delegate(authority, input), }), + serviceTool(context, { + name: "thread_launch", + description: 'Launch an ordinary top-level Mcode thread for the user, in its own workspace. Use it only when the user asks for a separate thread or conversation; never just because they said sub-agent. The new thread runs on its own and you are not notified when it finishes; use delegate_task when you need the result back. workspace: {type:"new_worktree", baseRef, startFromOrigin?} creates a worktree from a local branch, or from the fetched origin branch; {type:"existing_worktree", worktreePath, baseBranch?} uses one of this project\'s worktrees (baseBranch is required when it is detached); omitted or {type:"root"} uses the project root, NOT your current worktree. Project setup finishes before the first message runs. Provider and model default to yours. Permission and interaction modes default to yours and cannot be broader. A supervised caller cannot use new_worktree. Returns the new threadId; link it for the user.', + input: ThreadLaunchInputSchema(), result: ThreadLaunchResultSchema(), + run: (service, authority, input) => service.launch(authority, input), + }), serviceTool(context, { name: "task_status", description: "Read the current state and bounded result of a task delegated by this Coordinator. Status follows the original task turn and includes its threadId when available; later user turns do not replace its result. Finished/interrupted can become working after recovery. Check later after ending the turn or doing other work; do not poll in a loop.", diff --git a/apps/server/src/features/projects/git/git-repository-service.ts b/apps/server/src/features/projects/git/git-repository-service.ts index 853827c40..e6f5adcae 100644 --- a/apps/server/src/features/projects/git/git-repository-service.ts +++ b/apps/server/src/features/projects/git/git-repository-service.ts @@ -281,6 +281,21 @@ export class GitRepositoryService { } } + /** Resolve only a local branch, returning null when its ref does not exist. */ + async resolveLocalBranchCommit(repoPath: string, branch: string): Promise { + validateBranchName(branch); + try { + const { stdout } = await this.gitExecutor.exec([ + "-C", repoPath, "rev-parse", "--verify", "--quiet", `refs/heads/${branch}^{commit}`, + ]); + return stdout.trim(); + } catch (error) { + const failure = gitCommandFailureSchema.safeParse(error); + if (failure.success && failure.data.code === 1) return null; + throw error; + } + } + /** Fetch an origin branch at an arbitrary repository path. */ async fetchBranchAt(repoPath: string, branch: string, prNumber?: number): Promise { validateBranchName(branch); diff --git a/apps/server/src/features/thread-control/composition/register-thread-control.ts b/apps/server/src/features/thread-control/composition/register-thread-control.ts index 3b6c6b232..b9f077500 100644 --- a/apps/server/src/features/thread-control/composition/register-thread-control.ts +++ b/apps/server/src/features/thread-control/composition/register-thread-control.ts @@ -14,11 +14,14 @@ import { ThreadControlApprovalRepo } from "../authority/persistence/thread-contr import { ThreadControlAuditRepo } from "../authority/persistence/thread-control-audit-repo.js"; import { DelegatedTaskRepo } from "../delegation/delegated-task-repo.js"; import { DelegatedTaskService } from "../delegation/delegated-task-service.js"; +import { ThreadLaunchService } from "../launch/thread-launch-service.js"; import { ThreadRepo } from "../persistence/thread-repo.js"; import { ThreadStartupService } from "../../thread-startup/thread-startup-service.js"; import { ThreadStartupFetchDecisions } from "../../thread-startup/thread-startup-fetch-decisions.js"; import { WorkspaceEnvironmentService } from "../../projects/environment/workspace-environment-service.js"; import { GitWorktreeService } from "../../projects/git/git-worktree-service.js"; +import { GitRepositoryService } from "../../projects/git/git-repository-service.js"; +import { WorkspaceRepo } from "../../projects/persistence/workspace-repo.js"; import { AgentService } from "../../agents/orchestration/agent-service.js"; import { AgentTurnCommandPort } from "../../agents/orchestration/agent-turn-command-port.js"; import { CanonicalAgentBoundary } from "../../agents/canonical/canonical-agent-boundary.js"; @@ -28,6 +31,12 @@ import type { CancelThreadStartupDeps } from "../../thread-startup/cancel-thread /** Register thread-control mutation, approval, authority, and lifecycle services. */ export function registerThreadControlServices(container: DependencyContainer): void { + container.register(ThreadLaunchService, { useFactory: instanceCachingFactory((c) => new ThreadLaunchService({ + threads: c.resolve(ThreadRepo), workspaces: c.resolve(WorkspaceRepo), + targets: c.resolve(delay(() => DelegationTargetResolver)), + repository: c.resolve(delay(() => GitRepositoryService)), + commands: c.resolve(delay(() => AgentTurnCommandPort)), + })) }); container.register(DelegatedTaskRepo, { useClass: DelegatedTaskRepo }, { lifecycle: Lifecycle.Singleton }); container.register("CancelThreadStartupDeps", { useFactory: instanceCachingFactory((c) => ({ threadStartupService: c.resolve(ThreadStartupService), diff --git a/apps/server/src/features/thread-control/launch/thread-launch-service.test.ts b/apps/server/src/features/thread-control/launch/thread-launch-service.test.ts new file mode 100644 index 000000000..76c05ab15 --- /dev/null +++ b/apps/server/src/features/thread-control/launch/thread-launch-service.test.ts @@ -0,0 +1,225 @@ +import "reflect-metadata"; +import * as NodeCrypto from "node:crypto"; +import * as NodeEvents from "node:events"; +import * as NodePath from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { IAgentProvider, IProviderRegistry, ThreadLaunchResult } from "@mcode/contracts"; +import type { InternalThreadControlAuthority } from "@mcode/thread-orchestration"; +import { createOriginFixture, git } from "../../projects/git/__tests__/start-from-origin-fixture.js"; +import { ThreadRepo } from "../persistence/thread-repo.js"; +import { ThreadService } from "../lifecycle/thread-service.js"; +import { MessageRepo } from "../../agents/conversation/persistence/message-repo.js"; +import { ThreadStartupService } from "../../thread-startup/thread-startup-service.js"; +import { ThreadStartupRepo } from "../../thread-startup/persistence/thread-startup-repo.js"; +import { ThreadStartupFetchDecisions } from "../../thread-startup/thread-startup-fetch-decisions.js"; +import { SettingsService } from "../../settings/settings-service.js"; +import { ModelCacheService } from "../../providers/models/model-cache-service.js"; +import { ModelCacheRepo } from "../../providers/models/persistence/model-cache-repo.js"; +import { ModelSightingsRepo } from "../../providers/status/model-sightings-repo.js"; +import { DelegationTargetResolver } from "../../agents/collaboration/delegation-target-resolver.js"; +import { CanonicalAgentBoundary } from "../../agents/canonical/canonical-agent-boundary.js"; +import { CanonicalAgentWriterClient } from "../../agents/canonical/canonical-agent-writer-client.js"; +import { notifyCommittedCanonicalEvents } from "../../agents/canonical/committed-canonical-events.js"; +import { AttachmentService } from "../../attachments/storage/attachment-service.js"; +import { AgentRuntimeCommandPort, AgentTurnCommandPort } from "../../agents/orchestration/agent-turn-command-port.js"; +import { PlanTurnService } from "../../agents/planning/plan-turn-service.js"; +import { PlanQuestionService } from "../../agents/planning/plan-question-service.js"; +import { PlanRepo } from "../../agents/planning/persistence/plan-repo.js"; +import { PlanQuestionAnswersRepo } from "../../agents/planning/persistence/plan-question-answers-repo.js"; +import { startupEnvironmentFixture } from "../../projects/environment/testing/startup-environment-fixture.js"; +import { TurnAdmissionDispatchCoordinator, type TurnRuntimeAdmissionAuthority } from "../../agents/turns/turn-admission-dispatch-coordinator.js"; +import { ThreadCreationCoordinator } from "../../agents/turns/thread-creation-coordinator.js"; +import { TurnRevertStore } from "../../agents/turns/persistence/turn-revert-store.js"; +import { ProviderAvailabilityService } from "../../providers/availability/provider-availability-service.js"; +import { GoalLifecycleService } from "../../agents/goals/goal-lifecycle-service.js"; +import { FileService } from "../../projects/files/file-service.js"; +import { ProjectWorktreeService } from "../../projects/worktrees/project-worktree-service.js"; +import { SandboxWorktreeCleanupPolicy } from "../../projects/worktrees/sandbox-worktree-cleanup-policy.js"; +import { WorktreeSafetyService } from "../../projects/git/worktree-safety-service.js"; +import { HandoffStorage } from "../../handoff/index.js"; +import type { TerminalCommandCompletion } from "../../terminal/commands/terminal-command-service.js"; +import { ThreadLaunchService } from "./thread-launch-service.js"; + +vi.mock("../../../application/transport/push.js", () => ({ broadcast: vi.fn() })); +const disposables: Array<() => Promise | void> = []; +afterEach(async () => { + for (const dispose of disposables.splice(0)) await dispose(); + vi.restoreAllMocks(); +}); + +async function harness(coordinatorMode: "direct" | "worktree" = "direct", setup = false) { + const fixture = await createOriginFixture(); + const { runtime, workspace, workspaces, repository, worktrees, executor, host } = fixture; + const threads = new ThreadRepo(runtime.reader, runtime.writer); + const messages = new MessageRepo(runtime.reader, runtime.writer); + const startups = new ThreadStartupService(new ThreadStartupRepo(runtime.reader, runtime.writer), runtime.writer); + const settings = new SettingsService(); + disposables.push(() => settings.dispose()); + const sendTurn = vi.fn(async () => {}); + const provider: IAgentProvider = Object.assign(new NodeEvents.EventEmitter(), { + id: "codex" as const, descriptor: { id: "codex" as const, capabilities: [] }, subagentReporting: null, + supportsCompletion: false, sessionForkOnResume: "unsupported" as const, maxInputCharactersPerTurn: 100_000, + forker: { fork: async () => { throw new Error("Launch starts a new session"); } }, + sendTurn, stopSession: () => {}, shutdown: () => {}, + listModels: async () => [{ id: "gpt-test", name: "Test" }], + }); + const providers: IProviderRegistry = { resolve: () => provider, resolveAll: () => [provider], shutdown: async () => {} }; + const models = new ModelCacheService(new ModelCacheRepo(runtime.reader, runtime.writer), providers, + new ModelSightingsRepo(runtime.reader, runtime.writer)); + disposables.push(() => models.close()); + const targets = new DelegationTargetResolver(providers, models, settings); + const canonical = new CanonicalAgentBoundary(runtime.reader, runtime.writer, new CanonicalAgentWriterClient(runtime.writer), notifyCommittedCanonicalEvents); + const attachments = new AttachmentService(); + const commands = new AgentRuntimeCommandPort(); + const plans = new PlanTurnService(threads, providers, new PlanQuestionService(messages, new PlanQuestionAnswersRepo(runtime.reader, runtime.writer)), + new PlanRepo(runtime.reader, runtime.writer), new AgentTurnCommandPort(commands)); + let finishSetup: (result: TerminalCommandCompletion) => void = () => { throw new Error("Setup not prepared"); }; + const completion = new Promise((resolve) => { finishSetup = resolve; }); + const environment = startupEnvironmentFixture({ + mcodeDir: fixture.root, database: runtime.reader, databaseWriter: runtime.writer, threads, workspaces, + threadStartups: startups, platform: "windows", + terminalCommands: { prepare: async () => ({ kind: "ready", command: { + snapshot: { checkoutPath: fixture.repoPath, terminal: { executable: "sh", arguments: [] } }, + start: () => completion, close: async () => ({ kind: "contained" }), waitForRelease: () => new Promise(() => {}), + } }) }, + }); + disposables.push(() => environment.dispose()); + if (setup) await environment.save({ workspaceId: workspace.id, sourceRevision: null, + document: { version: "0.1.0", actions: [{ id: "setup", name: "Setup", command: { default: "setup" }, runOnStartup: true }] } }); + const admission = new TurnAdmissionDispatchCoordinator(threads, workspaces, messages, new TurnRevertStore(runtime.reader), worktrees, + attachments, providers, new ProviderAvailabilityService(settings, providers), canonical, settings, plans, + new GoalLifecycleService(threads, providers, runtime.writer, commands, canonical), environment, + new FileService(workspaces, threads, worktrees, executor, host), host.platform, + { start: async ({ parentTurn }) => { await canonical.startParentTurn(parentTurn); } }); + const projects = new ProjectWorktreeService(threads, workspaces, runtime.writer, worktrees, + new SandboxWorktreeCleanupPolicy(new WorktreeSafetyService(executor, host), repository, host)); + const threadService = new ThreadService(threads, projects, attachments, new HandoffStorage(), { + teardownThread: async () => {}, deletePersistentData: async (_ids, remove) => remove(), + }); + const creation = new ThreadCreationCoordinator(threads, () => threadService, admission, repository, workspaces, + new ThreadStartupFetchDecisions(), undefined, undefined, () => startups); + const authority: TurnRuntimeAdmissionAuthority = { + reserve: (command) => ({ threadId: command.threadId, turnExecutionId: NodeCrypto.randomUUID(), mutationReservationToken: NodeCrypto.randomUUID(), generation: 1 }), + activate: () => {}, abort: async () => {}, release: () => {}, owns: () => true, + }; + const dispatch = async (command: Parameters[0]) => { + const admitted = await admission.admit(command, authority); + if (admitted.kind === "dispatch") await admitted.provider.sendTurn(admitted.request); + }; + environment.setAutomaticSetupDispatcher({ dispatch: async (submission) => { + await dispatch(admission.queuedCommand(submission)); + return { completion: Promise.resolve() }; + } }); + const service = new ThreadLaunchService({ threads, workspaces, targets, repository, + commands: { createAndSend: async (command, options) => { + const created = await creation.createInitialTurn(command, options); + if (created.kind === "dispatch") await dispatch(created.command); + return created.thread; + } }, + }); + const coordinator = coordinatorMode === "worktree" + ? await threadService.create(workspace.id, "Coordinator", "worktree", "main", { provider: "codex", branchless: true }) + : await threads.create(workspace.id, "Coordinator", "direct", "main", false, "codex"); + await threads.updateModel(coordinator.id, "gpt-test"); + const source: InternalThreadControlAuthority = { type: "internal", userId: "local-user", sourceThreadId: coordinator.id, + sourceTurnId: NodeCrypto.randomUUID(), sourceToolCallId: "launch-call", sourceProviderId: "codex", permissionMode: "full" }; + return { ...fixture, threads, service, threadService, environment, canonical, messages, coordinator, source, sendTurn, finishSetup }; +} + +function requireLaunch(result: ThreadLaunchResult) { + if (result.status === "rejected") throw new Error(`Launch rejected: ${result.error.code}: ${result.error.message}`); + return result; +} + +const input = { title: "Independent work", prompt: "Implement the requested work" }; + +describe("thread launch", { timeout: 30_000 }, () => { + it("starts at the local branch and dispatches the first turn only after setup completes", async () => { + const f = await harness("direct", true); + const launched = requireLaunch(await f.service.launch(f.source, { ...input, + workspace: { type: "new_worktree", baseRef: "main", startFromOrigin: false } })); + const thread = f.threads.findById(launched.threadId); + expect(git(thread?.worktree_path ?? "", "rev-parse", "HEAD")).toBe(f.localSha); + await vi.waitFor(() => expect(f.environment.getAutomaticSetup({ threadId: launched.threadId }).attempt?.state).toBe("running")); + expect(f.canonical.loadLatestTurn(launched.threadId)).toBeNull(); + expect(f.sendTurn).not.toHaveBeenCalled(); + f.finishSetup({ kind: "exited", exitCode: 0, output: "Ready", outputTruncated: false }); + await vi.waitFor(() => expect(f.canonical.loadLatestTurn(launched.threadId)?.status).toBe("Running")); + await vi.waitFor(() => expect(f.sendTurn).toHaveBeenCalledTimes(1)); + expect(f.messages.listByThreadForThreadControl(launched.threadId, 10).messages.find((message) => message.role === "user")) + .toMatchObject({ content: input.prompt, originType: "thread", sourceThreadId: f.coordinator.id, + sourceTurnId: f.source.sourceTurnId, sourceProviderId: "codex" }); + }); + + it("starts at fetched origin instead of the differing local branch", async () => { + const f = await harness(); + const launched = requireLaunch(await f.service.launch(f.source, { ...input, + workspace: { type: "new_worktree", baseRef: "main", startFromOrigin: true } })); + const thread = f.threads.findById(launched.threadId); + expect(git(thread?.worktree_path ?? "", "rev-parse", "HEAD")).toBe(f.originSha); + expect(git(thread?.worktree_path ?? "", "rev-parse", "HEAD")).not.toBe(f.localSha); + }); + + it("rejects a branch missing on origin as a non-retryable placement without inserting a thread", async () => { + const f = await harness(); + expect(await f.service.launch(f.source, { ...input, + workspace: { type: "new_worktree", baseRef: "missing", startFromOrigin: true } })) + .toMatchObject({ status: "rejected", error: { code: "invalid_placement", retryable: false } }); + expect(f.runtime.database.query("SELECT id FROM threads").all()).toEqual([{ id: f.coordinator.id }]); + }); + + it("rejects an unknown local branch without inserting a thread", async () => { + const f = await harness(); + expect(await f.service.launch(f.source, { ...input, + workspace: { type: "new_worktree", baseRef: "missing", startFromOrigin: false } })) + .toMatchObject({ status: "rejected", error: { code: "invalid_placement", retryable: false } }); + expect(f.runtime.database.query("SELECT id FROM threads").all()).toEqual([{ id: f.coordinator.id }]); + }); + + it("attaches a registered worktree", async () => { + const f = await harness("worktree"); + const path = f.coordinator.worktree_path; + if (!path) throw new Error("Expected coordinator worktree"); + const launched = requireLaunch(await f.service.launch(f.source, { ...input, + workspace: { type: "existing_worktree", worktreePath: path, baseBranch: "main" } })); + expect(NodePath.resolve(f.threads.findById(launched.threadId)?.worktree_path ?? "")).toBe(NodePath.resolve(path)); + }); + + it.each(["arbitrary directory", "repository root"])("rejects %s as an existing worktree without inserting a thread", async (placement) => { + const f = await harness(); + const path = placement === "repository root" ? f.repoPath : f.root; + expect(await f.service.launch(f.source, { ...input, workspace: { type: "existing_worktree", worktreePath: path } })) + .toMatchObject({ status: "rejected", error: { code: "invalid_placement", retryable: false } }); + expect(f.runtime.database.query("SELECT id FROM threads").all()).toEqual([{ id: f.coordinator.id }]); + }); + + it("rejects broader modes and inherits supervised and plan when omitted", async () => { + const f = await harness(); + const source = { ...f.source, permissionMode: "supervised" as const }; + await f.threads.updateSettings(f.coordinator.id, { permission_mode: "supervised", interaction_mode: "plan" }); + expect(await f.service.launch(source, { ...input, permissionMode: "full" })) + .toMatchObject({ status: "rejected", error: { code: "forbidden", retryable: false } }); + expect(await f.service.launch(source, { ...input, interactionMode: "build" })) + .toMatchObject({ status: "rejected", error: { code: "forbidden", retryable: false } }); + expect(await f.service.launch(source, { ...input, workspace: { type: "new_worktree", baseRef: "main" } })) + .toMatchObject({ status: "rejected", error: { code: "forbidden", retryable: false } }); + expect(f.threads.listByWorkspace(f.workspace.id).map((thread) => thread.id)).toEqual([f.coordinator.id]); + const launched = requireLaunch(await f.service.launch(source, input)); + expect(f.threads.findById(launched.threadId)).toMatchObject({ permission_mode: "supervised", interaction_mode: "plan" }); + }); + + it("defaults to the project root when the caller runs in a worktree", async () => { + const f = await harness("worktree"); + const launched = requireLaunch(await f.service.launch(f.source, input)); + expect(f.threads.findById(launched.threadId)).toMatchObject({ mode: "direct", worktree_path: null, branch: "main" }); + }); + + it("returns a visible ordinary thread without delegation lineage", async () => { + const f = await harness(); + const launched = requireLaunch(await f.service.launch(f.source, input)); + expect(f.threads.findById(launched.threadId)).toMatchObject({ + id: launched.threadId, delegation_coordinator_thread_id: null, parent_thread_id: null, + }); + expect(f.threads.listByWorkspace(f.workspace.id).map((thread) => thread.id)).toContain(launched.threadId); + }); +}); diff --git a/apps/server/src/features/thread-control/launch/thread-launch-service.ts b/apps/server/src/features/thread-control/launch/thread-launch-service.ts new file mode 100644 index 000000000..117469385 --- /dev/null +++ b/apps/server/src/features/thread-control/launch/thread-launch-service.ts @@ -0,0 +1,142 @@ +import * as NodeCrypto from "node:crypto"; +import { + ThreadLaunchInputSchema, ProviderIdSchema, + type ThreadLaunchInput, type ThreadLaunchResult, type ThreadLaunchWorkspace, type ThreadControlError, type Thread, +} from "@mcode/contracts"; +import type { InternalThreadControlAuthority } from "@mcode/thread-orchestration"; +import type { ThreadRepo } from "../persistence/thread-repo.js"; +import type { WorkspaceRepo } from "../../projects/persistence/workspace-repo.js"; +import type { DelegationTargetResolver, DelegationTargetResolution } from "../../agents/collaboration/delegation-target-resolver.js"; +import { OriginFetchError, type GitRepositoryService } from "../../projects/git/git-repository-service.js"; +import type { CreateAndSendCommand, CreateAndSendOptions } from "../../agents/turns/thread-creation-coordinator.js"; +import { ExistingWorktreePlacementError } from "../../agents/turns/turn-admission-dispatch-coordinator.js"; + +/** Existing owners used to prepare and start an ordinary thread. */ +export interface ThreadLaunchDeps { + threads: Pick; + workspaces: Pick; + targets: Pick; + repository: Pick; + commands: { + createAndSend(command: CreateAndSendCommand, options: CreateAndSendOptions): Promise>; + }; +} + +type LaunchRejection = Extract; +type LaunchModes = Required>; +interface LaunchPlacement { + fields: Pick; + worktreeStart?: CreateAndSendOptions["worktreeStart"]; +} + +/** Launch ordinary user-owned threads through the first-turn setup gate. */ +export class ThreadLaunchService { + constructor(private readonly deps: ThreadLaunchDeps) {} + + /** Prepare the requested workspace and start a thread within the caller's modes. */ + async launch(authority: InternalThreadControlAuthority, input: ThreadLaunchInput): Promise { + const parsed = ThreadLaunchInputSchema().parse(input); + const source = this.deps.threads.findById(authority.sourceThreadId); + if (!source || source.deleted_at) return rejection("not_found", "Source thread not found"); + const modes = resolveModes(authority, source, parsed); + if ("status" in modes) return modes; + if (authority.permissionMode === "supervised" && parsed.workspace.type === "new_worktree") { + return rejection("forbidden", "A supervised caller cannot create a worktree without the user's approval"); + } + const target = await this.resolveTarget(source, parsed); + if (target.status === "rejected") return target; + const placement = await this.placement(source.workspace_id, parsed.workspace); + if ("status" in placement) return placement; + try { + const thread = await this.deps.commands.createAndSend({ + workspaceId: source.workspace_id, + startupId: NodeCrypto.randomUUID(), + title: parsed.title, + content: parsed.prompt, + provider: target.providerId, + model: target.modelId, + ...modes, + ...placement.fields, + attachments: [], + }, { + messageId: NodeCrypto.randomUUID(), + turn: { + sourceTurnId: NodeCrypto.randomUUID(), + sourceThreadId: authority.sourceThreadId, + sourceProviderId: ProviderIdSchema.parse(authority.sourceProviderId), + originSourceTurnId: authority.sourceTurnId, + }, + worktreeStart: placement.worktreeStart, + }); + return { status: "launched", threadId: thread.id }; + } catch (error) { + if (error instanceof ExistingWorktreePlacementError) return rejection("invalid_placement", error.message); + throw error; + } + } + + private async resolveTarget(source: Thread, input: ThreadLaunchInput): Promise | LaunchRejection> { + const providerId = input.providerId ?? source.provider; + const modelId = input.modelId ?? (providerId === source.provider ? source.model : undefined); + if (!modelId) return rejection("invalid_model", "A model is required for the requested provider"); + const target = await this.deps.targets.resolve({ providerId, modelId }); + if (target.status === "resolved") return target; + if (target.status === "discovery_failed") return rejection("internal_error", "Provider target discovery failed", true); + return rejection(target.status === "invalid_provider" ? "invalid_provider" : "invalid_model", "The requested provider and model are unavailable"); + } + + private async placement(workspaceId: string, workspace: ThreadLaunchWorkspace): Promise { + if (workspace.type === "root") { + const branch = await this.deps.repository.getCurrentBranch(workspaceId); + if (!branch) return rejection("invalid_placement", "The project root has no current branch"); + return { fields: { mode: "direct", branch } }; + } + if (workspace.type === "existing_worktree") { + return { fields: { + mode: "worktree", + existingWorktreePath: workspace.worktreePath, + existingWorktreeBaseBranch: workspace.baseBranch, + } }; + } + const project = this.deps.workspaces.findById(workspaceId); + if (!project) return rejection("not_found", "Project not found"); + const commit = await this.resolveCommit(project.path, workspace); + if (typeof commit !== "string") return commit; + return { + fields: { mode: "worktree", branch: workspace.baseRef, existingWorktreeBaseBranch: workspace.baseRef, worktreeBranchMode: "branchless" }, + worktreeStart: { commit }, + }; + } + + private async resolveCommit(repoPath: string, workspace: Extract): Promise { + if (workspace.startFromOrigin) { + try { + return await this.deps.repository.fetchOriginTrackingBranch(repoPath, workspace.baseRef); + } catch (error) { + if (!(error instanceof OriginFetchError)) throw error; + const missing = error.code === "no_origin" || error.code === "branch_missing"; + return missing ? rejection("invalid_placement", error.message) : rejection("internal_error", error.message, true); + } + } + const commit = await this.deps.repository.resolveLocalBranchCommit(repoPath, workspace.baseRef); + return commit ?? rejection("invalid_placement", `No local branch ${workspace.baseRef}`); + } +} + +function resolveModes(authority: InternalThreadControlAuthority, source: Thread, input: ThreadLaunchInput): LaunchModes | LaunchRejection { + if (authority.permissionMode === "supervised" && input.permissionMode === "full") { + return rejection("forbidden", "The launched thread cannot have broader permissions than its caller"); + } + const interactionMode = source.interaction_mode ?? "build"; + if (interactionMode === "plan" && input.interactionMode === "build") { + return rejection("forbidden", "A plan thread cannot launch a build thread"); + } + return { + permissionMode: input.permissionMode ?? authority.permissionMode, + interactionMode: input.interactionMode ?? interactionMode, + }; +} + +function rejection(code: ThreadControlError["code"], message: string, retryable = false): LaunchRejection { + return { status: "rejected", error: { code, message, retryable } }; +} diff --git a/apps/web/src/components/chat/tool-renderers/constants.ts b/apps/web/src/components/chat/tool-renderers/constants.ts index 11d726072..a31c7f02b 100644 --- a/apps/web/src/components/chat/tool-renderers/constants.ts +++ b/apps/web/src/components/chat/tool-renderers/constants.ts @@ -31,6 +31,7 @@ export const TOOL_LABELS: Record = { workspace_search: "Searched projects", worktree_list: "Listed worktrees", thread_target_list: "Listed targets", + thread_launch: "Launched thread", thread_search: "Searched threads", thread_get: "Read thread", thread_send: "Sent to thread", @@ -56,6 +57,7 @@ export const TOOL_ICONS: Record = { delegate_task: StackedLayersIcon, task_status: ListChecks, task_cancel: CircleSlash, + thread_launch: StackedLayersIcon, }; export const DEFAULT_ICON: IconComponent = Wrench; @@ -131,6 +133,7 @@ export const TOOL_OUTCOME_VERBS: Record = { workspace_search: "Project search", worktree_list: "Worktree list", thread_target_list: "Target list", + thread_launch: "Thread launch", thread_search: "Thread search", thread_get: "Thread read", thread_send: "Thread send", @@ -154,6 +157,7 @@ export const TOOL_SUMMARY_VERBS: Record = { workspace_search: ["%d project search", "%d project searches"], worktree_list: ["Listed worktrees %d time", "Listed worktrees %d times"], thread_target_list: ["Listed targets %d time", "Listed targets %d times"], + thread_launch: ["Launched %d thread", "Launched %d threads"], thread_search: ["%d thread search", "%d thread searches"], thread_get: ["Read %d thread", "Read %d threads"], thread_send: ["Sent %d message", "Sent %d messages"], diff --git a/apps/web/src/features/conversation/narrative/__tests__/tool-row-overflow.test.tsx b/apps/web/src/features/conversation/narrative/__tests__/tool-row-overflow.test.tsx index 495863fc7..fec333b96 100644 --- a/apps/web/src/features/conversation/narrative/__tests__/tool-row-overflow.test.tsx +++ b/apps/web/src/features/conversation/narrative/__tests__/tool-row-overflow.test.tsx @@ -495,7 +495,7 @@ describe.each(MCODE_MCP_SPELLINGS)("Mcode MCP task tools named %s*", (prefix) => const MCODE_THREAD_TOOLS = [ "workspace_search", "worktree_list", "thread_target_list", "thread_search", - "thread_get", "thread_send", "thread_stop", "thread_wait", + "thread_get", "thread_send", "thread_stop", "thread_wait", "thread_launch", ] as const; function mcodeCall(toolName: string, overrides: Partial = {}): ToolCall { diff --git a/packages/contracts/src/__tests__/thread-control.test.ts b/packages/contracts/src/__tests__/thread-control.test.ts index ed8d93f5f..8d7a06167 100644 --- a/packages/contracts/src/__tests__/thread-control.test.ts +++ b/packages/contracts/src/__tests__/thread-control.test.ts @@ -17,8 +17,22 @@ import { WorktreeListResultSchema, ThreadTargetListInputSchema, ThreadTargetListResultSchema, + ThreadLaunchInputSchema, } from "../thread-control.js"; +describe("thread launch input", () => { + it("defaults an omitted workspace to the project root", () => { + expect(ThreadLaunchInputSchema().parse({ title: "Independent work", prompt: "Implement it" }).workspace) + .toEqual({ type: "root" }); + }); + + it("rejects baseRef on an existing worktree placement", () => { + expect(ThreadLaunchInputSchema().safeParse({ title: "Independent work", prompt: "Implement it", + workspace: { type: "existing_worktree", worktreePath: "/repo/worktree", baseRef: "main" }, + }).success).toBe(false); + }); +}); + describe("thread control discovery schemas", () => { it("accepts secret-free provider target discovery and validates defaults", () => { expect(ThreadTargetListInputSchema().parse({})).toEqual({}); diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 3c9c8a1b1..b11b7ca28 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -6,6 +6,9 @@ export * from "./models/task-list-tools.js"; export { DelegateTaskInputSchema, DelegateTaskResultSchema, + ThreadLaunchWorkspaceSchema, + ThreadLaunchInputSchema, + ThreadLaunchResultSchema, TaskStateSchema, TaskStatusInputSchema, TaskStatusResultSchema, @@ -13,6 +16,9 @@ export { TaskCancelResultSchema, type DelegateTaskInput, type DelegateTaskResult, + type ThreadLaunchWorkspace, + type ThreadLaunchInput, + type ThreadLaunchResult, type TaskState, type TaskStatusInput, type TaskStatusResult, diff --git a/packages/contracts/src/thread-control.ts b/packages/contracts/src/thread-control.ts index 3e720b16b..9ae701b76 100644 --- a/packages/contracts/src/thread-control.ts +++ b/packages/contracts/src/thread-control.ts @@ -154,6 +154,44 @@ export type DelegateTaskInput = z.infer z.object({ status: z.literal("rejected"), error: ThreadControlErrorSchema() }).strict()); +/** Workspace prepared before an ordinary thread's first turn. */ +export const ThreadLaunchWorkspaceSchema = lazySchema(() => z.discriminatedUnion("type", [ + z.object({ type: z.literal("root") }).strict(), + z.object({ + type: z.literal("new_worktree"), + baseRef: gitRef, + startFromOrigin: z.boolean().default(false), + }).strict(), + z.object({ + type: z.literal("existing_worktree"), + worktreePath: z.string().trim().min(1).max(4096), + baseBranch: gitRef.optional(), + }).strict(), +])); +/** Parsed workspace choice for an ordinary thread launch. */ +export type ThreadLaunchWorkspace = z.infer>; + +/** Launch an ordinary thread in the source thread's Project. */ +export const ThreadLaunchInputSchema = lazySchema(() => z.object({ + title: z.string().trim().min(1).max(THREAD_CREATE_TITLE_MAX_LENGTH), + prompt: z.string().min(1).max(THREAD_CREATE_PROMPT_MAX_LENGTH), + workspace: ThreadLaunchWorkspaceSchema().default({ type: "root" }), + providerId: executionId.optional(), + modelId: executionId.optional(), + permissionMode: PermissionModeSchema.optional(), + interactionMode: InteractionModeSchema.optional(), +}).strict()); +/** Thread launch request before workspace defaults are applied. */ +export type ThreadLaunchInput = z.input>; + +/** Ordinary thread identity or an authority-safe launch rejection. */ +export const ThreadLaunchResultSchema = lazySchema(() => z.union([ + z.object({ status: z.literal("launched"), threadId: opaqueId }).strict(), + taskRejection(), +])); +/** Result of launching an ordinary thread. */ +export type ThreadLaunchResult = z.infer>; + /** Durable receipt returned immediately after delegation admission. */ export const DelegateTaskResultSchema = lazySchema(() => z.union([ z.object({ taskId: opaqueId, threadId: opaqueId.optional(), state: TaskStateSchema(), effectivePermissionMode: PermissionModeSchema }).strict(), diff --git a/packages/thread-orchestration/src/mcode-instructions.test.ts b/packages/thread-orchestration/src/mcode-instructions.test.ts new file mode 100644 index 000000000..ab57bfe02 --- /dev/null +++ b/packages/thread-orchestration/src/mcode-instructions.test.ts @@ -0,0 +1,20 @@ +import * as NodeBuffer from "node:buffer"; +import { describe, expect, it } from "vitest"; +import { buildMcodeInstructionSections, isExplicitMcodeThreadRequest, MCODE_INSTRUCTION_SECTION_MAX_BYTES } from "./mcode-instructions.js"; + +describe("thread launch requests", () => { + it("recognizes a request naming thread_launch", () => { + expect(isExplicitMcodeThreadRequest("Use thread_launch to start a separate conversation")).toBe(true); + }); + + it("rejects a negated thread_launch request", () => { + expect(isExplicitMcodeThreadRequest("Don't use thread_launch")).toBe(false); + }); + + it("keeps launch guidance within the provider context budget", () => { + const sections = buildMcodeInstructionSections({ toolkits: { threadControl: true, browser: false } }); + for (const section of sections) { + expect(NodeBuffer.Buffer.byteLength(section.text, "utf8")).toBeLessThanOrEqual(MCODE_INSTRUCTION_SECTION_MAX_BYTES); + } + }); +}); diff --git a/packages/thread-orchestration/src/mcode-instructions.ts b/packages/thread-orchestration/src/mcode-instructions.ts index d0d4a2ac4..2f7ab9e99 100644 --- a/packages/thread-orchestration/src/mcode-instructions.ts +++ b/packages/thread-orchestration/src/mcode-instructions.ts @@ -3,8 +3,8 @@ /** Codex silently cuts the middle out of an additionalContext value above ~1,000 tokens (UTF-8 bytes / 4). */ export const MCODE_INSTRUCTION_SECTION_MAX_BYTES = 4_000; -const EXPLICIT_MCODE_THREAD_TOKEN = /\b(?:mcode(?:[-_\s]+managed)?[-_\s]+threads?|mcode[-_\s]+thread[-_\s]?control|thread_(?:target_list|create_batch|search|get|send|stop|wait)|delegate_task|task_(?:status|cancel))\b/i; -const NEGATED_MCODE_THREAD_REQUEST = /\b(?:do\s+not|don['’]?t|dont|never|without|no)\b[\s\S]{0,96}\b(?:mcode(?:[-_\s]+managed)?[-_\s]+threads?|mcode[-_\s]+thread[-_\s]?control|thread_(?:target_list|create_batch|search|get|send|stop|wait)|delegate_task|task_(?:status|cancel))\b/i; +const EXPLICIT_MCODE_THREAD_TOKEN = /\b(?:mcode(?:[-_\s]+managed)?[-_\s]+threads?|mcode[-_\s]+thread[-_\s]?control|thread_(?:target_list|create_batch|launch|search|get|send|stop|wait)|delegate_task|task_(?:status|cancel))\b/i; +const NEGATED_MCODE_THREAD_REQUEST = /\b(?:do\s+not|don['’]?t|dont|never|without|no)\b[\s\S]{0,96}\b(?:mcode(?:[-_\s]+managed)?[-_\s]+threads?|mcode[-_\s]+thread[-_\s]?control|thread_(?:target_list|create_batch|launch|search|get|send|stop|wait)|delegate_task|task_(?:status|cancel))\b/i; const MCODE_THREAD_REQUEST_ACTION = /\b(?:create|start|open|launch|use|manage|list|search|send|stop|wait|get|target|delegate|call|check|cancel)\b/i; /** Stable section names used as Codex context tags. */ @@ -61,6 +61,7 @@ const THREAD_CONTROL_BLOCK = [ "- Use the mcode MCP thread_* tools only when the user explicitly asks for Mcode threads. An Mcode thread is a persistent user-visible conversation, different from provider-native child agents.", "- If no Project is named, use the active source thread's Project without discovery. Resolve an explicitly named Project with workspace_search. Use worktree_list only to resolve an explicit worktree or reference.", "- Delegate child work with delegate_task. Preserve the user's title and task in the prompt. Choose isolation shared for the Coordinator's checkout or worktree for a new checkout at its local commit. Retry the same input with the same clientRequestId to return the existing task.", + "- thread_launch starts a separate user-owned thread only when the user asks for one; use delegate_task for child work whose result you need.", "- Call thread_target_list before choosing a provider or model, and pass the exact returned providerId and modelId. Pass any named provider or execution setting. Never assume or enumerate providers or models.", "- Use returned taskId and threadId references for follow-up work. delegate_task returns immediately: end the turn or continue other work and check later with task_status rather than polling in a loop. Use task_cancel to stop delegated work and thread_get to inspect its thread. Run each review round as a fresh task carrying the brief, prior findings and open objections.", "- Never target the active source thread. Keep prompts faithful to the user's request and inspect results before sending follow-up work.",