fix: don't rollback/unlock a candidate this session never locked

srxsync/orchestrator.py

@@ -90,10 +90,13 @@ async def check_one(target: Target) -> DriftLine:
     def _fetch_source(self) -> etree._Element:
         t = self._tx()
         secret = get_secret(host=self._inv.source.host, auth=self._inv.source.auth)
-        t.connect(
-            self._inv.source.host, secret.username, secret.password, ssh_key=secret.ssh_key_path
-        )
         try:
+            t.connect(
+                self._inv.source.host,
+                secret.username,
+                secret.password,
+                ssh_key=secret.ssh_key_path,
+            )
             return t.fetch(self._union_paths)
         finally:
             t.close()

srxsync/transport/pyez.py

@@ -20,6 +20,7 @@ class PyEZTransport(Transport):
     def __init__(self) -> None:
         self._dev: Device | None = None
         self._cfg: Config | None = None
+        self._locked: bool = False
 
     def connect(
         self,
@@ -41,6 +42,7 @@ def connect(
             self._dev.open()
             self._cfg = Config(self._dev, mode="exclusive")
             self._cfg.lock()
+            self._locked = True
         except (ConnectError, RpcError) as e:
             raise TransportError(f"connect failed for {host}: {_reason(e)}") from e
 
@@ -86,16 +88,17 @@ def confirm(self) -> None:
             raise TransportError(f"confirm commit failed: {_reason(e)}") from e
 
     def rollback(self) -> None:
-        if self._cfg is None:
+        if self._cfg is None or not self._locked:
             return
         with contextlib.suppress(RpcError):
             self._cfg.rollback()
 
     def close(self) -> None:
-        if self._cfg is not None:
+        if self._cfg is not None and self._locked:
             with contextlib.suppress(RpcError):
                 self._cfg.unlock()
-            self._cfg = None
+            self._locked = False
+        self._cfg = None
         if self._dev is not None:
             with contextlib.suppress(Exception):
                 self._dev.close()

srxsync/transport/rustez.py

@@ -124,7 +124,7 @@ def confirm(self) -> None:
             raise TransportError(f"confirm commit failed on {self._host}: {_reason(exc)}") from exc
 
     def rollback(self) -> None:
-        if self._cfg is None:
+        if self._cfg is None or not self._locked:
             return  # nothing to roll back; match PyEZ behavior
         with contextlib.suppress(Exception):
             self._cfg.rollback(0)

tests/unit/test_orchestrator.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 from typing import ClassVar, Literal
 
 from lxml import etree
@@ -164,6 +165,25 @@ def test_check_mixed_failures_report_every_target(monkeypatch):
     assert by_host[HOST_OK].in_sync is True
 
 
+HOST_SOURCE = "198.51.100.1"
+
+
+def test_fetch_source_closes_session_when_lock_fails(monkeypatch):
+    """connect() on the source device was called outside _fetch_source's
+    try/finally. If lock() raised there, close() never ran and the NETCONF
+    session to the source stayed open until GC or process exit."""
+    _FakeTransport.reset(connect_failures={HOST_SOURCE: _LockError("lock held")})
+    _patch_get_secret(monkeypatch, secret_failures=set())
+
+    inv = _inventory([HOST_OK])
+    orch = Orchestrator(inventory=inv, categories=_categories(), transport_factory=_FakeTransport)
+
+    with contextlib.suppress(_LockError):
+        orch._fetch_source()
+
+    assert HOST_SOURCE in _FakeTransport.closes
+
+
 def test_gather_return_exceptions_defends_against_an_escaped_error(monkeypatch):
     """Even if some future failure mode slips past the per-target except
     clause entirely, the batch must not abort — every target still gets a

tests/unit/test_transport_pyez.py

@@ -93,3 +93,30 @@ def test_connect_never_echoes_device_lock_error_body(monkeypatch):
     assert "alice" not in message
     assert "198.51.100.7" not in message
     assert "LockError" in message
+
+
+def test_rollback_and_close_are_noops_after_failed_lock(monkeypatch):
+    """If Config.lock() raises during connect(), we never held the lock.
+    rollback() must not send <rollback>, and close() must not send <unlock>,
+    for a candidate another session may still be editing — otherwise a
+    failed lock (e.g. because another operator already has uncommitted
+    changes) turns into us discarding their work."""
+    rsp = etree.fromstring(_LOCK_RPC_ERROR_XML)
+    lock_err = LockError(rsp)
+
+    fake_device = MagicMock()
+    fake_device.open.return_value = None
+    monkeypatch.setattr("srxsync.transport.pyez.Device", lambda **kw: fake_device)
+    fake_config = MagicMock()
+    fake_config.lock.side_effect = lock_err
+    monkeypatch.setattr("srxsync.transport.pyez.Config", lambda dev, mode: fake_config)
+
+    t = PyEZTransport()
+    with pytest.raises(TransportError):
+        t.connect("198.51.100.1", "svc", password="FAKEPASS")
+
+    t.rollback()
+    t.close()
+
+    fake_config.rollback.assert_not_called()
+    fake_config.unlock.assert_not_called()

tests/unit/test_transport_rustez.py

@@ -155,3 +155,38 @@ def test_load_and_fetch_errors_are_sanitized(rustez_module: types.ModuleType) ->
     with pytest.raises(TransportError) as info:
         tx.load(etree.Element("configuration"), "merge")
     assert SECRET_TEXT not in str(info.value)
+
+
+def test_rollback_and_close_are_noops_after_failed_lock(
+    rustez_module: types.ModuleType, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """If Config.lock() raises during connect(), we never held the lock.
+    rollback() must not send <rollback>, and close() must not send <unlock>,
+    for a candidate another session may still be editing — otherwise a
+    failed lock (e.g. because another operator already has uncommitted
+    changes) turns into us discarding their work."""
+    calls: list[str] = []
+
+    fake_dev = types.SimpleNamespace(
+        open=lambda **kw: calls.append("open"),
+        close=lambda: calls.append("close"),
+    )
+    monkeypatch.setattr(rustez_module, "Device", lambda **kw: fake_dev)
+
+    fake_cfg = types.SimpleNamespace(
+        lock=_raise(_RpcError("locked by another session")),
+        rollback=lambda amount: calls.append(f"rollback({amount})"),
+        unlock=lambda: calls.append("unlock"),
+    )
+    monkeypatch.setattr(rustez_module, "Config", lambda dev: fake_cfg)
+
+    tx = rustez_module.RustezTransport()
+    with pytest.raises(TransportError):
+        tx.connect("192.0.2.1", "u", "p")
+
+    tx.rollback()
+    tx.close()
+
+    assert "rollback(0)" not in calls
+    assert "unlock" not in calls
+    assert tx._locked is False