The borrow marks its own session stale when the login bounces

backend/druks/browser/__init__.py

@@ -1,3 +1,5 @@
+from druks.browser import subscribers  # noqa: F401  (connects the signal reaction)
+from druks.browser.exceptions import BrowserSessionSignedOutError
 from druks.browser.sessions import BrowserSession
 
-__all__ = ["BrowserSession"]
+__all__ = ["BrowserSession", "BrowserSessionSignedOutError"]

backend/druks/browser/constants.py

@@ -17,3 +17,6 @@
 
 VNC_PORT = 5900
 SCREEN_CHUNK_BYTES = 64 * 1024
+
+# The fatal a bounced borrow announces; a subscriber marks the named session stale.
+SESSION_SIGNED_OUT_SIGNAL = "browser.session_signed_out"

backend/druks/browser/exceptions.py

@@ -1,5 +1,8 @@
 from typing import ClassVar
 
+from druks.browser.constants import SESSION_SIGNED_OUT_SIGNAL
+from druks.durable.exceptions import FatalError
+
 
 class BrowserApiError(Exception):
     # Raised from a browser route; the app maps it to this status with its
@@ -19,6 +22,19 @@ def __init__(self, name: str, status: str) -> None:
         super().__init__(f"Browser session {name!r} is {status}; log in before borrowing it.")
 
 
+class BrowserSessionSignedOutError(FatalError):
+    # Raised by extension code inside a borrow when the site bounced the login.
+    # The door stamps which session; the run fails under this code and announces
+    # the bounce, and a subscriber marks the session stale. Nothing to catch.
+    code = "browser_session_signed_out"
+    broadcast_topic = SESSION_SIGNED_OUT_SIGNAL
+    session_name: str = ""
+
+    @property
+    def broadcast_facts(self) -> dict[str, str]:
+        return {"session_name": self.session_name}
+
+
 class BrowserLaunchError(BrowserApiError):
     status_code = 502
 

backend/druks/browser/sessions.py

@@ -14,6 +14,7 @@
     BrowserExportError,
     BrowserLaunchError,
     BrowserSessionNotReadyError,
+    BrowserSessionSignedOutError,
 )
 from druks.browser.locks import acquire_writer_lock, release_writer_lock
 from druks.browser.models import StoredBrowserSession
@@ -71,6 +72,11 @@ async def cdp(self):
                 listener = await browser.forward_local_port(CDP_PORT)
                 try:
                     yield f"http://127.0.0.1:{listener.get_port()}"
+                except BrowserSessionSignedOutError as error:
+                    # The extension says the login bounced; only the door knows
+                    # which session that was. The run machinery does the rest.
+                    error.session_name = self.name
+                    raise
                 finally:
                     listener.close()
                 if self.persist:
@@ -98,11 +104,6 @@ async def playwright(self):
             finally:
                 await connection.close()
 
-    def mark_stale(self) -> None:
-        """Report the login bounced — the site wants the operator back. The
-        pane shows the session as stale; the workflow decides whether to park."""
-        self.get_or_create_row().mark_stale()
-
     def get_or_create_row(self) -> StoredBrowserSession:
         """The declaration's stored half, written by the first action that
         needs it — a borrow, a login-window open, or a state import. Until

backend/druks/browser/subscribers.py

@@ -0,0 +1,10 @@
+from druks.browser.constants import SESSION_SIGNED_OUT_SIGNAL
+from druks.browser.models import StoredBrowserSession
+from druks.signals import subscribe
+
+
+@subscribe(SESSION_SIGNED_OUT_SIGNAL)
+async def signed_out_session_goes_stale(*, session_name: str, **_: object) -> None:
+    # A borrow bounced and the run failed; the stored login is dead, so the
+    # session goes stale — the pane shows it and refuses borrows until a re-login.
+    StoredBrowserSession.get_for_name(session_name).mark_stale()

backend/druks/durable/exceptions.py

@@ -1,4 +1,4 @@
-from typing import ClassVar
+from typing import Any, ClassVar
 
 
 class FatalError(Exception):
@@ -10,6 +10,15 @@ class FatalError(Exception):
     # recognize the domain stop without parsing its message. Empty for a crash.
     code: ClassVar[str] = ""
 
+    # A subclass names a signal to announce itself on when the run fails, so a
+    # subscriber can settle state the rolled-back body couldn't. Empty is silent.
+    broadcast_topic: ClassVar[str] = ""
+
+    @property
+    def broadcast_facts(self) -> dict[str, Any]:
+        # What rides that signal for the subscriber to react to.
+        return {}
+
 
 class WorkflowError(Exception):
     pass

backend/druks/workflows.py

@@ -456,6 +456,20 @@ def _log_run_event(
     return payload
 
 
+async def _broadcast_fatal(exc: FatalError) -> None:
+    # A fatal that names a topic announces itself once the run is recorded
+    # failed, so a subscriber can settle state the rolled-back body couldn't.
+    # Its own retrying checkpoint, like the run-lifecycle publishes.
+    if not exc.broadcast_topic:
+        return
+
+    async def _fan_out() -> None:
+        async with step_session():
+            await publish(exc.broadcast_topic, **exc.broadcast_facts)
+
+    await DBOS.run_step_async(StepOptions(name=exc.broadcast_topic, **_IO_RETRIES), _fan_out)
+
+
 # Park sets the gate pair together; resume and a failure clear it together, so
 # a terminal or resumed run never keeps a stale ask.
 _GATE_CLEARED: dict[str, Any] = {"input_gate": None, "input_request": None}
@@ -484,7 +498,11 @@ async def record_failed(exc: BaseException, code: str) -> None:
         result = await body()
     except (DBOSAwaitedWorkflowCancelledError, DBOSWorkflowCancelledError):
         raise
-    except (FatalError, HarnessError) as exc:
+    except FatalError as exc:
+        await record_failed(exc, exc.code)
+        await _broadcast_fatal(exc)
+        raise
+    except HarnessError as exc:
         await record_failed(exc, exc.code)
         raise
     except Exception as exc:

backend/tests/test_author_surface.py

@@ -6,7 +6,7 @@
 # exact names it exports. Pattern A — no root facade; druks stays thin.
 AUTHOR_SURFACE = {
     "druks.extensions": {"Extension", "ExtensionSettings", "Secret"},
-    "druks.browser": {"BrowserSession"},
+    "druks.browser": {"BrowserSession", "BrowserSessionSignedOutError"},
     "druks.services": {"Service", "ServiceConnectError", "ServiceNotConnectedError"},
     "druks.agents": {"Agent", "AgentOutput"},
     "druks.workflows": {

backend/tests/test_browser_borrow.py

@@ -9,6 +9,7 @@
     BrowserClientMissingError,
     BrowserLaunchError,
     BrowserSessionNotReadyError,
+    BrowserSessionSignedOutError,
     BrowserSessionWriterLockedError,
 )
 from druks.browser.models import StoredBrowserSession
@@ -215,14 +216,21 @@ async def test_launch_failure_raises_and_releases_the_lock(borrow, x_me):
     assert not redis.values
 
 
-def test_mark_stale_flags_the_row(borrow, x_me):
-    stored_session(x_me.docs)
+async def test_signed_out_borrow_stamps_the_session_and_stores_nothing(borrow, x_me):
+    """The extension raises through the borrow when the site bounced the login:
+    the door stamps which session bounced, and the dead state is never stored."""
+    browser, redis = borrow
+    stored_session(x_me.x, payload=b"live-state")
 
-    x_me.docs.mark_stale()
+    with pytest.raises(BrowserSessionSignedOutError) as caught:
+        async with x_me.x.cdp():
+            raise BrowserSessionSignedOutError("the site bounced the login")
 
-    assert (
-        StoredBrowserSession.get_for_name(x_me.docs.name).status == BrowserSessionStatus.STALE.value
-    )
+    assert caught.value.session_name == "x_me.x"
+    db_session().expire_all()
+    assert StoredBrowserSession.get_for_name(x_me.x.name).payload.decrypt() == b"live-state"
+    assert ["session-export"] not in browser.commands
+    assert not redis.values  # the writer lock released on the way out
 
 
 async def test_playwright_yields_the_logged_in_context(borrow, x_me, monkeypatch):

backend/tests/test_durable_sdk.py

@@ -482,6 +482,50 @@ async def test_fail_branch(rt):
     assert status == "ERROR"
 
 
+async def test_signed_out_run_fails_and_marks_the_session_stale(rt):
+    # The platform owns the whole bounce reaction: the run fails under the code
+    # and the stamped session row goes stale — committed apart from the failing
+    # body's rolled-back transaction.
+    from druks.browser.enums import BrowserSessionPayloadFormat, BrowserSessionStatus
+    from druks.browser.exceptions import BrowserSessionSignedOutError
+    from druks.browser.models import StoredBrowserSession
+
+    session = get_session(rt.engine)
+    try:
+        session.add(
+            StoredBrowserSession(
+                name="x_me.x",
+                payload_format=BrowserSessionPayloadFormat.STORAGE_STATE.value,
+                site="x.com",
+            )
+        )
+        session.commit()
+    finally:
+        session.close()
+
+    class BounceFlow(Workflow):
+        async def run(self) -> None:
+            error = BrowserSessionSignedOutError("the site bounced the login")
+            error.session_name = "x_me.x"
+            raise error
+
+    try:
+        wfid = await BounceFlow.start(subject=None)
+        failed = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FAILED)
+        assert failed.failure == "the site bounced the login"
+        assert failed.failure_code == "browser_session_signed_out"
+        session = get_session(rt.engine)
+        try:
+            stored = session.execute(
+                select(StoredBrowserSession).where(StoredBrowserSession.name == "x_me.x")
+            ).scalar_one()
+            assert stored.status == BrowserSessionStatus.STALE.value
+        finally:
+            session.close()
+    finally:
+        workflows._items.pop("bounce_flow", None)
+
+
 async def test_subjectless_gate_fails_loudly(rt):
     """A gate with no on_wait override fails a subjectless run now, instead of
     parking it unseen for the whole gate TTL."""

docs/writing-an-extension.md

@@ -342,9 +342,12 @@ wrapper.
 
 ``persist=True`` writes rotated state back after each borrow — for sites that
 expire an unused login. ``headless=True`` is an opt-in optimization for sites
-that don't fingerprint headless browsers. When a site logs the session out,
-call ``XMe.x.mark_stale()`` and decide in the workflow whether to park on a
-gate for the operator.
+that don't fingerprint headless browsers. When your code inside the borrow
+sees the site bounce the login, raise ``BrowserSessionSignedOutError`` (from
+``druks.browser``) and druks does the rest: the session goes stale — the pane
+shows it and refuses further borrows until the operator signs in again — and
+the run fails under that reason. There is nothing to catch; the next
+scheduled run proceeds once the login is back.
 
 Provider selection is an operator concern. Extension workspace code targets the
 Druks sandbox contract, not `exe`, AWS, or Docker directly.