soulcraft/brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

fix(src/augmentationPipeline): remove redundant semicolons and enforce consistent formatting

Browse source 
Made code more concise by removing unnecessary semicolons and adjusting the code formatting to align with the project's style. Improved readability and maintainability.
This commit is contained in:
David Snelling 2025-06-26 10:48:10 -07:00
parent 5a8a6c1ba3
commit 47ba4f4093
1 changed files with 340 additions and 189 deletions
Download patch file Download diff file Expand all files Collapse all files

527
src/augmentationPipeline.ts
View file

@ -20,14 +20,14 @@ import { executeInThread } from './utils/workerUtils.js'
* Type definitions for the augmentation registry * Type definitions for the augmentation registry
*/ */
type AugmentationRegistry = { type AugmentationRegistry = {
sense: BrainyAugmentations.ISenseAugmentation[]; sense: BrainyAugmentations.ISenseAugmentation[]
conduit: BrainyAugmentations.IConduitAugmentation[]; conduit: BrainyAugmentations.IConduitAugmentation[]
cognition: BrainyAugmentations.ICognitionAugmentation[]; cognition: BrainyAugmentations.ICognitionAugmentation[]
memory: BrainyAugmentations.IMemoryAugmentation[]; memory: BrainyAugmentations.IMemoryAugmentation[]
perception: BrainyAugmentations.IPerceptionAugmentation[]; perception: BrainyAugmentations.IPerceptionAugmentation[]
dialog: BrainyAugmentations.IDialogAugmentation[]; dialog: BrainyAugmentations.IDialogAugmentation[]
activation: BrainyAugmentations.IActivationAugmentation[]; activation: BrainyAugmentations.IActivationAugmentation[]
webSocket: IWebSocketSupport[]; webSocket: IWebSocketSupport[]
} }
/** /**
@ -38,18 +38,18 @@ export enum ExecutionMode {
PARALLEL = 'parallel', PARALLEL = 'parallel',
FIRST_SUCCESS = 'firstSuccess', FIRST_SUCCESS = 'firstSuccess',
FIRST_RESULT = 'firstResult', FIRST_RESULT = 'firstResult',
THREADED = 'threaded' // Execute in separate threads when available THREADED = 'threaded' // Execute in separate threads when available
} }
/** /**
* Options for pipeline execution * Options for pipeline execution
*/ */
export interface PipelineOptions { export interface PipelineOptions {
mode?: ExecutionMode; mode?: ExecutionMode
timeout?: number; timeout?: number
stopOnError?: boolean; stopOnError?: boolean
forceThreading?: boolean; // Force threading even if not in THREADED mode forceThreading?: boolean // Force threading even if not in THREADED mode
disableThreading?: boolean; // Disable threading even if in THREADED mode disableThreading?: boolean // Disable threading even if in THREADED mode
} }
/** /**
@ -86,78 +86,96 @@ export class AugmentationPipeline {
* @param augmentation The augmentation to register * @param augmentation The augmentation to register
* @returns The pipeline instance for chaining * @returns The pipeline instance for chaining
*/ */
public register<T extends IAugmentation>(augmentation: T): AugmentationPipeline { public register<T extends IAugmentation>(
augmentation: T
): AugmentationPipeline {
let registered = false let registered = false
// Check for specific augmentation types // Check for specific augmentation types
if (this.isAugmentationType<BrainyAugmentations.ISenseAugmentation>( if (
augmentation, this.isAugmentationType<BrainyAugmentations.ISenseAugmentation>(
'processRawData', augmentation,
'listenToFeed' 'processRawData',
)) { 'listenToFeed'
)
) {
this.registry.sense.push(augmentation) this.registry.sense.push(augmentation)
registered = true registered = true
} else if (this.isAugmentationType<BrainyAugmentations.IConduitAugmentation>( } else if (
augmentation, this.isAugmentationType<BrainyAugmentations.IConduitAugmentation>(
'establishConnection', augmentation,
'readData', 'establishConnection',
'writeData', 'readData',
'monitorStream' 'writeData',
)) { 'monitorStream'
)
) {
this.registry.conduit.push(augmentation) this.registry.conduit.push(augmentation)
registered = true registered = true
} else if (this.isAugmentationType<BrainyAugmentations.ICognitionAugmentation>( } else if (
augmentation, this.isAugmentationType<BrainyAugmentations.ICognitionAugmentation>(
'reason', augmentation,
'infer', 'reason',
'executeLogic' 'infer',
)) { 'executeLogic'
)
) {
this.registry.cognition.push(augmentation) this.registry.cognition.push(augmentation)
registered = true registered = true
} else if (this.isAugmentationType<BrainyAugmentations.IMemoryAugmentation>( } else if (
augmentation, this.isAugmentationType<BrainyAugmentations.IMemoryAugmentation>(
'storeData', augmentation,
'retrieveData', 'storeData',
'updateData', 'retrieveData',
'deleteData', 'updateData',
'listDataKeys' 'deleteData',
)) { 'listDataKeys'
)
) {
this.registry.memory.push(augmentation) this.registry.memory.push(augmentation)
registered = true registered = true
} else if (this.isAugmentationType<BrainyAugmentations.IPerceptionAugmentation>( } else if (
augmentation, this.isAugmentationType<BrainyAugmentations.IPerceptionAugmentation>(
'interpret', augmentation,
'organize', 'interpret',
'generateVisualization' 'organize',
)) { 'generateVisualization'
)
) {
this.registry.perception.push(augmentation) this.registry.perception.push(augmentation)
registered = true registered = true
} else if (this.isAugmentationType<BrainyAugmentations.IDialogAugmentation>( } else if (
augmentation, this.isAugmentationType<BrainyAugmentations.IDialogAugmentation>(
'processUserInput', augmentation,
'generateResponse', 'processUserInput',
'manageContext' 'generateResponse',
)) { 'manageContext'
)
) {
this.registry.dialog.push(augmentation) this.registry.dialog.push(augmentation)
registered = true registered = true
} else if (this.isAugmentationType<BrainyAugmentations.IActivationAugmentation>( } else if (
augmentation, this.isAugmentationType<BrainyAugmentations.IActivationAugmentation>(
'triggerAction', augmentation,
'generateOutput', 'triggerAction',
'interactExternal' 'generateOutput',
)) { 'interactExternal'
)
) {
this.registry.activation.push(augmentation) this.registry.activation.push(augmentation)
registered = true registered = true
} }
// Check if the augmentation supports WebSocket // Check if the augmentation supports WebSocket
if (this.isAugmentationType<IWebSocketSupport>( if (
augmentation, this.isAugmentationType<IWebSocketSupport>(
'connectWebSocket', augmentation,
'sendWebSocketMessage', 'connectWebSocket',
'onWebSocketMessage', 'sendWebSocketMessage',
'closeWebSocket' 'onWebSocketMessage',
)) { 'closeWebSocket'
)
) {
this.registry.webSocket.push(augmentation as IWebSocketSupport) this.registry.webSocket.push(augmentation as IWebSocketSupport)
registered = true registered = true
} }
@ -182,7 +200,9 @@ export class AugmentationPipeline {
// Remove from all registries // Remove from all registries
for (const type in this.registry) { for (const type in this.registry) {
const typedRegistry = this.registry[type as keyof AugmentationRegistry] const typedRegistry = this.registry[type as keyof AugmentationRegistry]
const index = typedRegistry.findIndex(aug => aug.name === augmentationName) const index = typedRegistry.findIndex(
(aug) => aug.name === augmentationName
)
if (index !== -1) { if (index !== -1) {
typedRegistry.splice(index, 1) typedRegistry.splice(index, 1)
@ -202,9 +222,12 @@ export class AugmentationPipeline {
const allAugmentations = this.getAllAugmentations() const allAugmentations = this.getAllAugmentations()
await Promise.all( await Promise.all(
allAugmentations.map(augmentation => allAugmentations.map((augmentation) =>
augmentation.initialize().catch(error => { augmentation.initialize().catch((error) => {
console.error(`Failed to initialize augmentation ${augmentation.name}:`, error) console.error(
`Failed to initialize augmentation ${augmentation.name}:`,
error
)
}) })
) )
) )
@ -219,9 +242,12 @@ export class AugmentationPipeline {
const allAugmentations = this.getAllAugmentations() const allAugmentations = this.getAllAugmentations()
await Promise.all( await Promise.all(
allAugmentations.map(augmentation => allAugmentations.map((augmentation) =>
augmentation.shutDown().catch(error => { augmentation.shutDown().catch((error) => {
console.error(`Failed to shut down augmentation ${augmentation.name}:`, error) console.error(
`Failed to shut down augmentation ${augmentation.name}:`,
error
)
}) })
) )
) )
@ -237,19 +263,30 @@ export class AugmentationPipeline {
*/ */
public async executeSensePipeline< public async executeSensePipeline<
M extends keyof BrainyAugmentations.ISenseAugmentation & string, M extends keyof BrainyAugmentations.ISenseAugmentation & string,
R extends BrainyAugmentations.ISenseAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never R extends BrainyAugmentations.ISenseAugmentation[M] extends (
...args: any[]
) => AugmentationResponse<infer U>
? U
: never
>( >(
method: M & (BrainyAugmentations.ISenseAugmentation[M] extends (...args: any[]) => any ? M : never), method: M &
args: Parameters<Extract<BrainyAugmentations.ISenseAugmentation[M], (...args: any[]) => any>>, (BrainyAugmentations.ISenseAugmentation[M] extends (...args: any[]) => any
? M
: never),
args: Parameters<
Extract<
BrainyAugmentations.ISenseAugmentation[M],
(...args: any[]) => any
>
>,
options: PipelineOptions = {} options: PipelineOptions = {}
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> { ): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options } const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
return this.executeTypedPipeline<BrainyAugmentations.ISenseAugmentation, M, R>( return this.executeTypedPipeline<
this.registry.sense, BrainyAugmentations.ISenseAugmentation,
method, M,
args, R
opts >(this.registry.sense, method, args, opts)
)
} }
/** /**
@ -262,19 +299,32 @@ export class AugmentationPipeline {
*/ */
public async executeConduitPipeline< public async executeConduitPipeline<
M extends keyof BrainyAugmentations.IConduitAugmentation & string, M extends keyof BrainyAugmentations.IConduitAugmentation & string,
R extends BrainyAugmentations.IConduitAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never R extends BrainyAugmentations.IConduitAugmentation[M] extends (
...args: any[]
) => AugmentationResponse<infer U>
? U
: never
>( >(
method: M & (BrainyAugmentations.IConduitAugmentation[M] extends (...args: any[]) => any ? M : never), method: M &
args: Parameters<Extract<BrainyAugmentations.IConduitAugmentation[M], (...args: any[]) => any>>, (BrainyAugmentations.IConduitAugmentation[M] extends (
...args: any[]
) => any
? M
: never),
args: Parameters<
Extract<
BrainyAugmentations.IConduitAugmentation[M],
(...args: any[]) => any
>
>,
options: PipelineOptions = {} options: PipelineOptions = {}
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> { ): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options } const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
return this.executeTypedPipeline<BrainyAugmentations.IConduitAugmentation, M, R>( return this.executeTypedPipeline<
this.registry.conduit, BrainyAugmentations.IConduitAugmentation,
method, M,
args, R
opts >(this.registry.conduit, method, args, opts)
)
} }
/** /**
@ -287,19 +337,32 @@ export class AugmentationPipeline {
*/ */
public async executeCognitionPipeline< public async executeCognitionPipeline<
M extends keyof BrainyAugmentations.ICognitionAugmentation & string, M extends keyof BrainyAugmentations.ICognitionAugmentation & string,
R extends BrainyAugmentations.ICognitionAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never R extends BrainyAugmentations.ICognitionAugmentation[M] extends (
...args: any[]
) => AugmentationResponse<infer U>
? U
: never
>( >(
method: M & (BrainyAugmentations.ICognitionAugmentation[M] extends (...args: any[]) => any ? M : never), method: M &
args: Parameters<Extract<BrainyAugmentations.ICognitionAugmentation[M], (...args: any[]) => any>>, (BrainyAugmentations.ICognitionAugmentation[M] extends (
...args: any[]
) => any
? M
: never),
args: Parameters<
Extract<
BrainyAugmentations.ICognitionAugmentation[M],
(...args: any[]) => any
>
>,
options: PipelineOptions = {} options: PipelineOptions = {}
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> { ): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options } const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
return this.executeTypedPipeline<BrainyAugmentations.ICognitionAugmentation, M, R>( return this.executeTypedPipeline<
this.registry.cognition, BrainyAugmentations.ICognitionAugmentation,
method, M,
args, R
opts >(this.registry.cognition, method, args, opts)
)
} }
/** /**
@ -312,19 +375,32 @@ export class AugmentationPipeline {
*/ */
public async executeMemoryPipeline< public async executeMemoryPipeline<
M extends keyof BrainyAugmentations.IMemoryAugmentation & string, M extends keyof BrainyAugmentations.IMemoryAugmentation & string,
R extends BrainyAugmentations.IMemoryAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never R extends BrainyAugmentations.IMemoryAugmentation[M] extends (
...args: any[]
) => AugmentationResponse<infer U>
? U
: never
>( >(
method: M & (BrainyAugmentations.IMemoryAugmentation[M] extends (...args: any[]) => any ? M : never), method: M &
args: Parameters<Extract<BrainyAugmentations.IMemoryAugmentation[M], (...args: any[]) => any>>, (BrainyAugmentations.IMemoryAugmentation[M] extends (
...args: any[]
) => any
? M
: never),
args: Parameters<
Extract<
BrainyAugmentations.IMemoryAugmentation[M],
(...args: any[]) => any
>
>,
options: PipelineOptions = {} options: PipelineOptions = {}
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> { ): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options } const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
return this.executeTypedPipeline<BrainyAugmentations.IMemoryAugmentation, M, R>( return this.executeTypedPipeline<
this.registry.memory, BrainyAugmentations.IMemoryAugmentation,
method, M,
args, R
opts >(this.registry.memory, method, args, opts)
)
} }
/** /**
@ -337,19 +413,32 @@ export class AugmentationPipeline {
*/ */
public async executePerceptionPipeline< public async executePerceptionPipeline<
M extends keyof BrainyAugmentations.IPerceptionAugmentation & string, M extends keyof BrainyAugmentations.IPerceptionAugmentation & string,
R extends BrainyAugmentations.IPerceptionAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never R extends BrainyAugmentations.IPerceptionAugmentation[M] extends (
...args: any[]
) => AugmentationResponse<infer U>
? U
: never
>( >(
method: M & (BrainyAugmentations.IPerceptionAugmentation[M] extends (...args: any[]) => any ? M : never), method: M &
args: Parameters<Extract<BrainyAugmentations.IPerceptionAugmentation[M], (...args: any[]) => any>>, (BrainyAugmentations.IPerceptionAugmentation[M] extends (
...args: any[]
) => any
? M
: never),
args: Parameters<
Extract<
BrainyAugmentations.IPerceptionAugmentation[M],
(...args: any[]) => any
>
>,
options: PipelineOptions = {} options: PipelineOptions = {}
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> { ): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options } const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
return this.executeTypedPipeline<BrainyAugmentations.IPerceptionAugmentation, M, R>( return this.executeTypedPipeline<
this.registry.perception, BrainyAugmentations.IPerceptionAugmentation,
method, M,
args, R
opts >(this.registry.perception, method, args, opts)
)
} }
/** /**
@ -362,19 +451,32 @@ export class AugmentationPipeline {
*/ */
public async executeDialogPipeline< public async executeDialogPipeline<
M extends keyof BrainyAugmentations.IDialogAugmentation & string, M extends keyof BrainyAugmentations.IDialogAugmentation & string,
R extends BrainyAugmentations.IDialogAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never R extends BrainyAugmentations.IDialogAugmentation[M] extends (
...args: any[]
) => AugmentationResponse<infer U>
? U
: never
>( >(
method: M & (BrainyAugmentations.IDialogAugmentation[M] extends (...args: any[]) => any ? M : never), method: M &
args: Parameters<Extract<BrainyAugmentations.IDialogAugmentation[M], (...args: any[]) => any>>, (BrainyAugmentations.IDialogAugmentation[M] extends (
...args: any[]
) => any
? M
: never),
args: Parameters<
Extract<
BrainyAugmentations.IDialogAugmentation[M],
(...args: any[]) => any
>
>,
options: PipelineOptions = {} options: PipelineOptions = {}
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> { ): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options } const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
return this.executeTypedPipeline<BrainyAugmentations.IDialogAugmentation, M, R>( return this.executeTypedPipeline<
this.registry.dialog, BrainyAugmentations.IDialogAugmentation,
method, M,
args, R
opts >(this.registry.dialog, method, args, opts)
)
} }
/** /**
@ -387,19 +489,32 @@ export class AugmentationPipeline {
*/ */
public async executeActivationPipeline< public async executeActivationPipeline<
M extends keyof BrainyAugmentations.IActivationAugmentation & string, M extends keyof BrainyAugmentations.IActivationAugmentation & string,
R extends BrainyAugmentations.IActivationAugmentation[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never R extends BrainyAugmentations.IActivationAugmentation[M] extends (
...args: any[]
) => AugmentationResponse<infer U>
? U
: never
>( >(
method: M & (BrainyAugmentations.IActivationAugmentation[M] extends (...args: any[]) => any ? M : never), method: M &
args: Parameters<Extract<BrainyAugmentations.IActivationAugmentation[M], (...args: any[]) => any>>, (BrainyAugmentations.IActivationAugmentation[M] extends (
...args: any[]
) => any
? M
: never),
args: Parameters<
Extract<
BrainyAugmentations.IActivationAugmentation[M],
(...args: any[]) => any
>
>,
options: PipelineOptions = {} options: PipelineOptions = {}
): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> { ): Promise<Promise<{ success: boolean; data: R; error?: string }>[]> {
const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options } const opts = { ...DEFAULT_PIPELINE_OPTIONS, ...options }
return this.executeTypedPipeline<BrainyAugmentations.IActivationAugmentation, M, R>( return this.executeTypedPipeline<
this.registry.activation, BrainyAugmentations.IActivationAugmentation,
method, M,
args, R
opts >(this.registry.activation, method, args, opts)
)
} }
/** /**
@ -461,14 +576,22 @@ export class AugmentationPipeline {
public getAvailableAugmentationTypes(): AugmentationType[] { public getAvailableAugmentationTypes(): AugmentationType[] {
const availableTypes: AugmentationType[] = [] const availableTypes: AugmentationType[] = []
if (this.registry.sense.length > 0) availableTypes.push(AugmentationType.SENSE) if (this.registry.sense.length > 0)
if (this.registry.conduit.length > 0) availableTypes.push(AugmentationType.CONDUIT) availableTypes.push(AugmentationType.SENSE)
if (this.registry.cognition.length > 0) availableTypes.push(AugmentationType.COGNITION) if (this.registry.conduit.length > 0)
if (this.registry.memory.length > 0) availableTypes.push(AugmentationType.MEMORY) availableTypes.push(AugmentationType.CONDUIT)
if (this.registry.perception.length > 0) availableTypes.push(AugmentationType.PERCEPTION) if (this.registry.cognition.length > 0)
if (this.registry.dialog.length > 0) availableTypes.push(AugmentationType.DIALOG) availableTypes.push(AugmentationType.COGNITION)
if (this.registry.activation.length > 0) availableTypes.push(AugmentationType.ACTIVATION) if (this.registry.memory.length > 0)
if (this.registry.webSocket.length > 0) availableTypes.push(AugmentationType.WEBSOCKET) availableTypes.push(AugmentationType.MEMORY)
if (this.registry.perception.length > 0)
availableTypes.push(AugmentationType.PERCEPTION)
if (this.registry.dialog.length > 0)
availableTypes.push(AugmentationType.DIALOG)
if (this.registry.activation.length > 0)
availableTypes.push(AugmentationType.ACTIVATION)
if (this.registry.webSocket.length > 0)
availableTypes.push(AugmentationType.WEBSOCKET)
return availableTypes return availableTypes
} }
@ -494,18 +617,18 @@ export class AugmentationPipeline {
...methods: (keyof T)[] ...methods: (keyof T)[]
): augmentation is T { ): augmentation is T {
// First check that the augmentation has all the required base methods // First check that the augmentation has all the required base methods
const baseMethodsExist = [ const baseMethodsExist = ['initialize', 'shutDown', 'getStatus'].every(
'initialize', (method) => typeof (augmentation as any)[method] === 'function'
'shutDown', )
'getStatus'
].every(method => typeof (augmentation as any)[method] === 'function')
if (!baseMethodsExist) { if (!baseMethodsExist) {
return false return false
} }
// Then check that it has all the specific methods for this type // Then check that it has all the specific methods for this type
return methods.every(method => typeof (augmentation as any)[method] === 'function') return methods.every(
(method) => typeof (augmentation as any)[method] === 'function'
)
} }
/** /**
@ -517,21 +640,21 @@ export class AugmentationPipeline {
private shouldUseThreading(options: PipelineOptions): boolean { private shouldUseThreading(options: PipelineOptions): boolean {
// If threading is explicitly disabled, don't use it // If threading is explicitly disabled, don't use it
if (options.disableThreading) { if (options.disableThreading) {
return false; return false
} }
// If threading is explicitly forced, use it if available // If threading is explicitly forced, use it if available
if (options.forceThreading) { if (options.forceThreading) {
return isThreadingAvailable(); return isThreadingAvailable()
} }
// If in THREADED mode, use threading if available // If in THREADED mode, use threading if available
if (options.mode === ExecutionMode.THREADED) { if (options.mode === ExecutionMode.THREADED) {
return isThreadingAvailable(); return isThreadingAvailable()
} }
// Otherwise, don't use threading // Otherwise, don't use threading
return false; return false
} }
/** /**
@ -546,26 +669,34 @@ export class AugmentationPipeline {
private async executeTypedPipeline< private async executeTypedPipeline<
T extends IAugmentation, T extends IAugmentation,
M extends keyof T & string, M extends keyof T & string,
R extends T[M] extends (...args: any[]) => AugmentationResponse<infer U> ? U : never R extends T[M] extends (...args: any[]) => AugmentationResponse<infer U>
? U
: never
>( >(
augmentations: T[], augmentations: T[],
method: M & (T[M] extends (...args: any[]) => any ? M : never), method: M & (T[M] extends (...args: any[]) => any ? M : never),
args: Parameters<Extract<T[M], (...args: any[]) => any>>, args: Parameters<Extract<T[M], (...args: any[]) => any>>,
options: PipelineOptions options: PipelineOptions
): Promise<Promise<{ ): Promise<
success: boolean Promise<{
data: R success: boolean
error?: string data: R
}>[]> { error?: string
}>[]
> {
// Filter out disabled augmentations // Filter out disabled augmentations
const enabledAugmentations = augmentations.filter(aug => aug.enabled !== false) const enabledAugmentations = augmentations.filter(
(aug) => aug.enabled !== false
)
if (enabledAugmentations.length === 0) { if (enabledAugmentations.length === 0) {
return [] return []
} }
// Create a function to execute the method on an augmentation // Create a function to execute the method on an augmentation
const executeMethod = async (augmentation: T): Promise<{ const executeMethod = async (
augmentation: T
): Promise<{
success: boolean success: boolean
data: R data: R
error?: string error?: string
@ -574,21 +705,25 @@ export class AugmentationPipeline {
// Create a timeout promise if a timeout is specified // Create a timeout promise if a timeout is specified
const timeoutPromise = options.timeout const timeoutPromise = options.timeout
? new Promise<{ ? new Promise<{
success: boolean success: boolean
data: R data: R
error?: string error?: string
}>((_, reject) => { }>((_, reject) => {
setTimeout(() => { setTimeout(() => {
reject(new Error(`Timeout executing ${String(method)} on ${augmentation.name}`)) reject(
}, options.timeout) new Error(
}) `Timeout executing ${String(method)} on ${augmentation.name}`
)
)
}, options.timeout)
})
: null : null
// Check if threading should be used // Check if threading should be used
const useThreading = this.shouldUseThreading(options); const useThreading = this.shouldUseThreading(options)
// Execute the method on the augmentation, using threading if appropriate // Execute the method on the augmentation, using threading if appropriate
let methodPromise: Promise<AugmentationResponse<R>>; let methodPromise: Promise<AugmentationResponse<R>>
if (useThreading) { if (useThreading) {
// Execute in a separate thread // Execute in a separate thread
@ -597,19 +732,32 @@ export class AugmentationPipeline {
const workerFn = (...workerArgs: any[]) => { const workerFn = (...workerArgs: any[]) => {
// This function will be stringified and executed in the worker // This function will be stringified and executed in the worker
// It needs to be self-contained // It needs to be self-contained
const augFn = augmentation[method as string] as Function; const augFn = augmentation[method as string] as Function
return augFn.apply(augmentation, workerArgs); return augFn.apply(augmentation, workerArgs)
}; }
methodPromise = executeInThread<AugmentationResponse<R>>(workerFn.toString(), args); methodPromise = executeInThread<AugmentationResponse<R>>(
workerFn.toString(),
args
)
} catch (threadError) { } catch (threadError) {
console.warn(`Failed to execute in thread, falling back to main thread: ${threadError}`); console.warn(
`Failed to execute in thread, falling back to main thread: ${threadError}`
)
// Fall back to executing in the main thread // Fall back to executing in the main thread
methodPromise = Promise.resolve((augmentation[method] as Function)(...args) as AugmentationResponse<R>); methodPromise = Promise.resolve(
(augmentation[method] as Function)(
...args
) as AugmentationResponse<R>
)
} }
} else { } else {
// Execute in the main thread // Execute in the main thread
methodPromise = Promise.resolve((augmentation[method] as Function)(...args) as AugmentationResponse<R>); methodPromise = Promise.resolve(
(augmentation[method] as Function)(
...args
) as AugmentationResponse<R>
)
} }
// Race the method promise against the timeout promise if a timeout is specified // Race the method promise against the timeout promise if a timeout is specified
@ -619,7 +767,10 @@ export class AugmentationPipeline {
return result return result
} catch (error) { } catch (error) {
console.error(`Error executing ${String(method)} on ${augmentation.name}:`, error) console.error(
`Error executing ${String(method)} on ${augmentation.name}:`,
error
)
return { return {
success: false, success: false,
data: null as unknown as R, data: null as unknown as R,
@ -637,26 +788,26 @@ export class AugmentationPipeline {
case ExecutionMode.THREADED: case ExecutionMode.THREADED:
// Execute all augmentations in parallel with threading enabled // Execute all augmentations in parallel with threading enabled
// Force threading for this mode // Force threading for this mode
const threadedOptions = { ...options, forceThreading: true }; const threadedOptions = { ...options, forceThreading: true }
// Create a new executeMethod function that uses the threaded options // Create a new executeMethod function that uses the threaded options
const executeMethodThreaded = async (augmentation: T) => { const executeMethodThreaded = async (augmentation: T) => {
// Save the original options // Save the original options
const originalOptions = options; const originalOptions = options
// Set the options to the threaded options // Set the options to the threaded options
options = threadedOptions; options = threadedOptions
// Execute the method // Execute the method
const result = await executeMethod(augmentation); const result = await executeMethod(augmentation)
// Restore the original options // Restore the original options
options = originalOptions; options = originalOptions
return result; return result
}; }
return enabledAugmentations.map(executeMethodThreaded); return enabledAugmentations.map(executeMethodThreaded)
case ExecutionMode.FIRST_SUCCESS: case ExecutionMode.FIRST_SUCCESS:
// Execute augmentations sequentially until one succeeds // Execute augmentations sequentially until one succeeds