event API

Luna-Flow/mare_mark/event はランナーが出力するものを受け取ります。ObservationSink は 5 つのコールバックからなるレコードです。このパッケージは、インメモリのシンク、バッファリングする JSONL シンク、ストリーミングする JSONL シンク、およびファンアウトのコンビネータを提供します。JSONL 形式は、レポートとリプレイが読み込む監査記録です。event の設計も参照してください。

ソース: src/event/event.mbt。

import {
  "Luna-Flow/mare_mark/model",
  "Luna-Flow/mare_mark/event",
}

シンク

ObservationSink

ObservationSink はランナーとストレージの間のインターフェースです。

pub struct ObservationSink {
  emit_observation : (@model.Observation) -> Unit
  emit_validation : (@model.Validation) -> Unit
  emit_failure : (@model.ValidationFailure) -> Unit
  emit_calibration : (@model.CalibrationEvent) -> Unit
  finish : (@model.RunSummary) -> String
}
pub fn ObservationSink::new((@model.Observation) -> Unit, (@model.Validation) -> Unit, (@model.CalibrationEvent) -> Unit, (@model.RunSummary) -> String, emit_failure? : (@model.ValidationFailure) -> Unit) -> Self

finish は最後に一度だけ実行の要約を受け取り、ロケーション文字列を返します。ランナーはそれを RunSummary.artifact_location に格納します。new の引数の順序に注意してください。観測、検証、キャリブレーション、終了、そして省略可能な失敗コールバックの順で、失敗コールバックの既定は失敗を無視します。

test "a counting sink" {
  let count = Ref(0)
  let sink = @event.ObservationSink::new(
    _ => count.val += 1,
    _ => (),
    _ => (),
    summary => "counted://" + summary.run_id,
  )
  (sink.emit_observation)(
    @model.Observation::new("c", "a", "1", 0, 0, 0, Confirmatory, 1.5, 10, Kept, ExcludedFromMeasurement, true),
  )
  inspect(count.val, content="1")
  inspect((sink.finish)(@model.RunSummary::new("r", 1, 0, 0, true, None)), content="counted://r")
}

InMemorySink

InMemorySink はすべてのイベントを配列に保持します。

pub struct InMemorySink {
  observations : Array[@model.Observation]
  validations : Array[@model.Validation]
  failures : Array[@model.ValidationFailure]
  calibrations : Array[@model.CalibrationEvent]
}
pub fn InMemorySink::new() -> Self
pub fn InMemorySink::as_sink(Self) -> ObservationSink

as_sink は配列に追加します。その finish は "memory://run/" + run_id を返し、要約は保存しません。

JsonlSink

JsonlSink はすべてのイベントを 1 行の JSON として保持します。

pub struct JsonlSink {
  lines : Array[String]
}
pub fn JsonlSink::new() -> Self
pub fn JsonlSink::as_sink(Self) -> ObservationSink
pub fn JsonlSink::to_jsonl(Self) -> String

as_sink は要約を含め、イベントごとに 1 行を追加します。その finish は "jsonl://memory/" + run_id を返します。to_jsonl は行を "\n" で連結し、末尾に改行を付けません。

test "JSONL lines" {
  let jsonl = @event.JsonlSink::new()
  let sink = jsonl.as_sink()
  (sink.emit_calibration)(@model.CalibrationEvent::new("a", 0, 64, 1012.5, 1000.0, 1))
  inspect(
    jsonl.to_jsonl(),
    content="{\"artifact_version\":\"mmka_1\",\"type\":\"calibration\",\"implementation\":\"a\",\"dataset_id\":0,\"batch_iterations\":64,\"elapsed_us\":1012.5,\"target_elapsed_us\":1000,\"retries\":1}",
  )
}

streaming_jsonl

streaming_jsonl は、すべてのイベント行を即座に書き出すシンクを返します。

pub fn streaming_jsonl((String) -> Unit, String) -> ObservationSink

第 1 引数は改行を含まない各行を受け取り、第 2 引数は finish が返すロケーションです。実行の進行に合わせてファイルやパイプに追記するのに使います。こうすればクラッシュしてもストリームの有効な先頭部分が残ります。

test "stream lines as they happen" {
  let written : Array[String] = []
  let sink = @event.streaming_jsonl(line => written.push(line), "file://events.jsonl")
  let location = (sink.finish)(@model.RunSummary::new("r", 0, 0, 0, true, None))
  inspect(location, content="file://events.jsonl")
  inspect(written[0].contains("\"type\":\"summary\""), content="true")
}

tee

tee はすべてのイベントを 2 つのシンクに送ります。

pub fn tee(ObservationSink, ObservationSink) -> ObservationSink

イベントはまず左のシンクに送られます。finish は両方を呼び出し、右のシンクのロケーションを返します。

JSONL レコード

各行は "artifact_version": "mmka_1" と "type" を持つオブジェクトです:

type	フィールド
observation	case, implementation, implementation_version, dataset_id, repetition_id, block_id, phase, elapsed_us, iterations, batch_sink, valid
validation	status、implementation、oracle、scale。証拠がある場合はさらに case、dataset_id、step_id、operation、operands、context、rounding、expected、actual、expected_kind、actual_kind、expected_flags、actual_flags、trap、stderr、exit_code（存在する場合）、fingerprint、implementation_version、replay_command、replay_arguments、replay_timeout_ms
validation_failure	検証のすべてのフィールドに加えて、seed（10 進文字列）、original_fingerprint、minimal_fingerprint、shrink_path、minimal_input
calibration	implementation, dataset_id, batch_iterations, elapsed_us, target_elapsed_us, retries
summary	run_id、observation_count、validation_count、calibration_count、complete、passed_count、failed_count、unsupported_count、expected_difference_count、および判明している場合は semantic、performance、provenance オブジェクトを持つ environment

status は valid、invalid、skipped、expected_difference、unsupported、infrastructure_failure のいずれかで、ステータスの理由文字列は書き出されません。phase は exploratory または confirmatory で、batch_sink は kept または discarded:<reason> です。観測の setup_timing は書き出されません。