3600000 || $chunkMs < 1000 || $chunkMs > 120000) { self::invalid(); } FollowupAudioPipelineMedia::assertChunkPlan((float) $task['duration_seconds'], $chunkMs); $chunks = []; for ($start = 0; $start < $duration; $start += $chunkMs) { $end = min($duration, $start + $chunkMs); $chunks[] = ['id' => 'asr-' . hash('sha256', $task['sha256'] . ':' . $fingerprint . ':' . $start . ':' . $end), 'start_ms' => $start, 'end_ms' => $end, 'state' => 'pending']; } return ['schema_version' => 1, 'revision' => 0, 'source_sha256' => $task['sha256'], 'fingerprint' => $fingerprint, 'duration_ms' => $duration, 'chunk_ms' => $chunkMs, 'chunks' => $chunks, 'extraction' => ['state' => 'pending']]; } public static function validate(array $state, array $task): void { $ids = json_decode((string) ($task['upstream_ids_json'] ?? '{}'), true); if (!is_string($state['fingerprint'] ?? null) || !preg_match('/^[a-f0-9]{64}$/D', $state['fingerprint']) || !hash_equals((string) ($ids['provider_fingerprint'] ?? ''), $state['fingerprint']) || ($state['source_sha256'] ?? '') !== ($task['sha256'] ?? '') || !is_int($state['revision'] ?? null) || $state['revision'] < 1 || $state['revision'] > 15000 || !is_int($state['chunk_ms'] ?? null)) { self::invalid(); } $expected = self::initial($task, $state['fingerprint'], $state['chunk_ms']); foreach (['schema_version', 'source_sha256', 'fingerprint', 'duration_ms', 'chunk_ms'] as $field) { if (($state[$field] ?? null) !== $expected[$field]) { self::invalid(); } } if (array_diff(array_keys($state), array_keys($expected)) || !is_array($state['chunks'] ?? null) || !array_is_list($state['chunks']) || count($state['chunks']) !== count($expected['chunks'])) { self::invalid(); } $textBytes = 0; $unfinished = false; foreach ($state['chunks'] as $index => $chunk) { if (!is_array($chunk)) { self::invalid(); } foreach (['id', 'start_ms', 'end_ms'] as $field) { if (($chunk[$field] ?? null) !== $expected['chunks'][$index][$field]) { self::invalid(); } } self::entry($chunk, false); if ($unfinished && $chunk['state'] !== 'pending') { self::invalid(); } if ($chunk['state'] !== 'complete') { $unfinished = true; } $textBytes += strlen($chunk['text'] ?? ''); } if ($textBytes > 2000000 || !is_array($state['extraction'] ?? null)) { self::invalid(); } self::entry($state['extraction'], true); if ($unfinished && $state['extraction']['state'] !== 'pending') { self::invalid(); } } private static function entry(array $entry, bool $extraction): void { $allowed = $extraction ? ['state', 'request_id', 'answer'] : ['id', 'start_ms', 'end_ms', 'state', 'request_id', 'text']; if (array_diff(array_keys($entry), $allowed) || !in_array($entry['state'] ?? '', ['pending', 'intent', 'complete', 'rejected'], true)) { self::invalid(); } if ($entry['state'] === 'pending') { if (array_key_exists('request_id', $entry) || array_key_exists($extraction ? 'answer' : 'text', $entry)) { self::invalid(); } } elseif (!is_string($entry['request_id'] ?? null) || !preg_match('/^fa-[a-f0-9]{32}$/D', $entry['request_id'])) { self::invalid(); } $field = $extraction ? 'answer' : 'text'; if ($entry['state'] === 'complete') { if (!is_string($entry[$field] ?? null) || !mb_check_encoding($entry[$field], 'UTF-8') || strlen($entry[$field]) > ($extraction ? 8388608 : 200000)) { self::invalid(); } } elseif (array_key_exists($field, $entry)) { self::invalid(); } } public static function transition(array $previous, array $next, array $task, int $expectedRevision): void { self::validate($next, $task); if ($previous === []) { $initial = self::initial($task, $next['fingerprint'], $next['chunk_ms']); $initial['revision'] = 1; if ($expectedRevision !== 0 || FollowupAudioPolicy::canonical($next) !== FollowupAudioPolicy::canonical($initial)) { self::invalid(); } return; } self::validate($previous, $task); if ($expectedRevision !== $previous['revision'] || $next['revision'] !== $previous['revision'] + 1) { throw new FollowupAudioException('PIPELINE_CHECKPOINT_CONFLICT'); } foreach (['schema_version', 'source_sha256', 'fingerprint', 'duration_ms', 'chunk_ms'] as $key) { if ($next[$key] !== $previous[$key]) { self::invalid(); } } $changed = 0; foreach (array_merge($previous['chunks'], [$previous['extraction']]) as $index => $before) { $after = $index < count($next['chunks']) ? $next['chunks'][$index] : $next['extraction']; if (FollowupAudioPolicy::canonical($before) === FollowupAudioPolicy::canonical($after)) { continue; } $changed++; foreach (['id', 'start_ms', 'end_ms'] as $key) { if (($before[$key] ?? null) !== ($after[$key] ?? null)) { self::invalid(); } } $from = $before['state']; $to = $after['state']; if (!((in_array($from, ['pending', 'rejected'], true) && $to === 'intent') || ($from === 'intent' && in_array($to, ['complete', 'rejected'], true)))) { self::invalid(); } if ($from === 'intent' && $before['request_id'] !== $after['request_id']) { self::invalid(); } } if ($changed !== 1) { self::invalid(); } } public static function hasIntent(array $state): bool { foreach (array_merge($state['chunks'], [$state['extraction']]) as $entry) { if ($entry['state'] === 'intent') { return true; } } return false; } public static function segments(array $state): array { $result = []; foreach ($state['chunks'] as $chunk) { if ($chunk['state'] === 'complete') { $result[] = array_intersect_key($chunk, array_flip(['id', 'start_ms', 'end_ms', 'text'])); } } return $result; } private static function invalid(): void { throw new FollowupAudioException('PIPELINE_CHECKPOINT_INVALID', true); } }