fix(machine-config): isolate capability reads from sibling settings

loopx/agent_onboarding.py

@@ -441,10 +441,9 @@ def build_agent_onboarding_packet(
         {},
     )
     from .capabilities.machine_configuration.builtins import (
-        build_builtin_machine_configuration_registry,
+        read_builtin_goal_machine_configuration,
         project_goal_with_builtin_machine_configuration,
     )
-    from .capabilities.machine_configuration.store import read_machine_configuration
     from .paths import resolve_runtime_root
 
     runtime_root = resolve_runtime_root(
@@ -453,10 +452,7 @@ def build_agent_onboarding_packet(
     )
     goal = project_goal_with_builtin_machine_configuration(
         goal,
-        read_machine_configuration(
-            runtime_root,
-            registry=build_builtin_machine_configuration_registry(),
-        ),
+        read_builtin_goal_machine_configuration(runtime_root),
     )
     active_project_skill_ids = (
         [CHANGE_QUALITY_SKILL_ID]

loopx/capabilities/change_quality/receipt.py

@@ -93,15 +93,11 @@ def _goal_from_registry(
         raise ValueError(f"goal_id not found in registry: {goal_id}")
     if runtime_root is not None:
         from ..machine_configuration.builtins import (
-            build_builtin_machine_configuration_registry,
+            read_builtin_goal_machine_configuration,
             project_goal_with_builtin_machine_configuration,
         )
-        from ..machine_configuration.store import read_machine_configuration
 
-        machine_configuration = read_machine_configuration(
-            runtime_root,
-            registry=build_builtin_machine_configuration_registry(),
-        )
+        machine_configuration = read_builtin_goal_machine_configuration(runtime_root)
         return project_goal_with_builtin_machine_configuration(
             goal, machine_configuration
         )

loopx/capabilities/explore/machine_defaults.py

@@ -104,4 +104,5 @@ def resolve_explore_goal(goal: Mapping[str, Any], runtime_root: Path) -> dict[st
 
     return apply_explore_machine_default(goal, read_machine_configuration(
         runtime_root, registry=build_builtin_machine_configuration_registry(),
+        namespace_ids=("explore_harness",),
     ))

loopx/capabilities/machine_configuration/builtins.py

@@ -2,6 +2,7 @@
 
 from collections.abc import Callable, Mapping
 from importlib import import_module
+from pathlib import Path
 from typing import Any
 
 from .contract import MachineConfigurationNamespace, MachineConfigurationRegistry
@@ -80,6 +81,16 @@ def build_builtin_machine_configuration_registry() -> MachineConfigurationRegist
     return registry
 
 
+def read_builtin_goal_machine_configuration(runtime_root: Path) -> dict[str, Any] | None:
+    """Read the owners composed by project_goal_with_builtin_machine_configuration."""
+    from .store import read_machine_configuration
+
+    return read_machine_configuration(
+        runtime_root, registry=build_builtin_machine_configuration_registry(),
+        namespace_ids=("explore_harness", "todo_replan_cadence", "change_quality_qualification"),
+    )
+
+
 def project_goal_with_builtin_machine_configuration(
     goal: Mapping[str, Any],
     machine_configuration: Mapping[str, Any] | None,
@@ -135,6 +146,7 @@ def builtin_machine_inheritable_goal_overrides(
 
 __all__ = [
     "build_builtin_machine_configuration_registry",
+    "read_builtin_goal_machine_configuration",
     "builtin_machine_inheritable_goal_overrides",
     "project_goal_with_builtin_machine_configuration",
 ]

loopx/capabilities/machine_configuration/contract.py

@@ -217,8 +217,15 @@ def _mapping(value: object, label: str) -> dict[str, Any]:
 
 
 def normalize_machine_configuration(
-    raw: object, *, registry: MachineConfigurationRegistry
+    raw: object, *, registry: MachineConfigurationRegistry,
+    namespace_ids: tuple[str, ...] | None = None,
 ) -> dict[str, Any]:
+    """Validate a whole document, or its explicitly selected read-only owners.
+
+    Scoped readers still validate the envelope and every selected owner. They
+    neither interpret nor certify sibling settings. Mutation and full inspection
+    keep the default whole-document validation.
+    """
     payload = _mapping(raw, "machine_configuration")
     unknown = sorted(set(payload) - {"schema_version", "namespaces"})
     if unknown:
@@ -232,8 +239,15 @@ def normalize_machine_configuration(
     namespaces = _mapping(payload.get("namespaces"), "machine_configuration.namespaces")
     if not namespaces:
         raise ValueError("machine_configuration.namespaces must not be empty")
+    selected = set(namespaces)
+    if namespace_ids is not None:
+        if not namespace_ids:
+            raise ValueError("machine-configuration read scope must not be empty")
+        for namespace in namespace_ids:
+            registry.resolve(namespace)
+        selected.intersection_update(namespace_ids)
     normalized: dict[str, Any] = {}
-    for namespace in sorted(namespaces):
+    for namespace in sorted(selected):
         contract = registry.resolve(namespace)
         candidate = _mapping(
             namespaces[namespace], f"machine_configuration.namespaces.{namespace}"

loopx/capabilities/machine_configuration/goal_storage.py

@@ -46,7 +46,9 @@ def new_goal_storage_target(runtime_root: Path) -> dict[str, Any]:
     from .builtins import build_builtin_machine_configuration_registry
     from .store import read_machine_configuration
     registry = build_builtin_machine_configuration_registry()
-    configuration = read_machine_configuration(runtime_root, registry=registry)
+    configuration = read_machine_configuration(
+        runtime_root, registry=registry, namespace_ids=("goal_storage",),
+    )
     raw = (configuration or {}).get("namespaces", {}).get("goal_storage")
     if raw is None:
         # Reuse the registered configuration owner also projected by Settings.

loopx/capabilities/machine_configuration/store.py

@@ -103,15 +103,16 @@ def _secure_write(
 
 
 def read_machine_configuration(
-    runtime_root: Path, *, registry: MachineConfigurationRegistry
+    runtime_root: Path, *, registry: MachineConfigurationRegistry,
+    namespace_ids: tuple[str, ...] | None = None,
 ) -> dict[str, Any] | None:
     path = machine_configuration_store_path(runtime_root)
     if not path.is_file():
         if path.exists() or path.is_symlink():
             raise ValueError("machine configuration must be a readable file")
         return None
     normalized: dict[str, Any] = normalize_machine_configuration(
-        read_json(path), registry=registry
+        read_json(path), registry=registry, namespace_ids=namespace_ids,
     )
     return normalized
 

loopx/capabilities/periodic_report/machine_store.py

@@ -29,7 +29,8 @@ def machine_defaults_store_path(runtime_root: Path) -> Path:
 
 def read_periodic_report_machine_defaults(runtime_root: Path) -> dict[str, Any] | None:
     return read_machine_configuration(
-        runtime_root, registry=build_builtin_machine_configuration_registry()
+        runtime_root, registry=build_builtin_machine_configuration_registry(),
+        namespace_ids=("periodic_report",),
     )
 
 

loopx/capabilities/todo_replan_cadence/machine_defaults.py

@@ -211,6 +211,7 @@ def resolve_todo_replan_cadence_goal(
     machine = (
         read_machine_configuration(
             runtime_root, registry=build_builtin_machine_configuration_registry(),
+            namespace_ids=("todo_replan_cadence",),
         ) if runtime_root is not None else None
     )
     return apply_todo_replan_cadence_machine_default(goal, machine)

loopx/cli_commands/context_machine.py

@@ -99,7 +99,9 @@ def apply_context_machine_options(
         from ..capabilities.machine_configuration.store import read_machine_configuration
 
         machine_registry = build_builtin_machine_configuration_registry()
-        configuration = read_machine_configuration(runtime_root, registry=machine_registry)
+        configuration = read_machine_configuration(
+            runtime_root, registry=machine_registry, namespace_ids=("context_delivery",),
+        )
         value = (configuration or {}).get("namespaces", {}).get("context_delivery")
         using_default = value is None
         if using_default:

loopx/cli_commands/pr_review.py

@@ -255,7 +255,10 @@ def handle_pr_review_command(
         target_exact_heads=target_exact_heads,
     )
     try:
-        machine_configuration = (read_machine_configuration(runtime_root, registry=build_builtin_machine_configuration_registry()) if runtime_root is not None else None)
+        machine_configuration = (read_machine_configuration(
+            runtime_root, registry=build_builtin_machine_configuration_registry(),
+            namespace_ids=("pull_request_review",),
+        ) if runtime_root is not None else None)
         goal = None
         goal_id = getattr(args, "goal_id", None)
         if goal_id:

loopx/history.py

@@ -407,18 +407,14 @@ def collect_history(
     registry: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     from .capabilities.machine_configuration.builtins import (
-        build_builtin_machine_configuration_registry,
+        read_builtin_goal_machine_configuration,
         project_goal_with_builtin_machine_configuration,
     )
-    from .capabilities.machine_configuration.store import read_machine_configuration
     from .quota import goal_quota_with_spend_ledger
 
     if registry is None:
         registry = load_registry(registry_path)
-    machine_configuration = read_machine_configuration(
-        runtime_root,
-        registry=build_builtin_machine_configuration_registry(),
-    )
+    machine_configuration = read_builtin_goal_machine_configuration(runtime_root)
     registry_goal_rows = registry_goals(registry)
     registry_goal_digests = {
         str(goal.get("id")): goal_registry_digest(goal)

tests/capabilities/test_machine_configuration_contract.py

@@ -106,6 +106,59 @@ def test_unknown_namespace_and_unknown_envelope_field_fail_closed() -> None:
         normalize_machine_configuration(config, registry=_registry())
 
 
+def test_read_scope_uses_only_its_registered_owner_and_leaves_siblings_opaque(tmp_path):
+    import json
+    from loopx.capabilities.machine_configuration.store import (
+        configure_machine_configuration, read_machine_configuration,
+    )
+
+    config = _config()
+    config["namespaces"]["future_setting"] = {"schema_version": "unrecognized_v1"}
+    path = tmp_path / "machine/configuration.json"
+    path.parent.mkdir()
+    path.write_text(json.dumps(config))
+    before = path.read_bytes()
+    selected = read_machine_configuration(tmp_path, registry=_registry(), namespace_ids=("example",))
+    assert selected == _config()
+    assert path.read_bytes() == before
+    with pytest.raises(ValueError, match="unsupported machine-configuration namespace"):
+        read_machine_configuration(tmp_path, registry=_registry())
+    with pytest.raises(ValueError, match="unsupported machine-configuration namespace"):
+        configure_machine_configuration(runtime_root=tmp_path, configuration=config, registry=_registry())
+    assert path.read_bytes() == before
+
+
+@pytest.mark.parametrize("field,value,error", [
+    ("schema_version", "example_v1", ValueError),
+    ("enabled", "false", TypeError),
+    ("surprise", True, ValueError),
+])
+def test_scoped_read_still_rejects_an_invalid_selected_owner(field, value, error):
+    config = _config()
+    config["namespaces"]["example"][field] = value
+    with pytest.raises(error):
+        normalize_machine_configuration(config, registry=_registry(), namespace_ids=("example",))
+
+
+def test_known_absent_read_owner_is_absent_without_certifying_an_unknown_sibling():
+    config = {"schema_version": "loopx_machine_configuration_v0", "namespaces": {"future": None}}
+    assert normalize_machine_configuration(config, registry=_registry(), namespace_ids=("example",)) == {
+        "schema_version": "loopx_machine_configuration_v0", "namespaces": {},
+    }
+    for scope in ((), ("unregistered_owner",)):
+        with pytest.raises(ValueError):
+            normalize_machine_configuration(config, registry=_registry(), namespace_ids=scope)
+
+
+@pytest.mark.parametrize("change", [
+    {"arbitrary": True}, {"schema_version": "unrecognized_v1"},
+    {"namespaces": []}, {"namespaces": {}},
+])
+def test_read_scope_cannot_hide_a_malformed_machine_envelope(change):
+    with pytest.raises((ValueError, TypeError)):
+        normalize_machine_configuration({**_config(), **change}, registry=_registry(), namespace_ids=("example",))
+
+
 def test_consumer_schema_version_and_fields_are_enforced() -> None:
     config = _config()
     config["namespaces"]["example"]["schema_version"] = "example_v1"

tests/capabilities/test_machine_configuration_goal_defaults.py

@@ -156,6 +156,28 @@ def test_machine_defaults_reach_real_history_and_change_quality_paths(
     assert prepared["policy"]["enabled"] is True
 
 
+def test_goal_readers_keep_their_defaults_when_an_unrelated_owner_is_incompatible(tmp_path):
+    repo, registry_path, runtime_root = _fixture(tmp_path)
+    path = runtime_root / "machine/configuration.json"
+    config = json.loads(path.read_text())
+    config["namespaces"]["steward_executor"] = {
+        "schema_version": "steward_executor_machine_defaults_v1",
+        "executor_endpoint": "future_executor",
+    }
+    path.write_text(json.dumps(config))
+    original = path.read_bytes()
+    goal = _history_goal(registry_path, runtime_root)
+    assert goal["execution_profile"]["replan_after_completed_todos"] == 2
+    (repo / "app.py").write_text("value = 2\n", encoding="utf-8")
+    prepared = build_change_quality_prepare_packet(
+        registry_path=registry_path, runtime_root=runtime_root,
+        goal_id=GOAL_ID, repo_path=repo, base_ref="HEAD",
+    )
+    assert prepared["status"] == "review_required"
+    assert prepared["policy"]["enabled"] is True
+    assert path.read_bytes() == original
+
+
 @pytest.mark.parametrize("update", [{}, {"change_quality_enabled": False}, {"execution_turn_granularity": "fine"}])
 def test_unrelated_configuration_preserves_explicit_default_cadence(tmp_path: Path, update) -> None:
     _repo, registry_path, runtime_root = _fixture(tmp_path)

tests/capabilities/test_pr_review_configuration.py

@@ -55,6 +55,42 @@ def test_ci_configuration_rejects_non_boolean_input(invalid) -> None:
         apply_change({}, {"wait_for_ci": invalid}, clear=False)
 
 
+def test_review_reads_its_own_machine_policy_without_reconfiguring_siblings(tmp_path, capsys):
+    runtime = tmp_path / "runtime"
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"goals": []}))
+    config = runtime / "machine/configuration.json"
+    config.parent.mkdir(parents=True)
+    config.write_text(json.dumps({
+        "schema_version": "loopx_machine_configuration_v0",
+        "namespaces": {
+            "pull_request_review": {
+                "schema_version": "pull_request_review_machine_defaults_v0",
+                "wait_for_ci": False,
+            },
+            "steward_executor": {
+                "schema_version": "steward_executor_machine_defaults_v1",
+                "executor_endpoint": "future_executor",
+            },
+        },
+    }))
+    original = config.read_bytes()
+    fixtures = runpy.run_path(str(Path(__file__).parents[1] / "test_pr_review_github_scan.py"))
+    pr = fixtures["_merge_ready_pr"]()
+    pr["statusCheckRollup"] = [{"name": "test", "status": "IN_PROGRESS"}]
+    pr["review_thread_summary"] = {"complete": True, "total_count": 0, "unresolved_count": 0}
+    fixture = tmp_path / "prs.json"
+    fixture.write_text(json.dumps({"repository": "owner/repo", "pull_requests": [pr]}))
+    code = main([
+        "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", "pr-review",
+        "--fixture", str(fixture),
+    ])
+    packet = json.loads(capsys.readouterr().out)
+    assert code == 0, packet
+    assert packet["request"]["review_configuration"]["wait_for_ci"] is False
+    assert config.read_bytes() == original
+
+
 def test_goal_editor_save_clear_and_cli_observe_the_same_policy(
     tmp_path, capsys
 ) -> None:

tests/control_plane/test_context_machine_configuration.py

@@ -82,6 +82,33 @@ def test_machine_catalog_discovers_context_delivery_without_activation(tmp_path)
     assert not (runtime / "context-delivery").exists()
 
 
+@pytest.mark.parametrize("mode", ["on", "off"])
+def test_context_mode_does_not_depend_on_an_unrelated_executor(tmp_path, mode):
+    registry, runtime = tmp_path / "absent.json", tmp_path / "runtime"
+    path = runtime / "machine/configuration.json"
+    path.parent.mkdir(parents=True)
+    path.write_text(json.dumps({"schema_version": "loopx_machine_configuration_v0", "namespaces": {
+        "context_delivery": {"schema_version": CONTEXT_DELIVERY_MACHINE_SCHEMA, "mode": mode},
+        "steward_executor": {"schema_version": "steward_executor_machine_defaults_v1",
+                             "executor_endpoint": "future_executor"},
+    }}))
+    original = path.read_bytes()
+    code, result = _run_cli(registry, runtime, "quota", "context-config", "--codex-app",
+        "--goal-id", "fixture", "--agent-id", "worker")
+    if mode == "on" and os.name != "posix":
+        assert code != 0 and "unsupported" in result["reason"], result
+    else:
+        assert code == 0, result
+        assert result["context_delivery"]["enabled"] is (mode == "on"), result
+    assert path.read_bytes() == original
+    assert not registry.exists()
+    assert not (runtime / "goals").exists()
+    if mode == "off":
+        assert not (runtime / "context-delivery").exists()
+    elif os.name == "posix":
+        assert (runtime / "context-delivery").stat().st_mode & 0o077 == 0
+
+
 @pytest.mark.parametrize("invalid", [True, False, None, 1, "auto", "codex_app_heartbeat"])
 def test_invalid_machine_mode_never_writes(tmp_path, invalid):
     registry, runtime = tmp_path / "absent.json", tmp_path / "runtime"