Fix focused Activities profile crediting and ADO continuation

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())

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

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()},

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