diff --git a/docsgpt/api/connector/connections.py b/docsgpt/api/connector/connections.py index 352d8b2e..e3e30ce8 100644 --- a/docsgpt/api/connector/connections.py +++ b/docsgpt/api/connector/connections.py @@ -326,6 +326,14 @@ def _start_sync(user_id: str, row: dict, sync: dict): return ("Pick a GitHub repository", 400) items = {**items, "repo_url": repo} name = name or repo + elif definition.sync_ingestor == "linear": + from docsgpt.connectors import linear + + try: + items = linear.normalize_selection(items) + except ValueError as err: + return (str(err), 400) + name = name or linear.selection_name(items) name = name or definition.name # Validate before claiming the idempotency key: a rejected request must # leave the key free for the corrected retry. @@ -346,11 +354,15 @@ def _start_sync(user_id: str, row: dict, sync: dict): "sync_frequency": frequency, } else: - fields = {f.key for f in definition.setup_fields} - source_data = {k: v for k, v in items.items() if k in fields and v not in (None, "")} - missing = [f.label for f in definition.setup_fields if f.required and f.key not in source_data] - if missing: - return (f"Missing: {', '.join(missing)}", 400) + if definition.sync_ingestor == "linear": + # The teams and projects picked, read with the connection's MCP sign-in. + source_data = items + else: + fields = {f.key for f in definition.setup_fields} + source_data = {k: v for k, v in items.items() if k in fields and v not in (None, "")} + missing = [f.label for f in definition.setup_fields if f.required and f.key not in source_data] + if missing: + return (f"Missing: {', '.join(missing)}", 400) task_fn = ingest_remote kwargs = { "source_data": source_data, @@ -418,6 +430,36 @@ class ConnectionRepositories(Resource): ) +@connections_ns.route("/connections//linear") +class LinearWorkspace(Resource): + @api.doc( + description=( + "Linear: the teams and projects the connection can see, for the sync picker. " + "Read through Linear's MCP server with the connection's sign-in." + ) + ) + def get(self, connection_id: str): + from docsgpt.connectors import linear, mcp + + user_id = _user_id() + if not user_id: + return _unauthorized() + with db_readonly() as conn: + row = _owned(conn, connection_id, user_id) + if row is None or catalog.connector_key_for_row(row) != linear.LINEAR_CONNECTOR: + return _not_found() + try: + workspace = mcp.run_connection_session(row, linear.mcp_url(), linear.list_workspace) + except service.ConnectionUnavailable: + return _error("Reconnect to continue", 409, code="reconnect") + except service.TransientConnectionError: + return _error("Linear is not responding. Try again.", 503) + except Exception as err: + current_app.logger.error(f"Error listing Linear teams: {err}", exc_info=True) + return _error("Failed to list Linear teams", 502) + return make_response(jsonify({"success": True, **workspace}), 200) + + @connections_ns.route("/connections//reconnect") class ConnectionReconnect(Resource): @api.doc( diff --git a/docsgpt/connectors/catalog.py b/docsgpt/connectors/catalog.py index ce3eefcc..a9e4e951 100644 --- a/docsgpt/connectors/catalog.py +++ b/docsgpt/connectors/catalog.py @@ -406,7 +406,9 @@ def _load_presets(path: Optional[Path] = None) -> tuple[ConnectorDefinition, ... auth_kind=entry.get("auth_kind", "mcp_oauth"), capabilities=tuple(entry.get("capabilities") or ("read", "write")), tool_templates=("mcp_tool",), - setup={"tools": "auto", "sync": "off"}, + # Linear also syncs into Knowledge, read through its MCP server. + sync_ingestor=entry.get("sync_ingestor"), + setup={"tools": "auto", "sync": "ask" if entry.get("sync_ingestor") else "off"}, mcp_url=entry["mcp_url"], publisher="preset", docs_url=entry.get("docs_url"), diff --git a/docsgpt/connectors/linear.py b/docsgpt/connectors/linear.py new file mode 100644 index 00000000..fec52efe --- /dev/null +++ b/docsgpt/connectors/linear.py @@ -0,0 +1,299 @@ +"""Linear through its MCP server: what a sync can pick, and reading issues and documents. + +Linear's hosted MCP server (``mcp.linear.app``) is its own OAuth issuer, and +the tokens it grants are for that server. Knowledge sync therefore reads +Linear through the same MCP tools the agents use (``list_issues``, +``get_issue``, ``list_comments``, ``list_documents``...), signed in with the +Linear connection, so one sign-in powers both and nobody registers an OAuth +app. Linear publishes no output schema for these tools, so records are read +defensively: a value may be a string or an object with a ``name``. +""" + +from __future__ import annotations + +import json +import re +from typing import Any, AsyncIterator, Optional + +from docsgpt.connectors import catalog + +LINEAR_CONNECTOR = "mcp:linear" +LINEAR_MCP_URL = "https://mcp.linear.app/mcp" +# Enough for a team's working set; a sync is a full read, so it stays bounded. +MAX_ISSUES = 500 +MAX_DOCUMENTS = 100 +MAX_COMMENTS = 100 +# Teams and projects offered in the picker. +MAX_PICKER_ITEMS = 250 +PAGE_SIZE = 100 +_MAX_PAGES = 50 + +_IDENTIFIER = re.compile(r"^[A-Za-z][A-Za-z0-9_]*-\d+$") +_TRUNCATED = re.compile(r"\(truncated\b", re.IGNORECASE) +_LIST_KEYS = ("nodes", "items", "results", "data") +_PRIORITIES = {0: "No priority", 1: "Urgent", 2: "High", 3: "Medium", 4: "Low"} + + +class LinearSyncError(ValueError): + """Linear's MCP server cannot do what the sync needs (a tool or filter is gone).""" + + +def mcp_url() -> str: + """The Linear preset's MCP endpoint.""" + definition = catalog.get_definition(LINEAR_CONNECTOR) + return (definition.mcp_url if definition and definition.mcp_url else None) or LINEAR_MCP_URL + + +# --------------------------------------------------------------------------- +# What a source syncs +# --------------------------------------------------------------------------- + + +def _truthy(value: Any, default: bool) -> bool: + if value is None: + return default + if isinstance(value, str): + return value.strip().lower() not in ("", "0", "false", "no", "off") + return bool(value) + + +def _picked(values: Any, fields: tuple[str, ...]) -> list[dict]: + picked: dict[str, dict] = {} + for value in values if isinstance(values, list) else []: + record = value if isinstance(value, dict) else {"id": value} + item_id = str(record.get("id") or "").strip() + if not item_id: + continue + current = picked.setdefault(item_id, {"id": item_id, **{f: "" for f in fields}}) + for field in fields: + current[field] = current[field] or str(record.get(field) or "").strip() + return list(picked.values()) + + +def normalize_selection(items: Any) -> dict: + """What a Linear source syncs, as it is stored in ``sources.remote_data``. + + Args: + items: The wizard's choice (or a stored ``remote_data``, possibly as + JSON): ``teams`` and ``projects`` as ids or ``{id, key, name}`` + records, ``include_comments`` (on unless turned off) and + ``include_documents`` (the picked projects' documents). + + Returns: + ``{teams: [{id, key, name}], projects: [{id, name}], + include_comments, include_documents}``. + + Raises: + ValueError: Nothing to sync was picked. + """ + if isinstance(items, str): + try: + items = json.loads(items) + except ValueError: + items = {} + items = items if isinstance(items, dict) else {} + selection = { + "teams": _picked(items.get("teams"), ("key", "name")), + "projects": _picked(items.get("projects"), ("name",)), + "include_comments": _truthy(items.get("include_comments"), True), + "include_documents": _truthy(items.get("include_documents"), False), + } + if not selection["teams"] and not selection["projects"]: + raise ValueError("Pick at least one Linear team or project") + return selection + + +def selection_name(selection: dict) -> str: + """``Linear · Engineering, Acme``: a source's default name, from what it syncs.""" + names = [item.get("name") for item in (*selection["teams"], *selection["projects"]) if item.get("name")] + return f"Linear · {', '.join(names)}" if names else "Linear" + + +# --------------------------------------------------------------------------- +# Reading tool answers +# --------------------------------------------------------------------------- + + +def items_of(payload: Any, key: str) -> list[dict]: + """The records in a list tool's answer: ``{key: [...]}``, a GraphQL ``nodes`` list or a bare list.""" + if isinstance(payload, list): + found = payload + elif isinstance(payload, dict): + found = payload.get(key) + if isinstance(found, dict): + found = found.get("nodes") + if not isinstance(found, list): + found = next((payload[k] for k in _LIST_KEYS if isinstance(payload.get(k), list)), None) + if found is None: + lists = [value for value in payload.values() if isinstance(value, list)] + found = lists[0] if len(lists) == 1 else [] + else: + found = [] + return [item for item in found if isinstance(item, dict)] + + +def next_cursor(payload: Any) -> Optional[str]: + """The cursor of the next page, or None on the last one.""" + if not isinstance(payload, dict): + return None + info = payload.get("pageInfo") if isinstance(payload.get("pageInfo"), dict) else payload + if info.get("hasNextPage") is False: + return None + cursor = info.get("cursor") or info.get("nextCursor") or info.get("endCursor") + return str(cursor) if cursor else None + + +def unwrap(payload: Any, key: str) -> dict: + """A single record from a get tool's answer, bare or as ``{key: {...}}``.""" + if isinstance(payload, dict) and isinstance(payload.get(key), dict): + return payload[key] + return payload if isinstance(payload, dict) else {} + + +def name_of(value: Any) -> str: + """A person's, state's or label's name, whether Linear sent a string or an object.""" + if isinstance(value, dict): + value = value.get("name") or value.get("displayName") or value.get("title") or value.get("key") + return str(value).strip() if value not in (None, "") else "" + + +def names_of(values: Any) -> list[str]: + """Label names from a list of strings or objects (or a ``{nodes: [...]}``).""" + if isinstance(values, dict): + values = values.get("nodes") + return [name for name in (name_of(value) for value in values or []) if name] if isinstance(values, list) else [] + + +def priority_of(value: Any) -> str: + """``High``: a priority given as a label, an object or Linear's 0-4 number.""" + if isinstance(value, dict): + return str(value.get("name") or _PRIORITIES.get(value.get("value"), "")).strip() + if isinstance(value, bool): + return "" + if isinstance(value, int): + return _PRIORITIES.get(value, "") if value else "" + return str(value or "").strip() + + +def issue_identifier(issue: dict) -> str: + """``ENG-123``: the identifier people use, which list tools may send as the ``id``.""" + identifier = issue.get("identifier") + if identifier: + return str(identifier) + item_id = str(issue.get("id") or "") + return item_id if _IDENTIFIER.match(item_id) else "" + + +def is_truncated(text: Any) -> bool: + """Whether Linear clipped a description in a list (it marks it ``(truncated…)``).""" + return isinstance(text, str) and bool(_TRUNCATED.search(text)) + + +# --------------------------------------------------------------------------- +# Calling the tools +# --------------------------------------------------------------------------- + + +async def _schema(session: Any, tool: str) -> dict: + schema = await session.input_schema(tool) + if schema is None: + raise LinearSyncError(f"Linear's MCP server no longer offers {tool}") + return schema + + +def argument(schema: dict, *names: str) -> Optional[str]: + """The first of ``names`` the tool takes; the first name when the schema lists no properties.""" + properties = schema.get("properties") if isinstance(schema, dict) else None + if not isinstance(properties, dict) or not properties: + return names[0] + return next((name for name in names if name in properties), None) + + +def _page_size(schema: dict, name: str) -> int: + spec = (schema.get("properties") or {}).get(name) if isinstance(schema, dict) else None + maximum = spec.get("maximum") if isinstance(spec, dict) else None + return min(PAGE_SIZE, int(maximum)) if isinstance(maximum, (int, float)) and maximum > 0 else PAGE_SIZE + + +async def paged( + session: Any, + tool: str, + key: str, + filters: dict, + *, + limit: int, + extra: Optional[dict] = None, +) -> AsyncIterator[dict]: + """Every record a list tool returns for ``filters``, page by page, up to ``limit``. + + Args: + session: The open MCP session. + tool: The list tool, e.g. ``list_issues``. + key: Where its answer holds the records, e.g. ``issues``. + filters: Filters the sync depends on, as ``{(name, alias...): value}``. + A filter the tool does not take is an error, never dropped: a + team's sync must not read the whole workspace. + limit: Most records to yield. + extra: Optional arguments, sent only when the tool takes them. + + Raises: + LinearSyncError: The tool is gone or cannot apply a filter. + """ + schema = await _schema(session, tool) + arguments: dict = {} + for names, value in filters.items(): + name = argument(schema, *names) + if name is None: + raise LinearSyncError(f"Linear's {tool} can no longer filter by {names[0]}") + arguments[name] = value + size_name = argument(schema, "limit", "first") + if size_name: + arguments[size_name] = _page_size(schema, size_name) + properties = schema.get("properties") if isinstance(schema, dict) else None + for name, value in (extra or {}).items(): + if isinstance(properties, dict) and name in properties: + arguments[name] = value + cursor_name = argument(schema, "cursor", "after") + cursor: Optional[str] = None + seen: set[str] = set() + yielded = 0 + for _ in range(_MAX_PAGES): + payload = await session.call(tool, {**arguments, **({cursor_name: cursor} if cursor else {})}) + for record in items_of(payload, key): + yield record + yielded += 1 + if yielded >= limit: + return + cursor = next_cursor(payload) + if not cursor or not cursor_name or cursor in seen: + return + seen.add(cursor) + + +async def list_workspace(session: Any) -> dict: + """The teams and projects a Linear account can see, for the sync picker. + + Returns: + ``{teams: [{id, key, name}], projects: [{id, name, state, teams}]}``, + each sorted by name. + """ + teams = [] + async for team in paged(session, "list_teams", "teams", {}, limit=MAX_PICKER_ITEMS): + if team.get("id"): + teams.append({"id": str(team["id"]), "key": str(team.get("key") or ""), + "name": name_of(team.get("name")) or str(team.get("key") or team["id"])}) + projects = [] + async for project in paged(session, "list_projects", "projects", {}, limit=MAX_PICKER_ITEMS, + extra={"includeArchived": False}): + if project.get("id"): + project_teams = project.get("teams") or ([project["team"]] if project.get("team") else []) + projects.append({ + "id": str(project["id"]), + "name": name_of(project.get("name")) or str(project["id"]), + "state": name_of(project.get("state") or project.get("status")), + "teams": names_of(project_teams), + }) + return { + "teams": sorted(teams, key=lambda t: t["name"].lower()), + "projects": sorted(projects, key=lambda p: p["name"].lower()), + } diff --git a/docsgpt/connectors/presets/mcp.yaml b/docsgpt/connectors/presets/mcp.yaml index ee1ec692..f185a518 100644 --- a/docsgpt/connectors/presets/mcp.yaml +++ b/docsgpt/connectors/presets/mcp.yaml @@ -9,6 +9,9 @@ # # Keys look like ``mcp:``. The frontend reads ``settings.connectors # .descriptions.mcp_`` for a translated description. +# +# A preset with ``sync_ingestor`` also syncs into Knowledge through its own +# MCP server, signed in with the same connection (Linear's issues). - key: mcp:notion name: Notion @@ -22,12 +25,13 @@ - key: mcp:linear name: Linear - description: Find, create and update Linear issues and projects. + description: Sync issues and documents into Knowledge, and let agents find, create and update issues. icon: linear category: projects mcp_url: https://mcp.linear.app/mcp auth_kind: mcp_oauth - capabilities: [read, write] + capabilities: [sync, read, write] + sync_ingestor: linear docs_url: https://linear.app/docs/mcp - key: mcp:atlassian diff --git a/docsgpt/parser/remote/linear_loader.py b/docsgpt/parser/remote/linear_loader.py new file mode 100644 index 00000000..c768112f --- /dev/null +++ b/docsgpt/parser/remote/linear_loader.py @@ -0,0 +1,219 @@ +"""Load Linear issues and documents as Knowledge, through Linear's MCP server. + +A Linear source names teams and projects (see +``docsgpt.connectors.linear.normalize_selection``). Each issue becomes one +document: its identifier and title, state, assignee, priority, labels, +description and, optionally, its comments. With ``include_documents`` the +picked projects' Linear documents come too. Every document carries its +Linear URL as ``source``, so answers cite the issue. + +Each sync reads everything again (up to ``MAX_ISSUES`` issues and +``MAX_DOCUMENTS`` documents): the index of a synced source is rebuilt +whole, so there is nothing to merge an incremental read into. +""" + +from __future__ import annotations + +import logging +import re +from typing import Any, Optional + +from docsgpt.connectors import linear +from docsgpt.connectors.mcp import run_connection_session +from docsgpt.parser.remote.base import BaseRemote +from docsgpt.parser.schema.base import Document + +logger = logging.getLogger(__name__) + +_UNSAFE_PATH = re.compile(r"[\\/\x00-\x1f]+") + + +def _connection(connection_id: str) -> Optional[dict]: + from docsgpt.storage.db.repositories.connector_sessions import ConnectorSessionsRepository + from docsgpt.storage.db.session import db_readonly + + with db_readonly() as conn: + return ConnectorSessionsRepository(conn).get(str(connection_id)) + + +def _segment(value: str, fallback: str) -> str: + """A file tree segment: no slashes or control characters, never empty.""" + cleaned = _UNSAFE_PATH.sub(" ", value or "").strip().strip(".") + return cleaned[:120] or fallback + + +def _day(value: Any) -> str: + return str(value)[:10] if value else "" + + +class LinearLoader(BaseRemote): + """Read a Linear source's issues and documents with its connection's sign-in.""" + + def load_data(self, inputs: dict) -> list[Document]: + """Read what ``inputs`` picks, signed in with its ``connection_id``. + + Args: + inputs: The source's selection plus ``connection_id``, merged in + by the worker from the source's connection. + + Raises: + ValueError: No connection, or nothing picked. + docsgpt.connectors.service.ConnectionUnavailable: The connection + needs reconnecting. + """ + inputs = inputs if isinstance(inputs, dict) else {} + connection_id = inputs.get("connection_id") + connection = _connection(connection_id) if connection_id else None + if connection is None: + raise ValueError("A Linear source needs its Linear connection") + selection = linear.normalize_selection(inputs) + return run_connection_session( + connection, linear.mcp_url(), lambda session: self.collect(session, selection), + ) + + async def collect(self, session: Any, selection: dict) -> list[Document]: + """The documents for ``selection``, read over an open MCP session.""" + issues: dict[str, dict] = {} + filters = [(("team", "teamId"), team["id"]) for team in selection["teams"]] + filters += [(("project", "projectId"), project["id"]) for project in selection["projects"]] + for names, value in filters: + remaining = linear.MAX_ISSUES - len(issues) + if remaining <= 0: + logger.info("Linear sync stopped at %d issues", linear.MAX_ISSUES) + break + async for issue in linear.paged( + session, "list_issues", "issues", {names: value}, limit=remaining, + extra={"orderBy": "updatedAt", "includeArchived": False}, + ): + key = linear.issue_identifier(issue) or str(issue.get("id") or "") + if key: + issues.setdefault(key, issue) + documents = [] + for key, issue in issues.items(): + if linear.is_truncated(issue.get("description")) or issue.get("descriptionTruncated"): + issue = await self._full_issue(session, key, issue) + comments = await self._comments(session, key) if selection["include_comments"] else [] + documents.append(self._issue_document(issue, comments, selection)) + if selection["include_documents"]: + documents.extend(await self._project_documents(session, selection["projects"])) + return documents + + async def _full_issue(self, session: Any, key: str, issue: dict) -> dict: + schema = await session.input_schema("get_issue") + if schema is None: + return issue + full = linear.unwrap(await session.call("get_issue", {linear.argument(schema, "id", "issueId"): key}), "issue") + return {**issue, **{k: v for k, v in full.items() if v not in (None, "")}} + + async def _comments(self, session: Any, key: str) -> list[dict]: + """An issue's comments, oldest first as Linear lists them; none when Linear cannot list them.""" + schema = await session.input_schema("list_comments") + names = ("issueId", "id", "issue") + if schema is None or linear.argument(schema, *names) is None: + return [] + return [ + comment async for comment in linear.paged( + session, "list_comments", "comments", {names: key}, limit=linear.MAX_COMMENTS, + ) + ] + + def _issue_document(self, issue: dict, comments: list[dict], selection: dict) -> Document: + identifier = linear.issue_identifier(issue) or str(issue.get("id") or "") + title = str(issue.get("title") or "").strip() or identifier + heading = f"{identifier}: {title}" if identifier else title + facts = [ + ("State", linear.name_of(issue.get("status") or issue.get("state"))), + ("Assignee", linear.name_of(issue.get("assignee"))), + ("Priority", linear.priority_of(issue.get("priority"))), + ("Labels", ", ".join(linear.names_of(issue.get("labels")))), + ("Project", linear.name_of(issue.get("project"))), + ("Team", linear.name_of(issue.get("team"))), + ("Due", _day(issue.get("dueDate"))), + ("Created", _day(issue.get("createdAt"))), + ("Updated", _day(issue.get("updatedAt"))), + ("Link", str(issue.get("url") or "")), + ] + lines = [f"# {heading}", ""] + lines += [f"- {label}: {value}" for label, value in facts if value] + description = str(issue.get("description") or "").strip() + if description: + lines += ["", "## Description", "", description] + written = [c for c in comments if str(c.get("body") or "").strip()] + if written: + lines += ["", "## Comments"] + for comment in written: + author = linear.name_of(comment.get("user") or comment.get("author")) or "Someone" + when = _day(comment.get("createdAt")) + lines += ["", f"**{author}**" + (f" · {when}" if when else ""), str(comment["body"]).strip()] + return Document( + text="\n".join(lines).strip() + "\n", + extra_info={ + "title": heading, + "source": str(issue.get("url") or ""), + "file_path": f"{self._team_folder(issue, identifier, selection)}/{_segment(identifier, 'issue')}.md", + "linear_id": str(issue.get("id") or identifier), + "updated_at": str(issue.get("updatedAt") or ""), + }, + ) + + @staticmethod + def _team_folder(issue: dict, identifier: str, selection: dict) -> str: + """The folder an issue is filed under: its team's key (``ENG``), else its team's name.""" + team = issue.get("team") + key = team.get("key") if isinstance(team, dict) else None + if not key and "-" in identifier: + key = identifier.rsplit("-", 1)[0] + return _segment(str(key or linear.name_of(team) or ""), "Issues") + + async def _project_documents(self, session: Any, projects: list[dict]) -> list[Document]: + """The Linear documents of ``projects``, each read in full, up to ``MAX_DOCUMENTS``.""" + documents: list[Document] = [] + paths: set[str] = set() + get_schema = await session.input_schema("get_document") + for project in projects: + remaining = linear.MAX_DOCUMENTS - len(documents) + if remaining <= 0: + break + async for record in linear.paged( + session, "list_documents", "documents", {("projectId", "project"): project["id"]}, limit=remaining, + ): + content = record.get("content") + if (not content or linear.is_truncated(content)) and get_schema is not None and record.get("id"): + full = linear.unwrap( + await session.call("get_document", {linear.argument(get_schema, "id", "documentId"): + str(record["id"])}), + "document", + ) + record = {**record, **{k: v for k, v in full.items() if v not in (None, "")}} + document = self._linear_document(record, project) + path = document.extra_info["file_path"] + if path in paths: + # Two documents with one title stay two files. + document.extra_info["file_path"] = f"{path[:-3]} ({str(record.get('id') or len(paths))[:8]}).md" + paths.add(document.extra_info["file_path"]) + documents.append(document) + return documents + + @staticmethod + def _linear_document(record: dict, project: dict) -> Document: + title = str(record.get("title") or "").strip() or "Untitled document" + content = str(record.get("content") or "").strip() + project_name = project.get("name") or linear.name_of(record.get("project")) + lines = [f"# {title}", ""] + if project_name: + lines.append(f"- Project: {project_name}") + if record.get("url"): + lines.append(f"- Link: {record['url']}") + if content: + lines += ["", content] + folder = _segment(project_name or "", "Documents") + return Document( + text="\n".join(lines).strip() + "\n", + extra_info={ + "title": title, + "source": str(record.get("url") or ""), + "file_path": f"{folder}/Documents/{_segment(title, 'Document')}.md", + "linear_id": str(record.get("id") or ""), + "updated_at": str(record.get("updatedAt") or ""), + }, + ) diff --git a/docsgpt/parser/remote/remote_creator.py b/docsgpt/parser/remote/remote_creator.py index b6f396c5..585ea9b0 100644 --- a/docsgpt/parser/remote/remote_creator.py +++ b/docsgpt/parser/remote/remote_creator.py @@ -5,6 +5,7 @@ from docsgpt.parser.remote.crawler_loader import CrawlerLoader from docsgpt.parser.remote.web_loader import WebLoader from docsgpt.parser.remote.reddit_loader import RedditPostsLoaderRemote from docsgpt.parser.remote.github_loader import GitHubLoader +from docsgpt.parser.remote.linear_loader import LinearLoader from docsgpt.parser.remote.s3_loader import S3Loader @@ -26,6 +27,7 @@ class RemoteCreator: "reddit": RedditPostsLoaderRemote, "github": GitHubLoader, "s3": S3Loader, + "linear": LinearLoader, } @classmethod @@ -88,5 +90,5 @@ def normalize_remote_data(source_type, remote_data): return json.dumps(remote_data) return remote_data - # s3's loader accepts a dict or JSON string; pass it through unchanged. + # s3's and linear's loaders accept a dict or JSON string; pass it through unchanged. return remote_data diff --git a/docsgpt/worker.py b/docsgpt/worker.py index 0e5242bb..2c3fe9ca 100755 --- a/docsgpt/worker.py +++ b/docsgpt/worker.py @@ -1512,6 +1512,10 @@ def sync( return {"status": "success"} +# Remote loaders that can only read with a connection (no public fallback). +_CONNECTION_ONLY_LOADERS = frozenset({"linear"}) + + def sync_worker(self, frequency): from sqlalchemy import text as sql_text @@ -1565,6 +1569,11 @@ def sync_worker(self, frequency): # notifying) on every schedule until then. sync_counts["sync_skipped"] += 1 continue + if source_type in _CONNECTION_ONLY_LOADERS and not doc.get("connection_id"): + # Linear is read only with a connection's sign-in; its + # connection was removed and the content kept. + sync_counts["sync_skipped"] += 1 + continue source_data = normalize_remote_data(source_type, doc.get("remote_data")) if not source_data: @@ -2217,9 +2226,10 @@ def _with_connection_credentials(source_data, connection_id: str): S3, Reddit and GitHub sources made from a connection keep their keys on the connection only, never in ``sources.remote_data``. A JSON string stays a JSON string and a dict a dict; any other string (a GitHub - repository URL) becomes ``{"url": ...}`` next to the keys. Returns None - when the connection is gone, needs reconnecting, or its connector is - turned off. + repository URL) becomes ``{"url": ...}`` next to the keys. An MCP + sign-in (Linear) gets its ``connection_id`` instead: its tokens stay + with the MCP client, which renews them. Returns None when the + connection is gone, needs reconnecting, or its connector is turned off. """ from docsgpt.connectors import service from docsgpt.storage.db.repositories.connector_sessions import ConnectorSessionsRepository @@ -2229,10 +2239,15 @@ def _with_connection_credentials(source_data, connection_id: str): enabled = row is not None and service.connector_enabled(conn, row) if row is None or not enabled: return None - try: - credentials = service.access_credentials(row) - except service.ConnectionUnavailable: - return None + if (row.get("auth_kind") or "") == "mcp_oauth": + if service.normalize_status(row) != service.STATUS_CONNECTED: + return None + credentials = {"connection_id": str(row["id"])} + else: + try: + credentials = service.access_credentials(row) + except service.ConnectionUnavailable: + return None if isinstance(source_data, str): try: parsed = json.loads(source_data) diff --git a/tests/connectors/test_linear.py b/tests/connectors/test_linear.py new file mode 100644 index 00000000..7e94ed10 --- /dev/null +++ b/tests/connectors/test_linear.py @@ -0,0 +1,279 @@ +"""Linear as Knowledge: the same MCP sign-in that gives agents Linear's tools syncs its issues.""" + +from __future__ import annotations + +import asyncio +from contextlib import contextmanager +from unittest.mock import MagicMock, patch + +import pytest +from flask import Flask +from sqlalchemy import text + +import docsgpt.api.user # noqa: F401 (loads mcp_tool without the circular import) +from docsgpt.connectors import catalog, linear, service +from docsgpt.security.encryption import encrypt_json + + +@pytest.fixture +def app(): + return Flask(__name__) + + +@contextmanager +def _db(conn): + @contextmanager + def _yield(): + yield conn + + with patch.multiple("docsgpt.api.connector.connections", db_session=_yield, db_readonly=_yield), \ + patch.multiple("docsgpt.connectors.service", db_session=_yield, db_readonly=_yield): + yield + + +def _call(app, resource, method, path, user="alice", body=None, args=()): + with app.test_request_context(path, method=method.upper(), json=body): + from flask import request + + request.decoded_token = {"sub": user} if user else None + return getattr(resource(), method)(*args) + + +def _linear(conn, user="alice", status="connected") -> str: + tokens = {"tokens": {"access_token": "lin-mcp", "token_type": "Bearer", "refresh_token": "r"}} + return str(conn.execute( + text( + "INSERT INTO connector_sessions (user_id, provider, connector_key, auth_kind, status, account_label, " + "server_url, encrypted_credentials) VALUES (:u, 'mcp:https://mcp.linear.app', 'mcp:linear', " + "'mcp_oauth', :s, 'alice@acme.com', 'https://mcp.linear.app', :e) RETURNING id" + ), + {"u": user, "s": status, "e": encrypt_json(tokens, user)}, + ).scalar()) + + +class FakeLinear: + def __init__(self, answers): + self.answers = answers + + async def input_schema(self, name): + return {"properties": {"limit": {}, "cursor": {}}} if name in self.answers else None + + async def call(self, name, arguments): + return self.answers[name] + + +def _session(answers): + """``run_connection_session`` answering from ``answers``, recording the connection it was given.""" + seen = {} + + def run(connection, server_url, work, **kwargs): + seen.update(connection=connection, server_url=server_url) + return asyncio.run(work(FakeLinear(answers))) + + return run, seen + + +class TestCatalog: + @pytest.fixture(autouse=True) + def _fresh_registry(self): + catalog.reset_registry_for_tests() + yield + catalog.reset_registry_for_tests() + + def test_one_linear_sign_in_syncs_and_gives_tools(self): + definition = catalog.get_definition("mcp:linear") + assert definition.capabilities == ("sync", "read", "write") + assert definition.sync_ingestor == "linear" + assert definition.setup == {"tools": "auto", "sync": "ask"} + assert definition.tool_templates == ("mcp_tool",) + assert definition.auth_kind == "mcp_oauth" + + def test_other_presets_still_only_give_tools(self): + notion = catalog.get_definition("mcp:notion") + assert notion.sync_ingestor is None + assert notion.setup == {"tools": "auto", "sync": "off"} + assert "sync" not in notion.capabilities + + def test_linear_sources_have_a_loader(self): + from docsgpt.parser.remote.linear_loader import LinearLoader + from docsgpt.parser.remote.remote_creator import RemoteCreator, normalize_remote_data + + assert isinstance(RemoteCreator.create_loader("linear"), LinearLoader) + stored = {"teams": [{"id": "t1"}], "include_comments": True} + assert normalize_remote_data("linear", stored) == stored + + +class TestWorkspace: + def test_lists_the_teams_and_projects_to_pick(self, app, pg_conn): + from docsgpt.api.connector.connections import LinearWorkspace + + cid = _linear(pg_conn) + run, seen = _session({ + "list_teams": {"teams": [{"id": "t2", "key": "OPS", "name": "Operations"}, + {"id": "t1", "key": "ENG", "name": "Engineering"}]}, + "list_projects": [{"id": "p1", "name": "Acme", "state": {"name": "Started"}, "teams": ["Engineering"]}], + }) + with _db(pg_conn), patch("docsgpt.connectors.mcp.run_connection_session", side_effect=run): + resp = _call(app, LinearWorkspace, "get", f"/api/connections/{cid}/linear", args=[cid]) + assert resp.status_code == 200 + data = resp.get_json() + assert data["teams"] == [ + {"id": "t1", "key": "ENG", "name": "Engineering"}, {"id": "t2", "key": "OPS", "name": "Operations"}, + ] + assert data["projects"] == [{"id": "p1", "name": "Acme", "state": "Started", "teams": ["Engineering"]}] + assert str(seen["connection"]["id"]) == cid + assert seen["server_url"] == "https://mcp.linear.app/mcp" + assert "lin-mcp" not in resp.get_data(as_text=True) + + def test_a_lost_sign_in_asks_to_reconnect(self, app, pg_conn): + from docsgpt.api.connector.connections import LinearWorkspace + + cid = _linear(pg_conn) + with _db(pg_conn), patch("docsgpt.connectors.mcp.run_connection_session", + side_effect=service.ConnectionUnavailable("expired", connection_id=cid)): + resp = _call(app, LinearWorkspace, "get", f"/api/connections/{cid}/linear", args=[cid]) + assert resp.status_code == 409 + assert resp.get_json()["code"] == "reconnect" + + def test_linear_not_answering_is_worth_a_retry(self, app, pg_conn): + from docsgpt.api.connector.connections import LinearWorkspace + + cid = _linear(pg_conn) + with _db(pg_conn), patch("docsgpt.connectors.mcp.run_connection_session", + side_effect=service.TransientConnectionError("rate limit")): + resp = _call(app, LinearWorkspace, "get", f"/api/connections/{cid}/linear", args=[cid]) + assert resp.status_code == 503 + + def test_only_the_owner_and_only_linear(self, app, pg_conn): + from docsgpt.api.connector.connections import LinearWorkspace + + theirs = _linear(pg_conn, user="bob") + github = str(pg_conn.execute(text( + "INSERT INTO connector_sessions (user_id, provider, connector_key, auth_kind, status) " + "VALUES ('alice', 'github', 'github', 'api_key', 'connected') RETURNING id" + )).scalar()) + with _db(pg_conn): + for cid in (theirs, github): + resp = _call(app, LinearWorkspace, "get", f"/api/connections/{cid}/linear", args=[cid]) + assert resp.status_code == 404 + + +class TestSetup: + def test_sync_queues_the_picked_teams_with_the_connection(self, app, pg_conn): + from docsgpt.api.connector.connections import ConnectionSetup + + cid = _linear(pg_conn) + with _db(pg_conn), patch("docsgpt.api.user.tasks.ingest_remote.apply_async", + return_value=MagicMock(id="t")) as apply: + resp = _call(app, ConnectionSetup, "post", f"/api/connections/{cid}/setup", body={ + "create_tools": False, + "sync": {"items": {"teams": [{"id": "t1", "key": "ENG", "name": "Engineering"}], + "include_comments": False}, "frequency": "daily"}, + }, args=[cid]) + assert resp.status_code == 200 + kwargs = apply.call_args.kwargs["kwargs"] + assert kwargs["loader"] == "linear" + assert kwargs["connection_id"] == cid + assert kwargs["sync_frequency"] == "daily" + assert kwargs["source_data"] == { + "teams": [{"id": "t1", "key": "ENG", "name": "Engineering"}], "projects": [], + "include_comments": False, "include_documents": False, + } + assert resp.get_json()["sources"][0]["name"] == "Linear · Engineering" + + def test_nothing_picked_is_a_bad_request(self, app, pg_conn): + from docsgpt.api.connector.connections import ConnectionSetup + + cid = _linear(pg_conn) + with _db(pg_conn), patch("docsgpt.api.user.tasks.ingest_remote.apply_async") as apply: + resp = _call(app, ConnectionSetup, "post", f"/api/connections/{cid}/setup", body={ + "create_tools": False, "sync": {"items": {"teams": []}}, + }, args=[cid]) + assert resp.status_code == 400 + apply.assert_not_called() + + +class TestWorker: + @staticmethod + @contextmanager + def _readonly(conn): + from docsgpt import worker + + @contextmanager + def _yield(): + yield conn + + with patch.object(worker, "db_readonly", _yield): + yield worker + + def test_the_loader_gets_the_connection_not_its_tokens(self, pg_conn): + cid = _linear(pg_conn) + with self._readonly(pg_conn) as worker: + data = worker._with_connection_credentials({"teams": [{"id": "t1"}]}, cid) + assert data == {"teams": [{"id": "t1"}], "connection_id": cid} + + def test_a_connection_that_needs_reconnecting_gives_nothing(self, pg_conn): + cid = _linear(pg_conn, status="reconnect_needed") + with self._readonly(pg_conn) as worker: + assert worker._with_connection_credentials({"teams": [{"id": "t1"}]}, cid) is None + + def test_scheduled_sync_reads_with_the_connection(self, pg_conn): + cid = _linear(pg_conn) + pg_conn.execute(text( + "INSERT INTO sources (user_id, name, type, sync_frequency, connection_id, remote_data) " + "VALUES ('alice', 'Linear · Engineering', 'linear', 'daily', CAST(:c AS uuid), " + "'{\"teams\": [{\"id\": \"t1\"}]}')" + ), {"c": cid}) + with self._readonly(pg_conn) as worker, patch.object(worker, "sync", return_value={"status": "success"}) as sync: + counts = worker.sync_worker(MagicMock(), "daily") + assert counts["sync_success"] == 1 + args, kwargs = sync.call_args + assert args[1] == {"teams": [{"id": "t1"}]} + assert args[4] == "linear" + assert kwargs["connection_id"] == cid + + def test_scheduled_sync_skips_a_source_paused_for_reconnect(self, pg_conn): + cid = _linear(pg_conn, status="reconnect_needed") + pg_conn.execute(text( + "INSERT INTO sources (user_id, name, type, sync_frequency, connection_id, remote_data, metadata) " + "VALUES ('alice', 'Linear', 'linear', 'daily', CAST(:c AS uuid), '{\"teams\": [{\"id\": \"t1\"}]}', " + "'{\"sync_state\": \"paused_reconnect\"}')" + ), {"c": cid}) + with self._readonly(pg_conn) as worker, patch.object(worker, "sync") as sync: + counts = worker.sync_worker(MagicMock(), "daily") + sync.assert_not_called() + assert counts["sync_skipped"] == 1 + + def test_scheduled_sync_skips_a_source_whose_connection_was_removed(self, pg_conn): + pg_conn.execute(text( + "INSERT INTO sources (user_id, name, type, sync_frequency, remote_data) " + "VALUES ('alice', 'Linear', 'linear', 'daily', '{\"teams\": [{\"id\": \"t1\"}]}')" + )) + with self._readonly(pg_conn) as worker, patch.object(worker, "sync") as sync: + counts = worker.sync_worker(MagicMock(), "daily") + sync.assert_not_called() + assert counts["sync_skipped"] == 1 + + def test_a_lost_sign_in_fails_the_ingest_without_retrying_forever(self): + from docsgpt import worker + + loader = MagicMock() + loader.load_data.side_effect = service.ConnectionUnavailable("expired", connection_id="c-1") + task = MagicMock() + task.request.retries = 1 + with patch.object(worker.RemoteCreator, "create_loader", return_value=loader), patch.object( + worker, "_with_connection_credentials", return_value={"teams": [{"id": "t1"}], "connection_id": "c-1"}, + ), patch.object(worker, "publish_user_event") as publish: + with pytest.raises(service.ConnectionUnavailable): + worker.remote_worker(task, {"teams": [{"id": "t1"}]}, "Linear", "alice", "linear", + connection_id="c-1") + assert publish.call_args.args[1] == "source.ingest.failed" + + +def test_the_picker_lists_are_capped(monkeypatch): + monkeypatch.setattr(linear, "MAX_PICKER_ITEMS", 2) + session = FakeLinear({ + "list_teams": {"teams": [{"id": f"t{n}", "name": f"Team {n}"} for n in range(5)]}, + "list_projects": {"projects": []}, + }) + assert len(asyncio.run(linear.list_workspace(session))["teams"]) == 2 diff --git a/tests/parser/remote/test_linear_loader.py b/tests/parser/remote/test_linear_loader.py new file mode 100644 index 00000000..8527cbf1 --- /dev/null +++ b/tests/parser/remote/test_linear_loader.py @@ -0,0 +1,290 @@ +"""Linear issues and documents as Knowledge, read through Linear's MCP server (mocked).""" + +from __future__ import annotations + +import asyncio +from unittest.mock import patch + +import pytest + +from docsgpt.connectors import linear +from docsgpt.parser.remote.linear_loader import LinearLoader + +# Input schemas shaped like the ones Linear's MCP server publishes. +SCHEMAS = { + "list_issues": {"properties": { + "team": {}, "project": {}, "limit": {"maximum": 250}, "cursor": {}, "orderBy": {}, "includeArchived": {}, + }}, + "get_issue": {"properties": {"id": {}}}, + "list_comments": {"properties": {"issueId": {}}}, + "list_documents": {"properties": {"projectId": {}, "limit": {}, "cursor": {}}}, + "get_document": {"properties": {"id": {}}}, + "list_teams": {"properties": {"limit": {}, "cursor": {}}}, + "list_projects": {"properties": {"limit": {}, "cursor": {}, "includeArchived": {}}}, +} + +ISSUE = { + "id": "ENG-1310", + "title": "Fix the loader", + "description": "The spinner stays for 10 s.", + "url": "https://linear.app/acme/issue/ENG-1310/fix-the-loader", + "status": "In Progress", + "priority": {"value": 2, "name": "High"}, + "assignee": "Sam", + "labels": ["bug", {"id": "l2", "name": "ui"}], + "project": "Acme", + "team": "Engineering", + "createdAt": "2026-09-20T10:00:00.000Z", + "updatedAt": "2026-09-25T09:30:00.000Z", +} + + +class FakeLinear: + """Linear's MCP server, answering tool calls from handlers.""" + + def __init__(self, handlers, schemas=None): + self.handlers = handlers + self.schemas = SCHEMAS if schemas is None else schemas + self.calls: list[tuple[str, dict]] = [] + + async def input_schema(self, name): + return self.schemas.get(name) + + async def call(self, name, arguments): + self.calls.append((name, dict(arguments))) + return self.handlers[name](arguments) + + def called(self, name): + return [args for tool, args in self.calls if tool == name] + + +def _collect(session, **selection): + selection = linear.normalize_selection({"teams": [], "projects": [], **selection}) + return asyncio.run(LinearLoader().collect(session, selection)) + + +class TestIssues: + def test_one_document_per_issue_with_what_it_says(self): + session = FakeLinear({ + "list_issues": lambda args: {"issues": [ISSUE], "hasNextPage": False}, + "list_comments": lambda args: {"comments": [ + {"id": "c1", "body": "Reproduced on the current release.", "user": {"name": "Alex"}, + "createdAt": "2026-09-21T11:00:00.000Z"}, + {"id": "c2", "body": "Fixed.", "author": "Sam", "createdAt": "2026-09-22T11:00:00.000Z"}, + ]}, + }) + [doc] = _collect(session, teams=[{"id": "team-1", "key": "ENG", "name": "Engineering"}]) + text = doc.text + assert text.startswith("# ENG-1310: Fix the loader") + for fragment in ("In Progress", "Sam", "High", "bug, ui", "Acme", "The spinner stays for 10 s.", + "Reproduced on the current release.", "Alex", "Fixed."): + assert fragment in text + assert doc.extra_info["title"] == "ENG-1310: Fix the loader" + assert doc.extra_info["source"] == ISSUE["url"] + assert doc.extra_info["file_path"] == "ENG/ENG-1310.md" + assert session.called("list_comments") == [{"issueId": "ENG-1310"}] + + def test_pages_through_a_teams_issues(self): + pages = { + None: {"issues": [{**ISSUE, "id": "ENG-2"}], "hasNextPage": True, "cursor": "p2"}, + "p2": {"issues": [{**ISSUE, "id": "ENG-1"}], "hasNextPage": False}, + } + session = FakeLinear({"list_issues": lambda args: pages[args.get("cursor")]}) + docs = _collect(session, teams=["team-1"], include_comments=False) + assert [d.extra_info["title"].split(":")[0] for d in docs] == ["ENG-2", "ENG-1"] + first, second = session.called("list_issues") + assert first == {"team": "team-1", "limit": 100, "orderBy": "updatedAt", "includeArchived": False} + assert second == {**first, "cursor": "p2"} + + def test_a_clipped_description_is_read_in_full(self): + clipped = {**ISSUE, "description": "The spinner (truncated, use get_issue to read the full description)"} + session = FakeLinear({ + "list_issues": lambda args: {"issues": [clipped]}, + "get_issue": lambda args: {"issue": {**ISSUE, "description": "The whole story."}}, + }) + [doc] = _collect(session, teams=["team-1"], include_comments=False) + assert "The whole story." in doc.text + assert "truncated" not in doc.text + assert session.called("get_issue") == [{"id": "ENG-1310"}] + + def test_comments_are_paged_through(self): + schemas = {**SCHEMAS, "list_comments": {"properties": {"issueId": {}, "cursor": {}}}} + pages = { + None: {"comments": [{"body": "First."}], "hasNextPage": True, "cursor": "c2"}, + "c2": {"comments": [{"body": "Second."}], "hasNextPage": False}, + } + session = FakeLinear({ + "list_issues": lambda args: [ISSUE], + "list_comments": lambda args: pages[args.get("cursor")], + }, schemas) + [doc] = _collect(session, teams=["team-1"]) + assert "First." in doc.text and "Second." in doc.text + + def test_comments_linear_cannot_list_are_skipped(self): + schemas = {k: v for k, v in SCHEMAS.items() if k != "list_comments"} + session = FakeLinear({"list_issues": lambda args: [ISSUE]}, schemas) + [doc] = _collect(session, teams=["team-1"]) + assert "## Comments" not in doc.text + + def test_comments_are_left_out_when_not_wanted(self): + session = FakeLinear({"list_issues": lambda args: [ISSUE]}) + [doc] = _collect(session, teams=["team-1"], include_comments=False) + assert session.called("list_comments") == [] + assert "## Comments" not in doc.text + + def test_an_issue_in_a_picked_team_and_project_is_synced_once(self): + session = FakeLinear({"list_issues": lambda args: {"issues": [ISSUE]}}) + docs = _collect(session, teams=["team-1"], projects=["project-1"], include_comments=False) + assert len(docs) == 1 + assert [args.get("team") or args.get("project") for args in session.called("list_issues")] == [ + "team-1", "project-1", + ] + + def test_stops_at_the_cap(self, monkeypatch): + monkeypatch.setattr(linear, "MAX_ISSUES", 3) + pages = iter(range(100)) + + def endless(args): + page = next(pages) + issues = [{**ISSUE, "id": f"ENG-{page}{n}"} for n in range(2)] + return {"issues": issues, "hasNextPage": True, "cursor": f"p{page + 1}"} + + session = FakeLinear({"list_issues": endless}) + docs = _collect(session, teams=["team-1"], projects=["project-1"], include_comments=False) + assert len(docs) == 3 + assert len(session.called("list_issues")) == 2 + + def test_a_repeated_cursor_ends_the_listing(self): + session = FakeLinear({"list_issues": lambda args: {"issues": [ISSUE], "hasNextPage": True, "cursor": "same"}}) + _collect(session, teams=["team-1"], include_comments=False) + assert len(session.called("list_issues")) == 2 + + def test_nested_records_and_graphql_page_info_are_read(self): + record = { + "id": "5c1e3d2a-uuid", "identifier": "DES-7", "title": "Logo", + "state": {"id": "s", "name": "Done"}, "assignee": {"id": "u", "displayName": "kim"}, + "team": {"id": "t", "key": "DES", "name": "Design"}, "labels": {"nodes": [{"name": "brand"}]}, + } + session = FakeLinear({"list_issues": lambda args: { + "nodes": [record], "pageInfo": {"hasNextPage": False, "endCursor": "x"}, + }}) + [doc] = _collect(session, teams=["t"], include_comments=False) + assert doc.text.startswith("# DES-7: Logo") + assert "Done" in doc.text and "kim" in doc.text and "brand" in doc.text + assert doc.extra_info["file_path"] == "DES/DES-7.md" + + def test_argument_names_follow_the_tools_schema(self): + schemas = {**SCHEMAS, "list_issues": {"properties": {"teamId": {}, "first": {"maximum": 50}, "after": {}}}} + pages = {None: {"issues": [ISSUE], "hasNextPage": True, "cursor": "p2"}, "p2": {"issues": []}} + session = FakeLinear({"list_issues": lambda args: pages[args.get("after")]}, schemas) + _collect(session, teams=["team-1"], include_comments=False) + assert session.called("list_issues") == [ + {"teamId": "team-1", "first": 50}, {"teamId": "team-1", "first": 50, "after": "p2"}, + ] + + def test_a_tool_that_cannot_filter_by_team_syncs_nothing(self): + schemas = {**SCHEMAS, "list_issues": {"properties": {"query": {}, "limit": {}}}} + session = FakeLinear({"list_issues": lambda args: pytest.fail("must not list the whole workspace")}, schemas) + with pytest.raises(linear.LinearSyncError, match="team"): + _collect(session, teams=["team-1"]) + + def test_a_missing_tool_is_an_error(self): + session = FakeLinear({}, schemas={}) + with pytest.raises(linear.LinearSyncError, match="list_issues"): + _collect(session, teams=["team-1"]) + + +class TestDocuments: + def test_documents_of_the_picked_projects(self): + session = FakeLinear({ + "list_issues": lambda args: {"issues": []}, + "list_documents": lambda args: {"documents": [ + {"id": "d1", "title": "Launch plan", "url": "https://linear.app/acme/document/launch-plan-d1"}, + ]}, + "get_document": lambda args: {"id": "d1", "title": "Launch plan", "content": "We ship on Monday.", + "url": "https://linear.app/acme/document/launch-plan-d1"}, + }) + [doc] = _collect(session, projects=[{"id": "project-1", "name": "Acme"}], include_documents=True) + assert doc.text.startswith("# Launch plan") + assert "We ship on Monday." in doc.text + assert doc.extra_info["source"] == "https://linear.app/acme/document/launch-plan-d1" + assert doc.extra_info["file_path"] == "Acme/Documents/Launch plan.md" + assert session.called("list_documents") == [{"projectId": "project-1", "limit": 100}] + + def test_a_listed_document_with_its_content_is_not_read_again(self): + session = FakeLinear({ + "list_issues": lambda args: [], + "list_documents": lambda args: [{"id": "d1", "title": "Notes", "content": "All here."}], + }) + [doc] = _collect(session, projects=["project-1"], include_documents=True) + assert "All here." in doc.text + assert session.called("get_document") == [] + + def test_documents_with_one_title_stay_two_files(self): + session = FakeLinear({ + "list_issues": lambda args: [], + "list_documents": lambda args: [ + {"id": "d1aaaaaaaa", "title": "Notes", "content": "One."}, + {"id": "d2bbbbbbbb", "title": "Notes", "content": "Two."}, + ], + }) + docs = _collect(session, projects=[{"id": "p1", "name": "Acme"}], include_documents=True) + assert [d.extra_info["file_path"] for d in docs] == [ + "Acme/Documents/Notes.md", "Acme/Documents/Notes (d2bbbbbb).md", + ] + + def test_documents_are_left_out_when_not_wanted(self): + session = FakeLinear({"list_issues": lambda args: []}) + assert _collect(session, projects=["project-1"]) == [] + assert session.called("list_documents") == [] + + +class TestSelection: + def test_ids_or_records_are_kept_once(self): + selection = linear.normalize_selection({ + "teams": ["t1", {"id": "t1", "key": "ENG", "name": "Engineering"}, {"id": "t2"}, "", None], + "projects": [{"id": "p1", "name": "Acme"}], + "include_comments": "false", + "include_documents": True, + "unrelated": "dropped", + }) + assert selection == { + "teams": [{"id": "t1", "key": "ENG", "name": "Engineering"}, {"id": "t2", "key": "", "name": ""}], + "projects": [{"id": "p1", "name": "Acme"}], + "include_comments": False, + "include_documents": True, + } + + def test_comments_are_on_by_default(self): + assert linear.normalize_selection({"teams": ["t1"]})["include_comments"] is True + + def test_something_must_be_picked(self): + with pytest.raises(ValueError): + linear.normalize_selection({"teams": [], "projects": []}) + + def test_a_json_string_is_read(self): + assert linear.normalize_selection('{"teams": ["t1"]}')["teams"] == [{"id": "t1", "key": "", "name": ""}] + + def test_default_name_lists_what_was_picked(self): + selection = linear.normalize_selection({ + "teams": [{"id": "t1", "name": "Engineering"}], "projects": [{"id": "p1", "name": "Acme"}], + }) + assert linear.selection_name(selection) == "Linear · Engineering, Acme" + assert linear.selection_name(linear.normalize_selection({"teams": ["t1"]})) == "Linear" + + +class TestLoadData: + def test_reads_with_the_sources_connection(self): + connection = {"id": "c1", "user_id": "alice", "server_url": "https://mcp.linear.app"} + loader = LinearLoader() + with patch("docsgpt.parser.remote.linear_loader._connection", return_value=connection) as load, \ + patch("docsgpt.parser.remote.linear_loader.run_connection_session", + return_value=["doc"]) as run: + result = loader.load_data({"teams": ["t1"], "connection_id": "c1"}) + assert result == ["doc"] + load.assert_called_once_with("c1") + assert run.call_args.args[:2] == (connection, "https://mcp.linear.app/mcp") + + def test_needs_a_connection(self): + with pytest.raises(ValueError): + LinearLoader().load_data({"teams": ["t1"]})