fix(browser): make the artifact boundary compose end-to-end and scope stores per profile

Addresses both merge blockers from @andrexibiza's review of #85351:

1. HTTP-uploaded artifacts could never be consumed by broker dispatch:
   artifact_scope_key hashed (principal, session, family), the HTTP routes
   store with an EMPTY session (API-key auth has no server session) while
   broker validation carries a session-bearing ControllerScope — every
   real upload->dispatch journey died with ArtifactScopeMismatch
   (reproduced before fixing). Canonical ownership is now
   principal/transport-family (documented in the scope-key docstring);
   ids stay unguessable server-minted 32-hex and downloads one-shot.
   New composition regression: HTTP-shape upload -> registered controller
   scope -> broker artifact dispatch, mutation-checked (re-adding session
   to the key makes it fail).

2. The 'profile-scoped' artifact store was first-profile-wins process
   state: one adapter-level singleton pinned profile B to profile A's
   physical root on multiplex listeners (same frozen-handle class as
   #88734). Stores are now cached by resolved profile, and the broker
   selects the store from the controller scope's profile_id (default-slot
   fallback preserves single-profile/test behaviour). New A/B multiplex
   regression proves distinct physical roots regardless of touch order.

Also documents the advertised ticket_expires_at as best-effort wall clock
(broker enforces expiry monotonically) per review feedback.

gateway/browser_control_artifacts.py

@@ -157,23 +157,23 @@ class ArtifactReceipt:
 def artifact_scope_key(scope: Any) -> str:
     """Derive the stable scope key an artifact is bound to.
 
-    Only server-derived identity fields participate: principal (mandatory),
-    plus session and transport family when the caller resolved them
-    (mirroring the broker's exact-identity contract).  Capabilities and
-    optional ids are intentionally excluded so a reconnect that refreshes
-    the same controller keeps its artifacts.
-
-    The API-server artifact routes authenticate by API key and bind to the
-    derived principal; the broker additionally binds to the full
-    controller scope.  A principal-only key and a full controller key
-    never collide because the digest input differs.
+    Only server-derived identity fields participate: principal (mandatory)
+    plus transport family. ``session_id`` is deliberately EXCLUDED: the HTTP
+    artifact routes authenticate by API key and can never resolve a server
+    session, while broker dispatch always carries a session-bearing
+    ControllerScope — including the session would make the two halves of the
+    intended journey (HTTP upload → broker artifact dispatch) hash to
+    different keys and never compose. Artifacts are therefore
+    principal/transport-family owned; ids are unguessable server-minted
+    32-hex and downloads are one-shot, so cross-session reuse within one
+    authenticated principal is by design. Capabilities and optional ids are
+    likewise excluded so a reconnect that refreshes the same controller
+    keeps its artifacts.
     """
     principal = ""
-    session = ""
     family = ""
     try:
         principal = str(getattr(scope, "principal_id", "") or "")
-        session = str(getattr(scope, "session_id", "") or "")
         family = str(getattr(scope, "transport_family", "") or "")
     except Exception:
         pass
@@ -181,7 +181,7 @@ def artifact_scope_key(scope: Any) -> str:
         # Fail closed: an artifact can only be minted for an authenticated
         # principal.
         raise ArtifactError("artifact scope must carry a resolved principal")
-    material = f"{principal}\x00{session}\x00{family}".encode("utf-8")
+    material = f"{principal}\x00{family}".encode("utf-8")
     return hashlib.sha256(material).hexdigest()
 
 

gateway/browser_control_broker.py

@@ -341,17 +341,46 @@ class BrowserControlBroker:
         if developer_mode is None:
             developer_mode = browser_control_developer_mode()
         self._developer_mode = developer_mode is True
-        self._artifact_store: Any = None
+        # Artifact stores keyed by resolved profile id; ``None`` is the
+        # default/unscoped store (tests, single-profile hosts). A multiplex
+        # listener attaches one store per profile so profile A touching the
+        # artifact route first can never pin profile B to A's physical root.
+        self._artifact_stores: Dict[Optional[str], Any] = {}
 
-    def attach_artifact_store(self, store: Any) -> None:
-        """Attach the process artifact store for "approved artifact id only".
+    def attach_artifact_store(
+        self, store: Any, *, profile_id: Optional[str] = None
+    ) -> None:
+        """Attach an artifact store for "approved artifact id only".
 
         ``store`` must expose ``validate(artifact_id, *, scope) -> receipt``
         raising the artifacts module's :class:`ArtifactError` subclasses.
-        None clears the reference; dispatching an artifact action without a
-        store fails closed.
+        ``profile_id`` scopes the store to one profile on multiplex hosts;
+        ``None`` registers the default store. ``store=None`` clears that
+        slot; dispatching an artifact action without a resolvable store
+        fails closed.
         """
-        self._artifact_store = store
+        if store is None:
+            self._artifact_stores.pop(profile_id, None)
+            return
+        self._artifact_stores[profile_id] = store
+
+    def _artifact_store_for_scope(self, scope: "ControllerScope") -> Any:
+        """Select the artifact store for one controller scope.
+
+        Prefers the exact profile-scoped store, falling back to the default
+        (``None``) slot so single-profile hosts and existing tests keep the
+        historical one-store behaviour.
+        """
+        profile = getattr(scope, "profile_id", None) or None
+        store = self._artifact_stores.get(profile)
+        if store is not None:
+            return store
+        return self._artifact_stores.get(None)
+
+    @property
+    def _artifact_store(self) -> Any:
+        """Back-compat view of the default artifact store (tests)."""
+        return self._artifact_stores.get(None)
 
     @property
     def developer_mode(self) -> bool:
@@ -831,7 +860,8 @@ class BrowserControlBroker:
         traversal, expiry, checksum, or scope mismatch all surface as
         :class:`ControllerRejected` before any frame is emitted.
         """
-        if self._artifact_store is None:
+        store = self._artifact_store_for_scope(scope)
+        if store is None:
             raise ControllerRejected(
                 f"{action} requires an attached artifact store"
             )
@@ -839,7 +869,7 @@ class BrowserControlBroker:
         if not isinstance(artifact_id, str) or not artifact_id.strip():
             raise ControllerRejected(f"{action} requires a non-empty artifact_id")
         try:
-            self._artifact_store.validate(artifact_id.strip(), scope=scope)
+            store.validate(artifact_id.strip(), scope=scope)
         except ControllerRejected:
             raise
         except Exception as exc:

gateway/platforms/api_server.py

@@ -1583,10 +1583,11 @@ class APIServerAdapter(BasePlatformAdapter):
         # and command lifecycle shared with the dashboard Gateway transport. This adapter only maps HTTP registration and the
         # controller WebSocket onto the broker; it owns no broker state.
         self._browser_control_broker = get_browser_control_broker()
-        # One-shot artifact transport (Phase 8 Task 29). Lazy store + limiter
-        # are created on first authenticated artifact use; tests inject their
-        # own store/limiter via _inject_browser_control_artifacts().
-        self._browser_control_artifacts: Optional[ArtifactStore] = None
+        # One-shot artifact transport (Phase 8 Task 29). Lazy per-profile
+        # stores + limiter are created on first authenticated artifact use;
+        # tests inject their own store/limiter via
+        # _inject_browser_control_artifacts().
+        self._browser_control_artifacts: Dict[str, ArtifactStore] = {}
         self._browser_control_artifact_limiter: Optional[ArtifactRateLimiter] = None
 
     def active_agent_work_count(self) -> int:
@@ -3533,6 +3534,9 @@ class APIServerAdapter(BasePlatformAdapter):
             {
                 "protocol_version": _BROWSER_CONTROL_PROTOCOL_VERSION,
                 "ticket": ticket.value,
+                # Best-effort wall-clock projection for clients; the broker
+                # enforces expiry on its monotonic clock, so after an NTP
+                # step trust ticket_expires_in_seconds, not this absolute.
                 "ticket_expires_at": time.time() + ticket_ttl,
                 "ticket_expires_in_seconds": ticket_ttl,
                 "ws_path": "/v1/browser-control/ws",
@@ -3761,11 +3765,17 @@ class APIServerAdapter(BasePlatformAdapter):
 
         The store root lives under the profile's data directory
         (``<HERMES_HOME>/plugin-data/.../artifacts``-style controlled root),
-        so artifacts never escape the profile boundary.  The root itself is
-        created on first use; TTL cleanup runs on every store/load/prune.
+        so artifacts never escape the profile boundary.  Stores are cached
+        BY RESOLVED PROFILE — on a multiplex listener, profile A touching
+        the artifact route first must never pin profile B to A's physical
+        root (same frozen-handle class as the per-profile session-storage
+        fix in #88734).  The root itself is created on first use; TTL
+        cleanup runs on every store/load/prune.
         """
-        if self._browser_control_artifacts is not None:
-            return self._browser_control_artifacts
+        profile_key = str(profile or "default")
+        store = self._browser_control_artifacts.get(profile_key)
+        if store is not None:
+            return store
         try:
             from hermes_cli.profiles import get_profile_dir
 
@@ -3788,12 +3798,14 @@ class APIServerAdapter(BasePlatformAdapter):
             allowed_mime_types=DEFAULT_ALLOWED_MIME_TYPES,
         )
         store.prune_expired()
-        self._browser_control_artifacts = store
+        self._browser_control_artifacts[profile_key] = store
         # Share the store with the broker so artifact actions dispatched to a
         # controller validate their artifact reference against the same
-        # controlled root ("approved artifact id only").
+        # profile's controlled root ("approved artifact id only").
         try:
-            self._browser_control_broker.attach_artifact_store(store)
+            self._browser_control_broker.attach_artifact_store(
+                store, profile_id=profile_key
+            )
         except Exception:
             logger.debug("could not attach artifact store to broker", exc_info=True)
         return store
@@ -3811,9 +3823,14 @@ class APIServerAdapter(BasePlatformAdapter):
         self,
         store: Optional[ArtifactStore],
         limiter: Optional[ArtifactRateLimiter] = None,
+        *,
+        profile: str = "default",
     ) -> None:
         """Inject a store/limiter (tests, diagnostics)."""
-        self._browser_control_artifacts = store
+        if store is None:
+            self._browser_control_artifacts.pop(profile, None)
+        else:
+            self._browser_control_artifacts[profile] = store
         if limiter is not None:
             self._browser_control_artifact_limiter = limiter
 

tests/gateway/test_browser_control_artifacts.py

@@ -589,3 +589,74 @@ def test_route_table_advertises_artifact_routes():
     routes = {(method, path) for method, path, _handler in adapter._http_route_table()}
     assert ("POST", "/v1/artifacts/upload") in routes
     assert ("GET", "/v1/artifacts/download/{artifact_id}") in routes
+
+
+def test_http_uploaded_artifact_composes_with_broker_dispatch(tmp_path):
+    """The real journey: HTTP upload (no session) -> broker artifact dispatch.
+
+    Regression for the scope-key mismatch review blocker: the HTTP artifact
+    routes can never resolve a server session, so artifact ownership is
+    principal/transport-family scoped and a session-bearing ControllerScope
+    must validate the same artifact.
+    """
+    store = ArtifactStore(tmp_path / "root")
+    # Upload-side scope: what api_server's facade carries (empty session).
+    receipt = store.store(
+        TEXT_BYTES,
+        filename="note.txt",
+        content_type="text/plain",
+        scope=_Scope(session=""),
+    )
+
+    broker = BrowserControlBroker()
+    broker.attach_artifact_store(store)
+    scope = _broker_scope(
+        capabilities=frozenset({"browser_artifact_upload", "controller.noop"})
+    )
+
+    def send(frame):
+        broker.complete(
+            frame["params"]["command_id"], ok=True, result={"ok": True}
+        )
+
+    broker.attach(scope, send)
+    result = broker.dispatch(
+        scope,
+        action="browser_artifact_upload",
+        arguments={"artifact_id": receipt.artifact_id},
+    )
+    assert result == {"ok": True}
+
+    # Cross-principal / cross-family access still fails closed.
+    with pytest.raises(ArtifactScopeMismatch):
+        store.validate(receipt.artifact_id, scope=_Scope(principal="other"))
+    with pytest.raises(ArtifactScopeMismatch):
+        store.validate(receipt.artifact_id, scope=_Scope(session="", family="remote-api"))
+
+
+def test_multiplex_profiles_get_distinct_stores_regardless_of_touch_order(tmp_path, monkeypatch):
+    """Profile A touching the artifact route first must not pin profile B."""
+    import gateway.platforms.api_server as api_server_mod
+
+    adapter = _adapter()
+    monkeypatch.setattr(
+        "hermes_cli.profiles.get_profile_dir",
+        lambda profile: str(tmp_path / f"home-{profile}"),
+    )
+
+    store_a = adapter._artifact_store_for("profile-a")
+    store_b = adapter._artifact_store_for("profile-b")
+    assert store_a is not store_b
+    assert str(store_a.root) != str(store_b.root)
+    assert "home-profile-a" in str(store_a.root)
+    assert "home-profile-b" in str(store_b.root)
+    # Repeat lookups return the same cached store per profile.
+    assert adapter._artifact_store_for("profile-a") is store_a
+    assert adapter._artifact_store_for("profile-b") is store_b
+
+    # The broker resolves each profile's own store from the controller scope.
+    broker = adapter._browser_control_broker
+    scope_a = _broker_scope(profile_id="profile-a")
+    scope_b = _broker_scope(profile_id="profile-b")
+    assert broker._artifact_store_for_scope(scope_a) is store_a
+    assert broker._artifact_store_for_scope(scope_b) is store_b