diff --git a/docs/engineering/workflow_first_jira_to_von_migration_jvnautosci_1223.md b/docs/engineering/workflow_first_jira_to_von_migration_jvnautosci_1223.md index b3cbbeab9..665767892 100644 --- a/docs/engineering/workflow_first_jira_to_von_migration_jvnautosci_1223.md +++ b/docs/engineering/workflow_first_jira_to_von_migration_jvnautosci_1223.md @@ -53,6 +53,18 @@ archives, and deduplicate source activity. Unresolved hierarchy/link targets can be repaired from retained import rows after all projects have imported. Never infer deletion authority from a missing source issue. +Start large recovery runs with one importer process. Keep raw payloads in private +files and return compact progress summaries to the calling agent. Monitor host +memory pressure and the worker's memory use; stop the owned worker and preserve +its checkpoints when the host is under critical pressure. Increasing process +count is not a substitute for diagnosing a slow phase. Source registration and +task import coalesce derived relationship-index refreshes through the existing +bulk-mutation context, which flushes before the operation returns. +An issue or project source capture reuses one scoped MCP helper, while each +read retains its authority check, operation timeout and transport recovery. +The helper closes when that capture ends; unrelated calls keep their existing +one-call lifetime. + `scripts/capture_jira_project_sources.py` captures project and site source configuration through named Jira MCP resources and associates board/filter collections. It includes screen tabs and ordered fields, issue-type/screen @@ -67,6 +79,17 @@ documents require explicit capture or an evidence-backed scope disposition; REST access failures are not proof that a resource is empty. Keep these private source artefacts outside the repository. +Linked Atlassian Projects/Goals (formerly Atlas) can be read through the same +configured Jira account using named `atlas_project*` and `atlas_goal*` resources. +`capture_linked_atlassian_project` retains the selected project/goal fields and +paginated activity, custom values, access records and integration references as +source data. It checks the primary record again after capture. API errors, +changing records, missing/non-advancing pagination and unfinished nested pages +remain explicit gaps. A successful outer page does not certify nested comments +or custom-field values. These reads do not activate integrations or copy source +permissions into Von. Associate the retained archives with the intended native +projects and tasks, and keep unavailable fields in the disposition register. + ### Native use and actor binding The Tasks panel supports project/collection selection, ordinary task edits, diff --git a/src/backend/integrations/internal_mcp/atlassian_project_migration_queries.py b/src/backend/integrations/internal_mcp/atlassian_project_migration_queries.py new file mode 100644 index 000000000..57aa5d11e --- /dev/null +++ b/src/backend/integrations/internal_mcp/atlassian_project_migration_queries.py @@ -0,0 +1,180 @@ +"""Named read-only Atlas project queries for Jira-linked source retention. + +Queries use the configured Atlassian site and existing account. Source IDs and +cursors are variables, never executable GraphQL or credential destinations. +""" + +import re +from uuid import UUID + +USER = "id accountId canonicalAccountId accountStatus name picture" +DATE = "confidence dateRange { start end }" +COMMENT = f"id uuid url commentText creationDate editDate creator {{ {USER} }}" +HIGHLIGHT = f"id summary description creationDate lastEditedDate creator {{ {USER} }} lastEditedBy {{ {USER} }}" +NOTE = f"id summary description creationDate index archived creator {{ {USER} }}" +MILESTONE = "id uuid title status completionDate targetDate targetDateType lastModifiedDatetime creationDatetime" + + +def _connection(selection, *, arguments="first: $first, after: $after", edge_fields=""): + return f"({arguments}) {{ edges {{ {edge_fields} node {{ {selection} }} }} pageInfo {{ hasNextPage endCursor }} }}" + + +_DEFINITION_FIELDS = f"""id name token description type creationDate lastModifiedDate + linkedEntityTypes creator {{ {USER} }}""" +_DEFINITION = "__typename " + " ".join( + f"... on Townsquare{kind}CustomFieldDefinition {{ {_DEFINITION_FIELDS} }}" + for kind in ("User", "Text", "Number", "TextSelect") +) +_CUSTOM_COMMON = f"""uuid creationDate lastModifiedDate creator {{ {USER} }} + definition {{ {_DEFINITION} }}""" +CUSTOM_FIELDS = f"""__typename + ... on TownsquareTextCustomField {{ {_CUSTOM_COMMON} textValue: value {{ id value }} }} + ... on TownsquareNumberCustomField {{ {_CUSTOM_COMMON} numberValue: value {{ id value }} }} + ... on TownsquareTextSelectCustomField {{ {_CUSTOM_COMMON} + textValues: values {_connection('id value', arguments='first: 100')} }} + ... on TownsquareUserCustomField {{ {_CUSTOM_COMMON} + userValues: values {_connection(USER, arguments='first: 100')} }} +""" + + +_UPDATE_COMMON = f"""id ari uuid url creationDate editDate summary updateType + missedUpdate creator {{ {USER} }} lastEditedBy {{ {USER} }} + newTargetDate newTargetDateConfidence oldTargetDate oldTargetDateConfidence + newDueDate {{ {DATE} }} oldDueDate {{ {DATE} }} + comments {_connection(COMMENT, arguments='first: 100')} + updateNotes {_connection(NOTE, arguments='first: 100')} + highlights {_connection(HIGHLIGHT, arguments='first: 100')} +""" +PROJECT_UPDATE = _UPDATE_COMMON + f""" + newState {{ label value }} oldState {{ label value }} + newPhaseNew {{ id name displayName }} oldPhaseNew {{ id name displayName }} + milestones {_connection(MILESTONE, arguments='first: 100')} + changelog {{ accountId creationDate field newValue oldValue }} +""" +GOAL_UPDATE = _UPDATE_COMMON + """ + newState { label value score } oldState { label value score } + newScore oldScore newProgress { percentage type } oldProgress { percentage type } +""" +SHARED_CONNECTIONS = { + "customFields": CUSTOM_FIELDS, + "access": "__typename", + "teams": "id displayName description state membershipSettings organizationId", + "comments": COMMENT, + "watchers": USER, + "tags": "id name description creationDate iconData url", + "highlights": HIGHLIGHT, + "risks": HIGHLIGHT + " resolvedDate", + "slackChannels": f"""slackConnectionId types updatedAt creationDate + subscriber {{ {USER} }} metadata {{ fieldTypes }} + channel {{ channelId name numMembers private slackTeamName slackTeamId }}""", + "msteamsChannels": f"""creationDate subscriber {{ {USER} }} + tenant {{ microsoftTenantId displayName }} + team {{ internalId displayName visibility isPrivate }} + channel {{ channelId displayName isPrivate membershipType }}""", +} +PROJECT_CONNECTIONS = { + **SHARED_CONNECTIONS, + "updates": PROJECT_UPDATE, + "changelog": "accountId creationDate field newValue oldValue", + "dependencies": "created linkType incomingProject { id key name } outgoingProject { id key name }", + "goals": "id key name", + "links": "id name url iconUrl provider type", + "contributors": f"""userContributor {{ {USER} }} teamContributor {{ + team {{ id displayName }} + contributingMembers {_connection(USER, arguments='first: 100')} + }}""", +} +GOAL_CONNECTIONS = { + **SHARED_CONNECTIONS, + "updates": GOAL_UPDATE, + "projects": "id key name", + "parentGoals": "id key name", + "subGoals": "id key name", + "successMeasures": "id key name", +} + +_DRAFT = f"""id summary status modifiedDate author {{ {USER} }} + targetDate {{ date confidence }} + updateNotes {{ summary description uuid archived updateNoteId }} + highlights {{ summary description type }}""" +PROJECT_FIELDS = f"""id key uuid name isArchived isPrivate iconData + creationDate url startDate latestUpdateDate userUpdateCount watcherCount accessLevel + owner {{ {USER} }} description {{ what why measurement }} + dueDate {{ {DATE} }} state {{ label value }} + fusion {{ issueAri synced }} latestDraftUpdate {{ {_DRAFT} }} +""" +GOAL_FIELDS = f"""id key uuid name isArchived isPrivate iconData + creationDate url startDate latestUpdateDate accessLevel softDeleted + description descriptionWhy descriptionMeasurement scoringMode + owner {{ {USER} }} dueDate {{ {DATE} }} targetDate {{ {DATE} }} + state {{ label value score }} status {{ value score }} + progress {{ percentage type }} latestDraftUpdate {{ {_DRAFT} score }} +""" + + +def project_query_request( + resource, + identifier="", + secondary_id="", + start_at=0, + max_results=100, + next_page_token=None, +): + variables = {"first": max(1, min(int(max_results), 100)), "after": next_page_token} + if resource == "atlas_projects": + site_id = str(UUID(str(identifier))) + variables["container"] = f"ari:cloud:townsquare::site/{site_id}" + query = """query MigrationProjects($container: String!, $first: Int!, $after: String) { + projects_search(containerId: $container, searchString: "", first: $first, after: $after) { + edges { node { __typename id name } } + pageInfo { hasNextPage endCursor } + } + }""" + elif resource in {"atlas_project", "atlas_goal"} or resource.startswith( + ("atlas_project_", "atlas_goal_") + ): + kind = "goal" if resource.startswith("atlas_goal") else "project" + site_id = str(UUID(str(identifier))) + object_id = str(UUID(str(secondary_id))) + variables["id"] = f"ari:cloud:townsquare:{site_id}:{kind}/{object_id}" + root = f"{kind}s_byId({kind}Id: $id)" + variable_type = "ID!" if kind == "goal" else "String!" + if resource == f"atlas_{kind}": + variables = {"id": variables["id"]} + declarations = f"$id: {variable_type}" + selection = GOAL_FIELDS if kind == "goal" else PROJECT_FIELDS + else: + field = resource.removeprefix(f"atlas_{kind}_") + connections = GOAL_CONNECTIONS if kind == "goal" else PROJECT_CONNECTIONS + if field not in connections: + raise ValueError("Unsupported Atlassian project migration resource") + declarations = f"$id: {variable_type}, $first: Int!, $after: String" + edge_fields = ( + "principalAri role isFollower isOwner appLevelEffectiveRole" + if field == "access" + else "" + ) + selection = ( + "id " + field + _connection(connections[field], edge_fields=edge_fields) + ) + query = ( + f"query MigrationProjectResource({declarations}) {{ " + f'{root} @optIn(to: "Townsquare") {{ {selection} }} }}' + ) + elif resource == "atlas_type_schema": + if not re.fullmatch(r"[_A-Za-z][_0-9A-Za-z]*", str(identifier)): + raise ValueError("A GraphQL type name is required") + variables = {"name": identifier} + query = """query MigrationTypeSchema($name: String!) { + __type(name: $name) { + name kind possibleTypes { name } fields(includeDeprecated: true) { + name description args { name defaultValue type { + kind name ofType { kind name ofType { kind name } } + } } + type { kind name ofType { kind name ofType { kind name } } } + } + } + }""" + else: + raise ValueError("Unsupported Atlassian project migration resource") + return {"query": query, "variables": variables} diff --git a/src/backend/integrations/internal_mcp/jira_proxy_mcp.py b/src/backend/integrations/internal_mcp/jira_proxy_mcp.py index fe4e1c4fa..ee8f57d0d 100644 --- a/src/backend/integrations/internal_mcp/jira_proxy_mcp.py +++ b/src/backend/integrations/internal_mcp/jira_proxy_mcp.py @@ -216,6 +216,10 @@ def __init__(self, config: JiraProxyConfig): ) ) + def source_capture_session(self): + """Keep one helper for a source capture; authority is checked per read.""" + return self._client.reuse_session() + async def _call( self, tool_name: str, diff --git a/src/backend/integrations/internal_mcp/mcp_proxy_base.py b/src/backend/integrations/internal_mcp/mcp_proxy_base.py index fd83af7e6..d80154412 100644 --- a/src/backend/integrations/internal_mcp/mcp_proxy_base.py +++ b/src/backend/integrations/internal_mcp/mcp_proxy_base.py @@ -12,6 +12,8 @@ import logging import os import time +from contextlib import asynccontextmanager +from contextvars import ContextVar from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional, Sequence from uuid import uuid4 @@ -135,6 +137,68 @@ def __init__(self, config: MCPServerConfig) -> None: self._total_duration_ms = 0.0 self._last_call_telemetry: Optional[MCPCallTelemetry] = None self._telemetry_history: List[MCPCallTelemetry] = [] + self._capture_session: ContextVar[dict[str, Any] | None] = ContextVar( + f"mcp_capture_session_{id(self)}", default=None + ) + + @asynccontextmanager + async def _new_session(self, params): + # Bound startup independently; borrowed calls retain their own existing + # operation deadlines and telemetry below. + async with asyncio.timeout(self._effective_operation_timeout_sec()) as startup: + async with stdio_client(params) as (read_stream, write_stream): + async with ClientSession(read_stream, write_stream) as session: + await session.initialize() + startup.reschedule(None) + yield session + + @asynccontextmanager + async def reuse_session(self): + """Reuse one helper in an explicitly bounded caller scope, then close it. + + Context-local state cannot lend a session to an unrelated caller. A + dropped transport invalidates the borrowed session; normal per-call + retries then open fresh helpers. This is not a persistent process pool. + """ + existing = self._capture_session.get() + if existing and existing["usable"]: + yield + return + state = None + token = None + body_failed = False + try: + async with self._new_session(self._build_server_params()) as session: + state = {"session": session, "usable": True} + token = self._capture_session.set(state) + try: + yield + except BaseException: + body_failed = True + raise + except Exception as exc: + # A failed borrowed transport can also report its close while the + # scope unwinds. Preserve successfully recovered per-call results; + # never suppress a caller failure or an unrelated close error. + if ( + body_failed + or not state + or state["usable"] + or not self._is_transport_closed_error(exc) + ): + raise + finally: + if token is not None: + self._capture_session.reset(token) + + @asynccontextmanager + async def _call_session(self, params): + state = self._capture_session.get() + if state and state["usable"]: + yield state["session"] + else: + async with self._new_session(params) as session: + yield session @property def helper_owner_token(self) -> str: @@ -408,63 +472,63 @@ async def call_tool( async with asyncio.timeout(operation_timeout_sec): while True: try: - async with stdio_client(params) as (read_stream, write_stream): - async with ClientSession( - read_stream, write_stream - ) as session: - await session.initialize() - result = await session.call_tool(tool_name, arguments) - if bool(getattr(result, "isError", False)): - raise _MCPToolResultError( + async with self._call_session(params) as session: + result = await session.call_tool(tool_name, arguments) + if bool(getattr(result, "isError", False)): + raise _MCPToolResultError( + tool_name=tool_name, + actor_message=self._tool_result_error_message( + result, tool_name=tool_name, - actor_message=self._tool_result_error_message( - result, - tool_name=tool_name, - ), - ) - duration_ms = (time.perf_counter() - start_time) * 1000 - self._call_count += 1 - self._total_duration_ms += duration_ms - - # Extract content types for telemetry - content_types = [] - char_count = 0 - if result and getattr(result, "content", None): - for item in result.content: - if isinstance(item, mcp_types.TextContent): - content_types.append("text") - char_count += len(item.text or "") - elif isinstance(item, mcp_types.ImageContent): - content_types.append("image") - elif isinstance( - item, mcp_types.EmbeddedResource - ): - content_types.append("resource") - else: - content_types.append(type(item).__name__) - - parsed, parsed_as = self._parse_result_with_telemetry( - result, text_parser=text_parser + ), ) + duration_ms = (time.perf_counter() - start_time) * 1000 + self._call_count += 1 + self._total_duration_ms += duration_ms + + # Extract content types for telemetry + content_types = [] + char_count = 0 + if result and getattr(result, "content", None): + for item in result.content: + if isinstance(item, mcp_types.TextContent): + content_types.append("text") + char_count += len(item.text or "") + elif isinstance(item, mcp_types.ImageContent): + content_types.append("image") + elif isinstance(item, mcp_types.EmbeddedResource): + content_types.append("resource") + else: + content_types.append(type(item).__name__) + + parsed, parsed_as = self._parse_result_with_telemetry( + result, text_parser=text_parser + ) - telemetry.duration_ms = duration_ms - telemetry.success = True - telemetry.response_content_types = content_types - telemetry.response_char_count = char_count - telemetry.parsed_as = parsed_as - self._record_telemetry(telemetry) - - logger.info( - "%s Tool %s completed in %.1fms (chars=%d, parsed_as=%s)", - self._config.log_tag, - tool_name, - duration_ms, - char_count, - parsed_as, - ) + telemetry.duration_ms = duration_ms + telemetry.success = True + telemetry.response_content_types = content_types + telemetry.response_char_count = char_count + telemetry.parsed_as = parsed_as + self._record_telemetry(telemetry) + + logger.info( + "%s Tool %s completed in %.1fms (chars=%d, parsed_as=%s)", + self._config.log_tag, + tool_name, + duration_ms, + char_count, + parsed_as, + ) - return parsed + return parsed except Exception as exc: + state = self._capture_session.get() + if state and ( + isinstance(exc, TimeoutError) + or self._is_transport_closed_error(exc) + ): + state["usable"] = False should_retry = ( self._is_transport_closed_error(exc) and attempt < max_retries @@ -491,6 +555,11 @@ async def call_tool( continue raise except Exception as exc: # pragma: no cover + state = self._capture_session.get() + if state and ( + isinstance(exc, TimeoutError) or self._is_transport_closed_error(exc) + ): + state["usable"] = False duration_ms = (time.perf_counter() - start_time) * 1000 self._error_count += 1 self._total_duration_ms += duration_ms diff --git a/src/backend/mcp_server/mcp_server.py b/src/backend/mcp_server/mcp_server.py index e5668c63c..40c1dda0b 100644 --- a/src/backend/mcp_server/mcp_server.py +++ b/src/backend/mcp_server/mcp_server.py @@ -1396,12 +1396,21 @@ async def call_tool( return [types.TextContent(type="text", text=text)] elif name == "jira_get_migration_resource": + from src.backend.integrations.internal_mcp.atlassian_project_migration_queries import ( + project_query_request, + ) from src.backend.integrations.internal_mcp.jira_migration_resources import ( resource_request, ) - path, params = resource_request(**arguments) - result = _request_json("GET", f"{JIRA_BASE_URL}{path}", params=params) + if str(arguments.get("resource", "")).startswith("atlas_"): + payload = project_query_request(**arguments) + result = _request_json( + "POST", f"{JIRA_BASE_URL}/gateway/api/graphql", payload=payload + ) + else: + path, params = resource_request(**arguments) + result = _request_json("GET", f"{JIRA_BASE_URL}{path}", params=params) return [types.TextContent(type="text", text=_tool_json(result))] elif name == "jira_get_issue": diff --git a/src/backend/services/jira_source_retention_service.py b/src/backend/services/jira_source_retention_service.py index 00493e1d7..3d38dc675 100644 --- a/src/backend/services/jira_source_retention_service.py +++ b/src/backend/services/jira_source_retention_service.py @@ -14,9 +14,12 @@ import gzip import hashlib import json +from contextlib import nullcontext from datetime import UTC, datetime from typing import Any +from .relationship_extent_index_service import defer_relationship_extent_index_sync + def canonical_json(value: Any) -> bytes: return json.dumps( @@ -139,6 +142,16 @@ async def capture_issue( proxy, issue_key: str, *, attachment_max_size_bytes: int = 100 * 1024 * 1024 ): """Return an unchanged source archive and a separate hydrated projection.""" + session = getattr(proxy, "source_capture_session", None) + async with session() if callable(session) else nullcontext(): + return await _capture_issue( + proxy, + issue_key, + attachment_max_size_bytes=attachment_max_size_bytes, + ) + + +async def _capture_issue(proxy, issue_key, *, attachment_max_size_bytes): original = _source_payload( await proxy.get_issue( issue_key=issue_key, @@ -270,6 +283,12 @@ async def capture_issue( async def capture_project(proxy, project_key): """Retain project data, configuration and nested properties/role memberships.""" + session = getattr(proxy, "source_capture_session", None) + async with session() if callable(session) else nullcontext(): + return await _capture_project(proxy, project_key) + + +async def _capture_project(proxy, project_key): resources = {"project": await capture_resource(proxy, "project", project_key)} pages = resources["project"]["pages"] project = pages[0] if pages and isinstance(pages[0], dict) else {} @@ -522,6 +541,127 @@ async def capture_site_configuration(proxy, *, site_url): } +async def capture_linked_atlassian_project(proxy, *, kind, site_id, object_id): + """Capture named project/goal resources without silently truncating pages. + + This retains source information; it neither copies its permissions nor + enables imported integrations. Nested pages beyond the returned first page + remain explicit gaps, so callers cannot certify them from outer pagination. + """ + from ..integrations.internal_mcp.atlassian_project_migration_queries import ( + GOAL_CONNECTIONS, + PROJECT_CONNECTIONS, + ) + from ..integrations.internal_mcp.jira_proxy_mcp import JiraProxyError + + if kind not in {"project", "goal"}: + raise ValueError("Expected an Atlassian project or goal") + base_resource = f"atlas_{kind}" + root_key = f"{kind}s_byId" + resources = {} + + async def request(resource, cursor=None): + return _source_payload( + await proxy.get_migration_resource( + resource=resource, + identifier=site_id, + secondary_id=object_id, + max_results=100, + next_page_token=cursor, + ) + ) + + def nested_pages(value, path=""): + gaps = [] + if isinstance(value, dict): + if (value.get("pageInfo") or {}).get("hasNextPage"): + gaps.append(path) + for name, child in value.items(): + gaps.extend(nested_pages(child, f"{path}/{name}")) + elif isinstance(value, list): + for index, child in enumerate(value): + gaps.extend(nested_pages(child, f"{path}/{index}")) + return gaps + + first = await request(base_resource) + source = (first.get("data") or {}).get(root_key) + errors = list(first.get("errors") or []) + resources[base_resource] = { + "pages": [first], + "complete": bool(source) and not errors, + "errors": errors, + } + if source and not errors: + connections = PROJECT_CONNECTIONS if kind == "project" else GOAL_CONNECTIONS + for field in connections: + resource = f"{base_resource}_{field}" + record = { + "pages": [], + "items": [], + "errors": [], + "complete": False, + "nested_page_gaps": [], + } + resources[resource] = record + cursor = None + seen = set() + while True: + try: + page = await request(resource, cursor) + except (JiraProxyError, TimeoutError, OSError) as exc: + record["errors"].append( + {"type": type(exc).__name__, "message": str(exc)} + ) + break + record["pages"].append(page) + record["errors"].extend(page.get("errors") or []) + parent = (page.get("data") or {}).get(root_key) or {} + connection = parent.get(field) + if not isinstance(connection, dict) or record["errors"]: + if not record["errors"]: + record["errors"].append( + {"message": "Source connection unavailable"} + ) + break + edges = connection.get("edges") or [] + record["items"].extend(edges) + record["nested_page_gaps"].extend(nested_pages(edges)) + info = connection.get("pageInfo") or {} + if "hasNextPage" not in info: + record["errors"].append( + {"message": "Source pagination metadata missing"} + ) + break + if not info.get("hasNextPage"): + record["complete"] = not record["nested_page_gaps"] + break + cursor = info.get("endCursor") + if not cursor or cursor in seen: + record["errors"].append( + {"message": "Source pagination did not advance"} + ) + break + seen.add(cursor) + after = await request(base_resource) + resources[base_resource]["pages"].append(after) + stable = not after.get("errors") and canonical_json( + (after.get("data") or {}).get(root_key) + ) == canonical_json(source) + else: + stable = False + return { + "schema": "jira_source_archive.v1", + "kind": base_resource, + "source": source or {"id": object_id}, + "site_id": site_id, + "resources": resources, + "source_stable_during_capture": stable, + "capture_scope": "Named Atlassian project or goal fields and connections; source permissions and integration settings are retained as data.", + "complete": stable and all(value["complete"] for value in resources.values()), + } + + +@defer_relationship_extent_index_sync() def store_source_archive( archive, *, actor_concept_id, organisation_concept_id=None, namespace=None ): diff --git a/src/backend/services/jira_task_import_service.py b/src/backend/services/jira_task_import_service.py index 0602ac24b..95e80375e 100644 --- a/src/backend/services/jira_task_import_service.py +++ b/src/backend/services/jira_task_import_service.py @@ -55,6 +55,7 @@ ) from .task_ontology_service import JIRA_IMPORTED_TASK_SOURCE_ID from .feature_flags import suppress_event_workflow_launches +from .relationship_extent_index_service import defer_relationship_extent_index_sync _normalise_issue_key = normalise_jira_issue_key @@ -1984,6 +1985,7 @@ def _build_pilot_validation_report( @suppress_event_workflow_launches("jira_source_import") +@defer_relationship_extent_index_sync() def import_jira_issues_to_tasks( *, issues: Sequence[Mapping[str, Any]], diff --git a/tests/backend/test_jira_task_import_service.py b/tests/backend/test_jira_task_import_service.py index 29b131c49..0f496f315 100644 --- a/tests/backend/test_jira_task_import_service.py +++ b/tests/backend/test_jira_task_import_service.py @@ -173,6 +173,62 @@ def _fake_update_task_fields(*_args, **_kwargs): assert update_calls["count"] >= 2 +def test_import_flushes_shared_extent_once_after_all_task_mutations(monkeypatch): + from src.backend.services import relationship_extent_index_service as extent + + canonical = [] + flushes = [] + + def create(**_kwargs): + cid = f"#V#task_{len(canonical) + 1}" + canonical.append(cid) + extent.sync_relationship_extent_index_for_concept_id(cid) + extent.sync_relationship_extent_index_for_concept_id("#V#task_specification") + assert not flushes + return {"task_concept_id": cid} + + def update(cid, **_kwargs): + extent.sync_relationship_extent_index_for_concept_id(cid) + extent.sync_relationship_extent_index_for_concept_id("#V#task_specification") + assert not flushes + return {"task": {"task_concept_id": cid}} + + def flush(ids): + flushes.append((set(ids), list(canonical))) + return {"success": True} + + monkeypatch.setattr( + import_service, "find_task_by_external_reference", lambda **_kwargs: None + ) + monkeypatch.setattr(import_service, "create_task", create) + monkeypatch.setattr(import_service, "update_task_fields", update) + monkeypatch.setattr( + import_service, "upsert_task_external_reference", lambda *_args, **_kwargs: {} + ) + monkeypatch.setattr( + extent, "_sync_relationship_extent_index_for_concept_ids_now", flush + ) + monkeypatch.setattr( + extent, + "_sync_relationship_extent_index_for_concept_id_now", + lambda *_args: (_ for _ in ()).throw( + AssertionError("unbatched extent rebuild") + ), + ) + + result = import_service.import_jira_issues_to_tasks( + issues=[_jira_issue("TEST-1"), _jira_issue("TEST-2")], dry_run=False + ) + + assert result["summary"]["created"] == 2 + assert flushes == [ + ( + {"#V#task_1", "#V#task_2", "#V#task_specification"}, + ["#V#task_1", "#V#task_2"], + ) + ] + + def test_import_jira_issues_marks_bulk_migration_collection_when_requested(monkeypatch): captured_fields: list[dict[str, Any]] = [] diff --git a/tests/backend/test_linked_atlassian_project_retention.py b/tests/backend/test_linked_atlassian_project_retention.py new file mode 100644 index 000000000..f8cc52062 --- /dev/null +++ b/tests/backend/test_linked_atlassian_project_retention.py @@ -0,0 +1,122 @@ +import asyncio + +import pytest + +from src.backend.integrations.internal_mcp.atlassian_project_migration_queries import ( + project_query_request, +) +from src.backend.services.jira_source_retention_service import ( + capture_linked_atlassian_project, +) + +SITE = "b2b12142-9002-4b11-b421-83f073678fd3" +OBJECT = "2b2f3cae-3a8a-4e40-b117-aac7febd44c9" + + +class Source: + def __init__(self, mode="complete"): + self.mode = mode + self.cursors = [] + self.primary_reads = 0 + + async def get_migration_resource(self, **kwargs): + resource = kwargs["resource"] + if resource == "atlas_project": + self.primary_reads += 1 + name = ( + "Changed" + if self.mode == "changed" and self.primary_reads > 1 + else "Project" + ) + return {"data": {"projects_byId": {"id": OBJECT, "name": name}}} + field = resource.removeprefix("atlas_project_") + connection = { + "edges": [], + "pageInfo": {"hasNextPage": False, "endCursor": None}, + } + if field == "updates": + cursor = kwargs.get("next_page_token") + self.cursors.append(cursor) + connection["edges"] = [ + {"node": {"id": cursor or "first", "summary": "Original rich text"}} + ] + connection["pageInfo"] = { + "hasNextPage": cursor is None, + "endCursor": "second", + } + if self.mode == "repeated_cursor": + connection["pageInfo"]["hasNextPage"] = True + if self.mode == "missing_page_info": + connection.pop("pageInfo") + if self.mode == "nested_gap": + connection["edges"][0]["node"]["comments"] = { + "pageInfo": {"hasNextPage": True, "endCursor": "more"} + } + if self.mode == "denied": + return { + "data": {"projects_byId": None}, + "errors": [{"message": "Permission denied"}], + } + return {"data": {"projects_byId": {"id": OBJECT, field: connection}}} + + +def capture(source): + return asyncio.run( + capture_linked_atlassian_project( + source, kind="project", site_id=SITE, object_id=OBJECT + ) + ) + + +def test_capture_retains_all_pages_and_checks_source_version(): + source = Source() + result = capture(source) + assert result["complete"] + updates = result["resources"]["atlas_project_updates"] + assert [edge["node"]["id"] for edge in updates["items"]] == ["first", "second"] + assert len(updates["pages"]) == 2 + assert source.cursors == [None, "second"] + assert source.primary_reads == 2 + + +@pytest.mark.parametrize( + "mode", ["repeated_cursor", "missing_page_info", "nested_gap", "denied", "changed"] +) +def test_incomplete_or_changing_source_never_certifies_retention(mode): + source = Source(mode) + result = capture(source) + assert not result["complete"] + if mode != "changed": + record = result["resources"]["atlas_project_updates"] + assert not record["complete"] + assert record["pages"] + assert record["errors"] or record["nested_page_gaps"] + + +def test_query_variables_do_not_become_query_text_or_destinations(): + cursor = '") { mutation { deleteAll } } #' + result = project_query_request( + "atlas_project_updates", SITE, OBJECT, max_results=10000, next_page_token=cursor + ) + assert result["variables"]["after"] == cursor + assert result["variables"]["first"] == 100 + assert cursor not in result["query"] + assert "mutation" not in result["query"] + assert result["variables"]["id"] == f"ari:cloud:townsquare:{SITE}:project/{OBJECT}" + + +@pytest.mark.parametrize( + "resource,identifier,secondary", + [ + ("atlas_delete", SITE, OBJECT), + ("atlas_project_delete", SITE, OBJECT), + ("atlas_project", "https://example.com", OBJECT), + ("atlas_project", SITE, "../credentials"), + ("atlas_type_schema", "Query) { secret", ""), + ], +) +def test_unknown_resources_and_invalid_source_identifiers_are_rejected( + resource, identifier, secondary +): + with pytest.raises(ValueError): + project_query_request(resource, identifier, secondary) diff --git a/tests/backend/test_mcp_source_capture_session.py b/tests/backend/test_mcp_source_capture_session.py new file mode 100644 index 000000000..d6d69f220 --- /dev/null +++ b/tests/backend/test_mcp_source_capture_session.py @@ -0,0 +1,125 @@ +import asyncio +from contextlib import asynccontextmanager +from types import SimpleNamespace + +import pytest +from mcp import types + +from src.backend.integrations.internal_mcp import mcp_proxy_base as module + + +def client_fixture(monkeypatch, *, timeout=1): + state = {"opened": [], "closed": [], "calls": []} + + @asynccontextmanager + async def stdio(_params): + yield object(), object() + + class Session: + def __init__(self, *_args): + self.number = len(state["opened"]) + state["opened"].append(self.number) + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + state["closed"].append(self.number) + if state.get("drop_cleanup") and self.number == 0: + raise RuntimeError("Transport closed") + + async def initialize(self): + pass + + async def call_tool(self, name, arguments): + state["calls"].append(self.number) + if name == "drop" and self.number == 0: + raise RuntimeError("Transport closed") + if name == "slow": + await asyncio.sleep(1) + return SimpleNamespace( + content=[types.TextContent(type="text", text='{"ok":true}')] + ) + + monkeypatch.setattr(module, "stdio_client", stdio) + monkeypatch.setattr(module, "ClientSession", Session) + client = module.MCPStdIOClient( + module.MCPServerConfig( + command="python", + args=["source.py"], + timeout_sec=timeout, + transport_retry_attempts=1, + transport_retry_backoff_sec=0, + ) + ) + return client, state + + +def test_source_scope_reuses_concurrent_reads_and_closes_before_next_scope(monkeypatch): + client, state = client_fixture(monkeypatch) + + async def run(): + async with client.reuse_session(): + results = await asyncio.gather( + *(client.call_tool("read", {}) for _ in range(4)) + ) + assert all(result == {"ok": True} for result in results) + assert state["opened"] == [0] + assert not state["closed"] + assert state["closed"] == [0] + await client.call_tool("read", {}) + + asyncio.run(run()) + assert state["opened"] == state["closed"] == [0, 1] + assert client.call_count == 5 + assert client._capture_session.get() is None + + +def test_dropped_borrowed_transport_recovers_on_fresh_helpers(monkeypatch): + client, state = client_fixture(monkeypatch) + state["drop_cleanup"] = True + + async def run(): + async with client.reuse_session(): + assert await client.call_tool("drop", {}) == {"ok": True} + assert await client.call_tool("read", {}) == {"ok": True} + + asyncio.run(run()) + assert state["calls"] == [0, 1, 2] + assert sorted(state["closed"]) == state["opened"] + assert client.call_count == 2 + assert client._capture_session.get() is None + + +def test_each_borrowed_call_keeps_its_timeout_without_a_whole_capture_timeout( + monkeypatch, +): + client, state = client_fixture(monkeypatch, timeout=0.08) + + async def run(): + async with client.reuse_session(): + await asyncio.sleep(0.12) + assert await client.call_tool("read", {}) == {"ok": True} + with pytest.raises(module.MCPToolClientError, match="deadline"): + await client.call_tool("slow", {}) + assert await client.call_tool("read", {}) == {"ok": True} + + asyncio.run(run()) + assert state["opened"] == [0, 1] + assert sorted(state["closed"]) == state["opened"] + assert client.error_count == 1 + + +def test_caller_failure_is_preserved_and_session_context_is_cleared(monkeypatch): + client, state = client_fixture(monkeypatch) + + async def run(): + with pytest.raises(ValueError, match="caller failure"): + async with client.reuse_session(): + await client.call_tool("read", {}) + raise ValueError("caller failure") + assert client._capture_session.get() is None + assert await client.call_tool("read", {}) == {"ok": True} + + asyncio.run(run()) + assert state["opened"] == state["closed"] == [0, 1]