diff --git a/backend/druks/browser/__init__.py b/backend/druks/browser/__init__.py index e1fd30e3..30daf1af 100644 --- a/backend/druks/browser/__init__.py +++ b/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"] diff --git a/backend/druks/browser/constants.py b/backend/druks/browser/constants.py index 0dc6d1ba..8b36df8b 100644 --- a/backend/druks/browser/constants.py +++ b/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" diff --git a/backend/druks/browser/exceptions.py b/backend/druks/browser/exceptions.py index e2f3ded5..dc059002 100644 --- a/backend/druks/browser/exceptions.py +++ b/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 diff --git a/backend/druks/browser/sessions.py b/backend/druks/browser/sessions.py index 822cfaed..6ddfd25b 100644 --- a/backend/druks/browser/sessions.py +++ b/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 diff --git a/backend/druks/browser/subscribers.py b/backend/druks/browser/subscribers.py new file mode 100644 index 00000000..b0904870 --- /dev/null +++ b/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() diff --git a/backend/druks/durable/exceptions.py b/backend/druks/durable/exceptions.py index 5842afc7..4d8726f9 100644 --- a/backend/druks/durable/exceptions.py +++ b/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 diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index 81aacbba..c8421d3e 100644 --- a/backend/druks/workflows.py +++ b/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: diff --git a/backend/tests/test_author_surface.py b/backend/tests/test_author_surface.py index b233736d..a097c86e 100644 --- a/backend/tests/test_author_surface.py +++ b/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": { diff --git a/backend/tests/test_browser_borrow.py b/backend/tests/test_browser_borrow.py index fc2160c8..e92e32b6 100644 --- a/backend/tests/test_browser_borrow.py +++ b/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): diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py index b91e9e50..6dd5bb94 100644 --- a/backend/tests/test_durable_sdk.py +++ b/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.""" diff --git a/docs/writing-an-extension.md b/docs/writing-an-extension.md index 3e7ae9e2..95996d20 100644 --- a/docs/writing-an-extension.md +++ b/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.