diff options
Diffstat (limited to 'lib/private/TextProcessing')
| -rw-r--r-- | lib/private/TextProcessing/Db/Task.php | 102 | ||||
| -rw-r--r-- | lib/private/TextProcessing/Db/TaskMapper.php | 102 | ||||
| -rw-r--r-- | lib/private/TextProcessing/Manager.php | 365 | ||||
| -rw-r--r-- | lib/private/TextProcessing/RemoveOldTasksBackgroundJob.php | 43 | ||||
| -rw-r--r-- | lib/private/TextProcessing/TaskBackgroundJob.php | 47 |
5 files changed, 659 insertions, 0 deletions
diff --git a/lib/private/TextProcessing/Db/Task.php b/lib/private/TextProcessing/Db/Task.php new file mode 100644 index 00000000000..d4ebc19e74a --- /dev/null +++ b/lib/private/TextProcessing/Db/Task.php @@ -0,0 +1,102 @@ +<?php + +declare(strict_types=1); + +/** + * SPDX-FileCopyrightText: 2023 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +namespace OC\TextProcessing\Db; + +use OCP\AppFramework\Db\Entity; +use OCP\TextProcessing\Task as OCPTask; + +/** + * @method setType(string $type) + * @method string getType() + * @method setLastUpdated(int $lastUpdated) + * @method int getLastUpdated() + * @method setInput(string $type) + * @method string getInput() + * @method setOutput(string $type) + * @method string getOutput() + * @method setStatus(int $type) + * @method int getStatus() + * @method setUserId(?string $userId) + * @method string|null getUserId() + * @method setAppId(string $type) + * @method string getAppId() + * @method setIdentifier(string $identifier) + * @method string getIdentifier() + * @method setCompletionExpectedAt(null|\DateTime $completionExpectedAt) + * @method null|\DateTime getCompletionExpectedAt() + */ +class Task extends Entity { + protected $lastUpdated; + protected $type; + protected $input; + protected $output; + protected $status; + protected $userId; + protected $appId; + protected $identifier; + protected $completionExpectedAt; + + /** + * @var string[] + */ + public static array $columns = ['id', 'last_updated', 'type', 'input', 'output', 'status', 'user_id', 'app_id', 'identifier', 'completion_expected_at']; + + /** + * @var string[] + */ + public static array $fields = ['id', 'lastUpdated', 'type', 'input', 'output', 'status', 'userId', 'appId', 'identifier', 'completionExpectedAt']; + + + public function __construct() { + // add types in constructor + $this->addType('id', 'integer'); + $this->addType('lastUpdated', 'integer'); + $this->addType('type', 'string'); + $this->addType('input', 'string'); + $this->addType('output', 'string'); + $this->addType('status', 'integer'); + $this->addType('userId', 'string'); + $this->addType('appId', 'string'); + $this->addType('identifier', 'string'); + $this->addType('completionExpectedAt', 'datetime'); + } + + public function toRow(): array { + return array_combine(self::$columns, array_map(function ($field) { + return $this->{'get' . ucfirst($field)}(); + }, self::$fields)); + } + + public static function fromPublicTask(OCPTask $task): Task { + /** @var Task $task */ + $task = Task::fromParams([ + 'id' => $task->getId(), + 'type' => $task->getType(), + 'lastUpdated' => time(), + 'status' => $task->getStatus(), + 'input' => $task->getInput(), + 'output' => $task->getOutput(), + 'userId' => $task->getUserId(), + 'appId' => $task->getAppId(), + 'identifier' => $task->getIdentifier(), + 'completionExpectedAt' => $task->getCompletionExpectedAt(), + ]); + return $task; + } + + public function toPublicTask(): OCPTask { + $task = new OCPTask($this->getType(), $this->getInput(), $this->getAppId(), $this->getuserId(), $this->getIdentifier()); + $task->setId($this->getId()); + $task->setStatus($this->getStatus()); + $task->setOutput($this->getOutput()); + $task->setCompletionExpectedAt($this->getCompletionExpectedAt()); + return $task; + } +} diff --git a/lib/private/TextProcessing/Db/TaskMapper.php b/lib/private/TextProcessing/Db/TaskMapper.php new file mode 100644 index 00000000000..b03e5833958 --- /dev/null +++ b/lib/private/TextProcessing/Db/TaskMapper.php @@ -0,0 +1,102 @@ +<?php + +declare(strict_types=1); + +/** + * SPDX-FileCopyrightText: 2023 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +namespace OC\TextProcessing\Db; + +use OCP\AppFramework\Db\DoesNotExistException; +use OCP\AppFramework\Db\Entity; +use OCP\AppFramework\Db\MultipleObjectsReturnedException; +use OCP\AppFramework\Db\QBMapper; +use OCP\AppFramework\Utility\ITimeFactory; +use OCP\DB\Exception; +use OCP\IDBConnection; + +/** + * @extends QBMapper<Task> + */ +class TaskMapper extends QBMapper { + public function __construct( + IDBConnection $db, + private ITimeFactory $timeFactory, + ) { + parent::__construct($db, 'textprocessing_tasks', Task::class); + } + + /** + * @param int $id + * @return Task + * @throws Exception + * @throws DoesNotExistException + * @throws MultipleObjectsReturnedException + */ + public function find(int $id): Task { + $qb = $this->db->getQueryBuilder(); + $qb->select(Task::$columns) + ->from($this->tableName) + ->where($qb->expr()->eq('id', $qb->createPositionalParameter($id))); + return $this->findEntity($qb); + } + + /** + * @param int $id + * @param string|null $userId + * @return Task + * @throws DoesNotExistException + * @throws Exception + * @throws MultipleObjectsReturnedException + */ + public function findByIdAndUser(int $id, ?string $userId): Task { + $qb = $this->db->getQueryBuilder(); + $qb->select(Task::$columns) + ->from($this->tableName) + ->where($qb->expr()->eq('id', $qb->createPositionalParameter($id))); + if ($userId === null) { + $qb->andWhere($qb->expr()->isNull('user_id')); + } else { + $qb->andWhere($qb->expr()->eq('user_id', $qb->createPositionalParameter($userId))); + } + return $this->findEntity($qb); + } + + /** + * @param string $userId + * @param string $appId + * @param string|null $identifier + * @return array + * @throws Exception + */ + public function findUserTasksByApp(string $userId, string $appId, ?string $identifier = null): array { + $qb = $this->db->getQueryBuilder(); + $qb->select(Task::$columns) + ->from($this->tableName) + ->where($qb->expr()->eq('user_id', $qb->createPositionalParameter($userId))) + ->andWhere($qb->expr()->eq('app_id', $qb->createPositionalParameter($appId))); + if ($identifier !== null) { + $qb->andWhere($qb->expr()->eq('identifier', $qb->createPositionalParameter($identifier))); + } + return $this->findEntities($qb); + } + + /** + * @param int $timeout + * @return int the number of deleted tasks + * @throws Exception + */ + public function deleteOlderThan(int $timeout): int { + $qb = $this->db->getQueryBuilder(); + $qb->delete($this->tableName) + ->where($qb->expr()->lt('last_updated', $qb->createPositionalParameter(time() - $timeout))); + return $qb->executeStatement(); + } + + public function update(Entity $entity): Entity { + $entity->setLastUpdated($this->timeFactory->now()->getTimestamp()); + return parent::update($entity); + } +} diff --git a/lib/private/TextProcessing/Manager.php b/lib/private/TextProcessing/Manager.php new file mode 100644 index 00000000000..3fe45ce55ec --- /dev/null +++ b/lib/private/TextProcessing/Manager.php @@ -0,0 +1,365 @@ +<?php + +declare(strict_types=1); + +/** + * SPDX-FileCopyrightText: 2023 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + +namespace OC\TextProcessing; + +use OC\AppFramework\Bootstrap\Coordinator; +use OC\TextProcessing\Db\Task as DbTask; +use OC\TextProcessing\Db\TaskMapper; +use OCP\AppFramework\Db\DoesNotExistException; +use OCP\AppFramework\Db\MultipleObjectsReturnedException; +use OCP\BackgroundJob\IJobList; +use OCP\Common\Exception\NotFoundException; +use OCP\DB\Exception; +use OCP\IConfig; +use OCP\IServerContainer; +use OCP\PreConditionNotMetException; +use OCP\TaskProcessing\IManager as TaskProcessingIManager; +use OCP\TaskProcessing\TaskTypes\TextToText; +use OCP\TaskProcessing\TaskTypes\TextToTextHeadline; +use OCP\TaskProcessing\TaskTypes\TextToTextSummary; +use OCP\TaskProcessing\TaskTypes\TextToTextTopics; +use OCP\TextProcessing\Exception\TaskFailureException; +use OCP\TextProcessing\FreePromptTaskType; +use OCP\TextProcessing\HeadlineTaskType; +use OCP\TextProcessing\IManager; +use OCP\TextProcessing\IProvider; +use OCP\TextProcessing\IProviderWithExpectedRuntime; +use OCP\TextProcessing\IProviderWithId; +use OCP\TextProcessing\SummaryTaskType; +use OCP\TextProcessing\Task; +use OCP\TextProcessing\Task as OCPTask; +use OCP\TextProcessing\TopicsTaskType; +use Psr\Log\LoggerInterface; +use RuntimeException; +use Throwable; + +class Manager implements IManager { + /** @var ?IProvider[] */ + private ?array $providers = null; + + private static array $taskProcessingCompatibleTaskTypes = [ + FreePromptTaskType::class => TextToText::ID, + HeadlineTaskType::class => TextToTextHeadline::ID, + SummaryTaskType::class => TextToTextSummary::ID, + TopicsTaskType::class => TextToTextTopics::ID, + ]; + + public function __construct( + private IServerContainer $serverContainer, + private Coordinator $coordinator, + private LoggerInterface $logger, + private IJobList $jobList, + private TaskMapper $taskMapper, + private IConfig $config, + private TaskProcessingIManager $taskProcessingManager, + ) { + } + + public function getProviders(): array { + $context = $this->coordinator->getRegistrationContext(); + if ($context === null) { + return []; + } + + if ($this->providers !== null) { + return $this->providers; + } + + $this->providers = []; + + foreach ($context->getTextProcessingProviders() as $providerServiceRegistration) { + $class = $providerServiceRegistration->getService(); + try { + $this->providers[$class] = $this->serverContainer->get($class); + } catch (Throwable $e) { + $this->logger->error('Failed to load Text processing provider ' . $class, [ + 'exception' => $e, + ]); + } + } + + return $this->providers; + } + + public function hasProviders(): bool { + // check if task processing equivalent types are available + $taskTaskTypes = $this->taskProcessingManager->getAvailableTaskTypes(); + foreach (self::$taskProcessingCompatibleTaskTypes as $textTaskTypeClass => $taskTaskTypeId) { + if (isset($taskTaskTypes[$taskTaskTypeId])) { + return true; + } + } + + $context = $this->coordinator->getRegistrationContext(); + if ($context === null) { + return false; + } + return count($context->getTextProcessingProviders()) > 0; + } + + /** + * @inheritDoc + */ + public function getAvailableTaskTypes(): array { + $tasks = []; + foreach ($this->getProviders() as $provider) { + $tasks[$provider->getTaskType()] = true; + } + + // check if task processing equivalent types are available + $taskTaskTypes = $this->taskProcessingManager->getAvailableTaskTypes(); + foreach (self::$taskProcessingCompatibleTaskTypes as $textTaskTypeClass => $taskTaskTypeId) { + if (isset($taskTaskTypes[$taskTaskTypeId])) { + $tasks[$textTaskTypeClass] = true; + } + } + + return array_keys($tasks); + } + + public function canHandleTask(OCPTask $task): bool { + return in_array($task->getType(), $this->getAvailableTaskTypes()); + } + + /** + * @inheritDoc + */ + public function runTask(OCPTask $task): string { + // try to run a task processing task if possible + $taskTypeClass = $task->getType(); + if (isset(self::$taskProcessingCompatibleTaskTypes[$taskTypeClass]) && isset($this->taskProcessingManager->getAvailableTaskTypes()[self::$taskProcessingCompatibleTaskTypes[$taskTypeClass]])) { + try { + $taskProcessingTaskTypeId = self::$taskProcessingCompatibleTaskTypes[$taskTypeClass]; + $taskProcessingTask = new \OCP\TaskProcessing\Task( + $taskProcessingTaskTypeId, + ['input' => $task->getInput()], + $task->getAppId(), + $task->getUserId(), + $task->getIdentifier(), + ); + + $task->setStatus(OCPTask::STATUS_RUNNING); + if ($task->getId() === null) { + $taskEntity = $this->taskMapper->insert(DbTask::fromPublicTask($task)); + $task->setId($taskEntity->getId()); + } else { + $this->taskMapper->update(DbTask::fromPublicTask($task)); + } + $this->logger->debug('Running a TextProcessing (' . $taskTypeClass . ') task with TaskProcessing'); + $taskProcessingResultTask = $this->taskProcessingManager->runTask($taskProcessingTask); + if ($taskProcessingResultTask->getStatus() === \OCP\TaskProcessing\Task::STATUS_SUCCESSFUL) { + $output = $taskProcessingResultTask->getOutput(); + if (isset($output['output']) && is_string($output['output'])) { + $task->setOutput($output['output']); + $task->setStatus(OCPTask::STATUS_SUCCESSFUL); + $this->taskMapper->update(DbTask::fromPublicTask($task)); + return $output['output']; + } + } + } catch (\Throwable $e) { + $this->logger->error('TextProcessing to TaskProcessing failed', ['exception' => $e]); + $task->setStatus(OCPTask::STATUS_FAILED); + $this->taskMapper->update(DbTask::fromPublicTask($task)); + throw new TaskFailureException('TextProcessing to TaskProcessing failed: ' . $e->getMessage(), 0, $e); + } + $task->setStatus(OCPTask::STATUS_FAILED); + $this->taskMapper->update(DbTask::fromPublicTask($task)); + throw new TaskFailureException('Could not run task'); + } + + // try to run the text processing task + if (!$this->canHandleTask($task)) { + throw new PreConditionNotMetException('No text processing provider is installed that can handle this task'); + } + $providers = $this->getPreferredProviders($task); + + foreach ($providers as $provider) { + try { + $task->setStatus(OCPTask::STATUS_RUNNING); + if ($provider instanceof IProviderWithExpectedRuntime) { + $completionExpectedAt = new \DateTime('now'); + $completionExpectedAt->add(new \DateInterval('PT' . $provider->getExpectedRuntime() . 'S')); + $task->setCompletionExpectedAt($completionExpectedAt); + } + if ($task->getId() === null) { + $taskEntity = $this->taskMapper->insert(DbTask::fromPublicTask($task)); + $task->setId($taskEntity->getId()); + } else { + $this->taskMapper->update(DbTask::fromPublicTask($task)); + } + $output = $task->visitProvider($provider); + $task->setOutput($output); + $task->setStatus(OCPTask::STATUS_SUCCESSFUL); + $this->taskMapper->update(DbTask::fromPublicTask($task)); + return $output; + } catch (\Throwable $e) { + $this->logger->error('LanguageModel call using provider ' . $provider->getName() . ' failed', ['exception' => $e]); + $task->setStatus(OCPTask::STATUS_FAILED); + $this->taskMapper->update(DbTask::fromPublicTask($task)); + throw new TaskFailureException('LanguageModel call using provider ' . $provider->getName() . ' failed: ' . $e->getMessage(), 0, $e); + } + } + + $task->setStatus(OCPTask::STATUS_FAILED); + $this->taskMapper->update(DbTask::fromPublicTask($task)); + throw new TaskFailureException('Could not run task'); + } + + /** + * @inheritDoc + */ + public function scheduleTask(OCPTask $task): void { + if (!$this->canHandleTask($task)) { + throw new PreConditionNotMetException('No LanguageModel provider is installed that can handle this task'); + } + $task->setStatus(OCPTask::STATUS_SCHEDULED); + $providers = $this->getPreferredProviders($task); + $equivalentTaskProcessingTypeAvailable = ( + isset(self::$taskProcessingCompatibleTaskTypes[$task->getType()]) + && isset($this->taskProcessingManager->getAvailableTaskTypes()[self::$taskProcessingCompatibleTaskTypes[$task->getType()]]) + ); + if (count($providers) === 0 && !$equivalentTaskProcessingTypeAvailable) { + throw new PreConditionNotMetException('No LanguageModel provider is installed that can handle this task'); + } + [$provider,] = $providers; + if ($provider instanceof IProviderWithExpectedRuntime) { + $completionExpectedAt = new \DateTime('now'); + $completionExpectedAt->add(new \DateInterval('PT' . $provider->getExpectedRuntime() . 'S')); + $task->setCompletionExpectedAt($completionExpectedAt); + } + $taskEntity = DbTask::fromPublicTask($task); + $this->taskMapper->insert($taskEntity); + $task->setId($taskEntity->getId()); + $this->jobList->add(TaskBackgroundJob::class, [ + 'taskId' => $task->getId() + ]); + } + + /** + * @inheritDoc + */ + public function runOrScheduleTask(OCPTask $task): bool { + if (!$this->canHandleTask($task)) { + throw new PreConditionNotMetException('No LanguageModel provider is installed that can handle this task'); + } + [$provider,] = $this->getPreferredProviders($task); + $maxExecutionTime = (int)ini_get('max_execution_time'); + // Offload the task to a background job if the expected runtime of the likely provider is longer than 80% of our max execution time + // or if the provider doesn't provide a getExpectedRuntime() method + if (!$provider instanceof IProviderWithExpectedRuntime || $provider->getExpectedRuntime() > $maxExecutionTime * 0.8) { + $this->scheduleTask($task); + return false; + } + $this->runTask($task); + return true; + } + + /** + * @inheritDoc + */ + public function deleteTask(Task $task): void { + $taskEntity = DbTask::fromPublicTask($task); + $this->taskMapper->delete($taskEntity); + $this->jobList->remove(TaskBackgroundJob::class, [ + 'taskId' => $task->getId() + ]); + } + + /** + * Get a task from its id + * + * @param int $id The id of the task + * @return OCPTask + * @throws RuntimeException If the query failed + * @throws NotFoundException If the task could not be found + */ + public function getTask(int $id): OCPTask { + try { + $taskEntity = $this->taskMapper->find($id); + return $taskEntity->toPublicTask(); + } catch (DoesNotExistException $e) { + throw new NotFoundException('Could not find task with the provided id'); + } catch (MultipleObjectsReturnedException $e) { + throw new RuntimeException('Could not uniquely identify task with given id', 0, $e); + } catch (Exception $e) { + throw new RuntimeException('Failure while trying to find task by id: ' . $e->getMessage(), 0, $e); + } + } + + /** + * Get a task from its user id and task id + * If userId is null, this can only get a task that was scheduled anonymously + * + * @param int $id The id of the task + * @param string|null $userId The user id that scheduled the task + * @return OCPTask + * @throws RuntimeException If the query failed + * @throws NotFoundException If the task could not be found + */ + public function getUserTask(int $id, ?string $userId): OCPTask { + try { + $taskEntity = $this->taskMapper->findByIdAndUser($id, $userId); + return $taskEntity->toPublicTask(); + } catch (DoesNotExistException $e) { + throw new NotFoundException('Could not find task with the provided id and user id'); + } catch (MultipleObjectsReturnedException $e) { + throw new RuntimeException('Could not uniquely identify task with given id and user id', 0, $e); + } catch (Exception $e) { + throw new RuntimeException('Failure while trying to find task by id and user id: ' . $e->getMessage(), 0, $e); + } + } + + /** + * Get a list of tasks scheduled by a specific user for a specific app + * and optionally with a specific identifier. + * This cannot be used to get anonymously scheduled tasks + * + * @param string $userId + * @param string $appId + * @param string|null $identifier + * @return array + */ + public function getUserTasksByApp(string $userId, string $appId, ?string $identifier = null): array { + try { + $taskEntities = $this->taskMapper->findUserTasksByApp($userId, $appId, $identifier); + return array_map(static function (DbTask $taskEntity) { + return $taskEntity->toPublicTask(); + }, $taskEntities); + } catch (Exception $e) { + throw new RuntimeException('Failure while trying to find tasks by appId and identifier: ' . $e->getMessage(), 0, $e); + } + } + + /** + * @param OCPTask $task + * @return IProvider[] + */ + public function getPreferredProviders(OCPTask $task): array { + $providers = $this->getProviders(); + $json = $this->config->getAppValue('core', 'ai.textprocessing_provider_preferences', ''); + if ($json !== '') { + $preferences = json_decode($json, true); + if (isset($preferences[$task->getType()])) { + // If a preference for this task type is set, move the preferred provider to the start + $provider = current(array_values(array_filter($providers, function ($provider) use ($preferences, $task) { + if ($provider instanceof IProviderWithId) { + return $provider->getId() === $preferences[$task->getType()]; + } + return $provider::class === $preferences[$task->getType()]; + }))); + if ($provider !== false) { + $providers = array_filter($providers, fn ($p) => $p !== $provider); + array_unshift($providers, $provider); + } + } + } + return array_values(array_filter($providers, fn (IProvider $provider) => $task->canUseProvider($provider))); + } +} diff --git a/lib/private/TextProcessing/RemoveOldTasksBackgroundJob.php b/lib/private/TextProcessing/RemoveOldTasksBackgroundJob.php new file mode 100644 index 00000000000..be5e4127d08 --- /dev/null +++ b/lib/private/TextProcessing/RemoveOldTasksBackgroundJob.php @@ -0,0 +1,43 @@ +<?php + +declare(strict_types=1); + +/** + * SPDX-FileCopyrightText: 2023 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + + +namespace OC\TextProcessing; + +use OC\TextProcessing\Db\TaskMapper; +use OCP\AppFramework\Utility\ITimeFactory; +use OCP\BackgroundJob\TimedJob; +use OCP\DB\Exception; +use Psr\Log\LoggerInterface; + +class RemoveOldTasksBackgroundJob extends TimedJob { + public const MAX_TASK_AGE_SECONDS = 60 * 60 * 24 * 7; // 1 week + + public function __construct( + ITimeFactory $timeFactory, + private TaskMapper $taskMapper, + private LoggerInterface $logger, + ) { + parent::__construct($timeFactory); + $this->setInterval(60 * 60 * 24); + $this->setTimeSensitivity(self::TIME_INSENSITIVE); + } + + /** + * @param mixed $argument + * @inheritDoc + */ + protected function run($argument) { + try { + $this->taskMapper->deleteOlderThan(self::MAX_TASK_AGE_SECONDS); + } catch (Exception $e) { + $this->logger->warning('Failed to delete stale language model tasks', ['exception' => $e]); + } + } +} diff --git a/lib/private/TextProcessing/TaskBackgroundJob.php b/lib/private/TextProcessing/TaskBackgroundJob.php new file mode 100644 index 00000000000..2942fd7d0aa --- /dev/null +++ b/lib/private/TextProcessing/TaskBackgroundJob.php @@ -0,0 +1,47 @@ +<?php + +declare(strict_types=1); + +/** + * SPDX-FileCopyrightText: 2023 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: AGPL-3.0-or-later + */ + + +namespace OC\TextProcessing; + +use OCP\AppFramework\Utility\ITimeFactory; +use OCP\BackgroundJob\QueuedJob; +use OCP\EventDispatcher\IEventDispatcher; +use OCP\TextProcessing\Events\TaskFailedEvent; +use OCP\TextProcessing\Events\TaskSuccessfulEvent; +use OCP\TextProcessing\IManager; + +class TaskBackgroundJob extends QueuedJob { + public function __construct( + ITimeFactory $timeFactory, + private IManager $textProcessingManager, + private IEventDispatcher $eventDispatcher, + ) { + parent::__construct($timeFactory); + // We want to avoid overloading the machine with these jobs + // so we only allow running one job at a time + $this->setAllowParallelRuns(false); + } + + /** + * @param array{taskId: int} $argument + * @inheritDoc + */ + protected function run($argument) { + $taskId = $argument['taskId']; + $task = $this->textProcessingManager->getTask($taskId); + try { + $this->textProcessingManager->runTask($task); + $event = new TaskSuccessfulEvent($task); + } catch (\Throwable $e) { + $event = new TaskFailedEvent($task, $e->getMessage()); + } + $this->eventDispatcher->dispatchTyped($event); + } +} |