/** Typed projection over Harn's canonical durable event log. */
import { check_versioned } from "std/artifacts/typed"
import {
PairedObservation,
decide_experiment,
paired_observation_cell_key,
validate_paired_observation,
} from "std/eval/experiment/decision"
import { validate_compiled_hypothesis_plan } from "std/eval/hypothesis/compiler"
import { HypothesisPlan } from "std/eval/hypothesis/contracts"
import {
HypothesisApprovalRecorded,
HypothesisDecisionRecorded,
HypothesisEventAuthorityKind,
HypothesisEventContent,
HypothesisExecutionDrift,
HypothesisInvalidated,
HypothesisLedgerAppendReceipt,
HypothesisLedgerEvent,
HypothesisLedgerRecord,
HypothesisLedgerSnapshot,
HypothesisLedgerSnapshotRead,
HypothesisObservationRecorded,
HypothesisPlanRegistered,
HypothesisRegressionObserved,
HypothesisRelationshipRecorded,
HypothesisRunTransition,
hypothesis_event_content_contract,
} from "std/eval/hypothesis/ledger_contracts"
const HYPOTHESIS_LEDGER_TOPIC = "hypotheses.events.v1"
fn __fingerprint(value: unknown) -> string {
return "sha256:" + sha256(json_stringify(value))
}
fn __non_empty(value: string, field: string) {
if trim(value) == "" {
throw "hypothesis ledger: " + field + " must be non-empty"
}
}
fn __validate_content(content: HypothesisEventContent) {
__non_empty(content.event_id, "event_id")
__non_empty(content.hypothesis_id, "hypothesis_id")
__non_empty(content.occurred_at, "occurred_at")
__non_empty(content.actor, "actor")
__non_empty(content.source, "source")
if content.payload.kind == "observation_recorded" {
if content.payload.spend_delta_usd < 0.0
|| content.payload.compute_delta_ms < 0
|| content.payload.token_delta < 0
|| content.payload.api_call_delta < 0 {
throw "hypothesis ledger: evidence resource deltas must be non-negative"
}
if content.payload.telemetry_status != "observed"
|| len(content.payload.capability_degradations)
> 0 {
throw "hypothesis ledger: degraded observations cannot enter canonical evidence; record drift or invalidation instead"
}
}
if content.payload.kind == "decision_recorded" {
if content.payload.elapsed_ms < 0 {
throw "hypothesis ledger: decision elapsed time must be non-negative"
}
if (content.payload.belief_before != nil
&& (content.payload.belief_before < 0.0
|| content.payload
.belief_before
> 1.0))
|| (content.payload.belief_after != nil
&& (content.payload.belief_after < 0.0
|| content.payload
.belief_after
> 1.0)) {
throw "hypothesis ledger: recorded belief probabilities must be in [0, 1]"
}
}
if content.payload.kind == "approval_recorded" {
__non_empty(content.payload.approval_id, "approval_id")
__non_empty(content.payload.plan_fingerprint, "plan_fingerprint")
__non_empty(content.payload.rationale, "approval rationale")
}
}
fn __transition_allowed(previous: string?, next: string) -> bool {
if previous == nil {
return next == "scheduled"
}
if previous == "scheduled" {
return next == "running" || next == "cancelled" || next == "failed"
}
if previous == "running" {
return contains(["paused", "completed", "cancelled", "failed", "invalid"], next)
}
if previous == "paused" {
return contains(["running", "cancelled", "failed", "invalid"], next)
}
return false
}
fn __validate_observation_admission(
plan: HypothesisPlan,
snapshot: HypothesisLedgerSnapshot,
observation_ids: dict<string, bool>,
observed_cells: dict<string, bool>,
observation: HypothesisObservationRecorded,
) {
if plan.kind != "registered_experiment" || snapshot.run_state != "running" {
throw "hypothesis ledger: observations require a running registered experiment"
}
if observation_ids[observation.observation_id] ?? false {
throw "hypothesis ledger: observation identity is already registered"
}
if observation.operation_receipt_id != nil {
__non_empty(observation.operation_receipt_id, "observation operation_receipt_id")
}
const spend = snapshot.total_spend_usd + observation.spend_delta_usd
if spend > plan.design.budget.max_spend_usd
|| snapshot.total_compute_ms
+ observation.compute_delta_ms
> plan.design.budget.max_compute_ms
|| snapshot.total_tokens + observation.token_delta
> plan.design
.budget.max_tokens
|| snapshot.total_api_calls + observation.api_call_delta
> plan.design.budget
.max_api_calls {
throw "hypothesis ledger: observation exceeds the registered resource budget"
}
const _ = validate_paired_observation(plan.registration, observation.observation, observed_cells)
}
fn __validate_decision_admission(
plan: HypothesisPlan,
snapshot: HypothesisLedgerSnapshot,
observations: list<PairedObservation>,
decision: HypothesisDecisionRecorded,
) {
if plan.kind != "registered_experiment" || snapshot.run_state != "completed" {
throw "hypothesis ledger: a terminal decision requires a completed registered experiment"
}
if snapshot.completion == nil {
throw "hypothesis ledger: a terminal decision requires a typed completion reason"
}
const completion = snapshot.completion
if completion.kind == "wall_clock" {
if decision.elapsed_ms != completion.elapsed_ms
|| completion.elapsed_ms
< plan.design.budget
.max_wall_clock_ms {
throw "hypothesis ledger: wall-clock exhaustion does not match the registered ceiling"
}
} else if decision.elapsed_ms > plan.design.budget.max_wall_clock_ms {
throw "hypothesis ledger: decision exceeds the registered wall-clock budget"
}
const expected = decide_experiment(
plan.registration,
{
observations: observations,
phase_spend_usd: snapshot.total_spend_usd,
budget_spent: completion.kind != "statistical",
},
)
if __fingerprint(decision.decision) != __fingerprint(expected) {
throw "hypothesis ledger: decision is not the canonical result of the registered evidence"
}
}
fn __max_trials_exhausted(plan: HypothesisPlan, snapshot: HypothesisLedgerSnapshot) -> bool {
if plan.kind != "registered_experiment" {
return false
}
const expected = len(plan.registration.candidates)
* len(plan.registration.case_set.cases)
* plan.registration.budget.max_trials_per_case
return snapshot.observation_count == expected
}
fn __validate_completion(
plan: HypothesisPlan,
snapshot: HypothesisLedgerSnapshot,
observations: list<PairedObservation>,
transition: HypothesisRunTransition,
) {
if transition.state != "completed" {
if transition.completion != nil {
throw "hypothesis ledger: only a completed run may carry completion evidence"
}
return
}
if transition.completion == nil {
throw "hypothesis ledger: a completed run requires typed completion evidence"
}
const completion = transition.completion
if completion.kind == "statistical" {
const decision = decide_experiment(
plan.registration,
{observations: observations, phase_spend_usd: snapshot.total_spend_usd, budget_spent: false},
)
if decision.verdict == "RUNNING" {
throw "hypothesis ledger: statistical completion requires a terminal canonical decision"
}
return
}
if completion.kind == "max_trials" {
if !__max_trials_exhausted(plan, snapshot) {
throw "hypothesis ledger: max-trial completion requires every registered trial cell"
}
return
}
__non_empty(completion.receipt_id, "completion receipt_id")
if !transition.receipt_ids.contains(completion.receipt_id) {
throw "hypothesis ledger: native completion receipt must be bound to the lifecycle event"
}
if completion.kind == "wall_clock"
&& completion.elapsed_ms
< plan.design.budget.max_wall_clock_ms {
throw "hypothesis ledger: wall-clock completion precedes the registered ceiling"
}
}
fn __validate_plan_registration(event_count: int, content: HypothesisEventContent) {
const _ = validate_compiled_hypothesis_plan(content.payload.plan)
if event_count != 0
|| content.plan_id != content.payload.plan.plan_id
|| content.hypothesis_id
!= content.payload.plan
.hypothesis_id
|| content.run_id != nil {
throw "hypothesis ledger: plan registration must be the first matching aggregate event"
}
}
fn __validate_approval_admission(
plan: HypothesisPlan,
snapshot: HypothesisLedgerSnapshot,
content: HypothesisEventContent,
approval: HypothesisApprovalRecorded,
) {
if !plan.design.approval_required
|| content.plan_id != plan.plan_id
|| content.run_id != nil
|| snapshot.active_run_id != nil
|| snapshot.approval_status != nil
|| approval.approval_id != plan.design.approval_id
|| approval.plan_fingerprint != plan.fingerprint {
throw "hypothesis ledger: approval does not match the pending registered plan"
}
}
fn __validate_lifecycle_identity(
plan: HypothesisPlan,
snapshot: HypothesisLedgerSnapshot,
content: HypothesisEventContent,
) {
if content.plan_id != plan.plan_id || content.run_id == nil || trim(content.run_id) == "" {
throw "hypothesis ledger: lifecycle event identity does not match the registered plan and run"
}
if snapshot.active_run_id != nil && content.run_id != snapshot.active_run_id {
throw "hypothesis ledger: lifecycle event run does not match the active run"
}
}
fn __validate_admission(
previous_fingerprint: string?,
snapshot: HypothesisLedgerSnapshot,
observations: list<PairedObservation>,
observation_ids: dict<string, bool>,
observed_cells: dict<string, bool>,
event: HypothesisLedgerEvent,
) {
const content = event.content
if content.predecessor_fingerprint != previous_fingerprint {
throw "hypothesis ledger: event predecessor does not match the latest aggregate event"
}
if content.payload.kind == "plan_registered" {
__validate_plan_registration(snapshot.event_count, content)
return
}
if snapshot.plan == nil {
throw "hypothesis ledger: lifecycle events require a registered plan"
}
const plan = snapshot.plan
if content.payload.kind == "approval_recorded" {
__validate_approval_admission(plan, snapshot, content, content.payload)
return
}
__validate_lifecycle_identity(plan, snapshot, content)
if content.payload.kind == "run_transition" {
if plan.kind != "registered_experiment" {
throw "hypothesis ledger: observe-only plans cannot enter executable run states"
}
if snapshot.active_run_id == nil
&& plan.design.approval_required
&& snapshot.approval_status != "approved" {
throw "hypothesis ledger: the registered plan requires an approved host decision before scheduling"
}
if !__transition_allowed(snapshot.run_state, content.payload.state) {
throw "hypothesis ledger: invalid run-state transition"
}
__validate_completion(plan, snapshot, observations, content.payload)
return
}
if content.payload.kind == "observation_recorded" {
__validate_observation_admission(
plan,
snapshot,
observation_ids,
observed_cells,
content.payload,
)
return
}
if content.payload.kind == "decision_recorded" {
if snapshot.decision != nil {
throw "hypothesis ledger: a terminal decision is already registered for this run"
}
__validate_decision_admission(plan, snapshot, observations, content.payload)
}
}
/**
* Validate and fingerprint portable event content before persistence.
*
* @effects: []
* @errors: [validation]
* @api_stability: experimental
*/
pub fn hypothesis_event(value: unknown) -> HypothesisLedgerEvent {
const checked = check_versioned(value, hypothesis_event_content_contract())
if !is_ok(checked) {
throw "hypothesis ledger: invalid event content: " + unwrap_err(checked).detail
}
const content: HypothesisEventContent = unwrap(checked).value
__validate_content(content)
return {content: content, fingerprint: __fingerprint(content)}
}
fn __headers(content: HypothesisEventContent, fingerprint: string) -> dict<string, string> {
let headers: dict<string, string> = {
schema: content.schema,
hypothesis_id: content.hypothesis_id,
event_id: content.event_id,
fingerprint: fingerprint,
}
if content.plan_id != nil {
headers = headers + {plan_id: content.plan_id}
}
if content.run_id != nil {
headers = headers + {run_id: content.run_id}
}
return headers
}
fn __authority_kind(event: HypothesisLedgerEvent) -> HypothesisEventAuthorityKind {
if event.content.payload.kind == "plan_registered" {
return "plan_admission"
}
if event.content.payload.kind == "approval_recorded" {
return "native_approval"
}
if event.content.payload.kind == "observation_recorded" {
return "native_observation"
}
return "lifecycle_audit"
}
fn __authority_plan_fingerprint(
snapshot: HypothesisLedgerSnapshot,
event: HypothesisLedgerEvent,
) -> string {
if event.content.payload.kind == "plan_registered" {
return event.content.payload.plan.fingerprint
}
if snapshot.plan == nil {
throw "hypothesis ledger: lifecycle authority requires a registered plan"
}
return snapshot.plan.fingerprint
}
/**
* Atomically append one event identity. Replays return the original cursor;
* reusing an identity for different content fails closed.
*
* @effects: [state.write@harness.runtime]
* @errors: [backend, validation]
* @api_stability: experimental
*/
pub fn hypothesis_ledger_append(
obs: HarnessObs,
event: HypothesisLedgerEvent,
authority_proof: resource,
) -> HypothesisLedgerAppendReceipt {
__validate_content(event.content)
const expected = __fingerprint(event.content)
if event.fingerprint != expected {
throw "hypothesis ledger: event fingerprint does not match canonical content"
}
const state = __read_verified_snapshot(obs, event.content.hypothesis_id)
const existing = state.records
let replay = false
for record in existing {
if record.event.content.event_id == event.content.event_id {
if record.event.fingerprint != event.fingerprint {
throw "hypothesis ledger: event identity conflicts with different persisted content"
}
replay = true
}
}
if !replay {
__validate_admission(
state.previous_fingerprint,
state.snapshot,
state.observations,
state.observation_ids,
state.observed_cells,
event,
)
}
const plan_fingerprint = __authority_plan_fingerprint(state.snapshot, event)
const idempotency_key = event.fingerprint
const outcome = obs.hypothesis_event_append(
authority_proof,
"hypothesis." + event.content.payload.kind,
idempotency_key,
state.last_hash,
event.fingerprint,
plan_fingerprint,
event.content.hypothesis_id,
event.content.run_id,
event,
__headers(event.content, event.fingerprint),
)
if outcome.event?.payload?.fingerprint != event.fingerprint {
throw "hypothesis ledger: event identity conflicts with different persisted content"
}
const record_hash = outcome.event?.headers?.["harn.provenance.record_hash"]
if type_of(record_hash) != "string" || record_hash == "" {
throw "hypothesis ledger: persisted event is missing its integrity hash"
}
return {
schema: "harn.hypothesis.ledger_append.v1",
topic: HYPOTHESIS_LEDGER_TOPIC,
cursor: outcome.event_id,
inserted: outcome.inserted,
event_id: event.content.event_id,
fingerprint: event.fingerprint,
record_hash: record_hash,
}
}
fn __decode_record(raw) -> HypothesisLedgerRecord {
const event = hypothesis_event(raw?.payload?.content)
if event.fingerprint != raw?.payload?.fingerprint {
throw "hypothesis ledger: persisted event fingerprint is corrupt"
}
const record_hash = raw?.headers?.["harn.provenance.record_hash"]
if type_of(record_hash) != "string" || record_hash == "" {
throw "hypothesis ledger: persisted event is missing its integrity hash"
}
const authority_schema = raw?.headers?.["harn.hypothesis.authority.schema"]
const authority_kind = raw?.headers?.["harn.hypothesis.authority.kind"]
const authority_plan_fingerprint = raw?.headers?.["harn.hypothesis.plan_fingerprint"]
if authority_schema != "harn.hypothesis-event-authority.v1"
|| !contains(
["plan_admission", "native_approval", "native_observation", "lifecycle_audit"],
authority_kind,
)
|| type_of(authority_plan_fingerprint) != "string"
|| trim(authority_plan_fingerprint) == "" {
throw "hypothesis ledger: persisted event is missing native authority provenance"
}
const typed_authority: HypothesisEventAuthorityKind = authority_kind
return {
cursor: raw.cursor,
record_hash: record_hash,
authority_kind: typed_authority,
authority_plan_fingerprint: authority_plan_fingerprint,
event: event,
}
}
fn __validate_record_authority(snapshot: HypothesisLedgerSnapshot, record: HypothesisLedgerRecord) {
if record.authority_kind != __authority_kind(record.event) {
throw "hypothesis ledger: native authority kind does not match the event kind"
}
const expected = __authority_plan_fingerprint(snapshot, record.event)
if record.authority_plan_fingerprint != expected {
throw "hypothesis ledger: native authority is bound to a different plan"
}
}
/**
* Read one hypothesis history from the globally ordered portable ledger.
*
* @effects: [state.read@harness.runtime]
* @errors: [backend, validation]
* @api_stability: experimental
*/
pub fn hypothesis_ledger_read(
obs: HarnessObs,
hypothesis_id: string,
) -> list<HypothesisLedgerRecord> {
__non_empty(hypothesis_id, "hypothesis_id")
return __read_verified_snapshot(obs, hypothesis_id).records
}
/**
* Read and project one hypothesis in a single pass.
*
* Integrity metadata covers the retained global hypothesis topic from its
* retained genesis through the reported head. It does not prove that an
* external checkpoint retained an earlier head.
*
* @effects: [state.read@harness.runtime]
* @errors: [backend, validation]
* @api_stability: experimental
*/
pub fn hypothesis_ledger_snapshot(
obs: HarnessObs,
hypothesis_id: string,
) -> HypothesisLedgerSnapshotRead {
__non_empty(hypothesis_id, "hypothesis_id")
const state = __read_verified_snapshot(obs, hypothesis_id)
return {
schema: "harn.hypothesis.ledger_snapshot.v1",
integrity: {
scope: "retained_topic_chain",
verified: true,
scanned_event_count: state.scanned_event_count,
last_hash: state.last_hash,
},
snapshot: state.snapshot,
observations: state.observations,
observed_cells: state.observed_cells,
}
}
fn __read_verified_snapshot(obs: HarnessObs, hypothesis_id: string) {
const retained = obs.hypothesis_event_snapshot(hypothesis_id)
if !retained.verified {
throw "hypothesis ledger: persisted event-log integrity verification failed: "
+ join(
retained.errors,
"; ",
)
}
let records: list<HypothesisLedgerRecord> = []
for raw in retained.records {
records = records + [__decode_record(raw)]
}
const folded = __fold_records(records, hypothesis_id)
return folded
+ {
records: records,
last_hash: retained.last_hash,
scanned_event_count: retained.scanned_event_count,
}
}
fn __empty_snapshot(hypothesis_id: string) -> HypothesisLedgerSnapshot {
return {
schema: "harn.hypothesis.snapshot.v1",
hypothesis_id: hypothesis_id,
event_count: 0,
latest_cursor: nil,
latest_event_fingerprint: nil,
plan: nil,
active_run_id: nil,
approval_status: nil,
approval_id: nil,
run_state: nil,
completion: nil,
decision: nil,
observation_count: 0,
observed_arms: [],
receipt_ids: [],
drift: [],
relationships: [],
invalidations: [],
regressions: [],
total_spend_usd: 0.0,
total_compute_ms: 0,
total_tokens: 0,
total_api_calls: 0,
telemetry_degradations: [],
}
}
fn __apply_record(
snapshot: HypothesisLedgerSnapshot,
record: HypothesisLedgerRecord,
) -> HypothesisLedgerSnapshot {
const payload = record.event.content.payload
const updated = snapshot
+ {
event_count: snapshot.event_count + 1,
latest_cursor: record.cursor,
latest_event_fingerprint: record.event.fingerprint,
}
if payload.kind == "plan_registered" {
const registered: HypothesisPlanRegistered = payload
return updated + {plan: registered.plan}
}
if payload.kind == "approval_recorded" {
const approval: HypothesisApprovalRecorded = payload
return updated + {approval_status: approval.decision, approval_id: approval.approval_id}
}
if payload.kind == "run_transition" {
const transition: HypothesisRunTransition = payload
return updated
+ {
active_run_id: snapshot.active_run_id ?? record.event.content.run_id,
run_state: transition.state,
completion: transition.completion,
}
}
if payload.kind == "observation_recorded" {
const evidence: HypothesisObservationRecorded = payload
return updated
+ {
observation_count: snapshot.observation_count + 1,
total_spend_usd: snapshot.total_spend_usd + evidence.spend_delta_usd,
total_compute_ms: snapshot.total_compute_ms + evidence.compute_delta_ms,
total_tokens: snapshot.total_tokens + evidence.token_delta,
total_api_calls: snapshot.total_api_calls + evidence.api_call_delta,
}
}
if payload.kind == "decision_recorded" {
const decision: HypothesisDecisionRecorded = payload
return updated + {decision: decision}
}
if payload.kind == "relationship_recorded" {
const linked: HypothesisRelationshipRecorded = payload
return updated + {relationships: snapshot.relationships + [linked.relationship]}
}
if payload.kind == "execution_drift" {
const drift: HypothesisExecutionDrift = payload
return updated + {drift: snapshot.drift + [drift]}
}
if payload.kind == "invalidated" {
const invalidated: HypothesisInvalidated = payload
return updated
+ {
invalidations: snapshot.invalidations + [invalidated],
run_state: "invalid",
completion: nil,
decision: nil,
}
}
const regression: HypothesisRegressionObserved = payload
return updated + {regressions: snapshot.regressions + [regression]}
}
fn __fold_records(records: list<HypothesisLedgerRecord>, hypothesis_id: string) {
let snapshot = __empty_snapshot(hypothesis_id)
let previous_fingerprint: string? = nil
let previous_cursor: int? = nil
let observations: list<PairedObservation> = []
let observation_ids: dict<string, bool> = {}
let observed_cells: dict<string, bool> = {}
let receipt_ids: list<string> = []
let receipt_id_set: dict<string, bool> = {}
let observed_arms: list<string> = []
let observed_arm_set: dict<string, bool> = {}
for record in records {
__validate_content(record.event.content)
if record.event.content.hypothesis_id != hypothesis_id
|| (previous_cursor != nil
&& record.cursor
<= previous_cursor) {
throw "hypothesis ledger: aggregate event order or identity is invalid"
}
if record.event.fingerprint != __fingerprint(record.event.content) {
throw "hypothesis ledger: projected event fingerprint does not match canonical content"
}
__validate_record_authority(snapshot, record)
__validate_admission(
previous_fingerprint,
snapshot,
observations,
observation_ids,
observed_cells,
record.event,
)
snapshot = __apply_record(snapshot, record)
if record.event.content.payload.kind == "run_transition" {
const transition: HypothesisRunTransition = record.event.content.payload
for receipt_id in transition.receipt_ids {
if !(receipt_id_set[receipt_id] ?? false) {
receipt_ids = receipt_ids + [receipt_id]
receipt_id_set = receipt_id_set + {[receipt_id]: true}
}
}
}
if record.event.content.payload.kind == "observation_recorded" {
const evidence: HypothesisObservationRecorded = record.event.content.payload
observations = observations + [evidence.observation]
observation_ids = observation_ids + {[evidence.observation_id]: true}
const cell = paired_observation_cell_key(evidence.observation)
observed_cells = observed_cells + {[cell]: true}
for arm_id in [
evidence.observation.baseline_assignment.arm_id,
evidence.observation.candidate_assignment.arm_id,
] {
if !(observed_arm_set[arm_id] ?? false) {
observed_arms = observed_arms + [arm_id]
observed_arm_set = observed_arm_set + {[arm_id]: true}
}
}
for receipt_id in [
evidence.observation.baseline_assignment.assignment_id,
evidence.observation.candidate_assignment.assignment_id,
] {
if !(receipt_id_set[receipt_id] ?? false) {
receipt_ids = receipt_ids + [receipt_id]
receipt_id_set = receipt_id_set + {[receipt_id]: true}
}
}
if evidence.operation_receipt_id != nil
&& !(receipt_id_set[evidence.operation_receipt_id]
?? false) {
receipt_ids = receipt_ids + [evidence.operation_receipt_id]
receipt_id_set = receipt_id_set + {[evidence.operation_receipt_id]: true}
}
}
previous_fingerprint = record.event.fingerprint
previous_cursor = record.cursor
}
snapshot = snapshot + {observed_arms: observed_arms, receipt_ids: receipt_ids}
return {
snapshot: snapshot,
observations: observations,
observation_ids: observation_ids,
observed_cells: observed_cells,
previous_fingerprint: previous_fingerprint,
}
}
/**
* Validate and project append-only records into the current read model.
*
* Every event is replayed through the same admission rules used on append, so
* records written through a lower-level event-log API cannot bypass lifecycle,
* assignment, budget, approval, or canonical-decision checks.
*
* @effects: []
* @errors: [validation]
* @api_stability: experimental
*/
pub fn hypothesis_ledger_project(
records: list<HypothesisLedgerRecord>,
hypothesis_id: string,
) -> HypothesisLedgerSnapshot {
return __fold_records(records, hypothesis_id).snapshot
}