Add policy set command and live runtime defaults

Author: BB-84C

README.md

@@ -9,7 +9,7 @@ Simulator-first Python bridge between Nanonis SPM controller interfaces and QCod
 - Strict write semantics:
   - `set` is always a guarded single-step write.
   - `ramp` is always an explicit multi-step trajectory.
-- Safety-first defaults (`allow_writes=false`, `dry_run=true`).
+- Default runtime policy (`allow_writes=true`, `dry_run=false`).
 
 ## v1 API support contract
 
@@ -298,10 +298,11 @@ nqctl observables list
 nqctl actions list
 ```
 
-Inspect active runtime policy:
+Inspect and update runtime policy:
 
 ```powershell
 nqctl policy show
+nqctl policy set --allow-writes true --dry-run false
 ```
 
 ### Execute operations

nanonis_qcodes_controller/cli.py

@@ -6,6 +6,7 @@
 import inspect
 import json
 import math
+import os
 import re
 import sqlite3
 import sys
@@ -17,6 +18,8 @@
 from pathlib import Path
 from typing import Any
 
+import yaml
+
 from nanonis_qcodes_controller.client import create_client, probe_host_ports, report_to_dict
 from nanonis_qcodes_controller.client.errors import (
     NanonisCommandUnavailableError,
@@ -57,6 +60,9 @@
     re.IGNORECASE,
 )
 
+_TRUE_BOOL_TOKENS = frozenset({"1", "true", "yes", "on"})
+_FALSE_BOOL_TOKENS = frozenset({"0", "false", "no", "off"})
+
 
 @dataclass(frozen=True)
 class ActionDescriptor:
@@ -365,6 +371,23 @@ def _build_parser() -> argparse.ArgumentParser:
     parser_policy_show.add_argument("--config-file")
     parser_policy_show.set_defaults(handler=_cmd_policy_show)
 
+    parser_policy_set = policy_subparsers.add_parser(
+        "set", help="Set runtime policy write/dry-run flags."
+    )
+    _add_json_arg(parser_policy_set)
+    parser_policy_set.add_argument(
+        "--allow-writes",
+        type=_parse_cli_bool_arg,
+        help="Override safety.allow_writes (1/0, true/false, yes/no, on/off).",
+    )
+    parser_policy_set.add_argument(
+        "--dry-run",
+        type=_parse_cli_bool_arg,
+        help="Override safety.dry_run (1/0, true/false, yes/no, on/off).",
+    )
+    parser_policy_set.add_argument("--config-file")
+    parser_policy_set.set_defaults(handler=_cmd_policy_set)
+
     parser_trajectory = subparsers.add_parser("trajectory", help="Trajectory log utilities.")
     trajectory_subparsers = parser_trajectory.add_subparsers(
         dest="trajectory_command", required=True
@@ -826,6 +849,44 @@ def _cmd_policy_show(args: argparse.Namespace) -> int:
     return EXIT_OK
 
 
+def _cmd_policy_set(args: argparse.Namespace) -> int:
+    allow_writes = args.allow_writes
+    dry_run = args.dry_run
+    if allow_writes is None and dry_run is None:
+        raise ValueError("Provide at least one policy flag: --allow-writes and/or --dry-run.")
+
+    config_path = _resolve_policy_config_path(args.config_file)
+    config_values = _load_runtime_config_yaml(config_path)
+    safety_section = config_values.get("safety")
+    if safety_section is None:
+        updated_safety: dict[str, Any] = {}
+    elif isinstance(safety_section, Mapping):
+        updated_safety = dict(safety_section)
+    else:
+        raise ValueError("Runtime config section 'safety' must be a mapping.")
+
+    if allow_writes is not None:
+        updated_safety["allow_writes"] = bool(allow_writes)
+    if dry_run is not None:
+        updated_safety["dry_run"] = bool(dry_run)
+    config_values["safety"] = updated_safety
+
+    config_path.parent.mkdir(parents=True, exist_ok=True)
+    with config_path.open("w", encoding="utf-8") as handle:
+        yaml.safe_dump(config_values, handle, sort_keys=False)
+
+    settings = load_settings(config_file=config_path)
+    payload = {
+        "allow_writes": settings.safety.allow_writes,
+        "dry_run": settings.safety.dry_run,
+        "default_ramp_interval_s": settings.safety.default_ramp_interval_s,
+        "config_file": str(config_path),
+        "timestamp_utc": _now_utc_iso(),
+    }
+    _print_payload(payload, as_json=args.json)
+    return EXIT_OK
+
+
 def _cmd_parameters_discover(args: argparse.Namespace) -> int:
     commands = list(_discover_nanonis_spm_commands(args.match))
     if args.limit > 0:
@@ -1428,6 +1489,42 @@ def _parse_float_arg(*, name: str, raw_value: str) -> float:
         raise ValueError(f"{name} must be numeric.") from exc
 
 
+def _parse_cli_bool_arg(raw_value: object) -> bool:
+    if isinstance(raw_value, bool):
+        return raw_value
+
+    normalized = str(raw_value).strip().lower()
+    if normalized in _TRUE_BOOL_TOKENS:
+        return True
+    if normalized in _FALSE_BOOL_TOKENS:
+        return False
+
+    raise argparse.ArgumentTypeError("Expected boolean value: 1/0, true/false, yes/no, on/off.")
+
+
+def _resolve_policy_config_path(config_file: str | None) -> Path:
+    if config_file is not None:
+        return Path(config_file).expanduser()
+    env_override = os.environ.get("NANONIS_CONFIG_FILE")
+    if env_override:
+        return Path(env_override).expanduser()
+    return Path("config/default_runtime.yaml")
+
+
+def _load_runtime_config_yaml(config_path: Path) -> dict[str, Any]:
+    if not config_path.exists():
+        return {}
+
+    with config_path.open("r", encoding="utf-8") as handle:
+        loaded = yaml.safe_load(handle)
+
+    if loaded is None:
+        return {}
+    if not isinstance(loaded, Mapping):
+        raise ValueError("Runtime config file must contain a top-level mapping.")
+    return dict(loaded)
+
+
 def _parse_positive_float_arg(*, name: str, raw_value: str) -> float:
     value = _parse_float_arg(name=name, raw_value=raw_value)
     if value <= 0:

nanonis_qcodes_controller/config/settings.py

@@ -28,8 +28,8 @@ class NanonisConnectionSettings:
 
 @dataclass(frozen=True)
 class SafetySettings:
-    allow_writes: bool = False
-    dry_run: bool = True
+    allow_writes: bool = True
+    dry_run: bool = False
     default_ramp_interval_s: float = DEFAULT_RAMP_INTERVAL_S
 
 

nanonis_qcodes_controller/resources/config/default_runtime.yaml

@@ -11,8 +11,8 @@ nanonis:
   backend: "adapter"
 
 safety:
-  allow_writes: false
-  dry_run: true
+  allow_writes: true
+  dry_run: false
   default_ramp_interval_s: 0.05
 
 trajectory:

tests/test_cli.py

@@ -59,6 +59,21 @@ def test_showall_parser_available() -> None:
     assert args.command == "showall"
 
 
+def test_policy_set_parser_accepts_boolean_overrides() -> None:
+    parser = cli._build_parser()
+    args = parser.parse_args(["policy", "set", "--allow-writes", "true", "--dry-run", "false"])
+    assert args.command == "policy"
+    assert args.policy_command == "set"
+    assert args.allow_writes is True
+    assert args.dry_run is False
+
+
+def test_policy_set_parser_rejects_invalid_boolean_token() -> None:
+    parser = cli._build_parser()
+    with pytest.raises(SystemExit):
+        _ = parser.parse_args(["policy", "set", "--allow-writes", "maybe"])
+
+
 def test_act_parser_supports_repeatable_arg_flags() -> None:
     parser = cli._build_parser()
     args = parser.parse_args(
@@ -210,6 +225,52 @@ def fake_instrument_context(*_args, **_kwargs):
     assert captured_step_values == pytest.approx([1e-11, 1e-11])
 
 
+def test_cmd_policy_set_updates_config_and_emits_effective_policy(monkeypatch, tmp_path) -> None:
+    captured_payloads: list[dict[str, object]] = []
+
+    def fake_print_payload(payload, *, as_json: bool) -> None:
+        del as_json
+        captured_payloads.append(dict(payload))
+
+    monkeypatch.setattr(cli, "_print_payload", fake_print_payload)
+
+    config_file = tmp_path / "runtime.yaml"
+    config_file.write_text(
+        "\n".join(
+            (
+                "nanonis:",
+                '  host: "127.0.0.1"',
+                "  ports: [3364]",
+                "  timeout_s: 2.0",
+                "  retry_count: 1",
+                '  backend: "adapter"',
+                "safety:",
+                "  allow_writes: false",
+                "  dry_run: true",
+                "  default_ramp_interval_s: 0.05",
+                "trajectory:",
+                "  enabled: false",
+                '  directory: "artifacts/trajectory"',
+                "  queue_size: 2048",
+                "  max_events_per_file: 5000",
+            )
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+
+    args = argparse.Namespace(
+        allow_writes=True, dry_run=False, config_file=str(config_file), json=True
+    )
+    exit_code = getattr(cli, "_cmd_policy_set")(args)
+
+    assert exit_code == cli.EXIT_OK
+    assert captured_payloads
+    payload = captured_payloads[-1]
+    assert payload["allow_writes"] is True
+    assert payload["dry_run"] is False
+
+
 def test_parse_action_args_rejects_invalid_entries() -> None:
     with pytest.raises(ValueError, match="key=value"):
         _ = cli._parse_action_args(raw_args=("Scan_action",))

tests/test_default_file_resolution.py

@@ -70,8 +70,14 @@ def _tracking_resolver(name: str):
     settings = load_settings()
 
     assert calls == ["default_runtime.yaml"]
-    assert settings.safety.allow_writes is False
-    assert settings.safety.dry_run is True
+    assert settings.safety.allow_writes is True
+    assert settings.safety.dry_run is False
+
+
+def test_safety_settings_defaults_are_live_write_mode() -> None:
+    defaults = settings_module.SafetySettings()
+    assert defaults.allow_writes is True
+    assert defaults.dry_run is False
 
 
 def test_load_parameter_specs_falls_back_when_default_path_missing(tmp_path, monkeypatch) -> None: