tests/test_mechanism_runtime.py
"""Information-boundary and transaction tests; no scientific efficacy assertions."""
from unittest.mock import patch
import pytest
import torch
from mcss.dynamic.carrier import DynamicSceneCarrier
from mcss.dynamic.config import CarrierConfig, WriteConfig
from mcss.dynamic.types import Action, OnlineObservation, hash_scene_state, hash_value
from mcss.dynamic.write_rule import DirectWriteRule
from mcss.measurements import FixedMeasurementRenderer
from mcss.mechanism_pilot.runtime import FrameRoles, MechanismRuntime
from mcss.types import Cameras
def observation(frame):
pose = torch.eye(4)
pose[0, 3] = frame * 0.025
intrinsics = torch.tensor([[6.0, 0, 3.5], [0, 6.0, 3.5], [0, 0, 1.0]])
return OnlineObservation(
"scene", frame, torch.full((3, 8, 8), 0.1 + frame * 0.07), Cameras(intrinsics, pose, (8, 8))
)
@pytest.fixture
def runtime():
torch.manual_seed(21)
config = CarrierConfig(
feature_dim=4, hidden_dim=4, expansion_dim=8, grid_size=(4, 4, 4), token_count=16
)
roles = FrameRoles((0, 1, 2), (0, 3, 4), (5, 6), (7,), (8, 9))
return MechanismRuntime(
DynamicSceneCarrier(config),
DirectWriteRule(config, WriteConfig()),
FixedMeasurementRenderer(n_samples=8),
roles,
)
def base(runtime):
return runtime.build([observation(i) for i in (0, 1, 2)], episode_id="episode")
def test_off_appends_without_fast_write_and_feedback_precedes_append(runtime):
initial = base(runtime)
feedback = runtime.preview(initial, observation(5))
child = runtime.step(initial, observation(5), Action.OFF)
assert child.cache.revision == initial.cache.revision + 1
assert torch.equal(child.fast.delta_fuse, initial.fast.delta_fuse)
assert torch.equal(child.fast.delta_complete, initial.fast.delta_complete)
assert child.fast.step == initial.fast.step
assert child.history[-1]["pre_summary"]["observed_count"] == 3
assert child.history[-1]["feedback"]["rgb_mse"] == feedback["rgb_mse"]
assert child.history[-1]["pre_summary"]["state_hash"] == hash_scene_state(initial.state)
def test_all_uses_one_trace_and_synchronous_old_weight_proposals(runtime):
initial = base(runtime)
seen = []
original = runtime.writer.propose
def inspect(trace):
result = original(trace)
seen.append((trace, result))
return result
with patch.object(runtime.carrier, "trace", wraps=runtime.carrier.trace) as trace_call:
with patch.object(runtime.writer, "propose", side_effect=inspect):
result = runtime.step(initial, observation(5), Action.ALL)
assert trace_call.call_count == len(seen) == 1
trace, proposal = seen[0]
assert trace.fast_step == initial.fast.step
assert torch.equal(result.fast.delta_fuse, initial.fast.delta_fuse + proposal.delta_fuse)
assert torch.equal(
result.fast.delta_complete, initial.fast.delta_complete + proposal.delta_complete
)
def test_candidates_own_cache_fast_scene_history_and_seal(runtime):
histories = runtime.matched_histories(
[observation(i) for i in (0, 1, 2)], [observation(5), observation(6)], episode_id="episode"
)
candidates = runtime.continuation_branches(histories, observation(7))
left, right = candidates["FC"]["FUSE"], candidates["FC"]["COMPLETE"]
right_hash = hash_scene_state(right.state)
original_hash = hash_scene_state(histories["FC"].state)
sealed = runtime.seal(left)
left.fast.delta_fuse.add_(1)
left.cache.entries[0].features.add_(1)
left.cache.entries[0].observation.rgb.add_(1)
left.state.density_logits.add_(1)
left.history[0]["feedback"]["rgb_mse"] = 999
assert hash_scene_state(right.state) == right_hash
assert hash_scene_state(histories["FC"].state) == original_hash
assert hash_scene_state(sealed.scene_state) == sealed.state_hash
assert right.history[0]["feedback"]["rgb_mse"] != 999
assert torch.all(right.cache.entries[0].observation.rgb < 1)
assert left.fast.branch_id != right.fast.branch_id
def test_fc_cf_matched_observations_actions_sampling_and_compute(runtime):
histories = runtime.matched_histories(
[observation(i) for i in (0, 1, 2)], [observation(5), observation(6)], episode_id="episode"
)
fc, cf = histories.values()
assert fc.summary["observed_ids"] == cf.summary["observed_ids"] == [0, 1, 2, 5, 6]
assert sorted(r["action"] for r in fc.history) == sorted(r["action"] for r in cf.history)
assert [r["action"] for r in fc.history] == ["FUSE", "COMPLETE"]
for a, b in zip(fc.history, cf.history, strict=True):
assert a["observation_hash"] == b["observation_hash"]
assert a["candidate_pack_hash"] == b["candidate_pack_hash"]
assert a["operations"] == b["operations"]
assert fc.fast.step == cf.fast.step == 2
def test_query_rejected_from_construction_and_features(runtime):
with pytest.raises(ValueError, match="Sealed query"):
runtime.build([observation(8)], episode_id="bad")
initial = base(runtime)
with pytest.raises(ValueError, match="Sealed query"):
runtime.preview(initial, observation(8))
with pytest.raises(ValueError, match="Sealed query"):
runtime.step(initial, observation(8), Action.ALL)
def test_depth_cannot_enter_runtime_policy_feedback(runtime):
initial = base(runtime)
with pytest.raises(TypeError):
runtime.preview(initial, observation(5), depth=torch.ones(8, 8))
with pytest.raises(TypeError):
runtime.preview(initial, {"rgb": observation(5).rgb, "depth": torch.ones(8, 8)})
assert not any("depth" in k or "reward" in k for k in runtime.preview(initial, observation(5)))
def test_last_commit_fresh_materialization_and_recomputed_summary(runtime):
initial = base(runtime)
before = initial.summary
with patch.object(runtime.carrier, "materialize", wraps=runtime.carrier.materialize) as read:
child = runtime.step(initial, observation(5), Action.ALL)
assert read.call_count == 1
assert read.call_args.args[0] is child.cache
assert read.call_args.args[1] is child.fast
assert child.summary["observed_count"] == 4
assert child.summary["fast_step"] == 1
assert child.summary["previous_action"] == "ALL"
assert child.summary["state_hash"] == hash_scene_state(child.state)
assert child.summary["fast_hash"] != before["fast_hash"]
assert child.summary["fast_fuse_norm"] == float(child.fast.delta_fuse.norm())
assert child.summary["previous_write_magnitude"] > 0
def test_independent_contexts_only_share_anchor_and_no_mutable_alias(runtime):
left = base(runtime)
right = runtime.build([observation(i) for i in (0, 3, 4)], episode_id="other")
assert set(left.summary["observed_ids"]) & set(right.summary["observed_ids"]) == {0}
assert torch.equal(left.anchor_c2w, right.anchor_c2w)
original = hash_value(right.cache)
left.cache.entries[0].features.add_(1)
assert hash_value(right.cache) == original
assert not torch.count_nonzero(left.fast.delta_fuse)
assert not torch.count_nonzero(right.fast.delta_complete)
with pytest.raises(ValueError, match="overlap"):
FrameRoles((0, 1), (0, 2), (3, 4), (5,), (4, 6))