ars-backend/app/Services/RunDispatcher.php

<?php

namespace App\Services;

use App\Jobs\AgentRunJob;
use App\Models\ChatSession;
use App\Models\Message;
use Illuminate\Database\Eloquent\ModelNotFoundException;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Str;

class RunDispatcher
{
    public function __construct(
        private readonly ChatService $chatService,
        private readonly OutputSink $outputSink,
    ) {
    }

    /**
     * @throws ModelNotFoundException
     */
    public function dispatchForPrompt(string $sessionId, string $triggerMessageId): string
    {
        $triggerMessage = $this->chatService->getMessage($sessionId, $triggerMessageId);
        if (! $triggerMessage) {
            throw (new ModelNotFoundException())->setModel(Message::class, [$triggerMessageId]);
        }

        $shouldDispatch = false;

        $runId = DB::transaction(function () use ($sessionId, $triggerMessageId, &$shouldDispatch) {
            ChatSession::query()
                ->whereKey($sessionId)
                ->lockForUpdate()
                ->firstOrFail();

            $latestStatus = Message::query()
                ->where('session_id', $sessionId)
                ->where('type', 'run.status')
                ->orderByDesc('seq')
                ->first();

            if ($latestStatus && ($latestStatus->payload['status'] ?? null) === 'RUNNING' && ($latestStatus->payload['run_id'] ?? null)) {
                logger('existing run found', ['sessionId' => $sessionId, 'runId' => $latestStatus->payload['run_id']]);
                return $latestStatus->payload['run_id'];
            }

            $candidateRunId = (string) Str::uuid();
            $wasDeduped = null;

            $statusMessage = $this->outputSink->appendRunStatus($sessionId, $candidateRunId, 'RUNNING', [
                'trigger_message_id' => $triggerMessageId,
                'dedupe_key' => 'run:trigger:'.$triggerMessageId,
            ], $wasDeduped);

            $finalRunId = $statusMessage->payload['run_id'] ?? $candidateRunId;

            if ($wasDeduped) {
                logger('existing run found', ['sessionId' => $sessionId, 'runId' => $finalRunId]);
                return $finalRunId;
            }

            $shouldDispatch = true;

            return $finalRunId;
        });

        if ($shouldDispatch) {
            logger('dispatching run', ['sessionId' => $sessionId, 'runId' => $runId]);
            dispatch(new AgentRunJob($sessionId, $runId));
        }

        return $runId;
    }
}