+"""
+ from app.routers.dashboard import _sidebar_data
+ env = Environment(loader=FileSystemLoader("app/templates"))
+ sidebar = _sidebar_data(request, [])
+ block_tpl = env.from_string(
+ '{% extends "base.html" %}{% block content %}{{ content_html|safe }}{% endblock %}'
+ )
+ return block_tpl.render(
+ **sidebar,
+ request=request,
+ content_html=content_html,
+ page_title=ts["name"],
+ title_prefix="Teamspace",
+ page_icon="🏛️",
+ )
+
+
+@router.post("/api/v2/wiki/teamspaces")
+async def create_teamspace(request: Request):
+ user = _user(request)
+ try:
+ body = await request.json()
+ except Exception:
+ body = {}
+ name = (body.get("name") or "").strip()
+ if not name or len(name) > 120:
+ raise HTTPException(400, "name required (max 120 chars)")
+ wid = int(body.get("workspace_id") or 0)
+ with get_conn() as conn:
+ ws = conn.execute("SELECT owner_id FROM workspaces WHERE id=?", (wid,)).fetchone()
+ if not ws:
+ raise HTTPException(404, "Workspace not found")
+ admin = conn.execute("SELECT is_admin FROM users WHERE id=?", (user["id"],)).fetchone()
+ member = conn.execute("SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=?",
+ (wid, user["id"])).fetchone()
+ allowed = (ws["owner_id"] == user["id"] or (admin and admin["is_admin"])
+ or (member and member["role"] in ("admin", "editor", "owner")))
+ if not allowed:
+ raise HTTPException(403, "Editor role required in the workspace")
+ try:
+ tsid = wiki.create_teamspace(wid, name, user["id"],
+ description=body.get("description") or "",
+ private=bool(body.get("private")))
+ except ValueError as exc:
+ raise HTTPException(409, str(exc)) from None
+ audit_log(user, "teamspace.create", "teamspace", tsid, name, request)
+ return JSONResponse(status_code=201, content={"id": tsid, "name": name})
+
+
+@router.get("/api/v2/wiki/teamspaces/{teamspace_id}")
+async def get_teamspace(teamspace_id: int, request: Request):
+ user = _user(request)
+ ts = _teamspace_or_404(teamspace_id, user["id"])
+ ts["role"] = wiki.get_teamspace_role(user["id"], teamspace_id)
+ ts["member_count"] = len(wiki.teamspace_member_ids(teamspace_id))
+ return ts
+
+
+@router.get("/api/v2/wiki/teamspaces/{teamspace_id}/members")
+async def list_members(teamspace_id: int, request: Request):
+ user = _user(request)
+ _teamspace_or_404(teamspace_id, user["id"])
+ with get_conn() as conn:
+ rows = conn.execute(
+ """SELECT m.user_id, m.role, u.login, u.full_name FROM teamspace_members m
+ JOIN users u ON u.id = m.user_id WHERE m.teamspace_id=? ORDER BY u.login""",
+ (teamspace_id,)).fetchall()
+ return {"members": [dict(r) for r in rows]}
+
+
+@router.put("/api/v2/wiki/teamspaces/{teamspace_id}/members/{member_id}")
+async def set_member(teamspace_id: int, member_id: int, request: Request):
+ user = _user(request)
+ _teamspace_or_404(teamspace_id, user["id"])
+ if not wiki.can_write_teamspace(user["id"], teamspace_id):
+ raise HTTPException(403, "Editor role required")
+ try:
+ body = await request.json()
+ except Exception:
+ body = {}
+ role = body.get("role")
+ if role not in wiki.TEAMSPACE_ROLES:
+ raise HTTPException(400, f"role must be one of {', '.join(wiki.TEAMSPACE_ROLES)}")
+ with get_conn() as conn:
+ if not conn.execute("SELECT 1 FROM users WHERE id=?", (member_id,)).fetchone():
+ raise HTTPException(404, "User not found")
+ conn.execute(
+ """INSERT INTO teamspace_members (teamspace_id, user_id, role) VALUES (?,?,?)
+ ON CONFLICT(teamspace_id, user_id) DO UPDATE SET role=excluded.role""",
+ (teamspace_id, member_id, role))
+ conn.commit()
+ audit_log(user, "teamspace.member.set", "teamspace", teamspace_id,
+ f"u{member_id}={role}", request)
+ return {"status": "ok", "user_id": member_id, "role": role}
+
+
+@router.delete("/api/v2/wiki/teamspaces/{teamspace_id}/members/{member_id}")
+async def remove_member(teamspace_id: int, member_id: int, request: Request):
+ user = _user(request)
+ _teamspace_or_404(teamspace_id, user["id"])
+ if not wiki.can_write_teamspace(user["id"], teamspace_id):
+ raise HTTPException(403, "Editor role required")
+ with get_conn() as conn:
+ cur = conn.execute("DELETE FROM teamspace_members WHERE teamspace_id=? AND user_id=?",
+ (teamspace_id, member_id))
+ conn.commit()
+ if not cur.rowcount:
+ raise HTTPException(404, "Not a member")
+ return {"status": "removed", "user_id": member_id}
+
+
+# ── verified pages ─────────────────────────────────────────────────────────
+
+@router.get("/api/v2/wiki/pages/{page_id}/verification")
+async def get_verification(page_id: int, request: Request):
+ _user(request)
+ _page_or_404(page_id)
+ return {"verification": wiki.verification(page_id)}
+
+
+@router.post("/api/v2/wiki/pages/{page_id}/verify")
+async def verify_page(page_id: int, request: Request):
+ user = _user(request)
+ page = _page_or_404(page_id)
+ if not _can_verify(user, page):
+ raise HTTPException(403, "Editor role required to verify a page")
+ try:
+ body = await request.json()
+ except Exception:
+ body = {}
+ out = wiki.verify_page(page_id, user["id"],
+ days=body.get("days") or wiki.VERIFICATION_DAYS_DEFAULT,
+ note=body.get("note") or "")
+ audit_log(user, "page.verify", "page", page_id, out.get("expires_at") or "", request)
+ return {"verification": out}
+
+
+@router.delete("/api/v2/wiki/pages/{page_id}/verify")
+async def unverify_page(page_id: int, request: Request):
+ user = _user(request)
+ _page_or_404(page_id)
+ if not wiki.unverify_page(page_id):
+ raise HTTPException(404, "Page is not verified")
+ audit_log(user, "page.unverify", "page", page_id, "", request)
+ return {"status": "unverified", "page_id": page_id}
+
+
+@router.get("/api/v2/wiki/verified")
+async def list_verified(request: Request):
+ """Verified (non-expired) pages of a workspace — the ✅ wiki index."""
+ user = _user(request)
+ wid = _workspace_id(request)
+ with get_conn() as conn:
+ rows = conn.execute(
+ """SELECT p.id, p.title, p.page_icon, p.teamspace_id,
+ v.verified_at, v.expires_at, v.note, u.login
+ FROM page_verifications v
+ JOIN pages p ON p.id = v.page_id
+ LEFT JOIN users u ON u.id = v.verified_by
+ WHERE p.workspace_id=? AND p.deleted_at IS NULL""",
+ (wid,)).fetchall()
+ out = []
+ for r in rows:
+ item = dict(r)
+ if wiki.is_expired(r):
+ continue # badge lapsed → not listed
+ if item["teamspace_id"] and not wiki.can_read_teamspace(user["id"],
+ item["teamspace_id"]):
+ continue # private teamspace → hidden
+ item["active"] = True
+ out.append(item)
+ return {"pages": out}
+
+
+# ── follows ────────────────────────────────────────────────────────────────
+
+@router.post("/api/v2/wiki/pages/{page_id}/follow")
+async def follow_page(page_id: int, request: Request):
+ user = _user(request)
+ _page_or_404(page_id)
+ now = wiki.toggle_follow(page_id, user["id"])
+ return {"page_id": page_id, "following": now}
+
+
+@router.get("/api/v2/wiki/pages/{page_id}/followers")
+async def list_followers(page_id: int, request: Request):
+ _user(request)
+ _page_or_404(page_id)
+ ids = wiki.followers(page_id)
+ if not ids:
+ return {"followers": []}
+ with get_conn() as conn:
+ rows = conn.execute(
+ f"SELECT id, login, full_name FROM users WHERE id IN ({','.join('?' * len(ids))})",
+ ids).fetchall()
+ return {"followers": [dict(r) for r in rows]}
+
+
+# ── comment reactions ──────────────────────────────────────────────────────
+
+@router.post("/api/v2/wiki/comments/{comment_id}/reactions")
+async def react(comment_id: int, request: Request):
+ user = _user(request)
+ try:
+ body = await request.json()
+ except Exception:
+ body = {}
+ emoji = (body.get("emoji") or "").strip()
+ if not emoji:
+ raise HTTPException(400, "emoji required")
+ try:
+ counts = wiki.toggle_reaction(comment_id, user["id"], emoji)
+ except LookupError:
+ raise HTTPException(404, "Comment not found") from None
+ return {"comment_id": comment_id, "reactions": counts}
+
+
+@router.get("/api/v2/wiki/comments/{comment_id}/reactions")
+async def list_reactions(comment_id: int, request: Request):
+ _user(request)
+ return {"comment_id": comment_id, "reactions": wiki.reactions(comment_id)}
+
+
+# ── guest shares ───────────────────────────────────────────────────────────
+
+@router.post("/api/v2/wiki/pages/{page_id}/guests")
+async def create_guest(page_id: int, request: Request):
+ user = _user(request)
+ page = _page_or_404(page_id)
+ if page.get("teamspace_id") and not wiki.can_write_teamspace(user["id"], page["teamspace_id"]):
+ raise HTTPException(403, "Editor role required to share")
+ try:
+ body = await request.json()
+ except Exception:
+ body = {}
+ try:
+ share = wiki.create_guest_share(page_id, body.get("email") or "",
+ body.get("role") or "viewer",
+ user["id"], days=body.get("days", 30))
+ except ValueError as exc:
+ raise HTTPException(400, str(exc)) from None
+ audit_log(user, "page.guest_share", "page", page_id, share["email"], request)
+ return JSONResponse(status_code=201, content={
+ "id": share["id"], "token": share["token"], "role": share["role"],
+ "expires_at": share["expires_at"], "url": f"/g/{share['token']}"})
+
+
+@router.get("/api/v2/wiki/pages/{page_id}/guests")
+async def list_guests(page_id: int, request: Request):
+ _user(request)
+ _page_or_404(page_id)
+ with get_conn() as conn:
+ rows = conn.execute(
+ "SELECT id, email, role, expires_at, revoked, created_at FROM guest_shares"
+ " WHERE page_id=? ORDER BY id DESC", (page_id,)).fetchall()
+ return {"guests": [dict(r) for r in rows]}
+
+
+@router.delete("/api/v2/wiki/guests/{share_id}")
+async def revoke_guest(share_id: int, request: Request):
+ user = _user(request)
+ with get_conn() as conn:
+ if not conn.execute("SELECT 1 FROM guest_shares WHERE id=?", (share_id,)).fetchone():
+ raise HTTPException(404, "Guest share not found")
+ conn.execute("UPDATE guest_shares SET revoked=1 WHERE id=?", (share_id,))
+ conn.commit()
+ audit_log(user, "page.guest_revoke", "guest_share", share_id, "", request)
+ return {"status": "revoked", "id": share_id}
+
+
+_GUEST_404 = """
+
+Link unavailable — FlowDeck
+
+
This link is unavailable
+
It may have expired, been revoked, or never existed.
+Ask the person who shared it with you for a new link.
"""
+
+
+@router.get("/g/{token}", response_class=HTMLResponse)
+async def guest_page(token: str, request: Request):
+ """Account-less page access (read-only or commenter). 404 if inactive."""
+ share = wiki.resolve_guest_share(token)
+ if not share:
+ return HTMLResponse(_GUEST_404, status_code=404)
+ with get_conn() as conn:
+ page = conn.execute("SELECT id, title, content, created_at, updated_at, deleted_at"
+ " FROM pages WHERE id=?", (share["page_id"],)).fetchone()
+ if not page or page["deleted_at"]:
+ return HTMLResponse(_GUEST_404, status_code=404)
+ wiki.record_view(share["page_id"])
+ return f"""
+
+{page['title']} — FlowDeck guest
+
+
You are viewing this page as a guest
+({share['role']}{' — expires ' + str(share['expires_at']) if share['expires_at'] else ''}).
+Editing is disabled.
+
{page['title']}
{page['content'] or ''}
"""
+
+
+# ── page views ─────────────────────────────────────────────────────────────
+
+@router.get("/api/v2/wiki/pages/{page_id}/views")
+async def page_views(page_id: int, request: Request):
+ user = _user(request)
+ page = _page_or_404(page_id)
+ if page.get("teamspace_id") and not wiki.can_write_teamspace(user["id"], page["teamspace_id"]):
+ raise HTTPException(403, "Editor role required to read analytics")
+ return wiki.view_stats(page_id, days=request.query_params.get("days", 30))
+
+
+# ── wiki home ──────────────────────────────────────────────────────────────
+
+@router.get("/api/v2/wiki/home")
+async def wiki_home(request: Request):
+ """Aggregated knowledge home: verified pages + recents + teamspaces."""
+ user = _user(request)
+ wid = _workspace_id(request)
+ with get_conn() as conn:
+ recents = conn.execute(
+ """SELECT id, title, page_icon, updated_at FROM pages
+ WHERE workspace_id=? AND deleted_at IS NULL
+ ORDER BY updated_at DESC LIMIT 20""", (wid,)).fetchall()
+ verified = conn.execute(
+ """SELECT v.page_id, v.verified_at, v.expires_at FROM page_verifications v
+ JOIN pages p ON p.id = v.page_id
+ WHERE p.workspace_id=? AND p.deleted_at IS NULL
+ AND (v.expires_at IS NULL OR v.expires_at > ?)""",
+ (wid, __import__("datetime").datetime.now(
+ __import__("datetime").timezone.utc).replace(microsecond=0).isoformat()),
+ ).fetchall()
+ return {"workspace_id": wid,
+ "teamspaces": wiki.list_teamspaces(user["id"], wid),
+ "verified": [dict(r) for r in verified],
+ "recents": [dict(r) for r in recents]}
+
+
+@router.post("/api/v2/wiki/verify-expiry-sweep")
+async def sweep_expiry(request: Request):
+ """Notify verifiers whose ✅ expires within 7 days (idempotent-ish job)."""
+ user = _user(request)
+ with get_conn() as conn:
+ if not conn.execute("SELECT is_admin FROM users WHERE id=?", (user["id"],)).fetchone()[0]:
+ raise HTTPException(403, "Admin required")
+ sent = 0
+ for row in wiki.expiring_verifications(days=7):
+ create_notification(row["owner_id"], None, "page.verification_expiring",
+ "Verification expiring soon",
+ f"“{row['title']}” loses its verified badge on {row['expires_at']}.",
+ resource_type="page", resource_id=row["page_id"])
+ sent += 1
+ return {"notified": sent}
+
+
+# ── blocks (mermaid / equation_inline / progress) ───────────────────────────
+
+@router.post("/api/v2/wiki/blocks/preview")
+async def preview_blocks(request: Request):
+ """Render v7.3 blocks to HTML (same renderer used by the export pipeline)."""
+ _user(request)
+ try:
+ body = await request.json()
+ except Exception:
+ body = {}
+ blocks = body.get("blocks")
+ if not isinstance(blocks, list):
+ raise HTTPException(400, "blocks must be a list")
+ if len(blocks) > 200:
+ raise HTTPException(400, "max 200 blocks per preview")
+ from app.services.wiki_blocks import mmdc_available, render_block
+ out = [{"type": b.get("type"), "html": render_block(b)} for b in blocks
+ if isinstance(b, dict) and b.get("type") in ("mermaid", "equation_inline", "progress")]
+ return {"rendered": out, "mmdc_available": mmdc_available()}
diff --git a/app/routers/workers.py b/app/routers/workers.py
new file mode 100644
index 0000000..92ee4f6
--- /dev/null
+++ b/app/routers/workers.py
@@ -0,0 +1,216 @@
+"""FlowDeck — Workers API (v7.0.0): CRUD, manual run, history, fork, usage."""
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import JSONResponse
+
+from app.auth.session import SessionManager
+from app.db import get_conn
+from app.services import workers as worker_service
+from app.services.api_v2_helpers import (
+ audit_log,
+ has_scope,
+ paginate_headers,
+ parse_pagination,
+ resolve_bearer_token,
+ row_to_dict,
+)
+
+router = APIRouter(tags=["workers"])
+
+
+def _auth_user(request: Request, *, require_write: bool = False) -> dict:
+ sess = SessionManager.decode_session(request.cookies.get("flowdeck_session", ""))
+ if sess:
+ return sess
+ auth = request.headers.get("authorization") or request.headers.get("Authorization") or ""
+ if auth.lower().startswith("bearer "):
+ user = resolve_bearer_token(auth[7:].strip())
+ if not user:
+ raise HTTPException(401, "Invalid or expired API token")
+ if require_write and not has_scope(user.get("_token_scopes") or "read", "write"):
+ raise HTTPException(403, "Insufficient scope. Required: write")
+ return user
+ raise HTTPException(401, "Authentication required")
+
+
+def _row_to_api(row) -> dict:
+ d = row_to_dict(row)
+ d.pop("code_py", None) # code only via ?include_code=1 or owner fetch
+ return d
+
+
+@router.post("/api/v2/workers")
+async def create_worker(request: Request):
+ user = _auth_user(request, require_write=True)
+ try:
+ body = await request.json()
+ except Exception:
+ body = {}
+ name = (body.get("name") or "Untitled worker").strip()[:200]
+ code = body.get("code_py") or ""
+ try:
+ worker_service.validate_code(code)
+ except worker_service.WorkerRejected as exc:
+ raise HTTPException(400, f"code rejected: {exc}") from None
+ slug = worker_service.unique_slug(body.get("slug") or name)
+ with get_conn() as conn:
+ cur = conn.execute(
+ """INSERT INTO workers (slug, workspace_id, name, code_py, schedule_cron,
+ shared, daily_budget_s, created_by)
+ VALUES (?,?,?,?,?,?,?,?)""",
+ (slug, body.get("workspace_id"), name, code,
+ (body.get("schedule_cron") or "")[:60],
+ 1 if body.get("shared") else 0,
+ max(1, min(int(body.get("daily_budget_s") or 60), 3600)),
+ user["id"]))
+ conn.commit()
+ wid = cur.lastrowid
+ row = conn.execute("SELECT * FROM workers WHERE id=?", (wid,)).fetchone()
+ audit_log(user, "worker.create", "worker", wid, slug, request)
+ return JSONResponse(status_code=201, content={**_row_to_api(row), "code_py": code})
+
+
+@router.get("/api/v2/workers")
+async def list_workers(request: Request):
+ _auth_user(request)
+ limit, offset = parse_pagination(request)
+ with get_conn() as conn:
+ total = conn.execute("SELECT COUNT(*) FROM workers").fetchone()[0]
+ rows = conn.execute(
+ "SELECT * FROM workers ORDER BY id DESC LIMIT ? OFFSET ?",
+ (limit, offset)).fetchall()
+ resp = JSONResponse([_row_to_api(r) for r in rows])
+ for k, v in paginate_headers(total).items():
+ resp.headers[k] = v
+ return resp
+
+
+@router.get("/api/v2/workers/{worker_id}")
+async def get_worker(worker_id: int, request: Request):
+ user = _auth_user(request)
+ with get_conn() as conn:
+ row = conn.execute("SELECT * FROM workers WHERE id=?", (worker_id,)).fetchone()
+ if not row:
+ raise HTTPException(404, "Worker not found")
+ out = _row_to_api(row)
+ if (request.query_params.get("include_code") == "1" or row["created_by"] == user["id"]
+ or user.get("is_admin")):
+ out["code_py"] = row["code_py"]
+ return out
+
+
+@router.patch("/api/v2/workers/{worker_id}")
+async def update_worker(worker_id: int, request: Request):
+ user = _auth_user(request, require_write=True)
+ try:
+ body = await request.json()
+ except Exception:
+ body = {}
+ with get_conn() as conn:
+ row = conn.execute("SELECT * FROM workers WHERE id=?", (worker_id,)).fetchone()
+ if not row:
+ raise HTTPException(404, "Worker not found")
+ if row["created_by"] != user["id"] and not user.get("is_admin"):
+ raise HTTPException(403, "Only the owner can update this worker")
+ updates: dict = {}
+ if "name" in body:
+ updates["name"] = str(body["name"] or "")[:200]
+ if "code_py" in body:
+ try:
+ worker_service.validate_code(body["code_py"] or "")
+ except worker_service.WorkerRejected as exc:
+ raise HTTPException(400, f"code rejected: {exc}") from None
+ updates["code_py"] = body["code_py"] or ""
+ if "schedule_cron" in body:
+ updates["schedule_cron"] = str(body["schedule_cron"] or "")[:60]
+ if "shared" in body:
+ updates["shared"] = 1 if body["shared"] else 0
+ if "daily_budget_s" in body:
+ updates["daily_budget_s"] = max(1, min(int(body["daily_budget_s"] or 60), 3600))
+ if "slug" in body and body["slug"] != row["slug"]:
+ if not worker_service._SLUG_RE.match(str(body["slug"] or "")):
+ raise HTTPException(400, "Invalid slug")
+ if conn.execute("SELECT id FROM workers WHERE slug=? AND id!=?",
+ (body["slug"], worker_id)).fetchone():
+ raise HTTPException(409, "Slug already taken")
+ updates["slug"] = body["slug"]
+ if updates:
+ sets = ", ".join(f"{k}=?" for k in updates)
+ conn.execute(f"UPDATE workers SET {sets}, updated_at=CURRENT_TIMESTAMP WHERE id=?",
+ (*updates.values(), worker_id))
+ conn.commit()
+ row = conn.execute("SELECT * FROM workers WHERE id=?", (worker_id,)).fetchone()
+ audit_log(user, "worker.update", "worker", worker_id, ",".join(updates), request)
+ return _row_to_api(row)
+
+
+@router.delete("/api/v2/workers/{worker_id}")
+async def delete_worker(worker_id: int, request: Request):
+ user = _auth_user(request, require_write=True)
+ with get_conn() as conn:
+ row = conn.execute("SELECT * FROM workers WHERE id=?", (worker_id,)).fetchone()
+ if not row:
+ raise HTTPException(404, "Worker not found")
+ if row["created_by"] != user["id"] and not user.get("is_admin"):
+ raise HTTPException(403, "Only the owner can delete this worker")
+ conn.execute("DELETE FROM workers WHERE id=?", (worker_id,))
+ conn.commit()
+ audit_log(user, "worker.delete", "worker", worker_id, "", request)
+ return {"status": "deleted", "id": worker_id}
+
+
+@router.post("/api/v2/workers/{worker_id}/run")
+async def run_worker_endpoint(worker_id: int, request: Request):
+ user = _auth_user(request, require_write=True)
+ try:
+ body = await request.json() if request.headers.get("content-type") else {}
+ except Exception:
+ body = {}
+ with get_conn() as conn:
+ row = conn.execute("SELECT * FROM workers WHERE id=?", (worker_id,)).fetchone()
+ if not row:
+ raise HTTPException(404, "Worker not found")
+ if (row["created_by"] != user["id"] and not row["shared"]
+ and not user.get("is_admin")):
+ raise HTTPException(403, "Worker is private")
+ import asyncio
+ loop = asyncio.get_running_loop()
+ try:
+ out = await loop.run_in_executor(
+ None, worker_service.run_worker, worker_id, body.get("ctx") or {})
+ except HTTPException:
+ raise
+ audit_log(user, "worker.run", "worker", worker_id, out.get("status", ""), request)
+ return out
+
+
+@router.get("/api/v2/workers/{worker_id}/runs")
+async def worker_runs(worker_id: int, request: Request):
+ _auth_user(request)
+ limit, _offset = parse_pagination(request, default_limit=20)
+ with get_conn() as conn:
+ if not conn.execute("SELECT id FROM workers WHERE id=?", (worker_id,)).fetchone():
+ raise HTTPException(404, "Worker not found")
+ rows = conn.execute(
+ "SELECT * FROM worker_runs WHERE worker_id=? ORDER BY id DESC LIMIT ?",
+ (worker_id, limit)).fetchall()
+ return {"worker_id": worker_id, "runs": [row_to_dict(r) for r in rows]}
+
+
+@router.post("/api/v2/workers/{worker_id}/fork")
+async def fork_worker_endpoint(worker_id: int, request: Request):
+ user = _auth_user(request, require_write=True)
+ out = worker_service.fork_worker(worker_id, user["id"])
+ audit_log(user, "worker.fork", "worker", worker_id, "", request)
+ return JSONResponse(status_code=201, content=out)
+
+
+@router.get("/api/v2/workers-usage")
+async def workers_usage(request: Request):
+ user = _auth_user(request)
+ ws_raw = request.query_params.get("workspace_id")
+ wid = int(ws_raw) if ws_raw and str(ws_raw).isdigit() else None
+ return {"workspace_id": wid,
+ "used_seconds_today": round(worker_service.daily_usage_s(wid), 2),
+ "user_id": user.get("id")}
diff --git a/app/services/agent_engine.py b/app/services/agent_engine.py
index e619e73..a80202b 100644
--- a/app/services/agent_engine.py
+++ b/app/services/agent_engine.py
@@ -18,9 +18,10 @@ import re
from app.config import settings
from app.db import get_conn
+from app.services.agent_policies import check_tool, get_policy
from app.services.context_builder import ContextBuilder
from app.services.llm_client import LLMClient
-from app.services.permission_manager import PermissionManager
+from app.services.permission_manager import WRITE_TOOLS, PermissionManager
from app.services.tool_registry import ToolRegistry
logger = logging.getLogger(__name__)
@@ -185,7 +186,10 @@ class AgentEngine:
used_model = model or "" # peut être ajusté par un repli de modèle (404/410)
try:
- for _step in range(settings.agent_max_iterations or MAX_ITERATIONS):
+ # v7.2.0 — the workspace policy may cap iterations below the global max.
+ policy_max = get_policy(self.workspace_id).get("max_steps") or MAX_ITERATIONS
+ iterations = min(settings.agent_max_iterations or MAX_ITERATIONS, policy_max)
+ for _step in range(iterations):
if self._tokens >= settings.agent_max_tokens_budget:
yield self._event("error", {"message": "Budget de tokens dépassé"})
break
@@ -235,8 +239,30 @@ class AgentEngine:
tool, args = call["name"], call.get("arguments") or {}
call_id = tool_specs[idx]["id"]
denied = False
+ # v7.2.0 — workspace tool scope + human approval gate, checked
+ # *before* permissions so a scoped-out tool never reaches ACLs.
+ gov = check_tool(self.user_id, self.workspace_id, tool,
+ is_write=tool in WRITE_TOOLS,
+ conversation_id=conversation_id)
+ if not gov.get("allowed"):
+ detail = gov.get("reason") or "Refusé par la politique agent"
+ if gov.get("approval_id"):
+ detail = (f"Approbation requise (demande #{gov['approval_id']}) "
+ f"— action suspendue")
+ yield self._event("action", {
+ "tool": tool, "status": "approval_required" if gov.get("approval_id")
+ else "error", "detail": detail,
+ "approval_id": gov.get("approval_id")})
+ self._log_action(conversation_id, tool, args, {}, "error", detail=detail)
+ messages.append({
+ "role": "tool", "tool_call_id": call_id,
+ "content": json.dumps({"status": "error", "message": detail},
+ ensure_ascii=False),
+ })
+ denied = True
try:
- self.perms.assert_can(tool, args, self.workspace_id, approval_mode)
+ if not denied:
+ self.perms.assert_can(tool, args, self.workspace_id, approval_mode)
except Exception as exc: # permission / approval guard
detail = self._exc_detail(exc)
yield self._event("action", {"tool": tool, "status": "error", "detail": detail})
diff --git a/app/services/agent_policies.py b/app/services/agent_policies.py
new file mode 100644
index 0000000..afc8cd9
--- /dev/null
+++ b/app/services/agent_policies.py
@@ -0,0 +1,89 @@
+"""FlowDeck — agent governance (v7.2.0): workspace tool scope + approval gate.
+
+``agent_policies``: ``allowed_tools_json`` (null = all tools), ``max_steps``,
+``require_approval`` (write tools pause for a human). ``check_tool()`` is
+consulted by ``AgentEngine`` before ``PermissionManager.assert_can``.
+``agent.run.approval_requested`` is emitted on the outbound webhook bus so
+external systems can subscribe. See ``docs/V72_Enterprise_SCIM_2FA.md``.
+"""
+from __future__ import annotations
+
+import json
+
+from app.db import get_conn
+
+
+def get_policy(workspace_id: int | None) -> dict:
+ """Effective policy (workspace row, else global row, else defaults)."""
+ with get_conn() as conn:
+ row = None
+ if workspace_id is not None:
+ row = conn.execute("SELECT * FROM agent_policies WHERE workspace_id=?",
+ (workspace_id,)).fetchone()
+ if row is None:
+ row = conn.execute("SELECT * FROM agent_policies WHERE workspace_id IS NULL"
+ ).fetchone()
+ if not row:
+ return {"allowed_tools": None, "max_steps": 12, "require_approval": False}
+ d = dict(row)
+ try:
+ allowed = json.loads(d.get("allowed_tools_json")) if d.get("allowed_tools_json") else None
+ except (TypeError, ValueError):
+ allowed = None
+ return {"allowed_tools": allowed, "max_steps": d.get("max_steps") or 12,
+ "require_approval": bool(d.get("require_approval"))}
+
+
+def check_tool(user_id: int, workspace_id: int | None, tool: str,
+ is_write: bool, conversation_id: int = 0) -> dict:
+ """Policy gate for one tool call.
+
+ Returns {allowed: bool, approval_id: int|None}. Denied tools and gated
+ writes (pending approval) return allowed=False; the engine renders both
+ as action errors without executing.
+ """
+ from app.services.permission_manager import WRITE_TOOLS
+ policy = get_policy(workspace_id)
+ allowed = policy["allowed_tools"]
+ if allowed is not None and tool not in set(allowed):
+ return {"allowed": False, "approval_id": None, "reason": "tool not in policy scope"}
+ if is_write or tool in WRITE_TOOLS:
+ if policy["require_approval"]:
+ with get_conn() as conn:
+ cur = conn.execute(
+ """INSERT INTO agent_approvals
+ (conversation_id, tool, args_json, status, requester_id)
+ VALUES (?,?,?,?,?)""",
+ (conversation_id, tool, "{}", "pending", user_id))
+ conn.commit()
+ approval_id = cur.lastrowid
+ try:
+ import asyncio
+
+ from app.services.webhook_outbound import fire_event as _fire
+ try:
+ loop = asyncio.get_running_loop()
+ except RuntimeError:
+ loop = None
+ if loop is not None:
+ loop.create_task(_fire("agent.run.approval_requested", {
+ "approval_id": approval_id, "tool": tool,
+ "conversation_id": conversation_id}))
+ except Exception: # noqa: BLE001 — webhook never blocks policy
+ pass
+ return {"allowed": False, "approval_id": approval_id,
+ "reason": "approval requested"}
+ return {"allowed": True, "approval_id": None, "reason": ""}
+
+
+def decide_approval(approval_id: int, approver_id: int, approve: bool) -> dict | None:
+ with get_conn() as conn:
+ row = conn.execute("SELECT * FROM agent_approvals WHERE id=?",
+ (approval_id,)).fetchone()
+ if not row or row["status"] != "pending":
+ return None
+ conn.execute("UPDATE agent_approvals SET status=?, approver_id=? WHERE id=?",
+ ("approved" if approve else "rejected", approver_id, approval_id))
+ conn.commit()
+ return dict(conn.execute("SELECT * FROM agent_approvals WHERE id=?",
+ (approval_id,)).fetchone())
diff --git a/app/services/automations.py b/app/services/automations.py
index b5fac53..43f30a9 100644
--- a/app/services/automations.py
+++ b/app/services/automations.py
@@ -23,6 +23,7 @@ from __future__ import annotations
import asyncio
import json
import logging
+import time
from datetime import datetime, timedelta
import httpx
@@ -246,6 +247,43 @@ async def _run_action(action: dict, context: dict, trigger_source: str) -> str:
)
return f"notified user {user_id}"
+ if atype == "slack":
+ url = _secret_value(action.get("webhook_url") or action.get("url") or "")
+ if not url:
+ raise ValueError("slack action requires a webhook_url")
+ _, text = _maybe_convert_prediction(
+ action.get("text") or action.get("message") or "Automation fired", context)
+ return await _post_slack(url, text)
+
+ if atype == "email":
+ to = action.get("to", "")
+ _, subject = _maybe_convert_prediction(action.get("subject", "FlowDeck automation"), context)
+ _, body = _maybe_convert_prediction(action.get("body", action.get("message", "")), context)
+ return await _send_email_action(to, subject, body, context)
+
+ if atype == "forge_issue":
+ provider = (action.get("provider") or "gitea").lower()
+ owner = action.get("owner", "")
+ repo = action.get("repo", "")
+ if not owner or not repo:
+ raise ValueError("forge_issue requires owner + repo")
+ _, title = _maybe_convert_prediction(action.get("title", "Automation issue"), context)
+ _, body = _maybe_convert_prediction(action.get("body", ""), context)
+ return await _create_forge_issue(
+ provider, owner, repo, title, body,
+ labels=action.get("labels") or [],
+ user_id=context.get("created_by"),
+ )
+
+ if atype == "agent_trigger":
+ agent_id = action.get("agent_id")
+ if not agent_id:
+ raise ValueError("agent_trigger requires an agent_id")
+ _, message = _maybe_convert_prediction(action.get("message", ""), context)
+ return await _run_linked_agent(
+ int(agent_id), context.get("created_by") or 1,
+ context.get("workspace_id"), message, context)
+
raise ValueError(f"unknown action type: {atype!r}")
@@ -260,6 +298,11 @@ async def run_automation(automation_id: int, trigger_source: str, context: dict)
if not auto["enabled"]:
return {"status": "skipped", "detail": "automation disabled"}
+ # v7.0.0: chained steps take over when present (legacy path otherwise).
+ stepped = await _maybe_run_stepped(auto, trigger_source, context)
+ if stepped is not None:
+ return stepped
+
props = context.get("properties")
before = context.get("before_properties")
if not evaluate_conditions(auto["condition_json"], props, before):
@@ -307,6 +350,15 @@ async def run_automation(automation_id: int, trigger_source: str, context: dict)
async def fire_event(event: str, payload: dict):
"""Dispatch an event to outbound webhooks and matching automations."""
+ # v7.3.0: page.updated → in-app notification to followers (throttled).
+ if event == "page.updated":
+ try:
+ from app.services.wiki import notify_followers_of_page_update
+ notify_followers_of_page_update(
+ payload.get("page_id"), payload.get("actor_id"),
+ payload.get("title") or "")
+ except Exception: # noqa: BLE001 — notifications are best-effort
+ logger.debug("followers notification failed for page.updated")
# Outbound webhooks (v2.1.0 machinery, previously called nowhere).
try:
from app.services.webhook_outbound import fire_event as fire_webhooks
@@ -320,14 +372,29 @@ async def fire_event(event: str, payload: dict):
WHERE trigger_type='event' AND event=? AND enabled=1""",
(event,),
).fetchall()
+ stepped_ids: set[int] = set()
+ try:
+ with get_conn() as _c:
+ stepped_ids = {r[0] for r in _c.execute(
+ "SELECT DISTINCT automation_id FROM automation_steps").fetchall()}
+ except Exception: # noqa: BLE001 — table missing on very old DBs
+ pass
for row in rows:
auto = dict(row)
+ if auto["id"] in stepped_ids:
+ continue # v7.0.0: handled by fire_stepped_event below (no double run)
if auto["collection_id"] and payload.get("collection_id") != auto["collection_id"]:
continue
context = dict(payload)
context["event"] = event
await run_automation(auto["id"], "event", context)
+ # v7.0.0: step-based automations (multi-trigger any/all, chains).
+ try:
+ await fire_stepped_event(event, payload)
+ except Exception: # noqa: BLE001
+ logger.debug("stepped dispatch failed for %s", event)
+
# ═══════════ Cron scheduling (trigger_type='cron') ═══════════
@@ -409,6 +476,353 @@ async def automation_scheduler():
await run_automation(auto["id"], "cron", context)
except Exception: # noqa: BLE001
logger.warning("Cron automation %s errored", auto["id"])
+ # v7.0.0: workers on a cron schedule share the same 60s loop.
+ try:
+ from app.services.workers import run_due_workers
+ await run_due_workers()
+ except Exception: # noqa: BLE001
+ logger.debug("worker cron iteration failed")
except Exception: # noqa: BLE001
logger.warning("automation_scheduler iteration failed")
await asyncio.sleep(60)
+
+
+# ═══════════ v7.0.0 — multi-step automations (triggers/conditions/delay) ══
+
+STEP_KINDS = ("trigger", "condition", "delay", "action")
+STEP_ACTION_TYPES = ("webhook", "set_property", "create_page", "notify",
+ "slack", "email", "forge_issue", "agent_trigger")
+ALL_MODE_WINDOW_S = 300.0
+
+# mode=all bookkeeping (single-process): automation_id -> {event: timestamp}.
+_ALL_PENDING: dict[int, dict[str, float]] = {}
+
+
+def reset_all_pending() -> None:
+ """Test helper: clear the mode=all arrival window."""
+ _ALL_PENDING.clear()
+
+
+def _secret_value(stored: str | None) -> str:
+ """Decrypt a Fernet secret, falling back to raw plaintext (legacy/tests)."""
+ if not stored:
+ return ""
+ try:
+ from app.services.sso_provisioning import decrypt_secret
+ decrypted = decrypt_secret(stored)
+ if decrypted:
+ return decrypted
+ except Exception: # noqa: BLE001
+ pass
+ if isinstance(stored, str) and not stored.startswith("gAAAAA"):
+ return stored
+ return ""
+
+
+def validate_step(kind: str, config: dict) -> None:
+ """Validate a step payload. Raises ValueError with a human message."""
+ from fastapi import HTTPException
+ if kind not in STEP_KINDS:
+ raise HTTPException(400, f"invalid kind: {kind!r} (want trigger|condition|delay|action)")
+ config = config or {}
+ if kind == "trigger":
+ if not config.get("event"):
+ raise HTTPException(400, "trigger step requires an event")
+ elif kind == "condition":
+ if config.get("op", "eq") not in COND_OPS:
+ raise HTTPException(400, f"invalid op: {config.get('op')!r}")
+ elif kind == "delay":
+ try:
+ seconds = int(config.get("seconds", 0))
+ except (TypeError, ValueError):
+ raise HTTPException(400, "delay step requires integer seconds") from None
+ if seconds < 0 or seconds > 86400:
+ raise HTTPException(400, "delay seconds must be 0..86400")
+ elif kind == "action":
+ if config.get("type") not in STEP_ACTION_TYPES:
+ raise HTTPException(400, f"invalid action type: {config.get('type')!r}")
+
+
+def get_steps(automation_id: int) -> list[dict]:
+ """Ordered steps of an automation (empty when legacy single-mode)."""
+ with get_conn() as conn:
+ rows = conn.execute(
+ "SELECT * FROM automation_steps WHERE automation_id=? ORDER BY position, id",
+ (automation_id,),
+ ).fetchall()
+ out = []
+ for r in rows:
+ d = dict(r)
+ try:
+ d["config"] = json.loads(d.get("config_json") or "{}")
+ except (TypeError, json.JSONDecodeError):
+ d["config"] = {}
+ out.append(d)
+ return out
+
+
+def _steps_by_kind(steps: list[dict]) -> dict[str, list[dict]]:
+ grouped: dict[str, list[dict]] = {"trigger": [], "condition": [],
+ "delay": [], "action": []}
+ for s in steps:
+ if s.get("kind") in grouped:
+ grouped[s["kind"]].append(s)
+ return grouped
+
+
+def _step_trigger_matches(step_cfg: dict, event: str, payload: dict,
+ automation_collection_id: int | None) -> bool:
+ if step_cfg.get("event") != event:
+ return False
+ want_coll = step_cfg.get("collection_id") or automation_collection_id
+ if want_coll and payload.get("collection_id") != want_coll:
+ return False
+ return True
+
+
+async def _run_with_steps(auto: dict, steps: list[dict], trigger_source: str,
+ context: dict) -> dict:
+ """Execute a chained automation. Records one run row with per-step detail."""
+ grouped = _steps_by_kind(steps)
+ props = context.get("properties")
+ before = context.get("before_properties")
+ # Legacy single condition still applies on top of step conditions.
+ if not evaluate_conditions(auto.get("condition_json") or "[]", props, before):
+ _save_run(auto["id"], trigger_source, "skipped", "condition not met",
+ context.get("collection_id"), context.get("page_id"))
+ return {"status": "skipped", "detail": "condition not met"}
+ for cond in grouped["condition"]:
+ cfg = cond.get("config") or {}
+ if not match_condition_props(props, before, {
+ "property": cfg.get("property"), "op": cfg.get("op", "eq"),
+ "value": cfg.get("value")}):
+ _save_run(auto["id"], trigger_source, "skipped",
+ f"step condition not met: {cfg.get('property')}",
+ context.get("collection_id"), context.get("page_id"))
+ return {"status": "skipped", "detail": "step condition not met"}
+
+ ctx = dict(context)
+ ctx["automation_name"] = auto["name"]
+ ctx["created_by"] = auto["created_by"] or ctx.get("created_by")
+ ordered = sorted(steps, key=lambda s: (s.get("position", 0), s.get("id", 0)))
+ results = []
+ try:
+ for step in ordered:
+ kind = step.get("kind")
+ cfg = step.get("config") or {}
+ if kind in ("trigger", "condition"):
+ continue
+ if kind == "delay":
+ seconds = max(0, min(int(cfg.get("seconds", 0)), 300))
+ if seconds:
+ await asyncio.sleep(seconds)
+ results.append(f"delay {cfg.get('seconds', 0)}s")
+ elif kind == "action":
+ summary = await _run_action({"type": cfg.get("type"), **cfg}, ctx,
+ trigger_source)
+ results.append(summary)
+ detail = "; ".join(results) or "no steps executed"
+ _save_run(auto["id"], trigger_source, "fired", detail,
+ ctx.get("collection_id"), ctx.get("page_id"))
+ try:
+ from app.services.webhook_outbound import fire_event as _fire_wh
+ await _fire_wh("automation.fired", {
+ "automation_id": auto["id"], "name": auto["name"],
+ "trigger": trigger_source, "collection_id": ctx.get("collection_id"),
+ "page_id": ctx.get("page_id"), "detail": detail})
+ except Exception: # noqa: BLE001
+ logger.debug("automation.fired webhook dispatch failed")
+ return {"status": "fired", "detail": detail}
+ except Exception as exc: # noqa: BLE001
+ logger.warning("Automation %s (steps) failed: %s", auto["id"], exc)
+ _save_run(auto["id"], trigger_source, "error", str(exc),
+ ctx.get("collection_id"), ctx.get("page_id"))
+ return {"status": "error", "detail": str(exc)}
+
+
+async def _maybe_run_stepped(auto: dict, trigger_source: str, context: dict) -> dict | None:
+ """Run via steps when the automation has any; None → use legacy path."""
+ steps = get_steps(auto["id"])
+ if not steps:
+ return None
+ return await _run_with_steps(auto, steps, trigger_source, context)
+
+
+def _match_stepped_automations(event: str, payload: dict) -> list[tuple[dict, list[dict]]]:
+ """Automations (enabled) whose trigger steps match ``event`` + collection."""
+ with get_conn() as conn:
+ rows = conn.execute(
+ """SELECT a.* FROM automations a
+ JOIN automation_steps s ON s.automation_id = a.id
+ WHERE a.enabled=1 AND s.kind='trigger' GROUP BY a.id"""
+ ).fetchall()
+ matched = []
+ for row in rows:
+ auto = dict(row)
+ steps = get_steps(auto["id"])
+ triggers = [s for s in steps if s.get("kind") == "trigger"]
+ if any(_step_trigger_matches(t.get("config") or {}, event, payload,
+ auto.get("collection_id")) for t in triggers):
+ matched.append((auto, triggers))
+ return matched
+
+
+async def fire_stepped_event(event: str, payload: dict) -> None:
+ """Dispatch ``event`` to step-based automations (mode any/all).
+
+ Called from :func:`fire_event` after the legacy matcher. Unknown events
+ (not in the webhook catalogue) still work here — steps are independent
+ from outbound webhooks.
+ """
+ now = time.time()
+ for auto, triggers in _match_stepped_automations(event, payload):
+ # Skip automations already handled by the legacy matcher to avoid
+ # double runs (legacy = trigger_type event + no steps).
+ if not get_steps(auto["id"]):
+ continue
+ mode = (auto.get("trigger_mode") or "any").lower()
+ if mode == "all":
+ pending = _ALL_PENDING.setdefault(auto["id"], {})
+ pending[event] = now
+ # Expire arrivals outside the window.
+ for ev in [e for e, ts in pending.items() if now - ts > ALL_MODE_WINDOW_S]:
+ del pending[ev]
+ wanted = {t.get("config", {}).get("event") for t in triggers}
+ if not wanted <= set(pending):
+ continue
+ _ALL_PENDING.pop(auto["id"], None)
+ context = dict(payload)
+ context["event"] = event
+ await run_automation(auto["id"], "event", context)
+
+
+# ── v7.0.0 action backends (module-level = monkeypatchable in tests) ───────
+
+async def _post_slack(webhook_url: str, text: str) -> str:
+ async with httpx.AsyncClient(timeout=10) as client:
+ resp = await client.post(webhook_url, json={"text": text})
+ if resp.status_code >= 400:
+ raise RuntimeError(f"slack webhook returned HTTP {resp.status_code}")
+ return f"slack → ({resp.status_code})"
+
+
+async def _send_email_action(to: str, subject: str, body: str, context: dict) -> str:
+ from app.services import mailer
+ address = (to or "").strip()
+ if address.startswith("user:"):
+ try:
+ uid = int(address.split(":", 1)[1])
+ except ValueError:
+ raise ValueError(f"bad email target: {to!r}") from None
+ with get_conn() as conn:
+ row = conn.execute("SELECT email FROM users WHERE id=?", (uid,)).fetchone()
+ address = (row["email"] if row and row["email"] else "")
+ if not address:
+ raise ValueError(f"user {uid} has no email")
+ if not address:
+ address = None
+ with get_conn() as conn:
+ row = conn.execute("SELECT email FROM users WHERE id=?",
+ (context.get("created_by") or 1,)).fetchone()
+ if row and row["email"]:
+ address = row["email"]
+ if not address:
+ return "email skipped (no recipient)"
+ ok = mailer.send_email(address, subject or "FlowDeck automation", body or "")
+ return f"email → {address}" if ok else "email skipped (SMTP not configured)"
+
+
+async def _create_forge_issue(provider: str, owner: str, repo: str, title: str,
+ body: str, labels: list | None = None,
+ user_id: int | None = None) -> str:
+ token = ""
+ if user_id:
+ with get_conn() as conn:
+ row = conn.execute(
+ "SELECT access_token FROM user_oauth_tokens WHERE user_id=? AND provider=?",
+ (user_id, provider)).fetchone()
+ token = (row["access_token"] if row else "") or ""
+ if provider == "github":
+ if not token:
+ raise ValueError("github action needs a linked GitHub account (token)")
+ async with httpx.AsyncClient(timeout=15) as client:
+ resp = await client.post(
+ f"https://api.github.com/repos/{owner}/{repo}/issues",
+ headers={"Authorization": f"Bearer {token}",
+ "Accept": "application/vnd.github+json"},
+ json={"title": title, "body": body,
+ "labels": labels or []} if labels else {"title": title, "body": body},
+ )
+ if resp.status_code >= 400:
+ raise RuntimeError(f"github returned HTTP {resp.status_code}")
+ return f"github issue #{resp.json().get('number')} in {owner}/{repo}"
+ # gitea (default)
+ from app.services.gitea_client import GiteaClient
+ gitea = GiteaClient(user_token=token or None)
+ issue = await gitea.create_issue(owner, repo, title, body)
+ return f"gitea issue #{issue.get('number')} in {owner}/{repo}"
+
+
+async def _run_linked_agent(agent_id: int, user_id: int, workspace_id: int | None,
+ message: str, context: dict) -> str:
+ from app.services.agent_engine import AgentEngine
+ with get_conn() as conn:
+ agent = conn.execute("SELECT * FROM agents WHERE id=?", (agent_id,)).fetchone()
+ if not agent:
+ raise ValueError(f"agent {agent_id} not found")
+ cur = conn.execute(
+ """INSERT INTO agent_conversations (agent_id, user_id, title, context_json)
+ VALUES (?,?,?,?)""",
+ (agent_id, user_id,
+ f"Automation: {context.get('automation_name', 'run')}",
+ json.dumps({"workspace_id": workspace_id})),
+ )
+ conv_id = cur.lastrowid
+ conn.commit()
+ objective = ((agent["system_instructions"] or "").strip()
+ or f"Exécute l'agent « {agent['name']} ».")
+ if message:
+ objective = f"{objective}\n\n{message}"
+ engine = AgentEngine(user_id, workspace_id, agent["model"] or None)
+ final = ""
+ async for _ev in engine.run(conv_id, objective, model=agent["model"]):
+ pass
+ with get_conn() as conn:
+ row = conn.execute(
+ "SELECT content FROM agent_messages WHERE conversation_id=? AND role='assistant'"
+ " ORDER BY id DESC LIMIT 1", (conv_id,)).fetchone()
+ final = (row["content"][:300] if row and row["content"] else "")
+ return f"agent « {agent['name']} » ran (conversation {conv_id})" + (f": {final}" if final else "")
+
+
+# ── v7.0.0 native DB button ────────────────────────────────────────────────
+
+async def press_button(collection_id: int, row_id: int, prop_ref: str | int,
+ user_id: int) -> dict:
+ """Run the automation linked to a ``button`` property cell."""
+ with get_conn() as conn:
+ if isinstance(prop_ref, int) or str(prop_ref).isdigit():
+ prop = conn.execute(
+ "SELECT * FROM collection_properties WHERE id=? AND collection_id=?",
+ (int(prop_ref), collection_id)).fetchone()
+ else:
+ prop = conn.execute(
+ "SELECT * FROM collection_properties WHERE collection_id=? AND name=?",
+ (collection_id, prop_ref)).fetchone()
+ if not prop:
+ raise ValueError("button property not found")
+ prop = dict(prop)
+ if prop.get("prop_type") != "button":
+ raise ValueError("property is not a button")
+ auto_id = prop.get("button_automation_id")
+ if not auto_id:
+ raise ValueError("button has no linked automation")
+ row = conn.execute(
+ "SELECT id FROM collection_pages WHERE id=? AND collection_id=?",
+ (row_id, collection_id)).fetchone()
+ if not row:
+ raise ValueError("row not found")
+ context = get_page_context(row_id, collection_id)
+ context["created_by"] = user_id
+ return await run_automation(auto_id, "button", context)
+
diff --git a/app/services/calendar_sync.py b/app/services/calendar_sync.py
new file mode 100644
index 0000000..ed994fd
--- /dev/null
+++ b/app/services/calendar_sync.py
@@ -0,0 +1,528 @@
+"""FlowDeck — external calendar sync (v7.1.0).
+
+Bidirectional sync between a collection (date property) and an external
+calendar: Google Calendar (REST) or any CalDAV server (raw REPORT/PUT, no
+extra dependency). Tokens are Fernet-encrypted at rest.
+
+Matching: ``collection_pages.external_event_id`` ↔ remote event id.
+Conflicts (both sides changed since ``last_sync``): last-write-wins +
+in-app ``calendar.conflict`` notification (manual edit resolves).
+
+See ``docs/V71_Calendar_Meetings.md``.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+import uuid
+from datetime import UTC, datetime, timedelta
+
+import httpx
+
+from app.db import get_conn
+
+logger = logging.getLogger(__name__)
+
+PROVIDERS = ("google", "caldav")
+SYNC_LOOKBACK_DAYS = 30
+SYNC_LOOKAHEAD_DAYS = 90
+
+
+class SyncError(RuntimeError):
+ """Raised when the remote calendar cannot be reached/authorized."""
+
+
+# ── links ──────────────────────────────────────────────────────────────────
+
+def _encrypt_tokens(creds: dict) -> str:
+ from app.services.sso_provisioning import encrypt_secret
+ return encrypt_secret(json.dumps(creds or {}))
+
+
+def _decrypt_tokens(tokens_enc: str) -> dict:
+ if not tokens_enc:
+ return {}
+ try:
+ from app.services.sso_provisioning import decrypt_secret
+ raw = decrypt_secret(tokens_enc)
+ if raw:
+ return json.loads(raw)
+ except Exception: # noqa: BLE001
+ pass
+ try: # legacy plaintext (tests)
+ data = json.loads(tokens_enc)
+ return data if isinstance(data, dict) else {}
+ except Exception: # noqa: BLE001
+ return {}
+
+
+def save_link(user_id: int, provider: str, collection_id: int,
+ credentials: dict, calendar_id: str = "primary",
+ date_property: str = "") -> dict:
+ if provider not in PROVIDERS:
+ raise ValueError(f"provider must be google|caldav, got {provider!r}")
+ with get_conn() as conn:
+ if not conn.execute("SELECT id FROM collections WHERE id=?",
+ (collection_id,)).fetchone():
+ raise ValueError("collection not found")
+ cur = conn.execute(
+ """INSERT INTO calendar_links
+ (user_id, provider, tokens_enc, calendar_id, collection_id, date_property)
+ VALUES (?,?,?,?,?,?)
+ ON CONFLICT(user_id, provider, calendar_id) DO UPDATE SET
+ tokens_enc=excluded.tokens_enc, collection_id=excluded.collection_id,
+ date_property=excluded.date_property""",
+ (user_id, provider, _encrypt_tokens(credentials),
+ calendar_id or "primary", collection_id, date_property or ""))
+ conn.commit()
+ row = conn.execute(
+ "SELECT * FROM calendar_links WHERE user_id=? AND provider=? AND calendar_id=?",
+ (user_id, provider, calendar_id or "primary")).fetchone()
+ _ = cur
+ out = dict(row)
+ out.pop("tokens_enc", None)
+ return out
+
+
+def list_links(user_id: int) -> list[dict]:
+ with get_conn() as conn:
+ rows = conn.execute(
+ "SELECT id, user_id, provider, calendar_id, collection_id,"
+ " date_property, last_sync, created_at FROM calendar_links WHERE user_id=?"
+ " ORDER BY id", (user_id,)).fetchall()
+ return [dict(r) for r in rows]
+
+
+def delete_link(user_id: int, link_id: int) -> bool:
+ with get_conn() as conn:
+ cur = conn.execute("DELETE FROM calendar_links WHERE id=? AND user_id=?",
+ (link_id, user_id))
+ conn.commit()
+ return cur.rowcount > 0
+
+
+def _load_link(link_id: int) -> dict | None:
+ with get_conn() as conn:
+ row = conn.execute("SELECT * FROM calendar_links WHERE id=?", (link_id,)).fetchone()
+ return dict(row) if row else None
+
+
+# ── remote I/O (module-level = monkeypatchable) ────────────────────────────
+
+def _remote_event(eid: str, title: str, start: str, description: str = "",
+ updated: str = "") -> dict:
+ return {"id": str(eid), "title": title or "Untitled", "start": start,
+ "description": description or "", "updated": updated or ""}
+
+
+async def google_list_events(tokens: dict, calendar_id: str,
+ time_min: str, time_max: str) -> list[dict]:
+ access = tokens.get("access_token", "")
+ if not access:
+ raise SyncError("google link has no access_token — relink the calendar")
+ url = (f"https://www.googleapis.com/calendar/v3/calendars/{calendar_id}"
+ f"/events?singleEvents=true&orderBy=startTime"
+ f"&timeMin={time_min}&timeMax={time_max}")
+ async with httpx.AsyncClient(timeout=15) as client:
+ resp = await client.get(url, headers={"Authorization": f"Bearer {access}"})
+ if resp.status_code == 401:
+ raise SyncError("google token expired — relink the calendar")
+ if resp.status_code >= 400:
+ raise SyncError(f"google returned HTTP {resp.status_code}")
+ out = []
+ for item in resp.json().get("items", []):
+ start = (item.get("start") or {}).get("dateTime") or (item.get("start") or {}).get("date") or ""
+ out.append(_remote_event(item.get("id", ""), item.get("summary", ""),
+ start, item.get("description", ""),
+ item.get("updated", "")))
+ return out
+
+
+async def google_push_event(tokens: dict, calendar_id: str, event: dict,
+ remote_id: str = "") -> str:
+ access = tokens.get("access_token", "")
+ if not access:
+ raise SyncError("google link has no access_token — relink the calendar")
+ body = {"summary": event.get("title", ""),
+ "description": event.get("description", ""),
+ "start": {"date": event.get("start", "")[:10]},
+ "end": {"date": event.get("start", "")[:10]}}
+ base = f"https://www.googleapis.com/calendar/v3/calendars/{calendar_id}/events"
+ async with httpx.AsyncClient(timeout=15) as client:
+ if remote_id:
+ resp = await client.patch(f"{base}/{remote_id}",
+ headers={"Authorization": f"Bearer {access}"}, json=body)
+ else:
+ resp = await client.post(base, headers={"Authorization": f"Bearer {access}"},
+ json=body)
+ if resp.status_code == 401:
+ raise SyncError("google token expired — relink the calendar")
+ if resp.status_code >= 400:
+ raise SyncError(f"google returned HTTP {resp.status_code}")
+ return str(resp.json().get("id", remote_id or ""))
+
+
+_CALDAV_REPORT = """
+
+
+
+
+
+"""
+
+
+def _parse_caldav_events(xml_text: str) -> list[dict]:
+ """Minimal multistatus → event parser (UID/SUMMARY/DTSTART/DESCRIPTION)."""
+ import re
+ import xml.etree.ElementTree as ET
+ events = []
+ try:
+ root = ET.fromstring(xml_text)
+ except ET.ParseError:
+ return []
+ ns = {"D": "DAV:", "C": "urn:ietf:params:xml:ns:caldav"}
+ for resp in root.findall("D:response", ns):
+ href = resp.findtext("D:href", default="", namespaces=ns)
+ data_el = resp.find(".//{urn:ietf:params:xml:ns:caldav}calendar-data")
+ if data_el is None or not data_el.text:
+ continue
+ ics = data_el.text
+ uid = re.search(r"^UID:(.+)$", ics, re.M)
+ summary = re.search(r"^SUMMARY:(.+)$", ics, re.M)
+ dtstart = re.search(r"^DTSTART(?:;[^:]*)?:(.+)$", ics, re.M)
+ desc = re.search(r"^DESCRIPTION:(.+)$", ics, re.M)
+ events.append(_remote_event(
+ (uid.group(1).strip() if uid else href.strip("/").split("/")[-1]),
+ summary.group(1).strip() if summary else "Untitled",
+ _ics_to_date(dtstart.group(1).strip()) if dtstart else "",
+ desc.group(1).strip() if desc else ""))
+ return events
+
+
+def _ics_to_date(value: str) -> str:
+ value = value.strip()
+ if len(value) >= 8 and value[:8].isdigit():
+ return f"{value[:4]}-{value[4:6]}-{value[6:8]}"
+ return value[:10]
+
+
+def _event_to_ics(uid: str, title: str, date: str, description: str = "") -> str:
+ stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
+ day = (date or "")[:10].replace("-", "")
+ return (f"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//FlowDeck//Sync//EN\r\n"
+ f"BEGIN:VEVENT\r\nUID:{uid}\r\nDTSTAMP:{stamp}\r\nDTSTART;VALUE=DATE:{day}\r\n"
+ f"SUMMARY:{title}\r\nDESCRIPTION:{description}\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n")
+
+
+async def caldav_list_events(creds: dict, time_min: str, time_max: str) -> list[dict]:
+ url = creds.get("url", "")
+ if not url:
+ raise SyncError("caldav link needs a calendar url")
+ auth = (creds.get("username", ""), creds.get("password", ""))
+ body = _CALDAV_REPORT.format(
+ start=time_min.replace("-", "").split("T")[0] + "T000000Z",
+ end=time_max.replace("-", "").split("T")[0] + "T000000Z")
+ async with httpx.AsyncClient(timeout=15, auth=auth if auth[0] else None) as client:
+ resp = await client.request("REPORT", url, content=body,
+ headers={"Depth": "1",
+ "Content-Type": "application/xml"})
+ if resp.status_code == 401:
+ raise SyncError("caldav rejected credentials")
+ if resp.status_code >= 400:
+ raise SyncError(f"caldav returned HTTP {resp.status_code}")
+ return _parse_caldav_events(resp.text)
+
+
+async def caldav_push_event(creds: dict, event: dict, remote_id: str = "") -> str:
+ url = (creds.get("url", "") or "").rstrip("/")
+ if not url:
+ raise SyncError("caldav link needs a calendar url")
+ auth = (creds.get("username", ""), creds.get("password", ""))
+ uid = remote_id or f"flowdeck-{uuid.uuid4().hex}@flowdeck"
+ href = f"{url}/{uid}.ics" if not remote_id else (
+ remote_id if remote_id.startswith("http") else f"{url}/{remote_id}")
+ ics = _event_to_ics(uid.split("@")[0], event.get("title", ""),
+ event.get("start", ""), event.get("description", ""))
+ async with httpx.AsyncClient(timeout=15, auth=auth if auth[0] else None) as client:
+ resp = await client.put(href, content=ics, headers={"Content-Type": "text/calendar"})
+ if resp.status_code >= 400:
+ raise SyncError(f"caldav returned HTTP {resp.status_code}")
+ return uid
+
+
+# ── mapping + sync ─────────────────────────────────────────────────────────
+
+def _date_prop_id(conn, collection_id: int, wanted: str = "") -> tuple[str, str] | None:
+ props = conn.execute(
+ "SELECT id, name FROM collection_properties WHERE collection_id=? ORDER BY position",
+ (collection_id,)).fetchall()
+ if wanted:
+ for p in props:
+ if str(p["id"]) == str(wanted) or p["name"] == wanted:
+ return str(p["id"]), p["name"]
+ return None
+ for p in props:
+ # prop_type lives in the row; fetch full rows only when needed
+ full = conn.execute("SELECT prop_type FROM collection_properties WHERE id=?",
+ (p["id"],)).fetchone()
+ if full and full["prop_type"] == "date":
+ return str(p["id"]), p["name"]
+ return None
+
+
+def _row_date(values: dict, prop_id: str, prop_name: str) -> str:
+ raw = values.get(prop_id, values.get(prop_name, ""))
+ if isinstance(raw, dict):
+ raw = raw.get("date") or raw.get("value") or ""
+ return str(raw or "")
+
+
+def _to_epoch(value: str | None) -> float:
+ if not value:
+ return 0.0
+ text = str(value).strip()
+ try:
+ if text.endswith("Z"):
+ dt = datetime.fromisoformat(text.replace("Z", "+00:00"))
+ else:
+ dt = datetime.fromisoformat(text[:19] if "T" in text else text[:19])
+ if dt.tzinfo is None:
+ dt = dt.replace(tzinfo=UTC)
+ return dt.timestamp()
+ except Exception: # noqa: BLE001
+ try:
+ return time.mktime(time.strptime(text[:10], "%Y-%m-%d"))
+ except Exception: # noqa: BLE001
+ return 0.0
+
+
+def _window() -> tuple[str, str]:
+ now = datetime.now(UTC)
+ start = (now - timedelta(days=SYNC_LOOKBACK_DAYS)).strftime("%Y-%m-%dT00:00:00Z")
+ end = (now + timedelta(days=SYNC_LOOKAHEAD_DAYS)).strftime("%Y-%m-%dT00:00:00Z")
+ return start, end
+
+
+async def sync_link(link_id: int) -> dict:
+ """One bidirectional sync pass. Returns {pulled, pushed, conflicts}."""
+ link = _load_link(link_id)
+ if not link:
+ raise ValueError("link not found")
+ creds = _decrypt_tokens(link.get("tokens_enc") or "")
+ collection_id = link.get("collection_id")
+ if not collection_id:
+ raise ValueError("link has no collection")
+ with get_conn() as conn:
+ date_prop = _date_prop_id(conn, collection_id, link.get("date_property") or "")
+ if not date_prop:
+ raise ValueError("collection has no date property")
+ prop_id, prop_name = date_prop
+
+ tmin, tmax = _window()
+ if link["provider"] == "google":
+ remote = await google_list_events(creds, link.get("calendar_id") or "primary",
+ tmin, tmax)
+ else:
+ remote = await caldav_list_events(creds, tmin, tmax)
+
+ last_sync = _to_epoch(link.get("last_sync"))
+ pulled = pushed = conflicts = 0
+ with get_conn() as conn:
+ rows = conn.execute(
+ "SELECT id, title, property_values_json, updated_at,"
+ " COALESCE(external_event_id, '') AS xid FROM collection_pages"
+ " WHERE collection_id=?", (collection_id,)).fetchall()
+ local = {r["xid"]: dict(r) for r in rows if r["xid"]}
+ seen_remote: set[str] = set()
+ touched: set[int] = set() # rows written by this pull pass — never push back
+
+ for ev in remote:
+ eid = ev.get("id", "")
+ if not eid:
+ continue
+ seen_remote.add(eid)
+ day = (ev.get("start") or "")[:10]
+ if eid not in local:
+ values: dict = {}
+ values[prop_id] = day
+ max_pos = conn.execute(
+ "SELECT COALESCE(MAX(position), -1)+1 FROM collection_pages"
+ " WHERE collection_id=?", (collection_id,)).fetchone()[0]
+ conn.execute(
+ """INSERT INTO collection_pages
+ (collection_id, title, position, property_values_json, external_event_id)
+ VALUES (?,?,?,?,?)""",
+ (collection_id, ev.get("title") or "Untitled", max_pos,
+ json.dumps(values), eid))
+ pulled += 1
+ continue
+ row = local[eid]
+ try:
+ values = json.loads(row["property_values_json"] or "{}")
+ except (TypeError, json.JSONDecodeError):
+ values = {}
+ local_day = _row_date(values, prop_id, prop_name)[:10]
+ remote_newer = _to_epoch(ev.get("updated")) > _to_epoch(row["updated_at"])
+ local_dirty = _to_epoch(row["updated_at"]) > last_sync and local_day != day
+ if remote_newer and local_dirty and local_day and day and local_day != day:
+ # Conflict: both sides moved → last-write-wins + notify.
+ if _to_epoch(ev.get("updated")) >= _to_epoch(row["updated_at"]):
+ values[prop_id] = day
+ conn.execute(
+ "UPDATE collection_pages SET property_values_json=?,"
+ " updated_at=CURRENT_TIMESTAMP WHERE id=?",
+ (json.dumps(values), row["id"]))
+ touched.add(row["id"])
+ conflicts += 1
+ _notify_conflict(conn, link, row, ev)
+ elif day and day != local_day:
+ values[prop_id] = day
+ conn.execute(
+ "UPDATE collection_pages SET property_values_json=?,"
+ " updated_at=CURRENT_TIMESTAMP WHERE id=?",
+ (json.dumps(values), row["id"]))
+ touched.add(row["id"])
+ pulled += 1
+
+ # Push local changes (created locally or edited after last_sync).
+ for xid, row in local.items():
+ if row["id"] in touched:
+ continue
+ if xid in seen_remote:
+ # Edited locally since last sync and remote unchanged → push.
+ if last_sync and _to_epoch(row["updated_at"]) > last_sync:
+ await _push(link, creds, row, prop_id, prop_name, xid)
+ pushed += 1
+ continue
+ # Remote deleted the event → drop the local id (keep the row).
+ conn.execute("UPDATE collection_pages SET external_event_id='' WHERE id=?",
+ (row["id"],))
+ # Rows never linked and recently touched → create remotely.
+ fresh = conn.execute(
+ "SELECT id, title, property_values_json, updated_at FROM collection_pages"
+ " WHERE collection_id=? AND COALESCE(external_event_id, '')=''",
+ (collection_id,)).fetchall()
+ for row in fresh:
+ try:
+ values = json.loads(row["property_values_json"] or "{}")
+ except (TypeError, json.JSONDecodeError):
+ values = {}
+ day = _row_date(values, prop_id, prop_name)[:10]
+ if not day:
+ continue
+ new_id = await _push(link, creds, dict(row), prop_id, prop_name, "")
+ conn.execute("UPDATE collection_pages SET external_event_id=? WHERE id=?",
+ (new_id, row["id"]))
+ pushed += 1
+
+ conn.execute("UPDATE calendar_links SET last_sync=CURRENT_TIMESTAMP WHERE id=?",
+ (link_id,))
+ conn.commit()
+ return {"pulled": pulled, "pushed": pushed, "conflicts": conflicts}
+
+
+async def _push(link: dict, creds: dict, row: dict, prop_id: str,
+ prop_name: str, remote_id: str) -> str:
+ try:
+ values = json.loads(row.get("property_values_json") or "{}")
+ except (TypeError, json.JSONDecodeError):
+ values = {}
+ event = {"title": row.get("title") or "Untitled",
+ "start": _row_date(values, prop_id, prop_name),
+ "description": ""}
+ if link["provider"] == "google":
+ return await google_push_event(creds, link.get("calendar_id") or "primary",
+ event, remote_id)
+ return await caldav_push_event(creds, event, remote_id)
+
+
+def _notify_conflict(conn, link: dict, row: dict, ev: dict) -> None:
+ try:
+ from app.services.notifications import create_notification
+ create_notification(
+ link["user_id"], link["user_id"], "calendar",
+ "Calendar sync conflict",
+ f"« {row.get('title') or 'Untitled'} » changed on both sides;"
+ f" kept the newest ({ev.get('start', '')[:10]}). Edit the row to resolve.",
+ resource_type="collection", resource_id=link.get("collection_id") or 0,
+ url=f"/db/{link.get('collection_id')}", conn=conn, commit=False)
+ except Exception: # noqa: BLE001 — notify must never break sync
+ pass
+
+
+async def calendar_sync_scheduler(interval_seconds: int = 900) -> None:
+ """Background loop: sync every link with a collection (15 min default)."""
+ while True:
+ try:
+ with get_conn() as conn:
+ ids = [r["id"] for r in conn.execute(
+ "SELECT id FROM calendar_links WHERE collection_id IS NOT NULL"
+ ).fetchall()]
+ for link_id in ids:
+ try:
+ await sync_link(link_id)
+ except Exception as exc: # noqa: BLE001 — one link must not kill the loop
+ logger.debug("calendar sync link %s failed: %s", link_id, exc)
+ except Exception as exc: # noqa: BLE001
+ logger.debug("calendar_sync_scheduler: %s", exc)
+ await asyncio.sleep(interval_seconds)
+
+
+# ── free/busy ──────────────────────────────────────────────────────────────
+
+def freebusy(collection_id: int, date_from: str, date_to: str,
+ date_property: str = "") -> dict:
+ """Busy/free weekdays in [date_from, date_to] (day granularity).
+
+ Expands recurrence rules server-side (``recurrence.expand_rule``).
+ """
+ from app.services import recurrence as _rec
+ try:
+ start = datetime.strptime(date_from[:10], "%Y-%m-%d").date()
+ end = datetime.strptime(date_to[:10], "%Y-%m-%d").date()
+ except ValueError:
+ raise ValueError("use YYYY-MM-DD dates") from None
+ if end < start or (end - start).days > 370:
+ raise ValueError("range must be 1..370 days")
+ with get_conn() as conn:
+ date_prop = _date_prop_id(conn, collection_id, date_property)
+ if not date_prop:
+ raise ValueError("collection has no date property")
+ prop_id, prop_name = date_prop
+ rows = conn.execute(
+ "SELECT property_values_json FROM collection_pages WHERE collection_id=?",
+ (collection_id,)).fetchall()
+ busy: set[str] = set()
+ for r in rows:
+ try:
+ values = json.loads(r["property_values_json"] or "{}")
+ except (TypeError, json.JSONDecodeError):
+ continue
+ base = _row_date(values, prop_id, prop_name)
+ if not base:
+ continue
+ rec = (values.get("__recurrence__") or {})
+ rule = rec.get(prop_id) or rec.get(prop_name)
+ if rule:
+ try:
+ for occ in _rec.expand_rule(
+ base, rule, start.isoformat(), end.isoformat()):
+ busy.add(occ[:10])
+ except Exception: # noqa: BLE001 — bad rule, use base date only
+ busy.add(base[:10])
+ else:
+ if start.isoformat() <= base[:10] <= end.isoformat():
+ busy.add(base[:10])
+ days, free = [], []
+ day = start
+ while day <= end:
+ iso = day.isoformat()
+ days.append({"date": iso, "busy": iso in busy,
+ "weekend": day.weekday() >= 5})
+ if iso not in busy and day.weekday() < 5:
+ free.append(iso)
+ day += timedelta(days=1)
+ return {"collection_id": collection_id, "from": start.isoformat(),
+ "to": end.isoformat(), "days": days, "free_weekdays": free}
diff --git a/app/services/export.py b/app/services/export.py
index 13f0e1f..ea74438 100644
--- a/app/services/export.py
+++ b/app/services/export.py
@@ -639,6 +639,10 @@ def blocks_to_html(blocks: list) -> str:
icon = b.get("icon") or "💡"
bg = (b.get("style") or {}).get("bgColor", "#eef2ff")
parts.append(f'
{_text(icon, escape=False)}
{c}
')
+ elif t in ("mermaid", "equation_inline", "progress"):
+ # v7.3.0 blocks — server-rendered so export embeds real content
+ from app.services.wiki_blocks import render_block
+ parts.append(render_block(b))
elif t == "image":
src = b.get("src") or ""
alt = _text(b.get("alt"))
diff --git a/app/services/gitea_client.py b/app/services/gitea_client.py
index 7e5f7b9..8cbe1b5 100644
--- a/app/services/gitea_client.py
+++ b/app/services/gitea_client.py
@@ -73,6 +73,33 @@ class GiteaClient:
self._set_cache(cache_key, data)
return data
+ async def get_repo_info(self, owner: str, repo: str) -> dict:
+ """Repository metadata for an unfurl card (owner/name/branch/…)."""
+ cache_key = f"repo_info:{owner}:{repo}"
+ cached = self._cached(cache_key)
+ if cached:
+ return cached
+ async with httpx.AsyncClient(timeout=15) as client:
+ resp = await client.get(
+ f"{self._base}/repos/{owner}/{repo}",
+ headers=self._headers,
+ )
+ resp.raise_for_status()
+ info = resp.json()
+ repo_info = {
+ "id": info.get("id"),
+ "name": info.get("name"),
+ "owner": (info.get("owner") or {}).get("login", owner),
+ "full_name": info.get("full_name") or f"{owner}/{repo}",
+ "clone_url": info.get("clone_url", ""),
+ "html_url": info.get("html_url", ""),
+ "default_branch": info.get("default_branch", "main"),
+ "description": info.get("description") or "",
+ "language": info.get("language") or "",
+ }
+ self._set_cache(cache_key, repo_info)
+ return repo_info
+
async def get_user_orgs(self) -> list[dict]:
cache_key = "user_orgs"
cached = self._cached(cache_key)
diff --git a/app/services/meetings.py b/app/services/meetings.py
new file mode 100644
index 0000000..c84c9a3
--- /dev/null
+++ b/app/services/meetings.py
@@ -0,0 +1,112 @@
+"""FlowDeck — AI Meeting Notes v2 (v7.1.0).
+
+Upload audio → optional server transcription (``STT_COMMAND``, e.g. whisper)
+→ AI summary (``AIWritingService.summarize``, offline-capable) → fires
+``meeting.summarized`` so custom agents pick it up (Notion 07/2026 pattern).
+
+Without ``STT_COMMAND`` the server stores the audio and accepts a manual
+``transcript`` (client-side transcription). Nothing here requires new pip
+dependencies. See ``docs/V71_Calendar_Meetings.md``.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+from app.db import get_conn
+
+logger = logging.getLogger(__name__)
+
+AUDIO_EXTENSIONS = {"mp3", "wav", "m4a", "ogg", "flac", "aac"}
+MAX_AUDIO_BYTES = 100 * 1024 * 1024
+
+
+class TranscriptionUnavailable(RuntimeError):
+ """Raised when no transcription backend is configured."""
+
+
+def meetings_dir() -> Path:
+ root = Path(os.environ.get("FLOWDECK_DATA_DIR", "/data"))
+ d = root / "uploads" / "meetings"
+ d.mkdir(parents=True, exist_ok=True)
+ return d
+
+
+def save_transcript(page_id: int, transcript: str, language: str = "fr",
+ audio_path: str = "") -> int:
+ with get_conn() as conn:
+ if not conn.execute("SELECT id FROM pages WHERE id=?", (page_id,)).fetchone():
+ raise ValueError("page not found")
+ cur = conn.execute(
+ """INSERT INTO meeting_transcripts (page_id, audio_path, transcript, language)
+ VALUES (?,?,?,?)""",
+ (page_id, audio_path, transcript or "", language or "fr"))
+ conn.commit()
+ return cur.lastrowid
+
+
+def transcribe_audio(audio_path: str, language: str = "fr") -> str:
+ """Transcribe with ``STT_COMMAND`` (``{cmd} {file}` → stdout text).
+
+ Example: ``STT_COMMAND="whisper --language fr --output_format txt --output_dir /tmp"``
+ (command must print or be adapted — stdout is preferred). Raises
+ :class:`TranscriptionUnavailable` when unconfigured.
+ """
+ cmd_template = os.environ.get("STT_COMMAND", "").strip()
+ if not cmd_template:
+ raise TranscriptionUnavailable(
+ "no transcription backend (set STT_COMMAND or POST a manual transcript)")
+ if shutil.which(cmd_template.split()[0]) is None:
+ raise TranscriptionUnavailable(f"STT command not found: {cmd_template.split()[0]}")
+ try:
+ proc = subprocess.run(cmd_template.split() + [audio_path], # noqa: S603 — admin-configured
+ capture_output=True, text=True, timeout=600)
+ except subprocess.TimeoutExpired as exc:
+ raise TranscriptionUnavailable("transcription timed out") from exc
+ text = (proc.stdout or "").strip()
+ if proc.returncode != 0 or not text:
+ raise TranscriptionUnavailable(
+ f"transcription failed: {(proc.stderr or '')[:300]}")
+ return text
+
+
+async def summarize_transcript(transcript_id: int, user_id: int | None = None) -> dict:
+ """Summarize a stored transcript + fire ``meeting.summarized``.
+
+ Returns {transcript_id, summary, offline}. Emits the automation event so
+ custom agents (update tracker, post recap, file tickets) trigger.
+ """
+ from app.services.ai_writing import AIWritingService
+
+ with get_conn() as conn:
+ row = conn.execute("SELECT * FROM meeting_transcripts WHERE id=?",
+ (transcript_id,)).fetchone()
+ if not row:
+ raise ValueError("transcript not found")
+ tr = dict(row)
+ if not (tr.get("transcript") or "").strip():
+ raise ValueError("transcript is empty — transcribe first")
+ svc = AIWritingService(user_id=user_id)
+ res = await svc.run("summarize", context=tr["transcript"])
+ summary = (res.get("text") or "").strip() if res.get("ok") else ""
+ if not summary:
+ raise RuntimeError(f"summarization failed: {res.get('error', 'unknown')}")
+ with get_conn() as conn:
+ conn.execute("UPDATE meeting_transcripts SET summary=? WHERE id=?",
+ (summary, transcript_id))
+ page = conn.execute("SELECT id FROM pages WHERE id=?", (tr["page_id"],)).fetchone()
+ conn.commit()
+ try:
+ from app.services.automations import fire_event
+ await fire_event("meeting.summarized", {
+ "page_id": tr["page_id"] if page else 0,
+ "transcript_id": transcript_id,
+ "language": tr.get("language") or "fr",
+ })
+ except Exception as exc: # noqa: BLE001 — summary stands even if dispatch fails
+ logger.debug("meeting.summarized dispatch failed: %s", exc)
+ return {"transcript_id": transcript_id, "summary": summary,
+ "offline": bool(res.get("offline"))}
diff --git a/app/services/property_types.py b/app/services/property_types.py
index e01fd52..4cd6785 100644
--- a/app/services/property_types.py
+++ b/app/services/property_types.py
@@ -96,6 +96,10 @@ PROPERTY_TYPES: dict[str, dict] = {
"storage": "auto — {id, login}",
"default": None,
},
+ "button": {
+ "storage": "none — runs linked automation (button_automation_id)",
+ "default": None,
+ },
}
# CSV-friendly subset (no relation/rollup/formula)
diff --git a/app/services/semantic_search.py b/app/services/semantic_search.py
new file mode 100644
index 0000000..f77f420
--- /dev/null
+++ b/app/services/semantic_search.py
@@ -0,0 +1,535 @@
+"""FlowDeck — semantic (vector) search + Ask AI (v6.9.0).
+
+Hybrid retrieval = lexical (FTS5/LIKE via :mod:`app.services.search`) fused
+with vector cosine similarity via Reciprocal Rank Fusion, then filtered
+through :class:`PermissionManager` so unauthorized chunks never surface
+(and never enter an LLM prompt).
+
+Vectors use a dependency-free **hashed TF** encoder (``hash-256``): token →
+``md5 % 256`` with L2 normalization. Deterministic, offline-first, good
+enough for recall on small workspaces; the ``embed_texts`` entry point is
+pluggable should an LLM ``/embeddings`` provider be wired later.
+See ``docs/V69_Search_Ask_AI.md``.
+"""
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import logging
+import math
+import re
+import struct
+import time
+
+from app.db import get_conn
+
+logger = logging.getLogger(__name__)
+
+DIM = 256
+MODEL = "hash-256"
+CHUNK_SIZE = 1200
+CHUNK_OVERLAP = 150
+MAX_CHUNKS_PER_RESOURCE = 50
+RRF_K = 60
+
+_TOKEN_RE = re.compile(r"[\wÀ-ÿ]+", flags=re.UNICODE)
+
+# Ask cache: (question_hash, workspace_id, user_id) -> (expires_at, payload)
+_ask_cache: dict[tuple[str, int | None, int], tuple[float, dict]] = {}
+_ASK_CACHE_TTL = 600.0
+
+# Ask rate limit: user_id -> (window_start, count)
+_ask_rate: dict[int, tuple[float, int]] = {}
+_ASK_RATE_MAX = 30
+_ASK_RATE_WINDOW = 60.0
+
+
+# ── text extraction & chunking ─────────────────────────────────────────────
+
+def _blocks_to_text(blocks) -> list[str]:
+ parts: list[str] = []
+
+ def _walk(items) -> None:
+ for b in items or []:
+ if not isinstance(b, dict):
+ continue
+ for key in ("content", "text", "title"):
+ val = b.get(key)
+ if isinstance(val, str) and val.strip():
+ parts.append(val.strip())
+ break
+ children = b.get("children")
+ if isinstance(children, list):
+ _walk(children)
+
+ _walk(blocks if isinstance(blocks, list) else [])
+ return parts
+
+
+def extract_page_text(content: str | None, content_format: str | None) -> str:
+ """Full searchable text of a ``pages`` row (all blocks, recursive)."""
+ if not content:
+ return ""
+ if (content_format or "blocks") == "blocks":
+ try:
+ blocks = json.loads(content)
+ return "\n".join(_blocks_to_text(blocks))
+ except Exception:
+ return content
+ return content
+
+
+def chunk_text(text: str, size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
+ """Split text into overlapping chunks (char-based, word-boundary aware)."""
+ text = (text or "").strip()
+ if not text:
+ return []
+ if len(text) <= size:
+ return [text]
+ chunks: list[str] = []
+ start = 0
+ while start < len(text):
+ end = min(start + size, len(text))
+ if end < len(text):
+ space = text.rfind(" ", start, end)
+ if space > start + size // 2:
+ end = space
+ chunks.append(text[start:end].strip())
+ if end >= len(text):
+ break
+ start = max(end - overlap, start + 1)
+ if len(chunks) >= MAX_CHUNKS_PER_RESOURCE:
+ break
+ return [c for c in chunks if c]
+
+
+# ── hashed-TF embeddings ───────────────────────────────────────────────────
+
+def _tokens(text: str) -> list[str]:
+ return [t.lower() for t in _TOKEN_RE.findall(text or "") if t]
+
+
+def embed_text(text: str, dim: int = DIM) -> bytes:
+ """Deterministic L2-normalized hashed-TF vector, struct-packed float32."""
+ vec = [0.0] * dim
+ for tok in _tokens(text):
+ idx = int(hashlib.md5(tok.encode()).hexdigest(), 16) % dim
+ vec[idx] += 1.0
+ norm = math.sqrt(sum(v * v for v in vec))
+ if norm > 0:
+ vec = [v / norm for v in vec]
+ return struct.pack(f"<{dim}f", *vec)
+
+
+def embed_texts(texts: list[str], dim: int = DIM) -> list[bytes]:
+ """Batch entry point (pluggable: LLM /embeddings can replace hashing)."""
+ return [embed_text(t, dim) for t in texts]
+
+
+def cosine(a: bytes, b: bytes, dim: int = DIM) -> float:
+ """Cosine similarity of two packed normalized vectors (== dot product)."""
+ try:
+ va = struct.unpack(f"<{dim}f", a)
+ vb = struct.unpack(f"<{dim}f", b)
+ except struct.error:
+ return 0.0
+ return sum(x * y for x, y in zip(va, vb, strict=True))
+
+
+# ── indexing ───────────────────────────────────────────────────────────────
+
+def _resource_text(conn, resource_type: str, resource_id: int) -> str | None:
+ """Return indexable text, or None when the resource must not be indexed."""
+ if resource_type == "page":
+ row = conn.execute(
+ "SELECT title, content, content_format FROM pages "
+ "WHERE id=? AND (deleted_at IS NULL OR deleted_at='') "
+ "AND COALESCE(search_excluded, 0)=0",
+ (resource_id,),
+ ).fetchone()
+ if not row:
+ return None
+ body = extract_page_text(row["content"], row["content_format"])
+ return f"{row['title'] or ''}\n{body}".strip()
+ if resource_type == "collection":
+ row = conn.execute(
+ "SELECT name, description FROM collections WHERE id=?", (resource_id,)
+ ).fetchone()
+ if not row:
+ return None
+ return f"{row['name'] or ''}\n{row['description'] or ''}".strip()
+ return None
+
+
+def index_resource(resource_type: str, resource_id: int) -> int:
+ """(Re)index one resource. Returns the number of chunks stored."""
+ with get_conn() as conn:
+ text = _resource_text(conn, resource_type, resource_id)
+ conn.execute(
+ "DELETE FROM semantic_embeddings WHERE resource_type=? AND resource_id=?",
+ (resource_type, resource_id),
+ )
+ n = 0
+ if text:
+ for i, chunk in enumerate(chunk_text(text)[:MAX_CHUNKS_PER_RESOURCE]):
+ conn.execute(
+ """INSERT INTO semantic_embeddings
+ (resource_type, resource_id, chunk_id, chunk_text, embedding, model)
+ VALUES (?, ?, ?, ?, ?, ?)""",
+ (resource_type, resource_id, i, chunk, embed_text(chunk), MODEL),
+ )
+ n += 1
+ conn.execute(
+ """INSERT INTO semantic_index_state (resource_type, resource_id, indexed_at)
+ VALUES (?, ?, CURRENT_TIMESTAMP)
+ ON CONFLICT(resource_type, resource_id)
+ DO UPDATE SET indexed_at=CURRENT_TIMESTAMP""",
+ (resource_type, resource_id),
+ )
+ conn.commit()
+ return n
+
+
+def _stale_resources(conn, limit: int) -> list[tuple[str, int]]:
+ out: list[tuple[str, int]] = []
+ rows = conn.execute(
+ """SELECT p.id, p.updated_at FROM pages p
+ LEFT JOIN semantic_index_state s
+ ON s.resource_type='page' AND s.resource_id=p.id
+ WHERE (p.deleted_at IS NULL OR p.deleted_at='')
+ AND COALESCE(p.search_excluded, 0)=0
+ AND (s.indexed_at IS NULL OR p.updated_at > s.indexed_at)
+ ORDER BY p.updated_at DESC LIMIT ?""",
+ (limit,),
+ ).fetchall()
+ out += [("page", r["id"]) for r in rows]
+ if len(out) < limit:
+ rows = conn.execute(
+ """SELECT c.id, c.updated_at FROM collections c
+ LEFT JOIN semantic_index_state s
+ ON s.resource_type='collection' AND s.resource_id=c.id
+ WHERE s.indexed_at IS NULL OR c.updated_at > s.indexed_at
+ ORDER BY c.updated_at DESC LIMIT ?""",
+ (limit - len(out),),
+ ).fetchall()
+ out += [("collection", r["id"]) for r in rows]
+ return out
+
+
+def purge_orphans() -> int:
+ """Drop vectors for deleted/excluded resources. Returns rows removed."""
+ with get_conn() as conn:
+ cur = conn.execute(
+ """DELETE FROM semantic_embeddings
+ WHERE (resource_type='page' AND resource_id NOT IN (
+ SELECT id FROM pages WHERE (deleted_at IS NULL OR deleted_at='')
+ AND COALESCE(search_excluded, 0)=0))
+ OR (resource_type='collection' AND resource_id NOT IN (
+ SELECT id FROM collections))"""
+ )
+ conn.execute(
+ """DELETE FROM semantic_index_state
+ WHERE (resource_type='page' AND resource_id NOT IN (
+ SELECT id FROM pages WHERE (deleted_at IS NULL OR deleted_at='')
+ AND COALESCE(search_excluded, 0)=0))
+ OR (resource_type='collection' AND resource_id NOT IN (
+ SELECT id FROM collections))"""
+ )
+ conn.commit()
+ return cur.rowcount or 0
+
+
+def index_pending(limit: int = 50) -> dict:
+ """Index up to ``limit`` stale resources + purge orphans (scheduler job)."""
+ with get_conn() as conn:
+ stale = _stale_resources(conn, limit)
+ indexed = 0
+ for rtype, rid in stale:
+ try:
+ index_resource(rtype, rid)
+ indexed += 1
+ except Exception as exc: # never break the scheduler loop
+ logger.debug("semantic index failed for %s %s: %s", rtype, rid, exc)
+ purged = purge_orphans()
+ return {"checked": len(stale), "indexed": indexed, "purged": purged}
+
+
+async def semantic_index_scheduler(interval_seconds: int = 300) -> None:
+ """Background task: incremental indexing (wired in app lifespan)."""
+ while True:
+ try:
+ await asyncio.to_thread(index_pending)
+ except Exception as exc: # noqa: BLE001 — scheduler must survive
+ logger.debug("semantic index scheduler: %s", exc)
+ await asyncio.sleep(interval_seconds)
+
+
+# ── vector search ──────────────────────────────────────────────────────────
+
+def vector_search(query: str, *, limit: int = 20,
+ resource_types: tuple[str, ...] = ("page", "collection")) -> list[dict]:
+ """Brute-force cosine scan (fine at this scale). Returns ranked chunks."""
+ q = (query or "").strip()
+ if not q:
+ return []
+ qvec = embed_text(q)
+ with get_conn() as conn:
+ placeholders = ",".join("?" for _ in resource_types)
+ rows = conn.execute(
+ f"""SELECT resource_type, resource_id, chunk_id, chunk_text
+ FROM semantic_embeddings WHERE resource_type IN ({placeholders})""",
+ list(resource_types),
+ ).fetchall()
+ scored = []
+ for r in rows:
+ row = conn.execute(
+ "SELECT embedding FROM semantic_embeddings "
+ "WHERE resource_type=? AND resource_id=? AND chunk_id=?",
+ (r["resource_type"], r["resource_id"], r["chunk_id"]),
+ ).fetchone()
+ s = cosine(qvec, row["embedding"]) if row else 0.0
+ if s > 0:
+ scored.append({
+ "resource_type": r["resource_type"],
+ "resource_id": r["resource_id"],
+ "chunk_id": r["chunk_id"],
+ "chunk_text": r["chunk_text"],
+ "score": s,
+ })
+ scored.sort(key=lambda d: d["score"], reverse=True)
+ return scored[:limit]
+
+
+# ── hybrid (lexical + vector, RRF) + ACL ───────────────────────────────────
+
+def _rrf_fuse(ranked_lists: list[list[tuple[str, int]]], k: int = RRF_K) -> list[tuple[str, int, float]]:
+ scores: dict[tuple[str, int], float] = {}
+ for ranked in ranked_lists:
+ for rank, key in enumerate(ranked):
+ scores[key] = scores.get(key, 0.0) + 1.0 / (k + rank + 1)
+ fused = [(t, i, s) for (t, i), s in scores.items()]
+ fused.sort(key=lambda x: x[2], reverse=True)
+ return fused
+
+
+def hybrid_search(query: str, user: dict, *, limit: int = 20,
+ workspace_id: int | None = None,
+ resource_types: tuple[str, ...] = ("page", "collection")) -> tuple[list[dict], int]:
+ """Lexical + vector fusion, workspace-scoped, ACL-filtered.
+
+ Returns (results, total). Each result: {type, id, title, excerpt, url, score}.
+ """
+ from app.services.permission_manager import PermissionManager
+
+ q = (query or "").strip()
+ if not q:
+ return [], 0
+ limit = max(1, min(int(limit or 20), 100))
+ user_id = user.get("id")
+ pm = PermissionManager(user_id, bool(user.get("is_admin")))
+
+ # 1) lexical candidates (already workspace-membership scoped)
+ from app.services import search as search_service
+ lex = search_service.search(q, user_id, limit * 3)
+ lex_ranked: list[tuple[str, int]] = []
+ lex_by_key: dict[tuple[str, int], dict] = {}
+ for item in (lex.get("pages") or []) + (lex.get("collections") or []):
+ key = (item["type"], item["id"])
+ if key not in lex_by_key:
+ lex_by_key[key] = item
+ lex_ranked.append(key)
+
+ # 2) vector candidates
+ vec = vector_search(q, limit=limit * 3, resource_types=resource_types)
+ vec_ranked = [(d["resource_type"], d["resource_id"]) for d in vec]
+
+ # 3) fuse
+ fused = _rrf_fuse([lex_ranked, vec_ranked])
+
+ # 4) ACL + workspace filter, enrich
+ results: list[dict] = []
+ with get_conn() as conn:
+ for rtype, rid, score in fused:
+ if rtype == "page":
+ if not pm.can_view_page(rid):
+ continue
+ row = conn.execute(
+ "SELECT id, title, content, content_format, workspace_id, "
+ "COALESCE(search_excluded, 0) AS excluded "
+ "FROM pages WHERE id=?", (rid,)).fetchone()
+ if not row or row["excluded"]:
+ continue
+ if workspace_id and row["workspace_id"] != workspace_id:
+ continue
+ excerpt = (lex_by_key.get((rtype, rid), {}).get("excerpt")
+ or extract_page_text(row["content"], row["content_format"])[:160])
+ results.append({"type": "page", "id": rid,
+ "title": (row["title"] or "Untitled"),
+ "excerpt": excerpt, "url": f"/pages/{rid}",
+ "score": round(score, 5)})
+ else:
+ if not pm.can_view_collection(rid):
+ continue
+ row = conn.execute(
+ "SELECT id, name, description, workspace_id FROM collections WHERE id=?",
+ (rid,)).fetchone()
+ if not row:
+ continue
+ if workspace_id and row["workspace_id"] != workspace_id:
+ continue
+ excerpt = (lex_by_key.get((rtype, rid), {}).get("subtitle")
+ or (row["description"] or "")[:160])
+ results.append({"type": "collection", "id": rid,
+ "title": (row["name"] or "Untitled"),
+ "excerpt": excerpt, "url": f"/db/{rid}",
+ "score": round(score, 5)})
+ if len(results) >= limit:
+ break
+ return results, len(results)
+
+
+# ── Ask AI ─────────────────────────────────────────────────────────────────
+
+def _check_ask_rate(user_id: int) -> None:
+ from fastapi import HTTPException
+ now = time.time()
+ start, count = _ask_rate.get(user_id, (now, 0))
+ if now - start > _ASK_RATE_WINDOW:
+ _ask_rate[user_id] = (now, 1)
+ return
+ if count >= _ASK_RATE_MAX:
+ raise HTTPException(429, "Too many questions. Slow down.")
+ _ask_rate[user_id] = (start, count + 1)
+
+
+def _offline_answer(question: str, chunks: list[dict]) -> str:
+ """Extractive fallback: top sentences sharing query terms + citations."""
+ qterms = {t.lower() for t in _tokens(question)}
+ picked: list[str] = []
+ for ch in chunks[:8]:
+ for sent in re.split(r"(?<=[.!?])\s+", ch["chunk_text"] or ""):
+ words = {t.lower() for t in _tokens(sent)}
+ if qterms & words and len(sent.strip()) > 20:
+ picked.append((sent.strip(), ch))
+ if len(picked) >= 4:
+ break
+ if len(picked) >= 4:
+ break
+ if not picked:
+ # No lexical overlap: still cite the top vector matches.
+ lines = []
+ for ch in chunks[:3]:
+ snippet = (ch["chunk_text"] or "")[:200].replace("\n", " ")
+ lines.append(f"- {snippet} [[fdpage:{ch['resource_id']}]]"
+ if ch["resource_type"] == "page" else f"- {snippet}")
+ return ("Je n'ai pas trouvé de passage répondant directement, "
+ "mais voici les passages les plus proches :\n" + "\n".join(lines))
+ lines = []
+ for sent, ch in picked:
+ if ch["resource_type"] == "page":
+ lines.append(f"- {sent} [[fdpage:{ch['resource_id']}]]")
+ else:
+ lines.append(f"- {sent}")
+ return "Voici ce que j'ai trouvé dans votre workspace :\n" + "\n".join(lines)
+
+
+def _resolve_citations(conn, chunks: list[dict]) -> list[dict]:
+ seen: list[dict] = []
+ done: set[tuple[str, int]] = set()
+ for ch in chunks:
+ key = (ch["resource_type"], ch["resource_id"])
+ if key in done:
+ continue
+ done.add(key)
+ if ch["resource_type"] == "page":
+ row = conn.execute("SELECT title FROM pages WHERE id=?", (ch["resource_id"],)).fetchone()
+ seen.append({"type": "page", "id": ch["resource_id"],
+ "title": (row["title"] if row else "Deleted page") or "Untitled"})
+ else:
+ row = conn.execute("SELECT name FROM collections WHERE id=?",
+ (ch["resource_id"],)).fetchone()
+ seen.append({"type": "collection", "id": ch["resource_id"],
+ "title": (row["name"] if row else "Deleted") or "Untitled"})
+ return seen
+
+
+async def ask(question: str, user: dict, workspace_id: int | None = None) -> dict:
+ """RAG answer over the user's authorized chunks (LLM or offline fallback)."""
+ from app.services.permission_manager import PermissionManager
+
+ q = (question or "").strip()
+ if not q:
+ from fastapi import HTTPException
+ raise HTTPException(400, "question is required")
+ _check_ask_rate(user.get("id") or 0)
+ cache_key = (hashlib.sha256(q.encode()).hexdigest(), workspace_id, user.get("id"))
+ now = time.time()
+ hit = _ask_cache.get(cache_key)
+ if hit and hit[0] > now:
+ out = dict(hit[1])
+ out["cached"] = True
+ return out
+
+ pm = PermissionManager(user.get("id"), bool(user.get("is_admin")))
+ vec = vector_search(q, limit=24)
+ allowed = []
+ with get_conn() as conn:
+ for ch in vec:
+ if ch["resource_type"] == "page":
+ row = conn.execute(
+ "SELECT workspace_id, COALESCE(search_excluded, 0) AS excluded "
+ "FROM pages WHERE id=?", (ch["resource_id"],)).fetchone()
+ if not row or row["excluded"] or not pm.can_view_page(ch["resource_id"]):
+ continue
+ if workspace_id and row["workspace_id"] != workspace_id:
+ continue
+ else:
+ row = conn.execute(
+ "SELECT workspace_id FROM collections WHERE id=?",
+ (ch["resource_id"],)).fetchone()
+ if not row or not pm.can_view_collection(ch["resource_id"]):
+ continue
+ if workspace_id and row["workspace_id"] != workspace_id:
+ continue
+ allowed.append(ch)
+ if len(allowed) >= 8:
+ break
+
+ answer = ""
+ offline = True
+ if allowed:
+ try:
+ from app.services.llm_client import LLMClient
+ llm = LLMClient()
+ if await llm.is_available():
+ ctx = "\n\n".join(
+ f"[doc {i+1} page_id={c['resource_id']}]\n{c['chunk_text'][:1500]}"
+ for i, c in enumerate(allowed)
+ )
+ resp = await llm.complete([
+ {"role": "system",
+ "content": "Réponds en français en citant les sources avec "
+ "[[fdpage:ID]] (ID = page_id indiqué). Concis."},
+ {"role": "user", "content": f"Question : {q}\n\nContexte :\n{ctx}"},
+ ])
+ answer = (resp.text or "").strip()
+ offline = False
+ except Exception as exc: # noqa: BLE001 — fall back to extractive
+ logger.debug("ask LLM failed, offline fallback: %s", exc)
+ if not answer:
+ answer = ("Aucun contenu accessible ne correspond à votre question."
+ if not allowed else _offline_answer(q, allowed))
+
+ with get_conn() as conn:
+ citations = _resolve_citations(conn, allowed[:8])
+ out = {"answer_markdown": answer, "citations": citations,
+ "offline": offline, "cached": False}
+ _ask_cache[cache_key] = (now + _ASK_CACHE_TTL, out)
+ return out
+
+
+def reset_state() -> None:
+ """Test helper: clear ask cache + rate limiter."""
+ _ask_cache.clear()
+ _ask_rate.clear()
diff --git a/app/services/two_factor.py b/app/services/two_factor.py
new file mode 100644
index 0000000..b8fcb6f
--- /dev/null
+++ b/app/services/two_factor.py
@@ -0,0 +1,136 @@
+"""FlowDeck — TOTP 2FA + backup codes (v7.2.0).
+
+Secrets are Fernet-encrypted at rest (same construction as SSO secrets).
+Login flow: ``POST /auth/local-login`` returns ``2fa_required`` + a short-lived
+signed ``pending`` token; ``POST /auth/local-verify`` exchanges it for a
+session. See ``docs/V72_Enterprise_SCIM_2FA.md``.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import secrets
+
+from app.db import get_conn
+
+BACKUP_CODE_COUNT = 10
+
+
+def _fernet():
+ import base64
+
+ from cryptography.fernet import Fernet
+
+ from app.config import settings
+ key = hashlib.sha256((settings.app_secret_key or "flowdeck").encode()).digest()
+ return Fernet(base64.urlsafe_b64encode(key))
+
+
+def is_enabled(user_id: int) -> bool:
+ with get_conn() as conn:
+ row = conn.execute("SELECT totp_secret_enc FROM users WHERE id=?",
+ (user_id,)).fetchone()
+ if not row or not row["totp_secret_enc"]:
+ return False
+ try:
+ return bool(_fernet().decrypt(row["totp_secret_enc"].encode()).decode())
+ except Exception: # noqa: BLE001
+ return False
+
+
+def setup_secret(user_id: int) -> dict:
+ """Create a new TOTP secret (not yet active until verified)."""
+ import pyotp
+ secret = pyotp.random_base32()
+ with get_conn() as conn:
+ row = conn.execute("SELECT login, email FROM users WHERE id=?", (user_id,)).fetchone()
+ label = (row["email"] or row["login"]) if row else f"user{user_id}"
+ uri = pyotp.totp.TOTP(secret).provisioning_uri(name=label, issuer_name="FlowDeck")
+ return {"secret": secret, "otpauth_url": uri}
+
+
+def activate_secret(user_id: int, secret: str, code: str) -> list[str]:
+ """Verify ``code`` against ``secret``; on success store + return backup codes."""
+ import pyotp
+ if not pyotp.TOTP(secret).verify(code, valid_window=1):
+ raise ValueError("invalid code")
+ codes = [secrets.token_hex(4) for _ in range(BACKUP_CODE_COUNT)]
+ hashes = [hashlib.sha256(c.encode()).hexdigest() for c in codes]
+ with get_conn() as conn:
+ conn.execute("UPDATE users SET totp_secret_enc=?, totp_backup_hashes=? WHERE id=?",
+ (_fernet().encrypt(secret.encode()).decode(),
+ json.dumps(hashes), user_id))
+ conn.commit()
+ return codes
+
+
+def verify_code(user_id: int, code: str) -> bool:
+ """Check a TOTP code or consume a backup code."""
+ code = (code or "").strip().replace(" ", "")
+ if not code:
+ return False
+ with get_conn() as conn:
+ row = conn.execute("SELECT totp_secret_enc, totp_backup_hashes FROM users WHERE id=?",
+ (user_id,)).fetchone()
+ if not row or not row["totp_secret_enc"]:
+ return False
+ try:
+ secret = _fernet().decrypt(row["totp_secret_enc"].encode()).decode()
+ except Exception: # noqa: BLE001
+ return False
+ import pyotp
+ if secret and pyotp.TOTP(secret).verify(code, valid_window=1):
+ return True
+ # backup codes (single use)
+ try:
+ hashes = json.loads(row["totp_backup_hashes"] or "[]")
+ except (TypeError, json.JSONDecodeError):
+ hashes = []
+ digest = hashlib.sha256(code.encode()).hexdigest()
+ if digest in hashes:
+ hashes.remove(digest)
+ with get_conn() as conn:
+ conn.execute("UPDATE users SET totp_backup_hashes=? WHERE id=?",
+ (json.dumps(hashes), user_id))
+ conn.commit()
+ return True
+ return False
+
+
+def disable(user_id: int) -> None:
+ with get_conn() as conn:
+ conn.execute("UPDATE users SET totp_secret_enc='', totp_backup_hashes='[]'"
+ " WHERE id=?", (user_id,))
+ conn.commit()
+
+
+def remaining_backup_codes(user_id: int) -> int:
+ with get_conn() as conn:
+ row = conn.execute("SELECT totp_backup_hashes FROM users WHERE id=?",
+ (user_id,)).fetchone()
+ try:
+ return len(json.loads(row["totp_backup_hashes"] or "[]")) if row else 0
+ except (TypeError, json.JSONDecodeError):
+ return 0
+
+
+# ── pending 2FA challenge (signed, 5 min) ──────────────────────────────────
+
+def mint_pending(user_id: int) -> str:
+ from itsdangerous import URLSafeTimedSerializer
+
+ from app.config import settings
+ ser = URLSafeTimedSerializer(settings.app_secret_key, salt="totp-pending")
+ return ser.dumps({"user_id": user_id})
+
+
+def redeem_pending(token: str, max_age: int = 300) -> int | None:
+ from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer
+
+ from app.config import settings
+ ser = URLSafeTimedSerializer(settings.app_secret_key, salt="totp-pending")
+ try:
+ payload = ser.loads(token, max_age=max_age)
+ return int(payload.get("user_id", 0)) or None
+ except (BadSignature, SignatureExpired, ValueError):
+ return None
diff --git a/app/services/wiki.py b/app/services/wiki.py
new file mode 100644
index 0000000..efdd3e3
--- /dev/null
+++ b/app/services/wiki.py
@@ -0,0 +1,415 @@
+"""FlowDeck — teamspaces, verified pages, collab polish (v7.3.0).
+
+Design : ``docs/V73_Wiki_Teamspaces_Polish.md``.
+
+- ``teamspaces`` namespace pages + databases ; ``private=1`` → invisible aux
+ non-membres (404, comme une collection restricted).
+- ``page_verifications`` : badge ✅ avec expiration (90 j par défaut).
+- ``page_follows`` → notif ``page.updated`` ; ``comment_reactions`` ;
+ ``guest_shares`` (``/g/``) ; ``page_views`` (compteurs journaliers).
+"""
+from __future__ import annotations
+
+import datetime
+import secrets
+
+from app.db import get_conn
+
+VERIFICATION_DAYS_DEFAULT = 90
+# Roles, strongest first. Mirrors collection roles.
+TEAMSPACE_ROLES = ("owner", "editor", "commenter", "viewer")
+_ROLE_RANK = {r: i for i, r in enumerate(reversed(TEAMSPACE_ROLES))}
+
+
+def _utcnow() -> datetime.datetime:
+ return datetime.datetime.now(datetime.UTC)
+
+
+def _iso(dt: datetime.datetime) -> str:
+ return dt.replace(microsecond=0).isoformat()
+
+
+# ── teamspaces ─────────────────────────────────────────────────────────────
+
+def get_teamspace_role(user_id: int | None, teamspace_id: int) -> str | None:
+ """Explicit role, else workspace role, else ``None`` when unreachable."""
+ if not user_id:
+ return None
+ with get_conn() as conn:
+ ts = conn.execute("SELECT workspace_id, private FROM teamspaces WHERE id=?",
+ (teamspace_id,)).fetchone()
+ if not ts:
+ return None
+ row = conn.execute("SELECT role FROM teamspace_members WHERE teamspace_id=? AND user_id=?",
+ (teamspace_id, user_id)).fetchone()
+ if row:
+ return row["role"]
+ admin = conn.execute("SELECT is_admin FROM users WHERE id=?", (user_id,)).fetchone()
+ if admin and admin["is_admin"]:
+ return "owner"
+ owner = conn.execute("SELECT owner_id FROM workspaces WHERE id=?",
+ (ts["workspace_id"],)).fetchone()
+ if owner and owner["owner_id"] == user_id:
+ return "owner"
+ if not ts["private"]:
+ # a public teamspace is still workspace-scoped: no workspace
+ # membership means no access (otherwise any logged-in account on
+ # the instance could read every public teamspace).
+ member = conn.execute("SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=?",
+ (ts["workspace_id"], user_id)).fetchone()
+ return member["role"] if member else None
+ return None
+
+
+def can_read_teamspace(user_id: int | None, teamspace_id: int) -> bool:
+ return get_teamspace_role(user_id, teamspace_id) is not None
+
+
+def can_write_teamspace(user_id: int | None, teamspace_id: int) -> bool:
+ role = get_teamspace_role(user_id, teamspace_id)
+ return role in ("owner", "editor")
+
+
+def list_teamspaces(user_id: int, workspace_id: int | None = None) -> list[dict]:
+ """Teamspaces the user can see (private ones filtered out).
+
+ ``workspace_id=None`` lists across every workspace (global sidebar).
+ """
+ with get_conn() as conn:
+ if workspace_id is not None:
+ ws_member = conn.execute("SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=?",
+ (workspace_id, user_id)).fetchone()
+ if not ws_member:
+ owner = conn.execute("SELECT owner_id FROM workspaces WHERE id=?",
+ (workspace_id,)).fetchone()
+ admin = conn.execute("SELECT is_admin FROM users WHERE id=?", (user_id,)).fetchone()
+ if not owner or owner["owner_id"] != user_id:
+ if not (admin and admin["is_admin"]):
+ return []
+ rows = conn.execute("SELECT * FROM teamspaces WHERE workspace_id=? ORDER BY name",
+ (workspace_id,)).fetchall()
+ else:
+ rows = conn.execute("SELECT * FROM teamspaces ORDER BY workspace_id, name").fetchall()
+ out = []
+ for r in rows:
+ role = get_teamspace_role(user_id, r["id"])
+ if role is None:
+ continue
+ counts = conn.execute(
+ """SELECT (SELECT COUNT(*) FROM pages WHERE teamspace_id=?)
+ + (SELECT COUNT(*) FROM collections WHERE teamspace_id=?) AS n""",
+ (r["id"], r["id"])).fetchone()
+ item = dict(r)
+ item["role"] = role
+ item["workspace_name"] = _workspace_name(conn, r["workspace_id"])
+ item["item_count"] = counts["n"]
+ out.append(item)
+ return out
+
+
+def _workspace_name(conn, workspace_id: int) -> str:
+ row = conn.execute("SELECT name FROM workspaces WHERE id=?", (workspace_id,)).fetchone()
+ return row["name"] if row else ""
+
+
+def teamspace_pages(teamspace_id: int) -> list[dict]:
+ """Non-deleted pages belonging to a teamspace (title/URL order)."""
+ with get_conn() as conn:
+ return [dict(r) for r in conn.execute(
+ """SELECT id, title, workspace_id, teamspace_id
+ FROM pages WHERE teamspace_id=? AND deleted_at IS NULL
+ ORDER BY title""", (teamspace_id,)).fetchall()]
+
+
+def teamspace_collections(teamspace_id: int) -> list[dict]:
+ """Databases belonging to a teamspace."""
+ with get_conn() as conn:
+ return [dict(r) for r in conn.execute(
+ """SELECT id, name, icon FROM collections
+ WHERE teamspace_id=? ORDER BY name""", (teamspace_id,)).fetchall()]
+
+
+def create_teamspace(workspace_id: int, name: str, user_id: int,
+ description: str = "", private: bool = False) -> int:
+ with get_conn() as conn:
+ try:
+ cur = conn.execute(
+ "INSERT INTO teamspaces (workspace_id, name, description, private, created_by)"
+ " VALUES (?,?,?,?,?)",
+ (workspace_id, name.strip(), description[:400], 1 if private else 0, user_id))
+ except Exception as exc: # UNIQUE(workspace_id, name)
+ if "UNIQUE" in str(exc):
+ raise ValueError("A teamspace with this name already exists") from None
+ raise
+ conn.commit()
+ # the creator is owner
+ conn.execute("INSERT INTO teamspace_members (teamspace_id, user_id, role)"
+ " VALUES (?,?,'owner')", (cur.lastrowid, user_id))
+ conn.commit()
+ return cur.lastrowid
+
+
+def teamspace_member_ids(teamspace_id: int) -> list[int]:
+ with get_conn() as conn:
+ return [r["user_id"] for r in conn.execute(
+ "SELECT user_id FROM teamspace_members WHERE teamspace_id=?",
+ (teamspace_id,)).fetchall()]
+
+
+# ── verified pages ─────────────────────────────────────────────────────────
+
+def is_expired(row) -> bool:
+ if not row or not row["expires_at"]:
+ return False
+ try:
+ return _utcnow() > datetime.datetime.fromisoformat(row["expires_at"])
+ except ValueError:
+ return False
+
+
+def verify_page(page_id: int, user_id: int, days: int = VERIFICATION_DAYS_DEFAULT,
+ note: str = "") -> dict:
+ days = max(1, min(int(days or VERIFICATION_DAYS_DEFAULT), 365))
+ expires = _iso(_utcnow() + datetime.timedelta(days=days))
+ with get_conn() as conn:
+ conn.execute(
+ """INSERT INTO page_verifications (page_id, verified_by, note, expires_at)
+ VALUES (?,?,?,?)
+ ON CONFLICT(page_id) DO UPDATE SET
+ verified_by=excluded.verified_by, note=excluded.note,
+ verified_at=CURRENT_TIMESTAMP, expires_at=excluded.expires_at""",
+ (page_id, user_id, note[:400], expires))
+ conn.commit()
+ return verification(page_id)
+
+
+def verification(page_id: int) -> dict | None:
+ with get_conn() as conn:
+ row = conn.execute(
+ """SELECT v.*, u.full_name, u.login FROM page_verifications v
+ LEFT JOIN users u ON u.id = v.verified_by WHERE v.page_id=?""",
+ (page_id,)).fetchone()
+ if not row:
+ return None
+ d = dict(row)
+ d["expired"] = is_expired(row)
+ d["active"] = not d["expired"]
+ return d
+
+
+def unverify_page(page_id: int) -> bool:
+ with get_conn() as conn:
+ cur = conn.execute("DELETE FROM page_verifications WHERE page_id=?", (page_id,))
+ conn.commit()
+ return bool(cur.rowcount)
+
+
+def expiring_verifications(days: int = 7) -> list[dict]:
+ """Verifications expiring within ``days`` (drives the owner notification).
+
+ ``pages`` has no owner column in FlowDeck, so the reminder targets the user
+ who performed the verification.
+ """
+ horizon = _iso(_utcnow() + datetime.timedelta(days=days))
+ with get_conn() as conn:
+ rows = conn.execute(
+ """SELECT v.*, p.title, v.verified_by AS owner_id FROM page_verifications v
+ JOIN pages p ON p.id = v.page_id
+ WHERE v.expires_at IS NOT NULL AND v.expires_at <= ?""",
+ (horizon,)).fetchall()
+ return [dict(r) for r in rows]
+
+
+# ── follows ────────────────────────────────────────────────────────────────
+
+def is_following(page_id: int, user_id: int) -> bool:
+ with get_conn() as conn:
+ return bool(conn.execute("SELECT 1 FROM page_follows WHERE page_id=? AND user_id=?",
+ (page_id, user_id)).fetchone())
+
+
+def toggle_follow(page_id: int, user_id: int) -> bool:
+ """Returns the new state (True = now following)."""
+ with get_conn() as conn:
+ if conn.execute("SELECT 1 FROM page_follows WHERE page_id=? AND user_id=?",
+ (page_id, user_id)).fetchone():
+ conn.execute("DELETE FROM page_follows WHERE page_id=? AND user_id=?",
+ (page_id, user_id))
+ conn.commit()
+ return False
+ conn.execute("INSERT OR IGNORE INTO page_follows (page_id, user_id) VALUES (?,?)",
+ (page_id, user_id))
+ conn.commit()
+ return True
+
+
+def ensure_follow(page_id: int, user_id: int, conn=None) -> bool:
+ """Follow unless already following. Returns True when newly followed."""
+ if not user_id:
+ return False
+
+ def _run(c):
+ if c.execute("SELECT 1 FROM page_follows WHERE page_id=? AND user_id=?",
+ (page_id, user_id)).fetchone():
+ return False
+ c.execute("INSERT OR IGNORE INTO page_follows (page_id, user_id) VALUES (?,?)",
+ (page_id, user_id))
+ return True
+
+ if conn is not None:
+ added = _run(conn)
+ conn.commit()
+ return added
+ with get_conn() as _c:
+ added = _run(_c)
+ _c.commit()
+ return added
+
+
+def followers(page_id: int) -> list[int]:
+ with get_conn() as conn:
+ return [r["user_id"] for r in conn.execute(
+ "SELECT user_id FROM page_follows WHERE page_id=?", (page_id,)).fetchall()]
+
+
+# Rate-limit window for ``page.updated`` notifications: the block editor
+# autosaves every ~1.5 s; without it followers would be spammed per keystroke.
+_UPDATE_NOTIF_WINDOW_MIN = 10
+
+
+def notify_followers_of_page_update(page_id: int, actor_id: int | None,
+ title: str = "") -> int:
+ """Create a ``page.updated`` notification for every follower (except the
+ actor), throttled to one per :data:`_UPDATE_NOTIF_WINDOW_MIN` minutes.
+ Returns the number of notifications created."""
+ if not page_id:
+ return 0
+ from app.services.notifications import create_notification
+ with get_conn() as conn:
+ followed = conn.execute("SELECT user_id FROM page_follows WHERE page_id=?", (page_id,)).fetchall()
+ if not followed:
+ return 0
+ page = conn.execute("SELECT title FROM pages WHERE id=?", (page_id,)).fetchone()
+ title = (title or (page["title"] if page else "") or "Untitled")
+ created = 0
+ for f in followed:
+ uid = f["user_id"]
+ if uid == actor_id:
+ continue
+ recent = conn.execute(
+ """SELECT 1 FROM notifications
+ WHERE user_id=? AND resource_type='page' AND resource_id=?
+ AND ntype='page.updated'
+ AND created_at >= datetime('now', ?)""",
+ (uid, page_id, f"-{_UPDATE_NOTIF_WINDOW_MIN} minutes")).fetchone()
+ if recent:
+ continue
+ create_notification(
+ uid, actor_id, "page.updated",
+ title=f'"{title}" was updated',
+ message=f'Page "{title}" has been modified',
+ resource_type="page", resource_id=page_id,
+ url=f"/pages/{page_id}", conn=conn, commit=False,
+ )
+ created += 1
+ conn.commit()
+ return created
+
+
+# ── comment reactions ──────────────────────────────────────────────────────
+
+def toggle_reaction(comment_id: int, user_id: int, emoji: str) -> dict:
+ """Add or remove ``emoji``; returns the aggregated counts for the comment."""
+ emoji = (emoji or "").strip()[:16]
+ with get_conn() as conn:
+ if not conn.execute("SELECT 1 FROM comments WHERE id=?", (comment_id,)).fetchone():
+ raise LookupError("Comment not found")
+ existing = conn.execute(
+ "SELECT id FROM comment_reactions WHERE comment_id=? AND user_id=? AND emoji=?",
+ (comment_id, user_id, emoji)).fetchone()
+ if existing:
+ conn.execute("DELETE FROM comment_reactions WHERE id=?", (existing["id"],))
+ conn.commit()
+ else:
+ conn.execute("INSERT INTO comment_reactions (comment_id, user_id, emoji)"
+ " VALUES (?,?,?)", (comment_id, user_id, emoji))
+ conn.commit()
+ return reactions(comment_id)
+
+
+def reactions(comment_id: int) -> dict[str, dict]:
+ with get_conn() as conn:
+ rows = conn.execute(
+ """SELECT emoji, COUNT(*) AS n,
+ GROUP_CONCAT(user_id) AS users
+ FROM comment_reactions WHERE comment_id=? GROUP BY emoji ORDER BY emoji""",
+ (comment_id,)).fetchall()
+ return {r["emoji"]: {"count": r["n"],
+ "users": [int(u) for u in (r["users"] or "").split(",") if u]}
+ for r in rows}
+
+
+# ── guest shares ───────────────────────────────────────────────────────────
+
+def create_guest_share(page_id: int, email: str, role: str, created_by: int,
+ days: int | None = 30) -> dict:
+ if role not in ("viewer", "commenter"):
+ raise ValueError("role must be viewer or commenter")
+ token = f"g_{secrets.token_urlsafe(24)}"
+ expires = _iso(_utcnow() + datetime.timedelta(days=days)) if days else None
+ with get_conn() as conn:
+ cur = conn.execute(
+ """INSERT INTO guest_shares (page_id, email, token, role, created_by, expires_at)
+ VALUES (?,?,?,?,?,?)""",
+ (page_id, email[:200], token, role, created_by, expires))
+ conn.commit()
+ return dict(conn.execute("SELECT * FROM guest_shares WHERE id=?",
+ (cur.lastrowid,)).fetchone())
+
+
+def resolve_guest_share(token: str) -> dict | None:
+ """Active share for ``token``, or ``None`` (unknown / revoked / expired)."""
+ with get_conn() as conn:
+ row = conn.execute("SELECT * FROM guest_shares WHERE token=?", (token,)).fetchone()
+ if not row or row["revoked"]:
+ return None
+ if row["expires_at"]:
+ try:
+ if _utcnow() > datetime.datetime.fromisoformat(row["expires_at"]):
+ return None
+ except ValueError:
+ pass
+ return dict(row)
+
+
+# ── page views ─────────────────────────────────────────────────────────────
+
+def record_view(page_id: int, day: str | None = None) -> int:
+ day = day or _utcnow().strftime("%Y-%m-%d")
+ with get_conn() as conn:
+ conn.execute(
+ """INSERT INTO page_views (page_id, day, views) VALUES (?,?,1)
+ ON CONFLICT(page_id, day) DO UPDATE SET views = views + 1""",
+ (page_id, day))
+ conn.commit()
+ row = conn.execute("SELECT views FROM page_views WHERE page_id=? AND day=?",
+ (page_id, day)).fetchone()
+ return row["views"]
+
+
+def view_stats(page_id: int, days: int = 30) -> dict:
+ days = max(1, min(int(days or 30), 365))
+ since = (_utcnow() - datetime.timedelta(days=days - 1)).strftime("%Y-%m-%d")
+ with get_conn() as conn:
+ rows = conn.execute(
+ "SELECT day, views FROM page_views WHERE page_id=? AND day>=? ORDER BY day",
+ (page_id, since)).fetchall()
+ series = {r["day"]: r["views"] for r in rows}
+ # fill the gap so charts have no holes
+ out, cursor = [], _utcnow() - datetime.timedelta(days=days - 1)
+ for _ in range(days):
+ key = cursor.strftime("%Y-%m-%d")
+ out.append({"day": key, "views": series.get(key, 0)})
+ cursor += datetime.timedelta(days=1)
+ return {"page_id": page_id, "days": days, "total": sum(series.values()),
+ "series": out}
diff --git a/app/services/wiki_blocks.py b/app/services/wiki_blocks.py
new file mode 100644
index 0000000..a3ec6e5
--- /dev/null
+++ b/app/services/wiki_blocks.py
@@ -0,0 +1,196 @@
+"""FlowDeck — v7.3.0 blocks: mermaid, equation_inline, progress.
+
+Server-side rendering so HTML/PDF export embeds real content (the editor
+already has Prism + KaTeX client-side). Design §2 of
+``docs/V73_Wiki_Teamspaces_Polish.md``.
+
+- ``mermaid`` : SVG via ``mmdc`` when installed, else a ``
`` fallback
+ that mermaid.js can still render in the browser.
+- ``equation_inline`` : KaTeX delimiters wrapped in a span (client auto-render);
+ falls back to readable plaintext.
+- ``progress`` : ``{"rollup_ref": {collection_id, property_id}}`` → percent
+ bar computed with :class:`RollupEngine`.
+"""
+from __future__ import annotations
+
+import html
+import re
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+
+# ── mermaid ────────────────────────────────────────────────────────────────
+
+def mmdc_available() -> bool:
+ return shutil.which("mmdc") is not None
+
+
+def mermaid_to_svg(source: str, timeout: int = 20) -> str | None:
+ """Render mermaid source to an inline SVG, or ``None`` if unavailable.
+
+ Uses the local mermaid-cli (``mmdc``) when present; never raises — callers
+ degrade to the code fallback.
+ """
+ source = (source or "").strip()
+ if not source or not mmdc_available():
+ return None
+ with tempfile.TemporaryDirectory(prefix="fd_mermaid_") as tmp:
+ src = Path(tmp) / "diagram.mmd"
+ out = Path(tmp) / "diagram.svg"
+ src.write_text(source, encoding="utf-8")
+ try:
+ subprocess.run(
+ ["mmdc", "-i", str(src), "-o", str(out), "-b", "transparent"],
+ capture_output=True, timeout=timeout, check=True)
+ except (subprocess.SubprocessError, OSError):
+ return None
+ if not out.exists():
+ return None
+ svg = out.read_text(encoding="utf-8", errors="replace")
+ # strip the XML prolog / doctype so the SVG can be inlined
+ svg = re.sub(r"<\?xml.*?\?>", "", svg, flags=re.S).strip()
+ return svg or None
+
+
+def render_mermaid(source: str) -> str:
+ """Inline SVG when possible, else a mermaid-renderable code block."""
+ svg = mermaid_to_svg(source)
+ if svg:
+ return f'{svg}'
+ return (f'
'
+ f'{html.escape(source or "")}
')
+
+
+# ── equations ──────────────────────────────────────────────────────────────
+
+# Only characters that are meaningful inside a math expression are kept. Note
+# that `<`, `>` and `\` are NOT allowed: they would let the source close the
+# KaTeX delimiter early or inject markup into the exported HTML.
+_EQUATION_ALLOWED = re.compile(r"[^0-9A-Za-z\s+\-*/^_=!(){}\[\].|,';:]")
+
+
+def sanitize_equation(source: str) -> str:
+ return _EQUATION_ALLOWED.sub("", source or "").strip()
+
+
+def render_equation(source: str) -> str:
+ expr = sanitize_equation(source)
+ if not expr:
+ return ""
+ return (f''
+ f'\\({html.escape(expr)}\\)')
+
+
+# ── progress ───────────────────────────────────────────────────────────────
+
+def _rows_values(collection_id: int) -> list[dict]:
+ import json as _json
+
+ from app.db import get_conn
+ with get_conn() as conn:
+ rows = conn.execute(
+ "SELECT property_values_json FROM collection_pages WHERE collection_id=?",
+ (collection_id,)).fetchall()
+ out = []
+ for r in rows:
+ try:
+ out.append(_json.loads(r["property_values_json"] or "{}"))
+ except ValueError:
+ out.append({})
+ return out
+
+
+_DONE_TRUE = (True, 1, "true", "done", "yes", "checked", "✓", "complete", "completed")
+
+
+def _is_done(value) -> bool:
+ if isinstance(value, bool):
+ return value
+ if isinstance(value, (int, float)):
+ return bool(value)
+ if isinstance(value, str):
+ return value.strip().lower() in _DONE_TRUE
+ return False
+
+
+def progress_value(rollup_ref: dict | None) -> dict:
+ """Resolve ``{collection_id, done_property_id | property_id, func}`` to a
+ 0..100 percent.
+
+ Aggregates directly over ``collection_pages.property_values_json`` (the
+ relation-based :class:`RollupEngine` needs a relation chain + a source
+ page, which a page-embedded progress bar does not have).
+ """
+ from app.db import get_conn
+ if not rollup_ref or not rollup_ref.get("collection_id"):
+ return {"percent": None, "done": None, "total": None}
+ cid = rollup_ref["collection_id"]
+ values = _rows_values(cid)
+ total = len(values)
+
+ done_pid = rollup_ref.get("done_property_id")
+ if done_pid:
+ key = str(done_pid)
+ done = sum(1 for vals in values if _is_done(vals.get(key)))
+ if total:
+ return {"percent": round(done * 100 / total, 1), "done": done, "total": total}
+ return {"percent": 0.0, "done": 0, "total": 0}
+
+ pid = rollup_ref.get("property_id")
+ if not pid or not total:
+ return {"percent": None, "done": None, "total": total}
+ key = str(pid)
+ with get_conn() as conn:
+ prop = conn.execute("SELECT prop_type FROM collection_properties WHERE id=? AND"
+ " collection_id=?", (pid, cid)).fetchone()
+ if not prop:
+ return {"percent": None, "done": None, "total": total}
+ func = (rollup_ref.get("func") or "count").lower()
+ if prop["prop_type"] in ("checkbox", "status", "select"):
+ done = sum(1 for vals in values if _is_done(vals.get(key)))
+ elif prop["prop_type"] in ("number", "formula", "rollup"):
+ nums = []
+ for vals in values:
+ v = vals.get(key)
+ try:
+ nums.append(float(v))
+ except (TypeError, ValueError):
+ continue
+ if not nums:
+ return {"percent": None, "done": None, "total": total}
+ done = {"sum": sum, "avg": lambda xs: sum(xs) / len(xs),
+ "max": max, "min": min}.get(func, len)(nums)
+ else:
+ done = sum(1 for vals in values if vals.get(key) not in (None, "", [], {}))
+ if not total:
+ return {"percent": 0.0, "done": done, "total": 0}
+ return {"percent": round(min(float(done) * 100 / total, 100), 1),
+ "done": done, "total": total}
+
+
+def render_progress(block: dict) -> str:
+ stats = progress_value(block.get("rollup_ref") or block)
+ pct = stats["percent"]
+ label = block.get("label") or "Progress"
+ if pct is None:
+ return (f'