test(projects): prove shared creation admission

Julius Marminge committed 2 months ago

Commit 9fb259c1 parent adec2225 files · +260 −191Browse files

Unified · Split

5 files

apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +237 −6
apps/server/src/mcp/ProjectMcpService.test.ts +0 −155
apps/server/src/orchestration-v2/ClientCommandDispatch.test.ts +5 −9
apps/server/src/orchestration-v2/ClientCommandDispatch.ts +15 −15
apps/server/src/ws.ts +3 −6

apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts+237−6

@@ -19,6 +19,7 @@ import { · 18 unmodified lines
19	OrchestratorMcpThreadReadResult,	19	OrchestratorMcpThreadReadResult,
20	OrchestratorMcpThreadSendResult,	20	OrchestratorMcpThreadSendResult,
21	OrchestratorMcpThreadWaitResult,	21	OrchestratorMcpThreadWaitResult,
		22	type Project,
22	ProjectId,	23	ProjectId,
23	ProviderDriverKind,	24	ProviderDriverKind,
24	ProviderInstanceId,	25	ProviderInstanceId,
@@ -37,6 +38,7 @@ import * as DateTime from "effect/DateTime"; · 12 unmodified lines
37	import * as Deferred from "effect/Deferred";	38	import * as Deferred from "effect/Deferred";
38	import * as Effect from "effect/Effect";	39	import * as Effect from "effect/Effect";
39	import * as FileSystem from "effect/FileSystem";	40	import * as FileSystem from "effect/FileSystem";
		41	import * as Fiber from "effect/Fiber";
40	import * as Layer from "effect/Layer";	42	import * as Layer from "effect/Layer";
41	import * as Option from "effect/Option";	43	import * as Option from "effect/Option";
42	import * as PubSub from "effect/PubSub";	44	import * as PubSub from "effect/PubSub";
@@ -48,8 +50,9 @@ import { McpSchema, McpServer } from "effect/unstable/ai"; · 5 unmodified lines
48	import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts";	50	import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts";
49	import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts";	51	import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts";
50	import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts";	52	import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts";
		53	import * as ClientCommandDispatch from "../orchestration-v2/ClientCommandDispatch.ts";
51	import { OrchestratorV2, type OrchestratorV2Shape } from "../orchestration-v2/Orchestrator.ts";	54	import { OrchestratorV2, type OrchestratorV2Shape } from "../orchestration-v2/Orchestrator.ts";
52	import { layer as threadManagementServiceLayer } from "../orchestration-v2/ThreadManagementService.ts";	55	import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
53	import {	56	import {
54	type ProviderAdapterV2Event,	57	type ProviderAdapterV2Event,
55	ProviderAdapterProtocolError,	58	ProviderAdapterProtocolError,
@@ -71,13 +74,18 @@ import { · 15 unmodified lines
71	materializeReplayTranscriptWorkspace,	74	materializeReplayTranscriptWorkspace,
72	} from "../orchestration-v2/testkit/ReplayTranscriptNdjson.ts";	75	} from "../orchestration-v2/testkit/ReplayTranscriptNdjson.ts";
73	import { makeProviderRegistryLayer } from "../provider/testUtils/providerRegistryMock.ts";	76	import { makeProviderRegistryLayer } from "../provider/testUtils/providerRegistryMock.ts";
		77	import * as ProjectService from "../project/ProjectService.ts";
74	import { ScheduledTaskService } from "../scheduledTasks/ScheduledTaskService.ts";	78	import { ScheduledTaskService } from "../scheduledTasks/ScheduledTaskService.ts";
		79	import * as ServerSettings from "../serverSettings.ts";
		80	import * as SourceControlRepositoryService from "../sourceControl/SourceControlRepositoryService.ts";
75	import * as McpHttpServer from "./McpHttpServer.ts";	81	import * as McpHttpServer from "./McpHttpServer.ts";
76	import * as McpInvocationContext from "./McpInvocationContext.ts";	82	import * as McpInvocationContext from "./McpInvocationContext.ts";
77	import { delegatedTaskRun, hasPendingChildRuns } from "./OrchestratorMcpService.ts";	83	import { delegatedTaskRun, hasPendingChildRuns } from "./OrchestratorMcpService.ts";
78		84
79	const parentThreadId = ThreadId.make("thread:mcp-orchestrator-parent");	85	const parentThreadId = ThreadId.make("thread:mcp-orchestrator-parent");
80	const projectId = ProjectId.make("project:mcp-orchestrator");	86	const projectId = ProjectId.make("project:mcp-orchestrator");
		87	const deletionRaceProjectId = ProjectId.make("project:mcp-orchestrator-delete-race");
		88	const deletionRaceParentThreadId = ThreadId.make("thread:mcp-orchestrator-delete-race-parent");
81	const codexInstanceId = ProviderInstanceId.make("codex");	89	const codexInstanceId = ProviderInstanceId.make("codex");
82	const claudeInstanceId = ProviderInstanceId.make("claudeAgent");	90	const claudeInstanceId = ProviderInstanceId.make("claudeAgent");
83	const codexModel = "gpt-5.4";	91	const codexModel = "gpt-5.4";
@@ -472,6 +480,28 @@ describe("orchestrator MCP toolkit", () => { · 388 unmodified lines
472	Effect.scoped(	480	Effect.scoped(
473	Effect.gen(function* () {	481	Effect.gen(function* () {
474	const cwd = yield* checkpointWorkspace("orchestrator-mcp-toolkit");	482	const cwd = yield* checkpointWorkspace("orchestrator-mcp-toolkit");
		483	const deletionRaceProject = {
		484	id: deletionRaceProjectId,
		485	title: "MCP deletion race project",
		486	workspaceRoot: cwd,
		487	repositoryIdentity: null,
		488	faviconPath: null,
		489	defaultModelSelection: codexSelection,
		490	defaultThreadEnvMode: "local",
		491	scripts: [],
		492	createdAt: "2026-08-30T00:00:00.000Z",
		493	updatedAt: "2026-08-30T00:00:00.000Z",
		494	deletedAt: null,
		495	} satisfies Project;
		496	const deletionRaceProjectState = yield* Ref.make<Project>(deletionRaceProject);
		497	const creationAdmissionEntered = yield* Deferred.make<void>();
		498	const allowCreationAdmission = yield* Deferred.make<void>();
		499	const deletionLockAttempted = yield* Deferred.make<void>();
		500	const deletionLockAcquired = yield* Deferred.make<void>();
		501	const allowDeletion = yield* Deferred.make<void>();
		502	const deletionRaceCommandId = CommandId.make(
		503	"command:mcp:mcp-provider-session-delete-race:create-thread:delete-create-race:0",
		504	);
475	const capturedTurns = yield* Ref.make<ReadonlyArray<CapturedTurn>>([]);	505	const capturedTurns = yield* Ref.make<ReadonlyArray<CapturedTurn>>([]);
476	const parentTerminalGates = new Map<ThreadId, Deferred.Deferred<void>>();	506	const parentTerminalGates = new Map<ThreadId, Deferred.Deferred<void>>();
477	const deliveryTerminalGates = new Map<ThreadId, Deferred.Deferred<void>>();	507	const deliveryTerminalGates = new Map<ThreadId, Deferred.Deferred<void>>();
@@ -482,7 +512,9 @@ describe("orchestrator MCP toolkit", () => { · 4 unmodified lines
482	capabilities: CodexProviderCapabilitiesV2,	512	capabilities: CodexProviderCapabilitiesV2,
483	capturedTurns,	513	capturedTurns,
484	shouldComplete: (turn) =>	514	shouldComplete: (turn) =>
485	turn.threadId !== parentThreadId && turn.message.text !== cancellationPrompt,	515	turn.threadId !== parentThreadId &&
		516	turn.threadId !== deletionRaceParentThreadId &&
		517	turn.message.text !== cancellationPrompt,
486	terminalGate: (turn) =>	518	terminalGate: (turn) =>
487	turn.message.text.startsWith("Delegated task") \|\|	519	turn.message.text.startsWith("Delegated task") \|\|
488	turn.message.text.startsWith("Delegated tasks")	520	turn.message.text.startsWith("Delegated tasks")
@@ -554,10 +586,42 @@ describe("orchestrator MCP toolkit", () => { · 65 unmodified lines
554	},	586	},
555	registryLayer,	587	registryLayer,
556	).pipe(Layer.provide(continuationProbeLayer));	588	).pipe(Layer.provide(continuationProbeLayer));
557	const orchestrationLayer = Layer.merge(	589	const threadManagementLayer = ThreadManagement.layer.pipe(
558	orchestratorLayer,	590	Layer.provide(orchestratorLayer),
559	threadManagementServiceLayer.pipe(Layer.provide(orchestratorLayer)),
560	);	591	);
		592	const gatedThreadManagementLayer = Layer.effect(
		593	ThreadManagement.ThreadManagementService,
		594	Effect.gen(function* () {
		595	const actual = yield* ThreadManagement.ThreadManagementService;
		596	return ThreadManagement.ThreadManagementService.of({
		597	...actual,
		598	withProjectCreationAdmission: (input, effect) =>
		599	actual.withProjectCreationAdmission(input, (receipt) =>
		600	input.commandId === deletionRaceCommandId && Option.isNone(receipt)
		601	? Deferred.succeed(creationAdmissionEntered, undefined).pipe(
		602	Effect.andThen(Deferred.await(allowCreationAdmission)),
		603	Effect.andThen(effect(receipt)),
		604	)
		605	: effect(receipt),
		606	),
		607	withProjectMutationLock: (lockedProjectId, effect) =>
		608	lockedProjectId === deletionRaceProjectId
		609	? Deferred.succeed(deletionLockAttempted, undefined).pipe(
		610	Effect.andThen(
		611	actual.withProjectMutationLock(
		612	lockedProjectId,
		613	Deferred.succeed(deletionLockAcquired, undefined).pipe(
		614	Effect.andThen(Deferred.await(allowDeletion)),
		615	Effect.andThen(effect),
		616	),
		617	),
		618	),
		619	)
		620	: actual.withProjectMutationLock(lockedProjectId, effect),
		621	});
		622	}),
		623	).pipe(Layer.provide(threadManagementLayer));
		624	const orchestrationLayer = Layer.merge(orchestratorLayer, gatedThreadManagementLayer);
561	const providerRegistryLayer = makeProviderRegistryLayer([	625	const providerRegistryLayer = makeProviderRegistryLayer([
562	makeProviderSnapshot({	626	makeProviderSnapshot({
563	instanceId: codexInstanceId,	627	instanceId: codexInstanceId,
@@ -613,11 +677,40 @@ describe("orchestrator MCP toolkit", () => { · 49 unmodified lines
613	runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"),	677	runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"),
614	}),	678	}),
615	);	679	);
616	const testLayer = McpHttpServer.OrchestratorToolkitRegistrationLive.pipe(	680	const projectLayer = Layer.mock(ProjectService.ProjectService)({
		681	getById: (requestedProjectId, options) =>
		682	requestedProjectId === deletionRaceProjectId
		683	? Ref.get(deletionRaceProjectState).pipe(
		684	Effect.map((project) =>
		685	project.deletedAt === null \|\| options?.includeDeleted === true
		686	? Option.some(project)
		687	: Option.none(),
		688	),
		689	)
		690	: Effect.succeed(Option.none()),
		691	delete: ({ projectId: deletedProjectId }) =>
		692	deletedProjectId === deletionRaceProjectId
		693	? Ref.updateAndGet(deletionRaceProjectState, (project) => ({
		694	...project,
		695	deletedAt: "2026-08-30T00:01:00.000Z",
		696	updatedAt: "2026-08-30T00:01:00.000Z",
		697	}))
		698	: Effect.die(`Unexpected project deletion: ${deletedProjectId}`),
		699	});
		700	const toolkitRegistrationLayer = Layer.merge(
		701	McpHttpServer.OrchestratorToolkitRegistrationLive,
		702	McpHttpServer.ProjectToolkitRegistrationLive,
		703	);
		704	const testLayer = toolkitRegistrationLayer.pipe(
617	Layer.provideMerge(McpServer.McpServer.layer),	705	Layer.provideMerge(McpServer.McpServer.layer),
618	Layer.provideMerge(orchestrationLayer),	706	Layer.provideMerge(orchestrationLayer),
619	Layer.provide(providerRegistryLayer),	707	Layer.provide(providerRegistryLayer),
620	Layer.provide(scheduledTaskStubLayer),	708	Layer.provide(scheduledTaskStubLayer),
		709	Layer.provideMerge(projectLayer),
		710	Layer.provide(
		711	Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({}),
		712	),
		713	Layer.provide(ServerSettings.layerTest({})),
621	Layer.provide(NodeServices.layer),	714	Layer.provide(NodeServices.layer),
622	);	715	);
623		716
@@ -2876,6 +2969,144 @@ describe("orchestrator MCP toolkit", () => { · 2252 unmodified lines
2876	removedDelivery.subagents.find((task) => task.id === removeTask.id),	2969	removedDelivery.subagents.find((task) => task.id === removeTask.id),
2877	).toMatchObject({ result: expect.any(String), status: "completed" });	2970	).toMatchObject({ result: expect.any(String), status: "completed" });
2878	yield* expectOffersToStay(0);	2971	yield* expectOffersToStay(0);
		2972
		2973	yield* orchestrator.dispatch({
		2974	type: "thread.create",
		2975	createdBy: "user",
		2976	creationSource: "web",
		2977	commandId: CommandId.make("command:mcp-delete-race-parent:create"),
		2978	threadId: deletionRaceParentThreadId,
		2979	projectId: deletionRaceProjectId,
		2980	title: "MCP deletion race parent",
		2981	modelSelection: codexSelection,
		2982	runtimeMode: "full-access",
		2983	interactionMode: "default",
		2984	branch: null,
		2985	worktreePath: cwd,
		2986	});
		2987	yield* orchestrator.dispatch({
		2988	type: "message.dispatch",
		2989	createdBy: "user",
		2990	creationSource: "web",
		2991	commandId: CommandId.make("command:mcp-delete-race-parent:start"),
		2992	threadId: deletionRaceParentThreadId,
		2993	messageId: MessageId.make("message:mcp-delete-race-parent:start"),
		2994	text: "Stay active while project deletion races thread creation.",
		2995	attachments: [],
		2996	modelSelection: codexSelection,
		2997	dispatchMode: { type: "start_immediately" },
		2998	});
		2999	yield* waitForProjection(orchestrator, deletionRaceParentThreadId, (projection) =>
		3000	projection.runs.some((run) => run.status === "running"),
		3001	);
		3002
		3003	const deletionRaceInvocation: McpInvocationContext.McpInvocationScope = {
		3004	...invocation,
		3005	threadId: deletionRaceParentThreadId,
		3006	providerSessionId: "mcp-provider-session-delete-race",
		3007	};
		3008	const invokeDeletionRace = (name: string, args: Record<string, unknown>) =>
		3009	server
		3010	.callTool({ name, arguments: args })
		3011	.pipe(
		3012	Effect.provideService(
		3013	McpInvocationContext.McpInvocationContext,
		3014	deletionRaceInvocation,
		3015	),
		3016	Effect.provideService(McpSchema.McpServerClient, client),
		3017	);
		3018	const creation = yield* Effect.forkChild(
		3019	invokeDeletionRace("create_threads", {
		3020	clientRequestId: "delete-create-race",
		3021	threads: [{ title: "Racing child" }],
		3022	}),
		3023	{ startImmediately: true },
		3024	);
		3025	yield* Deferred.await(creationAdmissionEntered);
		3026	const deletion = yield* Effect.forkChild(
		3027	invokeDeletionRace("t3_project_delete", {
		3028	projectId: deletionRaceProjectId,
		3029	cascadeThreads: true,
		3030	clientRequestId: "delete-create-race-project",
		3031	}),
		3032	{ startImmediately: true },
		3033	);
		3034	yield* Deferred.await(deletionLockAttempted);
		3035	expect(yield* Deferred.isDone(deletionLockAcquired)).toBe(false);
		3036
		3037	yield* Deferred.succeed(allowCreationAdmission, undefined);
		3038	yield* Deferred.await(deletionLockAcquired);
		3039	const creationCall = yield* Fiber.join(creation);
		3040	expect(creationCall.isError).toBe(false);
		3041	const creationResult = yield* decodeCreateThreadsResult(
		3042	creationCall.structuredContent,
		3043	).pipe(Effect.orDie);
		3044	const createdThread = creationResult.threads[0];
		3045	if (createdThread === undefined) {
		3046	return yield* Effect.die(new Error("Racing MCP thread was not created."));
		3047	}
		3048	expect(
		3049	(yield* orchestrator.getThreadProjection(createdThread.threadId)).thread.projectId,
		3050	).toBe(deletionRaceProjectId);
		3051
		3052	yield* Deferred.succeed(allowDeletion, undefined);
		3053	const deletionCall = yield* Fiber.join(deletion);
		3054	expect(deletionCall.isError).toBe(false);
		3055	expect(deletionCall.structuredContent).toMatchObject({
		3056	projectId: deletionRaceProjectId,
		3057	deleted: true,
		3058	alreadyDeleted: false,
		3059	deletedThreadCount: 2,
		3060	workspaceFilesDeleted: false,
		3061	});
		3062	expect(
		3063	(yield* orchestrator.getThreadProjection(createdThread.threadId)).thread.deletedAt,
		3064	).not.toBeNull();
		3065	expect(
		3066	(yield* orchestrator.getThreadProjection(deletionRaceParentThreadId)).thread
		3067	.deletedAt,
		3068	).not.toBeNull();
		3069
		3070	const threads = yield* ThreadManagement.ThreadManagementService;
		3071	const dispatchClientCommand = yield* ClientCommandDispatch.make;
		3072	const acceptedCreateCommand = {
		3073	type: "thread.create",
		3074	createdBy: "agent",
		3075	creationSource: "mcp",
		3076	commandId: deletionRaceCommandId,
		3077	threadId: createdThread.threadId,
		3078	projectId: deletionRaceProjectId,
		3079	title: "Racing child",
		3080	modelSelection: codexSelection,
		3081	runtimeMode: "full-access",
		3082	interactionMode: "default",
		3083	branch: null,
		3084	worktreePath: cwd,
		3085	} as const;
		3086	const sequenceBeforeReplay = yield* threads.getThreadEventSequence(
		3087	createdThread.threadId,
		3088	);
		3089	const replayed = yield* dispatchClientCommand(acceptedCreateCommand);
		3090	expect(
		3091	replayed.storedEvents.some((stored) => stored.event.type === "thread.created"),
		3092	).toBe(true);
		3093	expect(yield* threads.getThreadEventSequence(createdThread.threadId)).toBe(
		3094	sequenceBeforeReplay,
		3095	);
		3096	expect(
		3097	(yield* orchestrator.getThreadProjection(createdThread.threadId)).thread.deletedAt,
		3098	).not.toBeNull();
		3099
		3100	const freshThreadId = ThreadId.make("thread:mcp-delete-race:fresh-after-delete");
		3101	const freshFailure = yield* dispatchClientCommand({
		3102	...acceptedCreateCommand,
		3103	commandId: CommandId.make("command:mcp-delete-race:fresh-after-delete"),
		3104	threadId: freshThreadId,
		3105	}).pipe(Effect.flip);
		3106	expect(freshFailure).toMatchObject({ _tag: "ProjectMutationError" });
		3107	expect(
		3108	Option.isNone(yield* Effect.option(orchestrator.getThreadProjection(freshThreadId))),
		3109	).toBe(true);
2879	}).pipe(Effect.provide(testLayer));	3110	}).pipe(Effect.provide(testLayer));
2880	}),	3111	}),
2881	),	3112	),

apps/server/src/mcp/ProjectMcpService.test.ts+0−155

@@ -462,161 +462,6 @@ it.effect("clones before registration and requires explicit cascading for nonemp · 461 unmodified lines
462	}),	462	}),
463	);	463	);
464		464
465	it.effect("serializes receipt-aware thread creation with project deletion", () =>
466	Effect.gen(function* () {
467	const projectId = ProjectId.make("project:mcp:delete-launch-race");
468	const threadId = ThreadId.make("thread:mcp:delete-launch-race");
469	const modelSelection = {
470	instanceId: ProviderInstanceId.make("codex"),
471	model: "gpt-5.1-codex",
472	} as const;
473	const project = makeProject({
474	id: projectId,
475	title: "Delete race",
476	workspaceRoot: "/work/delete-race",
477	});
478	const projectState = yield* Ref.make<Project>(project);
479	const creationEntered = yield* Deferred.make<void>();
480	const allowCreation = yield* Deferred.make<void>();
481	const deleteLockAttempted = yield* Deferred.make<void>();
482	const deleteLockAcquired = yield* Deferred.make<void>();
483	const adapter = {
484	instanceId: modelSelection.instanceId,
485	driver: ProviderDriverKind.make("codex"),
486	getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
487	planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
488	openSession: () => Effect.die("provider execution is disabled in project deletion tests"),
489	} as ProviderAdapterV2Shape;
490	const registry = ProviderAdapterRegistry.makeLayer([adapter]);
491	const orchestrator = makeOrchestratorV2ReplayLayerWithRegistry(
492	{ name: "project-delete-create-race" },
493	registry,
494	{ databaseLayer: SqlitePersistenceMemory, runEffectWorker: false },
495	);
496	const actualThreads = ThreadManagement.layer.pipe(Layer.provide(orchestrator));
497	const gatedThreads = Layer.effect(
498	ThreadManagement.ThreadManagementService,
499	Effect.gen(function* () {
500	const actual = yield* ThreadManagement.ThreadManagementService;
501	return ThreadManagement.ThreadManagementService.of({
502	...actual,
503	withProjectMutationLock: (lockedProjectId, effect) =>
504	Deferred.succeed(deleteLockAttempted, undefined).pipe(
505	Effect.andThen(
506	actual.withProjectMutationLock(
507	lockedProjectId,
508	Deferred.succeed(deleteLockAcquired, undefined).pipe(Effect.andThen(effect)),
509	),
510	),
511	),
512	});
513	}),
514	).pipe(Layer.provide(actualThreads));
515	const projectServiceLayer = Layer.mock(ProjectService.ProjectService)({
516	getById: () =>
517	Ref.get(projectState).pipe(
518	Effect.map((current) =>
519	current.deletedAt === null ? Option.some(current) : Option.none(),
520	),
521	),
522	delete: () => Ref.updateAndGet(projectState, (current) => ({ ...current, deletedAt: now })),
523	});
524	const testLayer = ProjectMcp.layer.pipe(
525	Layer.provideMerge(gatedThreads),
526	Layer.provide(projectServiceLayer),
527	Layer.provide(Layer.mock(T3ProjectFileLoader.T3ProjectFileLoader)({})),
528	Layer.provide(
529	Layer.mock(ServerSettingsService.ServerSettingsService)({
530	getSettings: Effect.succeed({ defaultThreadEnvMode: "local" } as ServerSettings),
531	}),
532	),
533	Layer.provide(Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({})),
534	Layer.provide(NodeServices.layer),
535	);
536
537	yield* Effect.gen(function* () {
538	const projects = yield* ProjectMcp.ProjectMcpService;
539	const threads = yield* ThreadManagement.ThreadManagementService;
540	const createCommand = {
541	type: "thread.create",
542	createdBy: "user",
543	creationSource: "web",
544	commandId: CommandId.make("command:mcp:delete-launch-race:create"),
545	threadId,
546	projectId,
547	title: "Racing thread",
548	modelSelection,
549	runtimeMode: "full-access",
550	interactionMode: "default",
551	branch: null,
552	worktreePath: null,
553	} as const;
554	const admitCreate = (command: typeof createCommand) =>
555	threads.withProjectCreationAdmission(
556	{ projectId, commandId: command.commandId },
557	(receipt) =>
558	Effect.gen(function* () {
559	if (Option.isNone(receipt)) {
560	if (command.commandId === createCommand.commandId) {
561	yield* Deferred.succeed(creationEntered, undefined);
562	yield* Deferred.await(allowCreation);
563	}
564	const current = yield* Ref.get(projectState);
565	if (current.deletedAt !== null) {
566	return yield* new ProjectMcpFailure({
567	code: "project_deleted",
568	message: "Project was deleted before thread creation was admitted.",
569	});
570	}
571	}
572	return yield* threads.dispatch(command);
573	}),
574	);
575
576	const creation = yield* Effect.forkChild(admitCreate(createCommand), {
577	startImmediately: true,
578	});
579	yield* Deferred.await(creationEntered);
580	const deletion = yield* Effect.forkChild(
581	projects.delete(scope, {
582	projectId,
583	cascadeThreads: true,
584	clientRequestId: "delete-launch-race",
585	}),
586	{ startImmediately: true },
587	);
588	yield* Deferred.await(deleteLockAttempted);
589	expect(yield* Deferred.isDone(deleteLockAcquired)).toBe(false);
590
591	yield* Deferred.succeed(allowCreation, undefined);
592	yield* Fiber.join(creation);
593	yield* Deferred.await(deleteLockAcquired);
594	const deleted = yield* Fiber.join(deletion);
595	expect(deleted).toMatchObject({ deleted: true, deletedThreadCount: 1 });
596
597	const sequenceBeforeReplay = yield* threads.getThreadEventSequence(threadId);
598	const replayed = yield* admitCreate(createCommand);
599	const sequenceAfterReplay = yield* threads.getThreadEventSequence(threadId);
600	expect(replayed.storedEvents.some((stored) => stored.event.type === "thread.created")).toBe(
601	true,
602	);
603	expect(sequenceAfterReplay).toBe(sequenceBeforeReplay);
604
605	const freshCommand = {
606	...createCommand,
607	commandId: CommandId.make("command:mcp:delete-launch-race:fresh-after-delete"),
608	threadId: ThreadId.make("thread:mcp:delete-launch-race:fresh-after-delete"),
609	};
610	const rejected = yield* admitCreate(freshCommand).pipe(Effect.flip);
611	expect(rejected).toMatchObject({ code: "project_deleted" });
612	expect(
613	Option.isNone(yield* Effect.option(threads.getThreadProjection(freshCommand.threadId))),
614	).toBe(true);
615	expect((yield* Ref.get(projectState)).deletedAt).toBe(now);
616	}).pipe(Effect.provide(testLayer));
617	}),
618	);
619
620	it.effect("serializes delegated thread admission with cascading project deletion", () =>	465	it.effect("serializes delegated thread admission with cascading project deletion", () =>
621	Effect.gen(function* () {	466	Effect.gen(function* () {
622	const projectId = ProjectId.make("project:mcp:delete-delegate-race");	467	const projectId = ProjectId.make("project:mcp:delete-delegate-race");

apps/server/src/orchestration-v2/ClientCommandDispatch.test.ts+5−9

 import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
 import * as ProjectService from "../project/ProjectService.ts";
 import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
-import { dispatchClientCommand } from "./ClientCommandDispatch.ts";
+import * as ClientCommandDispatch from "./ClientCommandDispatch.ts";
 import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";
 import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
 import * as ThreadManagement from "./ThreadManagementService.ts";
     yield* Effect.gen(function* () {
       const threads = yield* ThreadManagement.ThreadManagementService;
-      const projects = yield* ProjectService.ProjectService;
+      const dispatchClientCommand = yield* ClientCommandDispatch.make;
       const command = {
         type: "thread.create",
         createdBy: "user",
         worktreePath: null,
       } as const;
-      yield* dispatchClientCommand({ command, projects, threads });
+      yield* dispatchClientCommand(command);
       const sequenceBeforeReplay = yield* threads.getThreadEventSequence(command.threadId);
       yield* Ref.set(projectState, null);
-      yield* dispatchClientCommand({ command, projects, threads });
+      yield* dispatchClientCommand(command);
       expect(yield* threads.getThreadEventSequence(command.threadId)).toBe(sequenceBeforeReplay);
       const freshCommand = {
         commandId: CommandId.make("command:client-command-admission:fresh"),
         threadId: ThreadId.make("thread:client-command-admission:fresh"),
       };
-      const rejected = yield* dispatchClientCommand({
+      const rejected = yield* dispatchClientCommand(freshCommand).pipe(Effect.flip);
-        command: freshCommand,
-        projects,
-        threads,
-      }).pipe(Effect.flip);
       expect(rejected).toMatchObject({ _tag: "ProjectMutationError" });
       expect(
         Option.isNone(yield* Effect.option(threads.getThreadProjection(freshCommand.threadId))),

apps/server/src/orchestration-v2/ClientCommandDispatch.ts+15−15

@@ -2,25 +2,25 @@ import { ProjectMutationError, type OrchestrationV2Command } from "@t3tools/cont · 1 unmodified line
2	import * as Effect from "effect/Effect";	2	import * as Effect from "effect/Effect";
3	import * as Option from "effect/Option";	3	import * as Option from "effect/Option";
4		4
5	import type * as ProjectService from "../project/ProjectService.ts";	5	import * as ProjectService from "../project/ProjectService.ts";
6	import type * as ThreadManagement from "./ThreadManagementService.ts";	6	import * as ThreadManagement from "./ThreadManagementService.ts";
7		7
8	export const dispatchClientCommand = Effect.fn("orchestrationV2.dispatchClientCommand")(	8	export const make = Effect.gen(function* () {
9	function* (input: {	9	const projects = yield* ProjectService.ProjectService;
10	readonly command: OrchestrationV2Command;	10	const threads = yield* ThreadManagement.ThreadManagementService;
11	readonly projects: ProjectService.ProjectService["Service"];	11
12	readonly threads: ThreadManagement.ThreadManagementService["Service"];	12	return Effect.fn("orchestrationV2.dispatchClientCommand")(function* (
13	}) {	13	command: OrchestrationV2Command,
14	const command = input.command;	14	) {
15	if (command.type !== "thread.create") {	15	if (command.type !== "thread.create") {
16	return yield* input.threads.dispatch(command);	16	return yield* threads.dispatch(command);
17	}	17	}
18	return yield* input.threads.withProjectCreationAdmission(	18	return yield* threads.withProjectCreationAdmission(
19	{ projectId: command.projectId, commandId: command.commandId },	19	{ projectId: command.projectId, commandId: command.commandId },
20	(receipt) =>	20	(receipt) =>
21	Effect.gen(function* () {	21	Effect.gen(function* () {
22	if (Option.isNone(receipt)) {	22	if (Option.isNone(receipt)) {
23	const project = yield* input.projects.getById(command.projectId);	23	const project = yield* projects.getById(command.projectId);
24	if (Option.isNone(project)) {	24	if (Option.isNone(project)) {
25	return yield* new ProjectMutationError({	25	return yield* new ProjectMutationError({
26	commandId: command.commandId,	26	commandId: command.commandId,
@@ -28,8 +28,8 @@ export const dispatchClientCommand = Effect.fn("orchestrationV2.dispatchClientCo · 1 unmodified line
28	});	28	});
29	}	29	}
30	}	30	}
31	return yield* input.threads.dispatch(command);	31	return yield* threads.dispatch(command);
32	}),	32	}),
33	);	33	);
34	},	34	});
35	);	35	});

apps/server/src/ws.ts+3−6

 import * as Keybindings from "./keybindings.ts";
 import * as ExternalLauncher from "./process/externalLauncher.ts";
 import * as ThreadManagementService from "./orchestration-v2/ThreadManagementService.ts";
-import { dispatchClientCommand } from "./orchestration-v2/ClientCommandDispatch.ts";
+import * as ClientCommandDispatch from "./orchestration-v2/ClientCommandDispatch.ts";
 import { ProviderSessionManagerV2 } from "./orchestration-v2/ProviderSessionManager.ts";
 import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts";
 import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts";
       const currentSessionId = currentSession.sessionId;
       const sql = yield* SqlClient.SqlClient;
       const threadManagement = yield* ThreadManagementService.ThreadManagementService;
+      const dispatchClientCommand = yield* ClientCommandDispatch.make;
       const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
       const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
       const canReplayPersistedRange = Effect.fnUntraced(function* (
                   creationSource: "creationSource" in command ? command.creationSource : "web",
                 },
               );
-              const dispatchCommand = dispatchClientCommand({
+              const dispatchCommand = dispatchClientCommand(provenanceCommand);
-                command: provenanceCommand,
-                projects: projectService,
-                threads: threadManagement,
-              });
               return yield* startup
                 .enqueueCommand(dispatchCommand)
                 .pipe(

test(projects): prove shared creation admission

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code