Fluent(feat): Add find_or_create_session

why: give the fluent builder an idempotent session entry, so re-running a
build reuses the live session instead of duplicating it -- the tmuxp
load -a shape, in the forward-ref builder.

what:
- PlanBuilder.find_or_create_session(name): record the same create as
  new_session, made conditional via LazyPlan.ensure with a display-message
  probe that captures the session's id, first window, and first pane
- Cover the recorded shape and a live idempotent double-build

Author: tony

src/libtmux/experimental/fluent.py

@@ -58,6 +58,10 @@
 _CURSOR_FMT = "#{cursor_x},#{cursor_y}"
 _WAIT_PANE_POLLS = 40
 _WAIT_PANE_INTERVAL = 0.05
+#: The probe format for a session find-or-create -- the ids
+#: ``NewSession(capture_panes=True)`` captures, so a found session binds the
+#: same self/window/pane slots a created one would.
+_SESSION_PROBE = "#{session_id} #{window_id} #{pane_id}"
 
 
 def _pane_ready(cursor: str) -> bool:
@@ -158,6 +162,32 @@ def new_session(self, name: str) -> SessionRef:
         slot = self.plan.add(NewSession(session_name=name, capture_panes=True))
         return SessionRef(self.plan, name, slot)
 
+    def find_or_create_session(self, name: str) -> SessionRef:
+        """Reach session *name*, creating it only if it does not exist.
+
+        At build time this records the same create as :meth:`new_session`, but
+        makes it conditional (see :meth:`~..ops.plan.LazyPlan.ensure`): at
+        execution the plan probes for *name* and reuses the live session when it
+        is already there, so a re-run is idempotent instead of a duplicate.
+
+        Examples
+        --------
+        >>> from libtmux.experimental.engines.concrete import ConcreteEngine
+        >>> p = plan()
+        >>> _ = p.find_or_create_session("dev").window().pane()
+        >>> [op.kind for op in p.plan.operations]
+        ['new_session']
+        >>> p.run(ConcreteEngine()).ok
+        True
+        """
+        create = NewSession(session_name=name, capture_panes=True)
+        slot = self.plan.add(create)
+        self.plan.ensure(
+            slot.slot,
+            DisplayMessage(target=NameRef(name), message=_SESSION_PROBE),
+        )
+        return SessionRef(self.plan, name, slot)
+
     def sleep(self, seconds: float) -> PlanBuilder:
         """Pause *seconds* after the last recorded op (a hard fold boundary).
 

tests/experimental/test_fluent.py

@@ -135,3 +135,32 @@ def test_sleep_runs_offline() -> None:
     p.sleep(0.0)
     pane.split().do(lambda c: c.send_keys("htop"))
     assert p.run(ConcreteEngine()).ok
+
+
+def test_find_or_create_session_records_a_conditional_create() -> None:
+    """find_or_create_session records one create, made conditional via ensure."""
+    p = plan()
+    pane = p.find_or_create_session("dev").window().pane()
+    pane.do(lambda c: c.send_keys("vim"))
+    assert [op.kind for op in p.plan.operations] == ["new_session", "send_keys"]
+    assert 0 in p.plan._ensures  # the create is conditional
+
+
+def test_find_or_create_session_is_idempotent_live(session: Session) -> None:
+    """Building the same session name twice reuses it instead of duplicating."""
+    from libtmux.experimental.engines.subprocess import SubprocessEngine
+
+    engine = SubprocessEngine.for_server(session.server)
+
+    def build() -> None:
+        p = plan()
+        p.find_or_create_session("fluent-idem").window().pane().do(
+            lambda c: c.send_keys("echo hi", enter=False),
+        )
+        p.run(engine).raise_for_status()
+
+    build()
+    build()  # second run must find the existing session, not create a duplicate
+
+    named = [s for s in session.server.sessions if s.session_name == "fluent-idem"]
+    assert len(named) == 1