[Platform] Introduce asynchronous jobs and move MiniMax polling out of the result converter

UPGRADE.md

@@ -1,3 +1,43 @@
+UPGRADE FROM 0.13 to 0.14
+=========================
+
+Platform
+--------
+
+ * The MiniMax bridge no longer blocks inside its result converter while an asynchronous task runs.
+   Video generation and asynchronous speech synthesis (`async: true`) now return a `Result\JobResult`
+   carrying a serializable `Job\JobHandle`, and waiting for the job became explicit. Reading the
+   result directly through `asBinary()`/`asFile()` therefore throws an `UnexpectedResultTypeException`:
+
+   ```diff
+   +use Symfony\AI\Platform\Job\JobRunner;
+   +
+   -$result = $platform->invoke('MiniMax-Hailuo-02', $prompt, ['duration' => 6]);
+   -$result->asFile('video.mp4');
+   +$provider = MiniMaxFactory::createProvider($apiKey);
+   +$handle = $provider->invoke('MiniMax-Hailuo-02', $prompt, ['duration' => 6])->asJob();
+   +
+   +$result = (new JobRunner())->wait($provider->getJobClient(), $handle);
+   +$result->asFile('video.mp4');
+   ```
+
+   The former budgets — 120 seconds for audio, 600 for video — are now stated on the handle rather
+   than baked into the bridge, so waiting for a job needs no knowledge of the provider's timings.
+   Pass `maxDuration` (in seconds) to `wait()` to bound a single call instead, for instance inside a
+   web request. A job that does not finish in time raises a `JobTimeoutException` that carries the
+   handle, so the job can be picked up later instead of being lost, including from another process:
+   the client resolving it comes from `ProviderInterface::getJobClient()`, or from
+   `Bridge\MiniMax\Factory::createJobClient()` in a worker that only resolves jobs.
+
+   Accordingly, `Bridge\MiniMax\MiniMaxResultConverter` no longer takes an HTTP client, API key,
+   endpoint or clock; polling moved to the new `Bridge\MiniMax\MiniMaxJobClient`. Code building the
+   bridge through `Bridge\MiniMax\Factory` is unaffected.
+
+   ```diff
+   -$converter = new MiniMaxResultConverter($httpClient, $apiKey, $endpoint, $clock);
+   +$converter = new MiniMaxResultConverter();
+   ```
+
 UPGRADE FROM 0.12 to 0.13
 =========================
 

docs/components/platform.rst

@@ -1039,6 +1039,131 @@ Code Examples
 * `PDF Input with GPT`_
 * `PDF Input with Claude`_
 
+Asynchronous Jobs
+-----------------
+
+Some work cannot be answered in the same request it was asked for: video generation runs for minutes,
+and several providers offer asynchronous or batch endpoints that accept a request now and produce the
+result later. Those providers answer the invocation with a job identifier, so ``invoke()`` returns a
+:class:`Symfony\\AI\\Platform\\Result\\JobResult` whose content is a
+:class:`Symfony\\AI\\Platform\\Job\\JobHandle`::
+
+    $handle = $platform->invoke('MiniMax-Hailuo-02', new Text('A cat playing the piano'), [
+        'duration' => 6,
+    ])->asJob();
+
+The handle holds no connection and no client, only what is needed to ask the provider about that job
+again, so it can be stored and picked up somewhere else entirely::
+
+    // in the process that started the job
+    $repository->save($handle->getId(), $handle->toString());
+
+    // in a worker, possibly much later
+    $handle = JobHandle::fromString($repository->load($id));
+
+    if ($jobClient->getStatus($handle)->is(JobStateCase::SUCCEEDED)) {
+        $result = $jobClient->getResult($handle);
+    }
+
+:method:`Symfony\\AI\\Platform\\Job\\JobHandle::toString` and its ``fromString`` counterpart cover
+storage that holds a single column; ``toArray()``/``fromArray()`` cover the structured case, and the
+handle is ``JsonSerializable`` so it also drops straight into a Messenger message. Note that the job
+identifier is the provider's, so an application storing handles of several providers keys them by
+provider and id rather than by id alone.
+
+The client resolving a job belongs to the provider that issued it, so a provider hands it out through
+:method:`Symfony\\AI\\Platform\\Job\\JobProviderInterface::getJobClient`, and a bridge builds one
+directly for a process that only resolves jobs and never invokes anything::
+
+    $jobClient = $provider->getJobClient();                 // the provider that started the job
+    $jobClient = MiniMaxFactory::createJobClient($apiKey);  // or straight from the bridge, in a worker
+
+:method:`Symfony\\AI\\Platform\\Job\\JobClientInterface::getStatus` performs exactly one request and
+never sleeps. To simply block until the job is done, hand it to a
+:class:`Symfony\\AI\\Platform\\Job\\JobRunner`, which owns the polling loop::
+
+    use Symfony\AI\Platform\Job\JobRunner;
+
+    $result = (new JobRunner())->wait($jobClient, $handle);
+
+    $result->asFile('video.mp4');
+
+What the runner hands back is a :class:`Symfony\\AI\\Platform\\Result\\DeferredResult`, the same
+thing ``invoke()`` returns, so finishing a job reads like any other invocation instead of leaving
+you to narrow a bare ``ResultInterface`` yourself.
+
+How long the work takes and how long you are willing to wait for it are two different questions. The
+first is the provider's: a bridge that knows its timings (MiniMax video generation runs for minutes
+where speech synthesis takes seconds) states it on the handle, and the runner honours it, so a
+caller who knows nothing about the provider still waits the right amount.
+
+The second is yours, and it usually belongs to the call rather than to the runner: the same job may
+be given ten minutes in a worker and five seconds inside a web request. Say so per call, in seconds::
+
+    $result = $runner->wait($jobClient, $handle, maxDuration: 5);
+
+A budget passed to the runner's constructor applies to every job it waits for and sits between the
+two: it overrules what a job asks for, and a single call overrules it in turn.
+
+In a Symfony application a runner using the application clock is available as
+``ai.platform.job_runner`` and autowired through :class:`Symfony\\AI\\Platform\\Job\\JobRunner`. It
+carries no budget of its own, so the same shared service serves a job finishing in seconds and one
+running for minutes. Each job-capable platform also registers its client as
+``ai.platform.job_client.<name>``, autowired by argument name::
+
+    public function __construct(
+        private JobRunner $jobRunner,
+        private JobClientInterface $minimaxJobClient,
+    ) {
+    }
+
+    public function __invoke(JobHandle $handle): void
+    {
+        // trust the job
+        $this->jobRunner->wait($this->minimaxJobClient, $handle);
+
+        // or bound it to what a request can afford
+        $this->jobRunner->wait($this->minimaxJobClient, $handle, maxDuration: 5);
+    }
+
+An application holding handles of several providers picks the client by the name the handle carries,
+from a locator over the ``ai.platform.job_client`` tag::
+
+    public function __construct(
+        #[AutowireLocator('ai.platform.job_client', indexAttribute: 'key')]
+        private ContainerInterface $jobClients,
+        private JobRunner $jobRunner,
+    ) {
+    }
+
+    public function __invoke(JobHandle $handle): void
+    {
+        $this->jobRunner->wait($this->jobClients->get($handle->getProvider()), $handle);
+    }
+
+The runner throws a :class:`Symfony\\AI\\Platform\\Exception\\JobFailedException` when the provider
+ends the job without a result, and a :class:`Symfony\\AI\\Platform\\Exception\\JobTimeoutException`
+when the budget runs out while the job is still going. The latter carries the handle, so the job can
+be handed on rather than lost.
+
+Providers spell their states differently, so a bridge maps them onto a
+:class:`Symfony\\AI\\Platform\\Job\\JobStateCase` while
+:method:`Symfony\\AI\\Platform\\Job\\JobStatus::getRaw` keeps the provider's own wording — the same
+split as ``FinishReason``. A state no bridge knows about is reported as ``UNKNOWN`` and treated as
+non-terminal, so a provider adding a state does not abort a running job.
+
+.. note::
+
+    Only bridges whose provider works this way return a ``JobResult``; everything else answers
+    synchronously as before. Currently this is the MiniMax bridge, for video generation and
+    asynchronous speech synthesis.
+
+Code Examples
+~~~~~~~~~~~~~
+
+* `Asynchronous Video Generation with MiniMax`_
+* `Resuming a MiniMax Video Job`_
+
 Audio Processing
 ----------------
 
@@ -1988,6 +2113,8 @@ Code Examples
 .. _`Binary Image Input with GPT`: https://github.com/symfony/ai/blob/main/examples/openai/image-input-binary.php
 .. _`Image URL Input with GPT`: https://github.com/symfony/ai/blob/main/examples/openai/image-input-url.php
 .. _`Audio Input with GPT`: https://github.com/symfony/ai/blob/main/examples/openai/audio-input.php
+.. _`Asynchronous Video Generation with MiniMax`: https://github.com/symfony/ai/blob/main/examples/minimax/text-to-video.php
+.. _`Resuming a MiniMax Video Job`: https://github.com/symfony/ai/blob/main/examples/minimax/video-job-resume.php
 .. _`Audio Output with GPT`: https://github.com/symfony/ai/blob/main/examples/openai/audio-output.php
 .. _`ElevenLabs Speech-to-Text with SRT`: https://github.com/symfony/ai/blob/main/examples/elevenlabs/speech-to-text-srt.php
 .. _`PDF Input with GPT`: https://github.com/symfony/ai/blob/main/examples/openai/pdf-input-binary.php

examples/minimax/text-to-speech-async.php

@@ -10,14 +10,15 @@
  */
 
 use Symfony\AI\Platform\Bridge\MiniMax\Factory;
+use Symfony\AI\Platform\Job\JobRunner;
 use Symfony\AI\Platform\Message\Content\Text;
 
 require_once dirname(__DIR__).'/bootstrap.php';
 
-$platform = Factory::createPlatform(env('MINI_MAX_API_KEY'), http_client());
+$provider = Factory::createProvider(env('MINI_MAX_API_KEY'), http_client());
 
-// The async endpoint enqueues a task; the bridge transparently polls it until the audio is ready.
-$result = $platform->invoke('speech-2.6-hd', new Text('The real danger is not that computers start thinking like people, but that people start thinking like computers.'), [
+// The async endpoint enqueues a task, so the invocation hands back a job handle instead of audio.
+$handle = $provider->invoke('speech-2.6-hd', new Text('The real danger is not that computers start thinking like people, but that people start thinking like computers.'), [
     'async' => true,
     'voice_setting' => [
         'voice_id' => 'English_expressive_narrator',
@@ -31,6 +32,13 @@
         'format' => 'mp3',
         'channel' => 1,
     ],
-]);
+])->asJob();
 
-echo $result->asBinary();
+$result = (new JobRunner())->wait($provider->getJobClient(), $handle);
+
+// MiniMax delivers the asynchronous result as a tar bundling the mp3 with a `.titles` and an
+// `.extra` file; the bridge unpacks the audio, so this is the same mp3 the synchronous endpoint
+// would have returned.
+$result->asFile(__DIR__.'/minimax-speech.mp3');
+
+echo 'Speech saved to minimax-speech.mp3'.\PHP_EOL;

examples/minimax/text-to-video.php

@@ -10,17 +10,25 @@
  */
 
 use Symfony\AI\Platform\Bridge\MiniMax\Factory;
+use Symfony\AI\Platform\Job\JobRunner;
 use Symfony\AI\Platform\Message\Content\Text;
 
 require_once dirname(__DIR__).'/bootstrap.php';
 
-$platform = Factory::createPlatform(env('MINI_MAX_API_KEY'), http_client());
+$provider = Factory::createProvider(env('MINI_MAX_API_KEY'), http_client());
 
-// Video generation is asynchronous; the bridge polls the task until the file is ready.
-$result = $platform->invoke('MiniMax-Hailuo-02', new Text('A cat playing the piano on a stage, cinematic lighting'), [
+// Video generation is asynchronous: MiniMax accepts the request and answers with a task, so the
+// invocation returns a handle instead of a video.
+$handle = $provider->invoke('MiniMax-Hailuo-02', new Text('A cat playing the piano on a stage, cinematic lighting'), [
     'duration' => 6,
     'resolution' => '768P',
-]);
+])->asJob();
+
+echo 'Started job '.$handle->getId().', waiting for it to finish...'.\PHP_EOL;
+
+// Waiting is explicit, but how long is not something the caller has to know: the handle states that
+// video generation may run for minutes, and the runner honours that unless it is told otherwise.
+$result = (new JobRunner())->wait($provider->getJobClient(), $handle);
 
 $result->asFile(__DIR__.'/minimax-video.mp4');
 

examples/minimax/video-job-resume.php

@@ -0,0 +1,67 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+use Symfony\AI\Platform\Bridge\MiniMax\Factory;
+use Symfony\AI\Platform\Job\JobHandle;
+use Symfony\AI\Platform\Job\JobRunner;
+use Symfony\AI\Platform\Job\JobStateCase;
+use Symfony\AI\Platform\Message\Content\Text;
+
+require_once dirname(__DIR__).'/bootstrap.php';
+
+/*
+ * A long-running job does not have to be waited for in the process that started it. Run this example
+ * once to start a video job - it exits immediately, leaving only a handle on disk - and run it again
+ * to pick the job up and download the result once it is done.
+ *
+ * The same handle is what you would put into a Messenger message or a database row to let a worker
+ * finish the job.
+ */
+
+$provider = Factory::createProvider(env('MINI_MAX_API_KEY'), http_client());
+$storage = __DIR__.'/minimax-video-job.json';
+
+if (!is_file($storage)) {
+    $handle = $provider->invoke('MiniMax-Hailuo-02', new Text('A cat playing the piano on a stage, cinematic lighting'), [
+        'duration' => 6,
+        'resolution' => '768P',
+    ])->asJob();
+
+    file_put_contents($storage, $handle->toString());
+
+    echo 'Started job '.$handle->getId().'. Run this example again to pick it up.'.\PHP_EOL;
+
+    exit(0);
+}
+
+$handle = JobHandle::fromString((string) file_get_contents($storage));
+
+// A worker that never invoked anything builds the client on its own: Factory::createJobClient().
+$jobClient = $provider->getJobClient();
+
+$status = $jobClient->getStatus($handle);
+
+echo 'Job '.$handle->getId().' is "'.$status->getRaw().'".'.\PHP_EOL;
+
+if (!$status->is(JobStateCase::SUCCEEDED)) {
+    echo $status->isTerminal()
+        ? 'It will not produce a result'.(null !== $status->getError() ? ': '.$status->getError() : '.').\PHP_EOL
+        : 'Still running - run this example again in a moment.'.\PHP_EOL;
+
+    exit(0);
+}
+
+// The job is done, so the runner returns without waiting - and hands back the same kind of result
+// a synchronous invocation would have.
+(new JobRunner())->wait($jobClient, $handle)->asFile(__DIR__.'/minimax-video.mp4');
+unlink($storage);
+
+echo 'Video saved to minimax-video.mp4'.\PHP_EOL;

src/ai-bundle/CHANGELOG.md

@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+0.14
+----
+
+ * Register `ai.platform.job_runner` (autowired as `Platform\Job\JobRunner`) using the application clock and, for a platform running asynchronous jobs, its `ai.platform.job_client.<name>` client tagged `ai.platform.job_client`; render an asynchronous job in the profiler as the handle it carries instead of as a result
+
 0.13
 ----
 

src/ai-bundle/config/services.php

@@ -73,6 +73,7 @@
 use Symfony\AI\Platform\Event\InvocationEvent;
 use Symfony\AI\Platform\EventListener\StringToMessageBagListener;
 use Symfony\AI\Platform\EventListener\TemplateRendererListener;
+use Symfony\AI\Platform\Job\JobRunner;
 use Symfony\AI\Platform\Message\TemplateRenderer\ExpressionLanguageTemplateRenderer;
 use Symfony\AI\Platform\Message\TemplateRenderer\StringTemplateRenderer;
 use Symfony\AI\Platform\Message\TemplateRenderer\TemplateRendererRegistry;
@@ -168,6 +169,16 @@
         ->set('ai.platform.string_to_message_bag_listener', StringToMessageBagListener::class)
             ->tag('kernel.event_listener', ['event' => InvocationEvent::class])
 
+        // asynchronous jobs
+        // Registered with the application's clock so a test can control the waiting. No budget is
+        // configured on purpose: each job states how long it may take, so one shared runner serves
+        // a speech job finishing in seconds and a video job running for minutes alike.
+        ->set('ai.platform.job_runner', JobRunner::class)
+            ->args([
+                service('clock'),
+            ])
+        ->alias(JobRunner::class, 'ai.platform.job_runner')
+
         // structured output
         ->set('ai.platform.response_format_factory', ResponseFormatFactory::class)
             ->args([

src/ai-bundle/src/AiBundle.php

@@ -77,6 +77,7 @@
 use Symfony\AI\Platform\Bridge\HuggingFace\Factory as HuggingFaceFactory;
 use Symfony\AI\Platform\Bridge\LmStudio\Factory as LmStudioFactory;
 use Symfony\AI\Platform\Bridge\MiniMax\Factory as MiniMaxFactory;
+use Symfony\AI\Platform\Bridge\MiniMax\MiniMaxJobClient;
 use Symfony\AI\Platform\Bridge\Mistral\Factory as MistralFactory;
 use Symfony\AI\Platform\Bridge\Ollama\Factory as OllamaFactory;
 use Symfony\AI\Platform\Bridge\Ollama\ModelCatalog;
@@ -93,6 +94,7 @@
 use Symfony\AI\Platform\Capability;
 use Symfony\AI\Platform\Contract\JsonSchema\Provider\SchemaProviderInterface;
 use Symfony\AI\Platform\Exception\RuntimeException;
+use Symfony\AI\Platform\Job\JobClientInterface;
 use Symfony\AI\Platform\Message\Content\File;
 use Symfony\AI\Platform\Message\Template;
 use Symfony\AI\Platform\ModelCatalog\ModelCatalogInterface;
@@ -995,6 +997,21 @@ private function processPlatformConfig(string $type, array $platform, ContainerB
             $container->setDefinition($platformId, $definition);
             $container->registerAliasForArgument($platformId, PlatformInterface::class, 'minimax');
 
+            // The job client is registered next to the platform, since a worker resolving a stored
+            // handle has the handle but not the invocation that produced it. Tagged with the provider
+            // name the handle carries, so an application holding handles of several providers can
+            // pick the right client from a locator.
+            $jobClientId = 'ai.platform.job_client.minimax';
+            $container->setDefinition($jobClientId, (new Definition(MiniMaxJobClient::class))
+                ->setFactory(MiniMaxFactory::class.'::createJobClient')
+                ->setArguments([
+                    $platform['api_key'],
+                    new Reference($platform['http_client'], ContainerInterface::NULL_ON_INVALID_REFERENCE),
+                    $platform['endpoint'],
+                ])
+                ->addTag('ai.platform.job_client', ['key' => 'minimax']));
+            $container->registerAliasForArgument($jobClientId, JobClientInterface::class, 'minimax');
+
             return;
         }