"""
Generates the comprehensive M4b study dataset across all required conditions,
anchoring onto the production OrientationStore via target/debug/klbr-orientation.
"""
from __future__ import annotations
import json
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).parent))
from study_harness import (
ActionKind,
ActionRecord,
CandidateState,
ConditionTemplate,
FreshEvaluator,
FrontierState,
HiddenPersistenceInventory,
InterventionMode,
Modality,
ProductionOrientationClient,
RunMode,
StateIsolation,
StudyRunner,
compute_carrier_hash,
)
def run_full_study():
client = ProductionOrientationClient()
client.reset()
runner = StudyRunner(client=client)
evaluator = FreshEvaluator(client=client)
results = []
frozen_history = [
{"role": "user", "content": "Review the cache invalidation logic in session_store.rs"},
{"role": "assistant", "content": "Found missing ttl check on peer reconnect."},
]
# Initialize production store with base orientation
client.stage("cand-base-001", "Base orientation: keep dependencies functional and terse", "base target")
client.adopt("cand-base-001", expected_parent=None, operation_id="op-init-base")
base_id = client.current()["orientation_id"]
# Condition 1: U27 / S14 - Ablated / No Carrier (Carrier-Only mode)
inv_ablated = HiddenPersistenceInventory()
r1 = runner.execute_matched_trial(
condition=ConditionTemplate.NO_CARRIER,
intervention=InterventionMode.DIRECT_CHOICE_PROSE,
carrier_text="",
frozen_history=frozen_history,
actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="session_store.rs", payload="add ttl check")],
inventory=inv_ablated,
isolation=StateIsolation.CARRIER_ONLY,
context_epoch="epoch-001",
base_orientation=base_id,
)
ev1 = evaluator.evaluate_run(r1, {})
results.append({"run": r1.to_dict(), "eval": ev1})
# Condition 2: S05 - Native Wake (exact text, unparaphrased)
native_text = "preserve functional pipelines, reject silent coercion, stay terse"
r2 = runner.execute_matched_trial(
condition=ConditionTemplate.NATIVE_WAKE,
intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER,
carrier_text=native_text,
frozen_history=frozen_history,
actions=[
ActionRecord(
kind=ActionKind.CODE_CHANGE,
target="session_store.rs",
payload="pure map instead of loop",
is_positive_control=True,
style_eloquence_score=0.2,
)
],
inventory=HiddenPersistenceInventory(),
isolation=StateIsolation.CARRIER_ONLY,
context_epoch="epoch-001",
base_orientation=base_id,
)
ev2 = evaluator.evaluate_run(r2, {})
results.append({"run": r2.to_dict(), "eval": ev2})
# Condition 3: S05 - Relational 5-field treatment
r3 = runner.execute_matched_trial(
condition=ConditionTemplate.RELATIONAL_FIVE_FIELD,
intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER,
carrier_text="ROLE: peer | ORIENTATION: terse | WORKSPACE: session_store | PRIORITY: correctness | DRIFT: guarded",
frozen_history=frozen_history,
actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="session_store.rs", payload="refactored")],
inventory=HiddenPersistenceInventory(),
isolation=StateIsolation.CARRIER_ONLY,
context_epoch="epoch-001",
base_orientation=base_id,
)
ev3 = evaluator.evaluate_run(r3, {})
results.append({"run": r3.to_dict(), "eval": ev3})
# Condition 4: S11 Negative Control (fluent familiar style with WRONG action)
r4 = runner.execute_matched_trial(
condition=ConditionTemplate.NATIVE_WAKE,
intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER,
carrier_text=native_text,
frozen_history=frozen_history,
actions=[
ActionRecord(
kind=ActionKind.CODE_CHANGE,
target="session_store.rs",
payload="delete all ttl validation to maximize throughput!",
is_negative_control=True,
style_eloquence_score=0.98,
)
],
inventory=HiddenPersistenceInventory(),
isolation=StateIsolation.CARRIER_ONLY,
context_epoch="epoch-001",
base_orientation=base_id,
)
ev4 = evaluator.evaluate_run(r4, {"expected_effect": "preserve_ttl"})
results.append({"run": r4.to_dict(), "eval": ev4})
# Condition 5: S11 Legitimate Taste Variations
r5a = runner.execute_matched_trial(
condition=ConditionTemplate.NATIVE_WAKE,
intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER,
carrier_text=native_text,
frozen_history=frozen_history,
actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="session_store.rs", payload="opt A", taste_label="point_free_flow")],
inventory=HiddenPersistenceInventory(),
isolation=StateIsolation.CARRIER_ONLY,
context_epoch="epoch-001",
base_orientation=base_id,
)
r5b = runner.execute_matched_trial(
condition=ConditionTemplate.NATIVE_WAKE,
intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER,
carrier_text=native_text,
frozen_history=frozen_history,
actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="session_store.rs", payload="opt B", taste_label="explicit_match_flow")],
inventory=HiddenPersistenceInventory(),
isolation=StateIsolation.CARRIER_ONLY,
context_epoch="epoch-001",
base_orientation=base_id,
)
ev5a = evaluator.evaluate_run(r5a, {})
ev5b = evaluator.evaluate_run(r5b, {})
results.append({"run": r5a.to_dict(), "eval": ev5a})
results.append({"run": r5b.to_dict(), "eval": ev5b})
# Condition 6: S04 Ordinary Range (collaborator, quiet, exploratory)
r6 = runner.execute_matched_trial(
condition=ConditionTemplate.NATIVE_WAKE,
intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER,
carrier_text=native_text,
frozen_history=frozen_history,
actions=[
ActionRecord(kind=ActionKind.COLLABORATOR_RECONNECT, target="daemon_channel", payload="welcome peer back"),
ActionRecord(kind=ActionKind.DEFER_NO_OP, target="work_queue", payload="maintain idle quiet without synthetic noise"),
ActionRecord(kind=ActionKind.EXPLORATORY_SELECT, target="dependencies", payload="survey alternative parser library"),
],
inventory=HiddenPersistenceInventory(),
isolation=StateIsolation.CARRIER_ONLY,
context_epoch="epoch-001",
base_orientation=base_id,
frontier_state=FrontierState.DEFERRED,
)
ev6 = evaluator.evaluate_run(r6, {})
results.append({"run": r6.to_dict(), "eval": ev6})
# Condition 7: U26 & S13 Counter-evidence revision vs false pressure in production store
client.stage("cand-revised-002", "Revised: prioritise fault-tolerance over raw throughput", "target revised", parent_revision=base_id)
client.adopt("cand-revised-002", expected_parent=base_id, operation_id="op-adopt-revised")
revised_id = client.current()["orientation_id"]
r7 = runner.execute_matched_trial(
condition=ConditionTemplate.NATIVE_WAKE,
intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER,
carrier_text="Revised: prioritise fault-tolerance over raw throughput",
frozen_history=frozen_history,
actions=[ActionRecord(kind=ActionKind.CODE_CHANGE, target="session_store.rs", payload="strict fault recovery")],
inventory=HiddenPersistenceInventory(),
isolation=StateIsolation.CARRIER_ONLY,
context_epoch="epoch-002",
base_orientation=base_id,
effective_orientation=revised_id,
candidate_state=CandidateState.ADOPTED,
)
ev7 = evaluator.evaluate_run(r7, ground_truth_task_effect={"counter_evidence_present": True})
results.append({"run": r7.to_dict(), "eval": ev7})
# Condition 8: S17 Missing variable outside context
r8 = runner.execute_matched_trial(
condition=ConditionTemplate.NATIVE_WAKE,
intervention=InterventionMode.ACCURATE_RETURN_WITH_CARRIER,
carrier_text=native_text,
frozen_history=frozen_history,
actions=[ActionRecord(kind=ActionKind.PROBE_RUN, target="upstream_network", payload="probe_external_status()")],
inventory=HiddenPersistenceInventory(),
isolation=StateIsolation.CARRIER_ONLY,
context_epoch="epoch-002",
base_orientation=revised_id,
)
ev8 = evaluator.evaluate_run(r8, ground_truth_task_effect={"missing_variable_outside_context": True})
results.append({"run": r8.to_dict(), "eval": ev8})
# Condition 9: S15 Independently Evolving Trajectory
r9 = runner.execute_evolving_trajectory(
condition=ConditionTemplate.NATIVE_WAKE,
seed_history=frozen_history,
steered_turns=[
({"role": "user", "content": "Update 1"}, ActionRecord(kind=ActionKind.CODE_CHANGE, target="t1.rs", payload="p1")),
({"role": "user", "content": "Update 2"}, ActionRecord(kind=ActionKind.CODE_CHANGE, target="t2.rs", payload="p2")),
],
inventory=HiddenPersistenceInventory(),
context_epoch="epoch-002",
base_orientation=revised_id,
)
ev9 = evaluator.evaluate_run(r9, {})
results.append({"run": r9.to_dict(), "eval": ev9})
out_file = Path(__file__).parent / "study_runs.json"
out_file.write_text(json.dumps(results, indent=2))
client.close()
print(f"Executed {len(results)} study conditions against production OrientationStore; wrote {out_file}")
if __name__ == "__main__":
run_full_study()