scripts/run_3d_mechanism_smoke.py
"""Synthetic engineering run only; never establishes any scientific hypothesis."""
import argparse
import json
import time
from dataclasses import asdict
from pathlib import Path
import torch
from mcss.dynamic.carrier import DynamicSceneCarrier
from mcss.dynamic.config import CarrierConfig, WriteConfig
from mcss.dynamic.feedback import batched_camera
from mcss.dynamic.types import hash_scene_state, hash_value
from mcss.dynamic.write_rule import DirectWriteRule
from mcss.geometry import project_world
from mcss.measurements import FixedMeasurementRenderer
from mcss.mechanism_pilot.contracts import EvaluationLedger
from mcss.mechanism_pilot.runtime import FrameRoles, MechanismRuntime
from mcss.mechanism_pilot.statistics import analyze_history, measurement_metrics, scene_macro
from mcss.mechanism_pilot.synthetic import arrived_observation, fixture, fixture_camera
ROOT = Path(__file__).resolve().parents[1]
NAME = "EXP-3D-20260927-state-history-pilot-v1"
def write(path, value):
path.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n")
@torch.no_grad()
def run(output):
if output.exists() and any(output.iterdir()):
raise FileExistsError("Refusing to overwrite an existing smoke run")
output.mkdir(parents=True, exist_ok=True)
torch.set_num_threads(1)
torch.manual_seed(20260927)
started = time.perf_counter()
config = CarrierConfig()
carrier = DynamicSceneCarrier(config)
writer = DirectWriteRule(config, WriteConfig())
renderer = FixedMeasurementRenderer()
roles = FrameRoles((0, 1, 2), (0, 3, 4), (5, 6), (7,), (8, 9))
runtime = MechanismRuntime(carrier, writer, renderer, roles)
# Freeze implementation and actual random-weight content before inference.
write(
output / "run_lock.json",
{
"scope": "SYNTHETIC_ENGINEERING_ONLY_RANDOM_UNTRAINED_WEIGHTS",
"seed": 20260927,
"carrier_config": asdict(config),
"write_config": asdict(WriteConfig()),
"frame_roles": asdict(roles),
"image_size": [16, 16],
"renderer_samples": 64,
"checkpoint_hash": runtime.checkpoint_hash,
"config_hash": runtime.config_hash,
"bootstrap_draws": 10000,
"tie_tolerance": 1e-8,
"primary_metric": "depth_absrel",
"scientific_scenes": 0,
"synthetic_fixtures": 3,
},
)
ids = [
f"{s}/{method}"
for s in range(3)
for method in ["A", "B", "anchor"]
+ [f"{h}/{a}" for h in ["FC", "CF"] for a in ["OFF", "FUSE", "COMPLETE", "ALL"]]
]
ledger = EvaluationLedger(ids)
sealed, branches, histories, predictions_pre = {}, {}, {}, {}
history_raw, numerical = [], []
construction_started = time.perf_counter()
for s in range(3):
observed = {i: arrived_observation(s, i) for i in range(8)}
for name, frames in [("A", roles.context_a), ("B", roles.context_b), ("anchor", (0,))]:
branch = runtime.build([observed[i] for i in frames], episode_id=f"s{s}-{name}")
branches[f"{s}/{name}"] = branch
matched = runtime.matched_histories(
[observed[i] for i in roles.context_a],
[observed[i] for i in roles.stream],
episode_id=f"s{s}-matched",
)
histories[s] = matched
for h, branch in matched.items():
pre = renderer(branch.state, batched_camera(observed[7].camera), ("rgb", "depth"))
predictions_pre[s, h] = pre
numerical.append(
{
"scene_id": f"synthetic-{s}",
"continuation_id": 7,
"fast_distance": float(
torch.sqrt(
sum(
(getattr(matched["FC"].fast, n) - getattr(matched["CF"].fast, n))
.square()
.sum()
for n in ["delta_fuse", "delta_complete"]
)
)
),
"pre_rgb_rms_difference": float(
(predictions_pre[s, "FC"]["rgb"] - predictions_pre[s, "CF"]["rgb"])
.square()
.mean()
.sqrt()
),
"pre_depth_rms_difference": float(
(predictions_pre[s, "FC"]["depth"] - predictions_pre[s, "CF"]["depth"])
.square()
.mean()
.sqrt()
),
}
)
candidates = runtime.continuation_branches(matched, observed[7])
for h, actions in candidates.items():
for action, branch in actions.items():
branches[f"{s}/{h}/{action}"] = branch
history_raw.append(
{
"scene_id": f"synthetic-{s}",
"history": h,
"action": action,
"summary": branch.summary,
"history_records": branch.history,
}
)
construction_seconds = time.perf_counter() - construction_started
# The evaluator receives only sealed states, never a live runtime or writer.
for key, branch in branches.items():
sealed[key] = runtime.seal(branch)
ledger.seal(key, sealed[key])
write(
output / "state_manifest.json",
{
key: {
"state_hash": v.state_hash,
"observed_ids": v.observed_ids,
"fast_state_hash": v.fast_state_hash,
}
for key, v in sealed.items()
},
)
static_rows, dynamic_rows, prediction_rows = [], [], []
post_effects = []
evaluate_started = time.perf_counter()
for s in range(3):
for q in roles.sealed_query:
camera = ledger.read_query_camera(f"{s}/{q}", lambda q=q: fixture_camera(q))
truth = ledger.read_query_ground_truth(f"{s}/{q}", lambda s=s, q=q: fixture(s, q)[1:])
rgb, depth, points = truth
common = torch.ones_like(depth, dtype=torch.bool)
for ctx in [roles.context_a, roles.context_b]:
visible = torch.zeros_like(common)
for frame in ctx:
_, _, valid = project_world(
points.reshape(-1, 3), arrived_observation(s, frame).camera
)
visible |= valid.reshape_as(common)
common &= visible
query_camera = batched_camera(camera)
camera_hash = hash_value(query_camera)
for method in ["A", "B", "anchor", "prior", "wrong_scene"]:
if method == "prior":
pred = {
"rgb": torch.full((1, 1, 3, *depth.shape), 0.5),
"depth": torch.full((1, 1, 1, *depth.shape), 5.0),
"visibility": torch.ones(1, 1, 1, *depth.shape),
}
else:
key = f"{(s + 1) % 3}/A" if method == "wrong_scene" else f"{s}/{method}"
state = sealed[key]
assert hash_scene_state(state.scene_state) == state.state_hash
pred = renderer(state.scene_state, query_camera, ("rgb", "depth", "visibility"))
assert hash_scene_state(state.scene_state) == state.state_hash
assert camera_hash == hash_value(query_camera)
metrics = measurement_metrics(
pred["rgb"][0, 0],
rgb,
pred["depth"][0, 0, 0],
depth,
pred["visibility"][0, 0, 0],
)
region_metrics = (
measurement_metrics(
pred["rgb"][0, 0],
rgb,
pred["depth"][0, 0, 0],
depth,
pred["visibility"][0, 0, 0],
region_mask=common.numpy(),
)
if common.any()
else None
)
static_rows.append(
dict(
scene_id=f"synthetic-{s}",
query_id=q,
method=method,
**metrics,
common_visible=region_metrics,
query_camera_hash=camera_hash,
)
)
post = {}
for h in ["FC", "CF"]:
for action in ["OFF", "FUSE", "COMPLETE", "ALL"]:
state = sealed[f"{s}/{h}/{action}"]
assert hash_scene_state(state.scene_state) == state.state_hash
pred = renderer(state.scene_state, query_camera, ("rgb", "depth", "visibility"))
post[h, action] = pred
metrics = measurement_metrics(
pred["rgb"][0, 0],
rgb,
pred["depth"][0, 0, 0],
depth,
pred["visibility"][0, 0, 0],
)
dynamic_rows.append(
dict(
scene_id=f"synthetic-{s}",
continuation_id=7,
query_id=q,
history=h,
action=action,
**metrics,
)
)
assert hash_scene_state(state.scene_state) == state.state_hash
prediction_rows.append(
{
"scene_id": s,
"query": q,
"history": h,
"action": action,
"rgb_hash": hash_value(pred["rgb"]),
"depth_hash": hash_value(pred["depth"]),
}
)
for action in ["OFF", "FUSE", "COMPLETE", "ALL"]:
effect = {"scene_id": f"synthetic-{s}", "query_id": q, "action": action}
for channel in ["rgb", "depth"]:
diff = post["FC", action][channel] - post["CF", action][channel]
effect[channel + "_rms"] = float(diff.square().mean().sqrt())
effect[channel + "_maxabs"] = float(diff.abs().max())
post_effects.append(effect)
write(output / "post_prediction_history_effect.json", post_effects)
analysis = analyze_history(dynamic_rows, draws=10000, seed=20260927, tie_tolerance=1e-8)
write(output / "controlled_history_results.json", dynamic_rows)
write(output / "static_state_results.json", static_rows)
write(output / "history_traces.json", history_raw)
write(output / "prediction_hashes.json", prediction_rows)
write(output / "numerical_history_effect.json", numerical)
write(output / "history_analysis.json", analysis)
metrics = [
"rgb_psnr",
"rgb_ssim",
"depth_absrel",
"depth_rmse",
"depth_delta1",
"opacity",
"coverage",
"valid_depth_fraction",
]
static_analysis = {
method: {
m: scene_macro([r for r in static_rows if r["method"] == method], m) for m in metrics
}
for method in ["A", "B", "anchor", "prior", "wrong_scene"]
}
static_analysis["R-Residual"] = {"status": "NOT_RUN_UNTRAINED_HEAD_NOT_FAIR_COMPARISON"}
static_analysis["depth_absrel_deltas"] = {
label: static_analysis[left]["depth_absrel"]["mean"]
- static_analysis[right]["depth_absrel"]["mean"]
for label, left, right in [
("context_B_minus_A", "B", "A"),
("anchor_minus_A", "anchor", "A"),
("wrong_scene_minus_A", "wrong_scene", "A"),
]
}
write(output / "static_analysis.json", static_analysis)
write(
output / "access_log.json",
{
"scope": "SYNTHETIC_ONLY",
"events": ledger.events,
"formal_reads": 0,
"note": "Synthetic labels only; no depth enters runtime.",
},
)
evaluation_seconds = time.perf_counter() - evaluate_started
# Pairwise tensor storage independence, including candidate caches/fast/state.
all_branches = list(branches.values())
ptrs = [b.fast.delta_fuse.data_ptr() for b in all_branches]
isolation = len(set(ptrs)) == len(ptrs)
assert isolation
write(
output / "summary.json",
{
"scope": "SYNTHETIC_ENGINEERING_ONLY_NOT_SCIENTIFIC_EVIDENCE",
"synthetic_scenes": 3,
"scientific_scenes": 0,
"controlled_pairs": 3,
"dynamic_query_rows": len(dynamic_rows),
"static_query_rows": len(static_rows),
"candidate_isolation_verified": isolation,
"all_states_sealed_before_query_reads": True,
"carrier_parameters": sum(p.numel() for p in carrier.parameters()),
"writer_parameters": sum(p.numel() for p in writer.parameters()),
"fixed_renderer_parameters": sum(p.numel() for p in renderer.parameters()),
"construction_seconds": construction_seconds,
"evaluation_seconds": evaluation_seconds,
"total_seconds": time.perf_counter() - started,
"device": "cpu",
"threads": 1,
"candidate_branch_steps": 24,
"matched_history_steps": 12,
"bidirectional_matching_calls": 0,
"policy_training_runs": 0,
},
)
print(json.dumps(json.loads((output / "summary.json").read_text()), indent=2))
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--output", type=Path, default=ROOT / "outputs" / NAME / "smoke")
run(parser.parse_args().output)