Merge pull request #607 from AYin-Z/fix/tray-notice-capability

fix(desktop): grant tray capability to the notice and evidence windows

CHANGELOG.md

@@ -8,10 +8,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this
 
 ### Added
 
+- Provider usage shows Grok subscription periods, product usage, and on-demand credit state for hosts signed in with `xai-oauth`. Android shows SuperGrok by default when no provider visibility choice is saved.
 - Guided Secure Link setup in Dashboard and the Desktop Relay pane, with shared read-only host CLI checks, restart instructions, and signed pairing handoff.
 
 ### Fixed
 
+- Desktop tray notices, screenshot evidence, and grant prompts stay reusable after dismissal; screenshot evidence keeps the most recently selected image. (#606)
+- Proactive phone Thread messages render relay-token and host-path media as attachments while preserving multiline text; notification previews omit media markers. (#485)
+- Desktop computer screenshots attach validated image bytes to the host tool result instead of returning base64 as plain text.
+- Relay-owned media uploads are removed on token expiry, eviction, and shutdown; media activity logs omit tokens, file paths, and screenshot bytes.
+- Plugin screenshot and navigation tools resolve Android's authenticated media token, attach the actual bounded image to host vision, and keep legacy inline screenshots readable. (#593)
 - Android Chat can open the model picker before the first turn, loads Gateway models when opened, and distinguishes loading, unavailable, and empty catalogs.
 - Secure Link configuration failures leave ordinary Relay available; route details and pairing previews resolve the advertised service namespaces.
 - Dashboard pairing QR codes support larger certificate-bearing Secure Link invites.

app/src/main/kotlin/com/hermesandroid/relay/data/ProviderUsagePreferences.kt

@@ -27,7 +27,7 @@ data class ProviderUsagePreferences(
     val visibleProviders: Set<String> = DEFAULT_VISIBLE_PROVIDERS,
 ) {
     companion object {
-        val DEFAULT_VISIBLE_PROVIDERS = setOf("openai-codex", "nous", "opencode-go")
+        val DEFAULT_VISIBLE_PROVIDERS = setOf("openai-codex", "nous", "opencode-go", "supergrok")
     }
 }
 

app/src/main/kotlin/com/hermesandroid/relay/network/relay/ProactiveMessageHandler.kt

@@ -131,7 +131,7 @@ class ProactiveMessageHandler(
         ProactiveMessageNotifier.notify(
             context = context,
             title = msg.title,
-            text = msg.text,
+            text = mediaFreeProactivePreview(msg.text),
             messageId = msg.messageId,
             chatId = msg.chatId,
         )
@@ -156,6 +156,30 @@ class ProactiveMessageHandler(
     }
 }
 
+/** Notification text is a preview; the Thread owns attachment rendering. */
+internal fun mediaFreeProactivePreview(text: String): String {
+    var fence: String? = null
+    val lines = mutableListOf<String>()
+    for (line in text.lines()) {
+        val trimmed = line.trim()
+        val delimiter = when {
+            trimmed.startsWith("```") -> "```"
+            trimmed.startsWith("~~~") -> "~~~"
+            else -> null
+        }
+        if (delimiter != null) {
+            fence = if (fence == delimiter) null else if (fence == null) delimiter else fence
+        }
+        val markerOnly = fence == null && (
+            trimmed.startsWith("MEDIA:hermes-relay://") ||
+                trimmed.startsWith("MEDIA:/") ||
+                Regex("^MEDIA:[A-Za-z]:\\\\").containsMatchIn(trimmed)
+            )
+        if (!markerOnly && trimmed.isNotEmpty()) lines += trimmed
+    }
+    return lines.joinToString(" ").ifBlank { "Attachment" }
+}
+
 /**
  * A parsed agent-initiated message. `surfacing` is the optional route hint
  * (null = app default); Phase 2 keys inbox/session delivery off it.

app/src/main/kotlin/com/hermesandroid/relay/network/relay/RelayHttpClient.kt

@@ -336,11 +336,11 @@ class RelayHttpClient(
                 Result.success(FetchedMedia(contentType, bytes, fileName, sensitive))
             }
         } catch (e: IOException) {
-            Log.w(TAG, "fetchMedia failed: ${e.message}")
-            Result.failure(e)
+            Log.w(TAG, "fetchMedia failed")
+            Result.failure(if (e is RelayMediaLimitException) e else IOException("Relay media request failed"))
         } catch (e: Exception) {
-            Log.w(TAG, "fetchMedia unexpected error: ${e.message}")
-            Result.failure(e)
+            Log.w(TAG, "fetchMedia unexpected error")
+            Result.failure(IOException("Relay media request failed"))
         }
     }
 
@@ -416,7 +416,7 @@ class RelayHttpClient(
                     val reason = when (response.code) {
                         401 -> "Unauthorized — re-pair with the relay"
                         403 -> "Path not allowed by relay sandbox"
-                        404 -> "File not found on relay: $path"
+                        404 -> "File not found on relay"
                         400 -> "Bad request — missing path"
                         in 500..599 -> "Relay error (HTTP ${response.code})"
                         else -> "HTTP ${response.code}: ${response.message.ifBlank { "request failed" }}"
@@ -446,15 +446,15 @@ class RelayHttpClient(
                 Result.success(FetchedMedia(contentType, bytes, fileName, sensitive))
             }
         } catch (e: IOException) {
-            Log.w(TAG, "fetchMediaByPath failed for $path: ${e.message}")
+            Log.w(TAG, "fetchMediaByPath failed")
             if (e is RelayMediaLimitException) {
                 Result.failure(e)
             } else {
-                Result.failure(IOException("Relay unreachable: ${e.message ?: "IO error"}"))
+                Result.failure(IOException("Relay media request failed"))
             }
         } catch (e: Exception) {
-            Log.w(TAG, "fetchMediaByPath unexpected error for $path: ${e.message}")
-            Result.failure(e)
+            Log.w(TAG, "fetchMediaByPath unexpected error")
+            Result.failure(IOException("Relay media request failed"))
         }
     }
 

app/src/main/kotlin/com/hermesandroid/relay/network/upstream/ChatHandler.kt

@@ -479,12 +479,15 @@ class ChatHandler {
         arrivedWhileAway: Boolean = false,
     ) {
         val id = messageId?.let { "proactive-$it" } ?: "proactive-${java.util.UUID.randomUUID()}"
+        val mediaHits = mutableListOf<Pair<String, MediaMarkerHit>>()
+        val cleanedText = extractMediaMarkersFromContent(id, text, mediaHits)
+        val visibleText = if (mediaHits.isEmpty()) text else cleanedText
         _messages.update { list ->
             if (messageId != null && list.any { it.id == id }) return@update list
             val msg = ChatMessage(
                 id = id,
                 role = MessageRole.ASSISTANT,
-                content = text,
+                content = visibleText,
                 timestamp = System.currentTimeMillis(),
                 agentName = agentName,
                 badges = if (arrivedWhileAway) listOf("While away") else emptyList(),
@@ -492,6 +495,8 @@ class ChatHandler {
             )
             (list + msg).let { if (it.size > MAX_MESSAGES) it.drop(it.size - MAX_MESSAGES) else it }
         }
+        // The row must exist before the ViewModel attaches a loading card.
+        mediaHits.forEach { (_, hit) -> dispatchMediaHit(id, hit) }
     }
 
     /**
@@ -1735,34 +1740,7 @@ class ChatHandler {
         // Now that the reloaded messages are in state, fire callbacks so the
         // ViewModel can insert LOADING/FAILED attachments via mutateMessage.
         for ((messageId, hit) in pendingMediaHits) {
-            when (hit) {
-                is MediaMarkerHit.RelayToken -> {
-                    val dedupeKey = "$messageId:relay:${hit.token}"
-                    if (dispatchedMediaMarkers.add(dedupeKey)) {
-                        val alreadyHydrated = _messages.value
-                            .firstOrNull { it.matchesIdentity(messageId) }
-                            ?.attachments
-                            ?.any { it.relayToken == hit.token } == true
-                        if (!alreadyHydrated) {
-                            Log.d(TAG, "Media marker accepted from reloaded Relay history")
-                            onMediaAttachmentRequested(messageId, hit.token)
-                        }
-                    }
-                }
-                is MediaMarkerHit.BarePath -> {
-                    val dedupeKey = "$messageId:bare:${hit.path}"
-                    if (dispatchedMediaMarkers.add(dedupeKey)) {
-                        val alreadyHydrated = _messages.value
-                            .firstOrNull { it.matchesIdentity(messageId) }
-                            ?.attachments
-                            ?.any { it.relayToken == hit.path } == true
-                        if (!alreadyHydrated) {
-                            Log.d(TAG, "Media marker (bare-path, reload): ${hit.path}")
-                            onMediaBarePathRequested(messageId, hit.path)
-                        }
-                    }
-                }
-            }
+            dispatchMediaHit(messageId, hit)
         }
         for ((messageId, path) in pendingPersistedUserImages) {
             onPersistedUserImageRequested(messageId, path)
@@ -1996,29 +1974,33 @@ class ChatHandler {
         content: String,
         out: MutableList<Pair<String, MediaMarkerHit>>,
     ): String {
-        var cleaned = content
+        val visibleLines = mutableListOf<String>()
         var openFence: String? = null
         for (rawLine in content.lines()) {
             val trimmed = rawLine.trim()
-            if (trimmed.isEmpty()) continue
+            if (trimmed.isEmpty()) {
+                visibleLines += rawLine
+                continue
+            }
             val delimiter = fenceDelimiter(rawLine)
             if (delimiter != null) {
                 openFence = if (openFence == delimiter) null else if (openFence == null) delimiter else openFence
+                visibleLines += rawLine
+                continue
+            }
+            if (openFence != null) {
+                visibleLines += rawLine
                 continue
             }
-            if (openFence != null) continue
 
             val hits = parseMediaMarkerLine(trimmed)
             if (hits.isNotEmpty()) {
                 hits.forEach { out.add(messageId to it) }
-                cleaned = cleaned
-                    .replace("\n$rawLine\n", "\n")
-                    .replace("\n$rawLine", "")
-                    .replace("$rawLine\n", "")
-                    .replace(rawLine, "")
+            } else {
+                visibleLines += rawLine
             }
         }
-        return cleaned.trim()
+        return visibleLines.joinToString("\n").trim()
     }
 
     /**
@@ -2625,27 +2607,28 @@ class ChatHandler {
      */
     private fun tryDispatchMediaMarker(messageId: String, line: String): Boolean {
         val hits = parseMediaMarkerLine(line)
-        for (hit in hits) {
-            when (hit) {
-                is MediaMarkerHit.RelayToken -> {
-                    val dedupeKey = "$messageId:relay:${hit.token}"
-                    if (dispatchedMediaMarkers.add(dedupeKey)) {
-                        Log.d(TAG, "Media marker accepted from Relay stream")
-                        onMediaAttachmentRequested(messageId, hit.token)
-                    }
-                }
-                is MediaMarkerHit.BarePath -> {
-                    val dedupeKey = "$messageId:bare:${hit.path}"
-                    if (dispatchedMediaMarkers.add(dedupeKey)) {
-                        Log.d(TAG, "Media marker (bare-path): ${hit.path}")
-                        onMediaBarePathRequested(messageId, hit.path)
-                    }
-                }
-            }
-        }
+        hits.forEach { dispatchMediaHit(messageId, it) }
         return hits.isNotEmpty()
     }
 
+    private fun dispatchMediaHit(messageId: String, hit: MediaMarkerHit) {
+        val (key, reference) = when (hit) {
+            is MediaMarkerHit.RelayToken -> "$messageId:relay:${hit.token}" to hit.token
+            is MediaMarkerHit.BarePath -> "$messageId:bare:${hit.path}" to hit.path
+        }
+        if (!dispatchedMediaMarkers.add(key)) return
+        val alreadyHydrated = _messages.value
+            .firstOrNull { it.matchesIdentity(messageId) }
+            ?.attachments
+            ?.any { it.relayToken == reference } == true
+        if (alreadyHydrated) return
+        Log.d(TAG, "Media marker accepted")
+        when (hit) {
+            is MediaMarkerHit.RelayToken -> onMediaAttachmentRequested(messageId, hit.token)
+            is MediaMarkerHit.BarePath -> onMediaBarePathRequested(messageId, hit.path)
+        }
+    }
+
     /**
      * Remove a matched annotation line from the message's displayed content.
      * This prevents the raw annotation text (e.g., `💻 terminal`) from showing

app/src/main/kotlin/com/hermesandroid/relay/ui/screens/UsageLimitsScreen.kt

@@ -642,6 +642,7 @@ private fun ProviderUsageDisplaySettings(
                     "openai-codex" to "Codex",
                     "nous" to "Nous",
                     "opencode-go" to "OpenCode Go",
+                    "supergrok" to "SuperGrok",
                 )
             } else {
                 providers.map { it.id to it.displayName }

app/src/test/kotlin/com/hermesandroid/relay/data/ProviderUsagePreferencesTest.kt

@@ -44,11 +44,20 @@ class ProviderUsagePreferencesTest {
         val preferences = repository.preferences.first()
         assertEquals(ProviderUsageLandingMode.Summary, preferences.landingMode)
         assertEquals(
-            setOf("openai-codex", "nous", "opencode-go"),
+            setOf("openai-codex", "nous", "opencode-go", "supergrok"),
             preferences.visibleProviders,
         )
     }
 
+    @Test
+    fun existingInstallWithoutProviderChoiceGetsCurrentDefaults() = runTest {
+        repository.setLandingMode(ProviderUsageLandingMode.Expanded)
+
+        val preferences = repository.preferences.first()
+        assertEquals(ProviderUsageLandingMode.Expanded, preferences.landingMode)
+        assertTrue("supergrok" in preferences.visibleProviders)
+    }
+
     @Test
     fun persistsDisplayMode() = runTest {
         repository.setLandingMode(ProviderUsageLandingMode.Expanded)
@@ -66,4 +75,13 @@ class ProviderUsagePreferencesTest {
         assertTrue("openai-codex" in preferences.visibleProviders)
         assertTrue("opencode-go" in preferences.visibleProviders)
     }
+
+    @Test
+    fun persistsSuperGrokHiddenChoice() = runTest {
+        repository.setProviderVisible("supergrok", false)
+
+        val preferences = repository.preferences.first()
+        assertTrue("nous" in preferences.visibleProviders)
+        assertFalse("supergrok" in preferences.visibleProviders)
+    }
 }

app/src/test/kotlin/com/hermesandroid/relay/network/relay/ProactiveMessageHandlerTest.kt

@@ -48,6 +48,15 @@ class ProactiveMessageHandlerTest {
         }
     }
 
+    @Test
+    fun `notification previews omit media markers while thread text remains complete`() {
+        val text = "Headline\nDetail\nMEDIA:hermes-relay://private-token-123456\nMEDIA:/tmp/report.png"
+        assertEquals("Headline Detail", mediaFreeProactivePreview(text))
+        assertEquals("Attachment", mediaFreeProactivePreview("MEDIA:hermes-relay://private-token-123456"))
+        val fenced = "Example\n```\nMEDIA:/tmp/example.png\n```"
+        assertTrue(mediaFreeProactivePreview(fenced).contains("MEDIA:/tmp/example.png"))
+    }
+
     @Test
     fun `inbox surfacing persists silently`() {
         val persisted = mutableListOf<ProactiveMessage>()

app/src/test/kotlin/com/hermesandroid/relay/network/upstream/ChatHandlerTest.kt

@@ -182,6 +182,49 @@ class ChatHandlerTest {
         assertTrue(handler.messages.value.single().content.contains("MEDIA:/tmp/example.pdf"))
     }
 
+    @Test
+    fun proactiveThreadMessage_dispatchesMediaAfterInsertionAndPreservesTextLines() {
+        val tokens = mutableListOf<Pair<String, String>>()
+        val paths = mutableListOf<Pair<String, String>>()
+        handler.onMediaAttachmentRequested = { id, token ->
+            assertTrue(handler.messages.value.any { it.id == id })
+            tokens += id to token
+        }
+        handler.onMediaBarePathRequested = { id, path -> paths += id to path }
+
+        val content = "Headline\nDetail one\n\nMEDIA:hermes-relay://tok123\nDetail two\nMEDIA:/tmp/report.png"
+        handler.addAgentThreadMessage(content, "push-1", "Agent")
+        handler.addAgentThreadMessage(content, "push-1", "Agent")
+
+        assertEquals(1, handler.messages.value.size)
+        assertEquals("Headline\nDetail one\n\nDetail two", handler.messages.value.single().content)
+        assertEquals(listOf("proactive-push-1" to "tok123"), tokens)
+        assertEquals(listOf("proactive-push-1" to "/tmp/report.png"), paths)
+    }
+
+    @Test
+    fun proactiveThreadMessage_keepsFencedAndProseMediaExamples() {
+        val tokens = mutableListOf<String>()
+        handler.onMediaAttachmentRequested = { _, token -> tokens += token }
+        val content = "Intro\n```\nMEDIA:hermes-relay://example123\n```\nExample: MEDIA:hermes-relay://example456"
+
+        handler.addAgentThreadMessage(content, "push-2", null)
+
+        assertEquals(content, handler.messages.value.single().content)
+        assertTrue(tokens.isEmpty())
+    }
+
+    @Test
+    fun proactiveThreadMessage_doesNotRemoveIdenticalMarkerInsideCodeFence() {
+        val tokens = mutableListOf<String>()
+        handler.onMediaAttachmentRequested = { _, token -> tokens += token }
+        val marker = "MEDIA:hermes-relay://same-token-123456"
+        handler.addAgentThreadMessage("Text\n```\n$marker\n```\n$marker", "push-3", null)
+
+        assertEquals("Text\n```\n$marker\n```", handler.messages.value.single().content)
+        assertEquals(listOf("same-token-123456"), tokens)
+    }
+
     @Test
     fun onTextDelta_setsStreamingFlag() {
         handler.onTextDelta("assist-1", "delta")

desktop/README.md

@@ -433,6 +433,8 @@ The server-side plugin (`plugin/tools/desktop_tool.py`) registers `desktop_*` to
 
 `desktop_computer_status`, `desktop_computer_screenshot`, `desktop_computer_action`, `desktop_computer_grant_request`, and `desktop_computer_cancel` are registered server-side but the desktop client advertises and serves them only when explicitly enabled:
 
+The screenshot tool attaches a validated PNG or JPEG as a native host image result when the bounded relay response contains image bytes. A `save_to` capture remains a saved-path response. The desktop wire budget still rejects oversized captures rather than placing unbounded image data in a tool result.
+
 ```sh
 hermes-relay computer-use enable
 hermes-relay computer-use status

desktop/src/tools/handlers/computer.ts

@@ -329,6 +329,7 @@ async function cuaSnapshot(args: Record<string, unknown>, ctx: ToolContext): Pro
     elements: safeElements,
     tree_markdown: raw.tree_markdown,
     screenshot_base64: raw.screenshot_base64,
+    screenshot_mime_type: raw.screenshot_mime_type,
     screenshot_width: raw.screenshot_width,
     screenshot_height: raw.screenshot_height,
     truncated: elements.length > safeElements.length

desktop/tests/computerCuaHandlers.test.ts

@@ -51,7 +51,9 @@ class FakeCuaSession {
     return {
       snapshot_id: `s0000000${this.snapshotNumber}`,
       elements: [{ element_index: 7, element_token: 'e1234abcd', role: 'button', label: 'Seven' }],
-      tree_markdown: '[7] button Seven'
+      tree_markdown: '[7] button Seven',
+      screenshot_base64: 'aW1hZ2U=',
+      screenshot_mime_type: 'image/png'
     }
   }
 
@@ -140,9 +142,13 @@ test('CUA handlers issue a Hermes token, execute once, and verify with a fresh s
       ok: boolean
       backend: string
       elements: Array<{ snapshot_token: string; element_token?: string }>
+      screenshot_base64: string
+      screenshot_mime_type: string
     }
     assert.equal(observed.ok, true)
     assert.equal(observed.backend, 'cua_driver')
+    assert.equal(observed.screenshot_base64, 'aW1hZ2U=')
+    assert.equal(observed.screenshot_mime_type, 'image/png')
     assert.equal(observed.elements[0]!.element_token, undefined)
     assert.match(observed.elements[0]!.snapshot_token, /^hermes-snapshot-/)
 

desktop/tray/capabilities/secondary-windows.json

@@ -0,0 +1,7 @@
+{
+  "$schema": "../gen/schemas/desktop-schema.json",
+  "identifier": "secondary-windows",
+  "description": "Dismiss the connection notice and screenshot evidence windows",
+  "windows": ["notice", "evidence"],
+  "permissions": ["core:window:allow-hide"]
+}

desktop/tray/src/main.rs

@@ -2514,10 +2514,14 @@ mod app {
                 label,
                 event: WindowEvent::CloseRequested { api, .. },
                 ..
-            } if label == "main" => {
+            } if matches!(label.as_str(), "main" | "grant" | "notice" | "evidence") => {
                 api.prevent_close();
-                if let Some(window) = handle.get_webview_window("main") {
-                    request_main_hide(&window);
+                if let Some(window) = handle.get_webview_window(&label) {
+                    if label == "main" {
+                        request_main_hide(&window);
+                    } else {
+                        let _ = window.hide();
+                    }
                 }
             }
             _ => {}

desktop/tray/tests/management_contract.rs

@@ -340,3 +340,47 @@ fn management_window_keeps_the_reviewed_compact_geometry() {
     assert!(ui.contains("hide().finally(() => setWindowVisible(true))"));
     assert!(ui.contains("document.visibilityState === 'visible'"));
 }
+
+#[test]
+fn secondary_windows_have_only_the_dismissal_permission() {
+    let config: serde_json::Value =
+        serde_json::from_str(include_str!("../tauri.conf.json")).unwrap();
+    let main: serde_json::Value =
+        serde_json::from_str(include_str!("../capabilities/default.json")).unwrap();
+    let secondary: serde_json::Value =
+        serde_json::from_str(include_str!("../capabilities/secondary-windows.json")).unwrap();
+
+    let configured: std::collections::BTreeSet<&str> = config["app"]["windows"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|window| window["label"].as_str().unwrap())
+        .collect();
+    let main_windows: std::collections::BTreeSet<&str> = main["windows"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|label| label.as_str().unwrap())
+        .collect();
+    let secondary_windows: std::collections::BTreeSet<&str> = secondary["windows"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|label| label.as_str().unwrap())
+        .collect();
+
+    assert_eq!(main_windows, ["main", "grant"].into_iter().collect());
+    assert_eq!(
+        secondary_windows,
+        ["notice", "evidence"].into_iter().collect()
+    );
+    assert_eq!(
+        configured,
+        main_windows.union(&secondary_windows).copied().collect()
+    );
+    assert!(main_windows.is_disjoint(&secondary_windows));
+    assert_eq!(
+        secondary["permissions"],
+        serde_json::json!(["core:window:allow-hide"])
+    );
+}

desktop/tray/ui/App.tsx

@@ -286,16 +286,20 @@ function EvidenceWindow() {
   const [evidenceId, setEvidenceId] = useState<string | null>(null)
   const [source, setSource] = useState<string | null>(null)
   const [error, setError] = useState<string | null>(null)
+  const requestGeneration = useRef(0)
   useEffect(() => {
     const receive = (event: Event) => {
       const id = (event as CustomEvent<{ evidenceId: string }>).detail.evidenceId
+      const generation = ++requestGeneration.current
       setEvidenceId(id); setSource(null); setError(null)
-      void call<string>('get_activity_screenshot', { evidenceId: id }).then(setSource).catch(value => setError(String(value)))
+      void call<string>('get_activity_screenshot', { evidenceId: id })
+        .then(value => { if (generation === requestGeneration.current) setSource(value) })
+        .catch(value => { if (generation === requestGeneration.current) setError(String(value)) })
     }
     const close = (event: KeyboardEvent) => { if (event.key === 'Escape') void getCurrentWindow().hide() }
     window.addEventListener('hermes-screenshot-evidence', receive)
     window.addEventListener('keydown', close)
-    return () => { window.removeEventListener('hermes-screenshot-evidence', receive); window.removeEventListener('keydown', close) }
+    return () => { requestGeneration.current++; window.removeEventListener('hermes-screenshot-evidence', receive); window.removeEventListener('keydown', close) }
   }, [])
   return <div className="evidence-shell">
     <header><span><Eye /><strong>Screenshot evidence</strong><small>Stored locally with this activity event</small></span><button aria-label="Close screenshot" onClick={() => getCurrentWindow().hide()}><X /></button></header>

docs/relay-server.md

@@ -379,10 +379,11 @@ See [`docs/spec.md` §3.3](spec.md) for the full auth flow and the QR wire forma
 | `/sessions/{token_prefix}` | PATCH | Bearer-auth'd, self-targeted, and reduction-only. Body `{"ttl_seconds": 3600}`, `{"grants": {"terminal": 600}}`, or both may shorten the caller's current session policy. A bearer cannot target another session, extend its lifetime, add or lengthen grants, or change a finite expiry to never-expire; authority-increasing changes require a fresh operator-approved pairing flow. Omitted grants retain their existing absolute ceilings and are clamped if the parent session is shortened. Returns 200 with the reduced `{expires_at, grants}`; 400 on missing/invalid or unknown grants; 403 on cross-session targets or policy expansion; 404 on prefix miss; 409 on ambiguous prefix. |
 | `/chat/image-activity` | GET | Optional read-only Standard Gateway compatibility route. Requires a valid Relay bearer with an active `chat` grant and query parameters `profile`, `session_id`, and `since` (Unix seconds). Reads the selected profile's Hermes `state.db` without mutation and returns persisted `image_generate` calls as `running` or `completed`. Android polls only during an active turn, deduplicates against native Gateway tool events, and silently disables the bridge when the route is absent. |
 | `/clipboard/inbox` | POST | Bearer-auth'd clipboard rendezvous used by remote clients before native platform clipboard fallback. |
-| `/media/register` | POST | **Loopback only.** Register a file path with the in-memory `MediaRegistry` and receive an opaque token. Used by host-local tools (`android_screenshot` etc.) to make a file fetchable by the paired phone without leaking the filesystem path. Request body: `{"path": "/abs/path", "content_type": "image/jpeg", "file_name": "screenshot.jpg"}`. Response: `{"ok": true, "token": "<url-safe-16>", "expires_at": <unix>}`. Returns 403 for non-loopback callers, 400 on validation failure (relative path, missing file, oversized, outside allowed roots, etc). Path sandboxing is enforced server-side — see ADR 14. |
-| `/media/upload` | POST | Bearer-auth'd small upload endpoint for phone-originated media. Accepts JSON `{file_name, content_type, content}` where `content` is base64 and registers the decoded bytes with the media registry. |
+| `/media/register` | POST | **Loopback only.** Register a file path with the in-memory `MediaRegistry` and receive an opaque token. Host-owned files are never deleted by the registry. A relay-managed `android_screenshot_` temp file can opt into cleanup with `owned_file: true`; the registry validates its location and name. Request body: `{"path": "/abs/path", "content_type": "image/png", "file_name": "screenshot.png"}`. Response: `{"ok": true, "token": "<token>", "expires_at": <unix>}`. Returns 403 for non-loopback callers, 400 on validation failure. |
+| `/media/upload` | POST | Bearer-authenticated multipart upload for phone-originated media (`file` field). Streams to a size-bounded relay-owned temporary file and registers a token. Relay-owned files are removed on token expiry, LRU eviction, or orderly shutdown. |
+| `/media/{token}/sensitive` | POST | **Loopback only.** Mark an existing token sensitive before the host tool returns its marker. The subsequent media fetch includes `X-Media-Sensitive: 1`; no second image copy is created. Returns 404 for missing or expired tokens. |
 | `/media/{token}` | GET | Stream the bytes of a previously-registered file. Requires `Authorization: Bearer <session_token>` (same token the WSS channel uses; validated against `SessionManager`). Response has the registered `Content-Type` plus `Content-Disposition: inline; filename="..."` when a file name was provided. Returns 401 without auth or with an invalid bearer, 404 if the token is unknown or expired. The client never sees the underlying path — the token is the only handle. |
-| `/media/by-path` | GET | Stream the bytes of a file **addressed by absolute path** rather than by registry token. Covers the case where an agent's LLM freeform-emits a `MEDIA:/abs/path.ext` marker in its response text (upstream `hermes-agent/agent/prompt_builder.py` explicitly instructs the model to do this) — no loopback register step is needed. Query parameters: `path` (required, absolute) and `content_type` (optional; otherwise guessed from extension via Python's `mimetypes`). Requires `Authorization: Bearer <session_token>`. Path sandboxing is identical to `/media/register`: must be absolute, must `realpath`-resolve under an allowed root (`tempfile.gettempdir()` + `HERMES_WORKSPACE` + `RELAY_MEDIA_ALLOWED_ROOTS`), must exist, must be a regular file, must fit under `RELAY_MEDIA_MAX_SIZE_MB`. Response carries `Content-Type` and `Content-Disposition: inline; filename="<basename>"`. Error shapes: 400 missing `path`; 401 missing/invalid bearer; 403 outside sandbox / not absolute / too large; 404 file not found or not a regular file. See ADR 14. |
+| `/media/by-path` | GET | Bearer-authenticated fetch for an absolute-path `MEDIA:/...` marker. Requires a regular file within the size cap and always rejects credential/system paths. Allowed-root enforcement is opt-in through `RELAY_MEDIA_STRICT_SANDBOX=1`; default mode accepts other absolute readable paths. Optional `content_type` overrides extension guessing. Returns 400 for a missing path, 401 for invalid auth, 403 for policy/size failures, and 404 for a missing file. |
 | `/voice/transcribe` | POST | Bearer-auth'd via either a Relay session token with active `voice:stt` grant or a valid Hermes API bearer token. Non-loopback API-bearer calls require HTTPS unless `RELAY_ALLOW_INSECURE_API_BEARER=1`. `multipart/form-data` with an audio file field (any name — first field is used). Android may include `?profile=<name>` so the active profile context is recorded in the response/UI; execution still goes through the upstream STT helper. |
 | `/voice/synthesize` | POST | Bearer-auth'd via either a Relay session token with active `voice:tts` grant or a valid Hermes API bearer token. Non-loopback API-bearer calls require HTTPS unless `RELAY_ALLOW_INSECURE_API_BEARER=1`. JSON body `{"text": "...", "profile": "mizu"}` (max 5000 chars) runs the basic fallback TTS helper and serves the resulting mp3. Normal assistant speech prefers `/voice/output/*`. |
 | `/voice/config` | GET | Bearer-auth'd via either a Relay session token with active `voice:config` grant or a valid Hermes API bearer token. Optional `?profile=<name>` resolves `tts:` / `stt:` from `~/.hermes/profiles/<name>/config.yaml` where present, otherwise falls back. Returns provider info plus `profile`, `config_scope`, and `fallback_to_global`. |

docs/spec.md

@@ -980,7 +980,7 @@ Tools register against the Hermes plugin API in `plugin/tools/android_tool.py` (
 |------|-----------|---------|--------|
 | `android_ping` | `GET /ping` | Liveness check — does not require master enable | sideload Device Control |
 | `android_screen` | `GET /screen` | Serialize the accessibility tree → `ScreenContent` | sideload Device Control |
-| `android_screenshot` | `GET /screenshot` | `MediaProjection` PNG → `MEDIA:hermes-relay://<token>` | sideload Device Control |
+| `android_screenshot` | `GET /screenshot`, then authenticated `GET /media/<token>` | `MediaProjection` PNG → bounded native image tool result with the `MEDIA:hermes-relay://<token>` phone-delivery marker; older inline base64 responses remain readable | sideload Device Control |
 | `android_current_app` | `GET /current_app` | Best-effort foregrounded package name; use `/screen` for verification | sideload Device Control |
 | `android_get_apps` (`/apps` legacy) | `GET /get_apps` | Installed launcher apps | sideload Device Control |
 | `android_tap` | `POST /tap` | Tap at `(x, y)` or on resolved `node_id` | sideload Device Control |

plugin/relay/channels/desktop.py

@@ -54,7 +54,10 @@ ADB_RESPONSE_TIMEOUT = 300.0  # seconds — approval plus a bounded 120 s operat
 
 # Keys whose values must never appear in the ring buffer. Matched
 # case-insensitively against the full key name.
-_REDACT_KEYS = frozenset({"password", "token", "secret", "otp", "bearer", "api_key"})
+_REDACT_KEYS = frozenset({
+    "password", "token", "secret", "otp", "bearer", "api_key",
+    "bytes_base64", "screenshot_base64",
+})
 
 # Cap for the recent-commands ring buffer.
 RECENT_COMMANDS_MAX = 100
@@ -648,7 +651,7 @@ class DesktopHandler:
         payload = envelope.get("payload") or {}
         request_id = payload.get("request_id")
         if not isinstance(request_id, str) or not request_id:
-            logger.warning("desktop: response missing request_id: %s", payload)
+            logger.warning("desktop: response missing request_id")
             return
 
         async with self._lock:
@@ -693,9 +696,12 @@ class DesktopHandler:
             record.error = error_msg
         if result is not None and record.result_summary is None:
             try:
-                summary = json.dumps(result, default=str)
+                if record.tool == "desktop_computer_screenshot":
+                    summary = "Desktop screenshot response received"
+                else:
+                    summary = json.dumps(_redact_args(result), default=str)
             except (TypeError, ValueError):
-                summary = str(result)
+                summary = "Desktop result unavailable for activity summary"
             if len(summary) > 500:
                 summary = summary[:497] + "..."
             record.result_summary = summary

plugin/relay/client.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 import json
 import logging
 import os
+import re
 import urllib.error
 import urllib.request
 
@@ -68,7 +69,7 @@ def _post_loopback(
             try:
                 return json.loads(raw)
             except json.JSONDecodeError:
-                logger.warning("POST %s returned non-JSON body: %r", url, raw[:200])
+                logger.warning("Relay POST returned non-JSON body")
                 return None
     except (urllib.error.URLError, urllib.error.HTTPError, OSError, ValueError) as exc:
         logger.warning("POST %s failed: %s", url, exc)
@@ -83,6 +84,7 @@ def register_media(
     port: int | None = None,
     timeout: float = 5.0,
     sensitive: bool = False,
+    owned_file: bool = False,
 ) -> str | None:
     """Register ``path`` with the local relay and return an opaque token.
 
@@ -104,6 +106,7 @@ def register_media(
         "content_type": content_type,
         "file_name": file_name,
         "sensitive": bool(sensitive),
+        "owned_file": bool(owned_file),
     }
 
     data = _post_loopback(host, port, "/media/register", payload, timeout)
@@ -111,9 +114,7 @@ def register_media(
         return None
 
     if not data.get("ok"):
-        logger.warning(
-            "Relay rejected media registration: %s", data.get("error")
-        )
+        logger.warning("Relay rejected media registration")
         return None
 
     token = data.get("token")
@@ -122,3 +123,21 @@ def register_media(
         return None
 
     return token
+
+
+def mark_media_sensitive(token: str, port: int | None = None) -> bool:
+    """Mark an existing relay token private without copying its image bytes."""
+    if not isinstance(token, str) or not re.fullmatch(r"[A-Za-z0-9_-]{16,128}", token):
+        return False
+    relay_port = _default_port() if port is None else port
+    request = urllib.request.Request(
+        f"http://127.0.0.1:{relay_port}/media/{token}/sensitive",
+        data=b"{}",
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=5.0) as response:
+            return response.status == 200
+    except (urllib.error.URLError, OSError, ValueError):
+        return False

plugin/relay/media.py

@@ -226,6 +226,8 @@ class _MediaEntry:
     # at registration → False (not sensitive). See
     # docs/plans/2026-06-18-attachment-experience.md §C4.
     sensitive: bool = False
+    # Only relay-created upload files are deleted when this entry retires.
+    owned_file: bool = False
 
     @property
     def is_expired(self) -> bool:
@@ -531,16 +533,16 @@ class MediaRegistry:
         self.allowed_roots: list[str] = base_roots
 
         self._entries: "OrderedDict[str, _MediaEntry]" = OrderedDict()
+        self._owned_paths: set[str] = set()
         self._lock = asyncio.Lock()
 
         logger.info(
             "MediaRegistry initialized (max_entries=%d, ttl=%ds, "
-            "max_size=%d bytes, strict_sandbox=%s, roots=%s)",
+            "max_size=%d bytes, strict_sandbox=%s)",
             max_entries,
             ttl_seconds,
             max_size_bytes,
             strict_sandbox,
-            self.allowed_roots,
         )
 
     # ── Public API ──────────────────────────────────────────────────────
@@ -551,6 +553,7 @@ class MediaRegistry:
         content_type: str,
         file_name: str | None = None,
         sensitive: bool = False,
+        owned_file: bool = False,
     ) -> _MediaEntry:
         """Validate ``path`` and register a new media entry.
 
@@ -570,6 +573,13 @@ class MediaRegistry:
         real_path, size = validate_media_path(
             path, self.allowed_roots, self.max_size_bytes
         )
+        if owned_file and (
+            os.path.dirname(real_path) != os.path.realpath(tempfile.gettempdir())
+            or not os.path.basename(real_path).startswith(
+                ("hermes-relay-upload-", "android_screenshot_")
+            )
+        ):
+            raise MediaRegistrationError("managed upload path required")
 
         token = secrets.token_urlsafe(16)
         now = time.time()
@@ -583,24 +593,22 @@ class MediaRegistry:
             expires_at=now + self.ttl_seconds,
             last_accessed=now,
             sensitive=bool(sensitive),
+            owned_file=owned_file,
         )
 
         async with self._lock:
-            self._cleanup_locked()
             self._entries[token] = entry
+            if owned_file:
+                self._owned_paths.add(real_path)
+            self._cleanup_locked()
             # Evict oldest while over cap
             while len(self._entries) > self.max_entries:
-                evicted_token, evicted = self._entries.popitem(last=False)
-                logger.info(
-                    "MediaRegistry LRU eviction: token=%s... path=%s",
-                    evicted_token[:8],
-                    evicted.path,
-                )
+                _, evicted = self._entries.popitem(last=False)
+                self._retire_entry(evicted)
+                logger.info("MediaRegistry LRU eviction")
 
         logger.info(
-            "Registered media token=%s... path=%s size=%d type=%s sensitive=%s",
-            token[:8],
-            real_path,
+            "Registered media size=%d type=%s sensitive=%s",
             size,
             content_type,
             entry.sensitive,
@@ -622,9 +630,8 @@ class MediaRegistry:
                 return None
             if entry.is_expired:
                 del self._entries[token]
-                logger.info(
-                    "MediaRegistry expired on read: token=%s...", token[:8]
-                )
+                self._retire_entry(entry)
+                logger.info("MediaRegistry entry expired on read")
                 return None
             entry.last_accessed = time.time()
             self._entries.move_to_end(token)
@@ -635,6 +642,24 @@ class MediaRegistry:
         async with self._lock:
             return self._cleanup_locked()
 
+    async def mark_sensitive(self, token: str) -> bool:
+        """Increase the sensitivity of an existing token without copying bytes."""
+        async with self._lock:
+            self._cleanup_locked()
+            entry = self._entries.get(token)
+            if entry is None:
+                return False
+            entry.sensitive = True
+            return True
+
+    async def close(self) -> None:
+        """Release relay-owned upload files on shutdown; leave caller files alone."""
+        async with self._lock:
+            self._entries.clear()
+            for path in list(self._owned_paths):
+                if self._unlink_owned_path(path):
+                    self._owned_paths.remove(path)
+
     async def list_all(
         self, *, include_expired: bool = False
     ) -> list[dict]:
@@ -698,7 +723,33 @@ class MediaRegistry:
         """Prune expired entries. Caller must hold ``self._lock``."""
         expired = [k for k, v in self._entries.items() if v.is_expired]
         for k in expired:
-            del self._entries[k]
+            self._retire_entry(self._entries.pop(k))
+        # A prior unlink can fail while another reader has the file open.
+        # Retry it on the periodic cleanup sweep after that reader exits.
+        for path in list(self._owned_paths):
+            if not any(entry.path == path for entry in self._entries.values()):
+                if self._unlink_owned_path(path):
+                    self._owned_paths.remove(path)
         if expired:
             logger.debug("MediaRegistry cleaned up %d expired entries", len(expired))
         return len(expired)
+
+    def _retire_entry(self, entry: _MediaEntry) -> None:
+        path = entry.path
+        if path not in self._owned_paths:
+            return
+        if any(active.path == path for active in self._entries.values()):
+            return
+        if self._unlink_owned_path(path):
+            self._owned_paths.remove(path)
+
+    @staticmethod
+    def _unlink_owned_path(path: str) -> bool:
+        try:
+            os.unlink(path)
+        except FileNotFoundError:
+            return True
+        except OSError:
+            logger.warning("Could not remove relay-owned media file")
+            return False
+        return True

plugin/relay/provider_usage.py

@@ -2,8 +2,8 @@
 
 Hermes already owns provider credentials and the canonical account-usage model.
 Relay reuses that model, adds credential-pool and balance structure for Android,
-and supplies the missing OpenCode Go adapter. Provider keys remain host-side
-and are never serialized into the response.
+and supplies the missing OpenCode Go and Grok subscription adapters. Provider
+keys remain host-side and are never serialized into the response.
 """
 
 from __future__ import annotations
@@ -24,9 +24,18 @@ RELAY_CAPABILITIES = (
     "credential_pools",
     "structured_balances",
     "opencode_go",
+    "supergrok",
 )
 _OPENCODE_GO_DEFAULT_BASE_URL = "https://opencode.ai/zen/go/v1"
 _OPENCODE_GO_USER_AGENT = "curl/8.4.0"
+# Grok subscription usage is only served by xAI's CLI proxy, not the public API.
+_SUPERGROK_IDENTITY_URL = "https://cli-chat-proxy.grok.com/v1/user"
+_SUPERGROK_BILLING_URL = "https://cli-chat-proxy.grok.com/v1/billing?format=credits"
+_SUPERGROK_PERIOD_TYPE_PREFIX = "USAGE_PERIOD_TYPE_"
+_SUPERGROK_MAX_PRODUCT_WINDOWS = 8
+_SUPERGROK_CENTS_PER_USD = 100.0
+_SUPERGROK_CAMEL_BOUNDARY = re.compile(r"(?<=[a-z0-9])(?=[A-Z])")
+_SUPERGROK_SLUG = re.compile(r"[^a-z0-9]+")
 _MAX_DETAIL_LENGTH = 240
 _PROFILE_ID = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
 
@@ -505,6 +514,242 @@ async def fetch_opencode_go_usage(
     }
 
 
+def _supergrok_percent(value: Any) -> float | None:
+    if not isinstance(value, (int, float)) or isinstance(value, bool):
+        return None
+    number = float(value)
+    if not math.isfinite(number):
+        return None
+    return max(0.0, min(100.0, number))
+
+
+def _supergrok_cents(value: Any) -> float | None:
+    """Unwrap xAI's ``{"val": <int cents>}`` money wrapper."""
+    if not isinstance(value, dict):
+        return None
+    cents = value.get("val")
+    if not isinstance(cents, (int, float)) or isinstance(cents, bool):
+        return None
+    amount = float(cents)
+    if not math.isfinite(amount) or amount < 0:
+        return None
+    return amount
+
+
+def _supergrok_period_label(period_type: Any) -> str:
+    raw = str(period_type or "").strip()
+    if raw.startswith(_SUPERGROK_PERIOD_TYPE_PREFIX):
+        raw = raw[len(_SUPERGROK_PERIOD_TYPE_PREFIX) :]
+    return raw.replace("_", " ").strip().title() or "Current period"
+
+
+def _supergrok_product_label(product: Any) -> str | None:
+    raw = _bounded_text(product, 60)
+    if raw is None:
+        return None
+    return _SUPERGROK_CAMEL_BOUNDARY.sub(" ", raw).strip() or raw
+
+
+def _supergrok_windows(config: dict[str, Any]) -> list[dict[str, Any]]:
+    period = config.get("currentPeriod")
+    period = period if isinstance(period, dict) else {}
+    reset_at = _iso(period.get("end")) or _iso(config.get("billingPeriodEnd"))
+
+    percent = _supergrok_percent(config.get("creditUsagePercent"))
+    if percent is None:
+        used = _supergrok_cents(config.get("used"))
+        limit = _supergrok_cents(config.get("monthlyLimit"))
+        if used is not None and limit is not None and limit > 0:
+            percent = max(0.0, min(100.0, (used / limit) * 100))
+
+    windows: list[dict[str, Any]] = []
+    if percent is not None:
+        windows.append(
+            {
+                "id": "period",
+                "label": _supergrok_period_label(period.get("type")),
+                "used_percent": percent,
+                "reset_at": reset_at,
+                "detail": None,
+            }
+        )
+    elif reset_at:
+        # A period with no recorded usage yet omits the figure entirely rather
+        # than reporting zero. The window is real, so surface it without
+        # inventing a percentage for it.
+        windows.append(
+            {
+                "id": "period",
+                "label": _supergrok_period_label(period.get("type")),
+                "used_percent": None,
+                "reset_at": reset_at,
+                "detail": "No usage reported yet",
+            }
+        )
+
+    products = config.get("productUsage")
+    if isinstance(products, list):
+        for item in products[:_SUPERGROK_MAX_PRODUCT_WINDOWS]:
+            if not isinstance(item, dict):
+                continue
+            product_percent = _supergrok_percent(item.get("usagePercent"))
+            label = _supergrok_product_label(item.get("product"))
+            if product_percent is None or label is None:
+                continue
+            windows.append(
+                {
+                    "id": f"product_{_SUPERGROK_SLUG.sub('_', label.lower()).strip('_')[:32]}",
+                    "label": label,
+                    "used_percent": product_percent,
+                    "reset_at": reset_at,
+                    "detail": None,
+                }
+            )
+    return windows
+
+
+def _supergrok_details(config: dict[str, Any], payload: dict[str, Any]) -> list[str]:
+    details: list[str] = []
+    cap = _supergrok_cents(config.get("onDemandCap"))
+    used = _supergrok_cents(config.get("onDemandUsed"))
+    enabled = payload.get("onDemandEnabled") is True or config.get("onDemandEnabled") is True
+    if enabled or (cap or 0) > 0 or (used or 0) > 0:
+        parts = [
+            part
+            for part in (
+                f"${used / _SUPERGROK_CENTS_PER_USD:.2f} used" if used is not None else None,
+                f"of ${cap / _SUPERGROK_CENTS_PER_USD:.2f}" if cap is not None else None,
+            )
+            if part is not None
+        ]
+        if parts:
+            details.append(f"On-demand: {' '.join(parts)}")
+        elif enabled:
+            details.append("On-demand enabled")
+    prepaid = _supergrok_cents(config.get("prepaidBalance"))
+    if prepaid:
+        details.append(f"Prepaid balance: ${prepaid / _SUPERGROK_CENTS_PER_USD:.2f}")
+    return details
+
+
+async def fetch_supergrok_usage(
+    *,
+    profile_home: Path | None = None,
+    session_factory: Callable[[], Any] = aiohttp.ClientSession,
+    credential_resolver: Callable[..., Any] | None = None,
+) -> dict[str, Any]:
+    """Fetch Grok subscription usage behind the ``xai-oauth`` sign-in.
+
+    Subscription windows are only served by xAI's CLI proxy, so this adapter
+    speaks the pinned ``cli-chat-proxy.grok.com`` contract: the
+    Hermes-managed ``xai-oauth`` bearer reads the account identity, then the
+    credits billing snapshot scoped to that identity.
+    """
+    home_token = _set_home(profile_home)
+    try:
+        if credential_resolver is None:
+            from hermes_cli.auth import resolve_xai_oauth_runtime_credentials
+
+            credential_resolver = resolve_xai_oauth_runtime_credentials
+
+        credentials = await asyncio.to_thread(credential_resolver)
+    except Exception as exc:
+        if getattr(exc, "code", None) == "xai_auth_missing":
+            return unavailable_provider("supergrok", "SuperGrok")
+        return unavailable_provider(
+            "supergrok",
+            "SuperGrok",
+            status="unavailable",
+            message="Could not resolve SuperGrok credentials",
+        )
+    finally:
+        _reset_home(home_token)
+
+    access_token = str((credentials or {}).get("api_key") or "").strip()
+    if not access_token:
+        return unavailable_provider("supergrok", "SuperGrok")
+
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "Accept": "application/json",
+    }
+    try:
+        async with session_factory() as session:
+            async with session.get(
+                _SUPERGROK_IDENTITY_URL,
+                headers=headers,
+                timeout=aiohttp.ClientTimeout(total=15),
+            ) as response:
+                if response.status != 200:
+                    return unavailable_provider(
+                        "supergrok",
+                        "SuperGrok",
+                        status="unavailable",
+                        message=f"Provider returned HTTP {response.status}",
+                    )
+                identity = await response.json()
+            user_id = (
+                str(identity.get("userId") or "").strip()
+                if isinstance(identity, dict)
+                else ""
+            )
+            if not user_id:
+                return unavailable_provider(
+                    "supergrok",
+                    "SuperGrok",
+                    status="unavailable",
+                    message="Provider returned no account identity",
+                )
+            async with session.get(
+                _SUPERGROK_BILLING_URL,
+                headers={**headers, "x-userid": user_id},
+                timeout=aiohttp.ClientTimeout(total=15),
+            ) as response:
+                if response.status != 200:
+                    return unavailable_provider(
+                        "supergrok",
+                        "SuperGrok",
+                        status="unavailable",
+                        message=f"Provider returned HTTP {response.status}",
+                    )
+                payload = await response.json()
+    except (aiohttp.ClientError, asyncio.TimeoutError, ValueError, TypeError):
+        return unavailable_provider(
+            "supergrok",
+            "SuperGrok",
+            status="unavailable",
+            message="Could not load SuperGrok usage",
+        )
+
+    config = payload.get("config") if isinstance(payload, dict) else None
+    if not isinstance(config, dict):
+        return unavailable_provider(
+            "supergrok",
+            "SuperGrok",
+            status="unavailable",
+            message="Provider returned an unsupported usage payload",
+        )
+    windows = _supergrok_windows(config)
+    if not windows:
+        return unavailable_provider(
+            "supergrok",
+            "SuperGrok",
+            status="unavailable",
+            message="Provider returned no usage windows",
+        )
+    return {
+        "id": "supergrok",
+        "display_name": "SuperGrok",
+        "status": "available",
+        "source": "provider_api",
+        "fetched_at": _now_iso(),
+        "plan": _bounded_text(payload.get("subscriptionTier"), 80),
+        "windows": windows,
+        "details": _supergrok_details(config, payload),
+        "message": None,
+    }
+
+
 async def collect_provider_usage(
     *,
     profile_home: Path | None = None,
@@ -513,6 +758,7 @@ async def collect_provider_usage(
     codex_fetcher: Callable[..., Awaitable[dict[str, Any]]] = fetch_codex_usage,
     nous_fetcher: Callable[[Path | None], Awaitable[dict[str, Any]]] = fetch_nous_usage,
     opencode_fetcher: Callable[..., Awaitable[dict[str, Any]]] = fetch_opencode_go_usage,
+    supergrok_fetcher: Callable[..., Awaitable[dict[str, Any]]] = fetch_supergrok_usage,
 ) -> dict[str, Any]:
     providers = await asyncio.gather(
         codex_fetcher(
@@ -522,6 +768,7 @@ async def collect_provider_usage(
         ),
         nous_fetcher(profile_home),
         opencode_fetcher(profile_home=profile_home),
+        supergrok_fetcher(profile_home=profile_home),
     )
     return {
         "schema_version": SCHEMA_VERSION,

plugin/relay/server.py

@@ -31,6 +31,7 @@ import logging
 import math
 import mimetypes
 import os
+import re
 import secrets
 import signal
 import socket
@@ -317,6 +318,7 @@ class RelayServer:
         await self.desktop.close()
         await self.tui.close()
         await self.proactive.close()
+        await self.media.close()
 
         # Close all WebSocket connections
         for ws in list(self._clients):
@@ -1757,6 +1759,7 @@ async def handle_media_register(request: web.Request) -> web.Response:
     # Model-emitted sensitivity hint — transported verbatim. Absent/falsey
     # → not sensitive (back-compat with older callers that never send it).
     sensitive = _coerce_sensitive(payload.get("sensitive"))
+    owned_file = payload.get("owned_file") is True
 
     server: RelayServer = request.app["server"]
     try:
@@ -1765,9 +1768,10 @@ async def handle_media_register(request: web.Request) -> web.Response:
             content_type=content_type,
             file_name=file_name,
             sensitive=sensitive,
+            owned_file=owned_file,
         )
     except MediaRegistrationError as exc:
-        logger.info("Media registration rejected: %s", exc)
+        logger.info("Media registration rejected")
         return web.json_response(
             {"ok": False, "error": str(exc)}, status=400
         )
@@ -1901,23 +1905,22 @@ async def handle_media_upload(request: web.Request) -> web.Response:
             path=tmp.name,
             content_type=file_content_type,
             file_name=base_name,
+            owned_file=True,
         )
     except MediaRegistrationError as exc:
         try:
             os.unlink(tmp.name)
         except OSError:
             pass
-        logger.info("Media upload registration rejected: %s", exc)
+        logger.info("Media upload registration rejected")
         return web.json_response(
             {"ok": False, "error": str(exc)}, status=400
         )
 
     logger.info(
-        "Media uploaded: token=%s... bytes=%d type=%s file=%s",
-        entry.token[:8],
+        "Media uploaded: bytes=%d type=%s",
         bytes_written,
         file_content_type,
-        base_name,
     )
     return web.json_response(
         {
@@ -1988,17 +1991,22 @@ async def handle_media_get(request: web.Request) -> web.StreamResponse:
     else:
         headers["Content-Disposition"] = "inline"
 
-    logger.debug(
-        "Serving media token=%s... path=%s size=%d sensitive=%s to session=%s...",
-        token[:8],
-        entry.path,
-        entry.size,
-        entry.sensitive,
-        bearer[:8],
-    )
+    logger.debug("Serving media size=%d sensitive=%s", entry.size, entry.sensitive)
     return web.FileResponse(entry.path, headers=headers)
 
 
+async def handle_media_mark_sensitive(request: web.Request) -> web.Response:
+    """Allow a host-local screenshot tool to add the private-media hint."""
+    _require_loopback(request)
+    server: RelayServer = request.app["server"]
+    token = request.match_info["token"]
+    if not re.fullmatch(r"[A-Za-z0-9_-]{16,128}", token):
+        raise web.HTTPNotFound()
+    if not await server.media.mark_sensitive(token):
+        raise web.HTTPNotFound()
+    return web.json_response({"ok": True})
+
+
 async def handle_media_by_path(request: web.Request) -> web.StreamResponse:
     """Serve a media file by absolute path for LLM-emitted ``MEDIA:/abs/path`` markers.
 
@@ -2098,12 +2106,12 @@ async def handle_media_by_path(request: web.Request) -> web.StreamResponse:
         # retrying) from sandbox violations (phone should mark FAILED with
         # a different error). Both are non-retryable.
         if "does not exist" in msg or "not a regular file" in msg:
-            logger.info("Media by-path: not found — %s", msg)
+            logger.info("Media by-path: not found")
             raise web.HTTPNotFound(text=msg)
         # Everything else — not absolute, outside allowed root, too large,
         # bad stat — is a sandbox or policy violation. 403 signals this
         # distinctly from 401 (bad auth) and 400 (malformed request).
-        logger.info("Media by-path: sandbox violation — %s", msg)
+        logger.info("Media by-path: sandbox violation")
         raise web.HTTPForbidden(text=msg)
 
     # Content type: honor phone-provided hint if given; otherwise guess.
@@ -2130,12 +2138,10 @@ async def handle_media_by_path(request: web.Request) -> web.StreamResponse:
         headers["X-Media-Sensitive"] = "1"
 
     logger.debug(
-        "Serving media by-path %s size=%d type=%s sensitive=%s to session=%s...",
-        real_path,
+        "Serving media by-path size=%d type=%s sensitive=%s",
         size,
         content_type,
         sensitive,
-        bearer[:8],
     )
     return web.FileResponse(real_path, headers=headers)
 
@@ -4263,7 +4269,7 @@ async def handle_ws(request: web.Request) -> web.WebSocketResponse:
 
     server._clients[ws] = session_token
     server._client_tasks[ws] = set()
-    logger.info("Client authenticated from %s (token=%s...)", remote_ip, session_token[:8])
+    logger.info("Client authenticated from %s", remote_ip)
 
     try:
         async for msg in ws:
@@ -4926,6 +4932,7 @@ def create_app(config: RelayConfig) -> web.Application:
     app.router.add_post("/media/register", handle_media_register)
     # === PHASE3-bridge-server-followup: /media/upload ===
     app.router.add_post("/media/upload", handle_media_upload)
+    app.router.add_post("/media/{token}/sensitive", handle_media_mark_sensitive)
     # === END PHASE3-bridge-server-followup ===
     # Order matters: the fixed-path "/media/by-path" route must be declared
     # before the wildcard "/media/{token}" route or aiohttp will swallow
@@ -5184,17 +5191,29 @@ async def _on_app_startup(app: web.Application) -> None:
         _profile_rescan_loop(app), name="profile-rescan-loop"
     )
     app["_profile_rescan_task"] = task
+    app["_media_cleanup_task"] = asyncio.create_task(
+        _media_cleanup_loop(app), name="media-cleanup-loop"
+    )
+
+
+async def _media_cleanup_loop(app: web.Application) -> None:
+    """Retire expired relay-owned uploads even when no media is requested."""
+    server: RelayServer = app["server"]
+    while True:
+        await asyncio.sleep(300)
+        await server.media.cleanup()
 
 
 async def _on_app_cleanup(app: web.Application) -> None:
     """Cancel background tasks cleanly."""
-    task = app.get("_profile_rescan_task")
-    if task is not None and not task.done():
-        task.cancel()
-        try:
-            await task
-        except (asyncio.CancelledError, Exception):
-            pass
+    for key in ("_profile_rescan_task", "_media_cleanup_task"):
+        task = app.get(key)
+        if task is not None and not task.done():
+            task.cancel()
+            try:
+                await task
+            except (asyncio.CancelledError, Exception):
+                pass
 
 
 async def _on_app_shutdown(app: web.Application) -> None:
@@ -5465,5 +5484,6 @@ def main() -> None:
         host=config.host,
         port=config.port,
         ssl_context=ssl_ctx,
+        access_log=None,  # Media URLs carry private tokens or absolute paths.
         print=None,  # Suppress aiohttp's default startup banner
     )

plugin/tests/test_android_navigate.py

@@ -282,6 +282,27 @@ class TestSharedBridgeTransport(unittest.TestCase):
         response.raise_for_status.assert_called_once_with()
 
 
+class TestNavigateScreenshot(unittest.TestCase):
+    def test_token_response_writes_fetched_png_for_vision(self) -> None:
+        png = b"\x89PNG\r\n\x1a\nvision-bytes"
+        with mock.patch.object(nav, "_get", return_value={
+            "media": "MEDIA:hermes-relay://navigate-token-123456"
+        }), mock.patch.object(nav, "_bridge_request") as request:
+            response = request.return_value
+            response.status_code = 200
+            response.headers = {"Content-Type": "image/png"}
+            response.iter_content.return_value = iter([png])
+            shot = nav._capture_screenshot()
+        try:
+            self.assertEqual(Path(shot.local_path).read_bytes(), png)
+            self.assertEqual(shot.token, "hermes-relay://navigate-token-123456")
+            request.assert_called_once_with(
+                "GET", "/media/navigate-token-123456", timeout=nav._timeout(), stream=True
+            )
+        finally:
+            Path(shot.local_path).unlink(missing_ok=True)
+
+
 class TestNavigateLoop(unittest.TestCase):
     def setUp(self) -> None:
         # Belt-and-suspenders: make sure the stub env var never leaks

plugin/tests/test_android_screenshot_media.py

@@ -0,0 +1,74 @@
+"""Screenshot token retrieval and legacy payload safety checks."""
+
+import base64
+from unittest import mock
+
+import pytest
+import requests
+
+from plugin.tools.android_screenshot_media import (
+    MAX_SCREENSHOT_BYTES,
+    ScreenshotMediaError,
+    resolve_screenshot,
+)
+
+
+PNG = b"\x89PNG\r\n\x1a\nimage-bytes"
+TOKEN = "MEDIA:hermes-relay://valid-token-123456"
+
+
+def response(status=200, body=PNG, content_type="image/png", length=None):
+    value = mock.Mock(spec=requests.Response)
+    value.status_code = status
+    value.headers = {"Content-Type": content_type}
+    if length is not None:
+        value.headers["Content-Length"] = str(length)
+    value.iter_content.return_value = iter([body])
+    return value
+
+
+def test_token_bytes_and_authenticated_route():
+    fetched = response()
+    request = mock.Mock(return_value=fetched)
+    assert resolve_screenshot({"data": {"media": TOKEN}}, request, 3) == (PNG, "image/png", TOKEN)
+    request.assert_called_once_with("GET", "/media/valid-token-123456", timeout=3, stream=True)
+    fetched.close.assert_called_once_with()
+
+
+@pytest.mark.parametrize("status", [401, 403, 404, 500])
+def test_fetch_failure_is_safe(status):
+    fetched = response(status=status)
+    with pytest.raises(ScreenshotMediaError, match="unavailable or access denied") as error:
+        resolve_screenshot({"media": TOKEN}, mock.Mock(return_value=fetched), 3)
+    assert "valid-token" not in str(error.value)
+    fetched.close.assert_called_once_with()
+
+
+@pytest.mark.parametrize("marker", ["", "MEDIA:hermes-relay://../x", "MEDIA:https://other/x", 5])
+def test_invalid_marker_does_not_fetch(marker):
+    request = mock.Mock()
+    with pytest.raises(ScreenshotMediaError, match="Invalid screenshot media token"):
+        resolve_screenshot({"media": marker}, request, 3)
+    request.assert_not_called()
+
+
+def test_stream_cap_and_type_check():
+    fetched = response(body=PNG, length=MAX_SCREENSHOT_BYTES + 1)
+    with pytest.raises(ScreenshotMediaError, match="size limit"):
+        resolve_screenshot({"media": TOKEN}, mock.Mock(return_value=fetched), 3)
+    fetched = response(body=PNG, content_type="image/jpeg")
+    with pytest.raises(ScreenshotMediaError, match="type mismatch"):
+        resolve_screenshot({"media": TOKEN}, mock.Mock(return_value=fetched), 3)
+    fetched = response(body=b"x" * (MAX_SCREENSHOT_BYTES + 1))
+    with pytest.raises(ScreenshotMediaError, match="size limit"):
+        resolve_screenshot({"media": TOKEN}, mock.Mock(return_value=fetched), 3)
+
+
+def test_legacy_inline_is_bounded_and_validated():
+    request = mock.Mock()
+    assert resolve_screenshot({"image": base64.b64encode(PNG).decode()}, request, 3) == (PNG, "image/png", None)
+    request.assert_not_called()
+    with pytest.raises(ScreenshotMediaError, match="Invalid screenshot image data"):
+        resolve_screenshot({"image": "%%%"}, request, 3)
+    with pytest.raises(ScreenshotMediaError, match="size limit"):
+        resolve_screenshot({"image": "A" * (MAX_SCREENSHOT_BYTES * 2)}, request, 3)

plugin/tests/test_android_tool.py

@@ -1,3 +1,4 @@
+import base64
 import json
 import os
 import sys
@@ -241,16 +242,70 @@ class TestPressKey:
 
 class TestScreenshot:
     @responses.activate
-    def test_screenshot(self, bridge_url):
+    def test_screenshot_token_delivers_png_bytes(self, bridge_url):
+        png = b"\x89PNG\r\n\x1a\nreal-image-bytes"
         responses.add(
             responses.GET,
             f"{bridge_url}/screenshot",
-            json={"image": "aGVsbG8=", "width": 1080, "height": 1920},
+            json={"media": "MEDIA:hermes-relay://shot-token-123456"},
         )
-        with mock.patch("plugin.relay.client.register_media", return_value="shot-token"):
-            result = android_screenshot()
-        assert "Screenshot captured (1080x1920)" in result
-        assert "MEDIA:hermes-relay://shot-token" in result
+        responses.add(
+            responses.GET, f"{bridge_url}/media/shot-token-123456",
+            body=png, content_type="image/png",
+        )
+        result = android_screenshot()
+        assert result["_multimodal"] is True
+        assert "MEDIA:hermes-relay://shot-token-123456" in result["text_summary"]
+        assert result["content"][1]["image_url"]["url"] == (
+            "data:image/png;base64," + base64.b64encode(png).decode("ascii")
+        )
+        via_registry = _HANDLERS["android_screenshot"]({})
+        assert via_registry["_multimodal"] is True
+
+    @responses.activate
+    def test_screenshot_legacy_inline(self, bridge_url):
+        png = b"\x89PNG\r\n\x1a\nold-image"
+        responses.add(responses.GET, f"{bridge_url}/screenshot",
+                      json={"data": {"image": base64.b64encode(png).decode("ascii")}})
+        result = android_screenshot()
+        assert result["content"][1]["image_url"]["url"].endswith(
+            base64.b64encode(png).decode("ascii")
+        )
+
+    @responses.activate
+    def test_sensitive_screenshot_registers_private_media(self, bridge_url):
+        png = b"\x89PNG\r\n\x1a\nsensitive"
+        responses.add(responses.GET, f"{bridge_url}/screenshot",
+                      json={"media": "MEDIA:hermes-relay://old-token-123456"})
+        responses.add(responses.GET, f"{bridge_url}/media/old-token-123456",
+                      body=png, content_type="image/png")
+        with mock.patch("plugin.relay.client.mark_media_sensitive", return_value=True) as mark:
+            result = android_screenshot(sensitive=True)
+        mark.assert_called_once_with("old-token-123456")
+        assert "MEDIA:hermes-relay://old-token-123456" in result["text_summary"]
+
+    @responses.activate
+    def test_sensitive_legacy_screenshot_registers_managed_file(self, bridge_url):
+        png = b"\x89PNG\r\n\x1a\nlegacy-private"
+        responses.add(responses.GET, f"{bridge_url}/screenshot",
+                      json={"image": base64.b64encode(png).decode("ascii")})
+        with mock.patch("plugin.relay.client.register_media", return_value="private-token") as register:
+            result = android_screenshot(sensitive=True)
+        path = register.call_args.args[0]
+        try:
+            assert Path(path).read_bytes() == png
+            assert register.call_args.kwargs["sensitive"] is True
+            assert register.call_args.kwargs["owned_file"] is True
+            assert "MEDIA:hermes-relay://private-token" in result["text_summary"]
+        finally:
+            Path(path).unlink(missing_ok=True)
+
+    @responses.activate
+    def test_screenshot_rejects_invalid_token(self, bridge_url):
+        responses.add(responses.GET, f"{bridge_url}/screenshot",
+                      json={"media": "MEDIA:hermes-relay://../other"})
+        assert android_screenshot() == {"error": "Screenshot unavailable"}
+        assert len(responses.calls) == 1
 
 
 class TestScroll:

plugin/tests/test_desktop_computer_use.py

@@ -7,9 +7,13 @@ smoke path.
 
 from __future__ import annotations
 
+import base64
 import json
+import struct
 import unittest
+import zlib
 from typing import Any
+from unittest import mock
 
 from plugin.tools import desktop_tool
 
@@ -23,6 +27,17 @@ COMPUTER_TOOLS = [
 ]
 
 
+def one_pixel_png() -> bytes:
+    def chunk(kind: bytes, data: bytes) -> bytes:
+        return (struct.pack(">I", len(data)) + kind + data +
+                struct.pack(">I", zlib.crc32(kind + data)))
+
+    header = struct.pack(">IIBBBBB", 1, 1, 8, 6, 0, 0, 0)
+    return (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", header) +
+            chunk(b"IDAT", zlib.compress(b"\x00\xff\x00\x00\xff")) +
+            chunk(b"IEND", b""))
+
+
 class DesktopComputerUseToolTests(unittest.TestCase):
     def test_all_computer_tools_have_schema_and_handler(self) -> None:
         for name in COMPUTER_TOOLS:
@@ -114,6 +129,50 @@ class DesktopComputerUseToolTests(unittest.TestCase):
             ],
         )
 
+    def test_screenshot_attaches_actual_png_for_host_vision(self) -> None:
+        png = one_pixel_png()
+        encoded = base64.b64encode(png).decode("ascii")
+        result = {"ok": True, "request_id": "request-1", "result": {
+            "ok": True, "bytes_base64": encoded, "size_bytes": len(png),
+            "display": {"width": 800, "height": 600},
+        }}
+        with mock.patch.object(desktop_tool, "_post", return_value=result):
+            output = desktop_tool._HANDLERS["desktop_computer_screenshot"]({})
+        self.assertTrue(output["_multimodal"])
+        self.assertEqual(output["content"][1]["image_url"]["url"],
+                         f"data:image/png;base64,{encoded}")
+        self.assertNotIn(encoded, output["content"][0]["text"])
+        self.assertIn('"width": 800', output["content"][0]["text"])
+        self.assertIn('"request_id": "request-1"', output["content"][0]["text"])
+
+    def test_screenshot_keeps_saved_path_and_rejects_bad_bytes(self) -> None:
+        with mock.patch.object(desktop_tool, "_post", return_value={
+            "ok": True, "saved_path": "/tmp/shot.png", "size_bytes": 10,
+        }):
+            self.assertEqual(json.loads(desktop_tool.desktop_computer_screenshot())["saved_path"],
+                             "/tmp/shot.png")
+        with mock.patch.object(desktop_tool, "_post", return_value={
+            "ok": True, "bytes_base64": "%%%",
+        }):
+            self.assertIn("error", json.loads(desktop_tool.desktop_computer_screenshot()))
+
+    def test_cua_screenshot_preserves_element_metadata_with_image(self) -> None:
+        jpeg = b"\xff\xd8\xff\xe0private-window"
+        encoded = base64.b64encode(jpeg).decode("ascii")
+        with mock.patch.object(desktop_tool, "_post", return_value={
+            "ok": True, "result": {
+                "backend": "cua_driver",
+                "elements": [{"snapshot_token": "one-use-token"}],
+                "screenshot_base64": encoded,
+                "screenshot_mime_type": "image/jpeg",
+            },
+        }):
+            output = desktop_tool.desktop_computer_screenshot(pid=1, window_id=2)
+        self.assertEqual(output["content"][1]["image_url"]["url"],
+                         f"data:image/jpeg;base64,{encoded}")
+        self.assertIn("one-use-token", output["content"][0]["text"])
+        self.assertNotIn(encoded, output["content"][0]["text"])
+
     def test_grant_request_schema_only_requires_mode(self) -> None:
         schema = desktop_tool._SCHEMAS["desktop_computer_grant_request"]
         self.assertEqual(schema["parameters"]["required"], ["mode"])

plugin/tests/test_desktop_multidevice.py

@@ -10,6 +10,7 @@ from typing import Any
 from unittest.mock import AsyncMock, Mock, patch
 
 from plugin.relay.channels.desktop import (
+    DesktopCommandRecord,
     DesktopError,
     DesktopHandler,
     DesktopRequesterContext,
@@ -66,6 +67,18 @@ async def _register_two() -> tuple[DesktopHandler, _FakeWs, _FakeWs]:
 
 
 class DesktopMultiDeviceTests(unittest.IsolatedAsyncioTestCase):
+    async def test_screenshot_bytes_never_enter_activity_summary(self) -> None:
+        handler = DesktopHandler()
+        handler.recent_commands.append(DesktopCommandRecord(
+            request_id="req-1", tool="desktop_computer_screenshot"
+        ))
+        handler._update_record_from_response("req-1", {
+            "status": 200,
+            "result": {"bytes_base64": "private-pixels", "saved_path": "/private/shot.png"},
+        })
+        summary = handler.get_recent()[0]["result_summary"]
+        self.assertEqual(summary, "Desktop screenshot response received")
+
     async def test_tool_schema_exposes_script_and_device_selector(self) -> None:
         parameters = desktop_tool._SCHEMAS["desktop_powershell"]["parameters"]
         self.assertEqual(parameters["required"], ["script"])

plugin/tests/test_media_registry.py

@@ -9,6 +9,7 @@ stdlib.
 from __future__ import annotations
 
 import asyncio
+import logging
 import os
 import shutil
 import tempfile
@@ -18,6 +19,20 @@ import unittest
 from plugin.relay.media import MediaRegistrationError, MediaRegistry, _MediaEntry
 
 
+def test_registry_logs_omit_tokens_and_media_paths(caplog) -> None:
+    async def run() -> tuple[str, str]:
+        with tempfile.TemporaryDirectory() as root:
+            path = _write_file(root, "private-image.png")
+            registry = MediaRegistry(allowed_roots=[root])
+            entry = await registry.register(path, "image/png")
+            return path, entry.token
+
+    caplog.set_level(logging.INFO, logger="hermes_relay.media")
+    path, token = asyncio.run(run())
+    assert path not in caplog.text
+    assert token[:8] not in caplog.text
+
+
 # ── Helpers ─────────────────────────────────────────────────────────────────
 
 
@@ -97,6 +112,78 @@ class MediaRegistryTests(unittest.IsolatedAsyncioTestCase):
         # And the expired entry has been pruned.
         self.assertEqual(await registry.size(), 0)
 
+    async def test_owned_upload_is_deleted_on_expiry_but_caller_file_is_preserved(self) -> None:
+        registry = MediaRegistry()
+        with tempfile.NamedTemporaryFile(prefix="hermes-relay-upload-", delete=False) as upload:
+            upload.write(b"private-image")
+            owned_path = upload.name
+        caller_path = _write_file(self._sandbox, "caller.png")
+        try:
+            owned = await registry.register(owned_path, "image/png", owned_file=True)
+            caller = await registry.register(caller_path, "image/png")
+            async with registry._lock:
+                registry._entries[owned.token].expires_at = time.time() - 1
+                registry._entries[caller.token].expires_at = time.time() - 1
+            self.assertEqual(await registry.cleanup(), 2)
+            self.assertFalse(os.path.exists(owned_path))
+            self.assertTrue(os.path.exists(caller_path))
+        finally:
+            if os.path.exists(owned_path):
+                os.unlink(owned_path)
+
+    async def test_owned_upload_is_deleted_on_close(self) -> None:
+        registry = MediaRegistry()
+        with tempfile.NamedTemporaryFile(prefix="hermes-relay-upload-", delete=False) as upload:
+            upload.write(b"private-image")
+            owned_path = upload.name
+        try:
+            await registry.register(owned_path, "image/png", owned_file=True)
+            await registry.close()
+            self.assertFalse(os.path.exists(owned_path))
+        finally:
+            if os.path.exists(owned_path):
+                os.unlink(owned_path)
+
+    async def test_owned_upload_is_deleted_on_lru_eviction(self) -> None:
+        registry = MediaRegistry(max_entries=1)
+        with tempfile.NamedTemporaryFile(prefix="hermes-relay-upload-", delete=False) as upload:
+            upload.write(b"private-image")
+            owned_path = upload.name
+        try:
+            await registry.register(owned_path, "image/png", owned_file=True)
+            await registry.register(_write_file(self._sandbox, "keep.png"), "image/png")
+            self.assertFalse(os.path.exists(owned_path))
+        finally:
+            if os.path.exists(owned_path):
+                os.unlink(owned_path)
+
+    async def test_owned_file_survives_while_another_token_references_it(self) -> None:
+        registry = MediaRegistry()
+        with tempfile.NamedTemporaryFile(prefix="hermes-relay-upload-", delete=False) as upload:
+            upload.write(b"shared-image")
+            owned_path = upload.name
+        try:
+            owned = await registry.register(owned_path, "image/png", owned_file=True)
+            other = await registry.register(owned_path, "image/png")
+            async with registry._lock:
+                registry._entries[owned.token].expires_at = time.time() - 1
+            await registry.cleanup()
+            self.assertTrue(os.path.exists(owned_path))
+            async with registry._lock:
+                registry._entries[other.token].expires_at = time.time() - 1
+            await registry.cleanup()
+            self.assertFalse(os.path.exists(owned_path))
+        finally:
+            if os.path.exists(owned_path):
+                os.unlink(owned_path)
+
+    async def test_owned_file_cannot_delete_arbitrary_registered_path(self) -> None:
+        registry = _make_registry(self._sandbox)
+        path = _write_file(self._sandbox, "keep.png")
+        with self.assertRaisesRegex(MediaRegistrationError, "managed upload path"):
+            await registry.register(path, "image/png", owned_file=True)
+        self.assertTrue(os.path.exists(path))
+
     # ── LRU eviction ────────────────────────────────────────────────────
 
     async def test_lru_eviction_when_cap_exceeded(self) -> None:

plugin/tests/test_provider_usage.py

@@ -18,6 +18,7 @@ from plugin.relay.provider_usage import (
     fetch_codex_usage,
     fetch_nous_usage,
     fetch_opencode_go_usage,
+    fetch_supergrok_usage,
     resolve_profile_home,
     serialize_account_snapshot,
     unavailable_provider,
@@ -57,12 +58,34 @@ class _FakeSession:
         return self.response
 
 
+class _SequencedSession:
+    """Yield queued responses in call order, like aiohttp's request context manager."""
+
+    def __init__(self, responses: list[_FakeResponse]):
+        self._responses = list(responses)
+        self.calls: list[dict] = []
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    def get(self, url, *, headers=None, timeout=None):
+        self.calls.append({"url": url, "headers": headers or {}})
+        if not self._responses:
+            raise AssertionError("unexpected extra provider request")
+        return self._responses.pop(0)
+
+
 class ProviderUsageModelTests(unittest.IsolatedAsyncioTestCase):
     def test_profile_home_is_exact_and_rejects_traversal(self) -> None:
         with tempfile.TemporaryDirectory() as raw:
-            root = Path(raw)
+            # Resolve the temp root so macOS /var -> /private/var matches
+            # Path.resolve() inside resolve_profile_home.
+            root = Path(raw).resolve()
             (root / "config.yaml").write_text("model: {}\n", encoding="utf-8")
-            victor = root / "profiles" / "victor"
+            victor = (root / "profiles" / "victor").resolve()
             victor.mkdir(parents=True)
             (victor / "config.yaml").write_text("model: {}\n", encoding="utf-8")
             self.assertEqual(resolve_profile_home(str(root / "config.yaml"), "Victor"), victor)
@@ -159,6 +182,173 @@ class ProviderUsageModelTests(unittest.IsolatedAsyncioTestCase):
         self.assertNotIn("limits", result)
         self.assertEqual(fake.headers["Authorization"], "Bearer secret")
 
+    async def test_supergrok_without_oauth_is_not_configured(self) -> None:
+        result = await fetch_supergrok_usage(credential_resolver=lambda: {})
+
+        self.assertEqual(result["id"], "supergrok")
+        self.assertEqual(result["status"], "not_configured")
+        self.assertEqual(result["windows"], [])
+
+    async def test_supergrok_missing_oauth_state_is_not_configured(self) -> None:
+        class MissingOAuthState(Exception):
+            code = "xai_auth_missing"
+
+        def resolve_credentials() -> dict:
+            raise MissingOAuthState("No credentials stored")
+
+        result = await fetch_supergrok_usage(credential_resolver=resolve_credentials)
+
+        self.assertEqual(result["status"], "not_configured")
+
+    async def test_supergrok_oauth_refresh_failure_is_unavailable(self) -> None:
+        class RefreshFailure(Exception):
+            code = "xai_refresh_failed"
+
+        def resolve_credentials() -> dict:
+            raise RefreshFailure("private token details")
+
+        result = await fetch_supergrok_usage(credential_resolver=resolve_credentials)
+
+        self.assertEqual(result["status"], "unavailable")
+        self.assertEqual(result["message"], "Could not resolve SuperGrok credentials")
+        self.assertNotIn("private token details", str(result))
+
+    async def test_supergrok_maps_subscription_and_product_windows(self) -> None:
+        session = _SequencedSession(
+            [
+                _FakeResponse(payload={"userId": "user-1"}),
+                _FakeResponse(
+                    payload={
+                        "subscriptionTier": "SuperGrok",
+                        "onDemandEnabled": True,
+                        "config": {
+                            "creditUsagePercent": 14,
+                            "currentPeriod": {
+                                "type": "USAGE_PERIOD_TYPE_WEEKLY",
+                                "start": "2026-09-06T08:34:12.348291+00:00",
+                                "end": "2026-09-13T08:34:12.348291+00:00",
+                            },
+                            "productUsage": [
+                                {"product": "GrokBuild", "usagePercent": 11},
+                                {"product": "GrokImagine", "usagePercent": 2},
+                                {"product": "GrokChat", "usagePercent": None},
+                            ],
+                            "onDemandCap": {"val": 500},
+                            "onDemandUsed": {"val": 125},
+                            "prepaidBalance": {"val": 0},
+                        },
+                    }
+                ),
+            ]
+        )
+
+        result = await fetch_supergrok_usage(
+            session_factory=lambda: session,
+            credential_resolver=lambda: {"api_key": "secret"},
+        )
+
+        self.assertEqual(result["status"], "available")
+        self.assertEqual(result["source"], "provider_api")
+        self.assertEqual(result["plan"], "SuperGrok")
+        self.assertEqual(
+            [row["id"] for row in result["windows"]],
+            ["period", "product_grok_build", "product_grok_imagine"],
+        )
+        self.assertEqual(result["windows"][0]["label"], "Weekly")
+        self.assertEqual(result["windows"][0]["used_percent"], 14.0)
+        self.assertEqual(result["windows"][0]["reset_at"], "2026-09-13T08:34:12.348291+00:00")
+        self.assertEqual(result["windows"][1]["label"], "Grok Build")
+        self.assertEqual(result["windows"][1]["used_percent"], 11.0)
+        self.assertEqual(result["details"], ["On-demand: $1.25 used of $5.00"])
+        self.assertEqual(session.calls[0]["headers"]["Authorization"], "Bearer secret")
+        self.assertEqual(session.calls[1]["headers"]["x-userid"], "user-1")
+        self.assertIn("/billing?format=credits", session.calls[1]["url"])
+        self.assertNotIn("secret", str(result))
+
+    async def test_supergrok_stops_before_billing_without_account_identity(self) -> None:
+        session = _SequencedSession([_FakeResponse(payload={"userId": ""})])
+
+        result = await fetch_supergrok_usage(
+            session_factory=lambda: session,
+            credential_resolver=lambda: {"api_key": "secret"},
+        )
+
+        self.assertEqual(result["status"], "unavailable")
+        self.assertEqual(len(session.calls), 1)
+        self.assertNotIn("secret", str(result))
+
+    async def test_supergrok_reports_top_level_on_demand_state_without_amounts(self) -> None:
+        session = _SequencedSession(
+            [
+                _FakeResponse(payload={"userId": "user-1"}),
+                _FakeResponse(
+                    payload={
+                        "onDemandEnabled": True,
+                        "config": {
+                            "creditUsagePercent": 0,
+                            "currentPeriod": {"type": "USAGE_PERIOD_TYPE_WEEKLY"},
+                        },
+                    }
+                ),
+            ]
+        )
+
+        result = await fetch_supergrok_usage(
+            session_factory=lambda: session,
+            credential_resolver=lambda: {"api_key": "secret"},
+        )
+
+        self.assertEqual(result["status"], "available")
+        self.assertEqual(result["details"], ["On-demand enabled"])
+
+    async def test_supergrok_fresh_period_surfaces_window_without_inventing_a_percent(self) -> None:
+        session = _SequencedSession(
+            [
+                _FakeResponse(payload={"userId": "user-1"}),
+                _FakeResponse(
+                    payload={
+                        "config": {
+                            "currentPeriod": {
+                                "type": "USAGE_PERIOD_TYPE_WEEKLY",
+                                "start": "2026-09-13T08:34:12.348291+00:00",
+                                "end": "2026-09-20T08:34:12.348291+00:00",
+                            },
+                            "billingPeriodEnd": "2026-09-20T08:34:12.348291+00:00",
+                        }
+                    }
+                ),
+            ]
+        )
+
+        result = await fetch_supergrok_usage(
+            session_factory=lambda: session,
+            credential_resolver=lambda: {"api_key": "secret"},
+        )
+
+        self.assertEqual(result["status"], "available")
+        self.assertEqual(len(result["windows"]), 1)
+        self.assertEqual(result["windows"][0]["label"], "Weekly")
+        self.assertIsNone(result["windows"][0]["used_percent"])
+        self.assertEqual(result["windows"][0]["reset_at"], "2026-09-20T08:34:12.348291+00:00")
+        self.assertEqual(result["windows"][0]["detail"], "No usage reported yet")
+
+    async def test_supergrok_unusable_payload_is_unavailable(self) -> None:
+        session = _SequencedSession(
+            [
+                _FakeResponse(payload={"userId": "user-1"}),
+                _FakeResponse(payload={"config": {"isUnifiedBillingUser": True}}),
+            ]
+        )
+
+        result = await fetch_supergrok_usage(
+            session_factory=lambda: session,
+            credential_resolver=lambda: {"api_key": "secret"},
+        )
+
+        self.assertEqual(result["status"], "unavailable")
+        self.assertEqual(result["windows"], [])
+        self.assertEqual(result["message"], "Provider returned no usage windows")
+
     async def test_collection_keeps_provider_order_and_schema(self) -> None:
         async def codex(_home, **_kwargs):
             return unavailable_provider("openai-codex", "Codex")
@@ -169,19 +359,23 @@ class ProviderUsageModelTests(unittest.IsolatedAsyncioTestCase):
         async def opencode(*, profile_home=None):
             return unavailable_provider("opencode-go", "OpenCode Go")
 
+        async def supergrok(*, profile_home=None):
+            return unavailable_provider("supergrok", "SuperGrok")
+
         result = await collect_provider_usage(
             codex_fetcher=codex,
             nous_fetcher=nous,
             opencode_fetcher=opencode,
+            supergrok_fetcher=supergrok,
         )
         self.assertEqual(result["schema_version"], 2)
         self.assertEqual(
             result["capabilities"],
-            ["credential_pools", "structured_balances", "opencode_go"],
+            ["credential_pools", "structured_balances", "opencode_go", "supergrok"],
         )
         self.assertEqual(
             [row["id"] for row in result["providers"]],
-            ["openai-codex", "nous", "opencode-go"],
+            ["openai-codex", "nous", "opencode-go", "supergrok"],
         )
 
     async def test_codex_pool_marks_exact_live_session_credential_active(self) -> None:

plugin/tests/test_relay_media_routes.py

@@ -19,7 +19,7 @@ import time
 import unittest
 from unittest import mock
 
-from aiohttp import web
+from aiohttp import FormData, web
 from aiohttp.test_utils import AioHTTPTestCase
 
 from plugin.relay import media
@@ -186,6 +186,50 @@ class RelayMediaRoutesTests(AioHTTPTestCase):
         body = await resp.read()
         self.assertEqual(body, contents)
 
+    async def test_mark_sensitive_requires_loopback_and_preserves_image_bytes(self) -> None:
+        contents = b"\x89PNG\r\n\x1a\nprivate-image"
+        path = _write_file(self._sandbox, "private.png", content=contents)
+        entry = await self._server().media.register(path, "image/png")
+        response = await self.client.post(f"/media/{entry.token}/sensitive")
+        self.assertEqual(response.status, 200)
+        bearer = await self._create_session_token()
+        fetched = await self.client.get(
+            f"/media/{entry.token}", headers={"Authorization": f"Bearer {bearer}"}
+        )
+        self.assertEqual(fetched.headers.get("X-Media-Sensitive"), "1")
+        self.assertEqual(await fetched.read(), contents)
+        missing = await self.client.post("/media/missing-token-123456/sensitive")
+        self.assertEqual(missing.status, 404)
+        from plugin.relay.server import handle_media_mark_sensitive
+
+        forged = mock.Mock()
+        forged.remote = "203.0.113.10"
+        forged.app = self.app
+        forged.match_info = {"token": entry.token}
+        with self.assertRaises(web.HTTPForbidden):
+            await handle_media_mark_sensitive(forged)
+
+    async def test_uploaded_file_is_owned_and_deleted_when_token_expires(self) -> None:
+        bearer = await self._create_session_token()
+        form = FormData()
+        form.add_field("file", b"\x89PNG\r\n\x1a\nprivate-image",
+                       filename="capture.png", content_type="image/png")
+        self._server().media.allowed_roots.append(os.path.realpath(tempfile.gettempdir()))
+        response = await self.client.post(
+            "/media/upload", data=form,
+            headers={"Authorization": f"Bearer {bearer}"},
+        )
+        self.assertEqual(response.status, 200)
+        token = (await response.json())["token"]
+        entry = await self._server().media.get(token)
+        self.assertIsNotNone(entry)
+        path = entry.path
+        self.assertTrue(os.path.isfile(path))
+        async with self._server().media._lock:
+            self._server().media._entries[token].expires_at = time.time() - 1
+        await self._server().media.cleanup()
+        self.assertFalse(os.path.exists(path))
+
     async def test_fetch_expired_token_returns_404(self) -> None:
         path = _write_file(self._sandbox, "gone.bin", content=b"x")
         entry = await self._server().media.register(

plugin/tools/android_navigate.py

@@ -35,6 +35,7 @@ from typing import Any, Callable
 
 from .android_navigate_prompt import ParsedAction, build_prompt, parse_response
 from .android_tool import _bridge_request, _timeout
+from .android_screenshot_media import resolve_screenshot
 
 logger = logging.getLogger("hermes_relay.tools.android_navigate")
 
@@ -172,45 +173,29 @@ def _capture_screenshot() -> _Screenshot:
     an error-JSON envelope. The navigate loop needs structured data.
     """
 
-    import base64
     import tempfile
 
     raw = _get("/screenshot")
-    if "error" in raw:
-        raise RuntimeError(f"bridge /screenshot error: {raw['error']}")
-
-    # The bridge wraps its response in {"data": {...}} in some builds and
-    # returns a flat object in others — match android_tool.py's tolerant
-    # shape-check.
-    result = raw.get("data", raw)
-    img_b64 = result.get("image") or ""
-    if not img_b64:
-        raise RuntimeError("bridge /screenshot returned no image data")
-
-    img_bytes = base64.b64decode(img_b64)
+    img_bytes, mime, marker = resolve_screenshot(raw, _bridge_request, _timeout())
     tmp = tempfile.NamedTemporaryFile(
-        suffix=".jpg", prefix="android_navigate_", delete=False
+        suffix=".png" if mime == "image/png" else ".jpg",
+        prefix="android_navigate_", delete=False
     )
     try:
         tmp.write(img_bytes)
     finally:
         tmp.close()
 
-    # Try to register with the local relay for an opaque token. Fall
-    # back to the bare path marker if the relay isn't reachable (same
-    # graceful degradation as android_screenshot).
-    token_marker = f"file://{tmp.name}"
-    try:
-        from ..relay.client import register_media  # type: ignore
+    token_marker = marker.removeprefix("MEDIA:") if marker else f"file://{tmp.name}"
+    if marker is None:
+        try:
+            from ..relay.client import register_media  # type: ignore
 
-        token = register_media(tmp.name, "image/jpeg", file_name="nav_step.jpg")
-        if token:
-            token_marker = f"hermes-relay://{token}"
-    except Exception:
-        logger.debug(
-            "register_media unavailable — navigate trace will use file:// marker",
-            exc_info=True,
-        )
+            token = register_media(tmp.name, mime, file_name="nav_step.png" if mime == "image/png" else "nav_step.jpg")
+            if token:
+                token_marker = f"hermes-relay://{token}"
+        except Exception:
+            logger.debug("register_media unavailable for legacy screenshot")
 
     return _Screenshot(token=token_marker, local_path=tmp.name)
 

plugin/tools/android_screenshot_media.py

@@ -0,0 +1,98 @@
+"""Decode the Android bridge screenshot contract for host-side tools."""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import re
+from typing import Any, Callable
+
+import requests
+
+
+# Keep tool-result images well below the relay's configurable upload cap.
+MAX_SCREENSHOT_BYTES = 8 * 1024 * 1024
+_TOKEN_RE = re.compile(r"MEDIA:hermes-relay://([A-Za-z0-9_-]{16,128})\Z")
+
+
+class ScreenshotMediaError(ValueError):
+    """A screenshot could not be safely resolved."""
+
+
+def _image_type(data: bytes) -> str:
+    if data.startswith(b"\x89PNG\r\n\x1a\n"):
+        return "image/png"
+    if data.startswith(b"\xff\xd8\xff"):
+        return "image/jpeg"
+    raise ScreenshotMediaError("Screenshot is not a PNG or JPEG image")
+
+
+def resolve_screenshot(
+    raw: dict[str, Any],
+    request: Callable[..., requests.Response],
+    timeout: float,
+) -> tuple[bytes, str, str | None]:
+    """Return verified image bytes, MIME type, and optional relay marker.
+
+    Token fetches use the caller's existing bridge bearer and URL; the token
+    never becomes a URL or path supplied by the phone. Legacy inline images
+    remain readable, but are bounded before decoding.
+    """
+    if not isinstance(raw, dict):
+        raise ScreenshotMediaError("Invalid screenshot response")
+    if "error" in raw:
+        raise ScreenshotMediaError("Bridge screenshot failed")
+    result = raw.get("data", raw)
+    if not isinstance(result, dict):
+        raise ScreenshotMediaError("Invalid screenshot response")
+
+    marker = result.get("media")
+    if marker is not None:
+        if not isinstance(marker, str) or not (match := _TOKEN_RE.fullmatch(marker)):
+            raise ScreenshotMediaError("Invalid screenshot media token")
+        try:
+            response = request("GET", f"/media/{match.group(1)}", timeout=timeout, stream=True)
+        except requests.RequestException:
+            raise ScreenshotMediaError("Screenshot media fetch failed") from None
+        try:
+            if response.status_code != 200:
+                raise ScreenshotMediaError("Screenshot media unavailable or access denied")
+            declared = response.headers.get("Content-Type", "").split(";", 1)[0].lower()
+            if declared not in {"image/png", "image/jpeg"}:
+                raise ScreenshotMediaError("Screenshot media has an unsupported type")
+            length = response.headers.get("Content-Length")
+            if length:
+                try:
+                    declared_length = int(length)
+                except ValueError as exc:
+                    raise ScreenshotMediaError("Invalid screenshot media length") from exc
+                if declared_length < 0 or declared_length > MAX_SCREENSHOT_BYTES:
+                    raise ScreenshotMediaError("Screenshot exceeds size limit")
+            chunks: list[bytes] = []
+            size = 0
+            for chunk in response.iter_content(chunk_size=64 * 1024):
+                size += len(chunk)
+                if size > MAX_SCREENSHOT_BYTES:
+                    raise ScreenshotMediaError("Screenshot exceeds size limit")
+                chunks.append(chunk)
+            data = b"".join(chunks)
+            if _image_type(data) != declared:
+                raise ScreenshotMediaError("Screenshot media type mismatch")
+            return data, declared, marker
+        except requests.RequestException:
+            raise ScreenshotMediaError("Screenshot media fetch failed") from None
+        finally:
+            response.close()
+
+    encoded = result.get("image")
+    if not isinstance(encoded, str) or not encoded:
+        raise ScreenshotMediaError("No image data returned")
+    if len(encoded) > (MAX_SCREENSHOT_BYTES + 2) // 3 * 4:
+        raise ScreenshotMediaError("Screenshot exceeds size limit")
+    try:
+        data = base64.b64decode(encoded, validate=True)
+    except (binascii.Error, ValueError) as exc:
+        raise ScreenshotMediaError("Invalid screenshot image data") from exc
+    if len(data) > MAX_SCREENSHOT_BYTES:
+        raise ScreenshotMediaError("Screenshot exceeds size limit")
+    return data, _image_type(data), None

plugin/tools/android_tool.py

@@ -366,7 +366,7 @@ def _get(path: str, *, device: Optional[str] = None) -> dict:
     return r.json()
 
 
-def _dispatch_android_tool(func: Callable[..., str], args: Optional[dict] = None) -> str:
+def _dispatch_android_tool(func: Callable[..., Any], args: Optional[dict] = None) -> Any:
     """Call an android_* function with optional per-call device scoping.
 
     Public tool schemas can include a ``device`` selector without forcing every
@@ -665,79 +665,71 @@ def android_press_key(key: str) -> str:
         return json.dumps({"error": str(e)})
 
 
-def android_screenshot(sensitive: bool = False) -> str:
+def android_screenshot(sensitive: bool = False) -> dict:
     """
     Capture a screenshot of the Android screen.
 
-    Writes the JPEG to a temp file, then asks the local Hermes-Relay to
-    mint an opaque token via ``POST /media/register``. On success the
-    tool returns ``MEDIA:hermes-relay://<token>`` in its output, which
-    the phone parses and fetches via bearer-auth'd ``GET /media/<token>``
-    on the same relay.
+    Resolve the phone's relay media token with the same bridge bearer, then
+    attach the bounded image bytes as a native multimodal tool result. The
+    marker remains in the text summary for phone-side media delivery.
 
-    Set ``sensitive=True`` when the captured screen may contain private or
-    NSFW content (e.g. a lock screen, a 2FA code, a banking app). The bit is
-    transported verbatim to the phone via the relay's ``X-Media-Sensitive``
-    header so the client can blur the image per the user's setting — the
-    relay performs no classification of its own. Defaults to ``False``.
+    ``sensitive=True`` marks a current relay token private in place, preserving
+    the phone-side blur header without duplicating the image. Legacy inline
+    images use a managed temporary file that the relay retires with its token.
 
-    Fallback: if the relay is not running (or rejects the registration),
-    the tool returns the old ``MEDIA:/tmp/<path>`` form and logs a warning.
-    The phone's parser renders a "relay offline" placeholder for that case.
+    Older Android builds that return inline base64 are also accepted.
     """
     try:
         import base64
-        import logging
-        import tempfile
+        try:
+            from .android_screenshot_media import resolve_screenshot
+        except ImportError:  # direct-script compatibility
+            from android_screenshot_media import resolve_screenshot
 
         data = _get("/screenshot")
-        if "error" in data:
-            return json.dumps(data)
+        img_bytes, mime, marker = resolve_screenshot(data, _bridge_request, _timeout())
+        if sensitive:
+            if marker:
+                from ..relay.client import mark_media_sensitive
 
-        # Extract base64 image from the nested result
-        result = data.get("data", data)
-        img_b64 = result.get("image", "")
-        if not img_b64:
-            return json.dumps({"error": "No image data returned"})
+                if not mark_media_sensitive(marker.removeprefix("MEDIA:hermes-relay://")):
+                    return {"error": "Sensitive screenshot registration failed"}
+            else:
+                import os
+                import tempfile
+                from ..relay.client import register_media
 
-        # Save to temp file
-        img_bytes = base64.b64decode(img_b64)
-        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", prefix="android_screenshot_", delete=False)
-        tmp.write(img_bytes)
-        tmp.close()
-
-        w = result.get("width", "?")
-        h = result.get("height", "?")
-
-        # Try to register with the local relay so the phone gets an opaque
-        # token instead of a literal path. Relay must be running on the
-        # same host (it's loopback-only). Any failure falls back to the
-        # bare path form — the phone shows a placeholder in that case.
-        try:
-            from ..relay.client import register_media
-            token = register_media(
-                tmp.name,
-                "image/jpeg",
-                file_name="screenshot.jpg",
-                sensitive=bool(sensitive),
-            )
-        except Exception:
-            logging.getLogger("hermes_relay.tools").warning(
-                "register_media raised; falling back to bare MEDIA: path",
-                exc_info=True,
-            )
-            token = None
-
-        if token:
-            return f"Screenshot captured ({w}x{h})\nMEDIA:hermes-relay://{token}"
-
-        logging.getLogger("hermes_relay.tools").warning(
-            "relay not reachable; falling back to bare MEDIA: path "
-            "(phone will show a placeholder)"
-        )
-        return f"Screenshot captured ({w}x{h})\nMEDIA:{tmp.name}"
-    except Exception as e:
-        return json.dumps({"error": str(e)})
+                with tempfile.NamedTemporaryFile(
+                    suffix=".png" if mime == "image/png" else ".jpg",
+                    prefix="android_screenshot_", delete=False,
+                ) as tmp:
+                    tmp.write(img_bytes)
+                    path = tmp.name
+                token = None
+                try:
+                    token = register_media(path, mime, sensitive=True, owned_file=True)
+                finally:
+                    if not token:
+                        os.unlink(path)
+                if not token:
+                    return {"error": "Sensitive screenshot registration failed"}
+                marker = f"MEDIA:hermes-relay://{token}"
+        image_url = f"data:{mime};base64,{base64.b64encode(img_bytes).decode('ascii')}"
+        summary = "Screenshot captured; image attached for visual inspection."
+        if marker:
+            summary += f"\n{marker}"
+        if sensitive:
+            summary += "\nSensitive screen: handle privately."
+        return {
+            "_multimodal": True,
+            "content": [
+                {"type": "text", "text": summary},
+                {"type": "image_url", "image_url": {"url": image_url}},
+            ],
+            "text_summary": summary,
+        }
+    except Exception:
+        return {"error": "Screenshot unavailable"}
 
 
 def android_scroll(direction: str, node_id: Optional[str] = None) -> str:
@@ -1672,7 +1664,13 @@ def android_macro(steps: list, name: str = "unnamed", pace_ms: int = 500) -> str
             except (json.JSONDecodeError, TypeError):
                 parsed = {"raw": raw}
         elif isinstance(raw, dict):
-            parsed = raw
+            # A screenshot's native image belongs in the direct tool result;
+            # retaining its data URL in every macro trace would multiply
+            # memory use and expose pixels to text-only callers.
+            parsed = (
+                {"summary": raw.get("text_summary", "Screenshot captured")}
+                if raw.get("_multimodal") else raw
+            )
         else:
             parsed = {"raw": raw}
 
@@ -2062,7 +2060,7 @@ _SCHEMAS = {
     },
     "android_screenshot": {
         "name": "android_screenshot",
-        "description": "Take a screenshot of the current Android screen. Returns base64 PNG. Use when the accessibility tree is missing context or the screen uses canvas/game rendering.",
+        "description": "Take a screenshot of the current Android screen. Attaches a bounded image for visual inspection. Use when the accessibility tree is missing context or the screen uses canvas/game rendering.",
         "parameters": {
             "type": "object",
             "properties": {

plugin/tools/desktop_tool.py

@@ -69,6 +69,8 @@ when ``desktop_terminal`` would time out — useful for debugging.
 
 from __future__ import annotations
 
+import base64
+import binascii
 import json
 import os
 import time
@@ -752,7 +754,7 @@ def desktop_computer_screenshot(
     window_id: Optional[int] = None,
     query: Optional[str] = None,
     include_screenshot: bool = True,
-) -> str:
+) -> str | dict[str, Any]:
     """[EXPERIMENTAL] Capture a display or a structured CUA window snapshot."""
     payload: dict[str, Any] = {
         "display": display,
@@ -773,7 +775,41 @@ def desktop_computer_screenshot(
     if pid is not None or window_id is not None or not include_screenshot:
         payload["include_screenshot"] = bool(include_screenshot)
     data = _post("/desktop/desktop_computer_screenshot", payload)
-    return json.dumps(data)
+    if "error" in data or data.get("ok") is False:
+        return json.dumps(data)
+    result = data.get("result") if isinstance(data.get("result"), dict) else data
+    encoded = result.get("screenshot_base64") or result.get("bytes_base64")
+    if encoded is None:
+        return json.dumps(data)  # saved-path and text-only snapshots
+    max_image_bytes = 8 * 1024 * 1024
+    if not isinstance(encoded, str) or len(encoded) > ((max_image_bytes + 2) // 3) * 4:
+        return json.dumps({"error": "Desktop screenshot exceeds host image limit (8 MiB)"})
+    try:
+        image = base64.b64decode(encoded, validate=True)
+    except (binascii.Error, ValueError):
+        return json.dumps({"error": "Invalid desktop screenshot image data"})
+    mime = "image/png" if image.startswith(b"\x89PNG\r\n\x1a\n") else (
+        "image/jpeg" if image.startswith(b"\xff\xd8\xff") else None
+    )
+    if not mime or len(image) > max_image_bytes:
+        return json.dumps({"error": "Unsupported or oversized desktop screenshot image"})
+    declared = result.get("screenshot_mime_type")
+    if declared and declared != mime:
+        return json.dumps({"error": "Desktop screenshot media type mismatch"})
+    sanitized = {key: value for key, value in result.items()
+                 if key not in {"bytes_base64", "screenshot_base64"}}
+    metadata = {**data, "result": sanitized} if result is not data else sanitized
+    summary = "Desktop screenshot captured; image attached for visual inspection."
+    return {
+        "_multimodal": True,
+        "content": [
+            {"type": "text", "text": f"{summary}\n{json.dumps(metadata)}"},
+            {"type": "image_url", "image_url": {
+                "url": f"data:{mime};base64,{encoded}",
+            }},
+        ],
+        "text_summary": summary,
+    }
 
 
 def desktop_computer_action(action: str, **kwargs: Any) -> str:
@@ -1299,7 +1335,8 @@ _SCHEMAS: dict[str, dict[str, Any]] = {
         "description": (
             "[EXPERIMENTAL] Capture a desktop screenshot for observe-mode "
             "computer-use. Wraps the existing desktop screenshot backend and "
-            "returns PNG bytes or a saved path plus coordinate metadata. "
+            "attaches a bounded image for host vision, or returns a saved path "
+            "plus coordinate metadata when save_to is set. "
             "Requires an active observe/assist/control grant. "
             "Sensitive-window redaction and cursor inclusion are planned fields "
             "and are reported honestly when unavailable."

user-docs/reference/relay-server.md

@@ -264,13 +264,14 @@ hermes relay compat [status|install|remove]
 | `/sessions` | GET | Bearer-auth'd (same token the WSS channel uses). Returns all active paired devices with metadata — device name, token prefix (first 8 chars, full token never exposed), created/last-seen timestamps, session expiry, per-channel grants, transport hint, and `is_current` for the device matching the bearer. `math.inf` expiries serialize as `null` (never expire). |
 | `/sessions/{token_prefix}` | DELETE | Bearer-auth'd. Revoke a paired device by token-prefix (≥ 4 chars). 200 on exact match, 404 on zero, 409 on ambiguous matches. Self-revoke is allowed and flagged via `revoked_self: true`. |
 | `/sessions/{token_prefix}` | PATCH | Bearer-auth'd, self-targeted, and reduction-only. Body `{ttl_seconds?, grants?}` may shorten the caller's current lifetime or existing grants. Extending policy, adding grants, switching to never-expire, or changing another session requires a fresh operator-approved pairing flow. |
-| `/api/plugins/hermes-relay/provider-usage` | GET | Dashboard-authenticated provider usage for Codex, Nous, and OpenCode Go. Optional `profile=<id>` and `session_id=<id>` scope the lookup and let the Dashboard plugin identify the active Codex pool credential directly from the live session. |
+| `/api/plugins/hermes-relay/provider-usage` | GET | Dashboard-authenticated provider usage for Codex, Nous, OpenCode Go, and the Grok subscription behind `xai-oauth`. Optional `profile=<id>` and `session_id=<id>` scope the lookup and let the Dashboard plugin identify the active Codex pool credential directly from the live session. |
 | `/usage/providers` | GET | Bearer-authenticated standalone Relay fallback for the same provider-neutral data. Optional `profile=<id>` and `session_id=<id>` scope credentials and active-session correlation. Disabled by default; set `RELAY_PROVIDER_USAGE_ENABLED=1`. Never returns provider credentials. |
 | `/clipboard/inbox` | POST | Bearer-auth'd clipboard rendezvous used by remote clients before native platform clipboard fallback. |
-| `/media/register` | POST | **Loopback only.** Register a host-local file with the `MediaRegistry` and receive an opaque token. Body: `{"path": "/abs/path", "content_type": "image/jpeg", "file_name": "screenshot.jpg"}`. Used by tools like `android_screenshot` so the agent can emit `MEDIA:hermes-relay://<token>` in chat and have the phone fetch bytes out-of-band. Path is sandboxed to `tempfile.gettempdir()` + `HERMES_WORKSPACE` + any `RELAY_MEDIA_ALLOWED_ROOTS`; symlink escape is rejected via `realpath`. Returns 400 on validation failure. See ADR 14. |
-| `/media/upload` | POST | Bearer-auth'd small upload endpoint for phone-originated media. Accepts JSON `{file_name, content_type, content}` where `content` is base64 and registers the decoded bytes with the media registry. |
+| `/media/register` | POST | **Loopback only.** Register a host-local file with the `MediaRegistry` and receive an opaque token. Body: `{"path": "/abs/path", "content_type": "image/png", "file_name": "screenshot.png"}`. The optional `owned_file: true` is restricted to recognized temporary upload/screenshot files, which Relay removes when the last token retires. Other registered files remain caller-owned. Paths are sandboxed by real path and size. |
+| `/media/upload` | POST | Bearer-authenticated multipart upload for phone-originated media (`file` field). Relay streams it to a size-bounded temporary file, then removes that file on token expiry, eviction, or orderly shutdown. |
+| `/media/{token}/sensitive` | POST | **Loopback only.** Mark an existing token sensitive before returning it to the phone; later fetches include `X-Media-Sensitive: 1` without another image copy. |
 | `/media/{token}` | GET | Stream the bytes of a previously-registered file. Requires `Authorization: Bearer <session_token>` — same token the WSS channel uses (validated against `SessionManager`). Response carries the registered `Content-Type` plus `Content-Disposition: inline; filename="..."` if a file name was provided at register time. The client only ever sees the opaque token — the path is never exposed. 401 without/with bad auth, 404 for unknown/expired tokens. |
-| `/media/by-path` | GET | Fetch a file **by absolute path** rather than by registry token — covers the case where the agent's LLM freeform-emits `MEDIA:/abs/path.ext` in its response text (upstream `prompt_builder.py` tells it to). Query: `path` (required), `content_type` (optional hint). Bearer auth identical to `/media/{token}`. Path is sandbox-validated against the same allowed roots as `/media/register` (`tempfile.gettempdir()` + `HERMES_WORKSPACE` + `RELAY_MEDIA_ALLOWED_ROOTS`). 400 missing `path`; 401 auth; 403 sandbox violation; 404 file not found. See ADR 14 for the bare-path-LLM rationale. |
+| `/media/by-path` | GET | Bearer-authenticated absolute-path fetch for a `MEDIA:/...` marker. Regular-file, size, and credential/system-path checks always apply. Allowed-root enforcement is opt-in through `RELAY_MEDIA_STRICT_SANDBOX=1`; default mode accepts other absolute readable paths. Optional `content_type` overrides extension guessing. Returns 400 for missing path, 401 for auth, 403 for policy/size, and 404 for a missing file. |
 | `/voice/transcribe` | POST | Bearer-auth'd via either a Relay session token with active `voice:stt` grant or a valid Hermes API bearer token. Non-loopback API-bearer calls require HTTPS unless `RELAY_ALLOW_INSECURE_API_BEARER=1`. `multipart/form-data` with an audio file → `{"text": "...", "provider": "openai", "success": true}`. Android may include `?profile=<name>` so logs/UI retain the active profile context; execution still goes through the upstream STT helper. See [Voice Mode](/features/voice) for the full story. |
 | `/voice/synthesize` | POST | Bearer-auth'd via either a Relay session token with active `voice:tts` grant or a valid Hermes API bearer token. Non-loopback API-bearer calls require HTTPS unless `RELAY_ALLOW_INSECURE_API_BEARER=1`. JSON body `{"text": "...", "profile": "mizu"}` (max 5000 chars) → `audio/mpeg` file. This is the basic fallback TTS route; normal Android voice playback prefers `/voice/output/*`. |
 | `/voice/config` | GET | Bearer-auth'd via either a Relay session token with active `voice:config` grant or a valid Hermes API bearer token. Non-loopback API-bearer calls require HTTPS unless `RELAY_ALLOW_INSECURE_API_BEARER=1`. Optional `?profile=<name>` resolves `tts:` / `stt:` from `~/.hermes/profiles/<name>/config.yaml` when present, otherwise falls back to global config. Returns `config_scope`, `profile`, and `fallback_to_global` so the app can label Voice Settings accurately. |