Module: Cohere::Transcribe::State

Defined in:
lib/cohere/transcribe/state/io.rb,
lib/cohere/transcribe/state/locking.rb,
lib/cohere/transcribe/state/manifest.rb,
lib/cohere/transcribe/state/contracts.rb,
lib/cohere/transcribe/state/checkpoint.rb

Defined Under Namespace

Classes: BoundDirectory, Checkpoint, CheckpointRestore, DirectoryBinding, OutputLockTarget, OutputSetLock, PublicationVerification, SourceSnapshot

Constant Summary collapse

STATE_SCHEMA_VERSION = 
1
STATE_SUFFIX = 
".cohere-transcribe.manifest.json"
CHECKPOINT_SUFFIX = 
".cohere-transcribe.asr.json"
MAX_STATE_BYTES = 
64 * 1024 * 1024
DEFERRED_PUBLICATION_EXCEPTIONS = 
{
  SignalException => :never,
  SystemExit => :never
}.freeze
LOCK_DIRECTORY_PREFIX = 
"cohere-transcribe"
LOCK_SUFFIX = 
".lock"
CONTRACT_SCHEMA_VERSION = 
3
ASR_IMPLEMENTATION_FILES = 
%w[
  asr/native.rb
  audio/decoder.rb
  audio/ffmpeg_native.rb
  audio/segmentation.rb
  configuration.rb
  dense_converter.rb
  gguf_writer.rb
  model_identity.rb
  runtime/engine.rb
  runtime/model_provider.rb
  runtime/preparation.rb
  vad/silero.rb
  vad/silero_vad_v6.onnx
  vad/timestamps.rb
].freeze
RENDER_IMPLEMENTATION_FILES = 
%w[
  alignment/aligner.rb
  alignment/ctc.rb
  alignment/text.rb
  output/publication.rb
  output/rendering.rb
  output/timing.rb
  types.rb
].freeze

Class Method Summary collapse

Class Method Details

.asr_checkpoint_payload(result:, source_snapshot:, asr_contract_key:, speech_spans:, vad_provider_options:, generation_id:) ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 56

def asr_checkpoint_payload(result:, source_snapshot:, asr_contract_key:, speech_spans:,
                           vad_provider_options:, generation_id:)
  provenance = result.provenance
  {
    "kind" => "asr_complete",
    "generation_id" => generation_id,
    "asr_contract_key" => asr_contract_key,
    "source" => source_snapshot.payload,
    "updated_unix_seconds" => Time.now.to_f,
    "checkpoint" => {
      "duration" => result.duration,
      "segment_times" => result.segments.map { |segment| [segment.start, segment.end] },
      "speech_spans" => speech_spans.map { |start_time, end_time| [start_time, end_time] },
      "segment_texts" => result.segments.map(&:text),
      "generated_tokens" => provenance.generated_tokens_by_segment.sort,
      "repetition_stopped_segments" => provenance.repetition_stopped_segments.sort,
      "truncation_retried_segments" => provenance.truncation_retried_segments.sort,
      "token_limit_segments" => provenance.token_limit_segments.sort,
      "decode_backend" => provenance.decode_backend,
      "decode_fallback_reason" => provenance.decode_fallback_reason,
      "vad_engine_actual" => provenance.vad_engine_actual,
      "vad_provider" => provenance.vad_provider,
      "vad_provider_options" => vad_provider_options,
      "vad_fallback_reason" => provenance.vad_fallback_reason
    }
  }
end

.asr_contract_key(options, model_format: "dense", model_quantization: nil) ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 39

def asr_contract_key(options, model_format: "dense", model_quantization: nil)
  silero = options.vad == "silero"
  auditok = options.vad == "auditok"
  model_id = json_value(options.model)
  model_revision = ModelIdentity.default_model_revision(model_id, options.model_revision)
  configuration = {
    "language" => options.language,
    "device" => options.device,
    "dtype" => options.dtype,
    "audio_backend" => options.audio_backend,
    "vad" => options.vad,
    "vad_engine" => silero ? options.vad_engine : nil,
    "vad_batch_size" => silero ? options.vad_batch_size : nil,
    "vad_block_frames" => silero ? options.vad_block_frames : nil,
    "vad_threads" => silero ? options.vad_threads : nil,
    "vad_merge" => silero ? options.vad_merge : nil,
    "min_dur" => options.vad == "none" ? nil : options.min_dur,
    "max_dur" => options.max_dur,
    "max_silence" => auditok ? options.max_silence : nil,
    "energy_threshold" => auditok ? options.energy_threshold : nil,
    "vad_threshold" => silero ? options.vad_threshold : nil,
    "min_silence_ms" => silero ? options.min_silence_ms : nil,
    "speech_pad_ms" => silero ? options.speech_pad_ms : nil,
    "batch_size" => options.batch_size,
    "batch_max_size" => options.batch_max_size,
    "batch_audio_seconds" => options.batch_audio_seconds,
    "batch_vram_target" => options.batch_vram_target,
    "adaptive_batch" => options.adaptive_batch,
    "pin_memory" => options.pin_memory,
    "audio_memory_gb" => options.audio_memory_gb,
    "pipeline_preparation" => options.pipeline_preparation,
    "max_new_tokens" => options.max_new_tokens,
    "max_retry_tokens" => options.max_retry_tokens,
    "truncation_policy" => options.truncation_policy,
    "stop_repetition_loops" => options.stop_repetition_loops
  }
  fingerprint(
    "contract_schema_version" => CONTRACT_SCHEMA_VERSION,
    "configuration" => configuration,
    "models" => {
      "asr" => {
        "id" => model_id,
        "revision" => model_revision,
        "format" => model_format,
        "quantization" => model_quantization,
        "adapter" => options.adapter && {
          "id" => json_value(options.adapter),
          "revision" => options.adapter_revision
        }
      },
      "silero_version" => "6.2.1"
    },
    "implementation_sha256" => implementation_fingerprint(ASR_IMPLEMENTATION_FILES)
  )
end

.canonical_json(value) ⇒ Object 



466
467
468
 
# File 'lib/cohere/transcribe/state/io.rb', line 466

def canonical_json(value)
  JSON.generate(deep_sort(value))
end

.checkpoint_path_for_outputs(output_paths) ⇒ Object 



461
462
463
464
 
# File 'lib/cohere/transcribe/state/io.rb', line 461

def checkpoint_path_for_outputs(output_paths)
  parent, stem = output_parent_and_stem(output_paths)
  parent.join(".#{stem}#{CHECKPOINT_SUFFIX}")
end

.decode_state(path, directory_binding: nil, guard_bindings: nil) ⇒ Object 



478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
 
# File 'lib/cohere/transcribe/state/io.rb', line 478

def decode_state(path, directory_binding: nil, guard_bindings: nil)
  path = Pathname(path)
  contents = securely_read_regular_file(
    path,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  )
  decoded = JSON.parse(contents)
  return [nil, "state marker root is not an object"] unless decoded.is_a?(Hash)
  return [nil, "state marker schema is unsupported"] unless decoded["schema_version"] == STATE_SCHEMA_VERSION

  payload = decoded["payload"]
  return [nil, "state marker payload is not an object"] unless payload.is_a?(Hash)

  expected = Digest::SHA256.hexdigest(canonical_json(payload))
  return [nil, "state marker integrity check failed"] unless secure_digest_equal?(decoded["payload_sha256"], expected)

  [payload, nil]
rescue Errno::ENOENT
  [nil, "state marker is missing or not a regular file"]
rescue JSON::ParserError, JSON::GeneratorError, EncodingError, SystemCallError,
       TranscriptionRuntimeError, TypeError, ArgumentError => e
  [nil, "state marker is unreadable (#{e.class}: #{e.message})"]
end

.deep_freeze_json(value) ⇒ Object 



210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 210

def deep_freeze_json(value)
  case value
  when Hash
    value.to_h { |key, item| [key.to_s.freeze, deep_freeze_json(item)] }.freeze
  when Array
    value.map { |item| deep_freeze_json(item) }.freeze
  when String
    value.dup.freeze
  else
    value
  end
end

.deep_sort(value) ⇒ Object 



710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
 
# File 'lib/cohere/transcribe/state/io.rb', line 710

def deep_sort(value)
  case value
  when Hash
    value.keys.map(&:to_s).sort.to_h do |key|
      original_key = value.key?(key) ? key : value.keys.find { |candidate| candidate.to_s == key }
      [key, deep_sort(value.fetch(original_key))]
    end
  when Array
    value.map { |item| deep_sort(item) }
  when Pathname
    value.to_s
  else
    value
  end
end

.ensure_bound_directory(path, root_binding: nil, operation_hook: nil) ⇒ Object 



378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
 
# File 'lib/cohere/transcribe/state/io.rb', line 378

def ensure_bound_directory(path, root_binding: nil, operation_hook: nil)
  target = Pathname(path).expand_path.cleanpath
  bindings = []
  current = nil
  if root_binding
    root_binding.verify!
    relative = target.relative_path_from(root_binding.canonical_path)
    if relative.each_filename.first == ".."
      raise TranscriptionRuntimeError,
            "Publication directory escapes its bound root: #{target}"
    end
    current_binding = root_binding
    access_cursor = root_binding.canonical_path
    canonical_cursor = root_binding.canonical_path
    components = relative.each_filename.reject { |name| name == "." }
  else
    missing = []
    cursor = target
    until cursor.exist? || cursor.symlink?
      parent = cursor.dirname
      raise TranscriptionRuntimeError, "Cannot resolve publication directory #{target}" if parent == cursor

      missing.unshift(cursor.basename.to_s)
      cursor = parent
    end
    current_binding = DirectoryBinding.capture(cursor)
    access_cursor = current_binding.access_path
    canonical_cursor = current_binding.canonical_path
    components = missing
  end

  Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
    current = BoundDirectory.open(current_binding, guards: [current_binding])
    bindings << current
  end
  operation_hook&.call(:ancestor_opened, current.binding.access_path)
  components.each do |component|
    operation_hook&.call(:before_mkdir, access_cursor.join(component))
    current.mkdir(component)
    operation_hook&.call(:after_mkdir, access_cursor.join(component))
    access_cursor = access_cursor.join(component)
    canonical_cursor = canonical_cursor.join(component)
    Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
      current = current.open_child_directory(
        component,
        access_path: access_cursor,
        canonical_path: canonical_cursor
      )
      bindings << current
    end
  end
  bindings.each(&:verify!)
  current.binding
rescue ArgumentError => e
  raise TranscriptionRuntimeError, "Cannot bind publication directory #{path}: #{e.message}"
ensure
  bindings&.reverse_each(&:close)
end

.ensure_source_unchanged!(snapshot) ⇒ Object 



637
638
639
640
641
642
643
644
 
# File 'lib/cohere/transcribe/state/io.rb', line 637

def ensure_source_unchanged!(snapshot)
  return if SourceSnapshot.capture(snapshot.canonical_path) == snapshot

  raise TranscriptionRuntimeError, "Source changed while processing: #{snapshot.canonical_path}"
rescue SystemCallError, ArgumentError => e
  raise TranscriptionRuntimeError,
        "Cannot re-check source #{snapshot.canonical_path}: #{e.message}"
end

.envelope(payload) ⇒ Object 



470
471
472
473
474
475
476
 
# File 'lib/cohere/transcribe/state/io.rb', line 470

def envelope(payload)
  {
    "schema_version" => STATE_SCHEMA_VERSION,
    "payload_sha256" => Digest::SHA256.hexdigest(canonical_json(payload)),
    "payload" => payload
  }
end

.fingerprint(value) ⇒ Object 



132
133
134
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 132

def fingerprint(value)
  Digest::SHA256.hexdigest(canonical_json(value))
end

.finite_number(value, name) ⇒ Object 



197
198
199
200
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 197

def finite_number(value, name)
  validate!(value.is_a?(Numeric) && !value.is_a?(Complex) && value.finite?, "#{name} is not finite numeric data")
  value.to_f
end

.immutable_optional_string(value) ⇒ Object 



206
207
208
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 206

def immutable_optional_string(value)
  value&.dup&.freeze
end

.implementation_fingerprint(files) ⇒ Object 



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 116

def implementation_fingerprint(files)
  root = Pathname(__dir__).parent
  missing = files.reject { |relative| root.join(relative).file? }
  unless missing.empty?
    raise TranscriptionRuntimeError,
          "Implementation fingerprint references missing package artifacts: #{missing.join(", ")}"
  end

  fingerprint(
    "package_version" => VERSION,
    "artifacts_sha256" => files.to_h do |relative|
      [relative, Digest::SHA256.file(root.join(relative)).hexdigest]
    end
  )
end

.inspect_lock_path!(path) ⇒ Object 



179
180
181
182
183
184
185
186
 
# File 'lib/cohere/transcribe/state/locking.rb', line 179

def inspect_lock_path!(path)
  stat = path.lstat
  return if stat.file? && !stat.symlink?

  raise TranscriptionRuntimeError, "Output lock is not a regular file: #{path}"
rescue Errno::ENOENT
  nil
end

.json_value(value) ⇒ Object 



136
137
138
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 136

def json_value(value)
  value.is_a?(Pathname) ? value.to_s : value
end

.lock_registry_directory ⇒ Object 



120
121
122
123
 
# File 'lib/cohere/transcribe/state/locking.rb', line 120

def lock_registry_directory
  scope = Process.respond_to?(:uid) ? Process.uid.to_s : Digest::SHA256.hexdigest(Dir.home)[0, 16]
  Pathname(Dir.tmpdir).join("#{LOCK_DIRECTORY_PREFIX}-#{scope}")
end

.lock_target_for_outputs(output_paths) ⇒ Object 



103
104
105
106
107
108
109
110
111
 
# File 'lib/cohere/transcribe/state/locking.rb', line 103

def lock_target_for_outputs(output_paths)
  parent, stem = output_parent_and_stem(output_paths)
  identity = parent.join(stem).expand_path.cleanpath.to_s
  digest = Digest::SHA256.hexdigest(identity.downcase)
  OutputLockTarget.new(
    path: lock_registry_directory.join("#{digest}#{LOCK_SUFFIX}").freeze,
    identity: identity.freeze
  )
end

.open_lock_file(path) ⇒ Object 



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/cohere/transcribe/state/locking.rb', line 125

def open_lock_file(path)
  path = Pathname(path)
  succeeded = false
  validate_lock_directory!(path.dirname)
  inspect_lock_path!(path)
  flags = File::RDWR | File::CREAT
  flags |= File::NOFOLLOW if defined?(File::NOFOLLOW)
  flags |= File::CLOEXEC if defined?(File::CLOEXEC)
  descriptor = ::IO.sysopen(path.to_s, flags, 0o600)
  handle = File.new(descriptor, "r+", autoclose: true)
  descriptor = nil
  opened = verify_lock_identity!(
    path,
    handle,
    message: "Output lock changed while it was being opened or is not regular"
  )
  validate_owned_private_file!(opened, path)
  succeeded = true
  handle
rescue Errno::ELOOP, Errno::EISDIR, Errno::ENXIO => e
  raise TranscriptionRuntimeError, "Output lock is not a regular file: #{path}", cause: e
ensure
  if !succeeded && handle
    handle.close
  elsif descriptor
    ::IO.new(descriptor).close
  end
end

.output_parent_and_stem(output_paths) ⇒ Object

Raises:

  • (ArgumentError) 


441
442
443
444
445
446
447
448
449
450
451
452
453
454
 
# File 'lib/cohere/transcribe/state/io.rb', line 441

def output_parent_and_stem(output_paths)
  raise ArgumentError, "An output set must contain at least one path" if output_paths.empty?

  paths = output_paths.values.map { |path| Pathname(path) }
  parent = paths.first.dirname
  stem = paths.first.basename(paths.first.extname).to_s
  unless paths.drop(1).all? do |path|
    path.dirname == parent && path.basename(path.extname).to_s == stem
  end
    raise ArgumentError, "All formats in one output set must share a parent and stem"
  end

  [parent, stem]
end

.publication_mismatch_reason(payload, source_snapshot:, output_paths:, asr_contract_key:, render_contract_key:, directory_binding: nil, guard_bindings: nil) ⇒ Object 



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/cohere/transcribe/state/manifest.rb', line 88

def publication_mismatch_reason(payload, source_snapshot:, output_paths:, asr_contract_key:,
                                render_contract_key:, directory_binding: nil,
                                guard_bindings: nil)
  return "state is #{payload["kind"].inspect}, not published" unless payload["kind"] == "published"
  return "state marker ASR contract does not match" unless payload["asr_contract_key"] == asr_contract_key
  return "state marker render contract does not match" unless payload["render_contract_key"] == render_contract_key
  return "state marker source snapshot does not match" unless payload["source"] == source_snapshot.payload

  generation_id = payload["generation_id"]
  return "state marker generation ID is invalid" unless generation_id.is_a?(String) && !generation_id.empty?

  records = payload["outputs"]
  return "state marker output formats do not match" unless records.is_a?(Hash) && records.keys.sort == output_paths.keys.sort

  output_paths.each do |format, path_value|
    path = Pathname(path_value)
    record = records[format]
    return "state marker path for #{format} does not match" unless record.is_a?(Hash) && record["name"] == path.basename.to_s

    begin
      size, sha256 = if directory_binding
                       bound_output_record(
                         path,
                         directory_binding: directory_binding,
                         guard_bindings: guard_bindings
                       )
                     else
                       stat = path.lstat
                       return "#{format} output is missing or not regular" unless stat.file? && !stat.symlink?

                       [stat.size, Digest::SHA256.file(path).hexdigest]
                     end
      return "#{format} output does not match its state marker" if record["size"] != size || record["sha256"] != sha256
    rescue Errno::ENOENT
      return "#{format} output is missing or not regular"
    end
  end
  nil
end

.published_manifest_content(source_snapshot:, output_paths:, contents:, asr_contract_key:, render_contract_key:, generation_id:) ⇒ Object 



16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/cohere/transcribe/state/manifest.rb', line 16

def published_manifest_content(source_snapshot:, output_paths:, contents:, asr_contract_key:,
                               render_contract_key:, generation_id:)
  payload = published_payload(
    source_snapshot: source_snapshot,
    output_paths: output_paths,
    contents: contents,
    asr_contract_key: asr_contract_key,
    render_contract_key: render_contract_key,
    generation_id: generation_id
  )
  JSON.pretty_generate(envelope(payload)) << "\n"
end

.published_payload(source_snapshot:, output_paths:, contents:, asr_contract_key:, render_contract_key:, generation_id:) ⇒ Object

Raises:

  • (ArgumentError) 


29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/cohere/transcribe/state/manifest.rb', line 29

def published_payload(source_snapshot:, output_paths:, contents:, asr_contract_key:,
                      render_contract_key:, generation_id:)
  raise ArgumentError, "generation ID is invalid" unless generation_id.is_a?(String) && !generation_id.empty?

  {
    "kind" => "published",
    "generation_id" => generation_id,
    "asr_contract_key" => asr_contract_key,
    "render_contract_key" => render_contract_key,
    "source" => source_snapshot.payload,
    "updated_unix_seconds" => Time.now.to_f,
    "outputs" => output_paths.keys.sort.to_h do |format|
      content = contents.fetch(format)
      [
        format,
        {
          "name" => Pathname(output_paths.fetch(format)).basename.to_s,
          "size" => content.bytesize,
          "sha256" => Digest::SHA256.hexdigest(content)
        }
      ]
    end
  }
end

.render_contract_key(options) ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/cohere/transcribe/state/contracts.rb', line 95

def render_contract_key(options)
  formats = options.publication&.formats || []
  fingerprint(
    "contract_schema_version" => CONTRACT_SCHEMA_VERSION,
    "configuration" => {
      "alignment" => options.alignment,
      "align_batch_size" => options.align_batch_size,
      "align_dtype" => options.align_dtype,
      "formats" => formats.sort,
      "max_chars" => options.max_chars,
      "max_cue_dur" => options.max_cue_dur,
      "max_gap" => options.max_gap
    },
    "models" => {
      "align_revision" => options.alignment == "word" ? Alignment::ModelProvider::REVISION : nil
    },
    "output_schema_version" => Output::Publication::OUTPUT_SCHEMA_VERSION,
    "implementation_sha256" => implementation_fingerprint(RENDER_IMPLEMENTATION_FILES)
  )
end

.restore_asr_checkpoint(path:, source_snapshot:, asr_contract_key:, directory_binding: nil, guard_bindings: nil) ⇒ Object 



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 84

def restore_asr_checkpoint(path:, source_snapshot:, asr_contract_key:,
                           directory_binding: nil, guard_bindings: nil)
  payload, reason = decode_state(
    path,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  )
  return CheckpointRestore.new(checkpoint: nil, reason: reason.freeze) unless payload
  unless payload["kind"] == "asr_complete"
    return CheckpointRestore.new(
      checkpoint: nil,
      reason: "state is #{payload["kind"].inspect}, not an ASR checkpoint".freeze
    )
  end
  unless payload["asr_contract_key"] == asr_contract_key
    return CheckpointRestore.new(checkpoint: nil, reason: "ASR checkpoint contract does not match")
  end
  unless payload["source"] == source_snapshot.payload
    return CheckpointRestore.new(checkpoint: nil, reason: "state marker source snapshot does not match")
  end

  CheckpointRestore.new(checkpoint: validate_checkpoint(payload).freeze, reason: nil)
rescue TypeError, ArgumentError => e
  CheckpointRestore.new(checkpoint: nil, reason: "ASR checkpoint is invalid (#{e.message})".freeze)
end

.same_file?(first, second) ⇒ Boolean

Returns:

  • (Boolean) 


698
699
700
 
# File 'lib/cohere/transcribe/state/io.rb', line 698

def same_file?(first, second)
  first.dev == second.dev && first.ino == second.ino && first.file? && second.file?
end

.secure_digest_equal?(actual, expected) ⇒ Boolean

Returns:

  • (Boolean) 


702
703
704
705
706
707
708
 
# File 'lib/cohere/transcribe/state/io.rb', line 702

def secure_digest_equal?(actual, expected)
  return false unless actual.is_a?(String) && actual.bytesize == expected.bytesize

  difference = 0
  actual.bytes.zip(expected.bytes) { |left, right| difference |= left ^ right }
  difference.zero?
end

.securely_read_regular_file(path, directory_binding: nil, guard_bindings: nil) ⇒ Object 



646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
 
# File 'lib/cohere/transcribe/state/io.rb', line 646

def securely_read_regular_file(path, directory_binding: nil, guard_bindings: nil)
  if directory_binding
    return with_bound_parent(
      path,
      directory_binding: directory_binding,
      guard_bindings: guard_bindings
    ) do |bound, basename|
      bound.verify!
      handle = nil
      Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
        handle = bound.open_regular(basename)
      end
      opened = handle.stat
      raise TranscriptionRuntimeError, "State marker is too large: #{path}" if opened.size > MAX_STATE_BYTES

      contents = handle.read(MAX_STATE_BYTES + 1)
      raise TranscriptionRuntimeError, "State marker is too large: #{path}" if contents.bytesize > MAX_STATE_BYTES
      unless bound.same_regular_entry?(basename, opened)
        raise TranscriptionRuntimeError, "State marker changed while it was being read: #{path}"
      end

      bound.verify!
      contents
    ensure
      handle&.close
    end
  end

  before = path.lstat
  raise TranscriptionRuntimeError, "State marker is not a regular file: #{path}" unless before.file? && !before.symlink?
  raise TranscriptionRuntimeError, "State marker is too large: #{path}" if before.size > MAX_STATE_BYTES

  flags = File::RDONLY
  flags |= File::NOFOLLOW if defined?(File::NOFOLLOW)
  flags |= File::CLOEXEC if defined?(File::CLOEXEC)
  descriptor = ::IO.sysopen(path.to_s, flags)
  handle = ::IO.new(descriptor, "rb", autoclose: true)
  descriptor = nil
  opened = handle.stat
  current = path.lstat
  unless opened.file? && same_file?(opened, before) && same_file?(opened, current)
    raise TranscriptionRuntimeError, "State marker changed while it was being opened: #{path}"
  end

  handle.read(MAX_STATE_BYTES + 1).tap do |contents|
    raise TranscriptionRuntimeError, "State marker is too large: #{path}" if contents.bytesize > MAX_STATE_BYTES
  end
ensure
  handle&.close
  ::IO.new(descriptor).close if descriptor
end

.state_path_for_outputs(output_paths) ⇒ Object 



456
457
458
459
 
# File 'lib/cohere/transcribe/state/io.rb', line 456

def state_path_for_outputs(output_paths)
  parent, stem = output_parent_and_stem(output_paths)
  parent.join(".#{stem}#{STATE_SUFFIX}")
end

.time_nanoseconds(time) ⇒ Object 



437
438
439
 
# File 'lib/cohere/transcribe/state/io.rb', line 437

def time_nanoseconds(time)
  (time.to_i * 1_000_000_000) + time.nsec
end

.validate!(condition, message) ⇒ Object

Raises:

  • (ArgumentError) 


202
203
204
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 202

def validate!(condition, message)
  raise ArgumentError, message unless condition
end

.validate_checkpoint(payload) ⇒ Object 



110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 110

def validate_checkpoint(payload)
  generation_id = payload["generation_id"]
  validate!(generation_id.is_a?(String) && !generation_id.empty?, "generation ID is invalid")
  checkpoint = payload["checkpoint"]
  validate!(checkpoint.is_a?(Hash), "checkpoint is not an object")

  duration = finite_number(checkpoint["duration"], "duration")
  validate!(duration >= 0, "duration is invalid")
  segment_times = validated_spans(checkpoint["segment_times"], duration, "segment_times")
  speech_spans = validated_spans(checkpoint["speech_spans"], duration, "speech_spans")
  texts = checkpoint["segment_texts"]
  validate!(texts.is_a?(Array) && texts.all?(String), "segment_texts is invalid")
  validate!(texts.length == segment_times.length, "segment text/time counts differ")
  texts = texts.map { |text| text.dup.freeze }.freeze
  segment_count = segment_times.length

  generated_tokens = validated_generated_tokens(checkpoint.fetch("generated_tokens", []), segment_count)
  repetition = validated_indices(checkpoint.fetch("repetition_stopped_segments", []), segment_count,
                                 "repetition_stopped_segments")
  truncation = validated_indices(checkpoint.fetch("truncation_retried_segments", []), segment_count,
                                 "truncation_retried_segments")
  token_limit = validated_indices(checkpoint.fetch("token_limit_segments", []), segment_count,
                                  "token_limit_segments")
  optional_names = %w[
    decode_backend decode_fallback_reason vad_engine_actual vad_provider vad_fallback_reason
  ]
  optional_names.each do |name|
    value = checkpoint[name]
    validate!(value.nil? || value.is_a?(String), "checkpoint provenance strings are invalid")
  end
  provider_options = checkpoint["vad_provider_options"]
  validate!(provider_options.nil? || provider_options.is_a?(Hash), "vad_provider_options is invalid")

  Checkpoint.new(
    generation_id: generation_id.dup.freeze,
    duration: duration,
    segment_times: segment_times,
    speech_spans: speech_spans,
    segment_texts: texts,
    generated_tokens_by_segment: generated_tokens,
    repetition_stopped_segments: repetition,
    truncation_retried_segments: truncation,
    token_limit_segments: token_limit,
    decode_backend: immutable_optional_string(checkpoint["decode_backend"]),
    decode_fallback_reason: immutable_optional_string(checkpoint["decode_fallback_reason"]),
    vad_engine_actual: immutable_optional_string(checkpoint["vad_engine_actual"]),
    vad_provider: immutable_optional_string(checkpoint["vad_provider"]),
    vad_provider_options: deep_freeze_json(provider_options),
    vad_fallback_reason: immutable_optional_string(checkpoint["vad_fallback_reason"])
  )
end

.validate_lock_directory!(path) ⇒ Object 



154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/cohere/transcribe/state/locking.rb', line 154

def validate_lock_directory!(path)
  begin
    Dir.mkdir(path, 0o700)
  rescue Errno::EEXIST
    nil
  rescue SystemCallError => e
    raise TranscriptionRuntimeError, "Cannot prepare output lock directory #{path}: #{e.message}"
  end
  stat = path.lstat
  raise TranscriptionRuntimeError, "Output lock directory is not a real directory: #{path}" unless stat.directory? && !stat.symlink?

  return if Gem.win_platform?

  if Process.respond_to?(:uid) && stat.uid != Process.uid
    raise TranscriptionRuntimeError,
          "Output lock directory is not owned by the current user: #{path}"
  end
  return if stat.mode.nobits?(0o077)

  raise TranscriptionRuntimeError,
        "Output lock directory permissions must be private (0700): #{path}"
rescue Errno::ENOENT, Errno::ENOTDIR => e
  raise TranscriptionRuntimeError, "Cannot prepare output lock directory #{path}: #{e.message}"
end

.validate_owned_private_file!(stat, path) ⇒ Object

Raises:



188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/cohere/transcribe/state/locking.rb', line 188

def validate_owned_private_file!(stat, path)
  return if Gem.win_platform?

  if Process.respond_to?(:uid) && stat.uid != Process.uid
    raise TranscriptionRuntimeError, "Output lock is not owned by the current user: #{path}"
  end
  return if stat.mode.nobits?(0o077)

  raise TranscriptionRuntimeError,
        "Output lock permissions must be private (0600): #{path}"
end

.validated_generated_tokens(value, segment_count) ⇒ Object 



177
178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 177

def validated_generated_tokens(value, segment_count)
  validate!(value.is_a?(Array), "generated_tokens is invalid")
  seen = {}
  value.map do |row|
    valid = row.is_a?(Array) && row.length == 2 && row.all?(Integer)
    valid &&= row[0].between?(0, segment_count - 1) && row[1] >= 0
    valid &&= !seen.key?(row[0])
    validate!(valid, "generated_tokens contains an invalid or duplicate row")
    seen[row[0]] = true
    [row[0], row[1]].freeze
  end.freeze
end

.validated_indices(value, segment_count, name) ⇒ Object 



190
191
192
193
194
195
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 190

def validated_indices(value, segment_count, name)
  validate!(value.is_a?(Array), "#{name} is invalid")
  valid = value.all? { |index| index.is_a?(Integer) && index.between?(0, segment_count - 1) }
  validate!(valid && value.uniq.length == value.length, "#{name} is invalid or contains duplicate indices")
  value.dup.freeze
end

.validated_spans(value, duration, name) ⇒ Object 



162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 162

def validated_spans(value, duration, name)
  validate!(value.is_a?(Array), "#{name} is not a list")
  previous_end = nil
  value.map do |row|
    validate!(row.is_a?(Array) && row.length == 2, "#{name} contains an invalid row")
    start_time = finite_number(row[0], name)
    end_time = finite_number(row[1], name)
    valid = start_time.between?(0, end_time) && end_time <= duration + 1e-6
    valid &&= previous_end.nil? || start_time >= previous_end
    validate!(valid, "#{name} contains an overlapping, out-of-order, or out-of-range row")
    previous_end = end_time
    [start_time, [end_time, duration].min].freeze
  end.freeze
end

.verify_lock_identity!(path, handle, message:) ⇒ Object 



200
201
202
203
204
205
206
207
208
 
# File 'lib/cohere/transcribe/state/locking.rb', line 200

def verify_lock_identity!(path, handle, message:)
  opened = handle.stat
  current = Pathname(path).lstat
  return opened if opened.file? && same_file?(opened, current)

  raise TranscriptionRuntimeError, "#{message}: #{path}"
rescue SystemCallError => e
  raise TranscriptionRuntimeError, "#{message}: #{path}", cause: e
end

.verify_published_outputs(source_snapshot:, output_paths:, state_path:, asr_contract_key:, render_contract_key:, directory_binding: nil, guard_bindings: nil) ⇒ Object 



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/cohere/transcribe/state/manifest.rb', line 54

def verify_published_outputs(source_snapshot:, output_paths:, state_path:, asr_contract_key:,
                             render_contract_key:, directory_binding: nil,
                             guard_bindings: nil)
  payload, reason = decode_state(
    state_path,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  )
  return PublicationVerification.new(verified: false, generation_id: nil, reason: reason.freeze) unless payload

  mismatch = publication_mismatch_reason(
    payload,
    source_snapshot: source_snapshot,
    output_paths: output_paths,
    asr_contract_key: asr_contract_key,
    render_contract_key: render_contract_key,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  )
  return PublicationVerification.new(verified: false, generation_id: nil, reason: mismatch.freeze) if mismatch

  PublicationVerification.new(
    verified: true,
    generation_id: payload.fetch("generation_id").dup.freeze,
    reason: nil
  )
rescue EncodingError, SystemCallError, TypeError, ArgumentError => e
  PublicationVerification.new(
    verified: false,
    generation_id: nil,
    reason: "cannot verify output generation (#{e.class}: #{e.message})".freeze
  )
end

.with_bound_parent(path, directory_binding: nil, guard_bindings: nil) ⇒ Object 



360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
 
# File 'lib/cohere/transcribe/state/io.rb', line 360

def with_bound_parent(path, directory_binding: nil, guard_bindings: nil)
  path = Pathname(path).expand_path
  directory_binding ||= DirectoryBinding.capture(path.dirname)
  unless [directory_binding.access_path, directory_binding.canonical_path].include?(path.dirname)
    raise TranscriptionRuntimeError,
          "Publication path is outside its planned parent: #{path}"
  end
  guards = Array(guard_bindings).compact
  guards << directory_binding
  bound = nil
  Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
    bound = BoundDirectory.open(directory_binding, guards: guards.uniq)
  end
  yield bound, path.basename.to_s
ensure
  bound&.close
end

.with_output_lock(target, blocking: false) ⇒ Object 



113
114
115
116
117
118
 
# File 'lib/cohere/transcribe/state/locking.rb', line 113

def with_output_lock(target, blocking: false)
  lock = OutputSetLock.acquire(target, blocking: blocking)
  yield lock
ensure
  lock&.release
end

.write_asr_checkpoint(path:, result:, source_snapshot:, asr_contract_key:, speech_spans:, vad_provider_options: nil, generation_id: nil, directory_binding: nil, guard_bindings: nil) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/cohere/transcribe/state/checkpoint.rb', line 33

def write_asr_checkpoint(path:, result:, source_snapshot:, asr_contract_key:,
                         speech_spans:, vad_provider_options: nil, generation_id: nil,
                         directory_binding: nil, guard_bindings: nil)
  generation_id = generation_id.to_s
  generation_id = SecureRandom.hex(16) if generation_id.empty?
  payload = asr_checkpoint_payload(
    result: result,
    source_snapshot: source_snapshot,
    asr_contract_key: asr_contract_key,
    speech_spans: speech_spans,
    vad_provider_options: vad_provider_options,
    generation_id: generation_id
  )
  write_state_atomic(
    path,
    payload,
    source_snapshot: source_snapshot,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  )
  generation_id.freeze
end

.write_state_atomic(path, payload, source_snapshot:, directory_binding: nil, guard_bindings: nil, operation_hook: nil, rename: nil) ⇒ Object 



503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
 
# File 'lib/cohere/transcribe/state/io.rb', line 503

def write_state_atomic(path, payload, source_snapshot:, directory_binding: nil,
                       guard_bindings: nil, operation_hook: nil, rename: nil)
  path = Pathname(path).expand_path
  directory_binding ||= ensure_bound_directory(path.dirname)
  with_bound_parent(
    path,
    directory_binding: directory_binding,
    guard_bindings: guard_bindings
  ) do |bound, basename|
    operation_hook&.call(:directory_opened, path)
    temporary_name = nil
    temporary = nil
    backup_name = nil
    backup = nil
    source = nil
    committed = false
    preserved_backup = false
    primary_failed = false
    begin
      mode = bound_state_mode(bound, basename)
      Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
        temporary_name, temporary = bound.create_temporary(basename, ".tmp")
      end
      temporary.chmod(mode)
      operation_hook&.call(:before_stage_write, path)
      temporary.write(JSON.pretty_generate(envelope(payload)))
      temporary.write("\n")
      temporary.flush
      temporary.fsync
      temporary.close
      temporary = nil
      bound.verify!

      operation_hook&.call(:before_backup_open, path)
      Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
        source = begin
          bound.open_regular(basename)
        rescue Errno::ENOENT
          nil
        end
      end
      if source
        Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
          backup_name, backup = bound.create_temporary(basename, ".bak")
        end
        begin
          backup.chmod(source.stat.mode & 0o7777)
          operation_hook&.call(:before_backup_copy, path)
          IO.copy_stream(source, backup)
          backup.flush
          backup.fsync
        ensure
          source.close
          backup.close
        end
      end

      ensure_source_unchanged!(source_snapshot)
      bound.verify!
      operation_hook&.call(:before_rename, path)
      Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
        if rename
          committed = true
          rename.call(bound, temporary_name, basename, :commit)
        else
          bound.rename(temporary_name, basename)
          committed = true
        end
      end
      operation_hook&.call(:after_rename, path)
      bound.verify!
      bound.fsync
      path
    rescue Exception => e # rubocop:disable Lint/RescueException -- rollback includes interrupts
      primary_failed = true
      if committed
        begin
          Thread.handle_interrupt(DEFERRED_PUBLICATION_EXCEPTIONS) do
            bound.unlink(basename, missing_ok: true)
            if backup_name
              if rename
                rename.call(bound, backup_name, basename, :rollback)
              else
                bound.rename(backup_name, basename)
              end
            end
            backup_name = nil
          end
          bound.fsync
        rescue SystemCallError, TranscriptionRuntimeError => rollback_error
          retained = backup_name ? bound.display_path(backup_name) : nil
          preserved_backup = !backup_name.nil?
          raise TranscriptionRuntimeError,
                "State commit failed and rollback was incomplete (#{rollback_error.message}); " \
                "preserved backup: #{retained}",
                cause: e
        end
      end
      raise e
    ensure
      cleanup_failure = nil
      cleanup_operations = [
        -> { temporary&.close unless temporary&.closed? },
        -> { backup&.close unless backup&.closed? },
        -> { bound.unlink(temporary_name, missing_ok: true) if temporary_name },
        lambda do
          bound.unlink(backup_name, missing_ok: true) if backup_name && !preserved_backup
        end
      ]
      cleanup_operations.each do |operation|
        operation.call
      rescue SystemCallError, IOError => e
        cleanup_failure ||= e
      end
      raise cleanup_failure if cleanup_failure && !primary_failed
    end
  end
end