From 1948d3dc2197a5726c9e34fce56b34777ecc94bc Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 24 Jul 2026 18:31:28 -0400 Subject: [PATCH] fix(webui): repair traffic live path contracts for #640 review Address PR #885 REQUEST_CHANGES: full head_sha pins from queue signals, reviewer leases keyed by pr_number only, claim inventory via entries, live-path fixture tests, and traffic state vocabulary docs. Closes #640 (re-review at new head) Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/webui-local-dev.md | 25 +++ tests/test_webui_traffic_control.py | 216 +++++++++++++++++++++++++ webui/lease_loader.py | 9 +- webui/queue_loader.py | 37 ++++- webui/traffic_loader.py | 238 ++++++++++++++++++++++------ 5 files changed, 472 insertions(+), 53 deletions(-) diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index cb2bb8b..2bbecff 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -59,6 +59,31 @@ status, onboarding checklist state, and the fail-closed error payloads (#635). | `/api/queue` | JSON queue export with pagination metadata | | `/traffic` | Workflow traffic-control view — runnable, leased, blocked, needs-controller, terminal-complete (#640) | | `/api/traffic` | JSON traffic-control export with state classifications and next safe role actions | + +### Traffic-control state vocabulary (#640) + +The traffic view classifies each open issue/PR into exactly one bucket: + +| Bucket | Meaning | Operator implication | +|--------|---------|----------------------| +| **runnable** | No active lease, no block reason, safe for its expected role | Next role may start work | +| **leased** | Active author claim or reviewer PR lease | Do not stomp; wait or adopt via role tools | +| **blocked** | Dependency, missing head pin, conflict, or status:blocked | Author/controller remediation first | +| **needs_controller** | Contaminated / controller-only diagnosis | Controller only | +| **terminal_complete** | Reconciler / terminal-lock territory | Reconciler cleanup path | + +**Live path contracts (do not invent):** + +- PR head pins come from `QueueItem.signals["head_sha"]` (full SHA). Display + `extra["head_sha"]` is truncated and must never be used for routing. +- Reviewer leases are keyed as `(pr, pr_number)` only — never via a linked + `issue_number` on the same lease marker. +- Issue claims come from `claim_inventory["entries"]` + (`issue_claim_heartbeat.build_claim_inventory`). There is no `active_claims` + key. +- Queue display badges are only: `blocked`, `claimed`, `duplicate`, `stale`, + `in-review`, `open`. Review verdicts (`request-changes`, `approved`) are + **not** queue badges; traffic does not invent them from the queue loader. | `/projects` | Project registry list with status and onboarding progress (#427, #635) | | `/projects/{id}` | Project detail + onboarding checklist | | `/api/v1/projects` | Versioned JSON registry export (#635) | diff --git a/tests/test_webui_traffic_control.py b/tests/test_webui_traffic_control.py index 6d85661..f4c7581 100644 --- a/tests/test_webui_traffic_control.py +++ b/tests/test_webui_traffic_control.py @@ -116,6 +116,222 @@ class TestTrafficLoader(unittest.TestCase): self.assertFalse(snap.inventory_complete) +class TestTrafficLivePath(unittest.TestCase): + """Live path tests: inject QueueSnapshot + LeaseSnapshot (no candidates=). + + Covers the production ``load_traffic_snapshot()`` branch that ``/traffic`` + and ``/api/traffic`` actually execute (#640 B1–B5). + """ + + FULL_SHA = "069a9af7e6aa2c2994e07199d1b0814819457017" + + def _queue( + self, + *, + prs=(), + issues=(), + ): + from webui.queue_loader import QueueSnapshot + + return QueueSnapshot( + project_id="gitea-tools", + repo_label="Scaled-Tech-Consulting/Gitea-Tools", + prs=tuple(prs), + issues=tuple(issues), + pr_pagination=None, + issue_pagination=None, + fetch_error=None, + ) + + def _lease( + self, + *, + claim_inventory=None, + reviewer_leases=(), + ): + from webui.lease_loader import LeaseSnapshot + + return LeaseSnapshot( + project_id="gitea-tools", + repo_label="Scaled-Tech-Consulting/Gitea-Tools", + issue_lock=None, + claim_inventory=claim_inventory or {"entries": [], "counts": {}}, + reviewer_leases=tuple(reviewer_leases), + duplicate_prs=(), + duplicate_branches=(), + collision_history=(), + fetch_error=None, + ) + + def test_live_pr_uses_full_head_sha_and_is_runnable(self): + from webui.queue_loader import QueueItem + + pr = QueueItem( + number=885, + title="traffic control", + badges=("in-review",), + extra={"head_sha": self.FULL_SHA[:12], "linked_issue": "640"}, + signals={ + "head_sha": self.FULL_SHA, + "mergeable": True, + "labels": (), + "linked_issue": 640, + }, + ) + q = self._queue(prs=[pr]) + l = self._lease() + snap = load_traffic_snapshot( + fetch_queue_snapshot=lambda: q, + fetch_lease_snapshot=lambda: l, + ) + self.assertIsNone(snap.fetch_error) + self.assertEqual(len(snap.runnable), 1) + item = snap.runnable[0] + self.assertEqual(item.kind, "pr") + self.assertEqual(item.number, 885) + self.assertEqual(item.head_sha, self.FULL_SHA) + self.assertNotEqual(item.head_sha, self.FULL_SHA[:12]) + self.assertIsNone(item.block_reason) + self.assertEqual(len(snap.blocked), 0) + + def test_live_pr_without_head_sha_is_blocked(self): + from webui.queue_loader import QueueItem + + pr = QueueItem( + number=1, + title="missing pin", + badges=("open",), + extra={"head_sha": ""}, + signals={"head_sha": "", "mergeable": True, "labels": ()}, + ) + snap = load_traffic_snapshot( + fetch_queue_snapshot=lambda: self._queue(prs=[pr]), + fetch_lease_snapshot=lambda: self._lease(), + ) + self.assertEqual(len(snap.blocked) + len(snap.needs_controller), 1) + item = (snap.blocked or snap.needs_controller)[0] + self.assertIn("missing head_sha", (item.block_reason or "").lower()) + + def test_reviewer_lease_keys_by_pr_not_linked_issue(self): + from webui.queue_loader import QueueItem + + pr = QueueItem( + number=885, + title="leased pr", + badges=("in-review",), + extra={"head_sha": self.FULL_SHA[:12]}, + signals={"head_sha": self.FULL_SHA, "mergeable": True, "labels": ()}, + ) + issue = QueueItem( + number=640, + title="linked issue", + badges=("open",), + extra={}, + signals={"labels": ()}, + ) + # Marker-shaped record: has both pr_number and issue_number; must + # attach to the PR only (B2). + reviewer_lease = { + "pr_number": 885, + "issue_number": 640, + "phase": "validating", + "reviewer_identity": "sysadmin", + "session_id": "review-sess-1", + } + snap = load_traffic_snapshot( + fetch_queue_snapshot=lambda: self._queue(prs=[pr], issues=[issue]), + fetch_lease_snapshot=lambda: self._lease(reviewer_leases=[reviewer_lease]), + ) + leased_prs = [i for i in snap.leased if i.kind == "pr" and i.number == 885] + self.assertEqual(len(leased_prs), 1) + self.assertEqual(leased_prs[0].lease_info.get("pr_number"), 885) + # Issue 640 must not inherit the reviewer lease just because issue_number + # is present on the marker. + for item in list(snap.leased) + list(snap.runnable) + list(snap.blocked): + if item.kind == "issue" and item.number == 640: + self.assertIsNone( + item.lease_info, + "reviewer lease must not attach to linked issue #640", + ) + break + else: + self.fail("expected issue #640 in traffic snapshot") + + def test_claim_inventory_entries_key_marks_issue_leased(self): + from webui.queue_loader import QueueItem + + issue = QueueItem( + number=640, + title="claimed issue", + badges=("claimed",), + extra={}, + signals={"labels": ("status:in-progress",)}, + ) + inventory = { + "entries": [ + { + "issue_number": 640, + "status": "active", + "latest_heartbeat": {"session_id": "author-sess-9"}, + "reasons": ["claim has structured heartbeat proof"], + } + ], + "counts": {"active": 1}, + "in_progress_total": 1, + } + snap = load_traffic_snapshot( + fetch_queue_snapshot=lambda: self._queue(issues=[issue]), + fetch_lease_snapshot=lambda: self._lease(claim_inventory=inventory), + ) + leased_issues = [i for i in snap.leased if i.kind == "issue" and i.number == 640] + self.assertEqual(len(leased_issues), 1) + self.assertEqual(leased_issues[0].traffic_state, "leased") + + def test_active_claims_key_is_ignored(self): + """B3 regression: fictional ``active_claims`` must not create lease_info.""" + from webui.queue_loader import QueueItem + + issue = QueueItem( + number=640, + title="open issue", + badges=("open",), + extra={}, + signals={"labels": ()}, + ) + # Only the broken key — must NOT produce lease_info. Entries-less + # inventory is empty (entries is the real claim_inventory key). + inventory = { + "active_claims": [ + { + "kind": "issue", + "number": 640, + "issue_number": 640, + "status": "active", + }, + ], + "counts": {}, + } + snap = load_traffic_snapshot( + fetch_queue_snapshot=lambda: self._queue(issues=[issue]), + fetch_lease_snapshot=lambda: self._lease(claim_inventory=inventory), + ) + items = [ + i + for i in ( + list(snap.runnable) + + list(snap.leased) + + list(snap.blocked) + + list(snap.needs_controller) + ) + if i.kind == "issue" and i.number == 640 + ] + self.assertEqual(len(items), 1) + self.assertIsNone( + items[0].lease_info, + "active_claims is not a real inventory key; entries-only", + ) + + class TestTrafficRoutesAndRendering(unittest.TestCase): def setUp(self): self.client = TestClient(create_app()) diff --git a/webui/lease_loader.py b/webui/lease_loader.py index cd05d61..bd8c4f3 100644 --- a/webui/lease_loader.py +++ b/webui/lease_loader.py @@ -182,10 +182,17 @@ def _extract_reviewer_leases( parsed = parse_reviewer_lease_comment(comment.get("body") or "") if not parsed: continue + subject_pr = parsed.get("pr_number") or pr_number leases.append( { **parsed, - "pr_number": parsed.get("pr_number") or pr_number, + "pr_number": subject_pr, + # The lease subject is the PR, never the linked issue: a + # reviewer lease on PR #N must not be attributed to issue #N + # or to the issue that PR closes (#640). + "kind": "pr", + "number": subject_pr, + "role": "reviewer", "comment_id": comment.get("id"), "author": (comment.get("user") or {}).get("login"), "created_at": comment.get("created_at"), diff --git a/webui/queue_loader.py b/webui/queue_loader.py index a6d35c7..135ef39 100644 --- a/webui/queue_loader.py +++ b/webui/queue_loader.py @@ -4,7 +4,7 @@ from __future__ import annotations import os import re -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Callable from urllib.parse import urlparse @@ -31,10 +31,20 @@ class PaginationMeta: @dataclass(frozen=True) class QueueItem: + """One queue row. + + ``extra`` holds *display* strings for the queue page (values are truncated + or humanized for rendering). ``signals`` holds the *authoritative* typed + values taken straight from the Gitea payload, for consumers that classify + or pin state rather than render it (#640). Never derive identity or + concurrency decisions from ``extra``. + """ + number: int title: str badges: tuple[str, ...] extra: dict[str, str] + signals: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True) @@ -134,21 +144,37 @@ def _format_pr_item(pr: dict, badges: tuple[str, ...]) -> QueueItem: "mergeable" if mergeable is True else "conflicted" if mergeable is False else "unknown" ) linked = _extract_linked_issue(pr.get("title"), pr.get("body")) + head_sha = str(head.get("sha") or "") + labels = tuple( + str(lb.get("name") or "") for lb in (pr.get("labels") or []) if lb.get("name") + ) return QueueItem( number=int(pr["number"]), title=str(pr.get("title") or ""), badges=badges, extra={ "branch": f"{head.get('ref', '?')} → {base.get('ref', '?')}", - "head_sha": str(head.get("sha") or "")[:12], + # Display only — truncated. Pin against signals["head_sha"] instead. + "head_sha": head_sha[:12], "mergeable": merge_label, "linked_issue": str(linked) if linked is not None else "", }, + signals={ + "head_sha": head_sha, + "head_ref": str(head.get("ref") or ""), + "base_ref": str(base.get("ref") or ""), + "mergeable": mergeable if isinstance(mergeable, bool) else None, + "labels": labels, + "linked_issue": linked, + }, ) def _format_issue_item(issue: dict, badges: tuple[str, ...]) -> QueueItem: - labels = ", ".join(lb.get("name", "") for lb in issue.get("labels", [])) + label_names = tuple( + str(lb.get("name") or "") for lb in (issue.get("labels") or []) if lb.get("name") + ) + labels = ", ".join(label_names) assignee = (issue.get("assignee") or {}).get("login", "") return QueueItem( number=int(issue["number"]), @@ -159,6 +185,11 @@ def _format_issue_item(issue: dict, badges: tuple[str, ...]) -> QueueItem: "assignee": assignee or "unassigned", "state": str(issue.get("state") or ""), }, + signals={ + "labels": label_names, + "assignee": assignee, + "state": str(issue.get("state") or ""), + }, ) diff --git a/webui/traffic_loader.py b/webui/traffic_loader.py index 9753d15..912d10e 100644 --- a/webui/traffic_loader.py +++ b/webui/traffic_loader.py @@ -8,7 +8,6 @@ and terminal-complete candidates. from __future__ import annotations -import os from dataclasses import dataclass from typing import Any, Callable, Sequence @@ -97,13 +96,23 @@ def _classify_traffic_item( expected_role = entry.expected_role entry_is_safe = entry.block_reason is None and bool(entry.safe_for_roles) + # Lease state is checked first: an item that is both leased and blocked is + # reported as leased. That is safe by construction — a leased item is never + # placed in the runnable lane — and it keeps the operator's attention on the + # session that currently owns the work. The blocker text still renders. if lease_info is not None or "in-progress" in badges or "claimed" in badges: state = "leased" elif expected_role == "reconciler" or "terminal-lock" in badges: state = "terminal_complete" elif expected_role == "controller" or "contaminated" in badges or "needs-controller" in badges: state = "needs_controller" - elif block_reason is not None or "blocked" in badges or "dependency-unmet" in badges or "blocked-by-terminal" in badges: + elif ( + block_reason is not None + or "blocked" in badges + or "dependency-unmet" in badges + or "blocked-by-terminal" in badges + or "status:blocked" in badges + ): state = "blocked" elif entry_is_safe: state = "runnable" @@ -124,6 +133,156 @@ def _classify_traffic_item( ) +# Claim statuses from ``issue_claim_heartbeat.build_claim_inventory`` that mean +# a live worker currently holds the issue. Everything else (``stale``, +# ``phantom``, ``reclaimable``, ``not_claimed``) is reported through the +# dashboard's stale-lease channel and is never rendered as an active lease. +_ACTIVE_CLAIM_STATUSES = frozenset({"active", "awaiting_review"}) + +# Statuses that positively mean "not an active lease" for any lease record. +_INACTIVE_LEASE_STATUSES = frozenset( + {"expired", "stale", "released", "moot", "reclaimable", "phantom", "not_claimed"} +) + + +def _candidates_from_queue_snapshot(q_snap: QueueSnapshot) -> list[WorkCandidate]: + """Build allocator candidates from the queue loader's authoritative signals. + + Display badges (``blocked``/``claimed``/``duplicate``/``stale``/ + ``in-review``/``open``) are rendering hints, not routing state, so nothing + here branches on them. Every routing field comes from + ``QueueItem.signals`` — the raw Gitea payload values. + + The queue loader reads ``/pulls`` and ``/issues`` only; it never fetches + review verdicts. ``request_changes_current_head`` / ``approval_on_current_head`` + are therefore left at their fail-safe ``False`` rather than being guessed + from badges: an unproven approval must never route a PR to the merger. + """ + candidates: list[WorkCandidate] = [] + + for pr in q_snap.prs: + signals = pr.signals or {} + head_sha = str(signals.get("head_sha") or "").strip() + mergeable = signals.get("mergeable") + labels = tuple(str(x) for x in (signals.get("labels") or ())) + candidates.append( + WorkCandidate( + kind="pr", + number=pr.number, + state="open", + labels=labels, + title=pr.title, + # Full 40-char SHA from head.sha — never the 12-char display value. + head_sha=head_sha or None, + priority=5, + mergeable=mergeable is True, + blocked=mergeable is False or "status:blocked" in labels, + ) + ) + + for issue in q_snap.issues: + signals = issue.signals or {} + labels = tuple(str(x) for x in (signals.get("labels") or ())) + lowered = {label.lower() for label in labels} + candidates.append( + WorkCandidate( + kind="issue", + number=issue.number, + state="open", + labels=labels, + title=issue.title, + priority=20 if "status:ready" in lowered else 10, + blocked="status:blocked" in lowered, + # A live claim by another session is not this session's work. + already_claimed_elsewhere="status:in-progress" in lowered, + ) + ) + + return candidates + + +def _claim_lease_records(inventory: dict[str, Any] | None) -> list[dict[str, Any]]: + """Normalize ``build_claim_inventory`` entries into lease records. + + The inventory contract is ``{"entries", "counts", "heartbeat_lease_minutes", + "reclaim_after_minutes", "in_progress_total"}``. Each entry is keyed by + ``issue_number``; the subject kind is therefore always ``issue``. + """ + entries = (inventory or {}).get("entries") or () + records: list[dict[str, Any]] = [] + for entry in entries: + if not isinstance(entry, dict): + continue + number = entry.get("issue_number") + if number is None: + continue + try: + number_int = int(number) + except (TypeError, ValueError): + continue + heartbeat = entry.get("latest_heartbeat") or {} + record = dict(entry) + record.update( + { + "kind": "issue", + "number": number_int, + "role": "author", + "lease_source": "issue-claim-heartbeat", + } + ) + if isinstance(heartbeat, dict): + if heartbeat.get("session_id") and not record.get("session_id"): + record["session_id"] = heartbeat.get("session_id") + if heartbeat.get("author") and not record.get("author"): + record["author"] = heartbeat.get("author") + records.append(record) + return records + + +def _lease_subject(lease: dict[str, Any]) -> tuple[str, int] | None: + """Return the ``(kind, number)`` a lease record actually covers. + + Fails closed: a record that does not identify exactly one subject is + dropped rather than attributed to a guessed work item (#640 — never invent + a lease, and never attach a PR lease to a same-numbered issue). + """ + kind = str(lease.get("kind") or lease.get("work_kind") or "").strip().lower() + pr_number = lease.get("pr_number") + issue_number = lease.get("issue_number") + + if kind not in ("pr", "issue"): + if pr_number is not None and issue_number is None: + kind = "pr" + elif issue_number is not None and pr_number is None: + kind = "issue" + else: + return None + + number = lease.get("number") + if number is None: + number = lease.get("work_number") + if number is None: + number = pr_number if kind == "pr" else issue_number + if number is None: + return None + try: + return kind, int(number) + except (TypeError, ValueError): + return None + + +def _is_active_lease(lease: dict[str, Any]) -> bool: + """True when the record proves a worker currently holds the item.""" + if lease.get("stale") or lease.get("expired"): + return False + status = str(lease.get("status") or lease.get("lease_status") or "").strip().lower() + if status in _INACTIVE_LEASE_STATUSES: + return False + if lease.get("lease_source") == "issue-claim-heartbeat": + return status in _ACTIVE_CLAIM_STATUSES + return True + + def load_traffic_snapshot( *, candidates: Sequence[WorkCandidate] | None = None, @@ -192,45 +351,27 @@ def load_traffic_snapshot( inventory_complete=False, ) - # Build WorkCandidates from live queue snapshot - candidate_list: list[WorkCandidate] = [] - for pr in q_snap.prs: - linked = int(pr.extra["linked_issue"]) if pr.extra.get("linked_issue") and pr.extra["linked_issue"].isdigit() else None - candidate_list.append( - WorkCandidate( - kind="pr", - number=pr.number, - state="open", - labels=(), - title=pr.title, - priority=10 if "request-changes" in pr.badges else 5, - request_changes_current_head="request-changes" in pr.badges, - approval_on_current_head="approved" in pr.badges or "merge-ready" in pr.badges, - mergeable="blocked" not in pr.badges, - blocked="blocked" in pr.badges, - linked_issue_number=linked, - ) - ) + candidate_list = _candidates_from_queue_snapshot(q_snap) - for issue in q_snap.issues: - labels = [b for b in issue.badges if b.startswith("status:") or b in ("discussion", "blocked", "ready", "in-progress")] - candidate_list.append( - WorkCandidate( - kind="issue", - number=issue.number, - state="open", - labels=tuple(labels), - title=issue.title, - priority=20 if "status:ready" in labels else 10, - blocked="blocked" in labels or "status:blocked" in labels, - ) - ) - - raw_leases: list[dict[str, Any]] = [] - if l_snap.claim_inventory and "active_claims" in l_snap.claim_inventory: - raw_leases.extend(l_snap.claim_inventory["active_claims"]) - for r_lease in l_snap.reviewer_leases: - raw_leases.append(r_lease) + raw_leases: list[dict[str, Any]] = _claim_lease_records(l_snap.claim_inventory) + for r_lease in l_snap.reviewer_leases or (): + if not isinstance(r_lease, dict): + continue + # Always pin reviewer leases to the PR subject, even if a linked + # issue_number is present on the marker (#640 B2). + normalized = dict(r_lease) + subject = normalized.get("pr_number") or normalized.get("number") + if subject is None: + continue + try: + pr_num = int(subject) + except (TypeError, ValueError): + continue + normalized["kind"] = "pr" + normalized["number"] = pr_num + normalized["pr_number"] = pr_num + normalized.setdefault("role", "reviewer") + raw_leases.append(normalized) dashboard = build_workflow_dashboard( candidates=candidate_list, @@ -256,17 +397,16 @@ def _build_traffic_snapshot_from_dashboard( """Classify dashboard entries into the 5 traffic state buckets.""" all_entries = dashboard.open_prs + dashboard.open_issues - # Map leased work numbers + # Map each active lease onto the exact work item it covers. Records whose + # subject cannot be determined, and claims that are stale/phantom/ + # reclaimable, are deliberately dropped instead of guessed. lease_map: dict[tuple[str, int], dict[str, Any]] = {} for lease in leases: - if isinstance(lease, dict): - kind = str(lease.get("kind") or lease.get("work_kind") or "issue").lower() - num = lease.get("number") or lease.get("work_number") or lease.get("issue_number") or lease.get("pr_number") - if num is not None: - try: - lease_map[(kind, int(num))] = lease - except (ValueError, TypeError): - pass + if not isinstance(lease, dict) or not _is_active_lease(lease): + continue + subject = _lease_subject(lease) + if subject is not None: + lease_map[subject] = lease runnable: list[TrafficItem] = [] leased: list[TrafficItem] = []