mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-07-10 01:18:29 +00:00
bd584da2f4
Generalize the Editor storage/session path from Markdown-only to exact-text .md and .txt documents, including open, save, Save As, and rename refresh behavior. Expose Open in Editor as a visible row action in the File Browser for Editor-owned text files, keep non-Editor surface opens in the overflow menu, and route Desktop text MIME handling through the Editor bridge. Tests: node --check webui/components/modals/file-browser/file-browser-store.js; python -m py_compile plugins/_office/helpers/document_store.py plugins/_editor/helpers/markdown_sessions.py plugins/_editor/api/editor_session.py plugins/_editor/api/ws_editor.py plugins/_desktop/helpers/desktop_session.py plugins/_desktop/api/desktop_session.py plugins/_office/api/office_session.py plugins/_office/api/ws_office.py; pytest tests/test_office_document_store.py tests/test_file_browser_navigation.py tests/test_office_canvas_setup.py -q
75 lines
3.3 KiB
Python
75 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from helpers.ws import WsHandler
|
|
from helpers.ws_manager import WsResult
|
|
from plugins._editor.helpers import markdown_sessions
|
|
from plugins._office.helpers import document_store
|
|
|
|
|
|
class WsEditor(WsHandler):
|
|
async def on_disconnect(self, sid: str) -> None:
|
|
markdown_sessions.get_manager().close_sid(sid)
|
|
|
|
async def process(self, event: str, data: dict[str, Any], sid: str) -> dict[str, Any] | WsResult | None:
|
|
if not event.startswith("editor_"):
|
|
return None
|
|
try:
|
|
if event == "editor_open":
|
|
return self._open(data, sid)
|
|
if event == "editor_input":
|
|
return markdown_sessions.get_manager().input(
|
|
str(data.get("session_id") or ""),
|
|
text=data.get("text") if "text" in data else None,
|
|
patch=data.get("patch") if isinstance(data.get("patch"), dict) else None,
|
|
)
|
|
if event == "editor_save":
|
|
return markdown_sessions.get_manager().save(
|
|
str(data.get("session_id") or ""),
|
|
text=data.get("text") if "text" in data else None,
|
|
)
|
|
if event == "editor_activate":
|
|
return markdown_sessions.get_manager().activate(str(data.get("session_id") or ""))
|
|
if event == "editor_close":
|
|
return markdown_sessions.get_manager().close(str(data.get("session_id") or ""))
|
|
except FileNotFoundError as exc:
|
|
return WsResult.error(code="EDITOR_SESSION_NOT_FOUND", message=str(exc), correlation_id=data.get("correlationId"))
|
|
except Exception as exc:
|
|
return WsResult.error(code="EDITOR_ERROR", message=str(exc), correlation_id=data.get("correlationId"))
|
|
|
|
return WsResult.error(
|
|
code="UNKNOWN_EDITOR_EVENT",
|
|
message=f"Unknown editor event: {event}",
|
|
correlation_id=data.get("correlationId"),
|
|
)
|
|
|
|
def _open(self, data: dict[str, Any], sid: str) -> dict[str, Any] | WsResult:
|
|
context_id = str(data.get("ctxid") or data.get("context_id") or "")
|
|
file_id = str(data.get("file_id") or "").strip()
|
|
path = str(data.get("path") or "").strip()
|
|
if file_id:
|
|
doc = document_store.get_document(file_id)
|
|
elif path:
|
|
doc = document_store.register_document(path, context_id=context_id)
|
|
else:
|
|
fmt = str(data.get("format") or "md").lower().strip().lstrip(".")
|
|
if fmt not in document_store.EDITOR_TEXT_EXTENSIONS:
|
|
return WsResult.error(
|
|
code="UNSUPPORTED_EDITOR_DOCUMENT",
|
|
message=f"Editor can only create Markdown (.md) and text (.txt) documents, not .{fmt}.",
|
|
correlation_id=data.get("correlationId"),
|
|
)
|
|
doc = document_store.create_document(
|
|
kind="document",
|
|
title=str(data.get("title") or "Untitled"),
|
|
fmt=fmt,
|
|
content=str(data.get("content") or ""),
|
|
context_id=context_id,
|
|
)
|
|
return markdown_sessions.get_manager().open(
|
|
doc,
|
|
sid=sid,
|
|
context_id=context_id,
|
|
refresh=data.get("refresh") is True,
|
|
)
|