$GLOBALS['delayedTerminalNowNs'] / 1e9; $advanceMs = static function (int $ms): void { $GLOBALS['delayedTerminalNowNs'] += $ms * 1000000; }; $read = static fn (string $path): array => json_decode((string) file_get_contents($path), true, 512, JSON_THROW_ON_ERROR); $write = static function (string $path, array $state): void { file_put_contents($path, json_encode($state, JSON_THROW_ON_ERROR)); }; $reply = static fn (array $identity, string $status): array => $identity + ['ok' => true, 'status' => $status]; $make = static function () use (&$sequence, $directory, $expect, $advanceMs): array { $advanceMs(10000); $uuid = sprintf('d2345678-1234-4234-8234-%012d', ++$sequence); $control = new Control(7, 11, $uuid, $directory); $expect($control->begin(), 'fresh synthetic request begins'); $token = $control->beginSubmission('qwen', 'synthetic-user'); $task = 'synthetic-delayed-task-' . $sequence; $run = 'synthetic-delayed-run-' . $sequence; $control->bindTask('qwen', 'synthetic-user', $task, $run, $token); return [$control, $task, $run, $directory . '/' . hash('sha256', '7:11:' . $uuid) . '.json', $uuid, $token]; }; $unlocked = static function (string $path) use ($expect, $directory): void { foreach ([$directory . '/.gc.lock', $path] as $file) { $handle = fopen($file, 'c+'); $acquired = flock($handle, LOCK_EX | LOCK_NB); $expect($acquired, 'request and GC locks are released during I/O and retry wait'); if ($acquired) flock($handle, LOCK_UN); fclose($handle); } }; try { // The observed failure: stop ACK and producer stopped precede persistent // stopped after the owned pool's ten-second bounded teardown. [$c, $task, $run, $path, $uuid, $token] = $make(); $c->signalValidationFailure(); $stopCalls = 0; $c->stopBoundTask(static function () use (&$stopCalls): bool { ++$stopCalls; return true; }); $c->markTerminal($task, 'stopped', $run); $expect(!$c->canStartAttempt() && !$c->publicState()['upstream_terminal_confirmed'], 'ACK and producer receipt alone do not admit correction'); $started = $now(); $times = $timeouts = $identities = []; $GLOBALS['delayedTerminalSleepHook'] = static function () use ($unlocked, $path, $c, $expect): void { $unlocked($path); $expect(!$c->canStartAttempt(), 'correction remains blocked throughout retry spacing'); }; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity, int $ms) use (&$times, &$timeouts, &$identities, $now, $started, $reply, $unlocked, $path): array { $unlocked($path); $times[] = $now() - $started; $timeouts[] = $ms; $identities[] = $identity; return $reply($identity, $now() - $started < 10.25 ? 'running' : 'stopped'); }); unset($GLOBALS['delayedTerminalSleepHook']); $state = $read($path); $archive = array_values($state['completed_generations'])[0]; $expect($result['confirmed'] && $stopCalls === 1 && count($times) === 2, 'delayed independent stopped is confirmed with exactly two reads and one stop'); $expect($times[0] === 0.0 && abs($times[1] - 11.0) < 0.00001, 'first read is immediate and final read follows bounded teardown within original twelve-second window'); $expect($timeouts[0] === 1000 && $timeouts[1] > 0 && $timeouts[1] <= 1000, 'both I/O timeouts stay at most one second within original deadline'); $expect($identities[0] === $identities[1] && $identities[0] === ['profile' => 'qwen', 'user' => 'synthetic-user', 'task_id' => $task, 'workflow_run_id' => $run, 'generation' => 1], 'both reads use the exact original generation and scope'); $expect((float) $archive['verification_started_at'] === $started && (float) $archive['verification_deadline_at'] === $started + 12.0 && $archive['verification_attempts'] === 2 && array_column($archive['run_status_observations'], 'status') === ['running', 'stopped'], 'original budget, finite attempts and both observations survive archive'); $expect($c->canStartAttempt() && $c->publicState()['submitted_pending_attempt_count'] === 0, 'independent terminal and closed owned submission admit clinical correction'); $nextToken = $c->beginSubmission('qwen', 'synthetic-user'); $expect($nextToken !== $token && $c->publicState()['submitted_attempt_count'] === 2 && $read($path)['completed_generations'] === $state['completed_generations'], 'correction receives a new owned submission while preserving original completion evidence'); foreach (['stopped', 'succeeded', 'failed'] as $status) { [$c, $task, $run, $path, $uuid, $token] = $make(); $started = $now(); $calls = 0; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity) use (&$calls, $reply, $status): array { ++$calls; return $reply($identity, $status); }); $expect($result['confirmed'] && $calls === 1 && $now() === $started, 'already-persisted terminal completes immediately: ' . $status); } [$c, $task, $run, $path, $uuid, $token] = $make(); $c->markStop(true, $task); $c->markTerminal($task, 'stopped', $run); $calls = 0; $started = $now(); $running = static function (array $identity) use (&$calls, $reply): array { ++$calls; return $reply($identity, 'running'); }; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', $running); $expect(!$result['confirmed'] && $result['reason'] === 'attempts_exhausted' && $calls === 2 && abs($now() - $started - 11.0) < 0.00001, 'persistent running exhausts finite reads later while retaining identity'); $state = $read($path); $c->closeSubmittedGeneration($token, $task, 'unconfirmed', $running); $c->reconcileTerminal($task, $running); $c->finish('error'); $summary = $c->reconcileClosedTasks($running); $closedDeadline = $read($path)['closed_verification_deadline_at']; $expect(abs($closedDeadline - $read($path)['closed_verification_started_at'] - 12.0) < 0.0000001, 'fresh shared final recovery also freezes a twelve-second deadline'); $expect($calls === 2 && $summary['lookup_attempt_count'] === 0 && (float) $read($path)['verification_started_at'] === $started && $read($path)['verification_deadline_at'] === $state['verification_deadline_at'] && $read($path)['run_status_observations'] === $state['run_status_observations'], 'repeated close, direct and final recovery cannot renew deadline, attempts or observations'); $expect(!$c->canStartAttempt() && $c->publicState()['upstream_pending_task_count'] === 1, 'still-running execution remains a hard successor gate'); $advanceMs(1100); $result = $c->reconcileTerminal($task, $running); $expect($calls === 2 && $result['reason'] === 'budget_exhausted' && (float) $read($path)['verification_started_at'] === $started, 'elapsed original budget is never restarted'); $c->reconcileClosedTasks($running); $expect($read($path)['closed_verification_deadline_at'] === $closedDeadline && $calls === 2, 'repeated final recovery cannot replenish its shared deadline or exhausted generation reads'); foreach ([11.8, 12.1, -1.0] as $elapsed) { [$c, $task, $run, $path, $uuid, $token] = $make(); $state = $read($path); $state['verification_started_at'] = $now() - $elapsed; $state['verification_deadline_at'] = $state['verification_started_at'] + 12.0; $state['verification_attempts'] = 1; $write($path, $state); $calls = 0; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity, int $ms) use (&$calls, $expect, $reply): array { ++$calls; $expect($ms > 0 && $ms <= 200, 'resumed final lookup receives only original remaining 200ms'); return $reply($identity, 'stopped'); }); $expect(abs((float) $read($path)['verification_started_at'] - $state['verification_started_at']) < 0.0000001, 'resumed or reset clock cannot rewrite original start'); $expect($elapsed === 11.8 ? $result['confirmed'] && $calls === 1 : !$result['confirmed'] && $calls === 0 && $result['reason'] === 'budget_exhausted', 'original deadline and monotonic clock reset fail closed without reset'); } foreach ([1.8, 2.1] as $legacyElapsed) { [$c, $task, $run, $path, $uuid, $token] = $make(); $state = $read($path); $legacyStarted = $now() - $legacyElapsed; $state['verification_started_at'] = $legacyStarted; $state['verification_attempts'] = 1; $write($path, $state); $calls = 0; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity, int $ms) use (&$calls, $expect, $reply): array { ++$calls; $expect($ms > 0 && $ms <= 200, 'legacy start-only window retains its original two-second remainder'); return $reply($identity, 'stopped'); }); $expect($legacyElapsed === 1.8 ? $result['confirmed'] && $calls === 1 : !$result['confirmed'] && $calls === 0, 'deploying longer window never replenishes a previously started legacy budget'); if ($result['confirmed']) $expect(abs($read($path)['verification_deadline_at'] - $legacyStarted - 2.0) < 0.0000001, 'legacy deadline remains frozen at its original start plus two seconds'); } // Final read has only one second; an exact independent terminal arriving // after that timeout or the original deadline cannot release the gate. [$c, $task, $run, $path, $uuid, $token] = $make(); $calls = 0; $timeouts = []; $started = $now(); $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity, int $ms) use (&$calls, &$timeouts, $reply, $advanceMs): array { $timeouts[] = $ms; if (++$calls === 1) { $advanceMs(900); return $reply($identity, 'running'); } $advanceMs(1001); return $reply($identity, 'stopped'); }); $state = $read($path); $expect(!$result['confirmed'] && $result['reason'] === 'budget_exhausted' && $calls === 2 && $timeouts[0] === 1000 && $timeouts[1] <= 1000 && abs($now() - $started - 12.001) < 0.00001, 'slow first read and late final callback do not extend original twelve-second budget'); $expect(!$c->canStartAttempt() && isset($state['upstream']) && !$state['run_status_observations'][1]['within_budget'], 'late terminal remains observational only'); [$c, $task, $run, $path, $uuid, $token] = $make(); $calls = 0; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity, int $ms) use (&$calls, $advanceMs, $reply, $expect): array { ++$calls; $expect($ms === 1000, 'first read retains original one-second I/O cap'); $advanceMs(1001); return $reply($identity, 'succeeded'); }); $expect(!$result['confirmed'] && $calls === 1 && !$c->canStartAttempt(), 'late first callback cannot confirm even before generation deadline'); [$c, $task, $run, $path, $uuid, $token] = $make(); $peer = new Control(7, 11, $uuid, $directory); $calls = 0; $peerCalls = 0; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity) use (&$calls, &$peerCalls, $peer, $task, $reply, $expect, $unlocked, $path): array { ++$calls; $unlocked($path); $nested = $peer->reconcileTerminal($task, static function () use (&$peerCalls): array { ++$peerCalls; return []; }); $expect(in_array($nested['reason'], ['verification_in_progress', 'attempts_exhausted'], true), 'peer cannot duplicate an active final or first claim'); return $reply($identity, $calls === 1 ? 'running' : 'stopped'); }); $expect($result['confirmed'] && $calls === 2 && $peerCalls === 0, 'cross-worker claim protects both unlocked lookups'); foreach (['token', 'generation', 'workflow_run_id', 'submission_id'] as $changed) { [$c, $task, $run, $path, $uuid, $token] = $make(); $calls = 0; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity) use (&$calls, $changed, $path, $read, $write, $reply): array { if (++$calls === 1) return $reply($identity, 'running'); $state = $read($path); if ($changed === 'token') $state['verification_claim']['token'] = 'replaced-claim'; elseif ($changed === 'generation') ++$state['upstream']['generation']; else $state['upstream'][$changed] = 'replaced-owner'; $write($path, $state); return $reply($identity, 'stopped'); }); $expect($result['reason'] === 'stale_result' && $calls === 2 && $read($path)['persisted_run_status'] === 'running' && count($read($path)['run_status_observations']) === 1 && !$c->canStartAttempt(), 'late replaced final claim/identity/owner cannot accept old result: ' . $changed); } foreach (['cancel', 'timeout'] as $intent) { [$c, $task, $run, $path, $uuid, $token] = $make(); if ($intent === 'cancel') $c->signalCancel(); else $c->signalTimeout(); $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static fn (array $identity): array => $reply($identity, 'stopped')); $expect($result['confirmed'] && !$c->canStartAttempt() && $c->publicState()[$intent === 'cancel' ? 'cancel_requested' : 'timeout_reached'], 'independent cleanup never bypasses original intent gate: ' . $intent); try { $c->beginSubmission('openai', 'synthetic-user'); $expect(false, 'intent must reject another submission'); } catch (\RuntimeException $error) { $expect($error->getMessage() === 'UPSTREAM_CLEANUP_PENDING', 'cancel/timeout remains a hard submission gate'); } } // A cancel endpoint is a separate worker. It must be able to write and // acknowledge intent during the owner's longer unlocked retry wait. foreach (['stopped', 'running'] as $finalStatus) { [$c, $task, $run, $path, $uuid, $token] = $make(); $peer = new Control(7, 11, $uuid, $directory); $calls = $stopCalls = 0; $cancelLatency = null; $started = $now(); $GLOBALS['delayedTerminalSleepHook'] = static function () use ($peer, $c, $path, $task, $run, $read, $now, $unlocked, $expect, &$stopCalls, &$cancelLatency): void { unset($GLOBALS['delayedTerminalSleepHook']); $unlocked($path); $before = $read($path); $cancelStarted = $now(); $receipt = $peer->requestCancel(static function (array $identity) use (&$stopCalls, $expect, $task, $run, $unlocked, $path): bool { ++$stopCalls; $unlocked($path); $expect($identity['task_id'] === $task && $identity['workflow_run_id'] === $run, 'cancel during wait stops only original captured pair'); return true; }); $cancelLatency = $now() - $cancelStarted; $after = $read($path); $expect($cancelLatency === 0.0 && $receipt['cancel_requested'] && $c->shouldAbort(), 'cancel worker returns and owner sees intent without waiting for scheduled GET'); $expect($receipt['upstream_stop_confirmed'] && !$receipt['upstream_terminal_confirmed'], 'prompt cancel ACK does not certify stopped execution'); $expect($after['verification_started_at'] === $before['verification_started_at'] && $after['verification_attempts'] === $before['verification_attempts'] && $after['upstream'] === $before['upstream'], 'concurrent cancellation preserves original verification budget, attempts and identity'); }; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity, int $ms) use (&$calls, $reply, $finalStatus, $c, $expect): array { if (++$calls === 1) return $reply($identity, 'running'); $expect($c->shouldAbort() && $ms > 0 && $ms <= 1000, 'cancelled owner performs only remaining bounded cleanup lookup'); return $reply($identity, $finalStatus); }); $expect($calls === 2 && $stopCalls === 1 && $cancelLatency === 0.0 && abs($now() - $started - 11.0) < 0.00001, 'cancel endpoint is prompt while owner retains finite synchronous cleanup wait'); $expect($result['confirmed'] === ($finalStatus === 'stopped') && !$c->canStartAttempt() && $c->publicState()['upstream_pending_task_count'] === ($finalStatus === 'stopped' ? 0 : 1), 'cancel during wait preserves hard successor gate for both terminal and persistent-running cleanup'); $c->finish('error'); $expect($c->publicState()['status'] === 'cancelled', 'cancel during retry wait survives final error completion'); } // Scheduling stalls can overrun a sleep; wake-up must recheck the // persisted deadline before allowing another I/O claim. [$c, $task, $run, $path, $uuid, $token] = $make(); $calls = 0; $started = $now(); $GLOBALS['delayedTerminalSleepHook'] = static function () use ($advanceMs): void { unset($GLOBALS['delayedTerminalSleepHook']); $advanceMs(1100); }; $result = $c->closeSubmittedGeneration($token, $task, 'unconfirmed', static function (array $identity) use (&$calls, $reply): array { ++$calls; return $reply($identity, 'running'); }); $expect($calls === 1 && $result['reason'] === 'budget_exhausted' && $now() - $started > 12.0 && (float) $read($path)['verification_started_at'] === $started, 'overslept retry cannot claim another GET or renew original deadline'); $expect(!$c->canStartAttempt() && $c->publicState()['upstream_pending_task_count'] === 1, 'scheduler overrun remains pending with original task retained'); // The final transport path may narrow, never extend, this scheduling. [$c, $task, $run, $path, $uuid, $token] = $make(); $c->finish('error'); $state = $read($path); $state['closed_verification_started_at'] = $now() - 1.3; $write($path, $state); $started = $now(); $times = $timeouts = []; $summary = $c->reconcileClosedTasks(static function (array $identity, int $ms) use (&$times, &$timeouts, $now, $started, $reply, $advanceMs): array { $times[] = $now() - $started; $timeouts[] = $ms; if (count($times) === 1) return $reply($identity, 'running'); $advanceMs($ms + 1); return $reply($identity, 'stopped'); }); $expect(count($times) === 2 && $times[1] >= 0.249 && $times[1] < 0.251 && $timeouts[0] <= 700 && $timeouts[1] <= 450, 'narrower global deadline retains minimum spacing and caps each lookup'); $expect($summary['budget_exhausted'] && !$c->publicState()['upstream_terminal_confirmed'] && $now() - $started <= 0.701, 'final recovery cannot extend its original shared deadline or trust a deadline response'); echo "Delayed terminal confirmation: {$checks} assertions passed (synthetic controls and clock; no network/model/DB).\n"; } finally { unset($GLOBALS['delayedTerminalSleepHook']); $resolved = realpath($directory); $temp = realpath(sys_get_temp_dir()); if ($resolved !== false) { if ($temp === false || !str_starts_with(str_replace('\\', '/', $resolved), rtrim(str_replace('\\', '/', $temp), '/') . '/zyt-delayed-terminal-')) { throw new \RuntimeException('Synthetic cleanup boundary mismatch'); } foreach (new \DirectoryIterator($resolved) as $file) { if ($file->isDot()) continue; if ($file->isLink() || !$file->isFile()) throw new \RuntimeException('Unexpected synthetic cleanup entry'); unlink($file->getPathname()); } rmdir($resolved); } } }