859 lines
49 KiB
PHP
859 lines
49 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace app\adminapi\service;
|
|
|
|
use app\common\service\DifyChatService;
|
|
|
|
/**
|
|
* Shared between FPM workers on one host. Contains no credentials or clinical text.
|
|
* The caller MUST authorize the diagnosis before constructing or reading a control.
|
|
*/
|
|
final class AssistantRequestControl
|
|
{
|
|
private const STOP_LEASE_SECONDS = 1.25;
|
|
private const STOP_RETRY_DELAY_SECONDS = 0.25;
|
|
private const STOP_RETENTION_SECONDS = 30;
|
|
private const MAX_STOP_ATTEMPTS = 2;
|
|
private const VERIFICATION_BUDGET_SECONDS = 12.0;
|
|
private const LEGACY_VERIFICATION_BUDGET_SECONDS = 2.0;
|
|
private const VERIFICATION_IO_SECONDS = 1.0;
|
|
private const VERIFICATION_LEASE_SECONDS = 1.25;
|
|
private const VERIFICATION_RETRY_SECONDS = 0.25;
|
|
private const VERIFICATION_FINAL_IO_RESERVE_SECONDS = 1.0;
|
|
private const MAX_VERIFICATION_ATTEMPTS = 2;
|
|
|
|
private string $path;
|
|
private int $ttl;
|
|
private string $directory;
|
|
|
|
public function __construct(int $adminId, int $diagnosisId, string $requestId, ?string $directory = null, int $ttl = 1200)
|
|
{
|
|
if ($adminId <= 0 || $diagnosisId <= 0 || !self::validRequestId($requestId)) {
|
|
throw new \InvalidArgumentException('Invalid assistant request identity');
|
|
}
|
|
$directory = $directory ?? app()->getRuntimePath() . 'assistant-requests';
|
|
if (!is_dir($directory) && !@mkdir($directory, 0700, true) && !is_dir($directory)) {
|
|
throw new \RuntimeException('Assistant request control unavailable');
|
|
}
|
|
$this->path = rtrim($directory, '/\\') . DIRECTORY_SEPARATOR
|
|
. hash('sha256', $adminId . ':' . $diagnosisId . ':' . strtolower($requestId)) . '.json';
|
|
$this->ttl = max(600, $ttl);
|
|
$this->directory = $directory;
|
|
$this->collectExpired();
|
|
}
|
|
|
|
public static function validRequestId(string $id): bool
|
|
{
|
|
return preg_match('/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i', $id) === 1;
|
|
}
|
|
|
|
/** A cancel tombstone created before prepare is retained; a replay never restarts work. */
|
|
public function begin(): bool
|
|
{
|
|
return $this->update(static function (array &$state): bool {
|
|
if (!empty($state['begun']) || self::hasUnconfirmedIdentity($state)) {
|
|
return false;
|
|
}
|
|
$state['begun'] = true;
|
|
$state['status'] = !empty($state['cancel_requested']) ? 'cancel_requested' : 'running';
|
|
return true;
|
|
});
|
|
}
|
|
|
|
public function shouldAbort(): bool
|
|
{
|
|
return $this->update(static fn (array &$state): bool => !empty($state['cancel_requested']), false);
|
|
}
|
|
|
|
/** Read-only gate before every new generation/compatibility attempt. */
|
|
public function canStartAttempt(): bool
|
|
{
|
|
return $this->update(static fn (array &$state): bool => self::canStartState($state), false);
|
|
}
|
|
|
|
private static function canStartState(array $state, string $ignoreSubmission = ''): bool
|
|
{
|
|
if (empty($state['begun']) || !empty($state['cancel_requested']) || !empty($state['timeout_reached'])
|
|
|| in_array($state['status'] ?? '', ['done', 'error', 'cancelled'], true)) return false;
|
|
if (self::submissionCounts($state, $ignoreSubmission)['pending'] > 0) return false;
|
|
if (is_array($state['upstream'] ?? null) && !self::independentlyTerminal($state)) return false;
|
|
foreach ($state['handoff_cleanup'] ?? [] as $cleanup) {
|
|
if (is_array($cleanup['upstream'] ?? null) && !self::independentlyTerminal($cleanup)) return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
/** Atomic ownership record immediately before this generation's first POST execution. */
|
|
public function beginSubmission(string $profile, string $user): string
|
|
{
|
|
if (!in_array($profile, ['qwen', 'openai'], true) || strlen($user) > 128) throw new \RuntimeException('Invalid submission scope');
|
|
$token = bin2hex(random_bytes(12));
|
|
return $this->update(static function (array &$state) use ($profile, $user, $token): string {
|
|
if (!self::canStartState($state)) throw new \RuntimeException('UPSTREAM_CLEANUP_PENDING');
|
|
self::releaseTask($state);
|
|
$state['active_submission_id'] = $token;
|
|
$state['upstream_stop_confirmed'] = false;
|
|
$state['upstream_terminal_confirmed'] = false;
|
|
$state['upstream_terminal_status'] = '';
|
|
foreach (['producer_terminal_status', 'producer_terminal_observed_at', 'persisted_run_status',
|
|
'persisted_run_observed_at', 'persisted_terminal_confirmed', 'lifecycle_conflict', 'run_status_observations',
|
|
'transfer_closed', 'transfer_closed_at', 'generation_verification_elapsed_ms', 'stop_accepted_at',
|
|
'stop_requested', 'stop_retry_until'] as $field) unset($state[$field]);
|
|
$state['submission_sequence'] = (int) ($state['submission_sequence'] ?? 0) + 1;
|
|
$state['submitted_attempts'][$token] = ['generation' => $state['submission_sequence'], 'profile' => $profile,
|
|
'user' => $user, 'submitted_at' => microtime(true), 'transfer_closed' => false,
|
|
'task_id' => '', 'workflow_run_id' => '', 'disposition' => 'unconfirmed'];
|
|
return $token;
|
|
});
|
|
}
|
|
|
|
/** Empty-task closure is authoritative only for its own admitted attempt token. */
|
|
public function closeSubmittedGeneration(string $token, string $taskId, string $disposition, callable $lookup): array
|
|
{
|
|
if (!in_array($disposition, ['unconfirmed', 'not_submitted', 'rejected_before_execution'], true)) {
|
|
throw new \InvalidArgumentException('Invalid submission disposition');
|
|
}
|
|
$bound = $this->update(static function (array &$state) use ($token, $taskId, $disposition): ?string {
|
|
if (!preg_match('/\A[0-9a-f]{24}\z/D', $token) || !isset($state['submitted_attempts'][$token])) return null;
|
|
$record =& $state['submitted_attempts'][$token];
|
|
if ($record['task_id'] !== $taskId) return null;
|
|
if (empty($record['transfer_closed'])) {
|
|
$record['transfer_closed'] = true;
|
|
$record['transfer_closed_at'] = microtime(true);
|
|
if ($taskId === '') $record['disposition'] = $disposition;
|
|
if (($state['active_submission_id'] ?? '') === $token && $taskId === '') {
|
|
$state['transfer_closed'] = true;
|
|
$state['transfer_closed_at'] = $record['transfer_closed_at'];
|
|
}
|
|
}
|
|
return $record['task_id'];
|
|
});
|
|
if ($bound === null) return ['confirmed' => false, 'pending' => true, 'reason' => 'stale_submission'];
|
|
if ($bound === '') {
|
|
$pending = $this->update(static fn (array &$state): bool =>
|
|
!in_array($state['submitted_attempts'][$token]['disposition'], ['not_submitted', 'rejected_before_execution'], true), false);
|
|
return ['confirmed' => false, 'pending' => $pending,
|
|
'reason' => $pending ? 'identity_unavailable' : 'pre_execution_closed'];
|
|
}
|
|
// Exact owned task/run validation, lookup claims and budgets stay in the existing path.
|
|
return $this->closeGeneration($bound, $lookup);
|
|
}
|
|
|
|
/** Submission resolution never borrows a same-ID record from another attempt. */
|
|
private static function submissionCounts(array $state, string $ignore = ''): array
|
|
{
|
|
$counts = ['total' => 0, 'closed' => 0, 'pending' => 0, 'unidentified' => 0, 'additional_pending' => 0, 'pre_execution' => 0, 'not_submitted' => 0, 'rejected_before_execution' => 0];
|
|
foreach ($state['submitted_attempts'] ?? [] as $token => $submission) {
|
|
++$counts['total'];
|
|
if (!empty($submission['transfer_closed'])) ++$counts['closed'];
|
|
if ($token === $ignore) continue;
|
|
if (!empty($submission['transfer_closed']) && ($submission['task_id'] ?? '') === ''
|
|
&& in_array($submission['disposition'] ?? '', ['not_submitted', 'rejected_before_execution'], true)) {
|
|
++$counts['pre_execution'];
|
|
++$counts[$submission['disposition']]; continue;
|
|
}
|
|
$resolved = false; $activePending = false;
|
|
$targets = array_merge([$state], array_values($state['handoff_cleanup'] ?? []));
|
|
foreach ($targets as $target) {
|
|
$records = array_values($target['completed_generations'] ?? []);
|
|
if (is_array($target['upstream'] ?? null)) $records[] = $target;
|
|
foreach ($records as $record) {
|
|
$identity = $record['upstream'] ?? [];
|
|
if (($identity['submission_id'] ?? '') !== $token || ($submission['task_id'] ?? '') === ''
|
|
|| ($identity['task_id'] ?? null) !== $submission['task_id']
|
|
|| ($identity['workflow_run_id'] ?? null) !== ($submission['workflow_run_id'] ?? '')
|
|
|| ($identity['profile'] ?? null) !== ($submission['profile'] ?? null)
|
|
|| ($identity['user'] ?? null) !== ($submission['user'] ?? null)
|
|
|| ($identity['generation'] ?? null) !== ($submission['control_generation'] ?? null)) continue;
|
|
if (self::independentlyTerminal($record) && !empty($submission['transfer_closed'])) $resolved = true;
|
|
elseif (($target['upstream']['submission_id'] ?? '') === $token) $activePending = true;
|
|
}
|
|
}
|
|
if (!$resolved) {
|
|
++$counts['pending'];
|
|
if (($submission['task_id'] ?? '') === '' || ($submission['workflow_run_id'] ?? '') === '') ++$counts['unidentified'];
|
|
if (!$activePending) ++$counts['additional_pending'];
|
|
}
|
|
}
|
|
return $counts;
|
|
}
|
|
|
|
/** An already-created conflicting task must be stopped without user-cancel intent. */
|
|
public function hasHandoffTask(string $taskId): bool
|
|
{
|
|
return $this->update(static fn (array &$state): bool =>
|
|
($state['handoff_cleanup'][$taskId]['upstream']['task_id'] ?? '') === $taskId, false);
|
|
}
|
|
|
|
/** Keep normal and exceptional cleanup leases in the same request's private file. */
|
|
private static function &taskState(array &$state, ?string $taskId): array
|
|
{
|
|
if ($taskId === null || ($state['upstream']['task_id'] ?? '') === $taskId) return $state;
|
|
if (isset($state['handoff_cleanup'][$taskId])) return $state['handoff_cleanup'][$taskId];
|
|
$missing = [];
|
|
return $missing;
|
|
}
|
|
|
|
public function bindTask(string $profile, string $user, string $taskId, string $workflowRunId = '', string $submissionId = ''): void
|
|
{
|
|
if (!in_array($profile, ['qwen', 'openai'], true)
|
|
|| !self::validUpstreamId($taskId) || strlen($user) > 128
|
|
|| ($workflowRunId !== '' && !self::validUpstreamId($workflowRunId))) {
|
|
throw new \RuntimeException('Invalid upstream task identity');
|
|
}
|
|
$outcome = $this->update(static function (array &$state) use ($profile, $user, $taskId, $workflowRunId, $submissionId): string {
|
|
if ($submissionId !== '') {
|
|
$submission = $state['submitted_attempts'][$submissionId] ?? null;
|
|
if (!is_array($submission) || !empty($submission['transfer_closed'])
|
|
|| $submission['profile'] !== $profile || $submission['user'] !== $user
|
|
|| ($submission['task_id'] !== '' && $submission['task_id'] !== $taskId)
|
|
|| ($submission['workflow_run_id'] !== '' && $workflowRunId !== '' && $submission['workflow_run_id'] !== $workflowRunId)) return 'UPSTREAM_TASK_IDENTITY_CONFLICT';
|
|
}
|
|
$bind = static function (array &$state) use ($profile, $user, $taskId, $workflowRunId, $submissionId): string {
|
|
$completed = $state['completed_generations'] ?? [];
|
|
foreach ($state['handoff_cleanup'] ?? [] as $cleanup) array_push($completed, ...array_values($cleanup['completed_generations'] ?? []));
|
|
foreach ($completed as $record) {
|
|
if (($record['upstream']['task_id'] ?? '') === $taskId
|
|
|| ($workflowRunId !== '' && ($record['upstream']['workflow_run_id'] ?? '') === $workflowRunId)) return 'UPSTREAM_TASK_IDENTITY_CONFLICT';
|
|
}
|
|
$previous = $state['upstream'] ?? [];
|
|
if (($previous['task_id'] ?? '') === $taskId) {
|
|
return self::enrichRun($state, $profile, $user, $workflowRunId)
|
|
? '' : 'UPSTREAM_TASK_IDENTITY_CONFLICT';
|
|
}
|
|
if (isset($state['handoff_cleanup'][$taskId])) {
|
|
return self::enrichRun($state['handoff_cleanup'][$taskId], $profile, $user, $workflowRunId)
|
|
? 'UPSTREAM_TASK_HANDOFF_BLOCKED' : 'UPSTREAM_TASK_IDENTITY_CONFLICT';
|
|
}
|
|
// The first identity may arrive after the deadline/cancel frame.
|
|
// It belongs to the already-started transfer and must remain the
|
|
// primary cleanup target, not be mistaken for a new generation.
|
|
$lateFirst = !empty($state['begun']) && empty($previous) && (empty($state['task_generation']) || $submissionId !== '');
|
|
$state['task_generation'] = (int) ($state['task_generation'] ?? 0) + 1;
|
|
$incoming = ['profile' => $profile, 'user' => $user, 'task_id' => $taskId,
|
|
'workflow_run_id' => $workflowRunId, 'generation' => $state['task_generation']];
|
|
if ($submissionId !== '') $incoming['submission_id'] = $submissionId;
|
|
if (!$lateFirst && !self::canStartState($state, $submissionId)) {
|
|
// The pre-start gate normally prevents this race. If a task
|
|
// already exists remotely, retain BOTH identities and its own
|
|
// bounded stop lease; never replace the unconfirmed old task.
|
|
$state['handoff_cleanup'][$taskId] ??= ['upstream' => $incoming,
|
|
'stop_requested' => true, 'stop_attempts' => 0,
|
|
'stop_retry_until' => microtime(true) + self::STOP_RETENTION_SECONDS];
|
|
return 'UPSTREAM_TASK_HANDOFF_BLOCKED';
|
|
}
|
|
$state['upstream'] = $incoming;
|
|
$state['upstream_stop_confirmed'] = false;
|
|
$state['upstream_terminal_confirmed'] = false;
|
|
$state['upstream_terminal_status'] = '';
|
|
foreach (['producer_terminal_status', 'producer_terminal_observed_at', 'persisted_run_status',
|
|
'persisted_run_observed_at', 'persisted_terminal_confirmed', 'lifecycle_conflict', 'run_status_observations',
|
|
'transfer_closed', 'transfer_closed_at', 'generation_verification_elapsed_ms', 'stop_accepted_at'] as $field) unset($state[$field]);
|
|
$state['stop_attempts'] = 0;
|
|
// These belong to the previous generation. In particular an old
|
|
// expired retention deadline must not disable a later real cancel.
|
|
unset($state['stop_claim'], $state['stop_retry_after']);
|
|
unset($state['verification_claim'], $state['verification_attempts'],
|
|
$state['verification_started_at'], $state['verification_deadline_at'], $state['verification_retry_after']);
|
|
if (!$lateFirst) unset($state['stop_requested'], $state['stop_retry_until']);
|
|
return '';
|
|
};
|
|
$result = $bind($state);
|
|
if ($submissionId !== '' && in_array($result, ['', 'UPSTREAM_TASK_HANDOFF_BLOCKED'], true)) {
|
|
$target =& self::taskState($state, $taskId);
|
|
if (isset($target['upstream']) && ($target['upstream']['submission_id'] ?? $submissionId) === $submissionId) {
|
|
$target['upstream']['submission_id'] = $submissionId;
|
|
$state['submitted_attempts'][$submissionId]['task_id'] = $taskId;
|
|
$state['submitted_attempts'][$submissionId]['workflow_run_id'] = $target['upstream']['workflow_run_id'];
|
|
$state['submitted_attempts'][$submissionId]['control_generation'] = $target['upstream']['generation'];
|
|
} else return 'UPSTREAM_TASK_IDENTITY_CONFLICT';
|
|
}
|
|
return $result;
|
|
});
|
|
if ($outcome !== '') throw new \RuntimeException($outcome);
|
|
}
|
|
|
|
private static function validUpstreamId(string $id): bool
|
|
{
|
|
return preg_match('/\A[a-zA-Z0-9_-]{1,128}\z/', $id) === 1;
|
|
}
|
|
|
|
/** Same-generation enrichment never changes stop, cancel or verification state. */
|
|
private static function enrichRun(array &$target, string $profile, string $user, string $workflowRunId): bool
|
|
{
|
|
$task = $target['upstream'] ?? [];
|
|
if (($task['profile'] ?? '') !== $profile || ($task['user'] ?? '') !== $user) return false;
|
|
$captured = (string) ($task['workflow_run_id'] ?? '');
|
|
if ($workflowRunId !== '' && $captured !== '' && $captured !== $workflowRunId) return false;
|
|
if ($workflowRunId !== '' && $captured === '') $target['upstream']['workflow_run_id'] = $workflowRunId;
|
|
return true;
|
|
}
|
|
|
|
public function clearTask(?string $taskId = null): void
|
|
{
|
|
$this->update(static function (array &$state) use ($taskId): void {
|
|
$target =& self::taskState($state, $taskId);
|
|
if ($taskId !== null && ($target['upstream']['task_id'] ?? '') !== $taskId) return;
|
|
if (is_array($target['upstream'] ?? null)) {
|
|
$target['transfer_closed'] = true;
|
|
$target['transfer_closed_at'] ??= microtime(true);
|
|
}
|
|
self::releaseTask($target);
|
|
});
|
|
}
|
|
|
|
public function markStop(bool $confirmed, ?string $taskId = null): void
|
|
{
|
|
$this->update(static function (array &$state) use ($confirmed, $taskId): void {
|
|
$target =& self::taskState($state, $taskId);
|
|
if ($taskId !== null && ($target['upstream']['task_id'] ?? '') !== $taskId) return;
|
|
$target['upstream_stop_confirmed'] = !empty($target['upstream_stop_confirmed']) || $confirmed;
|
|
if ($confirmed) $target['stop_accepted_at'] ??= microtime(true);
|
|
});
|
|
}
|
|
|
|
/** Bound producer receipt only. It never proves the independent run status. */
|
|
public function markTerminal(string $taskId, string $status, string $workflowRunId = ''): void
|
|
{
|
|
if (!in_array($status, ['stopped', 'succeeded', 'failed'], true)) return;
|
|
$this->update(static function (array &$state) use ($taskId, $status, $workflowRunId): void {
|
|
$target =& self::taskState($state, $taskId);
|
|
if (($target['upstream']['task_id'] ?? '') !== $taskId) return;
|
|
$captured = (string) ($target['upstream']['workflow_run_id'] ?? '');
|
|
// Legacy live-stream task-only receipts remain compatible. They do
|
|
// not establish a run binding and cannot enable independent lookup.
|
|
if ($captured !== '' ? $captured !== $workflowRunId : $workflowRunId !== '') return;
|
|
if (isset($target['producer_terminal_status'])) {
|
|
if ($target['producer_terminal_status'] !== $status) $target['lifecycle_conflict'] = true;
|
|
return;
|
|
}
|
|
$target['producer_terminal_status'] = $status;
|
|
$target['producer_terminal_observed_at'] = microtime(true);
|
|
if (isset($target['persisted_run_status']) && $target['persisted_run_status'] !== $status) $target['lifecycle_conflict'] = true;
|
|
});
|
|
}
|
|
|
|
/** Called only after this generation's curl handle has closed, never from its callbacks. */
|
|
public function closeGeneration(string $taskId, callable $lookup): array
|
|
{
|
|
if (!self::validUpstreamId($taskId)) return ['confirmed' => false, 'pending' => true, 'reason' => 'identity_unavailable'];
|
|
$this->clearTask($taskId);
|
|
return $this->reconcileTerminalIdentity($taskId, $lookup);
|
|
}
|
|
|
|
private static function independentlyTerminal(array $target): bool
|
|
{
|
|
return ($target['persisted_terminal_confirmed'] ?? null) === true
|
|
&& !empty($target['upstream_terminal_confirmed'])
|
|
&& in_array($target['persisted_run_status'] ?? null, ['stopped', 'succeeded', 'failed'], true);
|
|
}
|
|
|
|
private static function observationTime($value): ?float
|
|
{
|
|
return (is_float($value) || is_int($value)) && is_finite((float) $value) && $value >= 0 ? (float) $value : null;
|
|
}
|
|
|
|
/**
|
|
* Separately invoked recovery for an already closed transport. No built-in
|
|
* lookup, network call or public route exists. The trusted read-only adapter
|
|
* receives only the exact private identity and an I/O timeout in milliseconds;
|
|
* it MUST honor that timeout. Late/throwing/malformed results fail closed.
|
|
* Arbitrary PHP callbacks cannot be preempted; elapsed time is checked again
|
|
* before applying any result. Neither returned bodies nor error prose persist.
|
|
* @return array{confirmed:bool,pending:bool,reason:string}
|
|
*/
|
|
public function reconcileTerminal(string $expectedTaskId, callable $lookup): array
|
|
{
|
|
return $this->reconcileTerminalIdentity($expectedTaskId, $lookup);
|
|
}
|
|
|
|
/**
|
|
* Recover only this authorized request's original captured pairs after finish.
|
|
* The shared monotonic budget survives repeated calls and covers all targets,
|
|
* I/O and retry spacing. The adapter must honor each supplied I/O timeout;
|
|
* arbitrary PHP callbacks cannot be preempted and late results are rejected.
|
|
* No private identity, response body or exception metadata leaves this method.
|
|
* @return array<string,bool|int|string>
|
|
*/
|
|
public function reconcileClosedTasks(callable $lookup): array
|
|
{
|
|
$invokedAt = hrtime(true) / 1e9;
|
|
$snapshot = $this->update(static function (array &$state) use ($invokedAt): array {
|
|
$closed = !empty($state['transport_closed']);
|
|
$tasks = [];
|
|
foreach (array_merge([$state], array_values($state['handoff_cleanup'] ?? [])) as $target) {
|
|
if (!is_array($target) || !is_array($target['upstream'] ?? null) || self::independentlyTerminal($target)) continue;
|
|
$identity = self::verificationIdentity($target);
|
|
if ($closed && $identity !== null) $tasks[] = $identity;
|
|
}
|
|
$deadline = $invokedAt;
|
|
if ($tasks !== []) {
|
|
$existingBudget = isset($state['closed_verification_started_at']);
|
|
$started = (float) ($state['closed_verification_started_at'] ?? $invokedAt);
|
|
$state['closed_verification_started_at'] = $started;
|
|
// Freeze new windows; existing start-only controls retain their
|
|
// original two-second deadline rather than gaining a new budget.
|
|
$originalDeadline = min($started + self::VERIFICATION_BUDGET_SECONDS,
|
|
(float) ($state['closed_verification_deadline_at'] ?? ($started + ($existingBudget
|
|
? self::LEGACY_VERIFICATION_BUDGET_SECONDS : self::VERIFICATION_BUDGET_SECONDS))));
|
|
$state['closed_verification_deadline_at'] = $originalDeadline;
|
|
// A host monotonic-clock reset must not renew a persisted budget.
|
|
$deadline = $started > $invokedAt || !is_finite($originalDeadline) ? $invokedAt : $originalDeadline;
|
|
}
|
|
return ['closed' => $closed, 'tasks' => $tasks, 'deadline' => $deadline];
|
|
});
|
|
$attempts = 0;
|
|
$queried = 0;
|
|
$confirmed = 0;
|
|
$exhausted = false;
|
|
foreach ($snapshot['tasks'] as $identity) {
|
|
if (hrtime(true) / 1e9 >= $snapshot['deadline']) {
|
|
$exhausted = true;
|
|
break;
|
|
}
|
|
$targetAttempts = 0;
|
|
$boundedLookup = static function (array $task, int $timeoutMs) use ($lookup, &$attempts, &$targetAttempts) {
|
|
++$attempts;
|
|
++$targetAttempts;
|
|
return $lookup($task, $timeoutMs);
|
|
};
|
|
$result = $this->reconcileTerminalIdentity($identity['task_id'], $boundedLookup, $snapshot['deadline'], $identity);
|
|
if ($targetAttempts > 0) ++$queried;
|
|
if ($result['confirmed']) ++$confirmed;
|
|
if ($result['reason'] === 'budget_exhausted' && hrtime(true) / 1e9 >= $snapshot['deadline']) $exhausted = true;
|
|
}
|
|
$pending = $this->update(static function (array &$state): int {
|
|
$count = 0;
|
|
foreach (array_merge([$state], array_values($state['handoff_cleanup'] ?? [])) as $target) {
|
|
if (is_array($target) && is_array($target['upstream'] ?? null) && !self::independentlyTerminal($target)) ++$count;
|
|
}
|
|
return $count + self::submissionCounts($state)['additional_pending'];
|
|
}, false);
|
|
return [
|
|
'status' => !$snapshot['closed'] ? 'transport_open' : ($pending === 0 ? 'complete' : ($exhausted ? 'budget_exhausted' : 'pending')),
|
|
'transport_closed' => $snapshot['closed'],
|
|
'eligible_task_count' => count($snapshot['tasks']),
|
|
'queried_task_count' => $queried,
|
|
'lookup_attempt_count' => $attempts,
|
|
'confirmed_task_count' => $confirmed,
|
|
'upstream_pending_task_count' => $pending,
|
|
'budget_exhausted' => $exhausted,
|
|
'elapsed_ms' => max(0, (int) floor((hrtime(true) / 1e9 - $invokedAt) * 1000)),
|
|
];
|
|
}
|
|
|
|
/** The optional caller deadline only narrows each generation's persisted budget. */
|
|
private function reconcileTerminalIdentity(string $expectedTaskId, callable $lookup, ?float $invocationDeadline = null,
|
|
?array $expectedIdentity = null): array
|
|
{
|
|
while (true) {
|
|
$token = bin2hex(random_bytes(12));
|
|
$claim = $this->update(static function (array &$state) use ($expectedTaskId, $token, $invocationDeadline, $expectedIdentity): array {
|
|
$target =& self::taskState($state, $expectedTaskId);
|
|
$task = self::verificationIdentity($target);
|
|
if (!is_array($target['upstream'] ?? null)) {
|
|
return ['result' => ['confirmed' => false, 'pending' => false, 'reason' => 'task_unavailable']];
|
|
}
|
|
if (self::independentlyTerminal($target)) {
|
|
return ['result' => ['confirmed' => true, 'pending' => false, 'reason' => 'already_confirmed']];
|
|
}
|
|
if ($task === null || $task['task_id'] !== $expectedTaskId) {
|
|
return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'identity_unavailable']];
|
|
}
|
|
if ($expectedIdentity !== null && $task !== $expectedIdentity) {
|
|
return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'stale_result']];
|
|
}
|
|
if (empty($state['transport_closed']) && empty($target['transfer_closed'])) {
|
|
return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'transport_open']];
|
|
}
|
|
$now = hrtime(true) / 1e9;
|
|
$existingBudget = isset($target['verification_started_at']);
|
|
$started = (float) ($target['verification_started_at'] ?? $now);
|
|
$originalDeadline = min($started + self::VERIFICATION_BUDGET_SECONDS,
|
|
(float) ($target['verification_deadline_at'] ?? ($started + ($existingBudget
|
|
? self::LEGACY_VERIFICATION_BUDGET_SECONDS : self::VERIFICATION_BUDGET_SECONDS))));
|
|
// A monotonic clock reset (e.g. host reboot) never renews a budget.
|
|
$deadline = min($originalDeadline, $invocationDeadline ?? INF);
|
|
$remaining = $deadline - $now;
|
|
if ($started > $now || !is_finite($deadline) || $remaining <= 0) {
|
|
return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'budget_exhausted']];
|
|
}
|
|
$attempts = (int) ($target['verification_attempts'] ?? 0);
|
|
if ($attempts >= self::MAX_VERIFICATION_ATTEMPTS) {
|
|
return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'attempts_exhausted']];
|
|
}
|
|
if ((float) ($target['verification_claim']['until'] ?? 0) > $now) {
|
|
return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'verification_in_progress']];
|
|
}
|
|
$retryAt = (float) ($target['verification_retry_after'] ?? 0);
|
|
if ($attempts === self::MAX_VERIFICATION_ATTEMPTS - 1) {
|
|
// Keep the first observation fast. Spend the final finite read
|
|
// after the owned pool's bounded ten-second teardown, reserving
|
|
// one second for I/O within the SAME persisted window.
|
|
$retryAt = max($retryAt, $deadline - self::VERIFICATION_FINAL_IO_RESERVE_SECONDS);
|
|
}
|
|
$wait = $retryAt - $now;
|
|
if ($wait > 0) return ['wait_us' => (int) ceil(min($wait, $remaining) * 1e6)];
|
|
$timeoutMs = (int) floor(min(self::VERIFICATION_IO_SECONDS, $remaining) * 1000);
|
|
if ($timeoutMs < 1) {
|
|
return ['result' => ['confirmed' => false, 'pending' => true, 'reason' => 'budget_exhausted']];
|
|
}
|
|
$target['verification_started_at'] = $started;
|
|
$target['verification_deadline_at'] = $originalDeadline;
|
|
$target['verification_attempts'] = $attempts + 1;
|
|
$target['verification_claim'] = ['token' => $token, 'until' => $now + self::VERIFICATION_LEASE_SECONDS];
|
|
return ['task' => $task, 'timeout_ms' => $timeoutMs,
|
|
'submission_owner' => $target['upstream']['submission_id'] ?? '', 'deadline' => $deadline];
|
|
});
|
|
if (isset($claim['result'])) return $claim['result'];
|
|
if (isset($claim['wait_us'])) {
|
|
usleep($claim['wait_us']);
|
|
continue;
|
|
}
|
|
$task = $claim['task'];
|
|
$status = null;
|
|
$observedStatus = null;
|
|
$reason = 'lookup_failed';
|
|
$ioStarted = hrtime(true) / 1e9;
|
|
// No request-file or GC lock is held across the injected I/O.
|
|
try {
|
|
$response = $lookup($task, $claim['timeout_ms']);
|
|
$observedStatus = self::observedStatus($response, $task);
|
|
$status = in_array($observedStatus, ['stopped', 'succeeded', 'failed'], true) ? $observedStatus : null;
|
|
$reason = $status === null ? 'unconfirmed' : 'confirmed';
|
|
} catch (\Throwable $ignored) {}
|
|
$ioFinished = hrtime(true) / 1e9;
|
|
if ($ioFinished - $ioStarted > $claim['timeout_ms'] / 1000 || $ioFinished >= $claim['deadline']) {
|
|
$status = null;
|
|
$reason = 'budget_exhausted';
|
|
}
|
|
$deadline = $claim['deadline'];
|
|
$submissionOwner = $claim['submission_owner'];
|
|
$result = $this->update(static function (array &$state) use ($task, $token, $status, $observedStatus, $reason, $deadline, $submissionOwner): array {
|
|
$target =& self::taskState($state, $task['task_id']);
|
|
if (self::verificationIdentity($target) !== $task
|
|
|| ($target['upstream']['submission_id'] ?? '') !== $submissionOwner
|
|
|| ($target['verification_claim']['token'] ?? '') !== $token) {
|
|
return ['confirmed' => false, 'pending' => true, 'reason' => 'stale_result'];
|
|
}
|
|
unset($target['verification_claim']);
|
|
$target['verification_retry_after'] = hrtime(true) / 1e9 + self::VERIFICATION_RETRY_SECONDS;
|
|
$target['generation_verification_elapsed_ms'] = max(0, (int) floor(
|
|
(hrtime(true) / 1e9 - $target['verification_started_at']) * 1000));
|
|
if ($observedStatus !== null) {
|
|
$at = microtime(true);
|
|
$target['persisted_run_status'] = $observedStatus;
|
|
$target['persisted_run_observed_at'] = $at;
|
|
$target['run_status_observations'][] = ['status' => $observedStatus, 'observed_at' => $at,
|
|
'within_budget' => $reason !== 'budget_exhausted'];
|
|
if (isset($target['producer_terminal_status']) && $target['producer_terminal_status'] !== $observedStatus) $target['lifecycle_conflict'] = true;
|
|
}
|
|
if (self::independentlyTerminal($target)) {
|
|
return ['confirmed' => true, 'pending' => false, 'reason' => 'already_confirmed'];
|
|
}
|
|
if ($status !== null && hrtime(true) / 1e9 < $deadline) {
|
|
$target['upstream_terminal_confirmed'] = true;
|
|
$target['upstream_terminal_status'] = $status;
|
|
$target['persisted_terminal_confirmed'] = true;
|
|
self::releaseTask($target);
|
|
return ['confirmed' => true, 'pending' => false, 'reason' => 'confirmed'];
|
|
}
|
|
return ['confirmed' => false, 'pending' => true,
|
|
'reason' => hrtime(true) / 1e9 >= $deadline ? 'budget_exhausted' : $reason];
|
|
});
|
|
if ($result['confirmed'] || $result['reason'] === 'stale_result' || $reason === 'budget_exhausted') return $result;
|
|
}
|
|
}
|
|
|
|
/** Only original captured pairs with a local generation are lookup-capable. */
|
|
private static function verificationIdentity(array $target): ?array
|
|
{
|
|
$task = $target['upstream'] ?? [];
|
|
if (!is_array($task) || !is_string($task['task_id'] ?? null) || !self::validUpstreamId($task['task_id'])
|
|
|| !is_string($task['workflow_run_id'] ?? null) || !self::validUpstreamId($task['workflow_run_id'])
|
|
|| !in_array($task['profile'] ?? '', ['qwen', 'openai'], true)
|
|
|| !is_string($task['user'] ?? null) || strlen($task['user']) > 128
|
|
|| !is_int($task['generation'] ?? null) || $task['generation'] <= 0) return null;
|
|
return ['profile' => $task['profile'], 'user' => $task['user'], 'task_id' => $task['task_id'],
|
|
'workflow_run_id' => $task['workflow_run_id'], 'generation' => $task['generation']];
|
|
}
|
|
|
|
private static function observedStatus($response, array $task): ?string
|
|
{
|
|
if (!is_array($response) || ($response['ok'] ?? null) !== true
|
|
|| ($response['workflow_run_id'] ?? null) !== $task['workflow_run_id']
|
|
|| !in_array($response['status'] ?? null, ['scheduled', 'running', 'succeeded', 'failed', 'stopped', 'partial-succeeded', 'paused'], true)) return null;
|
|
foreach (['task_id', 'user', 'profile'] as $field) {
|
|
if (array_key_exists($field, $response) && $response[$field] !== $task[$field]) return null;
|
|
}
|
|
return $response['status'];
|
|
}
|
|
|
|
/** Safe inside a transport callback: writes intent and never performs network I/O. */
|
|
public function signalCancel(): void
|
|
{
|
|
$this->update(static function (array &$state): void {
|
|
if (in_array($state['status'] ?? '', ['done', 'error'], true)) return;
|
|
$state['cancel_requested'] = true;
|
|
if (($state['status'] ?? '') !== 'cancelled') $state['status'] = 'cancel_requested';
|
|
$state['stop_retry_until'] = $state['stop_retry_until'] ?? microtime(true) + self::STOP_RETENTION_SECONDS;
|
|
});
|
|
}
|
|
|
|
/** A generation deadline requests remote cleanup without becoming a user cancellation. */
|
|
public function signalTimeout(): void
|
|
{
|
|
$this->update(static function (array &$state): void {
|
|
$state['timeout_reached'] = true;
|
|
$state['stop_requested'] = true;
|
|
$state['stop_retry_until'] = $state['stop_retry_until'] ?? microtime(true) + self::STOP_RETENTION_SECONDS;
|
|
});
|
|
}
|
|
|
|
/** Rejecting clinical output stops work without pretending the doctor cancelled. */
|
|
public function signalValidationFailure(): void
|
|
{
|
|
$this->update(static function (array &$state): void {
|
|
$state['validation_rejected'] = true;
|
|
$state['stop_requested'] = true;
|
|
$state['stop_retry_until'] = $state['stop_retry_until'] ?? microtime(true) + self::STOP_RETENTION_SECONDS;
|
|
});
|
|
}
|
|
|
|
/** Safe client view. A requested cancellation is never reported as confirmed stop. */
|
|
public function requestCancel(?callable $stop = null): array
|
|
{
|
|
$this->signalCancel();
|
|
$this->stopBoundTask($stop);
|
|
$cleanupIds = $this->update(static fn (array &$state): array => array_keys($state['handoff_cleanup'] ?? []), false);
|
|
foreach ($cleanupIds as $taskId) $this->stopBoundTask($stop, (string) $taskId);
|
|
return $this->publicState();
|
|
}
|
|
|
|
/**
|
|
* Shared by the cancel endpoint and stream owner. The lease exceeds the 1s stop
|
|
* transport budget; a crashed claimant can be replaced, while accepted stops
|
|
* never repeat. Failed tasks remain eligible for one short-lived compensation.
|
|
* @return array{accepted:bool,pending:bool}
|
|
*/
|
|
public function stopBoundTask(?callable $stop = null, ?string $expectedTaskId = null): array
|
|
{
|
|
$token = bin2hex(random_bytes(12));
|
|
$claim = $this->update(static function (array &$state) use ($token, $expectedTaskId): array {
|
|
$target =& self::taskState($state, $expectedTaskId);
|
|
$now = microtime(true);
|
|
$task = $target['upstream'] ?? null;
|
|
$accepted = !empty($target['upstream_stop_confirmed']);
|
|
if (!is_array($task) || ($expectedTaskId !== null && $task['task_id'] !== $expectedTaskId)) {
|
|
return ['accepted' => false, 'pending' => false];
|
|
}
|
|
// A cancel worker may load a task written by the previous local
|
|
// release. Upgrade its private identity atomically before claiming.
|
|
if (!isset($task['generation'])) {
|
|
$target['task_generation'] = (int) ($target['task_generation'] ?? 0) + 1;
|
|
$task['generation'] = $target['task_generation'];
|
|
$target['upstream'] = $task;
|
|
}
|
|
if ($accepted || self::independentlyTerminal($target) || isset($target['producer_terminal_status'])
|
|
|| (empty($target['cancel_requested']) && empty($target['stop_requested']))
|
|
|| (float) ($target['stop_retry_until'] ?? 0) <= $now) {
|
|
return ['accepted' => $accepted, 'pending' => false];
|
|
}
|
|
if ((float) ($target['stop_claim']['until'] ?? 0) > $now
|
|
|| (float) ($target['stop_retry_after'] ?? 0) > $now) {
|
|
return ['accepted' => false, 'pending' => true];
|
|
}
|
|
if ((int) ($target['stop_attempts'] ?? 0) >= self::MAX_STOP_ATTEMPTS) {
|
|
return ['accepted' => false, 'pending' => false];
|
|
}
|
|
$target['stop_claim'] = ['token' => $token, 'until' => $now + self::STOP_LEASE_SECONDS];
|
|
$target['stop_attempts'] = (int) ($target['stop_attempts'] ?? 0) + 1;
|
|
return ['task' => $task, 'accepted' => false, 'pending' => true];
|
|
});
|
|
if (!isset($claim['task'])) return $claim;
|
|
$task = $claim['task'];
|
|
$confirmed = false;
|
|
// Never hold either file lock while waiting for the remote stop endpoint.
|
|
try {
|
|
$confirmed = ($stop !== null ? $stop($task) : DifyChatService::stopTask(
|
|
$task['profile'], $task['task_id'], $task['user']
|
|
)) === true;
|
|
} catch (\Throwable $ignored) {}
|
|
return $this->update(static function (array &$state) use ($task, $token, $confirmed): array {
|
|
$target =& self::taskState($state, $task['task_id']);
|
|
if (($target['upstream']['generation'] ?? 0) !== $task['generation']
|
|
|| ($target['upstream']['task_id'] ?? '') !== $task['task_id']
|
|
|| ($target['upstream']['submission_id'] ?? '') !== ($task['submission_id'] ?? '')
|
|
|| ($target['stop_claim']['token'] ?? '') !== $token) {
|
|
return ['accepted' => false, 'pending' => false];
|
|
}
|
|
$target['upstream_stop_confirmed'] = !empty($target['upstream_stop_confirmed']) || $confirmed;
|
|
if ($confirmed) $target['stop_accepted_at'] ??= microtime(true);
|
|
unset($target['stop_claim']);
|
|
$target['stop_retry_after'] = microtime(true) + self::STOP_RETRY_DELAY_SECONDS;
|
|
return ['accepted' => !empty($target['upstream_stop_confirmed']), 'pending' => false];
|
|
});
|
|
}
|
|
|
|
private static function releaseTask(array &$state): void
|
|
{
|
|
// The stop I/O retry window may expire, but an unconfirmed identity
|
|
// survives TTL reset/GC. Expiry is not a terminal.
|
|
if (is_array($state['upstream'] ?? null) && !self::independentlyTerminal($state)) {
|
|
return;
|
|
}
|
|
if (is_array($state['upstream'] ?? null)) {
|
|
$fields = ['upstream', 'upstream_stop_confirmed', 'stop_accepted_at', 'producer_terminal_status',
|
|
'producer_terminal_observed_at', 'persisted_run_status', 'persisted_run_observed_at',
|
|
'persisted_terminal_confirmed', 'upstream_terminal_confirmed', 'upstream_terminal_status',
|
|
'lifecycle_conflict', 'run_status_observations', 'transfer_closed', 'transfer_closed_at',
|
|
'verification_attempts', 'verification_started_at', 'verification_deadline_at', 'generation_verification_elapsed_ms'];
|
|
$state['completed_generations'][$state['upstream']['generation']] = array_intersect_key($state, array_flip($fields));
|
|
}
|
|
unset($state['upstream'], $state['stop_claim']);
|
|
}
|
|
|
|
public function finish(string $status): void
|
|
{
|
|
if (!in_array($status, ['done', 'error', 'cancelled'], true)) {
|
|
throw new \InvalidArgumentException('Invalid assistant terminal state');
|
|
}
|
|
$this->update(static function (array &$state) use ($status): void {
|
|
if (in_array($state['status'] ?? '', ['done', 'error', 'cancelled'], true)) {
|
|
return;
|
|
}
|
|
$state['status'] = !empty($state['cancel_requested']) ? 'cancelled' : $status;
|
|
$state['transport_closed'] = true;
|
|
self::releaseTask($state);
|
|
});
|
|
}
|
|
|
|
public function publicState(): array
|
|
{
|
|
return $this->update(static function (array &$state): array {
|
|
$pending = is_array($state['upstream'] ?? null) && !self::independentlyTerminal($state) ? 1 : 0;
|
|
$stopConfirmed = !empty($state['upstream_stop_confirmed']);
|
|
$terminalConfirmed = self::independentlyTerminal($state);
|
|
$producer = in_array($state['producer_terminal_status'] ?? null, ['stopped', 'succeeded', 'failed'], true)
|
|
? $state['producer_terminal_status'] : '';
|
|
$persisted = in_array($state['persisted_run_status'] ?? null,
|
|
['scheduled', 'running', 'succeeded', 'failed', 'stopped', 'partial-succeeded', 'paused'], true) ? $state['persisted_run_status'] : '';
|
|
$conflict = !empty($state['lifecycle_conflict']);
|
|
foreach ($state['completed_generations'] ?? [] as $completed) $conflict = $conflict || !empty($completed['lifecycle_conflict']);
|
|
foreach ($state['handoff_cleanup'] ?? [] as $cleanup) {
|
|
$conflict = $conflict || !empty($cleanup['lifecycle_conflict']);
|
|
foreach ($cleanup['completed_generations'] ?? [] as $completed) $conflict = $conflict || !empty($completed['lifecycle_conflict']);
|
|
if (is_array($cleanup['upstream'] ?? null) && !self::independentlyTerminal($cleanup)) {
|
|
++$pending;
|
|
$stopConfirmed = $stopConfirmed && !empty($cleanup['upstream_stop_confirmed']);
|
|
$terminalConfirmed = false;
|
|
}
|
|
}
|
|
$submissions = self::submissionCounts($state);
|
|
$pending += $submissions['additional_pending'];
|
|
if ($submissions['pending'] > 0) $terminalConfirmed = false;
|
|
return [
|
|
'status' => $state['status'] ?? 'pending',
|
|
'cancel_requested' => !empty($state['cancel_requested']),
|
|
'timeout_reached' => !empty($state['timeout_reached']),
|
|
'transport_closed' => !empty($state['transport_closed']),
|
|
'upstream_stop_confirmed' => $stopConfirmed,
|
|
'upstream_terminal_confirmed' => $terminalConfirmed,
|
|
'upstream_terminal_status' => $terminalConfirmed ? (string) ($state['upstream_terminal_status'] ?? '') : '',
|
|
'upstream_pending_task_count' => $pending,
|
|
'producer_terminal_received' => $producer !== '',
|
|
'producer_terminal_status' => $producer,
|
|
'producer_terminal_observed_at' => self::observationTime($state['producer_terminal_observed_at'] ?? null),
|
|
'persisted_run_status' => $persisted,
|
|
'persisted_run_observed_at' => self::observationTime($state['persisted_run_observed_at'] ?? null),
|
|
'stop_accepted_at' => self::observationTime($state['stop_accepted_at'] ?? null),
|
|
'generation_transfer_closed' => !empty($state['transfer_closed']),
|
|
'generation_transfer_closed_at' => self::observationTime($state['transfer_closed_at'] ?? null),
|
|
'generation_verification_elapsed_ms' => max(0, (int) ($state['generation_verification_elapsed_ms'] ?? 0)),
|
|
'lifecycle_conflict' => $conflict,
|
|
'submitted_attempt_count' => $submissions['total'],
|
|
'submitted_transfer_closed_count' => $submissions['closed'],
|
|
'submitted_pending_attempt_count' => $submissions['pending'],
|
|
'submitted_without_identity_count' => $submissions['unidentified'],
|
|
'pre_execution_closed_attempt_count' => $submissions['pre_execution'],
|
|
'not_submitted_attempt_count' => $submissions['not_submitted'],
|
|
'rejected_before_execution_attempt_count' => $submissions['rejected_before_execution'],
|
|
];
|
|
}, false);
|
|
}
|
|
|
|
private function update(callable $apply, bool $persist = true)
|
|
{
|
|
$guard = @fopen($this->directory . DIRECTORY_SEPARATOR . '.gc.lock', 'c+');
|
|
if ($guard === false || !flock($guard, LOCK_SH)) {
|
|
if (is_resource($guard)) fclose($guard);
|
|
throw new \RuntimeException('Assistant request control unavailable');
|
|
}
|
|
$handle = @fopen($this->path, 'c+');
|
|
if ($handle === false || !flock($handle, $persist ? LOCK_EX : LOCK_SH)) {
|
|
if (is_resource($handle)) fclose($handle);
|
|
flock($guard, LOCK_UN);
|
|
fclose($guard);
|
|
throw new \RuntimeException('Assistant request control unavailable');
|
|
}
|
|
try {
|
|
@chmod($this->path, 0600);
|
|
$raw = stream_get_contents($handle);
|
|
$state = is_string($raw) ? json_decode($raw, true) : null;
|
|
if ($raw !== '' && !is_array($state)) throw new \RuntimeException('Assistant request control unavailable');
|
|
if (!is_array($state) || ((int) ($state['expires_at'] ?? 0) <= time() && !self::hasUnconfirmedIdentity($state))) {
|
|
$state = ['expires_at' => time() + $this->ttl];
|
|
}
|
|
$result = $apply($state);
|
|
if (!$persist) return $result;
|
|
$state['expires_at'] = time() + $this->ttl;
|
|
$encoded = json_encode($state, JSON_THROW_ON_ERROR);
|
|
rewind($handle);
|
|
if (!ftruncate($handle, 0) || fwrite($handle, $encoded) !== strlen($encoded) || !fflush($handle)) {
|
|
throw new \RuntimeException('Assistant request control unavailable');
|
|
}
|
|
return $result;
|
|
} finally {
|
|
flock($handle, LOCK_UN);
|
|
fclose($handle);
|
|
flock($guard, LOCK_UN);
|
|
fclose($guard);
|
|
}
|
|
}
|
|
|
|
private static function hasUnconfirmedIdentity(array $state): bool
|
|
{
|
|
if (self::submissionCounts($state)['pending'] > 0) return true;
|
|
if (is_array($state['upstream'] ?? null) && !self::independentlyTerminal($state)) return true;
|
|
foreach ($state['handoff_cleanup'] ?? [] as $cleanup) {
|
|
if (is_array($cleanup['upstream'] ?? null) && !self::independentlyTerminal($cleanup)) return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
/** The global guard prevents unlink/reopen races with any other worker. */
|
|
private function collectExpired(): void
|
|
{
|
|
$guard = @fopen($this->directory . DIRECTORY_SEPARATOR . '.gc.lock', 'c+');
|
|
if ($guard === false) return;
|
|
try {
|
|
if (!flock($guard, LOCK_EX | LOCK_NB)) return;
|
|
$paths = glob($this->directory . DIRECTORY_SEPARATOR . '*.json') ?: [];
|
|
// Bound each sweep; rotating the scan also reclaims older files in large directories.
|
|
if (count($paths) > 256) shuffle($paths);
|
|
foreach (array_slice($paths, 0, 256) as $path) {
|
|
if ((int) @filemtime($path) > time() - 600) continue;
|
|
$state = json_decode((string) @file_get_contents($path), true);
|
|
// Corrupt controls fail closed; their identity cannot be inferred.
|
|
if (is_array($state) && (int) ($state['expires_at'] ?? 0) <= time()
|
|
&& !self::hasUnconfirmedIdentity($state)) @unlink($path);
|
|
}
|
|
} finally {
|
|
flock($guard, LOCK_UN);
|
|
fclose($guard);
|
|
}
|
|
}
|
|
}
|