def describe_screen(
*,
question: str = "",
monitor: Optional[int] = None,
skip_rate_limit: bool = False,
) -> ToolResult:
"""Capture once and describe. Shared by one-shot tool and share loop."""
global _last_capture_monotonic
cfg = _vision_config()
if not _vision_enabled(cfg):
return ToolResult(
tool_name="screen_describe",
content=(
"Screen vision is disabled. Enable with "
"[desktop.vision] enabled = true "
"(and grant Screen Recording on macOS)."
),
success=False,
)
rate_ms = int(getattr(cfg, "rate_limit_ms", 1500) or 1500)
now = time.monotonic()
if (
not skip_rate_limit
and rate_ms > 0
and (now - _last_capture_monotonic) * 1000 < rate_ms
):
# While sharing, prefer cached summary if available
share = get_screen_share()
cached = share.latest_summary() if share.is_active() else ""
if cached:
return ToolResult(
tool_name="screen_describe",
content=cached,
success=True,
metadata={"cached": True, "sharing": True},
)
return ToolResult(
tool_name="screen_describe",
content="Please wait a moment before capturing the screen again.",
success=False,
metadata={"rate_limited": True},
)
q = (question or "").strip() or (
"Describe briefly what you see on the screen. "
"Focus on the main window and any text the user might care about. "
"2–5 short sentences."
)
if monitor is None:
mon = int(getattr(cfg, "monitor", 1) or 1)
else:
mon = int(monitor)
max_dim = int(getattr(cfg, "max_dimension", 1280) or 1280)
keep_temp = bool(getattr(cfg, "keep_temp", False))
allow_cloud = bool(getattr(cfg, "allow_cloud", False))
model = str(getattr(cfg, "model", "") or "").strip()
engine_key = str(getattr(cfg, "engine", "") or "").strip()
# ── AUTHORISE FIRST, CAPTURE SECOND ──────────────────────────────────
# The previous order captured the screen and only then decided whether it
# was allowed to be sent. capture_screen_b64 goes through
# capture_screen_to_temp, so a request that was about to be refused still
# wrote a full-screen image to a temp file. Refusing after the fact
# protects the network but not the disk.
#
# Everything that can refuse — no engine, remote engine, no model — now
# runs before any image of the user's screen is allowed to exist.
try:
from diapason.core.config import load_config
from diapason.core.local_mode import local_only
from diapason.engine._discovery import get_engine
config = load_config()
key = engine_key or (config.engine.default or "").strip() or None
# get_engine rend (clé, moteur) — le couple passait pour le moteur
# et .generate explosait au premier regard réel (bogue dormant tant
# que la vision restait éteinte ; réveillé à l'allumage, 24/08/2026).
paire = get_engine(config, key)
if paire is None:
return ToolResult(
tool_name="screen_describe",
content="No inference engine available for vision.",
success=False,
)
cle_resolue, engine = paire
engine_id = getattr(engine, "engine_id", "") or cle_resolue or key or ""
is_local = engine_id in LOCAL_VISION_ENGINES
if engine_id and engine_id not in LOCAL_VISION_ENGINES:
if getattr(engine, "is_cloud", False) or engine_id in {
"openai",
"anthropic",
"gemini",
"groq",
}:
is_local = False
else:
is_local = not bool(getattr(engine, "is_cloud", False))
# [privacy] local_only outranks [desktop.vision] allow_cloud. A
# per-domain switch may only ever be more restrictive than the global
# one, never less — otherwise the global switch is a suggestion.
if local_only(config):
allow_cloud = False
if not allow_cloud and not is_local:
return ToolResult(
tool_name="screen_describe",
content=(
f"Refusing to send screenshot to non-local engine "
f"('{engine_id}'). Use ollama (e.g. gemma3:4b / llava) "
"or set [desktop.vision] allow_cloud = true."
),
success=False,
metadata={"engine": engine_id, "local": False, "captured": False},
)
resolved_model = model or (config.intelligence.default_model or "").strip()
if not resolved_model:
return ToolResult(
tool_name="screen_describe",
content=(
"No vision model configured. Set [desktop.vision] model "
"or [intelligence] default_model (e.g. gemma3:4b)."
),
success=False,
)
except Exception as exc:
logger.exception("screen_describe: engine resolution failed")
return ToolResult(
tool_name="screen_describe",
content=f"Screen vision failed: {exc}",
success=False,
)
# Authorised — only now may an image of the screen exist.
try:
b64, meta = capture_screen_b64(
monitor=mon,
max_dimension=max_dim,
keep_temp=keep_temp,
)
except Exception as exc:
return ToolResult(tool_name="screen_describe", content=str(exc), success=False)
_last_capture_monotonic = time.monotonic()
if skip_rate_limit:
# La boucle de partage seulement : trame identique = description
# identique, sans réveiller le modèle.
import hashlib
global _derniere_empreinte_partage, _derniere_description_partage
empreinte = hashlib.sha256(b64.encode("ascii")).hexdigest()
if empreinte == _derniere_empreinte_partage and _derniere_description_partage:
return ToolResult(
tool_name="screen_describe",
content=_derniere_description_partage,
success=True,
metadata={"cached": True, "unchanged": True},
)
_derniere_empreinte_partage = empreinte
try:
result = engine.generate(
[Message(role=Role.USER, content=q, images=[b64])],
model=resolved_model,
temperature=0.2,
max_tokens=400,
)
content = ""
if isinstance(result, dict):
content = str(result.get("content") or "").strip()
if not content:
content = "I could not read the screen clearly."
if skip_rate_limit:
_derniere_description_partage = content
return ToolResult(
tool_name="screen_describe",
content=content,
success=True,
metadata={
"monitor": mon,
"bytes": meta.get("bytes"),
"engine": engine_id,
"model": resolved_model,
"local": is_local,
"sharing": get_screen_share().is_active(),
},
)
except Exception as exc:
logger.exception("screen_describe failed")
return ToolResult(
tool_name="screen_describe",
content=f"Screen vision failed: {exc}",
success=False,
)