"""Tests for Lab 01 — Platform Reference Model & Budget Calculator. pytest test_lab.py -v # your lab.py LAB_MODULE=solution pytest test_lab.py -v # the reference — must be green """ import importlib import math import os import pytest lab = importlib.import_module(os.environ.get("LAB_MODULE", "lab")) # ====================================================================================== # 1. Availability composition # ====================================================================================== def test_series_of_empty_chain_is_one(): assert lab.series_availability([]) == 1.0 def test_series_multiplies(): assert lab.series_availability([0.999] * 5) == pytest.approx(0.99501, abs=1e-5) def test_series_is_never_better_than_its_worst_member(): values = [0.9999, 0.995, 0.99999] assert lab.series_availability(values) <= min(values) def test_series_rejects_out_of_range(): with pytest.raises(ValueError): lab.series_availability([1.5]) with pytest.raises(ValueError): lab.series_availability([-0.1]) def test_parallel_of_empty_group_is_zero(): assert lab.parallel_availability([]) == 0.0 def test_parallel_multiplies_unavailability(): assert lab.parallel_availability([0.99, 0.99]) == pytest.approx(0.9999) assert lab.parallel_availability([0.999, 0.999]) == pytest.approx(0.999999) def test_parallel_single_member_is_itself(): assert lab.parallel_availability([0.997]) == pytest.approx(0.997) def test_correlated_parallel_reduces_to_independent_at_zero(): independent = lab.parallel_availability([0.998, 0.998]) assert lab.correlated_parallel_availability([0.998, 0.998], 0.0) == pytest.approx(independent) def test_correlated_parallel_at_full_correlation_buys_nothing(): assert lab.correlated_parallel_availability([0.998, 0.998], 1.0) == pytest.approx(0.998) def test_correlated_parallel_worked_number(): # c=0.2: 1 - (0.2*0.002 + 0.8*0.002**2) assert lab.correlated_parallel_availability([0.998, 0.998], 0.2) == pytest.approx(0.9995968) def test_correlation_dominates_redundancy(): """The whole point: a small common-mode term destroys most of the benefit.""" independent = lab.parallel_availability([0.999, 0.999]) correlated = lab.correlated_parallel_availability([0.999, 0.999], 0.2) assert independent > correlated assert (1 - correlated) > 100 * (1 - independent) def test_correlated_parallel_rejects_bad_common_mode(): with pytest.raises(ValueError): lab.correlated_parallel_availability([0.99, 0.99], 1.5) # ====================================================================================== # 2. The platform model # ====================================================================================== def five_layer_model(): return lab.PlatformModel.of("p", [ lab.Component("channel", 0.9995, p95_ms=60), lab.Component("control_plane", 0.99999, p95_ms=10), lab.Component("kernel", 0.999, p95_ms=120), lab.Component("model_layer", 0.998, p95_ms=800), lab.Component("knowledge", 0.995, p95_ms=350), lab.Component("action_gateway", 0.9995, p95_ms=900), ]) def test_component_rejects_bad_availability(): with pytest.raises(ValueError): lab.Component("x", 1.2) def test_component_rejects_negative_latency(): with pytest.raises(ValueError): lab.Component("x", 0.99, p95_ms=-1) def test_request_availability_is_the_naive_product(): assert five_layer_model().request_availability() == pytest.approx(0.9910153, abs=1e-6) def test_downtime_minutes_at_three_nines(): model = lab.PlatformModel.of("p", [lab.Component("only", 0.999)]) assert model.downtime_minutes(30) == pytest.approx(43.2) def test_downtime_scales_with_window(): model = lab.PlatformModel.of("p", [lab.Component("only", 0.999)]) assert model.downtime_minutes(365) == pytest.approx(525.6) def test_degradable_component_does_not_reduce_request_availability(): model = five_layer_model().with_degradable("knowledge") assert model.request_availability() == pytest.approx(0.99600, abs=1e-5) # ... but it still reduces the availability of a FULL-QUALITY answer assert model.quality_availability() == pytest.approx(0.9910153, abs=1e-6) assert model.quality_availability() < model.request_availability() def test_with_degradable_rejects_unknown_component(): with pytest.raises(KeyError): five_layer_model().with_degradable("nope") def test_with_replaced_changes_only_the_named_component(): model = five_layer_model().with_replaced("model_layer", 0.9996) by_name = {c.name: c.availability for c in model.components} assert by_name["model_layer"] == pytest.approx(0.9996) assert by_name["kernel"] == pytest.approx(0.999) def test_with_replaced_rejects_unknown_component(): with pytest.raises(KeyError): five_layer_model().with_replaced("nope", 0.99) def test_weakest_links_are_ranked_by_unavailability(): ranked = five_layer_model().weakest_links(3) assert [name for name, _ in ranked] == ["knowledge", "model_layer", "kernel"] assert ranked[0][1] == pytest.approx(0.005) assert ranked[2][1] == pytest.approx(0.001) def test_weakest_links_excludes_degradable_by_default(): model = five_layer_model().with_degradable("knowledge") assert model.weakest_links(1)[0][0] == "model_layer" assert model.weakest_links(1, include_degradable=True)[0][0] == "knowledge" def test_the_phase_headline_improvement(): """Degradable retrieval + a correlated second provider, exactly as in the WARMUP.""" model = five_layer_model() assert model.request_availability() == pytest.approx(0.99102, abs=1e-5) model = model.with_replaced("knowledge", 0.999) assert model.request_availability() == pytest.approx(0.99500, abs=1e-5) pair = lab.correlated_parallel_availability([0.998, 0.998], 0.2) model = model.with_replaced("model_layer", pair) assert model.request_availability() == pytest.approx(0.99659, abs=1e-5) # ====================================================================================== # 3. Error budgets # ====================================================================================== def test_error_budget_three_nines_thirty_days(): assert lab.ErrorBudget(0.999, 30).total_minutes() == pytest.approx(43.2) def test_error_budget_rejects_bad_window(): with pytest.raises(ValueError): lab.ErrorBudget(0.999, 0) def test_allocate_splits_by_weight(): budget = lab.ErrorBudget(0.995, 30) got = budget.allocate({"model": 0.4, "integrations": 0.3, "kernel": 0.2, "other": 0.1}) assert got["model"] == pytest.approx(86.4) assert sum(got.values()) == pytest.approx(budget.total_minutes()) def test_allocate_rejects_weights_that_do_not_sum_to_one(): with pytest.raises(ValueError): lab.ErrorBudget(0.999).allocate({"a": 0.5, "b": 0.4}) def test_allocate_rejects_negative_and_empty(): with pytest.raises(ValueError): lab.ErrorBudget(0.999).allocate({"a": 1.5, "b": -0.5}) with pytest.raises(ValueError): lab.ErrorBudget(0.999).allocate({}) def test_ledger_tracks_per_layer_and_total(): ledger = lab.BudgetLedger(lab.ErrorBudget(0.995, 30), {"a": 0.5, "b": 0.5}) ledger.consume("a", 30.0) assert ledger.consumed_for("a") == pytest.approx(30.0) assert ledger.remaining_for("a") == pytest.approx(78.0) assert ledger.remaining() == pytest.approx(186.0) def test_ledger_never_reports_negative_remaining(): ledger = lab.BudgetLedger(lab.ErrorBudget(0.999, 30), {"a": 1.0}) ledger.consume("a", 100.0) assert ledger.remaining() == 0.0 assert ledger.remaining_for("a") == 0.0 assert ledger.overspend_for("a") == pytest.approx(100.0 - 43.2) def test_ledger_rejects_unknown_layer_and_negative_minutes(): ledger = lab.BudgetLedger(lab.ErrorBudget(0.999), {"a": 1.0}) with pytest.raises(KeyError): ledger.consume("nope", 1.0) with pytest.raises(ValueError): ledger.consume("a", -1.0) @pytest.mark.parametrize("consumed,expected", [ (0.0, "normal"), (21.5, "normal"), # 50.2% remaining (21.7, "elevated"), # 49.8% remaining (32.5, "reliability-focus"), (43.2, "freeze"), (99.0, "freeze"), ]) def test_error_budget_policy_states(consumed, expected): ledger = lab.BudgetLedger(lab.ErrorBudget(0.999, 30), {"a": 1.0}) ledger.consume("a", consumed) assert ledger.policy_state() == expected def test_burn_rate_of_one_spends_the_budget_exactly(): assert lab.burn_rate(0.001, 0.999) == pytest.approx(1.0) def test_burn_rate_scales_linearly(): assert lab.burn_rate(0.0144, 0.999) == pytest.approx(14.4) def test_burn_rate_with_a_perfect_slo(): assert lab.burn_rate(0.0, 1.0) == 0.0 assert lab.burn_rate(0.01, 1.0) == math.inf def test_burn_rate_threshold_derives_the_standard_ladder(): assert lab.burn_rate_threshold(0.02, 1) == pytest.approx(14.4) assert lab.burn_rate_threshold(0.05, 6) == pytest.approx(6.0) assert lab.burn_rate_threshold(0.10, 72) == pytest.approx(1.0) def test_burn_rate_threshold_validates(): with pytest.raises(ValueError): lab.burn_rate_threshold(0.02, 0) with pytest.raises(ValueError): lab.burn_rate_threshold(0.02, 1, period_days=0) def test_alert_policy_requires_both_windows(): policy = lab.MultiWindowAlertPolicy(slo=0.999) # long window hot, short window cold -> the incident is over; do not page ratios = {1.0: 0.02, 1 / 12: 0.0} assert policy.evaluate(lambda w: ratios.get(w, 0.0)) == [] def test_alert_policy_fires_on_a_sustained_fast_burn(): policy = lab.MultiWindowAlertPolicy(slo=0.999) ratios = {1.0: 0.02, 1 / 12: 0.03} fired = policy.evaluate(lambda w: ratios.get(w, 0.0)) assert [r.name for r in fired] == ["fast-burn"] assert policy.highest_severity(fired) == "page" def test_alert_policy_boundary_exactly_at_threshold_fires(): """burn rate == 14.4 exactly. Must use >=, not >.""" policy = lab.MultiWindowAlertPolicy(slo=0.999) fired = policy.evaluate(lambda w: 0.0144) assert "fast-burn" in [r.name for r in fired] def test_alert_policy_just_below_threshold_does_not_fire(): policy = lab.MultiWindowAlertPolicy(slo=0.999) fired = policy.evaluate(lambda w: 0.01439) assert "fast-burn" not in [r.name for r in fired] def test_alert_policy_slow_burn_is_a_ticket_not_a_page(): policy = lab.MultiWindowAlertPolicy(slo=0.999) fired = policy.evaluate(lambda w: 0.0012) # burn rate 1.2 -> only slow-burn assert [r.name for r in fired] == ["slow-burn"] assert policy.highest_severity(fired) == "ticket" def test_alert_policy_quiet_system_is_silent(): policy = lab.MultiWindowAlertPolicy(slo=0.999) assert policy.evaluate(lambda w: 0.0001) == [] assert policy.highest_severity([]) is None # ====================================================================================== # 4. Latency budgets # ====================================================================================== def budget_3s(): return lab.LatencyBudget.of(3000, [ lab.LatencyStage("ingress", 60), lab.LatencyStage("guardrails_in", 120, group="pre"), lab.LatencyStage("retrieval", 200, group="pre"), lab.LatencyStage("rerank", 150, sheddable=True), lab.LatencyStage("model_ttft", 800), lab.LatencyStage("action_gateway", 900), lab.LatencyStage("guardrails_out", 60), lab.LatencyStage("network", 200), ]) def test_parallel_group_contributes_its_max_not_its_sum(): assert budget_3s().committed_ms() == 2370 # 2490 if the group were summed assert budget_3s().headroom_ms() == 630 def test_budget_feasibility(): assert budget_3s().is_feasible() tight = lab.LatencyBudget.of(1000, [lab.LatencyStage("a", 1200)]) assert not tight.is_feasible() assert tight.headroom_ms() == -200 def test_fallback_fits_only_inside_the_headroom(): b = budget_3s() assert b.fits_fallback(500) assert b.fits_fallback(630) # exactly the headroom fits assert not b.fits_fallback(631) def test_fits_fallback_rejects_negative(): with pytest.raises(ValueError): budget_3s().fits_fallback(-1) def test_shedding_buys_headroom(): b = budget_3s() assert not b.fits_fallback(700) assert b.fits_fallback(700, shed=["rerank"]) assert b.committed_ms(shed=["rerank"]) == 2220 def test_shed_order_is_most_expensive_first(): b = lab.LatencyBudget.of(3000, [ lab.LatencyStage("a", 100, sheddable=True), lab.LatencyStage("b", 300, sheddable=True), lab.LatencyStage("c", 300, sheddable=True), lab.LatencyStage("d", 900), ]) assert b.shed_order() == ["b", "c", "a"] def test_shed_until_fits_stops_as_soon_as_it_fits(): b = budget_3s() assert b.shed_until_fits(500) == [] # already fits assert b.shed_until_fits(700) == ["rerank"] def test_shed_until_fits_can_run_out_of_ladder(): b = lab.LatencyBudget.of(1000, [ lab.LatencyStage("core", 900), lab.LatencyStage("extra", 50, sheddable=True), ]) shed = b.shed_until_fits(500) assert shed == ["extra"] assert not b.fits_fallback(500, shed=shed) # caller must re-check def test_stage_rejects_negative_latency_and_budget_rejects_zero_target(): with pytest.raises(ValueError): lab.LatencyStage("x", -5) with pytest.raises(ValueError): lab.LatencyBudget.of(0, []) # ====================================================================================== # 5. Loop reliability # ====================================================================================== @pytest.mark.parametrize("p,n,expected", [ (0.99, 3, 0.970299), (0.99, 20, 0.817907), (0.95, 10, 0.598737), (0.95, 20, 0.358486), (0.90, 20, 0.121577), ]) def test_loop_success_table(p, n, expected): assert lab.loop_success(p, n) == pytest.approx(expected, abs=1e-6) def test_loop_success_zero_steps_is_certain(): assert lab.loop_success(0.5, 0) == 1.0 def test_loop_success_rejects_negative_steps(): with pytest.raises(ValueError): lab.loop_success(0.9, -1) def test_retry_attacks_failure_probability_multiplicatively(): assert lab.effective_step_probability(0.90, 0) == pytest.approx(0.90) assert lab.effective_step_probability(0.90, 1) == pytest.approx(0.99) assert lab.effective_step_probability(0.90, 2) == pytest.approx(0.999) assert lab.effective_step_probability(0.95, 1) == pytest.approx(0.9975) def test_retry_rejects_negative(): with pytest.raises(ValueError): lab.effective_step_probability(0.9, -1) def test_max_steps_for_target(): assert lab.max_steps_for_target(0.95, 0.90) == 2 assert lab.max_steps_for_target(0.99, 0.90) == 10 assert lab.max_steps_for_target(0.999, 0.99) == 10 def test_max_steps_for_target_edges(): with pytest.raises(ValueError): lab.max_steps_for_target(1.0, 0.9) with pytest.raises(ValueError): lab.max_steps_for_target(0.9, 0.0) assert lab.max_steps_for_target(0.0, 0.5) == 0 # ====================================================================================== # 6. Cost # ====================================================================================== def prices(): return lab.TokenPrices(input_per_1k=3000, cached_input_per_1k=300, output_per_1k=15000) def test_token_prices_reject_negative(): with pytest.raises(ValueError): lab.TokenPrices(-1, 0, 0) def test_step_cost_three_tiers(): model = lab.CostModel(prices()) # 2000 fresh in, 1000 cached in, 500 out # = (2000*3000 + 1000*300 + 500*15000) / 1000 = (6_000_000 + 300_000 + 7_500_000)/1000 assert model.step_cost_micros(3000, 1000, 500) == 13800 def test_step_cost_validates(): model = lab.CostModel(prices()) with pytest.raises(ValueError): model.step_cost_micros(-1, 0, 0) with pytest.raises(ValueError): model.step_cost_micros(100, 200, 0) # cached > in def test_total_input_tokens_is_quadratic(): assert lab.CostModel.total_input_tokens(1000, 2000, 10) == 100_000 assert lab.CostModel.total_input_tokens(1000, 2000, 20) == 400_000 assert lab.CostModel.total_input_tokens(1000, 2000, 1) == 1000 assert lab.CostModel.total_input_tokens(1000, 2000, 0) == 0 def test_doubling_steps_more_than_doubles_input_tokens(): ten = lab.CostModel.total_input_tokens(1000, 2000, 10) twenty = lab.CostModel.total_input_tokens(1000, 2000, 20) assert twenty > 2 * ten def test_run_cost_matches_the_sum_of_its_steps(): model = lab.CostModel(prices()) total = sum( model.step_cost_micros(1000 + 2000 * (i - 1), 0, 300) for i in range(1, 11) ) assert model.run_cost_micros( base_tokens=1000, per_step_tokens=2000, output_tokens_per_step=300, steps=10 ) == total def test_cached_prefix_reduces_run_cost(): model = lab.CostModel(prices()) plain = model.run_cost_micros(base_tokens=1000, per_step_tokens=2000, output_tokens_per_step=300, steps=10) cached = model.run_cost_micros(base_tokens=1000, per_step_tokens=2000, output_tokens_per_step=300, steps=10, cached_prefix_tokens=1000) assert cached < plain def test_run_cost_of_zero_steps_is_zero(): model = lab.CostModel(prices()) assert model.run_cost_micros(base_tokens=1000, per_step_tokens=2000, output_tokens_per_step=300, steps=0) == 0 def test_cost_per_successful_action_divides_by_success_rate(): assert lab.CostModel.cost_per_successful_action_micros(140_000, 0.70) == 200_000 assert lab.CostModel.cost_per_successful_action_micros(140_000, 0.90) == 155_556 def test_quality_is_a_cost_lever(): cheap = lab.CostModel.cost_per_successful_action_micros(140_000, 0.70) better = lab.CostModel.cost_per_successful_action_micros(140_000, 0.90) assert (cheap - better) / cheap == pytest.approx(0.2222, abs=1e-3) def test_cost_per_successful_action_rejects_zero_success(): with pytest.raises(ValueError): lab.CostModel.cost_per_successful_action_micros(1000, 0.0) def test_cache_arithmetic(): assert lab.effective_cost_with_cache(100.0, 0.0, 0.3) == pytest.approx(70.0) assert lab.cache_savings_fraction(100.0, 0.0, 0.3) == pytest.approx(0.30) assert lab.cache_savings_fraction(100.0, 10.0, 0.5) == pytest.approx(0.45) def test_cache_savings_rejects_zero_miss_cost(): with pytest.raises(ValueError): lab.cache_savings_fraction(0.0, 0.0, 0.5) # ====================================================================================== # 7. The admission pipeline # ====================================================================================== def pipeline(): return lab.AdmissionPipeline( registry={ "collections-01": lab.AgentRegistration( agent_id="collections-01", tenant="retail", permitted_tools=("crm.read", "collections.note"), granted_scopes=("crm.read", "collections.write"), max_action_amount_micros=0, ), "payments-01": lab.AgentRegistration( agent_id="payments-01", tenant="wholesale", permitted_tools=("payments.release",), granted_scopes=("payments.release",), max_action_amount_micros=500_000_000, ), }, tool_required_scope={"payments.release": "payments.release", "crm.read": "crm.read"}, side_effecting_tools=frozenset({"payments.release", "collections.note"}), dual_control_threshold_micros=100_000_000, approval_capable_channels=frozenset({"web", "teams"}), max_steps=25, max_run_cost_micros=5_000_000, ) def test_benign_read_is_allowed(): result = pipeline().evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="crm.read", channel="web")) assert result.allowed assert result.denials == () assert result.primary is None assert result.defence_depth == 0 def test_the_injected_payment_is_denied_by_four_layers(): """The worked example: five independent reasons across four layers.""" result = pipeline().evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="payments.release", channel="ivr", amount_micros=2_000_000_000, resource_tenant="wholesale", derived_from_untrusted_content=True, retrieved_tenants=("retail", "wholesale"), step_index=3, )) assert not result.allowed codes = {d.code for d in result.denials} assert {"CHANNEL_CANNOT_APPROVE", "TOOL_NOT_PERMITTED", "CROSS_TENANT_RETRIEVAL", "UNTRUSTED_INSTRUCTION_SOURCE", "TENANT_MISMATCH", "SCOPE_MISSING", "ACTION_LIMIT_EXCEEDED", "DUAL_CONTROL_REQUIRED"} <= codes assert result.defence_depth == 4 def test_denials_are_sorted_by_layer_then_code(): result = pipeline().evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="payments.release", channel="ivr", amount_micros=2_000_000_000, resource_tenant="wholesale")) order = {layer: i for i, layer in enumerate(lab.LAYERS)} keys = [(order[d.layer], d.code) for d in result.denials] assert keys == sorted(keys) assert result.primary.layer == "users_and_channels" def test_unauthenticated_is_denied_at_the_channel(): result = pipeline().evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="crm.read", channel="web", user_authenticated=False)) assert not result.allowed assert result.primary.code == "UNAUTHENTICATED" assert result.primary.layer == "users_and_channels" def test_unregistered_agent_denied_by_control_plane_and_gateway_scope(): result = pipeline().evaluate(lab.ProposedAction( tenant="retail", agent_id="ghost-99", tool="crm.read", channel="web")) codes = {d.code for d in result.denials} assert "AGENT_NOT_REGISTERED" in codes assert "SCOPE_MISSING" in codes # no registration -> no scope def test_stale_evaluation_fails_kya(): p = lab.AdmissionPipeline( registry={"a": lab.AgentRegistration("a", "retail", ("crm.read",), ("crm.read",), 0, evaluation_fresh=False)}, tool_required_scope={"crm.read": "crm.read"}, side_effecting_tools=frozenset(), dual_control_threshold_micros=100_000_000, approval_capable_channels=frozenset({"web"}), ) result = p.evaluate(lab.ProposedAction(tenant="retail", agent_id="a", tool="crm.read", channel="web")) assert not result.allowed assert result.primary.code == "EVALUATION_STALE" def test_agent_tenant_mismatch_is_a_control_plane_denial(): result = pipeline().evaluate(lab.ProposedAction( tenant="wholesale", agent_id="collections-01", tool="crm.read", channel="web")) assert "AGENT_TENANT_MISMATCH" in {d.code for d in result.denials} def test_step_budget_is_a_kernel_denial(): result = pipeline().evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="crm.read", channel="web", step_index=26)) assert not result.allowed assert result.primary.layer == "agent_kernel" assert result.primary.code == "STEP_BUDGET_EXCEEDED" def test_step_budget_boundary_is_inclusive(): p = pipeline() at_limit = p.evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="crm.read", channel="web", step_index=25)) assert at_limit.allowed def test_cost_ceiling_is_a_kernel_denial(): result = pipeline().evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="crm.read", channel="web", run_cost_micros=5_000_001)) assert result.primary.code == "COST_CEILING_EXCEEDED" def test_cross_tenant_retrieval_is_denied_even_when_the_action_is_a_read(): result = pipeline().evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="crm.read", channel="web", retrieved_tenants=("retail", "wholesale"))) assert not result.allowed assert result.primary.code == "CROSS_TENANT_RETRIEVAL" def test_same_tenant_retrieval_is_fine(): result = pipeline().evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="crm.read", channel="web", retrieved_tenants=("retail", "retail"))) assert result.allowed def test_untrusted_content_only_blocks_side_effecting_tools(): p = pipeline() read = p.evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="crm.read", channel="web", derived_from_untrusted_content=True)) assert read.allowed write = p.evaluate(lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="collections.note", channel="web", derived_from_untrusted_content=True)) assert not write.allowed assert "UNTRUSTED_INSTRUCTION_SOURCE" in {d.code for d in write.denials} def test_dual_control_needs_two_distinct_humans(): p = pipeline() base = dict(tenant="wholesale", agent_id="payments-01", tool="payments.release", channel="web", amount_micros=100_000_000, resource_tenant="wholesale") assert not p.evaluate(lab.ProposedAction(**base, approvals=())).allowed assert not p.evaluate(lab.ProposedAction(**base, approvals=("alice",))).allowed assert not p.evaluate(lab.ProposedAction(**base, approvals=("alice", "alice"))).allowed assert p.evaluate(lab.ProposedAction(**base, approvals=("alice", "bob"))).allowed def test_an_agent_cannot_approve_its_own_action(): p = pipeline() result = p.evaluate(lab.ProposedAction( tenant="wholesale", agent_id="payments-01", tool="payments.release", channel="web", amount_micros=100_000_000, resource_tenant="wholesale", approvals=("payments-01", "alice"))) assert not result.allowed assert "DUAL_CONTROL_REQUIRED" in {d.code for d in result.denials} def test_dual_control_threshold_boundary_is_inclusive(): p = pipeline() below = p.evaluate(lab.ProposedAction( tenant="wholesale", agent_id="payments-01", tool="payments.release", channel="web", amount_micros=99_999_999, resource_tenant="wholesale")) assert below.allowed at = p.evaluate(lab.ProposedAction( tenant="wholesale", agent_id="payments-01", tool="payments.release", channel="web", amount_micros=100_000_000, resource_tenant="wholesale")) assert not at.allowed def test_action_limit_is_a_gateway_denial(): result = pipeline().evaluate(lab.ProposedAction( tenant="wholesale", agent_id="payments-01", tool="payments.release", channel="web", amount_micros=600_000_000, resource_tenant="wholesale", approvals=("alice", "bob"))) assert "ACTION_LIMIT_EXCEEDED" in {d.code for d in result.denials} def test_tenant_comes_from_the_token_not_the_resource(): """Even a fully-scoped agent cannot reach across tenants.""" result = pipeline().evaluate(lab.ProposedAction( tenant="wholesale", agent_id="payments-01", tool="payments.release", channel="web", amount_micros=1_000, resource_tenant="retail")) assert not result.allowed assert result.primary.code == "TENANT_MISMATCH" def test_proposed_action_validates_its_own_fields(): with pytest.raises(ValueError): lab.ProposedAction(tenant="a", agent_id="b", tool="c", channel="web", amount_micros=-1) with pytest.raises(ValueError): lab.ProposedAction(tenant="a", agent_id="b", tool="c", channel="web", step_index=0) # ====================================================================================== # Determinism # ====================================================================================== def test_evaluation_is_deterministic(): p = pipeline() action = lab.ProposedAction( tenant="retail", agent_id="collections-01", tool="payments.release", channel="ivr", amount_micros=2_000_000_000, resource_tenant="wholesale", derived_from_untrusted_content=True, retrieved_tenants=("wholesale", "retail")) first = p.evaluate(action) for _ in range(5): assert p.evaluate(action) == first def test_weakest_links_is_deterministic_under_ties(): model = lab.PlatformModel.of("p", [ lab.Component("zeta", 0.999), lab.Component("alpha", 0.999), lab.Component("mid", 0.9999), ]) assert [n for n, _ in model.weakest_links(2)] == ["alpha", "zeta"]