thanhnv
9b4dc01c1a
- Phase 9: Extend JiraConnectDialog with Project Knowledge config, mapping, and Sync Now button - Phase 10: Wire JiraSyncService to CanonicalAuditLogger for sync start/complete/fail events - Phase 11: Add retrieval regression suite with synthetic corpus and baseline metrics - Phase 14: Add comprehensive production guide (docs/jira-knowledge-guide.md) - Fix UI status label references (self.status -> self.conn_status) - Implement real sync trigger logic in UI using JiraSyncService
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""Jira knowledge synchronization service.
|
|
|
|
Orchestrates full and incremental sync of Jira issues into the local
|
|
canonical knowledge index. Reuses ``core.jira_tool`` for HTTP access and
|
|
the existing atomic-write / telemetry infrastructure for persistence and
|
|
observability.
|
|
|
|
Design invariants:
|
|
- Bounded batches: each sync page fetches at most ``_BATCH_SIZE`` issues.
|
|
- Idempotent upserts: re-syncing the same issue overwrites cleanly.
|
|
- Partial failure tolerance: one malformed issue does not abort the batch.
|
|
- Credential isolation: credentials are resolved per-call, never stored on
|
|
the service instance.
|
|
- Operational state: every sync updates the manifest with counts, timing,
|
|
and error category so operators can inspect health without reading logs.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from ...domain.jira_knowledge.canonical_issue import CanonicalJiraIssue, normalize_jira_issue
|
|
from ...domain.jira_knowledge.sync_state import SyncManifest, load_manifest, save_manifest
|
|
from ...mcp_servers.project_context.foundation import ProviderError
|
|
from .credential_resolver import JiraCredentialResolver, JiraCredentials
|
|
from .index_repository import JiraKnowledgeIndex
|
|
from .target_resolver import JiraTarget, JiraTargetResolver
|
|
|
|
_BATCH_SIZE = 50
|
|
_MAX_PAGES_PER_SYNC = 200
|
|
_JQL_FIELDS = (
|
|
"summary,status,assignee,priority,description,labels,components,"
|
|
"issuetype,updated,created,issuelinks"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SyncResult:
|
|
"""Outcome of one sync run."""
|
|
processed: int
|
|
failed: int
|
|
total_indexed: int
|
|
cursor: str
|
|
duration_seconds: float
|
|
error_category: str = ""
|
|
|
|
|
|
class JiraSyncService:
|
|
"""Full and incremental sync of Jira issues into the knowledge index.
|
|
|
|
The service is stateless between calls — all operational state lives in
|
|
the persisted manifest. This makes it safe to call from a scheduler,
|
|
a manual trigger, or a test harness interchangeably.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
target_resolver: JiraTargetResolver,
|
|
credential_resolver: JiraCredentialResolver,
|
|
index: Optional[JiraKnowledgeIndex] = None,
|
|
index_root: Optional[Any] = None,
|
|
) -> None:
|
|
self._target_resolver = target_resolver
|
|
self._credential_resolver = credential_resolver
|
|
self._index = index or JiraKnowledgeIndex(index_root=index_root)
|
|
|
|
def full_sync(self, identity: Any) -> SyncResult:
|
|
"""Paginated full sync of all issues in the identity's Jira project.
|
|
|
|
Clears the existing index before importing so deleted/inaccessible
|
|
issues are naturally removed. The manifest cursor is reset.
|
|
"""
|
|
return self._run_sync(identity, incremental=False)
|
|
|
|
def incremental_sync(self, identity: Any) -> SyncResult:
|
|
"""Fetch only issues updated since the last successful sync cursor.
|
|
|
|
Falls back to full sync when no cursor exists (first run).
|
|
"""
|
|
return self._run_sync(identity, incremental=True)
|
|
|
|
def _run_sync(self, identity: Any, *, incremental: bool) -> SyncResult:
|
|
start = time.monotonic()
|
|
target = self._target_resolver.resolve(identity)
|
|
creds = self._credential_resolver.resolve(identity)
|
|
manifest = load_manifest(self._index._root, target.cowork_project_id)
|
|
manifest.mark_attempt()
|
|
save_manifest(self._index._root, manifest)
|
|
|
|
# Emit audit event for sync start
|
|
try:
|
|
from ...infrastructure.telemetry.audit_logger import CanonicalAuditLogger
|
|
from ...config import CONFIG_DIR
|
|
logger = CanonicalAuditLogger(CONFIG_DIR / "audit")
|
|
logger.record(
|
|
kind="jira_knowledge.sync.started",
|
|
name=f"{target.cowork_project_id}:{target.jira_project_key}",
|
|
ok=True,
|
|
detail=f"mode={'incremental' if incremental else 'full'}",
|
|
agent_role="system"
|
|
)
|
|
except Exception:
|
|
pass # Audit failure must not break sync
|
|
|
|
# Fall back to full sync when no cursor exists.
|
|
if incremental and not manifest.sync_cursor:
|
|
incremental = False
|
|
|
|
try:
|
|
if not incremental:
|
|
self._index.clear(target.cowork_project_id)
|
|
|
|
config = {
|
|
"base_url": target.jira_base_url,
|
|
"email": creds.email,
|
|
"api_token": creds.api_token,
|
|
}
|
|
jql = f"project = {target.jira_project_key} ORDER BY updated ASC"
|
|
if incremental and manifest.sync_cursor:
|
|
jql = (
|
|
f"project = {target.jira_project_key} "
|
|
f"AND updated >= '{manifest.sync_cursor}' "
|
|
f"ORDER BY updated ASC"
|
|
)
|
|
|
|
processed, failed, latest_cursor = self._fetch_and_index(
|
|
config=config,
|
|
jql=jql,
|
|
project_id=target.cowork_project_id,
|
|
base_url=target.jira_base_url,
|
|
)
|
|
total_indexed = self._index.count(target.cowork_project_id)
|
|
duration = time.monotonic() - start
|
|
manifest.mark_success(
|
|
processed=processed,
|
|
failed=failed,
|
|
cursor=latest_cursor or manifest.sync_cursor,
|
|
duration=duration,
|
|
total_indexed=total_indexed,
|
|
)
|
|
save_manifest(self._index._root, manifest)
|
|
|
|
# Emit audit event for sync success
|
|
try:
|
|
from ...infrastructure.telemetry.audit_logger import CanonicalAuditLogger
|
|
from ...config import CONFIG_DIR
|
|
logger = CanonicalAuditLogger(CONFIG_DIR / "audit")
|
|
logger.record(
|
|
kind="jira_knowledge.sync.completed",
|
|
name=f"{target.cowork_project_id}:{target.jira_project_key}",
|
|
ok=True,
|
|
detail=f"processed={processed},failed={failed},duration={duration:.2f}s",
|
|
agent_role="system"
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
return SyncResult(
|
|
processed=processed,
|
|
failed=failed,
|
|
total_indexed=total_indexed,
|
|
cursor=latest_cursor or manifest.sync_cursor,
|
|
duration_seconds=round(duration, 2),
|
|
)
|
|
except ProviderError as exc:
|
|
duration = time.monotonic() - start
|
|
manifest.mark_failure(category=exc.code, failed=0)
|
|
save_manifest(self._index._root, manifest)
|
|
|
|
# Emit audit event for sync failure
|
|
try:
|
|
from ...infrastructure.telemetry.audit_logger import CanonicalAuditLogger
|
|
from ...config import CONFIG_DIR
|
|
logger = CanonicalAuditLogger(CONFIG_DIR / "audit")
|
|
logger.record(
|
|
kind="jira_knowledge.sync.failed",
|
|
name=f"{target.cowork_project_id}:{target.jira_project_key}",
|
|
ok=False,
|
|
detail=f"error={exc.code},message={exc.safe_message[:100]}",
|
|
agent_role="system"
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
raise
|
|
except Exception as exc: # noqa: BLE001
|
|
duration = time.monotonic() - start
|
|
manifest.mark_failure(category="UNEXPECTED", failed=0)
|
|
save_manifest(self._index._root, manifest)
|
|
|
|
# Emit audit event for unexpected failure
|
|
try:
|
|
from ...infrastructure.telemetry.audit_logger import CanonicalAuditLogger
|
|
from ...config import CONFIG_DIR
|
|
logger = CanonicalAuditLogger(CONFIG_DIR / "audit")
|
|
logger.record(
|
|
kind="jira_knowledge.sync.failed",
|
|
name=f"{target.cowork_project_id}:{target.jira_project_key}",
|
|
ok=False,
|
|
detail=f"error=UNEXPECTED,type={type(exc).__name__}",
|
|
agent_role="system"
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
raise ProviderError(
|
|
"SYNC_FAILED",
|
|
f"Jira sync failed: {type(exc).__name__}",
|
|
retryable=True,
|
|
) from exc
|
|
|
|
def _fetch_and_index(
|
|
self,
|
|
*,
|
|
config: Dict[str, str],
|
|
jql: str,
|
|
project_id: str,
|
|
base_url: str,
|
|
) -> Tuple[int, int, str]:
|
|
"""Paginate through Jira search results, normalize and upsert each issue.
|
|
|
|
Returns ``(processed, failed, latest_updated_cursor)``.
|
|
"""
|
|
from ...core import jira_tool
|
|
|
|
processed = 0
|
|
failed = 0
|
|
latest_cursor = ""
|
|
start_at = 0
|
|
|
|
for _ in range(_MAX_PAGES_PER_SYNC):
|
|
try:
|
|
data = jira_tool._get(
|
|
config,
|
|
"/rest/api/2/search",
|
|
{
|
|
"jql": jql,
|
|
"startAt": start_at,
|
|
"maxResults": _BATCH_SIZE,
|
|
"fields": _JQL_FIELDS,
|
|
},
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise ProviderError(
|
|
"UPSTREAM_ERROR",
|
|
"Failed to fetch issues from Jira.",
|
|
retryable=True,
|
|
) from exc
|
|
|
|
issues: List[dict] = data.get("issues") or []
|
|
if not issues:
|
|
break
|
|
|
|
for raw in issues:
|
|
try:
|
|
canonical = normalize_jira_issue(
|
|
raw, project_id=project_id, jira_base_url=base_url,
|
|
)
|
|
self._index.upsert(canonical)
|
|
processed += 1
|
|
# Track the latest updated timestamp for incremental cursor.
|
|
updated = canonical.provenance.source_updated
|
|
if updated and updated > latest_cursor:
|
|
latest_cursor = updated
|
|
except Exception: # noqa: BLE001
|
|
failed += 1
|
|
continue
|
|
|
|
total = data.get("total", 0)
|
|
start_at += len(issues)
|
|
if start_at >= total:
|
|
break
|
|
|
|
return processed, failed, latest_cursor
|
|
|
|
|
|
__all__ = ["JiraSyncService", "SyncResult"] |