Roog
6d934f4e34
- 默认切换 AgentProvider 为 HttpAgentProvider，增强网络请求的容错和重试机制 - 优化 Run 逻辑，支持多场景去重与并发保护 - 添加 Redis 发布失败的日志记录以提升问题排查效率 - 扩展 OpenAPI 规范，新增 Error 和 Run 状态相关模型 - 增强测试覆盖，验证调度策略和重复请求的幂等性 - 增加数据库索引以优化查询性能 - 更新所有相关文档和配置文件
78 lines
2.9 KiB
PHP
78 lines
2.9 KiB
PHP
<?php
|
|
|
|
namespace App\Jobs;
|
|
|
|
use App\Models\Message;
|
|
use App\Services\Agent\ProviderException;
|
|
use App\Services\CancelChecker;
|
|
use App\Services\OutputSink;
|
|
use App\Services\RunLoop;
|
|
use Illuminate\Bus\Queueable;
|
|
use Illuminate\Contracts\Queue\ShouldQueue;
|
|
use Illuminate\Foundation\Bus\Dispatchable;
|
|
use Illuminate\Queue\InteractsWithQueue;
|
|
use Illuminate\Queue\SerializesModels;
|
|
|
|
class AgentRunJob implements ShouldQueue
|
|
{
|
|
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
|
|
|
|
public int $tries;
|
|
|
|
public int $timeout;
|
|
|
|
public int $backoff;
|
|
|
|
public function __construct(public string $sessionId, public string $runId)
|
|
{
|
|
$this->tries = (int) config('agent.job.tries', 1);
|
|
$this->timeout = (int) config('agent.job.timeout_seconds', 120);
|
|
$this->backoff = (int) config('agent.job.backoff_seconds', 5);
|
|
}
|
|
|
|
public function handle(RunLoop $loop, OutputSink $sink, CancelChecker $cancelChecker): void
|
|
{
|
|
try {
|
|
logger("Running run {$this->runId} for session {$this->sessionId}");
|
|
$loop->run($this->sessionId, $this->runId);
|
|
} catch (\Throwable $e) {
|
|
logger("Running error {$this->runId} for session {$this->sessionId}");
|
|
logger("error message:",[$e->getMessage(),$e->getTraceAsString()]);
|
|
$errorCode = $e instanceof ProviderException ? $e->errorCode : 'run.failed';
|
|
$dedupeKey = $e instanceof ProviderException
|
|
? "run:{$this->runId}:error:provider"
|
|
: "run:{$this->runId}:error:job";
|
|
|
|
$sink->appendError($this->sessionId, $this->runId, $errorCode, $e->getMessage(), [], $dedupeKey);
|
|
$sink->appendRunStatus($this->sessionId, $this->runId, 'FAILED', [
|
|
'error' => $e->getMessage(),
|
|
'dedupe_key' => "run:{$this->runId}:status:FAILED",
|
|
]);
|
|
|
|
throw $e;
|
|
} finally {
|
|
$latestStatus = Message::query()
|
|
->where('session_id', $this->sessionId)
|
|
->where('type', 'run.status')
|
|
->whereRaw("payload->>'run_id' = ?", [$this->runId])
|
|
->orderByDesc('seq')
|
|
->first();
|
|
|
|
$status = $latestStatus ? ($latestStatus->payload['status'] ?? null) : null;
|
|
|
|
if ($status === 'RUNNING' || ! $status) {
|
|
if ($cancelChecker->isCanceled($this->sessionId, $this->runId)) {
|
|
$sink->appendRunStatus($this->sessionId, $this->runId, 'CANCELED', [
|
|
'dedupe_key' => "run:{$this->runId}:status:CANCELED",
|
|
]);
|
|
} else {
|
|
$sink->appendRunStatus($this->sessionId, $this->runId, 'FAILED', [
|
|
'error' => 'JOB_ENDED_UNEXPECTEDLY',
|
|
'dedupe_key' => "run:{$this->runId}:status:FAILED",
|
|
]);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|