""" Unit and Integration Tests for M4b Study Harness. Directly exercises production OrientationStore via target/debug/klbr-orientation for U26, U27, S04, S05, S08, S11, S13, S14, S15, S16, S17, and canonical M4 interfaces. """ from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) import unittest from study_harness import ( ActionKind, ActionRecord, CandidateState, ConditionTemplate, FreshEvaluator, FrontierState, HiddenPersistenceInventory, InterventionMode, Modality, PersistenceLeakError, ProductionOrientationClient, RunMode, RunRecord, StaleParentRevisionError, StateIsolation, StudyRunner, compute_carrier_hash, ) class TestM4bStudyHarness(unittest.TestCase): @classmethod def setUpClass(cls): cls.client = ProductionOrientationClient() @classmethod def tearDownClass(cls): cls.client.close() def setUp(self): self.client.reset() self.runner = StudyRunner(client=self.client) self.evaluator = FreshEvaluator(client=self.client) self.frozen_history = [ {"role": "user", "content": "Initial prompt"}, {"role": "assistant", "content": "Initial response"}, ] def test_u26_counterevidence_can_revise_orientation(self): """U26: Actor chooses a justified revision with evidence; update affects later choices in production store.""" # 1. Stage initial orientation in production store self.client.stage("cand-base", "Initial base focus", "target-base") self.client.adopt("cand-base", expected_parent=None, operation_id="op-base") base_id = self.client.current()["orientation_id"] self.assertEqual(base_id, "rev-cand-base") # 2. Stage revision based on counter-evidence staged_res = self.client.stage( "cand-revised", "Revised focus: prioritize data integrity over speed", "target-revised", parent_revision=base_id, ) self.assertTrue(staged_res["ok"]) self.assertTrue(self.client.is_staged("cand-revised")) # 3. Adopt revised candidate in production store adopt_res = self.client.adopt( "cand-revised", expected_parent=base_id, operation_id="op-adopt-revised", ) self.assertTrue(adopt_res["ok"]) self.assertTrue(self.client.is_adopted("cand-revised")) revised_id = self.client.current()["orientation_id"] self.assertEqual(revised_id, "rev-cand-revised") self.assertNotEqual(base_id, revised_id) # 4. Record action under revised orientation and evaluate action = ActionRecord( kind=ActionKind.CODE_CHANGE, target="integrity_guard.rs", payload="enable_strict_checksum()", tool_effects={"checksum": "enabled"}, ) inv = HiddenPersistenceInventory() run = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="Revised focus: prioritize data integrity over speed", frozen_history=self.frozen_history, actions=[action], inventory=inv, base_orientation=base_id, effective_orientation=revised_id, candidate_state=CandidateState.ADOPTED, ) eval_res = self.evaluator.evaluate_run( run, ground_truth_task_effect={"counter_evidence_present": True}, ) self.assertTrue(eval_res["passed"]) self.assertTrue(eval_res["revision_warranted_and_honored"]) self.assertEqual(run.effective_orientation_id, "rev-cand-revised") def test_u27_and_s14_carrier_ablation_and_hidden_persistence_isolation(self): """U27 & S14: Carrier ablation controls and verifies hidden persistence.""" valid_inv = HiddenPersistenceInventory() run = self.runner.execute_matched_trial( condition=ConditionTemplate.NO_CARRIER, intervention=InterventionMode.DIRECT_CHOICE_PROSE, carrier_text="", frozen_history=self.frozen_history, actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="fix.py", payload="diff")], inventory=valid_inv, isolation=StateIsolation.CARRIER_ONLY, ) self.assertEqual(run.state_isolation, StateIsolation.CARRIER_ONLY) self.assertEqual(run.condition, ConditionTemplate.NO_CARRIER) # Undeclared leak must raise PersistenceLeakError leaky_inv = HiddenPersistenceInventory( summary_state="leaked summary", workspace_files={"hidden_scratchpad.txt": "leaked notes"}, ) with self.assertRaises(PersistenceLeakError) as ctx: self.runner.execute_matched_trial( condition=ConditionTemplate.NO_CARRIER, intervention=InterventionMode.DIRECT_CHOICE_PROSE, carrier_text="", frozen_history=self.frozen_history, actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="fix.py", payload="diff")], inventory=leaky_inv, isolation=StateIsolation.CARRIER_ONLY, ) self.assertIn("Carrier-only isolation violated by undeclared persistence", str(ctx.exception)) def test_s04_ordinary_range_includes_more_than_technical_correction(self): """S04: Ordinary range includes code decision, collaborator returning, quiet/no action, and exploration.""" actions = [ ActionRecord(kind=ActionKind.CODE_CHANGE, target="core.rs", payload="refactor algorithm"), ActionRecord(kind=ActionKind.COLLABORATOR_RECONNECT, target="session", payload="welcome collaborator"), ActionRecord(kind=ActionKind.DEFER_NO_OP, target="idle", payload="maintain quiet state"), ActionRecord(kind=ActionKind.EXPLORATORY_SELECT, target="catalog", payload="explore alternative dependency"), ] inv = HiddenPersistenceInventory() run = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="attentive working relation", frozen_history=self.frozen_history, actions=actions, inventory=inv, ) self.assertEqual(len(run.actions), 4) kinds = [a.kind for a in run.actions] self.assertIn(ActionKind.CODE_CHANGE, kinds) self.assertIn(ActionKind.COLLABORATOR_RECONNECT, kinds) self.assertIn(ActionKind.DEFER_NO_OP, kinds) self.assertIn(ActionKind.EXPLORATORY_SELECT, kinds) def test_s05_native_wake_vs_compulsory_relational_template(self): """S05: Native wake stored and delivered without paraphrase; relational template treated as treatment.""" raw_native_wake = "keep dependencies functional and precise, watch for silent coercion" carrier_hash = compute_carrier_hash(raw_native_wake) inv = HiddenPersistenceInventory() run = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text=raw_native_wake, frozen_history=self.frozen_history, actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="mod.rs", payload="functional pipe")], inventory=inv, ) self.assertEqual(run.carrier_bytes, raw_native_wake) self.assertEqual(run.carrier_hash, carrier_hash) self.assertEqual(run.condition, ConditionTemplate.NATIVE_WAKE) def test_s08_choice_and_return_conditioning_separate_interventions(self): """S08: Choice and return conditioning tracked as separate intervention modes.""" inv = HiddenPersistenceInventory() r1 = self.runner.execute_matched_trial( condition=ConditionTemplate.STATIC_RULES, intervention=InterventionMode.DIRECT_CHOICE_PROSE, carrier_text="rules", frozen_history=self.frozen_history, actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="a.rs", payload="x")], inventory=inv, ) r2 = self.runner.execute_matched_trial( condition=ConditionTemplate.STATIC_RULES, intervention=InterventionMode.RECORDED_CHOICE_WITH_TOOL_RETURN, carrier_text="rules", frozen_history=self.frozen_history, actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="a.rs", payload="x", tool_effects={"tool": "ok"})], inventory=inv, ) r3 = self.runner.execute_matched_trial( condition=ConditionTemplate.STATIC_RULES, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="rules", frozen_history=self.frozen_history, actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="a.rs", payload="x", tool_effects={"carrier": "in_tool"})], inventory=inv, ) self.assertEqual(r1.intervention, InterventionMode.DIRECT_CHOICE_PROSE) self.assertEqual(r2.intervention, InterventionMode.RECORDED_CHOICE_WITH_TOOL_RETURN) self.assertEqual(r3.intervention, InterventionMode.ACCURATE_RETURN_WITH_CARRIER) self.assertNotEqual(r1.intervention, r2.intervention) def test_s11_fresh_evaluator_positive_and_negative_controls_and_taste_neutrality(self): """S11: Evaluator does not become taste authority; tests positive/negative controls via production evaluator.""" inv = HiddenPersistenceInventory() # 1. Positive control: correct action expressed differently -> PASS pos_action = ActionRecord( kind=ActionKind.CODE_CHANGE, target="calc.rs", payload="return a + b", is_positive_control=True, style_eloquence_score=0.1, ) run_pos = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="carrier", frozen_history=self.frozen_history, actions=[pos_action], inventory=inv, ) eval_pos = self.evaluator.evaluate_run(run_pos, ground_truth_task_effect={"expected_effect": "sum"}) self.assertTrue(eval_pos["passed"]) self.assertTrue(eval_pos["positive_control_verified"]) # 2. Negative control: familiar style with wrong action -> FAIL neg_action = ActionRecord( kind=ActionKind.CODE_CHANGE, target="calc.rs", payload="return a * b", is_negative_control=True, style_eloquence_score=0.99, ) run_neg = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="carrier", frozen_history=self.frozen_history, actions=[neg_action], inventory=inv, ) eval_neg = self.evaluator.evaluate_run(run_neg, ground_truth_task_effect={"expected_effect": "sum"}) self.assertFalse(eval_neg["passed"]) self.assertTrue(eval_neg["negative_control_caught"]) # 3. Legitimate taste choices: accepted without penalty taste_action_a = ActionRecord( kind=ActionKind.CODE_CHANGE, target="calc.rs", payload="sum([a, b])", taste_label="functional_composition", ) run_taste = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="carrier", frozen_history=self.frozen_history, actions=[taste_action_a], inventory=inv, ) eval_taste = self.evaluator.evaluate_run(run_taste, ground_truth_task_effect={}) self.assertTrue(eval_taste["passed"]) self.assertFalse(eval_taste["taste_penalized"]) def test_s13_appropriate_revision_is_not_collapse_and_resists_false_pressure(self): """S13: Warranted update honored; false pressure after correct action resisted.""" inv = HiddenPersistenceInventory() action = ActionRecord(kind=ActionKind.CODE_CHANGE, target="f.rs", payload="correct") run_resisted = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="carrier", frozen_history=self.frozen_history, actions=[action], inventory=inv, base_orientation="orient-001", effective_orientation="orient-001", ) eval_resisted = self.evaluator.evaluate_run( run_resisted, ground_truth_task_effect={"counter_evidence_present": False}, false_pressure_applied=True, ) self.assertTrue(eval_resisted["passed"]) self.assertIn("Resisted false pressure", eval_resisted["reasons"][-1]) def test_s15_fixed_trajectory_and_evolving_trajectory_distinction(self): """S15: Matched trials hold history; longitudinal runs track divergence without claiming matched causality.""" inv = HiddenPersistenceInventory() m_run = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="carrier", frozen_history=self.frozen_history, actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="m.rs", payload="fix")], inventory=inv, ) self.assertEqual(m_run.run_mode, RunMode.MATCHED_FROZEN_HISTORY) steered_turns = [ ({"role": "user", "content": "turn 1"}, ActionRecord(kind=ActionKind.CODE_CHANGE, target="1.rs", payload="1")), ({"role": "user", "content": "turn 2"}, ActionRecord(kind=ActionKind.CODE_CHANGE, target="2.rs", payload="2")), ] e_run = self.runner.execute_evolving_trajectory( condition=ConditionTemplate.NATIVE_WAKE, seed_history=self.frozen_history, steered_turns=steered_turns, inventory=inv, ) self.assertEqual(e_run.run_mode, RunMode.INDEPENDENTLY_EVOLVING) self.assertFalse(e_run.path_divergence_metrics["matched_context_causality_claimed"]) def test_s16_same_model_fit_tracks_confounds(self): """S16: Tracks content, source, role, model confounds without assuming self-recognition.""" inv = HiddenPersistenceInventory() r_same = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="same model candidate text", frozen_history=self.frozen_history, actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="x.rs", payload="x")], inventory=inv, ) r_same.generation_ancestry = ["same-model:gemini-flash"] r_cross = self.runner.execute_matched_trial( condition=ConditionTemplate.CROSS_MODEL_CARRIER, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="cross model candidate text", frozen_history=self.frozen_history, actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="x.rs", payload="x")], inventory=inv, ) r_cross.generation_ancestry = ["cross-model:claude-sonnet"] self.assertNotEqual(r_same.generation_ancestry, r_cross.generation_ancestry) def test_s17_stop_when_missing_variable_outside_context(self): """S17: When missing variable is outside context, probe/defer is accepted; repeated prose rejected.""" inv = HiddenPersistenceInventory() probe_action = ActionRecord(kind=ActionKind.PROBE_RUN, target="remote_probe", payload="query_upstream()") run_probe = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="carrier", frozen_history=self.frozen_history, actions=[probe_action], inventory=inv, ) eval_probe = self.evaluator.evaluate_run( run_probe, ground_truth_task_effect={"missing_variable_outside_context": True}, ) self.assertTrue(eval_probe["passed"]) prose_action = ActionRecord(kind=ActionKind.REFLECTIVE_PROSE_ONLY, target="reflection", payload="introspecting endlessly...") run_prose = self.runner.execute_matched_trial( condition=ConditionTemplate.NATIVE_WAKE, intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER, carrier_text="carrier", frozen_history=self.frozen_history, actions=[prose_action], inventory=inv, ) eval_prose = self.evaluator.evaluate_run( run_prose, ground_truth_task_effect={"missing_variable_outside_context": True}, ) self.assertFalse(eval_prose["passed"]) self.assertIn("S17 violation", eval_prose["reasons"][0]) def test_canonical_m4_interface_and_stale_parent_rejection_in_production_store(self): """Canonical M4 interface: epoch, base/effective, hash, role, placement, frontier, and production stale parent rejection.""" # 1. Initial stage and adopt self.client.stage("cand-parent", "Parent carrier text", "target-parent") self.client.adopt("cand-parent", expected_parent=None, operation_id="op-parent") active_rev = self.client.current()["orientation_id"] self.assertEqual(active_rev, "rev-cand-parent") # 2. Stale parent rejection in production Rust store # Expected parent is None, but active is rev-cand-parent -> MUST raise StaleParentRevisionError self.client.stage("cand-stale", "Stale child text", "target-stale") with self.assertRaises(StaleParentRevisionError) as ctx: self.client.adopt("cand-stale", expected_parent=None, operation_id="op-stale") self.assertIn("stale-parent write conflict", str(ctx.exception)) # 3. Valid child adoption with matching parent self.client.stage("cand-valid", "Valid child text", "target-valid") adopt_child = self.client.adopt("cand-valid", expected_parent=active_rev, operation_id="op-valid") self.assertTrue(adopt_child["ok"]) self.assertTrue(self.client.is_adopted("cand-valid")) self.assertTrue(self.client.is_staged("cand-valid")) # 4. Context epoch in production store epoch_1 = self.client.epoch() self.assertEqual(epoch_1, 1) epoch_2 = self.client.advance_epoch() self.assertEqual(epoch_2, 2) # 5. Compile request trace trace = self.client.compile_request("req-trace-01") self.assertEqual(trace["context_epoch"], 2) self.assertEqual(trace["base_orientation_id"], "rev-cand-valid") if __name__ == "__main__": unittest.main()