diff --git a/dag/test/claim/claim_cost_observation_type_wall_witness_test.dag b/dag/test/claim/claim_cost_observation_type_wall_witness_test.dag new file mode 100644 index 00000000000..e90907d8bff --- /dev/null +++ b/dag/test/claim/claim_cost_observation_type_wall_witness_test.dag @@ -0,0 +1,27 @@ +module test.claim.claim_cost_observation_type_wall_witness_test + +// The wall's RED remains authorable at the compiler-fixture boundary even though it is +// unrepresentable in accepted source: the negative arm passes List to a +// function whose parameter is List. If remedy_partition widens back to the +// observation coproduct, the planted program compiles and this witness reds. +fn exact_population_compiles() -> Bool { + compile_dag_rust_emit_check( + "module fixture.cost_exact\nimport v2.std.collection { List }\nimport v2.workflow.claim_cost_observation { ExactCompletionCost, RemedyPartition, remedy_partition }\nfn size(xs: List) -> RemedyPartition { remedy_partition(exact: xs) }", + "src/claim_cost_exact_control.rs", + [], + [], + ) +} + +fn censored_population_refuses_to_compile() -> Bool { + !compile_dag_rust_emit_check( + "module fixture.cost_censored\nimport v2.std.collection { List }\nimport v2.workflow.claim_cost_observation { RightCensoredCost, RemedyPartition, remedy_partition }\nfn size(xs: List) -> RemedyPartition { remedy_partition(exact: xs) }", + "src/claim_cost_censored_red.rs", + [], + [], + ) +} + +test fn remedy_partition_parameter_is_exact_completion_only() -> Bool { + exact_population_compiles() && censored_population_refuses_to_compile() +} diff --git a/src/v1/stage0/src/cli_run.rs b/src/v1/stage0/src/cli_run.rs index 40dbc8333de..60e0acb0c69 100644 --- a/src/v1/stage0/src/cli_run.rs +++ b/src/v1/stage0/src/cli_run.rs @@ -41095,23 +41095,143 @@ pub enum RequiredFloorDisposition { /// line is the diagnostic predicate applied to the measurement, not a stored property of /// the row -- a stored flag is a second representation of a comparison the policy already /// owns, and it is what lets a summary disagree with its own members. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ObservationCoordinates { + pub commit: String, + pub prepared_subject_digest: String, + pub witness_identity: String, + pub executor_binary_identity: String, + pub workflow_and_job_identity: String, + pub execution_policy_identity: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ObservationBinding { + RequiredFloorGate(ObservationCoordinates), + CostMeasurement(ObservationCoordinates), +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ExactCompletionCost { + pub binding: ObservationBinding, + pub exact_cpu_ms: u64, + pub exact_wall_ms: u64, + pub terminal_result: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RightCensoredCost { + pub binding: ObservationBinding, + pub cpu_lower_bound_ms: u64, + pub wall_lower_bound_ms: u64, + pub censored_by_policy: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ClaimCostObservation { + ExactCompletionCost(ExactCompletionCost), + RightCensoredCost(RightCensoredCost), + CostObservationUnavailable { + binding: ObservationBinding, + cause: String, + }, +} + +/// Remedy sizing is deliberately partial over the observation coproduct: exact completions are +/// accepted, while a lower bound or unavailable reading refuses the WHOLE input population. +/// Returning a partial partition would let its caller quote the exact subset as though it were +/// the denominator and would recreate the interrupt-point-as-cost error downstream. +pub fn exact_completion_population( + observations: &[ClaimCostObservation], +) -> Result, String> { + let mut exact = Vec::with_capacity(observations.len()); + for observation in observations { + match observation { + ClaimCostObservation::ExactCompletionCost(cost) => exact.push(cost), + ClaimCostObservation::RightCensoredCost(_) => { + return Err(format!( + "RemedyPartitionRefused: exact completion observations required; \ + population={} exact={} censored=1", + observations.len(), + exact.len() + )); + } + ClaimCostObservation::CostObservationUnavailable { cause, .. } => { + return Err(format!( + "RemedyPartitionRefused: cost observation unavailable; population={} \ + exact={} cause={cause}", + observations.len(), + exact.len() + )); + } + } + } + Ok(exact) +} + +#[derive(Debug, PartialEq, Eq)] +pub struct RemedyPartition<'a> { + pub exact: Vec<&'a ExactCompletionCost>, +} + +/// The sizing function cannot receive a censored observation: callers must first project the +/// dynamic coproduct through `exact_completion_population`, whose refused arm cannot be mistaken +/// for a partial sizing result. +pub fn remedy_partition(exact: &[ExactCompletionCost]) -> RemedyPartition<'_> { + RemedyPartition { + exact: exact.iter().collect(), + } +} + #[derive(Clone)] pub struct WitnessExecutionOccurrence { pub identity: String, pub module_path: String, - pub outcome: String, - pub wall_nanos: u128, - pub cpu_nanos: u128, - /// Whether the claim reached a verdict rather than being safety-interrupted. Retained - /// because `exceeds_completed_cost_line` only judges COMPLETED claims: an interrupted - /// claim has no cost to be over, and inferring that from the duration would make a slow - /// claim and a killed one the same fact. - pub verdict_reached: bool, + pub observation: ClaimCostObservation, /// The policy line this claim was measured against -- an INPUT, not a derivation, and /// carried per row because a future per-claim line must not silently re-judge old rows. pub cost_line_ms: u64, } +#[cfg(test)] +mod claim_cost_observation_law { + use super::*; + + fn coordinates() -> ObservationCoordinates { + ObservationCoordinates { + commit: "fixture-commit".into(), + prepared_subject_digest: "fixture-subject".into(), + witness_identity: "test.claim.fixture.planted_slow_claim".into(), + executor_binary_identity: "fixture-executor".into(), + workflow_and_job_identity: "fixture-workflow/fixture-job".into(), + execution_policy_identity: "fixture-policy".into(), + } + } + + #[test] + fn remedy_partition_accepts_exact_control_and_refuses_censored_twin() { + let exact = ClaimCostObservation::ExactCompletionCost(ExactCompletionCost { + binding: ObservationBinding::CostMeasurement(coordinates()), + exact_cpu_ms: 7, + exact_wall_ms: 8, + terminal_result: "returned-false".into(), + }); + let admitted = exact_completion_population(std::slice::from_ref(&exact)).unwrap(); + let owned: Vec = admitted.into_iter().cloned().collect(); + assert_eq!(remedy_partition(&owned).exact.len(), 1); + + let censored = ClaimCostObservation::RightCensoredCost(RightCensoredCost { + binding: ObservationBinding::CostMeasurement(coordinates()), + cpu_lower_bound_ms: 5, + wall_lower_bound_ms: 6, + censored_by_policy: "fixture-tiny-finite-policy".into(), + }); + let refusal = exact_completion_population(&[censored]).unwrap_err(); + assert!(refusal.contains("exact completion observations required")); + assert!(refusal.contains("population=1 exact=0 censored=1")); + } +} + /// The occurrence's terminal outcome as a stable receipt token. /// /// EXHAUSTIVE ON PURPOSE -- there is no `_` arm. A new `ClaimOutcome` variant must fail to @@ -42653,6 +42773,16 @@ pub fn run_required_floor( ); } let claims_planned = claims.len(); + let executor_binary_identity = floor_discovery_snapshot::floor_tool_identity()?; + let workflow_and_job_identity = format!( + "{}/{}", + std::env::var("GITHUB_WORKFLOW").unwrap_or_else(|_| "local".to_string()), + std::env::var("GITHUB_JOB").unwrap_or_else(|_| "local".to_string()) + ); + let execution_policy_identity = format!( + "required-floor:cpu={}ms;wall={}ms", + claim_cpu_safety_limit_ms, claim_wall_safety_limit_ms + ); let mut outcome = RequiredFloorOutcome { subject_digest: prepared.subject_digest.clone(), modules_resolved: prepared.modules_resolved, @@ -42894,13 +43024,40 @@ pub fn run_required_floor( // over-cost population, its count and the identity-grain receipt are all folds over // `outcome.claim_cost` computed after the loop -- so the summary cannot disagree with // its own members, and there is no separately incremented tally to drift. + let observation_binding = ObservationBinding::RequiredFloorGate(ObservationCoordinates { + commit: commit.to_string(), + prepared_subject_digest: prepared.subject_digest.clone(), + witness_identity: claim.qualified.clone(), + executor_binary_identity: executor_binary_identity.clone(), + workflow_and_job_identity: workflow_and_job_identity.clone(), + execution_policy_identity: execution_policy_identity.clone(), + }); + let cost_observation = match &terminality { + ClaimTerminality::VerdictReached { + observed_cpu_ms, + observed_wall_ms, + .. + } => ClaimCostObservation::ExactCompletionCost(ExactCompletionCost { + binding: observation_binding, + exact_cpu_ms: *observed_cpu_ms, + exact_wall_ms: *observed_wall_ms, + terminal_result: witness_execution_outcome_label(&result).to_string(), + }), + ClaimTerminality::SafetyInterrupted { + elapsed_cpu_at_least_ms, + elapsed_wall_at_least_ms, + .. + } => ClaimCostObservation::RightCensoredCost(RightCensoredCost { + binding: observation_binding, + cpu_lower_bound_ms: *elapsed_cpu_at_least_ms, + wall_lower_bound_ms: *elapsed_wall_at_least_ms, + censored_by_policy: execution_policy_identity.clone(), + }), + }; outcome.claim_cost.push(WitnessExecutionOccurrence { identity: claim.qualified.clone(), module_path: claim.module_path.clone(), - outcome: witness_execution_outcome_label(&result).to_string(), - wall_nanos: receipt.wall_nanos, - cpu_nanos: receipt.cpu_nanos, - verdict_reached: matches!(terminality, ClaimTerminality::VerdictReached { .. }), + observation: cost_observation, cost_line_ms: claim.cost_line_ms, }); // RETURN WHAT THE FRAME NO LONGER OWNS, on a cadence rather than every row. @@ -43633,18 +43790,20 @@ pub fn run_required_floor( let over_cost_members: Vec<&WitnessExecutionOccurrence> = outcome .claim_cost .iter() - .filter(|row| row.verdict_reached && observed_cost_ms(row, cost_basis) > row.cost_line_ms) + .filter(|row| observed_cost_ms(row, cost_basis).is_some_and(|ms| ms > row.cost_line_ms)) .collect(); outcome.over_cost_line_diagnostic = over_cost_members.len(); for row in &over_cost_members { - eprintln!( - "[over-cost] {} wall_ms={} cpu_ms={} line_ms={} outcome={}", - row.identity, - row.wall_nanos / 1_000_000, - row.cpu_nanos / 1_000_000, - row.cost_line_ms, - row.outcome - ); + if let ClaimCostObservation::ExactCompletionCost(cost) = &row.observation { + eprintln!( + "[over-cost] {} exact_wall_ms={} exact_cpu_ms={} line_ms={} outcome={}", + row.identity, + cost.exact_wall_ms, + cost.exact_cpu_ms, + row.cost_line_ms, + cost.terminal_result + ); + } } if let Some(path) = claim_cost_path { write_required_floor_claim_cost_tsv(&path, &outcome.claim_cost, cost_basis)?; @@ -43757,12 +43916,17 @@ fn required_floor_disposition_matched_prefix(disposition: &RequiredFloorDisposit /// comparison; the other rides in the receipt beside it as a remedy discriminator (high wall /// with high CPU is algorithm or repeated evaluation, high wall with low CPU is waiting, I/O, /// subprocess or scheduling). It is not a second threshold. -fn observed_cost_ms(row: &WitnessExecutionOccurrence, basis: RequiredFloorCostBasis) -> u64 { - let exact = match basis { - RequiredFloorCostBasis::CpuCost => row.cpu_nanos, - RequiredFloorCostBasis::WallCost => row.wall_nanos, +fn observed_cost_ms( + row: &WitnessExecutionOccurrence, + basis: RequiredFloorCostBasis, +) -> Option { + let ClaimCostObservation::ExactCompletionCost(exact) = &row.observation else { + return None; }; - (exact / 1_000_000) as u64 + Some(match basis { + RequiredFloorCostBasis::CpuCost => exact.exact_cpu_ms, + RequiredFloorCostBasis::WallCost => exact.exact_wall_ms, + }) } /// The identity-grain cost receipt: one row for EVERY executed claim, not only the over-cost @@ -43793,19 +43957,78 @@ fn write_required_floor_claim_cost_tsv( .map_err(|e| format!("write_required_floor_claim_cost_tsv: write {path}: {e}"))?; writeln!( file, - "identity\tmodule\toutcome\tverdict_reached\twall_ms\tcpu_ms\tcost_line_ms" + "identity\tmodule\tcost_observation_kind\tterminal_result\texact_wall_ms\t\ + exact_cpu_ms\twall_lower_bound_ms\tcpu_lower_bound_ms\tcensored_by_policy\t\ + observation_purpose\tcommit\tprepared_subject_digest\texecutor_binary_identity\t\ + workflow_and_job_identity\texecution_policy_identity\tcost_line_ms" ) .map_err(|e| format!("write_required_floor_claim_cost_tsv: write {path}: {e}"))?; for row in rows { + let (kind, terminal, exact_wall, exact_cpu, lower_wall, lower_cpu, censored_by, binding) = + match &row.observation { + ClaimCostObservation::ExactCompletionCost(cost) => ( + "exact-completion", + cost.terminal_result.as_str(), + cost.exact_wall_ms.to_string(), + cost.exact_cpu_ms.to_string(), + String::new(), + String::new(), + "", + &cost.binding, + ), + ClaimCostObservation::RightCensoredCost(cost) => ( + "right-censored", + "", + String::new(), + String::new(), + cost.wall_lower_bound_ms.to_string(), + cost.cpu_lower_bound_ms.to_string(), + cost.censored_by_policy.as_str(), + &cost.binding, + ), + ClaimCostObservation::CostObservationUnavailable { binding, cause } => ( + "unavailable", + cause.as_str(), + String::new(), + String::new(), + String::new(), + String::new(), + "", + binding, + ), + }; + let (purpose, coordinates) = match binding { + ObservationBinding::RequiredFloorGate(coordinates) => { + ("required-floor-gate", coordinates) + } + ObservationBinding::CostMeasurement(coordinates) => ("cost-measurement", coordinates), + }; writeln!( file, - "{}\t{}\t{}\t{}\t{}\t{}\t{}", + "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}", row.identity.replace(['\t', '\n'], " "), row.module_path.replace(['\t', '\n'], " "), - row.outcome, - row.verdict_reached, - row.wall_nanos / 1_000_000, - row.cpu_nanos / 1_000_000, + kind, + terminal.replace(['\t', '\n'], " "), + exact_wall, + exact_cpu, + lower_wall, + lower_cpu, + censored_by.replace(['\t', '\n'], " "), + purpose, + coordinates.commit.replace(['\t', '\n'], " "), + coordinates + .prepared_subject_digest + .replace(['\t', '\n'], " "), + coordinates + .executor_binary_identity + .replace(['\t', '\n'], " "), + coordinates + .workflow_and_job_identity + .replace(['\t', '\n'], " "), + coordinates + .execution_policy_identity + .replace(['\t', '\n'], " "), row.cost_line_ms ) .map_err(|e| format!("write_required_floor_claim_cost_tsv: write {path}: {e}"))?; diff --git a/src/v1/stage0/src/cli_run/terminal_ledger_publish.rs b/src/v1/stage0/src/cli_run/terminal_ledger_publish.rs index 9f9c754b2c7..bedcec0bc77 100644 --- a/src/v1/stage0/src/cli_run/terminal_ledger_publish.rs +++ b/src/v1/stage0/src/cli_run/terminal_ledger_publish.rs @@ -236,6 +236,69 @@ pub fn publish_terminal_ledger( Ok(published) } +/// Read one published terminal ledger through the `.dag` grammar and derive exactly its +/// budget-refused identities. The grammar verifies binding, footer population and row digest; +/// this seed seam only transports the resulting closed arm. +pub fn budget_refused_population_from_ledger( + source_roots: &[String], + text: &str, + expected_repository_snapshot: &str, + expected_prepared_subject: &str, +) -> Result, String> { + let ctx = build_ledger_wire_ctx(source_roots)?; + let args = vec![ + (Some("text".to_string()), str_value(text)), + ( + Some("expected_repository_snapshot".to_string()), + str_value(expected_repository_snapshot), + ), + ( + Some("expected_prepared_subject".to_string()), + str_value(expected_prepared_subject), + ), + ]; + let value = v1_interpreter::with_active_context(&ctx, || { + v1_interpreter::run_in_context_with_args( + &ctx, + "budget_refused_population_from_ledger", + &args, + false, + ) + }) + .map_err(|e| format!("COST-MEASUREMENT REFUSAL cause=LedgerReadFailed detail={e}"))?; + let Value::Variant { + variant_name, + fields, + .. + } = value + else { + return Err("COST-MEASUREMENT REFUSAL cause=LedgerReadShape".to_string()); + }; + if ctx.sym_eq(variant_name, "BudgetRefusedPopulationRefused") { + return Err(format!( + "COST-MEASUREMENT REFUSAL cause={}", + variant_field(&fields, &ctx, "cause").unwrap_or_else(|| "unreadable-cause".into()) + )); + } + if !ctx.sym_eq(variant_name, "BudgetRefusedPopulationDerived") { + return Err("COST-MEASUREMENT REFUSAL cause=LedgerReadUnknownArm".to_string()); + } + let identities = fields + .iter() + .find_map(|(symbol, value)| ctx.sym_eq(*symbol, "identities").then_some(value)) + .ok_or_else(|| "COST-MEASUREMENT REFUSAL cause=LedgerPopulationAbsent".to_string())?; + let Value::List(items) = identities else { + return Err("COST-MEASUREMENT REFUSAL cause=LedgerPopulationNotList".to_string()); + }; + items + .iter() + .map(|value| match value { + Value::Str(identity) => Ok(identity.to_string()), + _ => Err("COST-MEASUREMENT REFUSAL cause=LedgerIdentityNotString".to_string()), + }) + .collect() +} + enum Render { Text(String), Refused { diff --git a/src/v2/workflow/claim_cost_observation.dag b/src/v2/workflow/claim_cost_observation.dag new file mode 100644 index 00000000000..e3fa712d448 --- /dev/null +++ b/src/v2/workflow/claim_cost_observation.dag @@ -0,0 +1,146 @@ +module v2.workflow.claim_cost_observation + +import v2.std.algebra { Cons, Empty } +import v2.std.collection { List } +import v2.std.integer { Int } +import v2.std.text { String } +import v2.workflow.floor_terminal_ledger { ClaimAttemptTerminal } + +// ONE COST FACT, TWO EXECUTION POLICIES. The required floor and the off-path measurement +// instrument answer different questions, but they mint this same coproduct rather than parallel +// "floor cost" and "measurement cost" authorities. Exact fields and lower-bound fields have +// different names by construction: a censored interrupt cannot be copied into an ordinary +// `cpu_ms` column and silently become a completion cost. +type ObservationCoordinates { + commit: String + prepared_subject_digest: String + witness_identity: String + executor_binary_identity: String + workflow_and_job_identity: String + execution_policy_identity: String +} + +// PURPOSE IS CLOSED AND WRAPPED, not a String flag. In particular a measurement binding cannot +// inhabit RequiredFloorObservation below, so a successful measurement attempt cannot certify a +// required-floor run even when commit and subject happen to agree. +type ObservationBinding + = RequiredFloorGateBinding { coordinates: ObservationCoordinates } + | CostMeasurementBinding { coordinates: ObservationCoordinates } + +type ExactCompletionCost { + binding: ObservationBinding + exact_cpu_ms: Int + exact_wall_ms: Int + terminal_result: ClaimAttemptTerminal +} + +type RightCensoredCost { + binding: ObservationBinding + cpu_lower_bound_ms: Int + wall_lower_bound_ms: Int + censored_by_policy: String +} + +type ClaimCostObservation + = ExactCompletionCostObserved { exact: ExactCompletionCost } + | RightCensoredCostObserved { censored: RightCensoredCost } + | CostObservationUnavailable { binding: ObservationBinding, cause: String } + +type RequiredFloorObservation { exact: ExactCompletionCost } + +type RequiredFloorObservationAdmission + = RequiredFloorObservationAdmitted { observation: RequiredFloorObservation } + | RequiredFloorObservationRefused { cause: String } + +fn cost_observation_for_required_floor( + observation: ClaimCostObservation, +) -> RequiredFloorObservationAdmission { + match observation { + ExactCompletionCostObserved { exact: exact } => + match exact.binding { + RequiredFloorGateBinding { coordinates: _ } => + RequiredFloorObservationAdmitted { + observation: RequiredFloorObservation { exact: exact } + } + CostMeasurementBinding { coordinates: _ } => + RequiredFloorObservationRefused { + cause: "cost-measurement-purpose-cannot-certify-required-floor" + } + } + RightCensoredCostObserved { censored: _ } => + RequiredFloorObservationRefused { cause: "right-censored-cost-cannot-certify-required-floor" } + CostObservationUnavailable { binding: _, cause: _ } => + RequiredFloorObservationRefused { cause: "unavailable-cost-cannot-certify-required-floor" } + } +} + +// Remedy sizing consumes exact completions only. The result refuses on the FIRST censored or +// unavailable member rather than returning a partial partition that could be quoted as though it +// described the input population. `population`, `exact`, and `censored` remain separate counts in +// the refusal so every board must state its denominator. +type ExactCompletionPopulationAdmission + = ExactCompletionPopulationAdmitted { exact: List } + | ExactCompletionPopulationRefused { + population: Int + exact: Int + censored: Int + cause: String + } + +fn exact_completion_population_loop( + remaining: List, + exact: List, + population: Int, + exact_count: Int, + censored_count: Int, +) -> ExactCompletionPopulationAdmission { + match remaining { + Empty {} => ExactCompletionPopulationAdmitted { exact: exact } + Cons { head: observation, tail: tail } => + match observation { + ExactCompletionCostObserved { exact: completed } => + exact_completion_population_loop( + remaining: tail, + exact: Cons { head: completed, tail: exact }, + population: population + 1, + exact_count: exact_count + 1, + censored_count: censored_count, + ) + RightCensoredCostObserved { censored: _ } => + ExactCompletionPopulationRefused { + population: population + 1, + exact: exact_count, + censored: censored_count + 1, + cause: "exact-completion-observations-required", + } + CostObservationUnavailable { binding: _, cause: _ } => + ExactCompletionPopulationRefused { + population: population + 1, + exact: exact_count, + censored: censored_count, + cause: "cost-observation-unavailable", + } + } + } +} + +fn exact_completion_population( + observations: List, +) -> ExactCompletionPopulationAdmission { + exact_completion_population_loop( + remaining: observations, + exact: Empty {}, + population: 0, + exact_count: 0, + censored_count: 0, + ) +} + +// SIZING IS CONSTRUCTION-SAFE: there is no censored arm to match because the parameter cannot +// contain one. The dynamic observation coproduct must first pass exact_completion_population; +// callers cannot accidentally feed an interrupt point into remedy sizing as an ordinary cost. +type RemedyPartition { exact: List } + +fn remedy_partition(exact: List) -> RemedyPartition { + RemedyPartition { exact: exact } +} diff --git a/src/v2/workflow/claim_cost_observation_test.dag b/src/v2/workflow/claim_cost_observation_test.dag new file mode 100644 index 00000000000..54db78f45eb --- /dev/null +++ b/src/v2/workflow/claim_cost_observation_test.dag @@ -0,0 +1,83 @@ +module v2.workflow.claim_cost_observation_test + +import v2.std.algebra { Cons, Empty } +import v2.workflow.claim_cost_observation { + ObservationCoordinates, + CostMeasurementBinding, + ExactCompletionCost, + ExactCompletionCostObserved, + RightCensoredCost, + RightCensoredCostObserved, + RemedyPartition, + ExactCompletionPopulationRefused, + RequiredFloorObservationRefused, + cost_observation_for_required_floor, + exact_completion_population, + remedy_partition, +} +import v2.workflow.floor_terminal_ledger { + Returned, + ObservedVerdict, +} + +fn fixture_coordinates() -> ObservationCoordinates { + ObservationCoordinates { + commit: "fixture-commit", + prepared_subject_digest: "fixture-subject", + witness_identity: "test.claim.fixture.planted_slow_claim", + executor_binary_identity: "fixture-executor", + workflow_and_job_identity: "fixture-workflow/fixture-job", + execution_policy_identity: "fixture-policy", + } +} + +fn fixture_exact() -> ExactCompletionCost { + ExactCompletionCost { + binding: CostMeasurementBinding { coordinates: fixture_coordinates() }, + exact_cpu_ms: 7, + exact_wall_ms: 8, + terminal_result: Returned { observation: ObservedVerdict { holds: true } }, + } +} + +test fn remedy_partition_accepts_exact_completion_control() -> Bool { + match remedy_partition(exact: Cons { + head: fixture_exact(), + tail: Empty {}, + }) { + RemedyPartition { exact: Cons { head: _, tail: Empty {} } } => true + } +} + +test fn remedy_partition_refuses_right_censored_cost() -> Bool { + match exact_completion_population(observations: Cons { + head: RightCensoredCostObserved { + censored: RightCensoredCost { + binding: CostMeasurementBinding { coordinates: fixture_coordinates() }, + cpu_lower_bound_ms: 5, + wall_lower_bound_ms: 6, + censored_by_policy: "fixture-tiny-finite-policy", + } + }, + tail: Empty {}, + }) { + ExactCompletionPopulationRefused { + population: 1, + exact: 0, + censored: 1, + cause: "exact-completion-observations-required", + } => true + _ => false + } +} + +test fn cost_measurement_observation_cannot_certify_required_floor() -> Bool { + match cost_observation_for_required_floor( + observation: ExactCompletionCostObserved { exact: fixture_exact() }, + ) { + RequiredFloorObservationRefused { + cause: "cost-measurement-purpose-cannot-certify-required-floor" + } => true + _ => false + } +} diff --git a/src/v2/workflow/floor_terminal_ledger_wire.dag b/src/v2/workflow/floor_terminal_ledger_wire.dag index 1fbadb27d53..53945f19c8f 100644 --- a/src/v2/workflow/floor_terminal_ledger_wire.dag +++ b/src/v2/workflow/floor_terminal_ledger_wire.dag @@ -657,6 +657,56 @@ fn readback_digest_agrees(readback: LedgerReadback) -> Bool { } } +// THE MEASUREMENT POPULATION IS DERIVED FROM ONE COMPLETE LEDGER, never transcribed from a +// console count or copied into a roster. Binding agreement is checked before any identity is +// returned. The footer count is compared with the parsed population and the digest is recomputed; +// a truncated or edited ledger therefore refuses instead of becoming a shorter measurement run. +type BudgetRefusedPopulation + = BudgetRefusedPopulationDerived { identities: List } + | BudgetRefusedPopulationRefused { cause: String } + +fn budget_refused_population_from_ledger( + text: String, + expected_repository_snapshot: String, + expected_prepared_subject: String, +) -> BudgetRefusedPopulation { + let readback = parse_ledger(text: text) + match readback { + LedgerUnreadable { reason: reason, at: _ } => + BudgetRefusedPopulationRefused { cause: reason as String } + LedgerRead { + repository_snapshot: snapshot, + prepared_subject: subject, + roster_identity: _, + rows: rows, + footer_row_count_wire: count, + footer_digest: _, + } => + if snapshot != expected_repository_snapshot { + BudgetRefusedPopulationRefused { cause: "repository-snapshot-disagrees" } + } else { + if subject != expected_prepared_subject { + BudgetRefusedPopulationRefused { cause: "prepared-subject-disagrees" } + } else { + if count != rows.length() as String { + BudgetRefusedPopulationRefused { cause: "footer-row-count-disagrees" } + } else { + if !readback_digest_agrees(readback: readback) { + BudgetRefusedPopulationRefused { cause: "ledger-row-digest-disagrees" } + } else { + BudgetRefusedPopulationDerived { + identities: map( + filter(rows, row => row.terminal_tag == "budget-refused"), + row => row.qualified, + ) + } + } + } + } + } + } +} + // THE DISPOSITION SURVIVES THE ROUND TRIP, AND THIS IS THE PROPERTY THE ARTIFACT EXISTS FOR. // // A file that preserved identities but lost what happened to them would satisfy every structural