From 03d1b10897d8a9ee4df1218826c7c3bdb0f8d54e Mon Sep 17 00:00:00 2001 From: arc53-machine <232052973+arc53-machine@users.noreply.github.com> Date: Tue, 29 Sep 2026 11:22:26 +0100 Subject: [PATCH] Sync Linear issues and documents into Knowledge with the Linear sign-in The Linear connector now syncs as well as giving agents its tools, from one connection. Linear's MCP server is its own OAuth issuer, so its tokens are read through the same MCP tools the agents use (list_issues, get_issue, list_comments, list_documents) rather than Linear's GraphQL API, and no OAuth app has to be registered. A source picks teams and projects, with comments (on by default) and the projects' documents. Each issue becomes one document with its state, assignee, priority, labels, description and comments, filed under its team and citing its Linear URL. Each sync reads up to 500 issues and 100 documents again. /api/connections//linear lists the teams and projects to pick from. Sources sync on their schedule with the owner's connection, and pause when the sign-in needs reconnecting. --- docsgpt/api/connector/connections.py | 52 +++- docsgpt/connectors/catalog.py | 4 +- docsgpt/connectors/linear.py | 299 ++++++++++++++++++++++ docsgpt/connectors/presets/mcp.yaml | 8 +- docsgpt/parser/remote/linear_loader.py | 219 ++++++++++++++++ docsgpt/parser/remote/remote_creator.py | 4 +- docsgpt/worker.py | 29 ++- tests/connectors/test_linear.py | 279 ++++++++++++++++++++ tests/parser/remote/test_linear_loader.py | 290 +++++++++++++++++++++ 9 files changed, 1168 insertions(+), 16 deletions(-) create mode 100644 docsgpt/connectors/linear.py create mode 100644 docsgpt/parser/remote/linear_loader.py create mode 100644 tests/connectors/test_linear.py create mode 100644 tests/parser/remote/test_linear_loader.py 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"]})