From 82f0e172a743fbab8c5270cae296e03fbd179518 Mon Sep 17 00:00:00 2001 From: Mengxiong Han Date: Fri, 13 Nov 2026 13:06:11 +0800 Subject: [PATCH] Fix focused Activities profile crediting and ADO resume --- scripts/local_usage_ado.py | 47 +++++++++++--- scripts/local_usage_ado.test.py | 95 +++++++++++++++++++++++++++++ scripts/local_usage_profile.py | 14 +++-- scripts/local_usage_profile.test.py | 77 +++++++++++++++++++++++ scripts/refresh-local-usage.py | 8 ++- scripts/refresh-local-usage.test.py | 15 ++++- 6 files changed, 238 insertions(+), 18 deletions(-) diff --git a/scripts/local_usage_ado.py b/scripts/local_usage_ado.py index e61c6e6b7..e8b75b3aa 100644 --- a/scripts/local_usage_ado.py +++ b/scripts/local_usage_ado.py @@ -38,6 +38,17 @@ class RepositoryQuantum(Exception): pass +class ReadInterrupted(Exception): + def __init__(self, stage, status=None): + self.stage = stage + self.status = status + + +class HttpReadError(OSError): + def __init__(self, status): + self.status = status + + def stamp(value): return value.astimezone(dt.timezone.utc).isoformat().replace("+00:00", "Z") @@ -154,7 +165,7 @@ def __init__(self, config, run, end): self.source_calls = 0 self.repo_limit = None - def get(self, target): + def get(self, target, stage="authenticated"): remaining = min(self.end, self.source_end) - time.monotonic() if self.source_calls >= REQUESTS_PER_PROJECT or remaining <= 0: raise BudgetExceeded("ADO request budget exhausted") @@ -165,9 +176,19 @@ def get(self, target): self.calls += 1 self.source_calls += 1 try: - body = json.loads(self.run(args, timeout=min(20, remaining))) - except (OSError, ValueError, TypeError, subprocess.SubprocessError): - raise ValueError("ADO authenticated read unavailable") from None + raw = self.run(args, timeout=min(20, remaining)) + except HttpReadError as error: + if error.status in (408, 429) or 500 <= error.status <= 599: + raise ReadInterrupted(stage, error.status) from None + raise ValueError(f"ADO {stage} HTTP {error.status}") from None + except (OSError, subprocess.SubprocessError): + raise ReadInterrupted(stage) from None + except (ValueError, TypeError): + raise ValueError(f"ADO {stage} read unavailable") from None + try: + body = json.loads(raw) + except (ValueError, TypeError): + raise ValueError("Invalid ADO response") from None if not isinstance(body, dict) or body.get("errors") or body.get("continuationToken"): raise ValueError("Invalid or unpaged ADO response") return body @@ -258,7 +279,7 @@ def reset_pr_pages(work): def default_head(reader, org, project, repo): branch = repo["defaultBranch"] path = "/_apis/git/repositories/" + repo["id"] + "/refs" - page = rows(reader.get(url(org, project, path, filter=branch[len("refs/"):])), 100) + page = rows(reader.get(url(org, project, path, filter=branch[len("refs/"):]), "default-head"), 100) exact = [row for row in page if row.get("name") == branch] if len(exact) != 1 or not SHA.fullmatch(exact[0].get("objectId", "")): raise ValueError("ADO default branch head unavailable") @@ -272,7 +293,7 @@ def commit_page(reader, org, project, repo_id, head, offset, start=None, end=Non "searchCriteria.itemVersion.versionType": "commit", "$top": PAGE, "$skip": offset} if start is not None: criteria.update({"searchCriteria.fromDate": start, "searchCriteria.toDate": end}) - body = reader.get(url(org, project, path, **criteria)) + body = reader.get(url(org, project, path, **criteria), "focused-commits" if start else "pinned-commits") return rows(body) @@ -286,7 +307,7 @@ def credited_commit(reader, org, project, repo_id, item, emails, first): if (detail.get("commentTruncated") is not False or not isinstance(detail.get("comment"), str) or not isinstance(detail.get("parents"), list)): path = "/_apis/git/repositories/" + repo_id + "/commits/" + commit_id - detail = reader.get(url(org, project, path)) + detail = reader.get(url(org, project, path), "commit-detail") truncated = detail.get("commentTruncated", False) if (detail.get("commitId", "").lower() != commit_id or type(truncated) is not bool or truncated): @@ -609,7 +630,7 @@ def collect(config, run, repository_category, *, at=None, days=60, previous=None raise ValueError("Duplicate retained ADO source") emails = author_aliases(config) email_hash = digest("ado-author-aliases", sorted(emails)) if emails else None - for org, project, configured_names in configured: + for source_index, (org, project, configured_names) in enumerate(configured): account = principal(config, org) scope = org.casefold() + "/" + project.casefold() scope_hash = digest("ado-remotes", scope, configured_names) @@ -665,7 +686,10 @@ def collect(config, run, repository_category, *, at=None, days=60, previous=None work = new_work(at, focus_start if can_focus else None, prior_work["prs"] if can_focus else None) reader.source_calls = 0 - reader.source_end = min(reader.end, time.monotonic() + SECONDS_PER_PROJECT) + source_started = time.monotonic() + remaining_sources = len(configured) - source_index + reader.source_end = min(reader.end, source_started + SECONDS_PER_PROJECT, + source_started + max(0, reader.end - source_started) / remaining_sources) reasons = ["legacy-repository-scope-unresolved"] if unresolved_scope else [] verified = False try: @@ -723,8 +747,11 @@ def collect(config, run, repository_category, *, at=None, days=60, previous=None work["reasons"].append("disabled-repository") advance_prs(reader, org, project, account, work) advance_repos(reader, org, project, work, cache, emails) - except (BudgetExceeded, OSError, KeyError, TypeError, ValueError, IndexError, AttributeError) as error: + except (BudgetExceeded, ReadInterrupted, OSError, KeyError, TypeError, ValueError, IndexError, AttributeError) as error: reasons.append("request-budget" if isinstance(error, BudgetExceeded) + else (f"{error.stage}-read-interrupted" + + (f"-http-{error.status}" if error.status is not None else "")) + if isinstance(error, ReadInterrupted) else str(error) if isinstance(error, ValueError) else "read-failure") cache = {**cache, **work["repoDone"]} reasons += work["reasons"] + list(work["repoErrors"].values()) diff --git a/scripts/local_usage_ado.test.py b/scripts/local_usage_ado.test.py index a8126f3bb..13ef31408 100644 --- a/scripts/local_usage_ado.test.py +++ b/scripts/local_usage_ado.test.py @@ -162,6 +162,101 @@ def open_request(request, timeout): for query in queries)) self.assertTrue(current["sources"][0]["coverage"]["paginationComplete"]) + def test_detail_transport_timeout_persists_pinned_prefix_and_resumes_through_protected_adapter(self): + api = Api() + requests = [] + interrupted = [False] + + def session(): + protected = usage.AzureMemorySession(Path("/fixture/unused-auth"), [EMAIL], + projects={("example", "project")}) + protected.token = lambda: "fixture-token" + + def open_request(request, timeout): + self.assertEqual(request.get_header("Authorization"), "Bearer fixture-token") + requests.append(request.full_url) + if request.full_url.split("?")[0].endswith("/commits/" + OLDER) and not interrupted[0]: + interrupted[0] = True + raise TimeoutError("synthetic transport timeout: secret") + args = ["/fixture/az", "rest", "--method", "get", "--url", request.full_url, + "--resource", ado.RESOURCE, "-o", "json"] + return io.BytesIO(api(args, timeout).encode()) + + protected.http = types.SimpleNamespace(open=open_request) + return protected + + failed = collect(session().run) + work = failed["state"]["projects"]["example/project"]["work"] + self.assertEqual(failed["sources"][0]["status"], "partial") + self.assertIn("commit-detail-read-interrupted", failed["sources"][0]["coverage"]["reasons"]) + self.assertNotIn("secret", json.dumps(failed)) + self.assertFalse(work["scanComplete"]) + self.assertEqual(len(work["progress"][REPO]["items"]), 1) + self.assertNotIn(REPO, work["repoErrors"]) + resumed = collect(session().run, previous=failed) + self.assertEqual(resumed["sources"][0]["status"], "ready") + self.assertEqual(resumed["sources"][0]["days"][-2]["commits"], 1) + self.assertEqual(sum(url.split("?")[0].endswith("/commits/" + HEAD) for url in requests), 1) + + def test_interrupted_final_head_check_retains_pinned_proof_but_requires_remote_recheck(self): + api = Api() + previous = collect(api) + newer = "d" * 40 + api.heads[REPO] = newer + api.histories[newer] = [commit(newer, "2026-11-10T00:01:00Z"), *api.histories[HEAD]] + api.details[newer] = api.histories[newer][0] + api.enforce_query_window = True + refs = [0] + + def interrupted(args, timeout): + if "/refs?" in args[5]: + refs[0] += 1 + if refs[0] == 2: + raise TimeoutError("synthetic final head check interrupted") + return api(args, timeout) + + at = AT + dt.timedelta(minutes=5) + failed = collect(interrupted, previous=previous, at=at, focus_start="2026-11-10") + state = failed["state"]["projects"]["example/project"] + self.assertIn("default-head-read-interrupted", failed["sources"][0]["coverage"]["reasons"]) + self.assertEqual(state["cache"][REPO]["head"], newer) + self.assertEqual(state["work"]["progress"][REPO]["head"], newer) + self.assertFalse(state["work"]["scanComplete"]) + resumed = collect(api, previous=failed, at=at, focus_start="2026-11-10") + self.assertEqual(resumed["sources"][0]["status"], "ready") + self.assertEqual(resumed["state"]["projects"]["example/project"]["cache"][REPO]["head"], newer) + + def test_one_slow_project_leaves_the_second_project_a_bounded_read_slice(self): + first = Api() + second = Api() + second.repos[0]["name"] = "other" + second.repos[0]["defaultBranch"] = None + second.prs = [] + config = copy.deepcopy(CONFIG) + config["adoRemotes"].append("https://dev.azure.com/other/project/_git/other") + config["adoPrincipals"]["other"] = {"stableId": OWNER, "alias": EMAIL} + elapsed = [0.0] + calls = [] + + def run(args, timeout): + target = args[5] + other = "/other/" in target + calls.append((other, target)) + elapsed[0] += 0.015 if not other else 0.005 + if other: + mapped = list(args) + mapped[5] = target.replace("/other/project/", "/example/project/").replace( + "/other/_apis/", "/example/_apis/") + return second(mapped, timeout) + return first(args, timeout) + + with patch.object(ado.time, "monotonic", lambda: elapsed[0]), \ + patch.object(ado, "SECONDS_TOTAL", 0.15), patch.object(ado, "SECONDS_PER_PROJECT", 0.15): + result = ado.collect(config, run, lambda remote, cfg: "work", at=AT) + self.assertIn("request-budget", result["sources"][0]["coverage"]["reasons"]) + self.assertEqual(result["sources"][1]["status"], "ready") + self.assertTrue(any(other for other, _ in calls)) + def test_warm_focus_bounds_pr_and_commit_queries_and_keeps_coauthor_author_day(self): api = Api() api.enforce_query_window = True diff --git a/scripts/local_usage_profile.py b/scripts/local_usage_profile.py index 9f763e6bb..286f4b8ef 100644 --- a/scripts/local_usage_profile.py +++ b/scripts/local_usage_profile.py @@ -204,7 +204,7 @@ def elapsed(self): def profile_records(config, profile, budget, first, at, repository_category, query_first=None): start = window_start(query_first) if query_first is not None else window_start(first) - dt.timedelta(days=1) - after, seen_cursors, prs, repositories, commit_repositories = None, set(), {}, {}, [] + after, seen_cursors, seen_prs, prs, repositories, commit_repositories = None, set(), set(), {}, {}, [] signature = None for _ in range(MAX_PROFILE_PAGES): data = budget.query(profile, PROFILE_QUERY, {"from": stamp(start), "to": stamp(at), "after": after}) @@ -267,7 +267,7 @@ def profile_records(config, profile, budget, first, at, repository_category, que pr = node["pullRequest"] occurred = instant(node["occurredAt"]) created = instant(pr["createdAt"]) - if (occurred < start or occurred > at or created > at + if (not start.date() <= occurred.date() <= at.date() or not isinstance(pr.get("author"), dict) or pr["author"].get("id") != profile["stableId"] or pr["author"].get("login", "").casefold() != profile["login"].casefold() @@ -279,13 +279,15 @@ def profile_records(config, profile, budget, first, at, repository_category, que raise ValueError("conflicting_profile_repository") repositories[repo["id"]] = repo key = digest("github-pr", profile["host"], pr["id"]) - if key in prs: + if key in seen_prs: raise ValueError("duplicate_profile_pr") - prs[key] = {"day": china_day(created), "createdAt": stamp(created), - "repository": repo, "key": key} + seen_prs.add(key) + if start <= created <= at: + prs[key] = {"day": china_day(created), "createdAt": stamp(created), + "repository": repo, "key": key} page = connection["pageInfo"] if page["hasNextPage"] is False: - if len(prs) != total: + if len(seen_prs) != total: raise ValueError("incomplete_pr_pages") return {"prs": list(prs.values()), "commits": [], "commitRepositories": commit_repositories, "prKeys": set(prs), "prRepositoryIds": {pr["repository"]["id"] for pr in prs.values()}, diff --git a/scripts/local_usage_profile.test.py b/scripts/local_usage_profile.test.py index 89a8c51c4..058584494 100644 --- a/scripts/local_usage_profile.test.py +++ b/scripts/local_usage_profile.test.py @@ -1,8 +1,13 @@ import copy import datetime as dt +import importlib.util +import io import json +from pathlib import Path import re import subprocess +import sys +import types import unittest import urllib.parse from unittest.mock import patch @@ -11,6 +16,10 @@ import local_usage_profile_private as private import local_usage_ado as ado +SPEC = importlib.util.spec_from_file_location("usage_memory", Path(__file__).with_name("refresh-local-usage.py")) +usage = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(usage) + AT = dt.datetime.fromisoformat("2026-11-09T12:00:00+00:00") ACCOUNT = {"host": "github.com", "login": "fixture-owner", "stableId": "fixture-account-id", @@ -200,6 +209,57 @@ def collect(api=None, account=None, at=AT, previous=None, focus_start=None): class ProfileBeijingTests(unittest.TestCase): + def test_partial_day_profile_validates_raw_pages_but_credits_only_created_at_inside_window(self): + api = Api() + api.profile_commit_at = "2026-11-13T00:00:00Z" + api.profile_pr_rows = [ + ("before-focus", "2026-11-12T15:33:32Z", PUBLIC), + ("inside-focus", "2026-11-13T01:00:00Z", PUBLIC), + ("after-cutoff", "2026-11-13T05:00:00Z", PUBLIC), + ] + at = dt.datetime.fromisoformat("2026-11-13T04:44:46+00:00") + config = {**CONFIG, "githubHelper": str(Path(usage.__file__).resolve())} + session = usage.GitHubFileSession({**config, "githubProfiles": [ACCOUNT]}, Path("/fixture/no-auth")) + session._credential = lambda name: {"token": "synthetic-token", "login": ACCOUNT["login"], + "stableId": ACCOUNT["stableId"]} + requests = [] + + def open_request(request, timeout): + self.assertEqual(request.get_header("Authorization"), "Bearer synthetic-token") + requests.append(request) + if request.full_url.endswith("/user"): + return io.BytesIO(json.dumps({"node_id": ACCOUNT["stableId"], + "login": ACCOUNT["login"], + "html_url": "https://github.com/" + ACCOUNT["login"]}).encode()) + self.assertTrue(request.full_url.endswith("/graphql")) + body = json.loads(request.data) + self.assertEqual(body["query"], profile.PROFILE_QUERY) + self.assertEqual(body["variables"]["from"], "2026-11-12T16:00:00Z") + self.assertEqual(body["variables"]["to"], "2026-11-13T04:44:46Z") + args = [sys.executable, config["githubHelper"], "gh", ACCOUNT["profile"], "api", "graphql"] + for key, value in body["variables"].items(): + if value is not None: + args.extend(("-f", f"{key}={value}")) + args.extend(("-f", "query=" + body["query"])) + return io.BytesIO(api(args, timeout).encode()) + + session.http = types.SimpleNamespace(open=open_request) + typed = profile.profile_records(config, ACCOUNT, profile.Budget(config, session.run), + dt.date(2026, 9, 15), at, + lambda remote, configured: "personal", dt.date(2026, 11, 13)) + self.assertEqual(typed["typedTotals"]["prs"], 3) + self.assertEqual([row["day"] for row in typed["prs"]], ["2026-11-13"]) + self.assertEqual(len(typed["prKeys"]), 1) + self.assertEqual(len(requests), 3) + + duplicate = Api() + duplicate.profile_commit_at = api.profile_commit_at + duplicate.profile_pr_rows = [api.profile_pr_rows[0], api.profile_pr_rows[0]] + with self.assertRaisesRegex(ValueError, "duplicate_profile_pr"): + profile.profile_records(CONFIG, ACCOUNT, profile.Budget(CONFIG, duplicate), + dt.date(2026, 9, 15), at, + lambda remote, configured: "personal", dt.date(2026, 11, 13)) + def assert_same_days_except_capture_time(self, first, second): for days in (first, second): captured = {day["capturedAt"] for day in days} @@ -360,6 +420,23 @@ def test_warm_focus_queries_only_the_requested_beijing_day_and_keeps_old_events( self.assertTrue(current["sources"][0]["coverage"]["paginationComplete"]) self.assertFalse(any("history(first:" in arg for call in api.calls for arg in call)) + def test_calendar_wide_profile_rows_keep_old_day_and_allow_empty_focus_correction(self): + api = Api() + api.profile_commit_at = "2026-11-10T08:00:00Z" + api.profile_pr_rows = [("old", "2026-11-09T15:00:00Z", PUBLIC), + ("today", "2026-11-10T09:00:00Z", PUBLIC)] + account = {**ACCOUNT, "privateRepositories": []} + initial = collect(api, account, at=dt.datetime.fromisoformat("2026-11-10T10:00:00+00:00")) + self.assertEqual(initial["sources"][0]["days"][-1]["knownPrs"], 1) + api.profile_pr_rows = [("old", "2026-11-09T15:00:00Z", PUBLIC), + ("after-cutoff", "2026-11-10T13:00:00Z", PUBLIC)] + current = collect(api, account, at=dt.datetime.fromisoformat("2026-11-10T12:00:00+00:00"), + previous=initial, focus_start="2026-11-10") + days = {row["day"]: row for row in current["sources"][0]["days"]} + self.assertEqual(days["2026-11-09"]["knownPrs"], 1) + self.assertEqual(days["2026-11-10"]["knownPrs"], 0) + self.assertTrue(current["sources"][0]["coverage"]["paginationComplete"]) + def test_private_pr_discovery_uses_the_same_focus_and_retains_older_known_pr(self): api = Api() api.enforce_query_window = True diff --git a/scripts/refresh-local-usage.py b/scripts/refresh-local-usage.py index fb12af9ee..aefdd8fc7 100644 --- a/scripts/refresh-local-usage.py +++ b/scripts/refresh-local-usage.py @@ -20,6 +20,7 @@ import tempfile import threading import time +import urllib.error import urllib.parse import urllib.request from zoneinfo import ZoneInfo @@ -27,6 +28,7 @@ import local_usage_merged import local_usage_github import local_usage_contributions +import local_usage_ado ADO_RESOURCE = "499b84ac-1321-427f-aa17-267ca6975798" LOCAL_COLLECT_TIMEOUT = 120 @@ -599,8 +601,12 @@ def request(self, url, timeout): headers = getattr(response, "headers", None) continuation = headers.get("x-ms-continuationtoken") if headers else None return response.read().decode("utf-8"), continuation + except urllib.error.HTTPError as error: + if type(error.code) is not int or not 100 <= error.code <= 599: + raise ValueError("The Azure DevOps HTTP status is unavailable") from None + raise local_usage_ado.HttpReadError(error.code) from None except http.client.HTTPException: - raise ValueError("The Azure DevOps response could not be read") from None + raise OSError("The Azure DevOps response could not be read") from None def ado_project_scopes(config): diff --git a/scripts/refresh-local-usage.test.py b/scripts/refresh-local-usage.test.py index 4458416c6..ddd6506c6 100644 --- a/scripts/refresh-local-usage.test.py +++ b/scripts/refresh-local-usage.test.py @@ -376,10 +376,23 @@ def open_response(request, timeout): return response self.session.http = types.SimpleNamespace(open=open_response) - with self.subTest(wire=wire), self.assertRaises(ValueError) as error: + with self.subTest(wire=wire), self.assertRaises(OSError) as error: self.session.run(command(), 7) self.assertEqual(str(error.exception), "The Azure DevOps response could not be read") + def test_http_status_is_bounded_for_repository_read_classification(self): + self.session.token = lambda: "fixture-access-token" + for status in (401, 429, 503): + def open_request(request, timeout): + raise urllib.error.HTTPError(request.full_url, status, "synthetic private response", {}, None) + + self.session.http = types.SimpleNamespace(open=open_request) + with self.subTest(status=status), self.assertRaises(usage.local_usage_ado.HttpReadError) as error: + self.session.run(command(), 7) + self.assertEqual(error.exception.status, status) + self.assertIsInstance(error.exception, OSError) + self.assertNotIn("private", str(error.exception)) + def test_refresh_reuses_collector_without_azure_cli_or_direct_gh(self): collector = self.root / "collector.py" output = self.root / "result.json"