Merge pull request #38854 from nextcloud/enh/llm-api

10 files changed, 568 insertions, 0 deletions

diff --git a/lib/private/AppFramework/Bootstrap/RegistrationContext.php b/lib/private/AppFramework/Bootstrap/RegistrationContext.php
index 8fcafab2d87..5aea2a7a744 100644
--- a/lib/private/AppFramework/Bootstrap/RegistrationContext.php
+++ b/lib/private/AppFramework/Bootstrap/RegistrationContext.php
@@ -33,6 +33,7 @@ use Closure;
 use OCP\Calendar\Resource\IBackend as IResourceBackend;
 use OCP\Calendar\Room\IBackend as IRoomBackend;
 use OCP\Collaboration\Reference\IReferenceProvider;
+use OCP\TextProcessing\IProvider as ITextProcessingProvider;
 use OCP\SpeechToText\ISpeechToTextProvider;
 use OCP\Talk\ITalkBackend;
 use OCP\Translation\ITranslationProvider;
@@ -115,6 +116,9 @@ class RegistrationContext {
 	/** @var ServiceRegistration<ISpeechToTextProvider>[] */
 	private $speechToTextProviders = [];
 
+	/** @var ServiceRegistration<ITextProcessingProvider>[] */
+	private $textProcessingProviders = [];
+
 	/** @var ServiceRegistration<ICustomTemplateProvider>[] */
 	private $templateProviders = [];
 
@@ -262,6 +266,12 @@ class RegistrationContext {
 					$providerClass
 				);
 			}
+			public function registerTextProcessingProvider(string $providerClass): void {
+				$this->context->registerTextProcessingProvider(
+					$this->appId,
+					$providerClass
+				);
+			}
 
 			public function registerTemplateProvider(string $providerClass): void {
 				$this->context->registerTemplateProvider(
@@ -429,6 +439,10 @@ class RegistrationContext {
 		$this->speechToTextProviders[] = new ServiceRegistration($appId, $class);
 	}
 
+	public function registerTextProcessingProvider(string $appId, string $class): void {
+		$this->textProcessingProviders[] = new ServiceRegistration($appId, $class);
+	}
+
 	public function registerTemplateProvider(string $appId, string $class): void {
 		$this->templateProviders[] = new ServiceRegistration($appId, $class);
 	}
@@ -708,6 +722,13 @@ class RegistrationContext {
 	}
 
 	/**
+	 * @return ServiceRegistration<ITextProcessingProvider>[]
+	 */
+	public function getTextProcessingProviders(): array {
+		return $this->textProcessingProviders;
+	}
+
+	/**
 	 * @return ServiceRegistration<ICustomTemplateProvider>[]
 	 */
 	public function getTemplateProviders(): array {
diff --git a/lib/private/Repair.php b/lib/private/Repair.php
index 9c6a6cd00f2..05624a2423a 100644
--- a/lib/private/Repair.php
+++ b/lib/private/Repair.php
@@ -34,6 +34,7 @@
  */
 namespace OC;
 
+use OC\Repair\AddRemoveOldTasksBackgroundJob;
 use OC\Repair\CleanUpAbandonedApps;
 use OCP\AppFramework\QueryException;
 use OCP\AppFramework\Utility\ITimeFactory;
@@ -210,6 +211,7 @@ class Repair implements IOutput {
 			\OCP\Server::get(AddTokenCleanupJob::class),
 			\OCP\Server::get(CleanUpAbandonedApps::class),
 			\OCP\Server::get(AddMissingSecretJob::class),
+			\OCP\Server::get(AddRemoveOldTasksBackgroundJob::class),
 		];
 	}
 
diff --git a/lib/private/Repair/AddRemoveOldTasksBackgroundJob.php b/lib/private/Repair/AddRemoveOldTasksBackgroundJob.php
new file mode 100644
index 00000000000..94ae39f2183
--- /dev/null
+++ b/lib/private/Repair/AddRemoveOldTasksBackgroundJob.php
@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * @copyright 2023 Marcel Klehr <mklehr@gmx.net>
+ *
+ * @author Marcel Klehr <mklehr@gmx.net>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+namespace OC\Repair;
+
+use OC\TextProcessing\RemoveOldTasksBackgroundJob;
+use OCP\BackgroundJob\IJobList;
+use OCP\Migration\IOutput;
+use OCP\Migration\IRepairStep;
+
+class AddRemoveOldTasksBackgroundJob implements IRepairStep {
+	private IJobList $jobList;
+
+	public function __construct(IJobList $jobList) {
+		$this->jobList = $jobList;
+	}
+
+	public function getName(): string {
+		return 'Add language model tasks cleanup job';
+	}
+
+	public function run(IOutput $output) {
+		$this->jobList->add(RemoveOldTasksBackgroundJob::class);
+	}
+}
diff --git a/lib/private/Server.php b/lib/private/Server.php
index f98ee051a32..03c03e1b6ed 100644
--- a/lib/private/Server.php
+++ b/lib/private/Server.php
@@ -1470,6 +1470,8 @@ class Server extends ServerContainer implements IServerContainer {
 
 		$this->registerAlias(IEventSourceFactory::class, EventSourceFactory::class);
 
+		$this->registerAlias(\OCP\TextProcessing\IManager::class, \OC\TextProcessing\Manager::class);
+
 		$this->connectDispatcher();
 	}
 
diff --git a/lib/private/Setup.php b/lib/private/Setup.php
index d847125cc79..0993fe54f47 100644
--- a/lib/private/Setup.php
+++ b/lib/private/Setup.php
@@ -53,6 +53,7 @@ use Exception;
 use InvalidArgumentException;
 use OC\Authentication\Token\PublicKeyTokenProvider;
 use OC\Authentication\Token\TokenCleanupJob;
+use OC\TextProcessing\RemoveOldTasksBackgroundJob;
 use OC\Log\Rotate;
 use OC\Preview\BackgroundCleanupJob;
 use OCP\AppFramework\Utility\ITimeFactory;
@@ -453,6 +454,7 @@ class Setup {
 		$jobList->add(TokenCleanupJob::class);
 		$jobList->add(Rotate::class);
 		$jobList->add(BackgroundCleanupJob::class);
+		$jobList->add(RemoveOldTasksBackgroundJob::class);
 	}
 
 	/**
diff --git a/lib/private/TextProcessing/Db/Task.php b/lib/private/TextProcessing/Db/Task.php
new file mode 100644
index 00000000000..8c2ddb74f1f
--- /dev/null
+++ b/lib/private/TextProcessing/Db/Task.php
@@ -0,0 +1,112 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * @copyright Copyright (c) 2023 Marcel Klehr <mklehr@gmx.net>
+ *
+ * @author Marcel Klehr <mklehr@gmx.net>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+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 $type)
+ * @method string getuserId()
+ * @method setAppId(string $type)
+ * @method string getAppId()
+ * @method setIdentifier(string $type)
+ * @method string getIdentifier()
+ */
+class Task extends Entity {
+	protected $lastUpdated;
+	protected $type;
+	protected $input;
+	protected $output;
+	protected $status;
+	protected $userId;
+	protected $appId;
+	protected $identifier;
+
+	/**
+	 * @var string[]
+	 */
+	public static array $columns = ['id', 'last_updated', 'type', 'input', 'output', 'status', 'user_id', 'app_id', 'identifier'];
+
+	/**
+	 * @var string[]
+	 */
+	public static array $fields = ['id', 'lastUpdated', 'type', 'input', 'output', 'status', 'userId', 'appId', 'identifier'];
+
+
+	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');
+	}
+
+	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(),
+		]);
+		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());
+		return $task;
+	}
+}
diff --git a/lib/private/TextProcessing/Db/TaskMapper.php b/lib/private/TextProcessing/Db/TaskMapper.php
new file mode 100644
index 00000000000..508f3fdf3b8
--- /dev/null
+++ b/lib/private/TextProcessing/Db/TaskMapper.php
@@ -0,0 +1,78 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * @copyright Copyright (c) 2023 Marcel Klehr <mklehr@gmx.net>
+ *
+ * @author Marcel Klehr <mklehr@gmx.net>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+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, 'llm_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 $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..f52482bbb32
--- /dev/null
+++ b/lib/private/TextProcessing/Manager.php
@@ -0,0 +1,182 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * @copyright Copyright (c) 2023 Marcel Klehr <mklehr@gmx.net>
+ *
+ * @author Marcel Klehr <mklehr@gmx.net>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+namespace OC\TextProcessing;
+
+use OC\AppFramework\Bootstrap\Coordinator;
+use OC\TextProcessing\Db\Task as DbTask;
+use OCP\TextProcessing\Task as OCPTask;
+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\IServerContainer;
+use OCP\TextProcessing\IManager;
+use OCP\TextProcessing\IProvider;
+use OCP\PreConditionNotMetException;
+use Psr\Log\LoggerInterface;
+use RuntimeException;
+use Throwable;
+
+class Manager implements IManager {
+	/** @var ?IProvider[] */
+	private ?array $providers = null;
+
+	public function __construct(
+		private IServerContainer $serverContainer,
+		private Coordinator $coordinator,
+		private LoggerInterface $logger,
+		private IJobList $jobList,
+		private TaskMapper $taskMapper,
+	) {
+	}
+
+	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 {
+		$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;
+		}
+		return array_keys($tasks);
+	}
+
+	public function canHandleTask(OCPTask $task): bool {
+		return in_array($task->getType(), $this->getAvailableTaskTypes());
+	}
+
+	/**
+	 * @inheritDoc
+	 */
+	public function runTask(OCPTask $task): string {
+		if (!$this->canHandleTask($task)) {
+			throw new PreConditionNotMetException('No text processing provider is installed that can handle this task');
+		}
+		foreach ($this->getProviders() as $provider) {
+			if (!$task->canUseProvider($provider)) {
+				continue;
+			}
+			try {
+				$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));
+				}
+				$output = $task->visitProvider($provider);
+				$task->setOutput($output);
+				$task->setStatus(OCPTask::STATUS_SUCCESSFUL);
+				$this->taskMapper->update(DbTask::fromPublicTask($task));
+				return $output;
+			} catch (\RuntimeException $e) {
+				$this->logger->info('LanguageModel call using provider ' . $provider->getName() . ' failed', ['exception' => $e]);
+				$task->setStatus(OCPTask::STATUS_FAILED);
+				$this->taskMapper->update(DbTask::fromPublicTask($task));
+				throw $e;
+			} catch (\Throwable $e) {
+				$this->logger->info('LanguageModel call using provider ' . $provider->getName() . ' failed', ['exception' => $e]);
+				$task->setStatus(OCPTask::STATUS_FAILED);
+				$this->taskMapper->update(DbTask::fromPublicTask($task));
+				throw new RuntimeException('LanguageModel call using provider ' . $provider->getName() . ' failed: ' . $e->getMessage(), 0, $e);
+			}
+		}
+
+		throw new RuntimeException('Could not run task');
+	}
+
+	/**
+	 * @inheritDoc
+	 * @throws Exception
+	 */
+	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);
+		$taskEntity = DbTask::fromPublicTask($task);
+		$this->taskMapper->insert($taskEntity);
+		$task->setId($taskEntity->getId());
+		$this->jobList->add(TaskBackgroundJob::class, [
+			'taskId' => $task->getId()
+		]);
+	}
+
+	/**
+	 * @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);
+		}
+	}
+}
diff --git a/lib/private/TextProcessing/RemoveOldTasksBackgroundJob.php b/lib/private/TextProcessing/RemoveOldTasksBackgroundJob.php
new file mode 100644
index 00000000000..89d329acfbb
--- /dev/null
+++ b/lib/private/TextProcessing/RemoveOldTasksBackgroundJob.php
@@ -0,0 +1,59 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * @copyright Copyright (c) 2023 Marcel Klehr <mklehr@gmx.net>
+ *
+ * @author Marcel Klehr <mklehr@gmx.net>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+
+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 * 50 * 24 * 7; // 1 week
+
+	public function __construct(
+		ITimeFactory $timeFactory,
+		private TaskMapper $taskMapper,
+		private LoggerInterface $logger,
+
+	) {
+		parent::__construct($timeFactory);
+		$this->setInterval(60 * 60 * 24);
+	}
+
+	/**
+	 * @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..4c24b3e531f
--- /dev/null
+++ b/lib/private/TextProcessing/TaskBackgroundJob.php
@@ -0,0 +1,63 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * @copyright Copyright (c) 2023 Marcel Klehr <mklehr@gmx.net>
+ *
+ * @author Marcel Klehr <mklehr@gmx.net>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+
+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);
+	}
+}