"""
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()