...

from typing import Any, Dict, List, Optional import logging import re from src.constants import MAX_READ_CHARS from src.tool_utils import _parse_tool_args logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Active document state # --------------------------------------------------------------------------- _active_document_id: Optional[str] = None _active_model: Optional[str] = None def set_active_document(doc_id: Optional[str]): """Set the active document ID for document tool execution.""" global _active_document_id _active_document_id = doc_id def set_active_model(model: Optional[str]): """Set the current model name for version summaries.""" global _active_model _active_model = model def get_active_document(): return _active_document_id def clear_active_document(doc_id: Optional[str] = None) -> bool: """Clear the in-memory active-document pointer. With ``doc_id`` given, only clears when it matches the current pointer, so a different active document is left untouched. Returns True if it was cleared. Called when a document is detached from its session or deleted (its tab is closed): without this, the stale pointer makes the last-resort doc-injection path re-surface a closed document in a later, unrelated chat — even one whose session no longer matches — because an unlinked doc has session_id NULL (#1160). """ global _active_document_id if doc_id is None or _active_document_id == doc_id: _active_document_id = None return True return False def _owned_document_query(query, Document, owner: Optional[str]): if owner is None: # A bare Python `False` is not a valid SQL expression — SQLAlchemy 1.4 # deprecates it and 2.0 raises ArgumentError. Use the SQL `false()` # literal to return zero rows for an unscoped (owner-less) query. from sqlalchemy import false return query.filter(false()) return query.filter(Document.owner == owner) def _get_owned_document(db, Document, doc_id: str, owner: Optional[str], active_only: bool = False): q = db.query(Document).filter(Document.id == doc_id) if active_only: q = q.filter(Document.is_active == True) q = _owned_document_query(q, Document, owner) return q.first() def _most_recent_owned_document(db, Document, owner: Optional[str], active_only: bool = False): q = db.query(Document) if active_only: q = q.filter(Document.is_active == True) q = _owned_document_query(q, Document, owner) return q.order_by(Document.updated_at.desc()).first() # --------------------------------------------------------------------------- # Document tools — create/update/edit/suggest living documents # --------------------------------------------------------------------------- def _sniff_doc_language(text: str) -> str: """Best-effort detect a document's language from its content when the model didn't specify one. Defaults to 'markdown' (prose). Recognizes the common markup/code types the editor supports so e.g. an SVG isn't saved as markdown.""" import json as _json, re as _re2 s = (text or "").strip() if not s: return "markdown" head = s[:600] hl = head.lower() if _looks_like_email_document(s): return "email" # Markup (unambiguous) if " bool: import re as _re title_l = (title or "").strip().lower() if title_l in {"new email", "new mail", "new message"}: return True s = (text or "").lstrip() if "\n---\n" in s and _re.search(r"(?im)^To:\s*", s) and _re.search(r"(?im)^Subject:\s*", s): return True return bool(_re.search(r"(?im)^To:\s*", s) and _re.search(r"(?im)^Subject:\s*", s)) def _split_email_header_body(text: str) -> tuple[str, str]: if "\n---\n" in (text or ""): header, body = (text or "").split("\n---\n", 1) return header.rstrip(), body.strip() return (text or "").strip(), "" def _split_email_reply_history(body: str) -> tuple[str, str]: """Split draft body from quoted/original email history. Email reply docs keep the original thread below the user's new reply. Models often rewrite only the fresh reply body; this helper keeps the historical block from being wiped when update_document/edit_document replaces content. """ text = body or "" literal = "---------- Previous message ----------" literal_idx = text.find(literal) if literal_idx >= 0: return text[:literal_idx].strip(), text[literal_idx:].strip() patterns = [ r"(?m)^On .+ wrote:\s*$", r"(?m)^> .+", ] starts = [] for pat in patterns: m = re.search(pat, text) if m: starts.append(m.start()) if not starts: return text.strip(), "" idx = min(starts) return text[:idx].strip(), text[idx:].strip() def _merge_email_headers(old_header: str, new_header: str) -> str: """Preserve routing/threading metadata if a model omits it.""" protected = ( "In-Reply-To", "References", "X-Source-UID", "X-Source-Folder", "X-Attachments", "X-Forward-Attachments", ) lines = [l for l in (new_header or "").splitlines() if l.strip()] present = {l.split(":", 1)[0].strip().lower() for l in lines if ":" in l} for old_line in (old_header or "").splitlines(): if ":" not in old_line: continue key = old_line.split(":", 1)[0].strip() if key in protected and key.lower() not in present: lines.append(old_line) present.add(key.lower()) return "\n".join(lines).rstrip() def _coerce_email_document_content(existing: str, incoming: str) -> str: """Keep email docs in the To/Subject/---/body shape even if a model writes only the body or dumps header labels without the separator.""" import re as _re old = existing or "" new = (incoming or "").strip() old_header, old_body = _split_email_header_body(old) _, old_history = _split_email_reply_history(old_body) if "\n---\n" in new: new_header, new_body = _split_email_header_body(new) new_own, new_history = _split_email_reply_history(new_body) if old_history and not new_history: new_body = (new_own + "\n\n" + old_history).strip() return _merge_email_headers(old_header, new_header).rstrip() + "\n---\n" + new_body header = old_header if old_header else "To: \nSubject: " if _looks_like_email_document(new): lines = new.splitlines() last_header_idx = -1 header_re = _re.compile(r"^(To|Cc|Bcc|Subject|In-Reply-To|References|X-Source-UID|X-Source-Folder|X-Attachments):", _re.I) for i, line in enumerate(lines): if header_re.match(line.strip()): last_header_idx = i body_lines = lines[last_header_idx + 1:] if last_header_idx >= 0 else lines while body_lines and not body_lines[0].strip(): body_lines.pop(0) body = "\n".join(body_lines).strip() else: body = new _, incoming_history = _split_email_reply_history(body) if old_history and not incoming_history: body = (body.strip() + "\n\n" + old_history).strip() return header.rstrip() + "\n---\n" + body def parse_edit_blocks(content: str) -> list: """Parse <<>>...<<>>...<<>> blocks.""" edits = [] pattern = r'<<>>\n(.*?)\n<<>>\n(.*?)\n<<>>' for m in re.finditer(pattern, content, re.DOTALL): edits.append({"find": m.group(1), "replace": m.group(2)}) return edits def parse_suggest_blocks(content: str) -> list: """Parse <<>>...<<>>...<<>>...<<>> blocks.""" suggestions = [] _skip_phrases = ["no change", "clear", "fine as", "looks good", "no improvement", "keep as"] pattern = r'<<>>\n(.*?)\n<<>>\n(.*?)\n<<>>\n(.*?)\n<<>>' for m in re.finditer(pattern, content, re.DOTALL): find_text = m.group(1) replace_text = m.group(2) reason = m.group(3).strip() # Skip no-op suggestions where find == replace or reason says no change if find_text.strip() == replace_text.strip(): continue if any(phrase in reason.lower() for phrase in _skip_phrases): continue suggestions.append({ "id": f"sugg-{len(suggestions)+1}", "find": find_text, "replace": replace_text, "reason": reason, }) return suggestions def _pdf_source_upload_id(content: str) -> Optional[str]: try: from src.pdf_form_doc import find_source_upload_id return find_source_upload_id(content or "") except Exception: return None def _strip_pdf_editor_markers(content: str) -> str: """Turn a PDF-wrapper markdown doc into ordinary editable markdown. PDF docs use hidden HTML comments for source-upload links, form fields, and page annotations. Those comments are necessary for rendering/exporting the original PDF, but they make a derived AI text edit keep showing the original PDF preview. Remove only the editor plumbing and keep the readable text. """ text = content or "" text = re.sub(r'(?im)^\s*\s*\n*', '', text) text = re.sub(r'\s*', '', text) text = re.sub(r'\s*', '', text) return text.strip() def _create_pdf_text_derivative(db, *, source_doc, content: str, owner: Optional[str], summary: str) -> dict: import uuid from src.database import Document, DocumentVersion clean = _strip_pdf_editor_markers(content) title_base = (getattr(source_doc, "title", None) or "PDF").strip() title = title_base if title_base.lower().endswith("edited") else f"{title_base} edited" doc_id = str(uuid.uuid4()) ver_id = str(uuid.uuid4()) new_doc = Document( id=doc_id, session_id=getattr(source_doc, "session_id", None), title=title, language="markdown", current_content=clean, version_count=1, is_active=True, owner=owner if owner is not None else getattr(source_doc, "owner", None), ) ver = DocumentVersion( id=ver_id, document_id=doc_id, version_number=1, content=clean, summary=summary, source="ai", ) db.add(new_doc) db.add(ver) db.commit() set_active_document(doc_id) return { "action": "create", "doc_id": doc_id, "title": title, "language": "markdown", "content": clean, "version": 1, "source_doc_id": getattr(source_doc, "id", None), } class CreateDocumentTool: async def execute(self, content: str, ctx: dict) -> dict: """Create a new document. Supports two formats: 1) Line-based: line 1 = title, line 2 (optional) = language, rest = content 2) XML-like tags: ......... Some models mix them — strip any XML-style tags and fall back to line parsing.""" import uuid, re as _re from src.database import SessionLocal, Document, DocumentVersion, Session as DbSession raw = content or "" session_id = ctx.get("session_id") owner = ctx.get("owner") # Known languages the editor understands (match the