) => {
+ setConnectionVerified(false);
+ setProjectSelectionLoaded(false);
+ setProjects([]);
+ setProjectSearch("");
+ patch(value);
+ };
+
+ const validateConnection = (
+ source: JiraSource,
+ requireToken = true
+ ) => {
+ const missingFields = [
+ !source.id.trim() && "source id",
+ !source.display_name.trim() && "display name",
+ !source.connection.site_url.trim() && "site URL",
+ !source.connection.email.trim() && "Atlassian email",
+ requireToken && !source.connection.api_token.trim() && "API token",
+ ].filter(Boolean) as string[];
+
+ if (missingFields.length > 0) {
+ const fields =
+ missingFields.length === 1
+ ? missingFields[0]
+ : `${missingFields.slice(0, -1).join(", ")} and ${
+ missingFields[missingFields.length - 1]
+ }`;
+ setConnectionFeedback({
+ type: "error",
+ text: `${
+ fields.charAt(0).toUpperCase() + fields.slice(1)
+ } ${missingFields.length === 1 ? "is" : "are"} required.`,
+ });
+ return false;
+ }
+ return true;
+ };
+
+ const persistSource = async (source: JiraSource) => {
+ const response = await fetch(
+ `/ui/${selectedGraph}/data-sources/${encodeURIComponent(source.id)}`,
+ {
+ method: "PUT",
+ headers: {
+ Authorization: sessionStorage.getItem("auth")!,
+ "Content-Type": "application/json",
+ },
+ body: JSON.stringify(source),
+ }
+ );
+ const data = await safeJson(response);
+ if (!response.ok) throw new Error(data.detail || `HTTP ${response.status}`);
+ return data.source as JiraSource;
+ };
+
+ const save = async () => {
+ if (
+ !editing ||
+ !selectedGraph ||
+ !connectionVerified ||
+ !validateConnection(editing, false)
+ ) {
+ return;
+ }
+ if (editing.scope.project_keys.length === 0) {
+ setSaveFeedback({
+ type: "error",
+ text: "Select at least one Jira project before saving the scope.",
+ });
+ return;
+ }
+ setBusy("save");
+ setSaveFeedback({
+ type: "pending",
+ text: "Saving Jira project scope…",
+ });
+ try {
+ const saved = await persistSource(editing);
+ setEditing(saved);
+ sessionStorage.setItem(editingSourceKey(selectedGraph), saved.id);
+ await loadSources();
+ setSaveFeedback({
+ type: "success",
+ text: `Project scope saved (${saved.scope.project_keys.length} selected).`,
+ });
+ } catch (error: any) {
+ setSaveFeedback({
+ type: "error",
+ text: `Failed to save project scope: ${error.message}`,
+ });
+ } finally {
+ setBusy("");
+ }
+ };
+
+ const test = async () => {
+ if (!editing || !selectedGraph) return;
+ setConnectionVerified(false);
+ setProjectSelectionLoaded(false);
+ setProjects([]);
+ setProjectSearch("");
+ setCountFeedback(null);
+ setSaveFeedback(null);
+ if (!validateConnection(editing)) return;
+ setScopeFeedback(null);
+ setBusy("test");
+ setConnectionFeedback({
+ type: "pending",
+ text: "Testing Jira connection…",
+ });
+ try {
+ const response = await fetch(
+ `/ui/${selectedGraph}/data-sources/${encodeURIComponent(editing.id)}/test`,
+ {
+ method: "POST",
+ headers: {
+ Authorization: sessionStorage.getItem("auth")!,
+ "Content-Type": "application/json",
+ },
+ body: JSON.stringify(editing),
+ }
+ );
+ const data = await safeJson(response);
+ if (!response.ok) throw new Error(data.detail || `HTTP ${response.status}`);
+ setConnectionFeedback({
+ type: "success",
+ text: `Connected as ${data.account?.display_name || "Jira user"}.`,
+ });
+ setConnectionVerified(true);
+ setEditing(data.source);
+ sessionStorage.setItem(editingSourceKey(selectedGraph), data.source.id);
+ await loadSources();
+ } catch (error: any) {
+ setConnectionVerified(false);
+ setConnectionFeedback({
+ type: "error",
+ text: `Connection failed: ${error.message}`,
+ });
+ } finally {
+ setBusy("");
+ }
+ };
+
+ const loadProjects = async () => {
+ if (!editing || !connectionVerified) return;
+ const projectKey = directProjectKey.trim().toUpperCase();
+ setProjectSelectionLoaded(false);
+ setProjects([]);
+ setProjectSearch("");
+ setCountFeedback(null);
+ setBusy("projects");
+ setScopeFeedback({
+ type: "pending",
+ text: projectKey
+ ? `Loading Jira project ${projectKey}…`
+ : "Loading visible Jira projects…",
+ });
+ try {
+ const query = projectKey
+ ? `?project_key=${encodeURIComponent(projectKey)}`
+ : "";
+ const response = await fetch(
+ `/ui/${selectedGraph}/data-sources/${encodeURIComponent(editing.id)}/projects${query}`,
+ { headers: { Authorization: sessionStorage.getItem("auth")! } }
+ );
+ const data = await safeJson(response);
+ if (!response.ok) throw new Error(data.detail || `HTTP ${response.status}`);
+ const visibleProjects = Array.isArray(data.projects) ? data.projects : [];
+ setProjects(visibleProjects);
+ setProjectSearch("");
+ setProjectSelectionLoaded(true);
+ if (projectKey && visibleProjects.length === 1) {
+ setEditing((current) =>
+ current
+ ? {
+ ...current,
+ scope: {
+ ...current.scope,
+ project_keys: [visibleProjects[0].key],
+ },
+ }
+ : current
+ );
+ }
+ setScopeFeedback({
+ type: "success",
+ text:
+ visibleProjects.length > 0
+ ? projectKey
+ ? `Loaded and selected ${visibleProjects[0].key} — ${visibleProjects[0].name}.`
+ : `Loaded ${visibleProjects.length} visible Jira projects.`
+ : "Connection succeeded, but this account has no visible Jira projects.",
+ });
+ } catch (error: any) {
+ setProjectSelectionLoaded(false);
+ setScopeFeedback({
+ type: "error",
+ text: `Failed to load projects: ${error.message}`,
+ });
+ } finally {
+ setBusy("");
+ }
+ };
+
+ const previewTicketCount = async () => {
+ if (
+ !editing ||
+ !connectionVerified ||
+ editing.scope.project_keys.length === 0
+ ) {
+ setCountFeedback({
+ type: "error",
+ text: "Select at least one Jira project before previewing tickets.",
+ });
+ return;
+ }
+ setBusy("count");
+ setCountFeedback({
+ type: "pending",
+ text: "Checking how many Jira tickets match this scope…",
+ });
+ try {
+ const response = await fetch(
+ `/ui/${selectedGraph}/data-sources/${encodeURIComponent(editing.id)}/issues/count`,
+ {
+ method: "POST",
+ headers: {
+ Authorization: sessionStorage.getItem("auth")!,
+ "Content-Type": "application/json",
+ },
+ body: JSON.stringify(editing.scope),
+ }
+ );
+ const data = await safeJson(response);
+ if (!response.ok) {
+ throw new Error(data.detail || `HTTP ${response.status}`);
+ }
+ const count = Number(data.count || 0);
+ setCountFeedback({
+ type: "success",
+ text: `Approximately ${count.toLocaleString()} Jira tickets match this scope.`,
+ });
+ } catch (error: any) {
+ setCountFeedback({
+ type: "error",
+ text: `Failed to preview ticket count: ${error.message}`,
+ });
+ } finally {
+ setBusy("");
+ }
+ };
+
+ const installSchema = async () => {
+ if (!selectedGraph) return;
+ setBusy("schema");
+ setSchemaFeedback({
+ type: "pending",
+ text: "Installing Jira schema…",
+ });
+ try {
+ const response = await fetch(
+ `/ui/${selectedGraph}/data-sources/jira/schema/install`,
+ {
+ method: "POST",
+ headers: { Authorization: sessionStorage.getItem("auth")! },
+ }
+ );
+ const data = await safeJson(response);
+ if (!response.ok) {
+ throw new Error(
+ errorDetail(data, `Schema installation failed (${response.status})`)
+ );
+ }
+ setSchema(data.schema);
+ setSchemaFeedback({
+ type: "success",
+ text: "Jira schema installed.",
+ });
+ } catch (error: any) {
+ setSchemaFeedback({
+ type: "error",
+ text: `Schema installation failed: ${error.message}`,
+ });
+ } finally {
+ setBusy("");
+ }
+ };
+
+ const pollSync = async (sourceId: string, runId: string) => {
+ for (;;) {
+ await new Promise((resolve) => setTimeout(resolve, 3000));
+ const response = await fetch(
+ `/ui/${selectedGraph}/data-sources/${encodeURIComponent(sourceId)}/sync/${runId}`,
+ { headers: { Authorization: sessionStorage.getItem("auth")! } }
+ );
+ const data = await safeJson(response);
+ if (!response.ok) {
+ throw new Error(
+ errorDetail(data, `Sync status failed (${response.status})`)
+ );
+ }
+ pingIdleTimer();
+ if (data.status === "failed") {
+ throw new Error(data.error || "Jira synchronization failed");
+ }
+ if (data.status === "completed") return data.result;
+ }
+ };
+
+ const sync = async (source: JiraSource) => {
+ setBusy(`sync:${source.id}`);
+ setSources((current) =>
+ current.map((item) =>
+ item.id === source.id
+ ? { ...item, sync: { ...item.sync, last_error: null } }
+ : item
+ )
+ );
+ setSyncFeedback({
+ type: "pending",
+ text: `Ingesting ${source.display_name} into the graph…`,
+ });
+ pauseIdleTimer();
+ try {
+ const response = await fetch(
+ `/ui/${selectedGraph}/data-sources/${encodeURIComponent(source.id)}/sync`,
+ {
+ method: "POST",
+ headers: { Authorization: sessionStorage.getItem("auth")! },
+ }
+ );
+ const data = await safeJson(response);
+ if (!response.ok) {
+ throw new Error(
+ errorDetail(data, `Jira synchronization failed (${response.status})`)
+ );
+ }
+ const result = await pollSync(source.id, data.run_id);
+ let buildStarted = false;
+ const missingChunkEmbeddings = Number(
+ result.missing_chunk_embeddings || 0
+ );
+ if (result.rebuild_required) {
+ setSyncFeedback({
+ type: "pending",
+ text:
+ missingChunkEmbeddings > 0 && result.documents_loaded === 0
+ ? `No Jira changes detected, but ${missingChunkEmbeddings} graph chunks are missing embeddings. Starting a recovery build…`
+ : "Jira ingestion complete. Starting the GraphRAG build…",
+ });
+ const rebuild = await fetch(`/ui/${selectedGraph}/rebuild_graph`, {
+ method: "POST",
+ headers: { Authorization: sessionStorage.getItem("auth")! },
+ });
+ const rebuildData = await safeJson(rebuild);
+ if (!rebuild.ok) {
+ throw new Error(
+ errorDetail(
+ rebuildData,
+ "Jira data was ingested, but the GraphRAG build failed to start."
+ )
+ );
+ }
+ buildStarted = true;
+
+ // Poll rebuild_status and show live ECC progress on the Data Sources
+ // page, identical to the progress bar shown on the KGAdmin page.
+ const creds = sessionStorage.getItem("auth")!;
+ const baseMsg =
+ `Jira ingestion complete: ${result.issues_upserted} issues updated, ` +
+ `${result.issues_deleted || 0} removed, and ` +
+ `${result.documents_loaded} changed documents loaded. `;
+ let pollDone = false;
+ while (!pollDone) {
+ await new Promise((r) => setTimeout(r, 3000));
+ try {
+ const statusResp = await fetch(
+ `/ui/${selectedGraph}/rebuild_status`,
+ { headers: { Authorization: creds } }
+ );
+ if (!statusResp.ok) break;
+ const statusData = await statusResp.json();
+ if (statusData.is_running) {
+ const stage = statusData.stage ? ` — ${statusData.stage}` : " — Building…";
+ setSyncFeedback({
+ type: "pending",
+ text: baseMsg + `GraphRAG build in progress${stage}`,
+ });
+ } else {
+ pollDone = true;
+ }
+ } catch {
+ break;
+ }
+ }
+ }
+ await loadSources();
+ setSyncFeedback({
+ type: "success",
+ text:
+ `Jira ingestion complete: ${result.issues_upserted} issues updated, ` +
+ `${result.issues_deleted || 0} removed, and ` +
+ `${result.documents_loaded} changed documents loaded.` +
+ (buildStarted
+ ? missingChunkEmbeddings > 0
+ ? ` GraphRAG recovery build complete.`
+ : " GraphRAG build for chunking and embedding complete."
+ : " No Jira changes were detected, so no new build was started."),
+ });
+ } catch (error: any) {
+ setSyncFeedback({
+ type: "error",
+ text: `Jira ingestion failed: ${error.message}`,
+ });
+ } finally {
+ resumeIdleTimer();
+ setBusy("");
+ }
+ };
+
+ const remove = async (source: JiraSource) => {
+ if (!window.confirm(`Remove data source "${source.display_name}"?`)) return;
+ setBusy(`delete:${source.id}`);
+ setSourceFeedback({
+ type: "pending",
+ text: `Removing ${source.display_name}…`,
+ });
+ try {
+ const response = await fetch(
+ `/ui/${selectedGraph}/data-sources/${encodeURIComponent(source.id)}`,
+ {
+ method: "DELETE",
+ headers: { Authorization: sessionStorage.getItem("auth")! },
+ }
+ );
+ const data = await safeJson(response);
+ if (!response.ok) throw new Error(data.detail || `HTTP ${response.status}`);
+ if (editing?.id === source.id) {
+ setEditing(null);
+ sessionStorage.removeItem(editingSourceKey(selectedGraph));
+ }
+ await loadSources();
+ setSourceFeedback({
+ type: "success",
+ text: `${source.display_name} was removed.`,
+ });
+ } catch (error: any) {
+ setSourceFeedback({
+ type: "error",
+ text: `Delete failed: ${error.message}`,
+ });
+ } finally {
+ setBusy("");
+ }
+ };
+
+ const normalizedProjectSearch = projectSearch.trim().toLowerCase();
+ const filteredProjects = normalizedProjectSearch
+ ? projects.filter(
+ (project) =>
+ project.key.toLowerCase().includes(normalizedProjectSearch) ||
+ project.name.toLowerCase().includes(normalizedProjectSearch)
+ )
+ : projects;
+
+ return (
+
+
+
+
+
+
+ Data sources
+
+
+ Connect, configure, ingest, and monitor external sources for the
+ selected knowledge graph.
+
+
+
+
+
{}}
+ onGraphChange={(graph) => {
+ setSelectedGraph(graph);
+ sessionStorage.setItem("selectedGraph", graph);
+ window.dispatchEvent(new Event("graphrag:selectedGraph"));
+ }}
+ />
+
+ {message && (
+
+ {message}
+
+ )}
+
+ {selectedGraph && (
+
+
+
+
+
+
+
+ Jira Cloud
+
+
+ Connect Jira, select projects, install its graph schema, then
+ ingest issues and build GraphRAG.
+
+
+
+
{
+ setEditing(emptySource());
+ setConnectionVerified(false);
+ setProjectSelectionLoaded(false);
+ setProjects([]);
+ setDirectProjectKey("");
+ setProjectSearch("");
+ setConnectionFeedback(null);
+ setScopeFeedback(null);
+ setSaveFeedback(null);
+ setCountFeedback(null);
+ }}
+ disabled={!!busy || !!editing}
+ >
+
+
+
+
+ Until per-user Jira ACL enforcement is enabled, graph users can
+ retrieve every selected-project issue visible to the connector account.
+
+
+
+ {sourceFeedback && (
+
+ )}
+
+
+
+ 1. Connection and scope
+
+
+ {loading ? "Loading…" : `${sources.length} configured source${sources.length === 1 ? "" : "s"}`}
+
+
+
+ {sources.length === 0 && !loading && (
+
+ No Jira sources configured for this graph.
+
+ )}
+
+ {sources.map((source) => (
+
+
+
+ {source.display_name}
+
+
+ {source.connection.site_url}
+ {source.scope.project_keys.length > 0 &&
+ ` · ${source.scope.project_keys.join(", ")}`}
+
+
+
+ {source.sync.last_tested_at
+ ? "Connection tested"
+ : "Connection not tested"}
+
+
+ {source.sync.last_error && (
+
{source.sync.last_error}
+ )}
+
+
+ {
+ setEditing(source);
+ setConnectionVerified(false);
+ setProjectSelectionLoaded(false);
+ setProjects([]);
+ setDirectProjectKey("");
+ setProjectSearch("");
+ setConnectionFeedback(null);
+ setScopeFeedback(null);
+ setSaveFeedback(null);
+ setCountFeedback(null);
+ sessionStorage.setItem(
+ editingSourceKey(selectedGraph),
+ source.id
+ );
+ }}
+ disabled={!!busy}
+ >
+ Edit
+
+ remove(source)}
+ disabled={!!busy}
+ >
+
+
+
+ ))}
+
+
+
+ {editing && (
+
+
+ Connection details
+
+
+
+ Source id
+
+
+ Display name
+
+
+
Site URL
+
+ patchConnectionDetails({
+ connection: {
+ ...editing.connection,
+ site_url: event.target.value,
+ },
+ })
+ }
+ className={inputClass}
+ placeholder="https://your-company.atlassian.net"
+ />
+
+ Jira tenant base URL—not graphql.atlassian.net or an issue URL.
+
+
+
+
Atlassian account email
+
+ patchConnectionDetails({
+ connection: {
+ ...editing.connection,
+ email: event.target.value,
+ },
+ })
+ }
+ className={inputClass}
+ />
+
+ Email address associated with the API token.
+
+
+
+ API token
+
+
+
Project key (optional)
+
{
+ setDirectProjectKey(event.target.value);
+ setProjectSelectionLoaded(false);
+ setProjects([]);
+ setProjectSearch("");
+ setConnectionFeedback(null);
+ setScopeFeedback(null);
+ setSaveFeedback(null);
+ setCountFeedback(null);
+ }}
+ className={inputClass}
+ placeholder="GML"
+ />
+
+ Enter a key to load only that Jira project, or leave this
+ blank to load every project visible to the account.
+
+
+
+
+ {busy === "test" && (
+
+
+ {connectionFeedback && (
+
+ )}
+ {connectionVerified && (
+ <>
+
+
+ Projects to ingest
+
+
+ Load the requested Jira project or all projects visible
+ to this account, then select the projects for this graph.
+
+
+
+
+ {busy === "projects" && (
+
+
+
+ )}
+
+
+ {projects.length > 0 && (
+
+
+ {directProjectKey.trim()
+ ? "Jira project"
+ : "Visible Jira projects"}
+
+ {!directProjectKey.trim() && (
+
setProjectSearch(event.target.value)}
+ className={`${inputClass} mb-2`}
+ placeholder="Search by project key or name"
+ aria-label="Search visible Jira projects"
+ />
+ )}
+
+ {filteredProjects.map((project) => (
+
+
+ ))}
+ {filteredProjects.length === 0 && (
+
+ No projects match this search.
+
+ )}
+
+
+ )}
+
+ {connectionVerified && (
+
+ )}
+
+ {connectionVerified &&
+ projectSelectionLoaded &&
+ projects.length > 0 && (
+
+
+ Ticket filters
+
+
+ Optionally limit which tickets are included from the
+ selected projects.
+
+
+
+
+ Created on or after
+
+
+
+
+ Updated on or after
+
+
+
+
Status categories
+
+ {jiraStatusCategories.map((category) => {
+ const selected =
+ editing.scope.status_categories || [];
+ return (
+
+
+ );
+ })}
+
+
+ Leave all categories unselected to include every status.
+
+
+
+
+
+
+ Advanced scope options
+
+
+
+ Additional JQL filter
+
+
+
+
+
+ {busy === "count" && (
+
+
+ {countFeedback && (
+
+ )}
+
+
+ )}
+
+ {connectionVerified && projectSelectionLoaded && saveFeedback && (
+
+ )}
+
+
+ {
+ setEditing(null);
+ setConnectionVerified(false);
+ setProjectSelectionLoaded(false);
+ setProjects([]);
+ setDirectProjectKey("");
+ setProjectSearch("");
+ setConnectionFeedback(null);
+ setScopeFeedback(null);
+ setSaveFeedback(null);
+ setCountFeedback(null);
+ sessionStorage.setItem(
+ editingSourceKey(selectedGraph),
+ "__closed__"
+ );
+ }}
+ disabled={!!busy}
+ >
+ Cancel
+
+ {connectionVerified && projectSelectionLoaded && (
+
+ {busy === "save" ? (
+
+ )}
+
+
+ )}
+
+ {sources.length > 0 && (
+
+
+
+
+
+
+ {schema?.status === "installed"
+ ? "Installed"
+ : schema?.status === "not_initialized"
+ ? "Initialize this graph before installing the Jira schema."
+ : schema?.status === "conflict"
+ ? "Existing graph schema conflicts with the Jira connector."
+ : schema?.status === "incomplete"
+ ? "The Jira schema is incomplete."
+ : sources.some(
+ (source) =>
+ source.sync.last_tested_at &&
+ source.scope.project_keys.length > 0
+ )
+ ? "Ready to install."
+ : sources.some(
+ (source) => source.sync.last_tested_at
+ )
+ ? "Select and save at least one project first."
+ : "Test a saved connection before installing the schema."}
+
+
+
+ {schema &&
+ schema.status !== "installed" &&
+ schema.status !== "conflict" &&
+ schema.status !== "not_initialized" && (
+
+ source.sync.last_tested_at &&
+ source.scope.project_keys.length > 0
+ )
+ }
+ >
+ {busy === "schema" && (
+
+ )}
+
+
+ {(schema?.conflicts?.length || 0) > 0 && (
+
+ {schema?.conflicts.map((conflict) => (
+ {conflict}
+ ))}
+
+ )}
+ {schemaFeedback && (
+
+ )}
+
+ )}
+
+ {sources.length > 0 && schema?.status === "installed" && (
+
+
+ 3. Ingest and build
+
+
+ Import Jira issues into this graph. The existing GraphRAG loading
+ pipeline creates chunks and embeddings after structural ingestion.
+
+ {syncFeedback && (
+
+ )}
+
+ {sources.map((source) => (
+
+
+
+ {source.display_name}
+
+
+ Last ingestion: {source.sync.last_completed_at || "Never"} ·
+ Issues: {source.sync.last_issue_count || 0}
+
+ {source.sync.last_error &&
+ busy !== `sync:${source.id}` && (
+
+ {source.sync.last_error}
+
+ )}
+
+
sync(source)}
+ disabled={
+ !!busy ||
+ !source.enabled ||
+ !source.sync.last_tested_at ||
+ source.scope.project_keys.length === 0
+ }
+ >
+ {busy === `sync:${source.id}` && (
+
+
+ ))}
+
+
+
+ )}
+
diff --git a/graphrag-ui/src/types/dataSources.ts b/graphrag-ui/src/types/dataSources.ts
new file mode 100644
index 00000000..eb87568f
--- /dev/null
+++ b/graphrag-ui/src/types/dataSources.ts
@@ -0,0 +1,41 @@
+export interface JiraSource {
+ id: string;
+ type: "jira_cloud";
+ enabled: boolean;
+ display_name: string;
+ connection: {
+ site_url: string;
+ email: string;
+ api_token: string;
+ cloud_id?: string | null;
+ };
+ scope: {
+ project_keys: string[];
+ created_after?: string | null;
+ updated_after?: string | null;
+ status_categories?: Array<"new" | "indeterminate" | "done">;
+ jql_extra: string;
+ include_comments: boolean;
+ story_points_field?: string | null;
+ };
+ sync: {
+ overlap_seconds: number;
+ checkpoint?: string | null;
+ last_tested_at?: string | null;
+ last_started_at?: string | null;
+ last_completed_at?: string | null;
+ last_error?: string | null;
+ last_issue_count: number;
+ };
+}
+
+export interface JiraSchemaStatus {
+ status:
+ | "not_initialized"
+ | "not_installed"
+ | "incomplete"
+ | "conflict"
+ | "installed";
+ missing: Record;
+ conflicts: string[];
+}
diff --git a/graphrag/app/connectors/__init__.py b/graphrag/app/connectors/__init__.py
new file mode 100644
index 00000000..a1cb1b33
--- /dev/null
+++ b/graphrag/app/connectors/__init__.py
@@ -0,0 +1 @@
+"""Enterprise data-source connectors."""
diff --git a/graphrag/app/connectors/jira/__init__.py b/graphrag/app/connectors/jira/__init__.py
new file mode 100644
index 00000000..5a5ef0bb
--- /dev/null
+++ b/graphrag/app/connectors/jira/__init__.py
@@ -0,0 +1,5 @@
+"""Jira Cloud connector."""
+
+from .config import JiraDataSource
+
+__all__ = ["JiraDataSource"]
diff --git a/graphrag/app/connectors/jira/adf.py b/graphrag/app/connectors/jira/adf.py
new file mode 100644
index 00000000..5e1ad67d
--- /dev/null
+++ b/graphrag/app/connectors/jira/adf.py
@@ -0,0 +1,122 @@
+"""Convert Atlassian Document Format (ADF) to retrieval-friendly markdown."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def _escape_inline(value: str) -> str:
+ return value.replace("\\", "\\\\").replace("*", "\\*").replace("_", "\\_")
+
+
+def _with_marks(text: str, marks: list[dict[str, Any]]) -> str:
+ for mark in marks:
+ mark_type = mark.get("type")
+ attrs = mark.get("attrs") or {}
+ if mark_type == "code":
+ escaped = text.replace("`", "\\`")
+ text = f"`{escaped}`"
+ elif mark_type == "strong":
+ text = f"**{text}**"
+ elif mark_type == "em":
+ text = f"*{text}*"
+ elif mark_type == "strike":
+ text = f"~~{text}~~"
+ elif mark_type == "link" and attrs.get("href"):
+ text = f"[{text}]({attrs['href']})"
+ return text
+
+
+def _inline(node: dict[str, Any]) -> str:
+ node_type = node.get("type")
+ attrs = node.get("attrs") or {}
+ if node_type == "text":
+ text = _escape_inline(str(node.get("text") or ""))
+ return _with_marks(text, node.get("marks") or [])
+ if node_type == "hardBreak":
+ return " \n"
+ if node_type == "mention":
+ return f"@{attrs.get('text') or attrs.get('displayName') or 'user'}"
+ if node_type == "emoji":
+ return str(attrs.get("text") or attrs.get("shortName") or "")
+ if node_type == "inlineCard":
+ url = str(attrs.get("url") or "")
+ return f"[{url}]({url})" if url else ""
+ return "".join(_inline(child) for child in node.get("content") or [])
+
+
+def _block(node: dict[str, Any], depth: int = 0) -> str:
+ node_type = node.get("type")
+ attrs = node.get("attrs") or {}
+ children = node.get("content") or []
+
+ if node_type == "doc":
+ return "\n\n".join(
+ value for child in children if (value := _block(child, depth)).strip()
+ )
+ if node_type == "paragraph":
+ return "".join(_inline(child) for child in children).strip()
+ if node_type == "heading":
+ level = min(max(int(attrs.get("level") or 1), 1), 6)
+ return f"{'#' * level} {''.join(_inline(c) for c in children).strip()}"
+ if node_type == "blockquote":
+ content = "\n".join(_block(child, depth) for child in children).strip()
+ return "\n".join(f"> {line}" for line in content.splitlines())
+ if node_type == "codeBlock":
+ language = attrs.get("language") or ""
+ content = "".join(_inline(child) for child in children)
+ return f"```{language}\n{content}\n```"
+ if node_type == "rule":
+ return "---"
+ if node_type in ("bulletList", "orderedList"):
+ ordered = node_type == "orderedList"
+ start = int(attrs.get("order") or 1)
+ lines: list[str] = []
+ for index, child in enumerate(children):
+ value = _block(child, depth + 1).strip()
+ if not value:
+ continue
+ prefix = f"{start + index}. " if ordered else "- "
+ indentation = " " * depth
+ continuation = "\n".join(
+ f"{indentation} {line}" for line in value.splitlines()[1:]
+ )
+ first = f"{indentation}{prefix}{value.splitlines()[0]}"
+ lines.append(f"{first}\n{continuation}".rstrip())
+ return "\n".join(lines)
+ if node_type == "listItem":
+ return "\n".join(
+ value for child in children if (value := _block(child, depth)).strip()
+ )
+ if node_type in ("table", "tableRow", "tableCell", "tableHeader"):
+ # ADF tables can contain arbitrary blocks. Tabs preserve cell
+ # boundaries for embedding without pretending to provide full GFM.
+ separator = "\n" if node_type in ("table", "tableRow") else " "
+ values = [_block(child, depth).strip() for child in children]
+ values = [value for value in values if value]
+ if node_type == "tableRow":
+ separator = " | "
+ return separator.join(values)
+ if node_type in ("panel", "expand", "nestedExpand"):
+ title = str(attrs.get("title") or "").strip()
+ body = "\n\n".join(
+ value for child in children if (value := _block(child, depth)).strip()
+ )
+ return f"**{title}**\n\n{body}".strip() if title else body
+ if node_type == "mediaSingle":
+ return "\n".join(_block(child, depth) for child in children).strip()
+ if node_type == "media":
+ name = attrs.get("alt") or attrs.get("id") or "attachment"
+ return f"[Attachment: {name}]"
+ return "".join(_inline(child) for child in children).strip()
+
+
+def adf_to_markdown(value: Any) -> str:
+ """Return markdown for an ADF document, or a safe string fallback."""
+ if value is None:
+ return ""
+ if isinstance(value, str):
+ return value.strip()
+ if not isinstance(value, dict):
+ return str(value).strip()
+ return _block(value).strip()
diff --git a/graphrag/app/connectors/jira/client.py b/graphrag/app/connectors/jira/client.py
new file mode 100644
index 00000000..7cae672a
--- /dev/null
+++ b/graphrag/app/connectors/jira/client.py
@@ -0,0 +1,442 @@
+"""Small, testable Jira Cloud REST API client."""
+
+from __future__ import annotations
+
+import concurrent.futures
+import logging
+import time
+from collections.abc import Iterator
+from datetime import datetime, timedelta, timezone
+from email.utils import parsedate_to_datetime
+from typing import Any, Callable
+from urllib.parse import quote
+
+import httpx
+
+from .config import JiraDataSource
+
+logger = logging.getLogger(__name__)
+
+
+ISSUE_FIELDS = [
+ "summary",
+ "status",
+ "issuetype",
+ "priority",
+ "resolution",
+ "project",
+ "assignee",
+ "reporter",
+ "created",
+ "updated",
+ "duedate",
+ "labels",
+ "components",
+ "fixVersions",
+ "parent",
+ "issuelinks",
+ "comment",
+ "description",
+ "attachment",
+]
+STATUS_CATEGORY_JQL = {
+ "new": "To Do",
+ "indeterminate": "In Progress",
+ "done": "Done",
+}
+
+# Number of threads used to complete comment pagination within a single page.
+# Each thread issues its own HTTP request, so raising this reduces wall-clock
+# time proportionally up to the Jira rate-limit ceiling (~100 req/min on free
+# plans, higher on paid). 10 is a safe default that won't trigger throttling
+# on typical Atlassian Cloud accounts.
+COMMENT_FETCH_WORKERS = 10
+
+
+class JiraAPIError(RuntimeError):
+ def __init__(self, status_code: int, message: str):
+ super().__init__(message)
+ self.status_code = status_code
+
+
+class JiraCloudClient:
+ def __init__(
+ self,
+ source: JiraDataSource,
+ *,
+ client: httpx.Client | None = None,
+ sleep: Callable[[float], None] = time.sleep,
+ max_attempts: int = 4,
+ ):
+ self.source = source
+ self._sleep = sleep
+ self._max_attempts = max_attempts
+ self._owns_client = client is None
+ self._client = client or httpx.Client(
+ base_url=source.connection.site_url,
+ auth=(source.connection.email, source.connection.api_token),
+ headers={"Accept": "application/json"},
+ timeout=httpx.Timeout(30.0, connect=10.0),
+ )
+
+ def close(self) -> None:
+ if self._owns_client:
+ self._client.close()
+
+ def __enter__(self) -> "JiraCloudClient":
+ return self
+
+ def __exit__(self, *_args) -> None:
+ self.close()
+
+ @staticmethod
+ def _retry_after(response: httpx.Response, attempt: int) -> float:
+ value = response.headers.get("Retry-After")
+ if value:
+ try:
+ return min(max(float(value), 0.0), 60.0)
+ except ValueError:
+ try:
+ retry_at = parsedate_to_datetime(value)
+ if retry_at.tzinfo is None:
+ retry_at = retry_at.replace(tzinfo=timezone.utc)
+ return min(
+ max((retry_at - datetime.now(timezone.utc)).total_seconds(), 0),
+ 60.0,
+ )
+ except (TypeError, ValueError):
+ pass
+ return min(2 ** attempt, 30)
+
+ @staticmethod
+ def _error_message(response: httpx.Response) -> str:
+ try:
+ payload = response.json()
+ except ValueError:
+ return f"Jira returned HTTP {response.status_code}"
+ if isinstance(payload, dict):
+ messages = payload.get("errorMessages")
+ if isinstance(messages, list) and messages:
+ return "; ".join(str(message) for message in messages)
+ errors = payload.get("errors")
+ if isinstance(errors, dict) and errors:
+ return "; ".join(f"{key}: {value}" for key, value in errors.items())
+ if payload.get("message"):
+ return str(payload["message"])
+ return f"Jira returned HTTP {response.status_code}"
+
+ def _request(self, method: str, path: str, **kwargs) -> Any:
+ last_error: Exception | None = None
+ for attempt in range(self._max_attempts):
+ try:
+ response = self._client.request(method, path, **kwargs)
+ except (httpx.TimeoutException, httpx.TransportError) as exc:
+ last_error = exc
+ if attempt + 1 == self._max_attempts:
+ break
+ self._sleep(min(2 ** attempt, 30))
+ continue
+
+ if response.status_code == 429 or response.status_code >= 500:
+ if attempt + 1 < self._max_attempts:
+ self._sleep(self._retry_after(response, attempt))
+ continue
+ if response.is_error:
+ raise JiraAPIError(
+ response.status_code,
+ self._error_message(response),
+ )
+ if response.status_code == 204:
+ return None
+ try:
+ return response.json()
+ except ValueError as exc:
+ raise JiraAPIError(
+ response.status_code,
+ "Jira returned an invalid JSON response",
+ ) from exc
+ raise JiraAPIError(503, f"Unable to reach Jira: {last_error}") from last_error
+
+ def myself(self) -> dict[str, Any]:
+ return self._request("GET", "/rest/api/3/myself")
+
+ def cloud_id(self) -> str:
+ payload = self._request("GET", "/_edge/tenant_info")
+ cloud_id = payload.get("cloudId") if isinstance(payload, dict) else None
+ if not cloud_id:
+ raise JiraAPIError(502, "Jira did not return a cloudId")
+ return str(cloud_id)
+
+ def projects(self) -> list[dict[str, Any]]:
+ start_at = 0
+ projects: list[dict[str, Any]] = []
+ while True:
+ payload = self._request(
+ "GET",
+ "/rest/api/3/project/search",
+ params={"startAt": start_at, "maxResults": 100, "orderBy": "key"},
+ )
+ values = payload.get("values") or []
+ projects.extend(values)
+ if payload.get("isLast", True) or not values:
+ break
+ start_at += len(values)
+ return projects
+
+ def project(self, project_key: str) -> dict[str, Any]:
+ """Return one project by key without enumerating every visible project."""
+ return self._request(
+ "GET",
+ f"/rest/api/3/project/{quote(project_key, safe='')}",
+ )
+
+ def _jql(
+ self,
+ *,
+ incremental: bool = True,
+ include_order: bool = True,
+ ) -> str:
+ scope = self.source.scope
+ projects = ", ".join(scope.project_keys)
+ clauses = [f"project in ({projects})"]
+ if scope.created_after:
+ clauses.append(f'created >= "{scope.created_after.isoformat()}"')
+ if scope.updated_after:
+ clauses.append(f'updated >= "{scope.updated_after.isoformat()}"')
+ if scope.status_categories:
+ categories = ", ".join(
+ f'"{STATUS_CATEGORY_JQL[category]}"'
+ for category in scope.status_categories
+ )
+ clauses.append(f"statusCategory in ({categories})")
+ if scope.jql_extra:
+ clauses.append(f"({scope.jql_extra})")
+ if incremental and self.source.sync.checkpoint:
+ checkpoint = self.source.sync.checkpoint
+ if checkpoint.tzinfo is None:
+ checkpoint = checkpoint.replace(tzinfo=timezone.utc)
+ checkpoint = checkpoint.astimezone(timezone.utc) - timedelta(
+ seconds=self.source.sync.overlap_seconds
+ )
+ clauses.append(f'updated >= "{checkpoint:%Y-%m-%d %H:%M}"')
+ jql = " AND ".join(clauses)
+ return f"{jql} ORDER BY updated ASC, key ASC" if include_order else jql
+
+ def approximate_issue_count(self) -> int:
+ payload = self._request(
+ "POST",
+ "/rest/api/3/search/approximate-count",
+ json={
+ "jql": self._jql(
+ incremental=False,
+ include_order=False,
+ )
+ },
+ headers={"Content-Type": "application/json"},
+ )
+ return int(payload.get("count") or 0)
+
+ def _iter_search_pages(
+ self,
+ *,
+ fields: list[str],
+ incremental: bool,
+ ) -> Iterator[list[dict[str, Any]]]:
+ token: str | None = None
+ jql = self._jql(incremental=incremental)
+ while True:
+ body: dict[str, Any] = {
+ "jql": jql,
+ "fields": fields,
+ "fieldsByKeys": False,
+ "maxResults": 100,
+ }
+ if token:
+ body["nextPageToken"] = token
+ payload = self._request(
+ "POST",
+ "/rest/api/3/search/jql",
+ json=body,
+ headers={"Content-Type": "application/json"},
+ )
+ issues = payload.get("issues") or []
+ if issues:
+ yield issues
+ token = payload.get("nextPageToken")
+ if not token:
+ break
+
+ def _iter_search(self, *, fields: list[str], incremental: bool):
+ for issues in self._iter_search_pages(
+ fields=fields,
+ incremental=incremental,
+ ):
+ yield from issues
+
+ def iter_issue_pages(self) -> Iterator[list[dict[str, Any]]]:
+ """Yield complete issue pages for durable page-level synchronization.
+
+ Comment completion and changelog fetching are parallelised / batched
+ across issues within each page so that extra Jira API calls run
+ concurrently rather than one at a time.
+ """
+ fields = list(ISSUE_FIELDS)
+ story_points = self.source.scope.story_points_field
+ if story_points:
+ fields.append(story_points)
+ for issues in self._iter_search_pages(fields=fields, incremental=True):
+ if self.source.scope.include_comments:
+ workers = min(len(issues), COMMENT_FETCH_WORKERS)
+ with concurrent.futures.ThreadPoolExecutor(
+ max_workers=workers, thread_name_prefix="jira-comment"
+ ) as pool:
+ futures = [
+ pool.submit(self._complete_comments, issue)
+ for issue in issues
+ ]
+ for fut in concurrent.futures.as_completed(futures):
+ fut.result()
+ # Bulk-fetch changelogs for all issues in this page in one POST.
+ self._bulk_fetch_changelogs(issues)
+ yield issues
+
+ def iter_issues_by_keys(self, keys: set[str]) -> Iterator[list[dict[str, Any]]]:
+ """Fetch specific Jira issues by key without a full incremental scan.
+
+ Used for recovery: when a previous sync wrote structural vertices but
+ failed before embedding, this fetches only the failed issues by key
+ (JQL ``key in (...)``), avoiding a full re-scan of all Jira issues.
+ Yields pages of up to 100 issues with comments completed, identical in
+ structure to ``iter_issue_pages()``.
+ """
+ if not keys:
+ return
+ fields = list(ISSUE_FIELDS)
+ story_points = self.source.scope.story_points_field
+ if story_points:
+ fields.append(story_points)
+ keys_sorted = sorted(keys)
+ for i in range(0, len(keys_sorted), 100):
+ batch = keys_sorted[i : i + 100]
+ jql = "key in (" + ", ".join(batch) + ") ORDER BY updated ASC, key ASC"
+ payload = self._request(
+ "POST",
+ "/rest/api/3/search/jql",
+ json={
+ "jql": jql,
+ "fields": fields,
+ "fieldsByKeys": False,
+ "maxResults": 100,
+ },
+ headers={"Content-Type": "application/json"},
+ )
+ issues = payload.get("issues") or []
+ if not issues:
+ continue
+ if self.source.scope.include_comments:
+ workers = min(len(issues), COMMENT_FETCH_WORKERS)
+ with concurrent.futures.ThreadPoolExecutor(
+ max_workers=workers, thread_name_prefix="jira-recovery"
+ ) as pool:
+ futures = [
+ pool.submit(self._complete_comments, issue)
+ for issue in issues
+ ]
+ for fut in concurrent.futures.as_completed(futures):
+ fut.result()
+ # Bulk-fetch changelogs for all issues in this page in one POST.
+ self._bulk_fetch_changelogs(issues)
+ yield issues
+
+ def iter_issues(self) -> Iterator[dict[str, Any]]:
+ for issues in self.iter_issue_pages():
+ yield from issues
+
+ def _complete_comments(self, issue: dict[str, Any]) -> None:
+ fields = issue.setdefault("fields", {})
+ page = fields.get("comment") or {}
+ comments = list(page.get("comments") or [])
+ total = int(page.get("total") or len(comments))
+ if len(comments) >= total:
+ page["comments"] = comments
+ fields["comment"] = page
+ return
+
+ issue_id = issue.get("id") or issue.get("key")
+ start_at = len(comments)
+ while start_at < total:
+ payload = self._request(
+ "GET",
+ f"/rest/api/3/issue/{issue_id}/comment",
+ params={"startAt": start_at, "maxResults": 100, "orderBy": "created"},
+ )
+ batch = payload.get("comments") or []
+ if not batch:
+ break
+ comments.extend(batch)
+ start_at += len(batch)
+ total = int(payload.get("total") or total)
+ page["comments"] = comments
+ page["total"] = total
+ fields["comment"] = page
+
+ def _bulk_fetch_changelogs(self, issues: list[dict[str, Any]]) -> None:
+ """Bulk-fetch changelogs for up to 1,000 issues in a single POST.
+
+ Uses POST /rest/api/3/changelog/bulkfetch which returns changelogs for
+ all requested issues in one round-trip instead of N individual GETs.
+ Paginates via nextPageToken if the response is truncated.
+ Results are attached at issue["changelog"]["histories"] keyed by
+ Jira issue ID so the mapper can access them as issue.get("changelog").
+ On any error the method returns silently — no issue is blocked.
+ """
+ if not issues:
+ return
+
+ # Build a mapping from Jira numeric id → issue dict for fast lookup.
+ id_to_issue: dict[str, dict[str, Any]] = {
+ str(issue["id"]): issue for issue in issues if issue.get("id")
+ }
+ keys = [issue["key"] for issue in issues if issue.get("key")]
+ if not keys:
+ return
+
+ # Initialise empty changelog on every issue so mapper never sees None.
+ for issue in issues:
+ issue.setdefault("changelog", {"histories": []})
+
+ try:
+ token: str | None = None
+ while True:
+ body: dict[str, Any] = {
+ "issueIdsOrKeys": keys,
+ "maxResults": 100,
+ }
+ if token:
+ body["nextPageToken"] = token
+ payload = self._request(
+ "POST",
+ "/rest/api/3/changelog/bulkfetch",
+ json=body,
+ headers={"Content-Type": "application/json"},
+ )
+ for entry in payload.get("issueChangeLogs") or []:
+ issue_id = str(entry.get("issueId") or "")
+ issue = id_to_issue.get(issue_id)
+ if issue is None:
+ continue
+ histories = issue["changelog"].setdefault("histories", [])
+ for h in entry.get("changeHistories") or []:
+ # Normalise ms-epoch timestamp → ISO string for mapper.
+ created = h.get("created")
+ if isinstance(created, (int, float)):
+ created = datetime.fromtimestamp(
+ created / 1000, tz=timezone.utc
+ ).strftime("%Y-%m-%dT%H:%M:%S+00:00")
+ h = dict(h, created=created)
+ histories.append(h)
+ token = payload.get("nextPageToken")
+ if not token:
+ break
+ except Exception as exc:
+ logger.warning("bulk changelog fetch failed: %s — skipping", exc)
diff --git a/graphrag/app/connectors/jira/config.py b/graphrag/app/connectors/jira/config.py
new file mode 100644
index 00000000..1c5dfddd
--- /dev/null
+++ b/graphrag/app/connectors/jira/config.py
@@ -0,0 +1,138 @@
+"""Validated configuration for the Jira Cloud connector."""
+
+from __future__ import annotations
+
+import re
+from datetime import date, datetime
+from typing import Literal
+from urllib.parse import urlparse
+
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+
+SOURCE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")
+PROJECT_KEY_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,31}$")
+
+
+class JiraConnection(BaseModel):
+ site_url: str
+ email: str = Field(min_length=3)
+ api_token: str = ""
+ cloud_id: str | None = None
+
+ @field_validator("site_url")
+ @classmethod
+ def validate_site_url(cls, value: str) -> str:
+ normalized = value.strip().rstrip("/")
+ parsed = urlparse(normalized)
+ if parsed.scheme != "https" or not parsed.hostname:
+ raise ValueError("site_url must be a valid HTTPS URL")
+ if parsed.path not in ("", "/") or parsed.query or parsed.fragment:
+ raise ValueError("site_url must not include a path, query, or fragment")
+ return normalized
+
+ @field_validator("email")
+ @classmethod
+ def normalize_email(cls, value: str) -> str:
+ value = value.strip()
+ if "@" not in value:
+ raise ValueError("email must be a valid Atlassian account email")
+ return value
+
+
+class JiraScope(BaseModel):
+ project_keys: list[str] = Field(default_factory=list)
+ created_after: date | None = None
+ updated_after: date | None = None
+ status_categories: list[
+ Literal["new", "indeterminate", "done"]
+ ] = Field(default_factory=list)
+ jql_extra: str = ""
+ include_comments: bool = True
+ story_points_field: str | None = None
+
+ @field_validator("project_keys")
+ @classmethod
+ def normalize_project_keys(cls, values: list[str]) -> list[str]:
+ normalized: list[str] = []
+ seen: set[str] = set()
+ for raw in values:
+ key = raw.strip().upper()
+ if not PROJECT_KEY_RE.fullmatch(key):
+ raise ValueError(f"invalid Jira project key: {raw!r}")
+ if key not in seen:
+ normalized.append(key)
+ seen.add(key)
+ return normalized
+
+ @field_validator("status_categories")
+ @classmethod
+ def deduplicate_status_categories(
+ cls,
+ values: list[Literal["new", "indeterminate", "done"]],
+ ) -> list[Literal["new", "indeterminate", "done"]]:
+ return list(dict.fromkeys(values))
+
+ @field_validator("jql_extra")
+ @classmethod
+ def validate_jql_extra(cls, value: str) -> str:
+ value = value.strip()
+ if re.search(r"\border\s+by\b", value, flags=re.IGNORECASE):
+ raise ValueError("jql_extra must not contain ORDER BY")
+ return value
+
+ @field_validator("story_points_field")
+ @classmethod
+ def validate_story_points_field(cls, value: str | None) -> str | None:
+ if value is None or not value.strip():
+ return None
+ value = value.strip()
+ if not re.fullmatch(r"customfield_\d+", value):
+ raise ValueError("story_points_field must look like customfield_10016")
+ return value
+
+
+class JiraSyncState(BaseModel):
+ overlap_seconds: int = Field(default=120, ge=0, le=3600)
+ checkpoint: datetime | None = None
+ migrating_legacy_comments: bool = False
+ last_tested_at: datetime | None = None
+ last_started_at: datetime | None = None
+ last_completed_at: datetime | None = None
+ last_error: str | None = None
+ last_issue_count: int = Field(default=0, ge=0)
+
+
+class JiraDataSource(BaseModel):
+ id: str
+ type: Literal["jira_cloud"] = "jira_cloud"
+ enabled: bool = True
+ display_name: str = Field(min_length=1, max_length=100)
+ connection: JiraConnection
+ scope: JiraScope
+ sync: JiraSyncState = Field(default_factory=JiraSyncState)
+
+ @field_validator("id")
+ @classmethod
+ def validate_id(cls, value: str) -> str:
+ value = value.strip()
+ if not SOURCE_ID_RE.fullmatch(value):
+ raise ValueError(
+ "id must be 1-64 characters using letters, numbers, '_' or '-'"
+ )
+ return value
+
+ @field_validator("display_name")
+ @classmethod
+ def normalize_display_name(cls, value: str) -> str:
+ return value.strip()
+
+class JiraSourceFile(BaseModel):
+ sources: list[JiraDataSource] = Field(default_factory=list)
+
+ @model_validator(mode="after")
+ def unique_source_ids(self) -> "JiraSourceFile":
+ ids = [source.id for source in self.sources]
+ if len(ids) != len(set(ids)):
+ raise ValueError("data-source ids must be unique")
+ return self
diff --git a/graphrag/app/connectors/jira/mapper.py b/graphrag/app/connectors/jira/mapper.py
new file mode 100644
index 00000000..184f2614
--- /dev/null
+++ b/graphrag/app/connectors/jira/mapper.py
@@ -0,0 +1,814 @@
+"""Deterministic Jira payload to GraphRAG mapping."""
+
+from __future__ import annotations
+
+import hashlib
+import re
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any
+
+from dateutil import parser as date_parser
+
+from common.chunkers.structured import StructuredChunker
+from common.config import get_graphrag_config
+
+from .adf import adf_to_markdown
+from .config import JiraDataSource
+from .schema import (
+ JIRA_ASSIGNEE_EDGE,
+ JIRA_CHANGE_AUTHOR_EDGE,
+ JIRA_CHANGE_EDGE,
+ JIRA_COMMENT_AFTER_EDGE,
+ JIRA_COMMENT_AUTHOR_EDGE,
+ JIRA_COMMENT_ISSUE_EDGE,
+ JIRA_COMMENT_REPLY_EDGE,
+ JIRA_LINK_EDGE,
+ JIRA_PARENT_EDGE,
+ JIRA_PROJECT_EDGE,
+ JIRA_REPORTER_EDGE,
+)
+
+
+@dataclass(frozen=True)
+class VertexRecord:
+ vertex_type: str
+ vertex_id: str
+ attributes: dict[str, Any]
+
+
+@dataclass(frozen=True)
+class EdgeRecord:
+ source_type: str
+ source_id: str
+ edge_type: str
+ target_type: str
+ target_id: str
+ attributes: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class MappedChunk:
+ chunk_id: str
+ index: int
+ text: str
+
+
+@dataclass
+class MappedComment:
+ comment_id: str
+ comment_vertex_id: str
+ issue_vertex_id: str
+ content_hash: str
+ chunks: list[MappedChunk]
+
+
+@dataclass
+class MappedFact:
+ """One short searchable fact for an issue.
+
+ entities are extra CONTAINS_ENTITY targets besides the issue itself.
+ """
+
+ chunk: MappedChunk
+ entities: tuple[tuple[str, str], ...] = ()
+
+
+@dataclass
+class MappedIssue:
+ issue_id: str
+ issue_vertex_id: str
+ updated: datetime | None
+ content_hash: str
+ vertices: list[VertexRecord]
+ edges: list[EdgeRecord]
+ facts: list[MappedFact]
+ legacy_document_id: str
+ change_vertex_ids: list[str]
+ comments: list[MappedComment]
+
+
+def _datetime(value: Any) -> str | None:
+ if not value:
+ return None
+ try:
+ return date_parser.parse(str(value)).strftime("%Y-%m-%d %H:%M:%S")
+ except (TypeError, ValueError, OverflowError):
+ return None
+
+
+def _parsed_datetime(value: Any) -> datetime | None:
+ if not value:
+ return None
+ try:
+ return date_parser.parse(str(value))
+ except (TypeError, ValueError, OverflowError):
+ return None
+
+
+def _timestamp(value: Any) -> float:
+ parsed = _parsed_datetime(value)
+ return parsed.timestamp() if parsed is not None else 0.0
+
+
+def _display_name(user: dict[str, Any] | None) -> str:
+ if not user:
+ return ""
+ return str(user.get("displayName") or user.get("accountId") or "").strip()
+
+
+def _comma_names(values: list[Any] | None) -> str:
+ names: list[str] = []
+ for value in values or []:
+ if isinstance(value, dict):
+ name = value.get("name")
+ else:
+ name = value
+ if name is not None and str(name).strip():
+ names.append(str(name).strip())
+ return ", ".join(names)
+
+
+_FENCED_BLOCK_RE = re.compile(r"```[^\n]*\n(.*?)```", re.DOTALL)
+_LOG_LINE_RE = re.compile(
+ r"""(?ix)
+ ^\s*(?:
+ \[?\d{4}[-/]\d{2}[-/]\d{2}[T\s]\d{2}:\d{2}:\d{2}
+ |\[?(?:TRACE|DEBUG|INFO|WARN(?:ING)?|ERROR|FATAL|CRITICAL)\]?\b
+ |(?:Traceback\s+\(most\s+recent\s+call\s+last\)|Caused\s+by:)
+ |(?:at\s+[\w.$]+\([^)]*(?::\d+)?\))
+ |(?:File\s+"[^"]+",\s+line\s+\d+)
+ |\{.*"(?:timestamp|time|level|severity|logger)"\s*:
+ )
+ """
+)
+_LOG_OMISSION = "[Log output omitted from search content.]"
+
+
+def _is_log_heavy(lines: list[str]) -> bool:
+ non_empty = [line for line in lines if line.strip()]
+ if len(non_empty) < 5:
+ return False
+ matched = sum(bool(_LOG_LINE_RE.search(line)) for line in non_empty)
+ return matched >= 3 and matched / len(non_empty) >= 0.6
+
+
+def _filter_long_log_output(text: str, chunk_size: int) -> str:
+ """Remove log-dominated blocks only when a comment needs chunking."""
+ if len(text) <= chunk_size:
+ return text
+
+ def replace_fence(match: re.Match[str]) -> str:
+ body = match.group(1)
+ return _LOG_OMISSION if _is_log_heavy(body.splitlines()) else match.group(0)
+
+ filtered = _FENCED_BLOCK_RE.sub(replace_fence, text)
+ lines = filtered.splitlines()
+ output: list[str] = []
+ index = 0
+ while index < len(lines):
+ end = index
+ while end < len(lines) and (
+ not lines[end].strip() or _LOG_LINE_RE.search(lines[end])
+ ):
+ end += 1
+ block = lines[index:end]
+ if _is_log_heavy(block):
+ if not output or output[-1] != _LOG_OMISSION:
+ output.append(_LOG_OMISSION)
+ index = end
+ continue
+ output.append(lines[index])
+ index += 1
+ return "\n".join(output).strip()
+
+
+class JiraIssueMapper:
+ def __init__(
+ self,
+ source: JiraDataSource,
+ graphname: str | None = None,
+ ):
+ self.source = source
+ chunker_config = get_graphrag_config(graphname).get(
+ "chunker_config",
+ {},
+ )
+ self.comment_chunker = StructuredChunker(
+ chunk_size=chunker_config.get("chunk_size", 0),
+ overlap_size=chunker_config.get("overlap_size", -1),
+ )
+
+ def _id(self, object_type: str, object_id: Any) -> str:
+ # cloud_id intentionally excluded — vertex IDs use only the portable
+ # object_type + object_id so they are stable across graph recreations
+ # and don't leak internal tenant identifiers.
+ return f"jira:{object_type}:{object_id}".lower()
+
+ def _issue_id(self, issue_key: str) -> str:
+ # Use the human-readable ticket key so GenerateFunction can construct
+ # the vertex ID directly from what the user says (e.g. "GML-2191"
+ # → "jira:gml-2191:issue"). cloud_id is NOT included — ticket keys
+ # are unique within a graph's connected Jira project scope.
+ return f"jira:{issue_key}:issue".lower()
+
+ def _user_vertex(
+ self, user: dict[str, Any] | None
+ ) -> VertexRecord | None:
+ if not user or not user.get("accountId"):
+ return None
+ account_id = str(user["accountId"])
+ return VertexRecord(
+ "JiraUser",
+ self._id("user", account_id),
+ {
+ "account_id": account_id,
+ "display_name": _display_name(user),
+ },
+ )
+
+ def map(self, issue: dict[str, Any]) -> MappedIssue:
+ issue_id = str(issue["id"])
+ issue_key = str(issue.get("key") or issue_id)
+ fields = issue.get("fields") or {}
+ project = fields.get("project") or {}
+ project_id = str(project.get("id") or project.get("key") or "unknown")
+ issue_vertex_id = self._issue_id(issue_key)
+ project_vertex_id = self._id("project", project_id)
+ site_url = self.source.connection.site_url
+ issue_url = f"{site_url}/browse/{issue_key}"
+
+ description = adf_to_markdown(fields.get("description"))
+
+ status = fields.get("status") or {}
+ status_category = status.get("statusCategory") or {}
+ priority = fields.get("priority") or {}
+ resolution = fields.get("resolution") or {}
+ issue_type = fields.get("issuetype") or {}
+ story_points_field = self.source.scope.story_points_field
+ story_points = fields.get(story_points_field) if story_points_field else None
+
+ issue_attrs = {
+ "issue_key": issue_key,
+ "summary": str(fields.get("summary") or ""),
+ "issue_type": str(issue_type.get("name") or ""),
+ "status": str(status.get("name") or ""),
+ "status_category": str(status_category.get("key") or ""),
+ "priority": str(priority.get("name") or ""),
+ "resolution": str(resolution.get("name") or ""),
+ "labels": _comma_names(fields.get("labels")),
+ "components": _comma_names(fields.get("components")),
+ "fix_versions": _comma_names(fields.get("fixVersions")),
+ "created": _datetime(fields.get("created")),
+ "updated": _datetime(fields.get("updated")),
+ "due": _datetime(fields.get("duedate")),
+ "url": issue_url,
+ }
+ if story_points is not None:
+ try:
+ issue_attrs["story_points"] = float(story_points)
+ except (TypeError, ValueError):
+ pass
+ issue_attrs = {
+ key: value for key, value in issue_attrs.items() if value not in (None, "")
+ }
+
+ vertices: dict[tuple[str, str], VertexRecord] = {}
+
+ def add_vertex(vertex: VertexRecord | None) -> None:
+ if not vertex:
+ return
+ key = (vertex.vertex_type, vertex.vertex_id)
+ existing = vertices.get(key)
+ if existing and existing.attributes.keys() - vertex.attributes.keys():
+ return
+ vertices[key] = vertex
+
+ add_vertex(
+ VertexRecord(
+ "JiraProject",
+ project_vertex_id,
+ {
+ "project_key": str(project.get("key") or ""),
+ "name": str(project.get("name") or ""),
+ "url": (
+ f"{site_url}/jira/software/projects/{project.get('key')}"
+ if project.get("key")
+ else site_url
+ ),
+ },
+ )
+ )
+ add_vertex(VertexRecord("JiraIssue", issue_vertex_id, issue_attrs))
+
+ edges = [
+ EdgeRecord(
+ "JiraIssue",
+ issue_vertex_id,
+ JIRA_PROJECT_EDGE,
+ "JiraProject",
+ project_vertex_id,
+ )
+ ]
+
+ for field_name, edge_type in (
+ ("assignee", JIRA_ASSIGNEE_EDGE),
+ ("reporter", JIRA_REPORTER_EDGE),
+ ):
+ user_vertex = self._user_vertex(fields.get(field_name))
+ add_vertex(user_vertex)
+ if user_vertex:
+ edges.append(
+ EdgeRecord(
+ "JiraIssue",
+ issue_vertex_id,
+ edge_type,
+ "JiraUser",
+ user_vertex.vertex_id,
+ )
+ )
+
+ mapped_comments: list[MappedComment] = []
+ previous_comment_vertex_id: str | None = None
+ comments = (
+ (fields.get("comment") or {}).get("comments") or []
+ if self.source.scope.include_comments
+ else []
+ )
+ comments = sorted(
+ comments,
+ key=lambda comment: (
+ _timestamp(comment.get("created")),
+ str(comment.get("id") or ""),
+ ),
+ )
+ for comment in comments:
+ raw_comment_id = comment.get("id")
+ if raw_comment_id is None:
+ continue
+
+ # Skip bot/automation comments — accountType "app" means a Jira
+ # automation rule, CI integration, or service-account bot. These
+ # produce high-volume noise (build status, deploy notifications,
+ # auto-transitions) with no useful search content.
+ author = comment.get("author") or {}
+ if author.get("accountType") == "app":
+ continue
+
+ # Skip empty comments — nothing meaningful to store or search.
+ comment_body = adf_to_markdown(comment.get("body"))
+ if not comment_body or not comment_body.strip():
+ continue
+
+ comment_id = str(raw_comment_id)
+ comment_vertex_id = self._id("comment", comment_id)
+ author_vertex = self._user_vertex(author)
+ add_vertex(author_vertex)
+ visibility = comment.get("visibility")
+ if not isinstance(visibility, dict):
+ visibility = {}
+ visibility_text = ":".join(
+ str(value)
+ for value in (
+ visibility.get("type"),
+ visibility.get("value"),
+ )
+ if value
+ )
+ comment_body = _filter_long_log_output(
+ comment_body,
+ self.comment_chunker.chunk_size,
+ )
+ comment_text = self._comment_document_text(
+ issue_key=issue_key,
+ comment=comment,
+ body=comment_body,
+ )
+ chunks = [
+ self._lead_with_issue(issue_key, str(chunk))
+ for chunk in self.comment_chunker.chunk(comment_text)
+ ]
+ chunks = [chunk for chunk in chunks if chunk]
+ comment_hash = hashlib.sha256(
+ (
+ "direct-comment-chunks\0"
+ + "\0".join(chunks)
+ ).encode("utf-8")
+ ).hexdigest()
+ comment_attrs = {
+ "comment_id": comment_id,
+ "body": comment_body,
+ "created": _datetime(comment.get("created")),
+ "updated": _datetime(comment.get("updated")),
+ "visibility": visibility_text,
+ "is_public": comment.get("jsdPublic"),
+ "ontology_class": "Event",
+ "content_hash": comment_hash,
+ }
+ add_vertex(
+ VertexRecord(
+ "JiraComment",
+ comment_vertex_id,
+ {
+ key: value
+ for key, value in comment_attrs.items()
+ if value not in (None, "")
+ },
+ )
+ )
+ edges.append(
+ EdgeRecord(
+ "JiraComment",
+ comment_vertex_id,
+ JIRA_COMMENT_ISSUE_EDGE,
+ "JiraIssue",
+ issue_vertex_id,
+ )
+ )
+ if author_vertex:
+ edges.append(
+ EdgeRecord(
+ "JiraComment",
+ comment_vertex_id,
+ JIRA_COMMENT_AUTHOR_EDGE,
+ "JiraUser",
+ author_vertex.vertex_id,
+ )
+ )
+ parent_comment = comment.get("parent")
+ if not isinstance(parent_comment, dict):
+ parent_comment = {}
+ parent_comment_id = comment.get("parentId") or parent_comment.get("id")
+ if parent_comment_id:
+ parent_comment_vertex_id = self._id(
+ "comment",
+ parent_comment_id,
+ )
+ add_vertex(
+ VertexRecord(
+ "JiraComment",
+ parent_comment_vertex_id,
+ {
+ "comment_id": str(parent_comment_id),
+ "ontology_class": "Event",
+ },
+ )
+ )
+ edges.append(
+ EdgeRecord(
+ "JiraComment",
+ comment_vertex_id,
+ JIRA_COMMENT_REPLY_EDGE,
+ "JiraComment",
+ parent_comment_vertex_id,
+ )
+ )
+ if previous_comment_vertex_id:
+ edges.append(
+ EdgeRecord(
+ "JiraComment",
+ comment_vertex_id,
+ JIRA_COMMENT_AFTER_EDGE,
+ "JiraComment",
+ previous_comment_vertex_id,
+ )
+ )
+ mapped_comments.append(
+ MappedComment(
+ comment_id=comment_id,
+ comment_vertex_id=comment_vertex_id,
+ issue_vertex_id=issue_vertex_id,
+ content_hash=comment_hash,
+ chunks=[
+ MappedChunk(
+ chunk_id=(
+ f"{comment_vertex_id}:chunk:{index}:"
+ f"{hashlib.sha256(text.encode('utf-8')).hexdigest()[:12]}"
+ ),
+ index=index,
+ text=text,
+ )
+ for index, text in enumerate(chunks)
+ ],
+ )
+ )
+ previous_comment_vertex_id = comment_vertex_id
+
+ parent = fields.get("parent") or {}
+ if parent.get("id"):
+ parent_key = str(parent.get("key") or parent["id"])
+ parent_id = self._issue_id(parent_key)
+ parent_fields = parent.get("fields") or {}
+ add_vertex(
+ VertexRecord(
+ "JiraIssue",
+ parent_id,
+ {
+ "issue_key": str(parent.get("key") or ""),
+ "summary": str(parent_fields.get("summary") or ""),
+ "url": (
+ f"{site_url}/browse/{parent.get('key')}"
+ if parent.get("key")
+ else site_url
+ ),
+ },
+ )
+ )
+ edges.append(
+ EdgeRecord(
+ "JiraIssue",
+ issue_vertex_id,
+ JIRA_PARENT_EDGE,
+ "JiraIssue",
+ parent_id,
+ )
+ )
+
+ for link in fields.get("issuelinks") or []:
+ link_type = link.get("type") or {}
+ target = link.get("outwardIssue")
+ relation = link_type.get("outward")
+ if not target:
+ target = link.get("inwardIssue")
+ relation = link_type.get("inward")
+ if not target or not target.get("id"):
+ continue
+ target_key = str(target.get("key") or target["id"])
+ target_id = self._issue_id(target_key)
+ target_fields = target.get("fields") or {}
+ add_vertex(
+ VertexRecord(
+ "JiraIssue",
+ target_id,
+ {
+ "issue_key": target_key,
+ "summary": str(target_fields.get("summary") or ""),
+ "url": f"{site_url}/browse/{target_key}" if target_key else site_url,
+ },
+ )
+ )
+ edges.append(
+ EdgeRecord(
+ "JiraIssue",
+ issue_vertex_id,
+ JIRA_LINK_EDGE,
+ "JiraIssue",
+ target_id,
+ {"link_type": str(relation or link_type.get("name") or "relates to")},
+ )
+ )
+
+ change_vertices, change_edges, change_parts = self._map_changes(
+ issue_id=issue_id,
+ issue_key=issue_key,
+ changelog=issue.get("changelog") or {},
+ )
+ for vertex in change_vertices:
+ add_vertex(vertex)
+ edges.extend(change_edges)
+ facts = self._facts(
+ issue_vertex_id,
+ [
+ (
+ self._record_text(
+ issue_key=issue_key,
+ issue_url=issue_url,
+ fields=fields,
+ project=project,
+ ),
+ (),
+ ),
+ *self._description_parts(issue_key, description),
+ *change_parts,
+ ],
+ )
+ content_hash = hashlib.sha256(
+ "\0".join(fact.chunk.text for fact in facts).encode("utf-8")
+ ).hexdigest()
+ issue_vertex = vertices[("JiraIssue", issue_vertex_id)]
+ issue_attributes = dict(issue_vertex.attributes)
+ issue_attributes["content_hash"] = content_hash
+ vertices[("JiraIssue", issue_vertex_id)] = VertexRecord(
+ "JiraIssue",
+ issue_vertex_id,
+ issue_attributes,
+ )
+
+ return MappedIssue(
+ issue_id=issue_id,
+ issue_vertex_id=issue_vertex_id,
+ updated=_parsed_datetime(fields.get("updated")),
+ content_hash=content_hash,
+ vertices=list(vertices.values()),
+ edges=edges,
+ facts=facts,
+ legacy_document_id=self._id("issue-doc", issue_id),
+ change_vertex_ids=[vertex.vertex_id for vertex in change_vertices],
+ comments=mapped_comments,
+ )
+
+ def _comment_document_text(
+ self,
+ *,
+ issue_key: str,
+ comment: dict[str, Any],
+ body: str,
+ ) -> str:
+ author = _display_name(comment.get("author")) or "Unknown user"
+ created = _datetime(comment.get("created")) or ""
+ when = f" at {created}" if created else ""
+ return (
+ f"Issue: {issue_key}\n"
+ f"Comment by {author}{when}:\n"
+ f"{body.strip()}"
+ )
+
+ def _lead_with_issue(self, issue_key: str, text: str) -> str:
+ cleaned = text.strip()
+ prefix = f"Issue: {issue_key}"
+ if not cleaned or cleaned.startswith(prefix):
+ return cleaned
+ return f"{prefix}\n{cleaned}"
+
+ def _record_text(
+ self,
+ *,
+ issue_key: str,
+ issue_url: str,
+ fields: dict[str, Any],
+ project: dict[str, Any],
+ ) -> str:
+ status = fields.get("status") or {}
+ status_category = status.get("statusCategory") or {}
+ issue_type = fields.get("issuetype") or {}
+ priority = fields.get("priority") or {}
+ resolution = fields.get("resolution") or {}
+ project_line = (
+ f"{project.get('key') or ''} {project.get('name') or ''}".strip()
+ )
+ status_name = str(status.get("name") or "")
+ category = str(status_category.get("key") or "")
+ status_line = (
+ f"{status_name} ({category})".strip()
+ if category
+ else status_name
+ )
+ lines = [f"Issue: {issue_key}"]
+ for label, value in (
+ ("Summary", fields.get("summary")),
+ ("URL", issue_url),
+ ("Project", project_line),
+ ("Type", issue_type.get("name")),
+ ("Status", status_line),
+ ("Priority", priority.get("name")),
+ ("Resolution", resolution.get("name")),
+ ("Assignee", _display_name(fields.get("assignee")) or "Unassigned"),
+ ("Reporter", _display_name(fields.get("reporter"))),
+ ("Labels", _comma_names(fields.get("labels"))),
+ ("Components", _comma_names(fields.get("components"))),
+ ("Fix versions", _comma_names(fields.get("fixVersions"))),
+ ("Due", _datetime(fields.get("duedate"))),
+ ("Updated", _datetime(fields.get("updated"))),
+ ):
+ text = str(value or "").strip()
+ if text:
+ lines.append(f"{label}: {text}")
+ return "\n".join(lines)
+
+ def _description_parts(
+ self,
+ issue_key: str,
+ description: str,
+ ) -> list[tuple[str, tuple[tuple[str, str], ...]]]:
+ if not description or not description.strip():
+ return []
+ source_text = f"Issue: {issue_key}\nDescription:\n{description.strip()}"
+ return [
+ (self._lead_with_issue(issue_key, str(piece)), ())
+ for piece in self.comment_chunker.chunk(source_text)
+ if str(piece).strip()
+ ]
+
+ _CHANGELOG_FIELDS = {"status", "assignee", "priority", "resolution"}
+
+ def _map_changes(
+ self,
+ *,
+ issue_id: str,
+ issue_key: str,
+ changelog: dict[str, Any],
+ ) -> tuple[
+ list[VertexRecord],
+ list[EdgeRecord],
+ list[tuple[str, tuple[tuple[str, str], ...]]],
+ ]:
+ """One JiraChange event and one fact per tracked changelog item."""
+ histories = sorted(
+ changelog.get("histories") or [],
+ key=lambda history: history.get("created") or "",
+ )
+ vertices: list[VertexRecord] = []
+ edges: list[EdgeRecord] = []
+ parts: list[tuple[str, tuple[tuple[str, str], ...]]] = []
+ change_index = 0
+ issue_vertex_id = self._issue_id(issue_key)
+ for history in histories:
+ author = history.get("author") or {}
+ author_name = _display_name(author) or "Unknown"
+ created = _datetime(history.get("created"))
+ history_id = str(history.get("id") or "")
+ for item in history.get("items") or []:
+ field = str(item.get("field") or "").lower()
+ if field not in self._CHANGELOG_FIELDS:
+ continue
+ from_val = str(item.get("fromString") or "").strip()
+ to_val = str(item.get("toString") or "").strip()
+ if not to_val:
+ continue
+ change_index += 1
+ change_key = (
+ f"{issue_id}:{history_id or change_index}:"
+ f"{field}:{change_index}"
+ )
+ change_vertex_id = self._id("change", change_key)
+ attributes = {
+ "change_id": change_key,
+ "field": field,
+ "from_value": from_val,
+ "to_value": to_val,
+ "created": created,
+ "ontology_class": "Event",
+ }
+ vertices.append(
+ VertexRecord(
+ "JiraChange",
+ change_vertex_id,
+ {
+ key: value
+ for key, value in attributes.items()
+ if value not in (None, "")
+ },
+ )
+ )
+ edges.append(
+ EdgeRecord(
+ "JiraIssue",
+ issue_vertex_id,
+ JIRA_CHANGE_EDGE,
+ "JiraChange",
+ change_vertex_id,
+ )
+ )
+ author_vertex = self._user_vertex(author)
+ if author_vertex:
+ vertices.append(author_vertex)
+ edges.append(
+ EdgeRecord(
+ "JiraChange",
+ change_vertex_id,
+ JIRA_CHANGE_AUTHOR_EDGE,
+ "JiraUser",
+ author_vertex.vertex_id,
+ )
+ )
+ when = f"{created}: " if created else ""
+ if from_val:
+ sentence = (
+ f"{when}{author_name} changed {field} from "
+ f"\"{from_val}\" to \"{to_val}\""
+ )
+ else:
+ sentence = (
+ f"{when}{author_name} set {field} to \"{to_val}\""
+ )
+ parts.append(
+ (
+ f"Issue: {issue_key}\n{sentence}",
+ (("JiraChange", change_vertex_id),),
+ )
+ )
+ return vertices, edges, parts
+
+ def _facts(
+ self,
+ issue_vertex_id: str,
+ parts: list[tuple[str, tuple[tuple[str, str], ...]]],
+ ) -> list[MappedFact]:
+ facts: list[MappedFact] = []
+ for text, entities in parts:
+ cleaned = text.strip()
+ if not cleaned:
+ continue
+ digest = hashlib.sha256(cleaned.encode("utf-8")).hexdigest()[:12]
+ index = len(facts)
+ facts.append(
+ MappedFact(
+ chunk=MappedChunk(
+ chunk_id=f"{issue_vertex_id}:fact:{index}:{digest}",
+ index=index,
+ text=cleaned + "\n",
+ ),
+ entities=entities,
+ )
+ )
+ return facts
diff --git a/graphrag/app/connectors/jira/schema.py b/graphrag/app/connectors/jira/schema.py
new file mode 100644
index 00000000..007b9344
--- /dev/null
+++ b/graphrag/app/connectors/jira/schema.py
@@ -0,0 +1,313 @@
+"""Predefined Jira domain schema."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from common.db.schema_utils import SchemaProposal, read_existing_schema
+
+
+JIRA_PROJECT_EDGE = "JIRA_BELONGS_TO"
+JIRA_ASSIGNEE_EDGE = "JIRA_ASSIGNED_TO"
+JIRA_REPORTER_EDGE = "JIRA_REPORTED_BY"
+JIRA_PARENT_EDGE = "JIRA_HAS_PARENT"
+JIRA_LINK_EDGE = "JIRA_LINKS_TO"
+JIRA_COMMENT_ISSUE_EDGE = "JIRA_COMMENT_ON"
+JIRA_COMMENT_AUTHOR_EDGE = "JIRA_COMMENTED_BY"
+JIRA_COMMENT_REPLY_EDGE = "JIRA_COMMENT_REPLIES_TO"
+JIRA_COMMENT_AFTER_EDGE = "JIRA_COMMENT_AFTER"
+JIRA_CHANGE_EDGE = "JIRA_HAS_CHANGE"
+JIRA_CHANGE_AUTHOR_EDGE = "JIRA_CHANGE_BY"
+
+
+def _attribute_types(metadata: dict[str, Any]) -> dict[str, str]:
+ attributes: dict[str, str] = {}
+ primary_id = (metadata.get("PrimaryId") or {}).get("AttributeName")
+ for attribute in metadata.get("Attributes") or []:
+ name = attribute.get("AttributeName")
+ if not name or name == primary_id:
+ continue
+ attribute_type = (
+ (attribute.get("AttributeType") or {}).get("Name") or "STRING"
+ )
+ attributes[str(name).casefold()] = str(attribute_type).upper()
+ return attributes
+
+
+def _edge_pairs(metadata: dict[str, Any]) -> set[tuple[str, str]]:
+ pairs: set[tuple[str, str]] = set()
+ source = metadata.get("FromVertexTypeName")
+ target = metadata.get("ToVertexTypeName")
+ if source and target and source != "*" and target != "*":
+ pairs.add((str(source).casefold(), str(target).casefold()))
+ for pair in metadata.get("EdgePairs") or []:
+ source = pair.get("From")
+ target = pair.get("To")
+ if source and target:
+ pairs.add((str(source).casefold(), str(target).casefold()))
+ return pairs
+
+
+def jira_schema_proposal() -> SchemaProposal:
+ """Return the bounded Jira schema consumed by GraphRAG."""
+ proposal = SchemaProposal(domain_label="Jira Cloud")
+ proposal.add_vertex(
+ "JiraProject",
+ (
+ "A Jira project, modeled as a POLE+O Object subtype and work "
+ "container. project_key is the short key, such as PAY."
+ ),
+ [
+ ("project_key", "STRING"),
+ ("name", "STRING"),
+ ("url", "STRING"),
+ ],
+ )
+ proposal.add_vertex(
+ "JiraIssue",
+ (
+ "A Jira work item modeled as a POLE+O Object subtype. Filter using "
+ "issue_key, status, status_category (new, indeterminate, or done), "
+ "priority, issue_type, resolution, labels, components, "
+ "fix_versions, created, updated, or due. Status, assignee, "
+ "priority, and resolution changes are JiraChange events."
+ ),
+ [
+ ("issue_key", "STRING"),
+ ("summary", "STRING"),
+ ("issue_type", "STRING"),
+ ("status", "STRING"),
+ ("status_category", "STRING"),
+ ("priority", "STRING"),
+ ("resolution", "STRING"),
+ ("labels", "STRING"),
+ ("components", "STRING"),
+ ("fix_versions", "STRING"),
+ ("created", "DATETIME"),
+ ("updated", "DATETIME"),
+ ("due", "DATETIME"),
+ ("url", "STRING"),
+ ("story_points", "DOUBLE"),
+ ("content_hash", "STRING"),
+ ],
+ )
+ proposal.add_vertex(
+ "JiraUser",
+ (
+ "An Atlassian account modeled as a POLE+O Person subtype. "
+ "account_id is the stable identity and display_name is the "
+ "human-readable name."
+ ),
+ [
+ ("account_id", "STRING"),
+ ("display_name", "STRING"),
+ ],
+ )
+ proposal.add_vertex(
+ "JiraComment",
+ (
+ "A Jira comment modeled as a POLE+O Event subtype. body is the "
+ "comment text. Searchable text is also stored as document chunks. "
+ "Use graph edges for issue, author, ordering, and explicit replies."
+ ),
+ [
+ ("comment_id", "STRING"),
+ ("body", "STRING"),
+ ("created", "DATETIME"),
+ ("updated", "DATETIME"),
+ ("visibility", "STRING"),
+ ("is_public", "BOOL"),
+ ("ontology_class", "STRING"),
+ ("content_hash", "STRING"),
+ ],
+ )
+ proposal.add_vertex(
+ "JiraChange",
+ (
+ "One status, assignee, priority, or resolution change, modeled as "
+ "a POLE+O Event subtype. field is the changed field, from_value "
+ "and to_value are the previous and new values, and created is "
+ "when the change happened. The author is JIRA_CHANGE_BY."
+ ),
+ [
+ ("change_id", "STRING"),
+ ("field", "STRING"),
+ ("from_value", "STRING"),
+ ("to_value", "STRING"),
+ ("created", "DATETIME"),
+ ("ontology_class", "STRING"),
+ ],
+ )
+
+ proposal.add_edge_pair(
+ JIRA_PROJECT_EDGE,
+ "JiraIssue",
+ "JiraProject",
+ "The Jira issue's current project.",
+ )
+ proposal.add_edge_pair(
+ JIRA_ASSIGNEE_EDGE,
+ "JiraIssue",
+ "JiraUser",
+ "The Jira issue's current assignee.",
+ )
+ proposal.add_edge_pair(
+ JIRA_REPORTER_EDGE,
+ "JiraIssue",
+ "JiraUser",
+ "The Atlassian account that reported the Jira issue.",
+ )
+ proposal.add_edge_pair(
+ JIRA_PARENT_EDGE,
+ "JiraIssue",
+ "JiraIssue",
+ "The parent Jira issue, including epic and subtask parents.",
+ )
+ proposal.add_edge_pair(
+ JIRA_LINK_EDGE,
+ "JiraIssue",
+ "JiraIssue",
+ (
+ "A directed Jira issue link. link_type is the phrase from the "
+ "source issue toward the target, such as blocks or is blocked by."
+ ),
+ [("link_type", "STRING")],
+ )
+ proposal.add_edge_pair(
+ JIRA_COMMENT_ISSUE_EDGE,
+ "JiraComment",
+ "JiraIssue",
+ "The Jira issue on which the comment was posted.",
+ )
+ proposal.add_edge_pair(
+ JIRA_COMMENT_AUTHOR_EDGE,
+ "JiraComment",
+ "JiraUser",
+ "The Atlassian account that authored the comment.",
+ )
+ proposal.add_edge_pair(
+ JIRA_COMMENT_REPLY_EDGE,
+ "JiraComment",
+ "JiraComment",
+ "An explicit source-provided parent comment; never inferred.",
+ )
+ proposal.add_edge_pair(
+ JIRA_COMMENT_AFTER_EDGE,
+ "JiraComment",
+ "JiraComment",
+ "Chronological order between adjacent comments on one issue.",
+ )
+ proposal.add_edge_pair(
+ JIRA_CHANGE_EDGE,
+ "JiraIssue",
+ "JiraChange",
+ "A status, assignee, priority, or resolution change on this issue.",
+ )
+ proposal.add_edge_pair(
+ JIRA_CHANGE_AUTHOR_EDGE,
+ "JiraChange",
+ "JiraUser",
+ "The Atlassian account that made the change.",
+ )
+ return proposal
+
+
+def jira_schema_status(conn) -> dict[str, Any]:
+ """Inspect whether the one current Jira schema is fully installed."""
+ proposal = jira_schema_proposal()
+ existing = read_existing_schema(conn)
+ conflicts: list[str] = []
+ missing_vertices: list[str] = []
+ missing_edges: list[str] = []
+ missing_pairs: list[str] = []
+
+ required_core = ("Document", "DocumentChunk", "Content")
+ missing_core = [name for name in required_core if not existing.has_vertex(name)]
+ required_core_edges = ("CONTAINS_ENTITY", "HAS_CHILD", "HAS_CONTENT")
+ missing_core_edges = [
+ name for name in required_core_edges if not existing.has_edge(name)
+ ]
+ if missing_core or missing_core_edges:
+ return {
+ "status": "not_initialized",
+ "missing": {
+ "core_vertices": missing_core,
+ "core_edges": missing_core_edges,
+ },
+ "conflicts": [],
+ }
+
+ for vertex in proposal.vertices:
+ if not existing.has_vertex(vertex.name):
+ missing_vertices.append(vertex.name)
+ continue
+ metadata = conn.getVertexType(vertex.name) or {}
+ actual = _attribute_types(metadata)
+ for attribute in vertex.attributes:
+ actual_type = actual.get(attribute.name.casefold())
+ if actual_type is None:
+ conflicts.append(
+ f"{vertex.name}.{attribute.name} is missing"
+ )
+ elif actual_type != attribute.type.upper():
+ conflicts.append(
+ f"{vertex.name}.{attribute.name} must be "
+ f"{attribute.type.upper()}, found {actual_type}"
+ )
+
+ for edge in proposal.edges:
+ if not existing.has_edge(edge.name):
+ missing_edges.append(edge.name)
+ continue
+ metadata = conn.getEdgeType(edge.name) or {}
+ if bool(metadata.get("IsDirected")) != edge.directed:
+ expected = "directed" if edge.directed else "undirected"
+ conflicts.append(f"{edge.name} must be {expected}")
+ actual_attributes = _attribute_types(metadata)
+ for attribute in edge.attributes:
+ actual_type = actual_attributes.get(attribute.name.casefold())
+ if actual_type is None:
+ conflicts.append(f"{edge.name}.{attribute.name} is missing")
+ elif actual_type != attribute.type.upper():
+ conflicts.append(
+ f"{edge.name}.{attribute.name} must be "
+ f"{attribute.type.upper()}, found {actual_type}"
+ )
+ actual_pairs = _edge_pairs(metadata)
+ for source, target in edge.pairs:
+ if (source.casefold(), target.casefold()) not in actual_pairs:
+ missing_pairs.append(f"{edge.name}: {source} -> {target}")
+
+ required_links = (
+ ("CONTAINS_ENTITY", "Document", "JiraIssue"),
+ ("CONTAINS_ENTITY", "DocumentChunk", "JiraIssue"),
+ ("CONTAINS_ENTITY", "Document", "JiraComment"),
+ ("CONTAINS_ENTITY", "DocumentChunk", "JiraComment"),
+ ("CONTAINS_ENTITY", "Document", "JiraChange"),
+ ("CONTAINS_ENTITY", "DocumentChunk", "JiraChange"),
+ )
+ for edge, source, target in required_links:
+ if not existing.has_edge_pair(edge, source, target):
+ missing_pairs.append(f"{edge}: {source} -> {target}")
+
+ missing = {
+ "vertices": missing_vertices,
+ "edges": missing_edges,
+ "pairs": missing_pairs,
+ }
+ if conflicts:
+ status = "conflict"
+ elif not any(missing.values()):
+ status = "installed"
+ elif any(
+ existing.has_vertex(vertex.name) for vertex in proposal.vertices
+ ):
+ status = "incomplete"
+ else:
+ status = "not_installed"
+
+ return {
+ "status": status,
+ "missing": missing,
+ "conflicts": conflicts,
+ }
diff --git a/graphrag/app/connectors/jira/state.py b/graphrag/app/connectors/jira/state.py
new file mode 100644
index 00000000..e99e1a4e
--- /dev/null
+++ b/graphrag/app/connectors/jira/state.py
@@ -0,0 +1,150 @@
+"""Atomic persistence and secret redaction for Jira data sources."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+from common.config import _config_file_lock, validate_graphname
+
+from .config import JiraDataSource, JiraSourceFile, JiraSyncState
+
+
+MASKED_SECRET = "********"
+NON_TENANT_ATLASSIAN_HOSTS = {
+ "api.atlassian.com",
+ "atlassian.net",
+ "graphql.atlassian.net",
+ "id.atlassian.com",
+}
+
+
+class JiraSourceStore:
+ def __init__(self, config_root: str = "configs/graph_configs"):
+ self.config_root = Path(config_root)
+
+ def path(self, graphname: str) -> Path:
+ validate_graphname(graphname)
+ return self.config_root / graphname / "data_sources.json"
+
+ def load(self, graphname: str) -> JiraSourceFile:
+ path = self.path(graphname)
+ if not path.exists():
+ return JiraSourceFile()
+ with _config_file_lock:
+ with path.open("r", encoding="utf-8") as stream:
+ payload = json.load(stream)
+ return JiraSourceFile.model_validate(payload)
+
+ def save(self, graphname: str, config: JiraSourceFile) -> None:
+ path = self.path(graphname)
+ path.parent.mkdir(parents=True, exist_ok=True)
+ tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
+ payload = config.model_dump(mode="json")
+ for source in payload["sources"]:
+ if not source["connection"].get("api_token"):
+ source["connection"].pop("api_token", None)
+ with _config_file_lock:
+ try:
+ with tmp.open("w", encoding="utf-8") as stream:
+ json.dump(payload, stream, indent=2)
+ stream.flush()
+ os.fsync(stream.fileno())
+ os.replace(tmp, path)
+ finally:
+ if tmp.exists():
+ tmp.unlink()
+
+ def list(self, graphname: str, *, redact: bool = True) -> list[dict]:
+ sources = self.load(graphname).sources
+ return [self.redact(source) if redact else source.model_dump(mode="json") for source in sources]
+
+ def get(self, graphname: str, source_id: str) -> JiraDataSource:
+ for source in self.load(graphname).sources:
+ if source.id == source_id:
+ return source
+ raise KeyError(source_id)
+
+ def upsert(self, graphname: str, submitted: JiraDataSource) -> JiraDataSource:
+ hostname = urlparse(submitted.connection.site_url).hostname
+ if hostname and hostname.lower() in NON_TENANT_ATLASSIAN_HOSTS:
+ raise ValueError(
+ "site_url must be your Jira tenant URL, for example "
+ "https://your-company.atlassian.net"
+ )
+ config = self.load(graphname)
+ sources = list(config.sources)
+ existing = next((s for s in sources if s.id == submitted.id), None)
+ overlap_seconds = submitted.sync.overlap_seconds
+ submitted.connection.api_token = (
+ existing.connection.api_token
+ if existing is not None
+ and existing.connection.api_token != MASKED_SECRET
+ else ""
+ )
+ if existing is not None:
+ credentials_changed = (
+ submitted.connection.email != existing.connection.email
+ )
+ if submitted.connection.site_url != existing.connection.site_url:
+ unused_draft = (
+ existing.sync.last_tested_at is None
+ and existing.sync.checkpoint is None
+ and existing.sync.last_issue_count == 0
+ )
+ if not unused_draft:
+ raise ValueError(
+ "site_url cannot be changed after a source has been used; "
+ "create a new source so existing Jira data can be reconciled"
+ )
+ credentials_changed = True
+ else:
+ pass
+ submitted.sync = existing.sync.model_copy(deep=True)
+ submitted.sync.overlap_seconds = overlap_seconds
+ if credentials_changed:
+ submitted.sync.last_tested_at = None
+ if submitted.scope != existing.scope:
+ submitted.sync.checkpoint = None
+ else:
+ submitted.sync = JiraSyncState(overlap_seconds=overlap_seconds)
+ sources = [source for source in sources if source.id != submitted.id]
+ sources.append(submitted)
+ sources.sort(key=lambda source: source.id)
+ config.sources = sources
+ self.save(graphname, config)
+ return submitted
+
+ def delete(self, graphname: str, source_id: str) -> None:
+ config = self.load(graphname)
+ remaining = [source for source in config.sources if source.id != source_id]
+ if len(remaining) == len(config.sources):
+ raise KeyError(source_id)
+ config.sources = remaining
+ self.save(graphname, config)
+
+ def update_runtime_state(
+ self,
+ graphname: str,
+ source: JiraDataSource,
+ ) -> None:
+ """Persist connector-owned sync fields without losing edits."""
+ config = self.load(graphname)
+ sources = list(config.sources)
+ for index, stored in enumerate(sources):
+ if stored.id != source.id:
+ continue
+ stored.sync = source.sync
+ sources[index] = stored
+ config.sources = sources
+ self.save(graphname, config)
+ return
+ raise KeyError(source.id)
+
+ @staticmethod
+ def redact(source: JiraDataSource) -> dict:
+ payload = source.model_dump(mode="json")
+ payload["connection"]["api_token"] = ""
+ return payload
diff --git a/graphrag/app/connectors/jira/sync.py b/graphrag/app/connectors/jira/sync.py
new file mode 100644
index 00000000..e26b3004
--- /dev/null
+++ b/graphrag/app/connectors/jira/sync.py
@@ -0,0 +1,1173 @@
+"""Jira Cloud synchronization into the existing GraphRAG pipeline."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import queue
+import threading
+import time
+from collections import defaultdict
+from datetime import datetime, timezone
+from typing import Any, Iterator
+
+from common.config import get_embedding_store
+from common.db.health import embedding_coverage
+
+from .client import JiraCloudClient
+from .config import JiraDataSource
+from .mapper import (
+ EdgeRecord,
+ JiraIssueMapper,
+ MappedChunk,
+ MappedComment,
+ MappedIssue,
+ VertexRecord,
+)
+from .schema import (
+ JIRA_ASSIGNEE_EDGE,
+ JIRA_CHANGE_AUTHOR_EDGE,
+ JIRA_CHANGE_EDGE,
+ JIRA_COMMENT_AFTER_EDGE,
+ JIRA_COMMENT_AUTHOR_EDGE,
+ JIRA_COMMENT_ISSUE_EDGE,
+ JIRA_COMMENT_REPLY_EDGE,
+ JIRA_LINK_EDGE,
+ JIRA_PARENT_EDGE,
+ JIRA_PROJECT_EDGE,
+ JIRA_REPORTER_EDGE,
+ jira_schema_status,
+)
+from .state import JiraSourceStore
+
+logger = logging.getLogger(__name__)
+
+CURRENT_STATE_EDGES = (
+ JIRA_PROJECT_EDGE,
+ JIRA_ASSIGNEE_EDGE,
+ JIRA_REPORTER_EDGE,
+ JIRA_PARENT_EDGE,
+ JIRA_LINK_EDGE,
+ JIRA_CHANGE_EDGE,
+)
+COMMENT_STATE_EDGES = (
+ JIRA_COMMENT_ISSUE_EDGE,
+ JIRA_COMMENT_AUTHOR_EDGE,
+ JIRA_COMMENT_REPLY_EDGE,
+ JIRA_COMMENT_AFTER_EDGE,
+)
+UPSERT_BATCH_SIZE = 500
+EMBEDDING_BATCH_SIZE = 32
+# Flush accumulated documents and comment-chunks to TigerGraph every this
+# many Jira pages. A higher value means fewer supportai.ingest() round-trips
+# (the dominant overhead for large initial loads) at the cost of a larger
+# in-memory accumulation. 10 pages × 100 issues = 1 000 issues per flush.
+DOC_LOAD_BATCH_PAGES = 10
+
+
+def _attrs(attributes: dict[str, Any]) -> dict[str, dict[str, Any]]:
+ return {
+ key: {"value": value}
+ for key, value in attributes.items()
+ if value is not None
+ }
+
+
+def _upsert_payload(
+ vertices: list[VertexRecord],
+ edges: list[EdgeRecord],
+) -> dict[str, Any]:
+ payload: dict[str, Any] = {
+ "vertices": defaultdict(dict),
+ "edges": defaultdict(
+ lambda: defaultdict(
+ lambda: defaultdict(lambda: defaultdict(dict))
+ )
+ ),
+ }
+ for vertex in vertices:
+ payload["vertices"][vertex.vertex_type][vertex.vertex_id] = _attrs(
+ vertex.attributes
+ )
+ for edge in edges:
+ payload["edges"][edge.source_type][edge.source_id][edge.edge_type][
+ edge.target_type
+ ][edge.target_id] = _attrs(edge.attributes)
+ return payload
+
+
+class JiraSyncService:
+ def __init__(
+ self,
+ graphname: str,
+ source: JiraDataSource,
+ conn,
+ *,
+ store: JiraSourceStore | None = None,
+ client: JiraCloudClient | None = None,
+ ):
+ self.graphname = graphname
+ self.source = source
+ self.conn = conn
+ self.store = store or JiraSourceStore()
+ self.client = client or JiraCloudClient(source)
+ self._owns_client = client is None
+
+ def run(self) -> dict[str, Any]:
+ if not self.source.enabled:
+ raise ValueError("Cannot sync a disabled data source")
+
+ started = datetime.now(timezone.utc)
+ self.source.sync.last_started_at = started
+ self.source.sync.last_error = None
+ self.store.update_runtime_state(self.graphname, self.source)
+
+ try:
+ schema_result = jira_schema_status(self.conn)
+ if schema_result["status"] != "installed":
+ raise RuntimeError(
+ "Jira schema is not installed for this graph. "
+ "Install it from Data Sources before synchronization."
+ )
+ self._prepare_legacy_comment_migration()
+
+ mapper = JiraIssueMapper(
+ self.source,
+ graphname=self.graphname,
+ )
+ existing_hashes = self._all_existing_hashes()
+ # If the graph has no JiraIssue vertices but a checkpoint exists,
+ # the graph was likely cleared or recreated. Reset the checkpoint
+ # so the next iteration performs a full sync instead of a no-op.
+ if not existing_hashes and self.source.sync.checkpoint is not None:
+ logger.warning(
+ "Graph appears empty but checkpoint is set — resetting "
+ "checkpoint for graph=%s source=%s to trigger a full sync",
+ self.graphname,
+ self.source.id,
+ )
+ self.source.sync.checkpoint = None
+ self.store.update_runtime_state(self.graphname, self.source)
+ existing_comment_hashes = self._all_existing_comment_hashes()
+ issues_upserted = 0
+ documents_loaded = 0
+ comments_deleted = 0
+
+ # Pending batches — accumulated across DOC_LOAD_BATCH_PAGES pages
+ # before being flushed to TigerGraph in one supportai.ingest call.
+ pending_docs: list[MappedIssue] = []
+ pending_comments: list[MappedComment] = []
+ pages_since_flush = 0
+ # Track the highest "updated" timestamp seen so far. We only
+ # advance the checkpoint AFTER a successful batch flush so that a
+ # crash before the flush leaves the checkpoint behind the lost
+ # pages, ensuring they are re-fetched on the next sync run.
+ pending_checkpoint = self.source.sync.checkpoint
+
+ # ── Recovery phase ────────────────────────────────────────────────
+ # Re-embed ingested issues whose content hash was cleared before
+ # embedding finished. This does not change the checkpoint and does
+ # not replace the project scan below.
+ recovery_keys = self._find_recovery_issue_keys()
+ if recovery_keys:
+ recovery_docs: list[MappedIssue] = []
+ recovery_comments: list[MappedComment] = []
+ for issues in self.client.iter_issues_by_keys(recovery_keys):
+ mapped = [mapper.map(issue) for issue in issues]
+ changed_docs, changed_comms, deleted = self._upsert_issue_records(
+ mapped, existing_hashes, existing_comment_hashes
+ )
+ recovery_docs.extend(changed_docs)
+ recovery_comments.extend(changed_comms)
+ documents_loaded += len(changed_docs)
+ comments_deleted += deleted
+ issues_upserted += len(mapped)
+
+ if recovery_docs or recovery_comments:
+ self._finalize_batch(
+ recovery_docs,
+ recovery_comments,
+ existing_hashes,
+ existing_comment_hashes,
+ )
+ # Do not move the checkpoint here. A cleared checkpoint means a
+ # new project must be scanned from the start. Link stubs are
+ # not failed embeddings, and their update times must not become
+ # the checkpoint for that scan.
+ logger.info(
+ "Recovery complete: %d doc(s), %d comment(s) re-embedded "
+ "for graph=%s source=%s. The project scan continues from "
+ "the saved checkpoint.",
+ len(recovery_docs), len(recovery_comments),
+ self.graphname, self.source.id,
+ )
+ # ── End recovery phase ────────────────────────────────────────────
+
+ for issues in self._iter_pages_pipelined():
+ mapped = [mapper.map(issue) for issue in issues]
+ changed_docs, changed_comms, deleted_count = (
+ self._upsert_issue_records(
+ mapped,
+ existing_hashes,
+ existing_comment_hashes,
+ )
+ )
+ pending_docs.extend(changed_docs)
+ pending_comments.extend(changed_comms)
+ documents_loaded += len(changed_docs)
+ comments_deleted += deleted_count
+ issues_upserted += len(mapped)
+ pages_since_flush += 1
+
+ page_max = max(
+ (item.updated for item in mapped if item.updated is not None),
+ default=None,
+ )
+ if page_max is not None and (
+ pending_checkpoint is None or page_max > pending_checkpoint
+ ):
+ pending_checkpoint = page_max
+
+ if pages_since_flush >= DOC_LOAD_BATCH_PAGES:
+ self._finalize_batch(
+ pending_docs,
+ pending_comments,
+ existing_hashes,
+ existing_comment_hashes,
+ )
+ pending_docs = []
+ pending_comments = []
+ pages_since_flush = 0
+ if pending_checkpoint != self.source.sync.checkpoint:
+ self.source.sync.checkpoint = pending_checkpoint
+ self.store.update_runtime_state(
+ self.graphname, self.source
+ )
+
+ # Final flush for any pages that did not fill a complete batch.
+ if pending_docs or pending_comments:
+ self._finalize_batch(
+ pending_docs,
+ pending_comments,
+ existing_hashes,
+ existing_comment_hashes,
+ )
+ if pending_checkpoint != self.source.sync.checkpoint:
+ self.source.sync.checkpoint = pending_checkpoint
+ self.store.update_runtime_state(self.graphname, self.source)
+
+ chunk_coverage = embedding_coverage(
+ self.conn,
+ "DocumentChunk",
+ )
+ logger.info(
+ "embedding_coverage graph=%s: %s",
+ self.graphname,
+ chunk_coverage,
+ )
+ missing_chunk_embeddings = (
+ int(chunk_coverage["missing"])
+ if chunk_coverage is not None
+ else 0
+ )
+
+ # Recovery check: if this sync loaded nothing new but there are
+ # DocumentChunks still at epoch_processed=0, a previous ECC rebuild
+ # must have failed partway through. Trigger a fresh rebuild so
+ # those chunks get embedded without the user having to intervene.
+ if not documents_loaded and not comments_deleted and not missing_chunk_embeddings:
+ if self._has_unprocessed_chunks():
+ logger.info(
+ "Detected unprocessed DocumentChunks from a prior failed "
+ "ECC rebuild for graph=%s source=%s — triggering recovery rebuild",
+ self.graphname,
+ self.source.id,
+ )
+ missing_chunk_embeddings = 1
+ self.source.sync.last_completed_at = datetime.now(timezone.utc)
+ self.source.sync.last_issue_count = issues_upserted
+ self.source.sync.last_error = None
+ self.source.sync.migrating_legacy_comments = False
+ self.store.update_runtime_state(self.graphname, self.source)
+ return {
+ "status": "completed",
+ "issues_upserted": issues_upserted,
+ "issues_deleted": 0,
+ "comments_deleted": comments_deleted,
+ "documents_loaded": documents_loaded,
+ "checkpoint": (
+ self.source.sync.checkpoint.isoformat()
+ if self.source.sync.checkpoint is not None
+ else None
+ ),
+ "schema": schema_result["status"],
+ "missing_chunk_embeddings": missing_chunk_embeddings,
+ "rebuild_required": bool(
+ documents_loaded
+ or comments_deleted
+ or missing_chunk_embeddings
+ ),
+ }
+ except Exception as exc:
+ logger.exception(
+ "Jira sync failed for graph=%s source=%s",
+ self.graphname,
+ self.source.id,
+ )
+ self.source.sync.last_error = str(exc)[:1000]
+ self.store.update_runtime_state(self.graphname, self.source)
+ raise
+ finally:
+ if self._owns_client:
+ self.client.close()
+
+ def _find_recovery_issue_keys(self) -> set[str]:
+ """Return issue keys whose own ingest was written but not embedded.
+
+ Link and parent placeholders also have an empty content_hash. They
+ were never ingested, so they are not failed embeddings. Only a full
+ issue (status or updated set) or a real comment (body or created set)
+ is recovered. This does not replace the project scan.
+ """
+ keys: set[str] = set()
+
+ try:
+ issue_verts = self.conn.getVertices(
+ "JiraIssue",
+ where='content_hash=""',
+ select="content_hash,status,updated",
+ limit=5000,
+ ) or []
+ for vertex in issue_verts:
+ if not self._issue_embed_incomplete(vertex):
+ continue
+ key = self._issue_key_from_vertex_id(str(vertex.get("v_id", "")))
+ if key:
+ keys.add(key)
+ except Exception as exc:
+ logger.warning(
+ "Recovery: could not query empty-hash JiraIssue vertices "
+ "graph=%s: %s", self.graphname, exc,
+ )
+
+ try:
+ comment_verts = self.conn.getVertices(
+ "JiraComment",
+ where='content_hash=""',
+ select="content_hash,body,created",
+ limit=5000,
+ ) or []
+ for vertex in comment_verts:
+ if not self._comment_embed_incomplete(vertex):
+ continue
+ comment_v_id = str(vertex.get("v_id", ""))
+ try:
+ edges = self.conn.getEdges(
+ "JiraComment",
+ comment_v_id,
+ JIRA_COMMENT_ISSUE_EDGE,
+ ) or []
+ except Exception:
+ continue
+ for edge in edges:
+ key = self._issue_key_from_vertex_id(str(edge.get("to_id", "")))
+ if key:
+ keys.add(key)
+ except Exception as exc:
+ logger.warning(
+ "Recovery: could not query empty-hash JiraComment vertices "
+ "graph=%s: %s", self.graphname, exc,
+ )
+
+ if keys:
+ logger.info(
+ "Recovery: found %d ingested issue(s) with missing embeddings "
+ "in graph=%s source=%s. Re-embedding them without moving the "
+ "project checkpoint.",
+ len(keys), self.graphname, self.source.id,
+ )
+ return keys
+
+ @staticmethod
+ def _vertex_attr(vertex: dict, name: str) -> str:
+ attributes = vertex.get("attributes") or {}
+ value = attributes.get(name)
+ if isinstance(value, dict):
+ value = value.get("value")
+ return str(value or "").strip()
+
+ @classmethod
+ def _issue_embed_incomplete(cls, vertex: dict) -> bool:
+ """True for an ingested issue whose content hash was not saved.
+
+ A link placeholder has no status and no real updated time.
+ """
+ status = cls._vertex_attr(vertex, "status")
+ updated = cls._vertex_attr(vertex, "updated")
+ return bool(status) or cls._is_real_timestamp(updated)
+
+ @classmethod
+ def _comment_embed_incomplete(cls, vertex: dict) -> bool:
+ """True for a stored comment whose content hash was not saved.
+
+ A parent-comment placeholder has no body and no created time.
+ """
+ body = cls._vertex_attr(vertex, "body")
+ created = cls._vertex_attr(vertex, "created")
+ return bool(body) or cls._is_real_timestamp(created)
+
+ @staticmethod
+ def _is_real_timestamp(value: str) -> bool:
+ return bool(value) and not value.startswith("1970-01-01")
+
+ @staticmethod
+ def _issue_key_from_vertex_id(vertex_id: str) -> str:
+ if not vertex_id.startswith("jira:") or not vertex_id.endswith(":issue"):
+ return ""
+ return vertex_id.removeprefix("jira:").removesuffix(":issue").upper()
+
+ def _all_content_hashes(self, vertex_type: str) -> dict[str, str]:
+ # pyTigerGraph's getVerticesById raises error 601 as soon as any
+ # requested STRING ID does not exist, which is the normal state during
+ # an initial or incremental ingestion. Read the lightweight hash
+ # projection once and filter it in memory.
+ #
+ # getVertices() in this version of pyTigerGraph does not support an
+ # offset parameter, and an unbounded call hits TigerGraph's 4 MB REST
+ # limit (REST-4000) on large vertex sets. Use a GSQL interpreted query
+ # with LIMIT/OFFSET for proper pagination without that constraint.
+ _PAGE = 10_000
+ result: dict[str, str] = {}
+ offset = 0
+ while True:
+ query = (
+ f"INTERPRET QUERY() FOR GRAPH {self.graphname} {{\n"
+ f" verts = {{{vertex_type}.*}};\n"
+ f" res = SELECT v FROM verts:v\n"
+ f" ORDER BY v.content_hash ASC\n"
+ f" LIMIT {_PAGE} OFFSET {offset};\n"
+ f" PRINT res[res.content_hash];\n"
+ f"}}"
+ )
+ response = self.conn.runInterpretedQuery(query) or []
+ page = response[0].get("res", []) if response else []
+ for vertex in page:
+ # TG prefixes the attribute with the result-set alias:
+ # "res.content_hash" rather than plain "content_hash".
+ result[str(vertex.get("v_id", ""))] = str(
+ (vertex.get("attributes") or {}).get("res.content_hash") or ""
+ )
+ if len(page) < _PAGE:
+ break
+ offset += _PAGE
+ return result
+
+ def _all_existing_hashes(self) -> dict[str, str]:
+ return self._all_content_hashes("JiraIssue")
+
+ def _all_existing_comment_hashes(self) -> dict[str, str]:
+ return self._all_content_hashes("JiraComment")
+
+ def _has_unprocessed_chunks(self) -> bool:
+ """True when at least one DocumentChunk has epoch_processed=0.
+
+ Called only when this sync loaded no new content (documents_loaded=0,
+ comments_deleted=0). In that case any epoch_processed=0 chunk is a
+ survivor from a prior ECC rebuild that failed before it could embed
+ everything. We signal rebuild_required so the next ECC run can
+ complete the job without the user having to force a re-ingest.
+ """
+ try:
+ hits = self.conn.getVertices(
+ "DocumentChunk",
+ where="epoch_processed=0",
+ select="epoch_processed",
+ limit=1,
+ ) or []
+ logger.info(
+ "_has_unprocessed_chunks graph=%s: found %d chunk(s) with epoch_processed=0",
+ self.graphname,
+ len(hits),
+ )
+ return len(hits) > 0
+ except Exception as exc:
+ logger.warning("_has_unprocessed_chunks check failed: %s", exc)
+ return False
+
+ def _prepare_legacy_comment_migration(self) -> None:
+ if self.source.sync.migrating_legacy_comments:
+ return
+ documents = self.conn.getVertices("Document", select="id") or []
+ has_legacy_comments = any(
+ ":comment-doc:" in str(document.get("v_id") or "")
+ for document in documents
+ )
+ if not has_legacy_comments:
+ return
+ self.source.sync.migrating_legacy_comments = True
+ self.source.sync.checkpoint = None
+ self.store.update_runtime_state(self.graphname, self.source)
+
+ def _existing_hashes(self, issue_ids: list[str]) -> dict[str, str]:
+ if not issue_ids:
+ return {}
+ requested = set(issue_ids)
+ return {
+ issue_id: content_hash
+ for issue_id, content_hash in self._all_existing_hashes().items()
+ if issue_id in requested
+ }
+
+ def _iter_pages_pipelined(self) -> Iterator[list[dict[str, Any]]]:
+ """Yield Jira issue pages pre-fetched by a background thread.
+
+ The background thread requests the next page from the Jira API while
+ the main thread is writing the current page to TigerGraph, overlapping
+ network I/O with graph I/O. A queue capacity of 2 bounds memory: at
+ most two extra pages are held in RAM at any moment.
+ """
+ page_queue: queue.Queue[tuple[str, Any]] = queue.Queue(maxsize=2)
+ error_holder: list[Exception] = []
+
+ def _fetch() -> None:
+ try:
+ for page in self.client.iter_issue_pages():
+ page_queue.put(("page", page))
+ except Exception as exc: # noqa: BLE001
+ error_holder.append(exc)
+ finally:
+ page_queue.put(("done", None))
+
+ thread = threading.Thread(
+ target=_fetch, daemon=True, name="jira-page-fetcher"
+ )
+ thread.start()
+ try:
+ while True:
+ kind, value = page_queue.get()
+ if kind == "done":
+ if error_holder:
+ raise error_holder[0]
+ break
+ yield value
+ finally:
+ thread.join(timeout=60)
+
+ def _upsert_issue_records(
+ self,
+ mapped: list[MappedIssue],
+ existing_hashes: dict[str, str],
+ existing_comment_hashes: dict[str, str],
+ ) -> tuple[list[MappedIssue], list[MappedComment], int]:
+ """Write graph vertices/edges for one page; return items to embed later.
+
+ Fact chunks and comment chunks are written by the caller in batches
+ via _finalize_batch().
+
+ Returns (changed_documents, changed_comments, comments_deleted).
+ """
+ changed_documents = self._changed_documents(mapped, existing_hashes)
+ mapped_comments = [
+ comment
+ for item in mapped
+ for comment in item.comments
+ ]
+ changed_comments = self._changed_comments(
+ mapped_comments,
+ existing_comment_hashes,
+ )
+ comments_deleted = 0
+ for item in mapped:
+ comments_deleted += self._reconcile_issue_comments(
+ item,
+ existing_comment_hashes,
+ )
+ self._reconcile_issue_changes(item)
+
+ all_vertices: dict[tuple[str, str], VertexRecord] = {}
+ all_edges: list[EdgeRecord] = []
+ for item in mapped:
+ for vertex in item.vertices:
+ key = (vertex.vertex_type, vertex.vertex_id)
+ previous = all_vertices.get(key)
+ if (
+ previous
+ and previous.attributes.keys() - vertex.attributes.keys()
+ ):
+ # Keep richer attributes if a linked-issue placeholder
+ # arrives after a full issue record in the same page.
+ continue
+ all_vertices[key] = vertex
+ all_edges.extend(item.edges)
+ if item.issue_vertex_id in existing_hashes:
+ self._delete_current_edges(item.issue_vertex_id)
+ for change_vertex_id in item.change_vertex_ids:
+ self._delete_change_author_edges(change_vertex_id)
+ for comment in item.comments:
+ if comment.comment_vertex_id in existing_comment_hashes:
+ self._delete_current_comment_edges(
+ comment.comment_vertex_id
+ )
+
+ # Do not advance a changed item's content hash until its document load
+ # succeeds. A failed batch is therefore safe to replay after restart.
+ def strip_content_hash(vertex_type: str, vertex_id: str) -> None:
+ key = (vertex_type, vertex_id)
+ vertex = all_vertices.get(key)
+ if vertex is not None:
+ attributes = dict(vertex.attributes)
+ attributes.pop("content_hash", None)
+ all_vertices[key] = VertexRecord(
+ vertex.vertex_type,
+ vertex.vertex_id,
+ attributes,
+ )
+
+ for item in changed_documents:
+ strip_content_hash("JiraIssue", item.issue_vertex_id)
+ for comment in changed_comments:
+ strip_content_hash("JiraComment", comment.comment_vertex_id)
+
+ self._upsert_records(list(all_vertices.values()), all_edges)
+ return changed_documents, changed_comments, comments_deleted
+
+ def _finalize_batch(
+ self,
+ changed_documents: list[MappedIssue],
+ changed_comments: list[MappedComment],
+ existing_hashes: dict[str, str],
+ existing_comment_hashes: dict[str, str],
+ ) -> None:
+ """Write issue fact chunks and embed comment chunks for a batch.
+
+ Called after every DOC_LOAD_BATCH_PAGES pages (and once at the end of
+ the sync). Facts are written by the connector, one short chunk per
+ record, description piece, and change, instead of one document blob.
+ """
+ if changed_documents:
+ self._write_issue_facts(changed_documents, existing_hashes)
+ if changed_comments:
+ self._upsert_comment_chunks(changed_comments)
+ self._embed_chunks(
+ [
+ chunk
+ for comment in changed_comments
+ for chunk in comment.chunks
+ ]
+ )
+ for comment in changed_comments:
+ self.conn.upsertVertex(
+ "JiraComment",
+ comment.comment_vertex_id,
+ attributes={"content_hash": comment.content_hash},
+ )
+ existing_comment_hashes[
+ comment.comment_vertex_id
+ ] = comment.content_hash
+
+ def _upsert_issue_page(
+ self,
+ mapped: list[MappedIssue],
+ existing_hashes: dict[str, str],
+ existing_comment_hashes: dict[str, str],
+ ) -> tuple[int, int]:
+ changed_documents = self._changed_documents(mapped, existing_hashes)
+ mapped_comments = [
+ comment
+ for item in mapped
+ for comment in item.comments
+ ]
+ changed_comments = self._changed_comments(
+ mapped_comments,
+ existing_comment_hashes,
+ )
+ comments_deleted = 0
+ for item in mapped:
+ comments_deleted += self._reconcile_issue_comments(
+ item,
+ existing_comment_hashes,
+ )
+ self._reconcile_issue_changes(item)
+
+ all_vertices: dict[tuple[str, str], VertexRecord] = {}
+ all_edges: list[EdgeRecord] = []
+ for item in mapped:
+ for vertex in item.vertices:
+ key = (vertex.vertex_type, vertex.vertex_id)
+ previous = all_vertices.get(key)
+ if (
+ previous
+ and previous.attributes.keys() - vertex.attributes.keys()
+ ):
+ # Keep richer attributes if a linked-issue placeholder
+ # arrives after a full issue record in the same page.
+ continue
+ all_vertices[key] = vertex
+ all_edges.extend(item.edges)
+ if item.issue_vertex_id in existing_hashes:
+ self._delete_current_edges(item.issue_vertex_id)
+ for change_vertex_id in item.change_vertex_ids:
+ self._delete_change_author_edges(change_vertex_id)
+ for comment in item.comments:
+ if comment.comment_vertex_id in existing_comment_hashes:
+ self._delete_current_comment_edges(
+ comment.comment_vertex_id
+ )
+
+ # Do not advance a changed issue's content hash until its document load
+ # succeeds. A failed page is therefore safe to replay after restart.
+ def strip_content_hash(vertex_type: str, vertex_id: str) -> None:
+ key = (vertex_type, vertex_id)
+ vertex = all_vertices.get(key)
+ if vertex is not None:
+ attributes = dict(vertex.attributes)
+ attributes.pop("content_hash", None)
+ all_vertices[key] = VertexRecord(
+ vertex.vertex_type,
+ vertex.vertex_id,
+ attributes,
+ )
+
+ for item in changed_documents:
+ strip_content_hash("JiraIssue", item.issue_vertex_id)
+ for comment in changed_comments:
+ strip_content_hash("JiraComment", comment.comment_vertex_id)
+
+ self._upsert_records(list(all_vertices.values()), all_edges)
+
+ documents_to_load: list[MappedIssue | MappedComment] = [
+ *changed_documents,
+ *changed_comments,
+ ]
+ if documents_to_load:
+ if changed_documents:
+ self._write_issue_facts(changed_documents, existing_hashes)
+ if changed_comments:
+ self._upsert_comment_chunks(changed_comments)
+ self._embed_chunks(
+ [
+ chunk
+ for comment in changed_comments
+ for chunk in comment.chunks
+ ]
+ )
+ for comment in changed_comments:
+ self.conn.upsertVertex(
+ "JiraComment",
+ comment.comment_vertex_id,
+ attributes={"content_hash": comment.content_hash},
+ )
+ existing_comment_hashes[
+ comment.comment_vertex_id
+ ] = comment.content_hash
+
+ return len(documents_to_load), comments_deleted
+
+ def _upsert_records(
+ self,
+ vertices: list[VertexRecord],
+ edges: list[EdgeRecord],
+ ) -> None:
+ # Vertices must exist before their edge batches are applied. Keeping
+ # payloads bounded avoids REST request-size failures on large projects.
+ for records, are_edges in ((vertices, False), (edges, True)):
+ for start in range(0, len(records), UPSERT_BATCH_SIZE):
+ batch = records[start : start + UPSERT_BATCH_SIZE]
+ payload = _upsert_payload(
+ [] if are_edges else batch,
+ batch if are_edges else [],
+ )
+ result = self.conn.upsertData(json.dumps(payload))
+ if isinstance(result, dict) and (
+ result.get("skipped_vertices") or result.get("skipped_edges")
+ ):
+ raise RuntimeError(
+ "TigerGraph rejected part of the Jira graph upsert"
+ )
+
+ def _changed_documents(
+ self,
+ mapped: list[MappedIssue],
+ existing: dict[str, str],
+ ) -> list[MappedIssue]:
+ changed: list[MappedIssue] = []
+ for item in mapped:
+ if existing.get(item.issue_vertex_id) == item.content_hash:
+ continue
+ if item.issue_vertex_id in existing:
+ self._delete_issue_search_content(item)
+ changed.append(item)
+ return changed
+
+ def _changed_comments(
+ self,
+ comments: list[MappedComment],
+ existing: dict[str, str],
+ ) -> list[MappedComment]:
+ changed: list[MappedComment] = []
+ for comment in comments:
+ if existing.get(comment.comment_vertex_id) == comment.content_hash:
+ continue
+ if comment.comment_vertex_id in existing:
+ self._delete_comment_content(
+ comment.issue_vertex_id,
+ comment.comment_vertex_id,
+ )
+ changed.append(comment)
+ return changed
+
+ def _issue_comment_ids(self, issue_vertex_id: str) -> set[str]:
+ try:
+ edges = self.conn.getEdges(
+ "JiraIssue",
+ issue_vertex_id,
+ f"reverse_{JIRA_COMMENT_ISSUE_EDGE}",
+ ) or []
+ except Exception as exc:
+ if "is not a valid vertex id" in str(exc):
+ return set()
+ raise
+ return {
+ str(edge.get("to_id"))
+ for edge in edges
+ if edge.get("to_id") is not None
+ }
+
+ def _reconcile_issue_comments(
+ self,
+ issue: MappedIssue,
+ existing_hashes: dict[str, str],
+ ) -> int:
+ current_ids = {
+ comment.comment_vertex_id for comment in issue.comments
+ }
+ stale_ids = self._issue_comment_ids(issue.issue_vertex_id) - current_ids
+ for comment_vertex_id in stale_ids:
+ self._delete_comment(
+ issue.issue_vertex_id,
+ comment_vertex_id,
+ )
+ existing_hashes.pop(comment_vertex_id, None)
+ return len(stale_ids)
+
+ def _reconcile_issue_changes(self, issue: MappedIssue) -> None:
+ current_ids = set(issue.change_vertex_ids)
+ for change_vertex_id in (
+ self._issue_change_ids(issue.issue_vertex_id) - current_ids
+ ):
+ self.conn.delVerticesById("JiraChange", [change_vertex_id])
+
+ def _issue_change_ids(self, issue_vertex_id: str) -> set[str]:
+ try:
+ edges = self.conn.getEdges(
+ "JiraIssue",
+ issue_vertex_id,
+ JIRA_CHANGE_EDGE,
+ ) or []
+ except Exception as exc:
+ if "is not a valid vertex id" in str(exc):
+ return set()
+ raise
+ return {
+ str(edge.get("to_id"))
+ for edge in edges
+ if edge.get("to_id") is not None
+ }
+
+ def _delete_change_author_edges(self, change_vertex_id: str) -> None:
+ try:
+ self.conn.delEdges(
+ "JiraChange",
+ change_vertex_id,
+ JIRA_CHANGE_AUTHOR_EDGE,
+ )
+ except Exception as exc:
+ if "is not a valid vertex id" in str(exc):
+ return
+ raise
+
+ def _delete_current_edges(self, issue_vertex_id: str) -> None:
+ for edge_type in CURRENT_STATE_EDGES:
+ self.conn.delEdges("JiraIssue", issue_vertex_id, edge_type)
+
+ def _delete_current_comment_edges(
+ self,
+ comment_vertex_id: str,
+ ) -> None:
+ for edge_type in COMMENT_STATE_EDGES:
+ self.conn.delEdges("JiraComment", comment_vertex_id, edge_type)
+
+ def _delete_comment(
+ self,
+ issue_vertex_id: str,
+ comment_vertex_id: str,
+ ) -> None:
+ self._delete_comment_content(issue_vertex_id, comment_vertex_id)
+ self.conn.delVerticesById("JiraComment", [comment_vertex_id])
+
+ def _comment_chunk_ids(self, comment_vertex_id: str) -> list[str]:
+ try:
+ edges = self.conn.getEdges(
+ "JiraComment",
+ comment_vertex_id,
+ "reverse_CONTAINS_ENTITY",
+ ) or []
+ except Exception as exc:
+ if "is not a valid vertex id" in str(exc):
+ return []
+ raise
+ return [
+ str(edge["to_id"])
+ for edge in edges
+ if edge.get("to_type") == "DocumentChunk"
+ and edge.get("to_id") is not None
+ ]
+
+ def _delete_comment_content(
+ self,
+ issue_vertex_id: str,
+ comment_vertex_id: str,
+ ) -> None:
+ direct_chunk_ids = self._comment_chunk_ids(comment_vertex_id)
+ if direct_chunk_ids:
+ get_embedding_store(
+ graphname=self.graphname
+ ).remove_embeddings(ids=direct_chunk_ids)
+ self.conn.delVerticesById("DocumentChunk", direct_chunk_ids)
+ self.conn.delVerticesById("Content", direct_chunk_ids)
+
+ # Remove records created by the former Document -> ECC chunking path.
+ comment_id = comment_vertex_id.rsplit(":comment:", 1)[-1]
+ legacy_document_id = f"{issue_vertex_id}:comment-doc:{comment_id}"
+ self._delete_document_chunks(legacy_document_id)
+ self.conn.delVerticesById("Document", [legacy_document_id])
+ self.conn.delVerticesById("Content", [legacy_document_id])
+
+ def _delete_document_chunks(self, document_id: str) -> None:
+ try:
+ edges = self.conn.getEdges("Document", document_id, "HAS_CHILD") or []
+ except Exception as exc:
+ if "is not a valid vertex id" in str(exc):
+ return
+ raise
+ chunk_ids = [
+ str(edge.get("to_id"))
+ for edge in edges
+ if edge.get("to_id") is not None
+ ]
+ if not chunk_ids:
+ return
+ get_embedding_store(graphname=self.graphname).remove_embeddings(ids=chunk_ids)
+ self.conn.delVerticesById("DocumentChunk", chunk_ids)
+ self.conn.delVerticesById("Content", chunk_ids)
+
+ def _write_issue_facts(
+ self,
+ issues: list[MappedIssue],
+ existing_hashes: dict[str, str],
+ ) -> None:
+ self._upsert_issue_facts(issues)
+ self._embed_chunks(
+ [fact.chunk for issue in issues for fact in issue.facts]
+ )
+ for issue in issues:
+ self.conn.upsertVertex(
+ "JiraIssue",
+ issue.issue_vertex_id,
+ attributes={"content_hash": issue.content_hash},
+ )
+ existing_hashes[issue.issue_vertex_id] = issue.content_hash
+
+ def _delete_issue_search_content(self, issue: MappedIssue) -> None:
+ chunk_ids = self._issue_fact_chunk_ids(issue.issue_vertex_id)
+ if chunk_ids:
+ get_embedding_store(
+ graphname=self.graphname
+ ).remove_embeddings(ids=chunk_ids)
+ self.conn.delVerticesById("DocumentChunk", chunk_ids)
+ self.conn.delVerticesById("Content", chunk_ids)
+ document_id = issue.legacy_document_id.lower()
+ self._delete_document_chunks(document_id)
+ self.conn.delVerticesById("Document", [document_id])
+ self.conn.delVerticesById("Content", [document_id])
+
+ def _issue_fact_chunk_ids(self, issue_vertex_id: str) -> list[str]:
+ try:
+ edges = self.conn.getEdges(
+ "JiraIssue",
+ issue_vertex_id,
+ "reverse_CONTAINS_ENTITY",
+ ) or []
+ except Exception as exc:
+ if "is not a valid vertex id" in str(exc):
+ return []
+ raise
+ return [
+ str(edge["to_id"])
+ for edge in edges
+ if edge.get("to_type") == "DocumentChunk"
+ and edge.get("to_id") is not None
+ and ":fact:" in str(edge["to_id"])
+ ]
+
+ def _upsert_issue_facts(self, issues: list[MappedIssue]) -> None:
+ epoch_added = int(time.time())
+ vertices: list[VertexRecord] = []
+ edges: list[EdgeRecord] = []
+ for issue in issues:
+ previous_chunk_id: str | None = None
+ for fact in issue.facts:
+ chunk = fact.chunk
+ vertices.extend(
+ self._chunk_vertices(chunk, epoch_added, "jira_issue")
+ )
+ edges.extend(
+ self._chunk_edges(
+ chunk.chunk_id,
+ issue.issue_vertex_id,
+ fact.entities,
+ previous_chunk_id,
+ )
+ )
+ previous_chunk_id = chunk.chunk_id
+ self._upsert_records(vertices, edges)
+
+ def _chunk_vertices(
+ self,
+ chunk: MappedChunk,
+ epoch_added: int,
+ ctype: str,
+ ) -> list[VertexRecord]:
+ return [
+ VertexRecord(
+ "DocumentChunk",
+ chunk.chunk_id,
+ {
+ "idx": chunk.index,
+ "epoch_added": epoch_added,
+ "epoch_processing": 0,
+ "epoch_processed": 0,
+ },
+ ),
+ VertexRecord(
+ "Content",
+ chunk.chunk_id,
+ {
+ "ctype": ctype,
+ "text": chunk.text,
+ "epoch_added": epoch_added,
+ },
+ ),
+ ]
+
+ def _chunk_edges(
+ self,
+ chunk_id: str,
+ issue_vertex_id: str,
+ entities: tuple[tuple[str, str], ...],
+ previous_chunk_id: str | None,
+ ) -> list[EdgeRecord]:
+ edges = [
+ EdgeRecord(
+ "DocumentChunk",
+ chunk_id,
+ "HAS_CONTENT",
+ "Content",
+ chunk_id,
+ ),
+ EdgeRecord(
+ "DocumentChunk",
+ chunk_id,
+ "CONTAINS_ENTITY",
+ "JiraIssue",
+ issue_vertex_id,
+ ),
+ ]
+ for vertex_type, vertex_id in entities:
+ edges.append(
+ EdgeRecord(
+ "DocumentChunk",
+ chunk_id,
+ "CONTAINS_ENTITY",
+ vertex_type,
+ vertex_id,
+ )
+ )
+ if previous_chunk_id:
+ edges.append(
+ EdgeRecord(
+ "DocumentChunk",
+ chunk_id,
+ "IS_AFTER",
+ "DocumentChunk",
+ previous_chunk_id,
+ )
+ )
+ return edges
+
+ def _upsert_comment_chunks(
+ self,
+ comments: list[MappedComment],
+ ) -> None:
+ epoch_added = int(time.time())
+ vertices: list[VertexRecord] = []
+ edges: list[EdgeRecord] = []
+ for comment in comments:
+ previous_chunk_id: str | None = None
+ for chunk in comment.chunks:
+ vertices.extend(
+ self._chunk_vertices(chunk, epoch_added, "jira_comment")
+ )
+ edges.extend(
+ self._chunk_edges(
+ chunk.chunk_id,
+ comment.issue_vertex_id,
+ (("JiraComment", comment.comment_vertex_id),),
+ previous_chunk_id,
+ )
+ )
+ previous_chunk_id = chunk.chunk_id
+ self._upsert_records(vertices, edges)
+
+ def _embed_chunks(self, chunks: list[MappedChunk]) -> None:
+ if not chunks:
+ return
+ store = get_embedding_store(graphname=self.graphname)
+
+ async def embed_batches() -> None:
+ # Cap at 5 concurrent aadd_embeddings calls. Each call makes ~32
+ # sequential Gemini requests; 20 concurrent was causing traffic
+ # spikes that trigger Gemini 500 INTERNAL (server overload).
+ # 5 concurrent × ~32 requests = ~160 in-flight, safe for the API.
+ sem = asyncio.Semaphore(5)
+
+ async def _run_batch(batch: list) -> None:
+ async with sem:
+ await store.aadd_embeddings(
+ [(chunk.text, []) for chunk in batch],
+ [
+ {
+ "vertex_id": (
+ chunk.chunk_id,
+ "DocumentChunk",
+ )
+ }
+ for chunk in batch
+ ],
+ )
+
+ tasks = [
+ _run_batch(chunks[s : s + EMBEDDING_BATCH_SIZE])
+ for s in range(0, len(chunks), EMBEDDING_BATCH_SIZE)
+ ]
+ await asyncio.gather(*tasks)
+
+ asyncio.run(embed_batches())
+ processed_at = int(time.time())
+ for chunk in chunks:
+ self.conn.upsertVertex(
+ "DocumentChunk",
+ chunk.chunk_id,
+ attributes={"epoch_processed": processed_at},
+ )
diff --git a/graphrag/app/main.py b/graphrag/app/main.py
index cfa79aaa..30c01eb4 100644
--- a/graphrag/app/main.py
+++ b/graphrag/app/main.py
@@ -99,6 +99,7 @@ async def lifespan(app: FastAPI):
app.include_router(routers.queryai_router, prefix=PATH_PREFIX)
app.include_router(routers.ui_router, prefix=PATH_PREFIX)
app.include_router(routers.mcp_servers_router, prefix=PATH_PREFIX)
+app.include_router(routers.data_sources_router, prefix=PATH_PREFIX)
excluded_metrics_paths = ("/docs", "/openapi.json", "/metrics")
diff --git a/graphrag/app/routers/__init__.py b/graphrag/app/routers/__init__.py
index 054cc955..d2785067 100644
--- a/graphrag/app/routers/__init__.py
+++ b/graphrag/app/routers/__init__.py
@@ -4,3 +4,4 @@
from .supportai import router as supportai_router
from .ui import router as ui_router
from .mcp_servers import router as mcp_servers_router
+from .data_sources import router as data_sources_router
diff --git a/graphrag/app/routers/data_sources.py b/graphrag/app/routers/data_sources.py
new file mode 100644
index 00000000..8be0fa64
--- /dev/null
+++ b/graphrag/app/routers/data_sources.py
@@ -0,0 +1,578 @@
+"""Graph-scoped enterprise data-source management."""
+
+from __future__ import annotations
+
+import logging
+import threading
+import uuid
+from datetime import datetime, timezone
+from typing import Annotated, Any
+
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
+from fastapi.security import HTTPBasicCredentials
+
+from common.config import validate_graphname
+from common.db.schema_utils import apply_proposal
+from common.utils.graph_locks import (
+ acquire_graph_lock,
+ get_current_operation,
+ release_graph_lock,
+)
+from connectors.jira.client import JiraAPIError, JiraCloudClient
+from connectors.jira.config import PROJECT_KEY_RE, JiraDataSource, JiraScope
+from connectors.jira.schema import (
+ jira_schema_proposal,
+ jira_schema_status,
+)
+from connectors.jira.state import JiraSourceStore
+from connectors.jira.sync import JiraSyncService
+
+logger = logging.getLogger(__name__)
+router = APIRouter(tags=["Data Sources"])
+route_prefix = "/ui"
+store = JiraSourceStore()
+
+_sync_state: dict[str, dict[str, Any]] = {}
+_sync_state_lock = threading.Lock()
+_jira_tokens: dict[tuple[str, str], str] = {}
+_jira_tokens_lock = threading.Lock()
+
+
+def _ui_basic_auth():
+ from routers.ui import ui_basic_auth
+
+ return ui_basic_auth
+
+
+def _require_access(
+ graphname: str,
+ auth: tuple[list[str], HTTPBasicCredentials],
+) -> HTTPBasicCredentials:
+ validate_graphname(graphname)
+ graphs, credentials = auth
+ if graphname not in graphs:
+ raise HTTPException(status_code=403, detail="Graph access is required.")
+ from routers.ui import _require_roles
+
+ _require_roles(credentials, {"superuser", "globaldesigner"})
+ return credentials
+
+
+def _source_or_404(graphname: str, source_id: str) -> JiraDataSource:
+ try:
+ return store.get(graphname, source_id)
+ except KeyError:
+ raise HTTPException(status_code=404, detail="Data source not found.")
+
+
+def _source_with_credentials(
+ graphname: str,
+ source_id: str,
+) -> JiraDataSource:
+ source = _source_or_404(graphname, source_id)
+ with _jira_tokens_lock:
+ token = _jira_tokens.get((graphname, source_id))
+ token = token or source.connection.api_token
+ if not token:
+ raise HTTPException(
+ status_code=409,
+ detail=(
+ "Jira API token is not available. Test the connection again "
+ "and provide the token."
+ ),
+ )
+ resolved = source.model_copy(deep=True)
+ resolved.connection.api_token = token
+ return resolved
+
+
+def _jira_error(exc: JiraAPIError) -> HTTPException:
+ if exc.status_code == 404:
+ detail = (
+ "Jira REST API was not found at the configured Site URL. "
+ "Use the Jira tenant base URL, for example "
+ "https://your-company.atlassian.net."
+ )
+ elif exc.status_code == 401:
+ detail = (
+ "Jira rejected the credentials. Use the Atlassian account email "
+ "that owns the API token and a valid Jira API token."
+ )
+ elif exc.status_code == 403:
+ detail = (
+ "Jira accepted the credentials but denied access. Verify that the "
+ "account can access Jira and browse the required projects."
+ )
+ else:
+ detail = str(exc)
+ status_code = 400 if 400 <= exc.status_code < 500 else 502
+ return HTTPException(status_code=status_code, detail=detail)
+
+
+def _connection(request: Request, graphname: str):
+ authorization = request.headers.get("Authorization")
+ if not authorization:
+ raise HTTPException(status_code=401, detail="Missing Authorization header.")
+ from routers.ui import ws_basic_auth
+
+ _, conn = ws_basic_auth(authorization, graphname)
+ return conn
+
+
+def _set_sync_state(sync_run_id: str, **updates: Any) -> None:
+ with _sync_state_lock:
+ current = dict(_sync_state.get(sync_run_id) or {})
+ current.update(updates)
+ _sync_state[sync_run_id] = current
+
+
+@router.get(f"{route_prefix}/{{graphname}}/data-sources")
+def list_data_sources(
+ graphname: str,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+):
+ _require_access(graphname, auth)
+ return {"sources": store.list(graphname)}
+
+
+@router.put(f"{route_prefix}/{{graphname}}/data-sources/{{source_id}}")
+def save_data_source(
+ graphname: str,
+ source_id: str,
+ source: JiraDataSource,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+):
+ _require_access(graphname, auth)
+ if source.id != source_id:
+ raise HTTPException(
+ status_code=400,
+ detail="Path source id must match the request body.",
+ )
+ try:
+ saved = store.upsert(graphname, source)
+ except ValueError as exc:
+ raise HTTPException(status_code=400, detail=str(exc))
+ return {"source": store.redact(saved)}
+
+
+@router.delete(f"{route_prefix}/{{graphname}}/data-sources/{{source_id}}")
+def delete_data_source(
+ graphname: str,
+ source_id: str,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+):
+ _require_access(graphname, auth)
+ try:
+ store.delete(graphname, source_id)
+ except KeyError:
+ raise HTTPException(status_code=404, detail="Data source not found.")
+ with _jira_tokens_lock:
+ _jira_tokens.pop((graphname, source_id), None)
+ return {"status": "deleted"}
+
+
+@router.post(f"{route_prefix}/{{graphname}}/data-sources/{{source_id}}/test")
+def test_data_source(
+ graphname: str,
+ source_id: str,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+ candidate: JiraDataSource | None = None,
+):
+ _require_access(graphname, auth)
+ if candidate is None or not candidate.connection.api_token.strip():
+ raise HTTPException(
+ status_code=400,
+ detail="Enter the Jira API token before testing the connection.",
+ )
+ if candidate.id != source_id:
+ raise HTTPException(
+ status_code=400,
+ detail="Path source id must match the request body.",
+ )
+ token = candidate.connection.api_token.strip()
+ try:
+ with JiraCloudClient(candidate) as client:
+ account = client.myself()
+ cloud_id = candidate.connection.cloud_id or client.cloud_id()
+ except JiraAPIError as exc:
+ raise _jira_error(exc)
+ try:
+ source = store.upsert(graphname, candidate)
+ except ValueError as exc:
+ raise HTTPException(status_code=400, detail=str(exc))
+ with _jira_tokens_lock:
+ _jira_tokens[(graphname, source_id)] = token
+ source.connection.cloud_id = cloud_id
+ source.sync.last_tested_at = datetime.now(timezone.utc)
+ store.update_runtime_state(graphname, source)
+ return {
+ "status": "connected",
+ "account": {
+ "account_id": account.get("accountId"),
+ "display_name": account.get("displayName"),
+ },
+ "cloud_id": cloud_id,
+ "tested_at": source.sync.last_tested_at.isoformat(),
+ "source": store.redact(source),
+ }
+
+
+@router.get(f"{route_prefix}/{{graphname}}/data-sources/{{source_id}}/projects")
+def list_jira_projects(
+ graphname: str,
+ source_id: str,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+ project_key: str | None = None,
+):
+ _require_access(graphname, auth)
+ source = _source_with_credentials(graphname, source_id)
+ if source.sync.last_tested_at is None:
+ raise HTTPException(
+ status_code=409,
+ detail="Test this Jira connection before loading projects.",
+ )
+ normalized_project_key = project_key.strip().upper() if project_key else None
+ if normalized_project_key and not PROJECT_KEY_RE.fullmatch(
+ normalized_project_key
+ ):
+ raise HTTPException(
+ status_code=400,
+ detail=(
+ "Project key must start with a letter and contain only letters, "
+ "numbers, or underscores."
+ ),
+ )
+ try:
+ with JiraCloudClient(source) as client:
+ projects = (
+ [client.project(normalized_project_key)]
+ if normalized_project_key
+ else client.projects()
+ )
+ except JiraAPIError as exc:
+ if normalized_project_key and exc.status_code == 404:
+ raise HTTPException(
+ status_code=400,
+ detail=(
+ f"Jira project {normalized_project_key} was not found or "
+ "is not visible to this account."
+ ),
+ )
+ raise _jira_error(exc)
+ return {
+ "projects": [
+ {
+ "id": project.get("id"),
+ "key": project.get("key"),
+ "name": project.get("name"),
+ }
+ for project in projects
+ ]
+ }
+
+
+@router.post(
+ f"{route_prefix}/{{graphname}}/data-sources/{{source_id}}/issues/count"
+)
+def preview_jira_issue_count(
+ graphname: str,
+ source_id: str,
+ scope: JiraScope,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+):
+ """Estimate how many Jira issues match the unsaved project scope."""
+ _require_access(graphname, auth)
+ if not scope.project_keys:
+ raise HTTPException(
+ status_code=400,
+ detail="Select at least one Jira project before previewing tickets.",
+ )
+ source = _source_with_credentials(graphname, source_id)
+ if source.sync.last_tested_at is None:
+ raise HTTPException(
+ status_code=409,
+ detail="Test this Jira connection before previewing tickets.",
+ )
+ preview_source = source.model_copy(deep=True)
+ preview_source.scope = scope
+ try:
+ with JiraCloudClient(preview_source) as client:
+ count = client.approximate_issue_count()
+ except JiraAPIError as exc:
+ raise _jira_error(exc)
+ return {"count": count, "approximate": True}
+
+
+@router.get(f"{route_prefix}/{{graphname}}/data-sources/jira/schema")
+def get_jira_schema_status(
+ graphname: str,
+ request: Request,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+):
+ _require_access(graphname, auth)
+ conn = _connection(request, graphname)
+ return jira_schema_status(conn)
+
+
+@router.post(f"{route_prefix}/{{graphname}}/data-sources/jira/schema/install")
+def install_jira_schema(
+ graphname: str,
+ request: Request,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+):
+ _require_access(graphname, auth)
+ sources = store.load(graphname).sources
+ if not sources:
+ raise HTTPException(
+ status_code=409,
+ detail="Save a Jira connection before installing its schema.",
+ )
+ if not any(source.sync.last_tested_at for source in sources):
+ raise HTTPException(
+ status_code=409,
+ detail="Test a Jira connection before installing its schema.",
+ )
+ if not any(
+ source.sync.last_tested_at and source.scope.project_keys
+ for source in sources
+ ):
+ raise HTTPException(
+ status_code=409,
+ detail="Select and save at least one Jira project before installing its schema.",
+ )
+ operation = "install_jira_schema"
+ if not acquire_graph_lock(graphname, operation):
+ current = get_current_operation(graphname) or "another operation"
+ raise HTTPException(
+ status_code=409,
+ detail=f"Graph '{graphname}' is busy with '{current}'.",
+ )
+ try:
+ conn = _connection(request, graphname)
+ before = jira_schema_status(conn)
+ if before["status"] == "not_initialized":
+ raise HTTPException(
+ status_code=409,
+ detail=(
+ "Initialize the knowledge graph before installing "
+ "the Jira connector schema."
+ ),
+ )
+ if before["status"] == "conflict":
+ raise HTTPException(
+ status_code=409,
+ detail={
+ "message": "Existing graph schema conflicts with Jira schema.",
+ "conflicts": before["conflicts"],
+ },
+ )
+
+ result = apply_proposal(conn, graphname, jira_schema_proposal())
+ if result.get("status") == "error":
+ raise HTTPException(
+ status_code=500,
+ detail=result.get("error") or "Jira schema installation failed.",
+ )
+ retrievers = result.get("retrievers") or {}
+ if retrievers.get("status") == "error":
+ raise HTTPException(
+ status_code=500,
+ detail=(
+ "Jira schema was applied, but retriever installation failed: "
+ f"{retrievers.get('error', 'unknown error')}"
+ ),
+ )
+
+ # The connection used to apply the SCHEMA_CHANGE JOB can retain the
+ # pre-migration schema snapshot. Verify through a fresh connection so
+ # a successful migration is not reported as a false 500.
+ verification_conn = _connection(request, graphname)
+ after = jira_schema_status(verification_conn)
+ if after["status"] != "installed":
+ logger.error(
+ "Jira schema verification failed for graph %s: %s",
+ graphname,
+ after,
+ )
+ raise HTTPException(
+ status_code=500,
+ detail={
+ "message": (
+ "Jira schema installation completed, but verification "
+ "did not find a compatible schema."
+ ),
+ "missing": after.get("missing"),
+ "conflicts": after.get("conflicts"),
+ },
+ )
+ return {
+ "status": "installed",
+ "schema": after,
+ "migration": {
+ "status": result.get("status"),
+ "statements": result.get("statements") or [],
+ "summary": result.get("summary") or {},
+ "retrievers": retrievers,
+ },
+ }
+ finally:
+ release_graph_lock(graphname, operation)
+
+
+@router.post(f"{route_prefix}/{{graphname}}/data-sources/{{source_id}}/sync")
+def start_sync(
+ graphname: str,
+ source_id: str,
+ request: Request,
+ background_tasks: BackgroundTasks,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+):
+ _require_access(graphname, auth)
+ source = _source_with_credentials(graphname, source_id)
+ if not source.enabled:
+ raise HTTPException(status_code=409, detail="Jira data source is disabled.")
+ if source.sync.last_tested_at is None:
+ raise HTTPException(
+ status_code=409,
+ detail="Test this Jira connection before synchronization.",
+ )
+ if not source.scope.project_keys:
+ raise HTTPException(
+ status_code=409,
+ detail="Select at least one Jira project before synchronization.",
+ )
+ authorization = request.headers.get("Authorization")
+ if not authorization:
+ raise HTTPException(status_code=401, detail="Missing Authorization header.")
+ conn = _connection(request, graphname)
+ schema = jira_schema_status(conn)
+ if schema["status"] != "installed":
+ raise HTTPException(
+ status_code=409,
+ detail=(
+ "Jira schema is not installed for this graph. "
+ "Install it from Data Sources before synchronization."
+ ),
+ )
+
+ operation = f"jira_sync:{source_id}"
+ if not acquire_graph_lock(graphname, operation):
+ current = get_current_operation(graphname) or "another operation"
+ raise HTTPException(
+ status_code=409,
+ detail=f"Graph '{graphname}' is busy with '{current}'.",
+ )
+
+ run_id = uuid.uuid4().hex
+ try:
+ _set_sync_state(
+ run_id,
+ run_id=run_id,
+ graphname=graphname,
+ source_id=source_id,
+ status="queued",
+ started_at=datetime.now(timezone.utc).isoformat(),
+ )
+ except Exception:
+ release_graph_lock(graphname, operation)
+ raise
+
+ def run() -> None:
+ result: dict[str, Any] | None = None
+ error: Exception | None = None
+ try:
+ _set_sync_state(run_id, status="running")
+ from routers.ui import ws_basic_auth
+
+ _, conn = ws_basic_auth(authorization, graphname)
+ source = _source_with_credentials(graphname, source_id)
+ result = JiraSyncService(
+ graphname,
+ source,
+ conn,
+ store=store,
+ ).run()
+ except Exception as exc:
+ error = exc
+ logger.exception(
+ "Jira data-source sync failed graph=%s source=%s",
+ graphname,
+ source_id,
+ )
+ finally:
+ # Release before publishing a terminal state. The UI starts the
+ # GraphRAG rebuild as soon as it observes "completed".
+ release_graph_lock(graphname, operation)
+
+ if error is not None:
+ _set_sync_state(
+ run_id,
+ status="failed",
+ completed_at=datetime.now(timezone.utc).isoformat(),
+ error=str(error)[:1000],
+ )
+ else:
+ _set_sync_state(
+ run_id,
+ status="completed",
+ completed_at=datetime.now(timezone.utc).isoformat(),
+ result=result,
+ )
+
+ try:
+ background_tasks.add_task(run)
+ except Exception:
+ release_graph_lock(graphname, operation)
+ raise
+ return {"status": "submitted", "run_id": run_id}
+
+
+@router.get(
+ f"{route_prefix}/{{graphname}}/data-sources/{{source_id}}/sync/{{run_id}}"
+)
+def get_sync_status(
+ graphname: str,
+ source_id: str,
+ run_id: str,
+ auth: Annotated[
+ tuple[list[str], HTTPBasicCredentials],
+ Depends(_ui_basic_auth()),
+ ],
+):
+ _require_access(graphname, auth)
+ with _sync_state_lock:
+ state = dict(_sync_state.get(run_id) or {})
+ if (
+ not state
+ or state.get("graphname") != graphname
+ or state.get("source_id") != source_id
+ ):
+ raise HTTPException(status_code=404, detail="Sync run not found.")
+ return state
diff --git a/graphrag/app/tools/generate_cypher.py b/graphrag/app/tools/generate_cypher.py
index d9199de4..b9f3dbac 100644
--- a/graphrag/app/tools/generate_cypher.py
+++ b/graphrag/app/tools/generate_cypher.py
@@ -13,6 +13,7 @@
# along with this program. If not, see