refac

backend/open_webui/utils/terminals.py

@@ -2,6 +2,18 @@
 
 from urllib.parse import quote
 
+from open_webui.utils.chat_id import is_saved_chat_id
+
+TERMINAL_CONTEXT_HEADER = 'X-Terminal-Context-Id'
+TERMINAL_CONTEXT_DEFAULT = 'default'
+TERMINAL_CONTEXT_TYPES = {'chat', 'automation'}
+TERMINAL_CONTEXT_ID_SOURCES = {'chat': 'chat_id', 'automation': 'automation_id'}
+
+
+def is_terminal_orchestrator(connection: dict) -> bool:
+    """Return whether this connection points at Terminals, not raw Open Terminal."""
+    return connection.get('server_type') == 'orchestrator' or bool(connection.get('policy_id'))
+
 
 def get_terminal_server_url(connection: dict) -> str:
     """Return the upstream base URL for a terminal connection.
@@ -14,3 +26,81 @@ def get_terminal_server_url(connection: dict) -> str:
     if policy_id:
         return f'{base_url}/p/{quote(policy_id, safe="")}'
     return base_url
+
+
+def terminal_context_config(connection: dict, context: str) -> dict | bool:
+    """Return config for an OpenWebUI terminal context.
+
+    Missing config is legacy behavior: available, shared default terminal.
+    """
+    if not is_terminal_orchestrator(connection):
+        return {}
+
+    contexts = (connection.get('config') or {}).get('contexts')
+    if not isinstance(contexts, dict):
+        return {}
+
+    value = contexts.get(context, {})
+    if value is False:
+        return False
+    return value if isinstance(value, dict) else {}
+
+
+def terminal_context_available(connection: dict, context: str) -> bool:
+    """Return whether this terminal is exposed in an OpenWebUI context."""
+    if context not in TERMINAL_CONTEXT_TYPES:
+        return False
+    return terminal_context_config(connection, context) is not False
+
+
+def terminal_context_id(
+    connection: dict,
+    metadata: dict | None = None,
+    context: str = 'chat',
+) -> str | None:
+    """Return the terminal runtime context for trusted request metadata."""
+    if not is_terminal_orchestrator(connection) or not terminal_context_available(connection, context):
+        return None
+
+    config = terminal_context_config(connection, context)
+    context_id_source = config.get('context_id') if isinstance(config, dict) else None
+    if not context_id_source or context_id_source == TERMINAL_CONTEXT_DEFAULT:
+        return None
+
+    if context_id_source != TERMINAL_CONTEXT_ID_SOURCES.get(context):
+        return None
+
+    metadata = metadata or {}
+
+    if context == 'automation':
+        automation_id = metadata.get('automation_id')
+        return f'automation:{automation_id}' if automation_id else None
+
+    chat_id = metadata.get('chat_id')
+    if context == 'chat' and chat_id and is_saved_chat_id(chat_id):
+        return f'chat:{chat_id}'
+    return None
+
+
+def terminal_contexts(connection: dict) -> dict:
+    """Return normalized sparse context config for clients."""
+    if not is_terminal_orchestrator(connection):
+        return {}
+
+    contexts = (connection.get('config') or {}).get('contexts')
+    if not isinstance(contexts, dict):
+        return {}
+
+    result = {}
+    for context, value in contexts.items():
+        if context not in TERMINAL_CONTEXT_TYPES:
+            continue
+        if value is False:
+            result[context] = False
+        elif isinstance(value, dict):
+            context_id_source = value.get('context_id')
+            if context_id_source in {TERMINAL_CONTEXT_DEFAULT, TERMINAL_CONTEXT_ID_SOURCES[context]}:
+                result[context] = {'context_id': context_id_source}
+            else:
+                result[context] = {}
+    return result