(.*?)

\", r\"```\\n\\1\\n```\", html_content, flags=re.DOTALL)"," html_content = re.sub(r\"(.*?)\", r\"`\\1`\", html_content, flags=re.DOTALL)"," # Convert links"," html_content = re.sub(r'(.*?)', r\"[\\2](\\1)\", html_content, flags=re.DOTALL)"," # Convert lists (basic support)"," html_content = re.sub(r\"

(.*?)

\", r\"- \\1\", html_content, flags=re.DOTALL)"," html_content = re.sub(r\"

|\", \"\", html_content)"," # Convert paragraphs"," html_content = re.sub(r\"

(.*?)

\", r\"\\1\\n\\n\", html_content, flags=re.DOTALL)"," # Convert line breaks"," html_content = re.sub(r\"\", \"\\n\", html_content)"," # Remove remaining HTML tags"," html_content = re.sub(r\"<[^>]+>\", \"\", html_content)"," # Clean up extra whitespace"," html_content = re.sub(r\"\\n{3,}\", \"\\n\\n\", html_content)"," return html.unescape(html_content.strip())"," def _encode_pat(self, token: str) -> str:"," Encode PAT for Basic authentication."," token: Azure DevOps PAT"," Base64-encoded token for Basic auth"," import base64"," return base64.b64encode(f\":{token}\".encode()).decode()"," def _try_refresh_oauth_token(self) -> dict[str, Any] | None:"," Attempt to refresh expired OAuth token using persistent token cache."," This uses the same persistent cache as the auth command, allowing automatic"," token refresh without user interaction (like Azure CLI)."," Refreshed token data dict if successful, None if refresh failed"," from azure.identity import ( # type: ignore[reportMissingImports]"," DeviceCodeCredential,"," TokenCachePersistenceOptions,"," # Use the same cache name as auth command for shared cache"," # Try encrypted first, fall back to unencrypted if libsecret unavailable"," cache_options = None"," cache_options = TokenCachePersistenceOptions("," name=\"specfact-azure-devops\","," allow_unencrypted_storage=False, # Prefer encrypted"," except Exception:"," # Encrypted cache not available, try unencrypted"," allow_unencrypted_storage=True, # Fallback: unencrypted"," except Exception:"," # Persistent cache completely unavailable, can't refresh"," # Create credential with same cache - it will use cached refresh token"," credential = DeviceCodeCredential(cache_persistence_options=cache_options)"," # Use the same resource and scopes as auth command"," # Note: Refresh tokens are automatically obtained via persistent token cache"," # offline_access is a reserved scope and cannot be explicitly requested"," azure_devops_resource = \"499b84ac-1321-427f-aa17-267ca6975798/.default\""," azure_devops_scopes = [azure_devops_resource]"," token = _get_access_token(credential, azure_devops_scopes)"," # Return refreshed token data"," from datetime import UTC, datetime"," expires_at = datetime.fromtimestamp(token.expires_on, tz=UTC).isoformat()"," \"access_token\": token.token,"," \"token_type\": \"bearer\","," \"expires_at\": expires_at,"," \"resource\": azure_devops_resource,"," \"issued_at\": datetime.now(tz=UTC).isoformat(),"," # Refresh failed (no cached refresh token, refresh token expired, etc.)"," def _auth_headers(self) -> dict[str, str | bytes]:"," \"\"\"Return authorization headers based on token type.\"\"\""," return {}"," if self.auth_scheme == \"bearer\":"," return {\"Authorization\": f\"Bearer {self.api_token}\"}"," return {\"Authorization\": f\"Basic {self._encode_pat(self.api_token)}\"}"," @ensure(lambda result: hasattr(result, \"raise_for_status\"), \"Result must support raise_for_status\")"," def _ado_get("," url: str,"," headers: dict[str, str | bytes] | None = None,"," params: dict[str, Any] | None = None,"," timeout: int = 30,"," retry_on_ambiguous_transport: bool = True,"," ) -> Any:"," \"\"\"Execute an idempotent ADO GET with retry policy for transient failures.\"\"\""," return cast("," Any,"," lambda: requests.get(url, headers=headers, params=params, timeout=timeout),"," retry_on_ambiguous_transport=retry_on_ambiguous_transport,"," def _ado_post("," json: dict[str, Any] | None = None,"," \"\"\"Execute ADO POST with retry policy. Safe for read-only WIQL endpoints.\"\"\""," request_kwargs: dict[str, Any] = {\"headers\": headers, \"json\": json, \"timeout\": timeout}"," if params:"," request_kwargs[\"params\"] = params"," lambda: requests.post(url, **request_kwargs),"," def _work_item_exists(self, work_item_id: int | str, org: str, project: str) -> bool:"," Check if a work item exists in Azure DevOps."," work_item_id: Work item ID to check"," True if work item exists, False otherwise (including if deleted)"," # Ensure work_item_id is an integer"," work_item_id = int(work_item_id)"," headers = {"," \"Accept\": \"application/json\","," **self._auth_headers(),"," response = self._ado_get(url, headers=headers, timeout=10)"," # Check if work item is deleted (System.State == \"Removed\")"," work_item_data = response.json()"," fields = work_item_data.get(\"fields\", {})"," state = fields.get(\"System.State\", \"\")"," # Consider \"Removed\" as non-existent for our purposes"," return state != \"Removed\""," except requests.HTTPError as e:"," if e.response is not None and e.response.status_code == 404:"," # On any error, assume it doesn't exist (safer to allow creation)"," def _get_work_item_data(self, work_item_id: int | str, org: str, project: str) -> dict[str, Any] | None:"," Get current work item data from Azure DevOps."," work_item_id: Work item ID to fetch"," Work item data dict with fields (title, state, etc.) or None if not found"," return _normalize_work_item_data(response.json())"," lambda self, issue_id: isinstance(issue_id, str) and len(issue_id.strip()) > 0, \"issue_id must be non-empty\""," @ensure(lambda result: result is None or isinstance(result, BacklogItem), \"Must return BacklogItem or None\")"," \"\"\"Fetch a single ADO work item directly by ID (bypasses WIQL list queries).\"\"\""," normalized_id = issue_id.strip()"," if not normalized_id.isdigit():"," if not self.org or not self.project:"," url = self._build_ado_url(f\"_apis/wit/workitems/{int(normalized_id)}\", api_version=\"7.1\")"," response = self._ado_get(url, headers=headers, params={\"$expand\": \"all\"}, timeout=30)"," work_item = response.json()"," from specfact_cli.backlog.converter import convert_ado_work_item_to_backlog_item"," return convert_ado_work_item_to_backlog_item("," work_item,"," base_url=self.base_url,"," org=self.org,"," project_name=self.project,"," def _find_work_item_by_change_id(self, change_id: str, org: str, project: str) -> dict[str, Any] | None:"," Find an existing ADO work item by OpenSpec change_id embedded in the description."," change_id: OpenSpec change ID (used in footer marker)"," Source tracking entry dict if found, otherwise None."," if not self.api_token or not change_id:"," project_escaped = project.replace(\"'\", \"''\")"," change_id_escaped = change_id.replace(\"'\", \"''\")"," wiql = {"," \"query\": ("," \"Select [System.Id] From WorkItems \""," f\"Where [System.TeamProject] = '{project_escaped}' \""," f\"And [System.Description] Contains 'OpenSpec Change Proposal: `{change_id_escaped}`'\""," url = f\"{self.base_url}/{org}/{project}/_apis/wit/wiql?api-version=7.1\""," \"Content-Type\": \"application/json\","," response = self._ado_post(url, json=wiql, headers=headers, timeout=10)"," if is_debug_mode():"," debug_log_operation("," \"ado_wiql\","," url,"," str(response.status_code),"," error=None if response.ok else (response.text[:200] if response.text else None),"," if response.status_code != 200:"," work_items = response.json().get(\"workItems\", [])"," work_item_ids = [item.get(\"id\") for item in work_items if item.get(\"id\")]"," if not work_item_ids:"," work_item_id = min(work_item_ids)"," work_item_url = f\"{self.base_url}/{org}/{project}/_workitems/edit/{work_item_id}\""," \"source_id\": str(work_item_id),"," \"source_url\": work_item_url,"," \"source_type\": \"ado\","," \"source_repo\": f\"{org}/{project}\","," debug_log_operation(\"ado_wiql\", url, \"error\", error=str(e))"," def _create_work_item_from_proposal("," Create ADO work item from change proposal."," body = raw_body or self._build_change_proposal_body(title, rationale, description, impact, change_id)"," # Get work item type"," work_item_type = self._get_work_item_type(org, project)"," ado_state = self._resolve_proposal_ado_state(proposal_data)"," # Create work item via Azure DevOps API"," url = f\"{self.base_url}/{org}/{project}/_apis/wit/workitems/${work_item_type}?api-version=7.1\""," \"Content-Type\": \"application/json-patch+json\","," patch_document = self._ado_create_patch_document(title, body, ado_state)"," lambda: requests.post(url, json=cast(Any, patch_document), headers=headers, timeout=30),"," \"ado_create\","," work_item_data = cast(dict[str, Any], response.json())"," work_item_id, work_item_url = self._parse_ado_create_work_item_response(work_item_data)"," self._merge_created_work_item_source_tracking("," proposal_data,"," _AdoCreatedWorkItemRef("," work_item_id=work_item_id,"," work_item_url=work_item_url,"," org=org,"," project=project,"," work_item_type=work_item_type,"," ado_state=ado_state,"," \"work_item_id\": work_item_id,"," \"work_item_url\": work_item_url,"," \"state\": ado_state,"," resp = getattr(e, \"response\", None)"," e.ado_user_message = user_msg # type: ignore[attr-defined]"," def _update_work_item_status("," Update ADO work item status based on change proposal status."," proposal_data: Change proposal data with source_tracking containing work item ID"," Dict with updated work item data: {\"work_item_id\": int, \"work_item_url\": str, \"state\": str}"," work_item_id = self._get_source_tracking_work_item_id("," proposal_data.get(\"source_tracking\", {}),"," f\"{org}/{project}\","," work_item_data = self._patch_work_item("," org,"," project,"," work_item_id,"," [{\"op\": \"replace\", \"path\": \"/fields/System.State\", \"value\": ado_state}],"," work_item_url = work_item_data.get(\"_links\", {}).get(\"html\", {}).get(\"href\", \"\")"," return {\"work_item_id\": work_item_id, \"work_item_url\": work_item_url, \"state\": ado_state}"," def _update_work_item_body("," Update ADO work item body/description from change proposal."," work_item_id: Work item ID to update"," # Update work item body and state via Azure DevOps API"," # Build JSON Patch document for work item update"," # Set multilineFieldsFormat to Markdown for proper rendering"," patch_document = ["," {\"op\": \"replace\", \"path\": \"/fields/System.Title\", \"value\": title},"," {\"op\": \"replace\", \"path\": \"/fields/System.Description\", \"value\": body},"," {\"op\": \"replace\", \"path\": \"/fields/System.State\", \"value\": ado_state},"," }, # Set format to Markdown"," lambda: requests.patch(url, json=patch_document, headers=headers, timeout=30)"," work_item_url = work_item_data.get(\"_links\", {}).get(\"html\", {}).get(\"href\", \"\")"," def sync_status_to_ado("," Sync OpenSpec change status to ADO work item state."," Updates ADO work item state based on OpenSpec change proposal status."," Dict with sync result: {\"work_item_id\": int, \"work_item_url\": str, \"state_updated\": bool}"," ValueError: If work item ID not found in source_tracking"," source_tracking = ("," proposal.source_tracking if isinstance(proposal, ChangeProposal) else proposal.get(\"source_tracking\")"," raise ValueError(\"Source tracking required for status sync (work item must be created first)\")"," work_item_id = self._get_source_tracking_work_item_id(source_tracking, target_repo)"," ado_state = self.map_openspec_status_to_backlog("," proposal.status if isinstance(proposal, ChangeProposal) else proposal.get(\"status\", \"proposed\")"," \"work_item_id\": work_item_id,"," \"work_item_url\": work_item_url,"," \"state_updated\": True,"," \"new_state\": ado_state,"," @require(lambda work_item_data: isinstance(work_item_data, dict), \"Work item data must be dict\")"," def sync_status_from_ado("," work_item_data: dict[str, Any],"," Sync ADO work item state to OpenSpec change proposal."," Maps ADO work item state to OpenSpec status and resolves conflicts if status differs."," work_item_data: ADO work item data (dict from API response)"," fields = work_item_data.get(\"fields\", {})"," openspec_status_from_ado = self.map_backlog_status_to_openspec(ado_state)"," openspec_status = ("," return self.resolve_status_conflict(openspec_status, openspec_status_from_ado, strategy)"," target_repo: Target repository identifier (e.g., \"org/project\") to filter source_tracking entries"," code_repo_path: Path to code repository for branch verification"," # Try to infer from change_id (common pattern: feature/)"," # Common branch naming patterns"," possible_branches = ["," f\"feature/{change_id}\","," f\"bugfix/{change_id}\","," f\"hotfix/{change_id}\","," # Check each possible branch in code repo"," if code_repo_path:"," for branch in possible_branches:"," if self._verify_branch_exists(branch, code_repo_path):"," # No repo path available, return first as reasonable default"," return possible_branches[0]"," return _git_branch_exists_via_local_commands(repo_path, branch_name)"," self.console.log(f\"[bold yellow]Warning:[/bold yellow] Error checking branch existence: {e}\") # type: ignore[attr-defined]"," def _get_work_item_comments(self, org: str, project: str, work_item_id: int) -> list[dict[str, Any]]:"," Fetch comments for an Azure DevOps work item."," work_item_id: Work item ID"," List of comment dicts with 'text' or 'body' field, or empty list on error"," url = f\"{self.base_url}/{org}/{project}/_apis/wit/workItems/{work_item_id}/comments\""," comments: list[dict[str, Any]] = []"," continuation_token: str | None = None"," seen_tokens: set[str] = set()"," while True:"," params: dict[str, Any] = {\"api-version\": _ADO_COMMENTS_API_VERSION, \"$top\": 200, \"order\": \"asc\"}"," if continuation_token:"," params[\"continuationToken\"] = continuation_token"," response = self._ado_get(url, headers=headers, params=params, timeout=30)"," response_data = response.json()"," raw_comments = response_data.get(\"comments\", [])"," if isinstance(raw_comments, list):"," comments.extend([c for c in raw_comments if isinstance(c, dict)])"," next_token = response.headers.get(\"x-ms-continuationtoken\") or response_data.get(\"continuationToken\")"," if not next_token or not isinstance(next_token, str):"," if next_token in seen_tokens:"," seen_tokens.add(next_token)"," continuation_token = next_token"," return comments"," @require(lambda org: isinstance(org, str) and org, \"Organization must be non-empty string\")"," @require(lambda project: isinstance(project, str) and project, \"Project must be non-empty string\")"," lambda work_item_id: isinstance(work_item_id, int) and work_item_id > 0, \"Work item ID must be positive int\""," lambda comment_text: isinstance(comment_text, str) and comment_text, \"Comment text must be non-empty string\""," def _add_work_item_comment("," comment_text: str,"," Add a comment to an Azure DevOps work item."," comment_text: Comment text (markdown supported)"," Dict with comment data: {\"work_item_id\": int, \"comment_id\": int, \"comment_added\": bool}"," ValueError: If API token is missing"," msg = \"Azure DevOps API token is required\""," # Azure DevOps API for adding comments to work items"," url = ("," f\"{self.base_url}/{org}/{project}/_apis/wit/workitems/{work_item_id}/comments\""," f\"?api-version={_ADO_COMMENTS_API_VERSION}\""," # Build request body for comment"," comment_body = {\"text\": comment_text}"," lambda: requests.post(url, json=cast(Any, comment_body), headers=headers, timeout=30),"," comment_data = response.json()"," comment_id = comment_data.get(\"id\")"," \"comment_id\": comment_id,"," user_msg = _log_ado_patch_failure(resp, [], url)"," @require(lambda proposal_data: isinstance(proposal_data, dict), \"Proposal data must be dict\")"," Add progress comment to Azure DevOps work item based on code changes."," work_item_id: Azure DevOps work item ID"," Dict with updated work item data: {\"work_item_id\": int, \"work_item_url\": str, \"comment_added\": bool}"," \"work_item_url\": f\"{self.base_url}/{org}/{project}/_workitems/edit/{work_item_id}\","," self._add_work_item_comment(org, project, work_item_id, comment_text)"," msg = f\"Failed to add progress comment to Azure DevOps work item #{work_item_id}: {e}\""," def _resolve_default_team_from_project_api(self) -> str | None:"," \"\"\"Fetch first team for the project and cache as _auto_resolved_team.\"\"\""," from urllib.parse import quote"," project_encoded = quote(self.project or \"\", safe=\"\")"," project_url = f\"{self.base_url}/{self.org}/_apis/projects/{project_encoded}\""," project_params = {\"api-version\": \"7.1\"}"," project_headers = {"," \"Accept\": \"application/json\","," project_response = self._ado_get(project_url, headers=project_headers, params=project_params, timeout=30)"," project_data = project_response.json()"," project_id = project_data.get(\"id\")"," if not project_id:"," teams_url = f\"{self.base_url}/{self.org}/_apis/projects/{project_id}/teams\""," teams_response = self._ado_get(teams_url, headers=project_headers, params=project_params, timeout=30)"," teams_data = teams_response.json()"," teams = teams_data.get(\"value\", [])"," if not teams:"," team_to_use = teams[0].get(\"name\")"," self._auto_resolved_team = team_to_use"," return team_to_use"," def _get_current_iteration_path_for_team(self, team_to_use: str) -> str | None:"," \"\"\"Query team current iteration; on 404 retry with project name as team.\"\"\""," team_encoded = quote(team_to_use, safe=\"\")"," url = f\"{self.base_url}/{self.org}/{self.project}/{team_encoded}/_apis/work/teamsettings/iterations\""," params = {\"$timeframe\": \"current\", \"api-version\": \"7.1\"}"," response = self._ado_get(url, headers=headers, params=params, timeout=30)"," data = response.json()"," iterations = data.get(\"value\", [])"," if iterations:"," return iterations[0].get(\"path\")"," if e.response is not None and e.response.status_code == 404 and team_to_use != self.project:"," project_encoded = quote(self.project or \"\", safe=\"\")"," fallback_url = ("," f\"{self.base_url}/{self.org}/{self.project}/{project_encoded}/_apis/work/teamsettings/iterations\""," fallback_response = self._ado_get(fallback_url, headers=headers, params=params, timeout=30)"," fallback_data = fallback_response.json()"," fallback_iterations = fallback_data.get(\"value\", [])"," if fallback_iterations:"," return fallback_iterations[0].get(\"path\")"," except requests.RequestException:"," pass"," def _get_current_iteration(self) -> str | None:"," Get the current active iteration for the team."," Current iteration path if found, None otherwise"," requests.RequestException: If API call fails"," team_to_use = self.team or getattr(self, \"_auto_resolved_team\", None)"," if not team_to_use:"," team_to_use = self._resolve_default_team_from_project_api()"," return self._get_current_iteration_path_for_team(team_to_use)"," def _list_available_iterations(self) -> list[str]:"," List all available iteration paths for the team."," List of iteration paths (empty list if unavailable)"," # If team is not set, try to get it (same logic as _get_current_iteration)"," # Try to get the default team for the project (same logic as _get_current_iteration)"," from urllib.parse import quote"," project_encoded = quote(self.project, safe=\"\")"," project_url = f\"{self.base_url}/{self.org}/_apis/projects/{project_encoded}\""," project_params = {\"api-version\": \"7.1\"}"," project_headers = {"," **self._auth_headers(),"," \"Accept\": \"application/json\","," project_response = self._ado_get("," project_url, headers=project_headers, params=project_params, timeout=30"," project_data = project_response.json()"," project_id = project_data.get(\"id\")"," if project_id:"," teams_url = f\"{self.base_url}/{self.org}/_apis/projects/{project_id}/teams\""," teams_response = self._ado_get("," teams_url, headers=project_headers, params=project_params, timeout=30"," teams_data = teams_response.json()"," teams = teams_data.get(\"value\", [])"," if teams:"," team_to_use = teams[0].get(\"name\")"," self._auto_resolved_team = team_to_use"," except requests.RequestException:"," return []"," # Team iterations API: /{org}/{project}/{team}/_apis/work/teamsettings/iterations"," # URL encode team name in case it has spaces or special characters"," params = {\"api-version\": \"7.1\"}"," return [it.get(\"path\", \"\") for it in iterations if it.get(\"path\")]"," # Fail silently - will be handled by caller"," def _resolve_sprint_filter_when_empty("," self, items: list[BacklogItem], apply_current_when_missing: bool"," ) -> tuple[str | None, list[BacklogItem]]:"," if not apply_current_when_missing:"," return None, items"," current_iteration = self._get_current_iteration()"," if current_iteration:"," filtered = [item for item in items if item.iteration and item.iteration == current_iteration]"," return current_iteration, filtered"," console.print(\"[yellow]⚠ No current iteration found; returning all items[/yellow]\")"," return None, items"," def _resolve_sprint_filter_by_name("," self, sprint_filter: str, items: list[BacklogItem]"," matching_items = ["," if item.sprint"," and BacklogFilters.normalize_filter_value(item.sprint)"," == BacklogFilters.normalize_filter_value(sprint_filter)"," if not matching_items:"," return sprint_filter, []"," unique_iterations = {item.iteration for item in matching_items if item.iteration}"," if len(unique_iterations) > 1:"," raise ValueError(_ambiguous_sprint_error_message(sprint_filter, unique_iterations))"," iteration_path = unique_iterations.pop() if unique_iterations else None"," return iteration_path, matching_items"," def _resolve_sprint_filter("," sprint_filter: str | None,"," items: list[BacklogItem],"," apply_current_when_missing: bool = True,"," Resolve sprint filter with path matching and ambiguity detection."," sprint_filter: Sprint filter value (name or full path)"," items: List of backlog items to filter"," Tuple of (resolved_iteration_path, filtered_items)"," ValueError: If ambiguous sprint name match is detected"," if not sprint_filter:"," return self._resolve_sprint_filter_when_empty(items, apply_current_when_missing)"," if \"\\\\\" in sprint_filter or \"/\" in sprint_filter:"," filtered = [item for item in items if item.iteration and item.iteration == sprint_filter]"," return sprint_filter, filtered"," return self._resolve_sprint_filter_by_name(sprint_filter, items)"," return \"ado\""," def _apply_iteration_filter_post_fetch("," self, filtered_items: list[BacklogItem], filters: BacklogFilters"," ) -> list[BacklogItem]:"," \"\"\"Restrict items by iteration when WIQL did not already scope by path.\"\"\""," if not filters.iteration:"," return filtered_items"," normalized_iteration = BacklogFilters.normalize_filter_value(filters.iteration)"," if normalized_iteration in (None, \"any\"):"," target_iteration = filters.iteration"," if normalized_iteration == \"current\":"," current_iteration = self._get_current_iteration()"," if not current_iteration:"," target_iteration = current_iteration"," for item in filtered_items"," if BacklogFilters.normalize_filter_value(item.iteration)"," == BacklogFilters.normalize_filter_value(target_iteration)"," def _filter_backlog_items_state_assignee_labels("," normalized_state = BacklogFilters.normalize_filter_value(filters.state)"," item for item in filtered_items if BacklogFilters.normalize_filter_value(item.state) == normalized_state"," normalized_assignee = BacklogFilters.normalize_filter_value(filters.assignee)"," if any("," BacklogFilters.normalize_filter_value(assignee) == normalized_assignee"," for assignee in item.assignees"," filtered_items = [item for item in filtered_items if any(label in item.tags for label in filters.labels)]"," def _apply_sprint_filter_post_fetch("," filtered_items: list[BacklogItem],"," filters: BacklogFilters,"," sprint_apply_current: bool | None,"," echo_sprint_value_error: bool,"," if not filters.sprint:"," apply_current = ("," sprint_apply_current"," if sprint_apply_current is not None"," else getattr(filters, \"use_current_iteration_default\", True)"," _, out = self._resolve_sprint_filter("," filters.sprint,"," filtered_items,"," apply_current_when_missing=apply_current,"," except ValueError as err:"," if echo_sprint_value_error:"," console.print(f\"[red]Error:[/red] {err}\")"," return out"," def _filter_backlog_items_by_release_post_fetch("," if not filters.release:"," normalized_release = BacklogFilters.normalize_filter_value(filters.release)"," if item.release and BacklogFilters.normalize_filter_value(item.release) == normalized_release"," def _apply_backlog_limit_post_fetch("," if filters.limit is None or len(filtered_items) <= filters.limit:"," return filtered_items[: filters.limit]"," def _try_fetch_backlog_by_direct_issue(self, filters: BacklogFilters) -> list[BacklogItem] | None:"," \"\"\"When issue_id is set, fetch that item and apply filters; otherwise return None.\"\"\""," requested_issue_id = str(getattr(filters, \"issue_id\", \"\") or \"\").strip()"," if not requested_issue_id:"," direct_item = self._fetch_backlog_item_by_id(requested_issue_id)"," if direct_item is None:"," return self._apply_post_fetch_filters_after_wiql("," [direct_item],"," filters,"," include_iteration=True,"," sprint_apply_current=False,"," echo_sprint_value_error=False,"," def _wiql_append_iteration_conditions(self, filters: BacklogFilters, conditions: list[str]) -> str | None:"," \"\"\"Add iteration-related WIQL conditions; return resolved iteration path for error messages.\"\"\""," resolved_iteration: str | None = None"," if filters.iteration:"," if filters.iteration.lower() == \"current\":"," current_iteration = self._get_current_iteration()"," if current_iteration:"," resolved_iteration = current_iteration"," conditions.append(f\"[System.IterationPath] = '{resolved_iteration}'\")"," suggestions = _rich_iteration_suggestions_block(self._list_available_iterations())"," error_msg = ("," f\"[red]Error:[/red] No current iteration found.\\n\\n\""," f\"{suggestions}\""," f\"[cyan]Tips:[/cyan]\\n\""," f\" • Specify a full iteration path: [bold]--iteration 'Project\\\\Sprint 1'[/bold]\\n\""," f\" • Use [bold]--sprint[/bold] with just the sprint name for automatic matching\\n\""," f\" • Check your project's iteration paths in Azure DevOps: Project Settings → Boards → Iterations\\n\""," f\" • Ensure your team has an active iteration configured\""," console.print(error_msg)"," raise ValueError(\"No current iteration found\")"," resolved_iteration = filters.iteration"," conditions.append(f\"[System.IterationPath] = '{resolved_iteration}'\")"," elif filters.sprint:"," elif getattr(filters, \"use_current_iteration_default\", True):"," if current_iteration:"," resolved_iteration = current_iteration"," console.print(\"[yellow]⚠ No current iteration found and no sprint/iteration filter provided[/yellow]\")"," return resolved_iteration"," def _post_wiql_handle_http_error("," e: requests.HTTPError,"," resolved_iteration: str | None,"," ) -> NoReturn:"," user_friendly_msg = None"," if e.response is not None:"," error_json = e.response.json()"," error_message = error_json.get(\"message\", \"\")"," if \"TF51011\" in error_message or \"iteration path does not exist\" in error_message.lower():"," match = re.search(r\"«'([^']+)'»\", error_message)"," bad_path = match.group(1) if match else (resolved_iteration if resolved_iteration else None)"," available_iterations = self._list_available_iterations()"," suggestions = _rich_iteration_suggestions_block(available_iterations)"," user_friendly_msg = ("," f\"[red]Error:[/red] The iteration path does not exist in Azure DevOps.\\n\""," f\"[yellow]Provided path:[/yellow] {bad_path}\\n\\n\""," f\" • Use [bold]--iteration current[/bold] to automatically use the current active iteration\\n\""," f\" • Use [bold]--sprint[/bold] with just the sprint name (e.g., 'Sprint 01') for automatic matching\\n\""," f\" • The iteration path must match exactly as shown in Azure DevOps (including project name)\\n\""," f\" • Check your project's iteration paths in Azure DevOps: Project Settings → Boards → Iterations\""," elif \"400\" in str(e.response.status_code) or \"Bad Request\" in str(e):"," f\"[red]Error:[/red] Invalid request to Azure DevOps API.\\n\""," f\"[yellow]Details:[/yellow] {error_message}\\n\\n\""," f\"Please check your parameters and try again.\""," if user_friendly_msg:"," console.print(user_friendly_msg)"," raise ValueError(f\"Iteration path error: {resolved_iteration}\") from e"," error_detail = \"\""," error_detail = f\"\\nResponse: {error_json}\""," error_detail = f\"\\nResponse status: {e.response.status_code}\""," error_msg = ("," f\"Azure DevOps API error: {e}{error_detail}\\n\""," f\"URL: {url}\\n\""," f\"Organization: {self.org}\\n\""," f\"Project: {self.project}\\n\""," f\"Base URL: {self.base_url}\\n\""," f\"Expected format: https://dev.azure.com/{{org}}/{{project}}/_apis/wit/wiql?api-version=7.1\\n\""," f\"If using Azure DevOps Server (on-premise), base_url format may differ.\""," new_exception = requests.HTTPError(error_msg)"," new_exception.response = e.response"," raise new_exception from e"," def _ado_workitems_batch_base_url(self) -> str:"," if self._is_on_premise():"," parts = [p for p in base_url_normalized.split(\"/\") if p and p not in [\"http:\", \"https:\"]]"," has_collection_in_base = \"/tfs/\" in base_url_normalized.lower() or len(parts) > 1"," if has_collection_in_base:"," return base_url_normalized"," if \"/tfs\" in base_url_normalized.lower():"," return f\"{base_url_normalized}/tfs/{self.org}\""," return f\"{base_url_normalized}/{self.org}\""," return base_url_normalized"," return f\"{base_url_normalized}/{self.org}\""," def _batch_fetch_work_items_as_backlog_items(self, work_item_ids: list[int]) -> list[BacklogItem]:"," batch_size = 200"," workitems_base_url = self._ado_workitems_batch_base_url()"," for i in range(0, len(work_item_ids), batch_size):"," batch = work_item_ids[i : i + batch_size]"," ids_str = \",\".join(str(wi_id) for wi_id in batch)"," url = f\"{workitems_base_url}/_apis/wit/workitems?api-version=7.1\""," params = {\"ids\": ids_str, \"$expand\": \"all\"}"," workitems_headers = {"," debug_print(f\"[dim]ADO WorkItems URL: {url}&ids={ids_str}[/dim]\")"," response = self._ado_get(url, headers=workitems_headers, params=params, timeout=30)"," if is_debug_mode():"," debug_log_operation("," \"ado_workitems_get\","," url,"," str(response.status_code),"," error=None if response.ok else (response.text[:200] if response.text else None),"," except requests.HTTPError as e:"," \"error\","," error=str(e.response.status_code) if e.response is not None else str(e),"," error_detail = \"\""," if e.response is not None:"," try:"," error_json = e.response.json()"," error_detail = f\"\\nResponse: {error_json}\""," except Exception:"," error_detail = f\"\\nResponse status: {e.response.status_code}\""," error_msg = ("," f\"Azure DevOps API error: {e}{error_detail}\\n\""," f\"URL: {url}\\n\""," f\"Organization: {self.org}\\n\""," f\"Project: {self.project}\\n\""," f\"Base URL: {self.base_url}\\n\""," f\"Expected format: https://dev.azure.com/{{org}}/{{project}}/_apis/wit/workitems?ids={{ids}}&api-version=7.1\\n\""," f\"If using Azure DevOps Server (on-premise), base_url format may differ.\""," new_exception = requests.HTTPError(error_msg)"," new_exception.response = e.response"," raise new_exception from e"," work_items_data = response.json()"," for work_item in work_items_data.get(\"value\", []):"," backlog_item = convert_ado_work_item_to_backlog_item("," work_item,"," provider=\"ado\","," base_url=self.base_url,"," org=self.org,"," project_name=self.project,"," items.append(backlog_item)"," return items"," def _apply_post_fetch_filters_after_wiql("," include_iteration: bool = False,"," sprint_apply_current: bool | None = None,"," echo_sprint_value_error: bool = True,"," filtered_items = self._filter_backlog_items_state_assignee_labels(filtered_items, filters)"," if include_iteration:"," filtered_items = self._apply_iteration_filter_post_fetch(filtered_items, filters)"," filtered_items = self._apply_sprint_filter_post_fetch("," filtered_items,"," sprint_apply_current=sprint_apply_current,"," echo_sprint_value_error=echo_sprint_value_error,"," filtered_items = self._filter_backlog_items_by_release_post_fetch(filtered_items, filters)"," return self._apply_backlog_limit_post_fetch(filtered_items, filters)"," Fetch Azure DevOps work items matching the specified filters."," Uses ADO Work Items API to query work items."," \"Azure DevOps API token required to fetch backlog items.\\n\""," \"Options:\\n\""," \" 2. Use --ado-token option\\n\""," \" 3. Store token via specfact backlog auth azure-devops\""," \"org (organization) required to fetch backlog items.\\n\""," \"For Azure DevOps Services (cloud), org is always required.\\n\""," \"For Azure DevOps Server (on-premise), org is the collection name.\\n\""," \"Provide via --ado-org option or ensure it's set in adapter configuration.\""," msg = \"project required to fetch backlog items. Provide via --ado-project option.\""," direct_result = self._try_fetch_backlog_by_direct_issue(filters)"," if direct_result is not None:"," return direct_result"," wiql_parts = [\"SELECT [System.Id], [System.Title], [System.State], [System.WorkItemType]\"]"," wiql_parts.append(\"FROM WorkItems\")"," wiql_parts.append(\"WHERE [System.TeamProject] = @project\")"," conditions: list[str] = []"," if filters.area:"," conditions.append(f\"[System.AreaPath] = '{filters.area}'\")"," resolved_iteration = self._wiql_append_iteration_conditions(filters, conditions)"," if conditions:"," wiql_parts.append(\"AND \" + \" AND \".join(conditions))"," wiql = \" \".join(wiql_parts)"," url = self._build_ado_url(\"_apis/wit/wiql\", api_version=\"7.1\")"," payload = {\"query\": wiql}"," debug_print(f\"[dim]ADO WIQL URL: {url}[/dim]\")"," if \"Authorization\" in headers:"," auth_header_preview = ("," headers[\"Authorization\"][:20] + \"...\""," if len(headers[\"Authorization\"]) > 20"," else headers[\"Authorization\"]"," debug_print(f\"[dim]ADO Auth: {auth_header_preview}[/dim]\")"," debug_print(\"[yellow]Warning: No Authorization header in request[/yellow]\")"," response = self._ado_post(url, headers=headers, json=payload, timeout=30)"," self._post_wiql_handle_http_error(e, url, resolved_iteration)"," query_result = response.json()"," work_item_ids = [item[\"id\"] for item in query_result.get(\"workItems\", [])]"," if not work_item_ids:"," items = self._batch_fetch_work_items_as_backlog_items(work_item_ids)"," return self._apply_post_fetch_filters_after_wiql(items, filters)"," def _build_create_issue_patch_document("," description = str(payload.get(\"description\") or payload.get(\"body\") or \"\").strip()"," description = self._strip_leading_description_heading(description)"," field_rendering_format = \"Markdown\" if description_format != \"classic\" else \"Html\""," custom_mapping_file = os.environ.get(\"SPECFACT_ADO_CUSTOM_MAPPING\")"," ado_mapper = AdoFieldMapper(custom_mapping_file=custom_mapping_file)"," description_field = ado_mapper.resolve_write_target_field(\"description\") or \"System.Description\""," acceptance_criteria_field = ("," ado_mapper.resolve_write_target_field(\"acceptance_criteria\") or \"Microsoft.VSTS.Common.AcceptanceCriteria\""," priority_field = ado_mapper.resolve_write_target_field(\"priority\") or \"Microsoft.VSTS.Common.Priority\""," story_points_field = ("," ado_mapper.resolve_write_target_field(\"story_points\") or \"Microsoft.VSTS.Scheduling.StoryPoints\""," patch_document: list[dict[str, Any]] = ["," {\"op\": \"add\", \"path\": f\"/fields/{description_field}\", \"value\": description},"," {\"op\": \"add\", \"path\": f\"/multilineFieldsFormat/{description_field}\", \"value\": field_rendering_format},"," _ado_patch_doc_append_acceptance_criteria_create_issue("," patch_document,"," acceptance_criteria=acceptance_criteria,"," acceptance_criteria_field=acceptance_criteria_field,"," field_rendering_format=field_rendering_format,"," _ado_patch_doc_append_priority_story_points_create_issue("," payload=payload,"," priority_field=priority_field,"," story_points_field=story_points_field,"," _ado_patch_doc_append_provider_fields_create_issue(patch_document, payload)"," _ado_patch_doc_append_sprint_parent_create_issue("," org=org,"," project=project,"," return patch_document"," \"\"\"Create an Azure DevOps work item from provider-agnostic backlog payload.\"\"\""," org, project = self._resolve_graph_project_context(project_id)"," raw_type = str(payload.get(\"type\") or \"task\").strip().lower()"," type_mapping = {"," \"epic\": \"Epic\","," \"feature\": \"Feature\","," \"story\": \"User Story\","," \"user story\": \"User Story\","," \"task\": \"Task\","," \"bug\": \"Bug\","," \"spike\": \"Task\","," work_item_type = type_mapping.get(raw_type, \"Task\")"," patch_document = self._build_create_issue_patch_document(org, project, payload, title=title)"," lambda: requests.post(url, json=cast(Any, patch_document), headers=headers, timeout=30),"," created_id = str(created.get(\"id\") or \"\")"," html_url = str(created.get(\"_links\", {}).get(\"html\", {}).get(\"href\") or \"\")"," fallback_url = str(created.get(\"url\") or \"\")"," \"id\": created_id,"," \"key\": created_id,"," \"url\": html_url or fallback_url,"," def _get_org_project(self) -> tuple[str | None, str | None]:"," \"\"\"Query: return current org and project without mutation.\"\"\""," return self.org, self.project"," def _set_org_project(self, org: str | None, project: str | None) -> None:"," \"\"\"Command: set org and project without reading current state.\"\"\""," \"\"\"Fetch all ADO work items as provider-agnostic dictionaries for graph building.\"\"\""," resolved_org, resolved_project = self._resolve_graph_project_context(project_id)"," saved_org, saved_project = self._get_org_project()"," self._set_org_project(resolved_org, resolved_project)"," return [item.model_dump() for item in self.fetch_backlog_items(backlog_filters)]"," self._set_org_project(saved_org, saved_project)"," def _edges_from_ado_work_item_relations(self, item: dict[str, Any], item_id: str) -> list[tuple[str, str, str]]:"," \"\"\"Collect normalized graph edges from an ADO work item's relation list.\"\"\""," for relation in _flatten_issue_relation_dicts(cast(dict[str, Any], item)):"," if not isinstance(relation, dict):"," rel_name = str(relation.get(\"rel\") or relation.get(\"relation\") or relation.get(\"type\") or \"\").lower()"," target_ref = str(relation.get(\"url\") or relation.get(\"target\") or \"\")"," target_wi = self._extract_work_item_id_from_reference(target_ref)"," if not target_wi:"," edge = _ado_graph_edge_from_relation(rel_name, item_id, target_wi)"," if edge:"," edges.append(edge)"," \"\"\"Fetch ADO relationship edges for graph building.\"\"\""," for item in self.fetch_all_issues(project_id):"," item_id = str(item.get(\"id\") or item.get(\"key\") or \"\").strip()"," for src, tgt, et in self._edges_from_ado_work_item_relations(item, item_id):"," _add_edge(src, tgt, et)"," @ensure(lambda result: isinstance(result, tuple) and len(result) == 2, \"Must return (org, project) tuple\")"," def _resolve_graph_project_context(self, project_id: str) -> tuple[str, str]:"," \"\"\"Resolve org/project context for graph APIs from linked project_id and adapter defaults.\"\"\""," normalized = project_id.strip()"," if \"/\" in normalized:"," org, project = normalized.split(\"/\", 1)"," resolved_org = org.strip()"," resolved_project = project.strip()"," if resolved_org and resolved_project:"," return resolved_org, resolved_project"," raise ValueError(f\"Invalid ADO project_id format: {project_id!r}. Expected '/'.\")"," # Backward compatibility: allow project-only identifiers when adapter org already exists."," return self.org, normalized"," raise ValueError("," f\"ADO project_id '{project_id}' missing organization. Use '/' or configure adapter org.\""," @ensure(lambda result: isinstance(result, str), \"Work item id extraction must return str\")"," def _extract_work_item_id_from_reference(self, reference: str) -> str:"," \"\"\"Extract ADO work item id from relation reference URL/string.\"\"\""," if not reference:"," match = re.search(r\"/workitems/(\\d+)\", reference, flags=re.IGNORECASE)"," return match.group(1) if match else \"\""," \"\"\"Whether this adapter can add comments (requires token, org, project).\"\"\""," return bool(self.api_token and self.org and self.project)"," work_item_id = int(item.id)"," self._add_work_item_comment(self.org, self.project, work_item_id, comment)"," if not item.id.isdigit():"," raw = self._get_work_item_comments(self.org, self.project, int(item.id))"," comment_texts: list[str] = []"," for comment in raw:"," if not isinstance(comment, dict):"," text = comment.get(\"text\") or comment.get(\"body\")"," if isinstance(text, str):"," stripped = text.strip()"," if stripped:"," comment_texts.append(stripped)"," return comment_texts"," def _patch_ops_backlog_title_and_body("," update_fields: list[str] | None,"," ado_mapper: AdoFieldMapper,"," provider_field_names: set[str],"," operations: list[dict[str, Any]] = []"," operations.append({\"op\": \"replace\", \"path\": \"/fields/System.Title\", \"value\": item.title})"," raw_body = item.body_markdown"," markdown_content = raw_body if raw_body is not None else \"\""," markdown_content = self._strip_leading_description_heading(markdown_content)"," todo_pattern = r\"^(\\s*)[-*]\\s*\\[TODO[:\\s]+([^\\]]+)\\](.*)$\""," markdown_content = re.sub("," todo_pattern,"," r\"\\1- [ ] \\2\","," markdown_content,"," flags=re.MULTILINE | re.IGNORECASE,"," description_field = ("," ado_mapper.resolve_write_target_field(\"description\", provider_field_names) or \"System.Description\""," operations.append({\"op\": \"add\", \"path\": f\"/multilineFieldsFormat/{description_field}\", \"value\": \"Markdown\"})"," operations.append({\"op\": \"replace\", \"path\": f\"/fields/{description_field}\", \"value\": markdown_content})"," return operations"," def _patch_ops_backlog_mapped_optional_fields("," operations.extend("," _ado_patch_ops_optional_acceptance_criteria(item, update_fields, ado_mapper, provider_field_names)"," operations.extend(_ado_patch_ops_optional_story_points(item, update_fields, ado_mapper, provider_field_names))"," operations.extend(_ado_patch_ops_optional_business_value(item, update_fields, ado_mapper, provider_field_names))"," operations.extend(_ado_patch_ops_optional_priority(item, update_fields, ado_mapper, provider_field_names))"," def _build_update_backlog_patch_operations("," self, item: BacklogItem, update_fields: list[str] | None"," provider_field_names: set[str] = set()"," provider_fields_payload = item.provider_fields.get(\"fields\")"," if isinstance(provider_fields_payload, dict):"," provider_field_names = {str(field_name) for field_name in provider_fields_payload}"," operations.extend(self._patch_ops_backlog_title_and_body(item, update_fields, ado_mapper, provider_field_names))"," self._patch_ops_backlog_mapped_optional_fields(item, update_fields, ado_mapper, provider_field_names)"," operations.append({\"op\": \"replace\", \"path\": \"/fields/System.State\", \"value\": item.state})"," def _backlog_ops_without_multiline_format(operations: list[dict[str, Any]]) -> list[dict[str, Any]]:"," return [op for op in operations if not (op.get(\"path\") or \"\").startswith(\"/multilineFieldsFormat/\")]"," def _backlog_ops_replace_multiline_add_with_replace(operations: list[dict[str, Any]]) -> list[dict[str, Any]]:"," out: list[dict[str, Any]] = []"," for op in operations:"," path = op.get(\"path\") or \"\""," if path.startswith(\"/multilineFieldsFormat/\"):"," out.append({\"op\": \"replace\", \"path\": path, \"value\": op[\"value\"]})"," out.append(op)"," def _backlog_ops_convert_markdown_fields_to_html(operations: list[dict[str, Any]]) -> list[dict[str, Any]]:"," markdown_formatted_fields = {"," str(op.get(\"path\", \"\")).replace(\"/multilineFieldsFormat/\", \"\", 1)"," for op in operations"," if str(op.get(\"path\", \"\")).startswith(\"/multilineFieldsFormat/\")"," and str(op.get(\"value\", \"\")).lower() == \"markdown\""," operations_html = ["," dict(op) for op in operations if not (op.get(\"path\") or \"\").startswith(\"/multilineFieldsFormat/\")"," for op in operations_html:"," field_path = str(op.get(\"path\", \"\"))"," if not field_path.startswith(\"/fields/\"):"," field_name = field_path.replace(\"/fields/\", \"\", 1)"," if field_name in markdown_formatted_fields:"," op[\"value\"] = _markdown_to_html_ado_fallback(str(op.get(\"value\") or \"\"))"," return operations_html"," def _ado_http_error_message(response: requests.Response | None) -> str:"," if not response:"," err = response.json()"," return str(err.get(\"message\", \"\") or \"\")"," def _backlog_patch_try_without_multiline_format("," headers: dict[str, Any],"," operations: list[dict[str, Any]],"," ) -> requests.Response | None:"," operations_no_format = self._backlog_ops_without_multiline_format(operations)"," if operations_no_format == operations:"," resp = requests.patch(url, headers=headers, json=cast(Any, operations_no_format), timeout=30)"," resp.raise_for_status()"," return resp"," except requests.HTTPError as retry_error:"," _log_ado_patch_failure("," retry_error.response,"," operations_no_format,"," context=str(retry_error),"," def _backlog_patch_try_replace_multiline_add("," operations_replace = self._backlog_ops_replace_multiline_add_with_replace(operations)"," resp = requests.patch(url, headers=headers, json=cast(Any, operations_replace), timeout=30)"," except requests.HTTPError:"," def _backlog_patch_try_html_conversion("," user_msg: str,"," \"[yellow]⚠ Markdown format metadata not supported, converting multiline markdown fields to HTML[/yellow]\""," operations_html = self._backlog_ops_convert_markdown_fields_to_html(operations)"," resp = requests.patch(url, headers=headers, json=cast(Any, operations_html), timeout=30)"," def _execute_backlog_patch_with_fallbacks("," ) -> requests.Response:"," return self._request_with_retry("," lambda: requests.patch(url, headers=headers, json=cast(Any, operations), timeout=30)"," user_msg = _log_ado_patch_failure(e.response, operations, url)"," response: requests.Response | None = None"," if e.response and e.response.status_code in (400, 422):"," error_message = self._ado_http_error_message(e.response)"," response = self._backlog_patch_try_without_multiline_format(url, headers, operations)"," if response is None and ("," \"already exists\" in error_message.lower() or \"cannot add\" in error_message.lower()"," response = self._backlog_patch_try_replace_multiline_add(url, headers, operations)"," if response is None:"," response = self._backlog_patch_try_html_conversion(url, headers, operations, user_msg)"," if response is None:"," console.print(f\"[bold red]✗[/bold red] {user_msg}\")"," raise"," return response"," lambda update_fields: update_fields is None or isinstance(update_fields, list),"," Update an Azure DevOps work item."," Updates the work item title and/or description based on update_fields."," msg = \"Azure DevOps API token required to update backlog items\""," msg = \"org and project required to update backlog items\""," url = self._build_ado_url(f\"_apis/wit/workitems/{work_item_id}\", api_version=\"7.1\")"," operations = self._build_update_backlog_patch_operations(item, update_fields)"," response = self._execute_backlog_patch_with_fallbacks(url, headers, operations)"," updated_work_item = response.json()"," # Store format metadata in provider_fields for round-trip"," if hasattr(item, \"provider_fields\") and isinstance(item.provider_fields, dict):"," item.provider_fields[\"description_format\"] = \"Markdown\""," item.provider_fields[\"description_markdown\"] = item.body_markdown"," updated_work_item,","BRIDGE_PROTOCOL_REGISTRY.register_implementation(\"backlog_graph\", \"ado\", AdoAdapter)",[7757,7758,7759,7760,7761,7762,7763,7764],{"_461":45,"_45":7777},{"_461":7766,"_183":5946,"_1024":7775,"_1025":7776},{"_461":7766,"_183":4063,"_1024":7773,"_1025":7774},{"_461":7766,"_183":1767,"_1024":7771,"_1025":7772},{"_461":7766,"_183":1455,"_1024":7769,"_1025":7770},{"_461":45,"_45":7768},{"_461":7766,"_183":1018,"_1024":7767,"_1025":827},{"_461":45,"_45":7765},"\n* Committed the fix on this branch: `f3474484bbb88330cc17917679514ca8ff645e52`.\n\n**Testing**\n* ✅ `hatch run format`\n* ✅ `hatch run type-check`\n* ✅ `hatch run lint`\n* ✅ `python -m pytest tests/unit/adapters/test_ado_backlog_adapter.py tests/unit/registry/test_module_installer.py tests/unit/specfact_cli/registry/test_versioned_bundle_deps.py -q`","repo_file_citation",770,"\n* Updated ADO adapter auth-header tests to assert via `str(...)` before `startswith(...)`, which keeps test intent while remaining type-safe with `str | bytes` header values.",168,192,2446,2454,563,574,1942,1977,"### Summary\n* Fixed the CI type-check regressions (24 errors) by widening request-header typing from `dict[str, str]` to `dict[str, str | bytes]` in the affected adapter/sync/validator paths so they match `requests`’ `HeadersType` expectations under strict basedpyright. This includes ADO auth helper signatures and request wrapper arguments, plus explicit header dict annotations in GitHub and sync callsites.","follow_up_diff",{"_115":991,"_994":201,"_150":7780,"_995":-5,"_555":7781,"_997":7782,"_999":7783,"_1001":7784,"_1003":7784,"_1005":1006,"_1007":7785},"82f478d2de62485981fb2b09005c948c0fb086bb","diff --git a/src/specfact_cli/adapters/ado.py b/src/specfact_cli/adapters/ado.py\nindex faa793b4bb96d9ace2e520a166eb679467b49dce..6e94ada870707b1ae76c59b64bf5fdfb1fcaa50e 100644\n--- a/src/specfact_cli/adapters/ado.py\n+++ b/src/specfact_cli/adapters/ado.py\n@@ -1917,85 +1917,85 @@ class AdoAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n \n # Create credential with same cache - it will use cached refresh token\n credential = DeviceCodeCredential(cache_persistence_options=cache_options)\n # Use the same resource and scopes as auth command\n # Note: Refresh tokens are automatically obtained via persistent token cache\n # offline_access is a reserved scope and cannot be explicitly requested\n azure_devops_resource = \"499b84ac-1321-427f-aa17-267ca6975798/.default\"\n azure_devops_scopes = [azure_devops_resource]\n token = _get_access_token(credential, azure_devops_scopes)\n \n # Return refreshed token data\n from datetime import UTC, datetime\n \n expires_at = datetime.fromtimestamp(token.expires_on, tz=UTC).isoformat()\n return {\n \"access_token\": token.token,\n \"token_type\": \"bearer\",\n \"expires_at\": expires_at,\n \"resource\": azure_devops_resource,\n \"issued_at\": datetime.now(tz=UTC).isoformat(),\n }\n except Exception:\n # Refresh failed (no cached refresh token, refresh token expired, etc.)\n return None\n \n- def _auth_headers(self) -> dict[str, str]:\n+ def _auth_headers(self) -> dict[str, str | bytes]:\n \"\"\"Return authorization headers based on token type.\"\"\"\n if not self.api_token:\n return {}\n if self.auth_scheme == \"bearer\":\n return {\"Authorization\": f\"Bearer {self.api_token}\"}\n return {\"Authorization\": f\"Basic {self._encode_pat(self.api_token)}\"}\n \n @beartype\n @ensure(lambda result: hasattr(result, \"raise_for_status\"), \"Result must support raise_for_status\")\n def _ado_get(\n self,\n url: str,\n *,\n- headers: dict[str, str] | None = None,\n+ headers: dict[str, str | bytes] | None = None,\n params: dict[str, Any] | None = None,\n timeout: int = 30,\n retry_on_ambiguous_transport: bool = True,\n ) -> Any:\n \"\"\"Execute an idempotent ADO GET with retry policy for transient failures.\"\"\"\n return cast(\n Any,\n self._request_with_retry(\n lambda: requests.get(url, headers=headers, params=params, timeout=timeout),\n retry_on_ambiguous_transport=retry_on_ambiguous_transport,\n ),\n )\n \n @beartype\n @ensure(lambda result: hasattr(result, \"raise_for_status\"), \"Result must support raise_for_status\")\n def _ado_post(\n self,\n url: str,\n *,\n- headers: dict[str, str] | None = None,\n+ headers: dict[str, str | bytes] | None = None,\n params: dict[str, Any] | None = None,\n json: dict[str, Any] | None = None,\n timeout: int = 30,\n retry_on_ambiguous_transport: bool = True,\n ) -> Any:\n \"\"\"Execute ADO POST with retry policy. Safe for read-only WIQL endpoints.\"\"\"\n request_kwargs: dict[str, Any] = {\"headers\": headers, \"json\": json, \"timeout\": timeout}\n if params:\n request_kwargs[\"params\"] = params\n return cast(\n Any,\n self._request_with_retry(\n lambda: requests.post(url, **request_kwargs),\n retry_on_ambiguous_transport=retry_on_ambiguous_transport,\n ),\n )\n \n def _work_item_exists(self, work_item_id: int | str, org: str, project: str) -> bool:\n \"\"\"\n Check if a work item exists in Azure DevOps.\n \n Args:\n work_item_id: Work item ID to check\n org: Azure DevOps organization\n project: Azure DevOps project\ndiff --git a/src/specfact_cli/adapters/github.py b/src/specfact_cli/adapters/github.py\nindex 97d567276bdfe6c9c4c39754b51dab448fb38288..7d1e1859e2219c6bca491076ef6e4288184ba96d 100644\n--- a/src/specfact_cli/adapters/github.py\n+++ b/src/specfact_cli/adapters/github.py\n@@ -539,51 +539,51 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n if story_points is not None:\n labels.append(f\"story-points:{story_points}\")\n return labels\n \n def _build_issue_search_query(self, filters: BacklogFilters) -> str:\n \"\"\"Build the GitHub search query for backlog issue fetches.\"\"\"\n query_parts = [f\"repo:{self.repo_owner}/{self.repo_name}\", \"type:issue\"]\n if filters.state:\n normalized_state = BacklogFilters.normalize_filter_value(filters.state) or filters.state\n query_parts.append(f\"state:{normalized_state}\")\n if filters.assignee:\n assignee_value = filters.assignee.lstrip(\"@\")\n normalized_assignee_value = BacklogFilters.normalize_filter_value(assignee_value)\n query_parts.append(\"assignee:@me\" if normalized_assignee_value == \"me\" else f\"assignee:{assignee_value}\")\n if filters.labels:\n query_parts.extend(f\"label:{label}\" for label in filters.labels)\n if filters.search:\n query_parts.append(filters.search)\n return \" \".join(query_parts)\n \n def _search_github_issues(self, query: str) -> list[BacklogItem]:\n \"\"\"Run a GitHub issue search and convert results to backlog items.\"\"\"\n from specfact_cli.backlog.converter import convert_github_issue_to_backlog_item\n \n url = f\"{self.base_url}/search/issues\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n params = {\"q\": query, \"per_page\": 100}\n items: list[BacklogItem] = []\n page = 1\n while True:\n params[\"page\"] = page\n response = self._request_with_retry(lambda: requests.get(url, headers=headers, params=params, timeout=30))\n response.raise_for_status()\n issues = response.json().get(\"items\", [])\n if not issues:\n return items\n items.extend(convert_github_issue_to_backlog_item(issue, provider=\"github\") for issue in issues)\n if len(issues) < 100:\n return items\n page += 1\n \n @staticmethod\n def _graph_type_alias_map() -> dict[str, str]:\n \"\"\"Return normalized graph type aliases.\"\"\"\n return {\n \"epic\": \"epic\",\n \"feature\": \"feature\",\n \"story\": \"story\",\n@@ -1210,51 +1210,51 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n if not repo_owner or not repo_name:\n msg = \"GitHub repository owner and name required. Provide via --repo-owner and --repo-name or bridge config\"\n raise ValueError(msg)\n \n return self._export_github_artifact_dispatch(artifact_key, artifact_data, repo_owner, repo_name, bridge_config)\n \n @beartype\n @require(lambda item_ref: isinstance(item_ref, str) and len(item_ref) > 0, \"Item reference must be non-empty\")\n @ensure(lambda result: isinstance(result, dict), \"Must return dict with issue data\")\n def fetch_backlog_item(self, item_ref: str) -> dict[str, Any]:\n \"\"\"\n Fetch GitHub issue data by ID or URL.\n \n Args:\n item_ref: Issue number, owner/repo#number, or issue URL\n \n Returns:\n Issue data dict from GitHub API\n \"\"\"\n if not self.api_token:\n msg = \"GitHub API token required to fetch backlog items\"\n raise ValueError(msg)\n \n repo_owner, repo_name, issue_number = self._parse_issue_reference(item_ref)\n url = f\"{self.base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_number}\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n response = requests.get(url, headers=headers, timeout=30)\n if is_debug_mode():\n debug_log_operation(\n \"github_api_get\",\n url,\n str(response.status_code),\n error=None if response.ok else (response.text[:200] if response.text else None),\n )\n response.raise_for_status()\n return response.json()\n \n def _extract_repo_from_issue(self, issue_data: dict[str, Any]) -> str | None:\n \"\"\"\n Extract repository identifier (owner/repo) from GitHub issue data.\n \n Args:\n issue_data: GitHub issue data dict\n \n Returns:\n Repository identifier string or None if not found\n \"\"\"\n candidates = [\n@@ -1476,51 +1476,51 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n \n body = self._render_issue_body(\n _IssueBodyRenderInput(\n title=title,\n description=description,\n rationale=rationale,\n impact=impact,\n change_id=change_id,\n raw_body=raw_body,\n )\n )\n \n # Check for API token before making request\n if not self.api_token:\n msg = (\n \"GitHub API token required to create issues. Options:\\n\"\n \" 1. Set GITHUB_TOKEN environment variable\\n\"\n \" 2. Use --github-token option\\n\"\n \" 3. Use GitHub CLI authentication (gh auth login)\\n\"\n \" 4. Store token via specfact backlog auth github\"\n )\n raise ValueError(msg)\n \n # Create issue via GitHub API\n url = f\"{self.base_url}/repos/{repo_owner}/{repo_name}/issues\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n # Determine issue state based on proposal status\n # Check if source_state and source_type are provided (from cross-adapter sync)\n issue_state = self._resolve_issue_state(proposal_data, status)\n state_reason = self._resolve_state_reason(status)\n \n payload = {\n \"title\": title,\n \"body\": body,\n \"labels\": self._get_labels_for_status(status),\n \"state\": issue_state,\n }\n if state_reason:\n payload[\"state_reason\"] = state_reason\n \n try:\n response = self._request_with_retry(\n lambda: requests.post(url, json=payload, headers=headers, timeout=30),\n retry_on_ambiguous_transport=False,\n )\n issue_data = response.json()\n \n # If issue was created as closed, add a comment explaining why\n@@ -1570,137 +1570,137 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n target_repo = f\"{repo_owner}/{repo_name}\"\n issue_number = self._resolve_issue_number_from_tracking(source_tracking, repo_owner, repo_name)\n \n if not issue_number:\n msg = (\n f\"Issue number not found in source_tracking for repository {target_repo}. Issue must be created first.\"\n )\n raise ValueError(msg)\n \n status = proposal_data.get(\"status\", \"proposed\")\n title = proposal_data.get(\"title\", \"Untitled\")\n \n # Map status to GitHub issue state and comment\n # Check if source_state and source_type are provided (from cross-adapter sync)\n issue_state = self._resolve_issue_state(proposal_data, status)\n should_close = issue_state == \"closed\"\n source_tracking = proposal_data.get(\"source_tracking\", {})\n # Note: code_repo_path not available in _update_issue_status context\n comment_text = self._get_status_comment(status, title, source_tracking, None)\n \n # Map status to GitHub state_reason\n state_reason = self._resolve_state_reason(status)\n \n # Update issue state\n url = f\"{self.base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_number}\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n payload = {\"state\": \"closed\" if should_close else \"open\"}\n if state_reason:\n payload[\"state_reason\"] = state_reason\n \n try:\n response = self._request_with_retry(\n lambda: requests.patch(url, json=cast(Any, payload), headers=headers, timeout=30)\n )\n issue_data = response.json()\n \n # Add comment explaining status change\n if comment_text:\n self._add_issue_comment(repo_owner, repo_name, issue_number, comment_text)\n \n return {\n \"issue_number\": issue_data[\"number\"],\n \"issue_url\": issue_data[\"html_url\"],\n \"state\": issue_data[\"state\"],\n }\n except requests.RequestException as e:\n msg = f\"Failed to update GitHub issue #{issue_number}: {e}\"\n console.print(f\"[bold red]✗[/bold red] {msg}\")\n raise\n \n def _get_issue_comments(self, repo_owner: str, repo_name: str, issue_number: int) -> list[dict[str, Any]]:\n \"\"\"\n Fetch comments for a GitHub issue.\n \n Args:\n repo_owner: GitHub repository owner\n repo_name: GitHub repository name\n issue_number: Issue number\n \n Returns:\n List of comment dicts with 'body' field, or empty list on error\n \"\"\"\n if not self.api_token:\n return []\n \n url = f\"{self.base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_number}/comments\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n \n try:\n response = requests.get(url, headers=headers, timeout=30)\n response.raise_for_status()\n return response.json()\n except requests.RequestException:\n # Return empty list on error - comments are optional\n return []\n \n def _add_issue_comment(self, repo_owner: str, repo_name: str, issue_number: int, comment: str) -> None:\n \"\"\"\n Add comment to GitHub issue.\n \n Args:\n repo_owner: GitHub repository owner\n repo_name: GitHub repository name\n issue_number: Issue number\n comment: Comment text\n \"\"\"\n url = f\"{self.base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_number}/comments\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n payload = {\"body\": comment}\n \n try:\n self._request_with_retry(\n lambda: requests.post(url, json=cast(Any, payload), headers=headers, timeout=30),\n retry_on_ambiguous_transport=False,\n )\n except requests.RequestException as e:\n # Log but don't fail - comment is non-critical\n console.print(f\"[yellow]⚠[/yellow] Failed to add comment to issue #{issue_number}: {e}\")\n \n def _fetch_issue_snapshot(self, repo_owner: str, repo_name: str, issue_number: int) -> tuple[str, str, str]:\n \"\"\"Fetch current issue body, title, and state for preservation-aware updates.\"\"\"\n url = f\"{self.base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_number}\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n try:\n response = requests.get(url, headers=headers, timeout=30)\n response.raise_for_status()\n issue_data = response.json()\n return (\n issue_data.get(\"body\", \"\") or \"\",\n issue_data.get(\"title\", \"\") or \"\",\n issue_data.get(\"state\", \"open\"),\n )\n except requests.RequestException:\n return \"\", \"\", \"open\"\n \n @staticmethod\n def _preserved_issue_sections(current_body: str, change_id: str) -> list[str]:\n \"\"\"Extract non-OpenSpec sections to preserve during issue body rewrites.\"\"\"\n if not current_body:\n return []\n metadata_marker = f\"*OpenSpec Change Proposal: `{change_id}`*\"\n if metadata_marker not in current_body:\n return []\n _, after_marker = current_body.split(metadata_marker, 1)\n preserved_content = after_marker.strip()\n@@ -1739,51 +1739,51 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n description = proposal_data.get(\"description\", \"\")\n rationale = proposal_data.get(\"rationale\", \"\")\n impact = proposal_data.get(\"impact\", \"\")\n change_id = str(proposal_data.get(\"change_id\", \"unknown\"))\n status = proposal_data.get(\"status\", \"proposed\")\n raw_title, raw_body = self._extract_raw_fields(proposal_data)\n if raw_title:\n title = raw_title\n \n current_body, current_title, current_state = self._fetch_issue_snapshot(repo_owner, repo_name, issue_number)\n preserved_sections = self._preserved_issue_sections(current_body, change_id)\n body = self._render_issue_body(\n _IssueBodyRenderInput(\n title=title,\n description=description,\n rationale=rationale,\n impact=impact,\n change_id=change_id,\n raw_body=raw_body,\n preserved_sections=preserved_sections,\n )\n )\n \n # Update issue body via GitHub API PATCH\n url = f\"{self.base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_number}\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n desired_state = self._resolve_issue_state(proposal_data, status)\n state_reason = self._resolve_state_reason(status)\n payload = self._issue_body_update_payload(\n body, title, current_title, current_state, desired_state, state_reason\n )\n \n try:\n response = self._request_with_retry(lambda: requests.patch(url, json=payload, headers=headers, timeout=30))\n issue_data = response.json()\n self._add_issue_status_comment(\n _IssueStatusCommentInput(\n proposal_data,\n repo_owner,\n repo_name,\n issue_number,\n code_repo_path,\n payload,\n current_state,\n status,\n title,\n )\n )\n@@ -1949,51 +1949,51 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n status = proposal.status\n source_tracking = proposal.source_tracking\n else:\n status = proposal.get(\"status\", \"proposed\")\n source_tracking = proposal.get(\"source_tracking\")\n \n if not source_tracking:\n msg = \"Source tracking required for status sync (issue must be created first)\"\n raise ValueError(msg)\n \n # Get issue number from source_tracking (handle both dict and list formats)\n issue_number = None\n target_repo = f\"{repo_owner}/{repo_name}\"\n \n issue_number = self._resolve_issue_number_from_tracking(source_tracking, repo_owner, repo_name)\n \n if not issue_number:\n msg = f\"Issue number not found in source_tracking for repository {target_repo}\"\n raise ValueError(msg)\n \n # Map OpenSpec status to GitHub labels\n new_labels = self.map_openspec_status_to_backlog(status)\n \n # Get current issue to retrieve existing labels\n url = f\"{self.base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_number}\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n \n try:\n response = requests.get(url, headers=headers, timeout=30)\n response.raise_for_status()\n current_issue = response.json()\n current_labels = [label.get(\"name\", \"\") for label in current_issue.get(\"labels\", [])]\n status_labels = [\"in-progress\", \"completed\", \"deprecated\", \"wontfix\"]\n keep_labels = [label for label in current_labels if label not in status_labels and label != \"openspec\"]\n all_labels = self._dedupe_strings(keep_labels + new_labels)\n \n # Update issue labels\n patch_url = f\"{self.base_url}/repos/{repo_owner}/{repo_name}/issues/{issue_number}\"\n patch_payload = {\"labels\": all_labels}\n \n self._request_with_retry(\n lambda: requests.patch(patch_url, json=cast(Any, patch_payload), headers=headers, timeout=30)\n )\n \n return {\n \"issue_number\": current_issue.get(\"number\", issue_number), # Use API response number (int)\n \"issue_url\": current_issue.get(\"html_url\", \"\"),\n \"labels_updated\": True,\n@@ -2605,51 +2605,51 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n if not self.api_token:\n msg = \"GitHub API token required to fetch backlog items\"\n raise ValueError(msg)\n \n if not self.repo_owner or not self.repo_name:\n msg = \"repo_owner and repo_name required to fetch backlog items\"\n raise ValueError(msg)\n \n if filters.issue_id:\n direct_item = self._fetch_backlog_item_by_id(filters.issue_id)\n direct_items = [direct_item] if direct_item is not None else []\n return self._apply_backlog_post_filters(direct_items, filters)\n \n items = self._search_github_issues(self._build_issue_search_query(filters))\n \n return self._apply_backlog_post_filters(items, filters)\n \n @beartype\n def _fetch_backlog_item_by_id(self, issue_id: str) -> BacklogItem | None:\n \"\"\"Fetch a single GitHub issue by number for deterministic ID lookup flows.\"\"\"\n normalized_id = issue_id.strip().lstrip(\"#\")\n if not normalized_id:\n return None\n \n url = f\"{self.base_url}/repos/{self.repo_owner}/{self.repo_name}/issues/{normalized_id}\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n try:\n response = self._request_with_retry(lambda: requests.get(url, headers=headers, timeout=30))\n response.raise_for_status()\n except requests.HTTPError as exc:\n status_code = exc.response.status_code if exc.response is not None else None\n if status_code == 404:\n return None\n raise\n \n issue_payload = response.json()\n if not isinstance(issue_payload, dict):\n return None\n ip = _as_str_dict(issue_payload)\n if ip.get(\"pull_request\") is not None:\n # Backlog issue commands should not resolve pull requests.\n return None\n \n from specfact_cli.backlog.converter import convert_github_issue_to_backlog_item\n \n return convert_github_issue_to_backlog_item(issue_payload, provider=\"github\")\n \n @beartype\n@@ -2743,51 +2743,51 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n return _github_linked_issue_edge(issue_id, linked)\n \n def _issue_relationship_edges(self, issue: dict[str, Any], issue_id: str) -> list[tuple[str, str, str]]:\n \"\"\"Collect relationship edges from provider fields and body text.\"\"\"\n edges: list[tuple[str, str, str]] = []\n provider_fields = issue.get(\"provider_fields\")\n if isinstance(provider_fields, dict):\n linked_issues = _as_str_dict(provider_fields).get(\"linked_issues\", [])\n if isinstance(linked_issues, list):\n for linked in linked_issues:\n if isinstance(linked, dict):\n edge = self._linked_issue_edge(issue_id, linked)\n if edge:\n edges.append(edge)\n body = str(issue.get(\"body_markdown\") or issue.get(\"description\") or \"\")\n for linked_id, relation_type, direction in self._body_relationship_matches(body):\n if direction == \"reverse\":\n edges.append((linked_id, issue_id, relation_type))\n else:\n edges.append((issue_id, linked_id, relation_type))\n return edges\n \n @beartype\n def _github_graphql(self, query: str, variables: dict[str, Any]) -> dict[str, Any]:\n \"\"\"Execute GitHub GraphQL request and return `data` payload.\"\"\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github+json\",\n }\n response = self._request_with_retry(\n lambda: requests.post(\n f\"{self.base_url}/graphql\",\n json={\"query\": query, \"variables\": variables},\n headers=headers,\n timeout=30,\n )\n )\n payload_raw = response.json()\n if not isinstance(payload_raw, dict):\n raise ValueError(\"GitHub GraphQL response must be an object\")\n payload = _as_str_dict(payload_raw)\n errors = payload.get(\"errors\")\n if isinstance(errors, list) and errors:\n raise ValueError(f\"GitHub GraphQL errors: {errors}\")\n data = payload.get(\"data\")\n return data if isinstance(data, dict) else {}\n \n @staticmethod\n @beartype\n def _resolve_github_type_mapping_id(mapping: dict[str, Any], issue_type: str) -> str:\n \"\"\"\n@@ -2958,51 +2958,51 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n console.print(f\"[yellow]⚠[/yellow] Could not set GitHub Projects Type field automatically: {error}\")\n \n @beartype\n @require(\n lambda project_id: isinstance(project_id, str) and len(project_id.strip()) > 0, \"project_id must be non-empty\"\n )\n @require(lambda payload: isinstance(payload, dict), \"payload must be dict\")\n @ensure(lambda result: isinstance(result, dict), \"Must return dict\")\n def create_issue(self, project_id: str, payload: dict[str, Any]) -> dict[str, Any]:\n \"\"\"Create a GitHub issue from provider-agnostic backlog payload.\"\"\"\n owner, repo = project_id.split(\"/\", 1) if \"/\" in project_id else (self.repo_owner, self.repo_name)\n if not owner or not repo:\n raise ValueError(\n \"GitHub project_id must be '/' or adapter must be configured with repo_owner/repo_name\"\n )\n if not self.api_token:\n raise ValueError(\"GitHub API token required to create issues\")\n \n title = self._required_issue_title(payload)\n issue_type = str(payload.get(\"type\") or \"task\").strip().lower()\n body = self._create_issue_body(payload)\n labels = self._labels_from_payload(\n issue_type, str(payload.get(\"priority\") or \"\").strip(), payload.get(\"story_points\")\n )\n url = f\"{self.base_url}/repos/{owner}/{repo}/issues\"\n- headers = {\n+ headers: dict[str, str | bytes] = {\n \"Authorization\": f\"token {self.api_token}\",\n \"Accept\": \"application/vnd.github.v3+json\",\n }\n response = self._request_with_retry(\n lambda: requests.post(\n url,\n json=cast(Any, {\"title\": title, \"body\": body, \"labels\": labels}),\n headers=headers,\n timeout=30,\n ),\n retry_on_ambiguous_transport=False,\n )\n created = response.json()\n self._apply_create_issue_post_hooks(owner, repo, created, payload, issue_type)\n \n canonical_issue_number = str(created.get(\"number\") or created.get(\"id\") or \"\")\n return {\n \"id\": canonical_issue_number,\n \"key\": canonical_issue_number,\n \"url\": str(created.get(\"html_url\") or created.get(\"url\") or \"\"),\n }\n \n def _apply_create_issue_post_hooks(\n self,\n owner: str,\n@@ -3227,51 +3227,51 @@ class GitHubAdapter(BridgeAdapter, BacklogAdapterMixin, BacklogAdapter):\n @require(\n lambda item, update_fields: update_fields is None or isinstance(update_fields, list),\n \"Update fields must be None or list\",\n )\n @ensure(lambda result: isinstance(result, BacklogItem), \"Must return BacklogItem\")\n @ensure(\n lambda result, item: ensure_backlog_update_preserves_identity(result, item),\n \"Updated item must preserve id and provider\",\n )\n def update_backlog_item(self, item: BacklogItem, update_fields: list[str] | None = None) -> BacklogItem:\n \"\"\"\n Update a GitHub issue.\n \n Updates the issue title and/or body ba