"""FlowDeck — Agent context builder (v4.14.0). Collects a compact, permission-filtered snapshot of the active workspace so the LLM can reason about real entities (workspaces, documents, collections, pages, Gitea issues) without touching the database directly. """ from __future__ import annotations import json from app.db import get_conn class ContextBuilder: """Builds the textual context that accompanies each agent run.""" def __init__(self, user_id: int, workspace_id: int | None = None): self.user_id = user_id self.workspace_id = workspace_id def build(self, *, mentions: list[str] | None = None, files: list[dict] | None = None, include_collections: bool = True) -> str: """Return a compact Markdown-ish snapshot of the workspace context.""" sections: list[str] = [] if include_collections: sections.append(self._collections_context()) sections.append(self._pages_context()) sections.append(self._documents_context()) sections.append(self._workspaces_context()) if mentions: sections.append(self._mentions_context(mentions)) if files: sections.append(self._files_context(files)) return "\n\n".join(s for s in sections if s) # ── Internals ── def _collections_context(self) -> str: with get_conn() as conn: rows = conn.execute( "SELECT id, name, icon, is_locked, schema_json FROM collections ORDER BY name" ).fetchall() if not rows: return "## Collections\n(no collections yet)" lines = ["## Collections"] lines.append("Collection IDs: " + ", ".join(str(r["id"]) for r in rows)) for r in rows: props = json.loads(r["schema_json"]) if r["schema_json"] else [] schema = ", ".join(p if isinstance(p, str) else p.get("name", "?") for p in props) or "none" lock = " [LOCKED]" if r["is_locked"] else "" lines.append(f"- #{r['id']} {r['icon']} **{r['name']}** (schema: {schema}){lock}") return "\n".join(lines) def _pages_context(self, limit: int = 40) -> str: with get_conn() as conn: rows = conn.execute( "SELECT id, collection_id, title, property_values_json " "FROM collection_pages ORDER BY updated_at DESC LIMIT ?", (limit,), ).fetchall() if not rows: return "## Pages\n(no pages yet)" lines = ["## Recent pages"] for r in rows: props = json.loads(r["property_values_json"]) if r["property_values_json"] else {} summary = ", ".join(str(v) for v in props.values() if v) if props else "" lines.append(f"- page #{r['id']} in collection #{r['collection_id']}: **{r['title']}**{(' — ' + summary) if summary else ''}") return "\n".join(lines) def _documents_context(self, limit: int = 30) -> str: """Recent editor documents (`pages`), usable with create/read/update tools.""" with get_conn() as conn: ws_names = dict( conn.execute("SELECT id, name FROM workspaces").fetchall() ) rows = conn.execute( "SELECT id, title, workspace_id, content_format, parent_id " "FROM pages WHERE deleted_at IS NULL ORDER BY updated_at DESC LIMIT ?", (limit,), ).fetchall() if not rows: return "## Documents\n(aucun document)" lines = ["## Documents (pages éditeur — outils: read_document, write_blocks, create_document)"] for r in rows: ws_name = ws_names.get(r["workspace_id"], str(r["workspace_id"]) if r["workspace_id"] else "racine") lines.append(f"- document #{r['id']} **{r['title'] or 'Sans titre'}** (espace: {ws_name})") return "\n".join(lines) def _workspaces_context(self) -> str: """Workspaces accessible to the current user (with counts).""" base_sql = ( "SELECT w.id, w.name, {role} AS role, " "(SELECT COUNT(*) FROM pages p WHERE p.workspace_id=w.id AND p.deleted_at IS NULL) AS document_count " "FROM workspaces w {join} {where} ORDER BY w.name" ) with get_conn() as conn: if self.user_id is None: rows = conn.execute( base_sql.format(role="'owner'", join="", where=""), [] ).fetchall() else: rows = conn.execute( base_sql.format( role="COALESCE(wm.role, CASE WHEN w.owner_id=? THEN 'owner' ELSE 'viewer' END)", join="LEFT JOIN workspace_members wm ON wm.workspace_id=w.id AND wm.user_id=?", where="WHERE w.owner_id=? OR wm.user_id IS NOT NULL", ), (self.user_id, self.user_id, self.user_id), ).fetchall() if not rows: return "## Espaces de travail\n(aucun espace)" lines = ["## Espaces de travail (outil: read_workspaces)"] for r in rows: lines.append(f"- espace #{r['id']} **{r['name']}** ({r['role']}, {r['document_count']} document(s))") return "\n".join(lines) def _mentions_context(self, mentions: list[str]) -> str: """Resolve @document:x / @collection:x / @page:y / @repo:o/r mentions. Mentions bring the *actual content* of the referenced object into the context so the LLM can summarise / rewrite / analyse it directly without needing a read tool round-trip (and so the offline mock stays useful). """ lines = ["## Mentioned context"] for m in mentions: if m.startswith("document:"): pid = m.split(":", 1)[1] lines.append(self._single_document(pid)) elif m.startswith("collection:"): cid = m.split(":", 1)[1] lines.append(self._single_collection(cid)) elif m.startswith("page:"): pid = m.split(":", 1)[1] lines.append(self._single_page(pid)) elif m.startswith("repo:"): lines.append(f"- @repo: {m.split(':', 1)[1]} (Gitea issues available via read_gitea_issues)") elif m == "ws": lines.append("- @ws: full workspace context included above") return "\n".join(lines) @staticmethod def _blocks_to_text(content: str, limit: int = 9000) -> str: """Flatten a ``blocks`` document JSON into plain readable text. Collects the textual payload of each block (content, title, caption, children, meeting notes/summary) — enough for the LLM to reason about a mentioned editor page without the full block schema. """ try: blocks = json.loads(content or "[]") except (json.JSONDecodeError, TypeError): return "" if not isinstance(blocks, list): return "" _TEXT_KEYS = ("content", "title", "caption", "plain_text", "notes", "summary") out: list[str] = [] total = 0 def walk(node): nonlocal total if total >= limit: return if isinstance(node, dict): for k in _TEXT_KEYS: v = node.get(k) if isinstance(v, str) and v.strip(): line = v.replace("\r\n", "\n").strip() out.append(line) total += len(line) + 1 if total >= limit: return for v in node.values(): walk(v) elif isinstance(node, list): for item in node: walk(item) walk(blocks) return "\n".join(out)[:limit] def _single_document(self, pid: str) -> str: """Full editor-document mention: title + workspace + real content.""" with get_conn() as conn: row = conn.execute( "SELECT p.*, w.name AS ws_name FROM pages p " "LEFT JOIN workspaces w ON w.id=p.workspace_id " "WHERE p.id=? AND p.deleted_at IS NULL", (pid,), ).fetchone() if not row: return f"- document #{pid}: not found" title = row["title"] or "Sans titre" ws = row["ws_name"] or "" loc = f" (espace: {ws})" if ws else "" fmt = row["content_format"] or "blocks" raw = row["content"] or "" if fmt == "markdown": body = raw.strip() elif fmt == "file": body = "" else: body = self._blocks_to_text(raw) head = f"- document #{row['id']} **{title}**{loc}" if body: return f"{head}:\n{body[:9000]}" return head def _single_collection(self, cid: str) -> str: with get_conn() as conn: row = conn.execute("SELECT id, name, icon, schema_json FROM collections WHERE id=?", (cid,)).fetchone() if not row: return f"- collection #{cid}: not found" return f"- collection #{row['id']} {row['icon']} **{row['name']}**" def _single_page(self, pid: str) -> str: with get_conn() as conn: row = conn.execute("SELECT id, title, property_values_json FROM collection_pages WHERE id=?", (pid,)).fetchone() if not row: return f"- page #{pid}: not found" props = json.loads(row["property_values_json"]) if row["property_values_json"] else {} return f"- page #{row['id']} **{row['title']}** props={json.dumps(props, ensure_ascii=False)}" def _files_context(self, files: list[dict]) -> str: return "## Attached files\n" + "\n".join( f"- {f.get('name', 'file')} ({f.get('size', '?')} bytes)" for f in files )