Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  sysadminandClaude Opus 4.8
|
211890f361 |
@@ -1,167 +0,0 @@
|
||||
# ADR: High-availability and rolling-restart architecture for Gitea MCP control plane
|
||||
|
||||
- **Status:** Proposed (Design ADR under [#668](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/668))
|
||||
- **Date:** 2026-07-25
|
||||
- **Tracking Issue:** [#668](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/668)
|
||||
- **Policy Version:** `mcp-ha-rolling-restart/v1`
|
||||
- **Related:**
|
||||
- Parent: [#655](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/655) — Governed MCP restart coordination and zero-disruption recovery
|
||||
- Governance Policy: [#656](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/656) / `docs/architecture/mcp-restart-governance.md`
|
||||
- Control-Plane DB Substrate: [#613](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/613) / `docs/architecture/control-plane-db-substrate.md`
|
||||
- Runtime Policy: [#615](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/615) / `docs/architecture/mcp-stable-control-runtime-policy-adr.md`
|
||||
- Product Vision: [#652](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/652) (Phase 5 Maturity)
|
||||
- Delivery Roadmap: [#653](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/653)
|
||||
|
||||
---
|
||||
|
||||
## 1. Context & Problem Statement
|
||||
|
||||
The Gitea MCP server operates as the authoritative **control plane** for managing issues, Pull Requests, code mutations, formal reviews, and workflow reconciliations. Under single-process governance ([#656](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/656)), process restarts are strictly controlled using pre-flight checks, drain phases, and operator approvals.
|
||||
|
||||
However, a single-instance control plane inherently presents fundamental constraints:
|
||||
|
||||
1. **Downtime during updates:** Even a perfectly executed single-process drain requires a window where incoming client requests must be paused or rejected while the server binary or python environment reloads.
|
||||
2. **Single point of failure:** Infrastructure issues, process crashes, or unhandled host-level terminations immediately disconnect active LLM sessions and leave transient workflows incomplete.
|
||||
3. **Multi-agent concurrency bottlenecks:** High volumes of concurrent multi-LLM tasks put all lock management, lease allocation, and Gitea API interactions through a single process event loop.
|
||||
|
||||
To achieve true zero-disruption operation and seamless rolling deployments without stopping active work, the system requires a high-availability (HA), multi-instance MCP architecture.
|
||||
|
||||
---
|
||||
|
||||
## 2. Architectural Principles & Non-Goals
|
||||
|
||||
### 2.1 Core Architectural Principles
|
||||
* **Gitea as Canonical Work SoT:** Gitea remains the ultimate System of Record (SoT) for issue states, pull requests, labels, and audit comments. The MCP control plane does not duplicate domain entities.
|
||||
* **Control-Plane DB as Multi-Instance State Substrate:** The control-plane SQLite/durable database ([#613](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/613)) acts as the single source of truth for workflow leases, session tokens, assignment records, and lock fences across all MCP nodes.
|
||||
* **Stateless Worker Nodes:** MCP role server processes (`gitea-author`, `gitea-reviewer`, `gitea-merger`, `gitea-reconciler`, `gitea-controller`) maintain no unique in-memory state; any node can handle any request given a valid session resume token.
|
||||
* **Fail-Closed Split-Brain Defense:** In any network partition or quorum loss scenario, nodes must fail closed rather than risk double-mutations or conflicting Gitea states.
|
||||
|
||||
### 2.2 Non-Goals
|
||||
* **Replacing Gitea:** We do not replace Gitea issue/PR tracking with an independent database.
|
||||
* **Immediate Multi-Node Cluster Execution in v1:** This ADR defines the target architecture and phased roadmap; immediate implementation occurs incrementally post-[#655] v1.
|
||||
|
||||
---
|
||||
|
||||
## 3. High-Availability & Rolling-Restart Architecture
|
||||
|
||||
### 3.1 Architecture Overview
|
||||
|
||||
```
|
||||
+----------------------------+
|
||||
| LLM Clients / IDE Sessions |
|
||||
+--------------+-------------+
|
||||
|
|
||||
v
|
||||
+----------------------------+
|
||||
| HA Proxy / Router |
|
||||
| (Health-based & Affinity) |
|
||||
+------+--------------+------+
|
||||
| |
|
||||
+--------------+ +--------------+
|
||||
v v
|
||||
+--------------------+ +--------------------+
|
||||
| MCP Instance Node A| | MCP Instance Node B|
|
||||
| (Version N) | | (Version N+1) |
|
||||
+---------+----------+ +---------+----------+
|
||||
| |
|
||||
+----------------------+----------------------+
|
||||
|
|
||||
v
|
||||
+----------------------------+
|
||||
| Control-Plane DB Substrate|
|
||||
| (Shared Lease & Locks) |
|
||||
+--------------+-------------+
|
||||
|
|
||||
v
|
||||
+----------------------------+
|
||||
| Gitea API |
|
||||
+----------------------------+
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### 3.2 Key System Components
|
||||
|
||||
#### A. Multiple MCP Instance Cohorts
|
||||
* The control plane runs across $N \ge 2$ redundant process nodes.
|
||||
* Dual-namespace deployment allows running the old version (Node A) alongside a updated version (Node B) during rolling upgrades.
|
||||
|
||||
#### B. Shared Durable Session Storage & Resume Tokens
|
||||
* Session context, preflight verification proofs, and capability resolution states are stored in the shared control-plane database.
|
||||
* Client requests carry an explicit `session_id` and `resume_token`. If an MCP instance restarts or a request routes to a different instance, the target node validates the token against the database without requiring full session re-initialization.
|
||||
|
||||
#### C. Shared Lease Authority & Fencing Counters
|
||||
* Workflow leases (`gitea_allocate_next_work`, `gitea_adopt_workflow_lease`) use monotonic fencing tokens (`lease_generation_id`).
|
||||
* When Node B acquires or renews a lease, it increments the generation counter. Any delayed or out-of-order write attempt from Node A using an older generation token is rejected by database constraints.
|
||||
|
||||
#### D. Leader Election & Coordinated Drain
|
||||
* Node clusters elect a primary coordinator node for administrative background tasks (such as stale lease cleanup or incident Watchdogs).
|
||||
* During a rolling deployment:
|
||||
1. Node B (new version) is launched and registers as healthy.
|
||||
2. Router directs new session creations to Node B.
|
||||
3. Node A enters `MAINTENANCE_DRAIN` status ([#659](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/659)), completing in-flight mutations while refusing new tasks.
|
||||
4. Once all active sessions migrate or complete, Node A shuts down cleanly.
|
||||
|
||||
#### E. Idempotent Mutations & Failover Safety
|
||||
* All state-changing tool executions (PR creation, review submission, merge operations, label changes) carry a deterministic `idempotency_key`.
|
||||
* If a network connection flaps or a node fails mid-mutation, the re-issued request with the same `idempotency_key` is recognized by the control-plane substrate, returning the existing recorded result without repeating side effects on Gitea.
|
||||
|
||||
#### F. Schema Version Compatibility
|
||||
* Database migrations follow non-breaking additive patterns.
|
||||
* During rolling upgrades where Node A (Version $N$) and Node B (Version $N+1$) run concurrently, both versions operate against the shared schema without structural conflicts.
|
||||
|
||||
---
|
||||
|
||||
## 4. Split-Brain & Failure Behavior
|
||||
|
||||
### 4.1 Split-Brain Risk Scenarios & Mitigation
|
||||
|
||||
| Scenario | Risk | Mitigation Strategy |
|
||||
|---|---|---|
|
||||
| **Network Partition between Nodes** | Both Node A and Node B attempt to process operations for the same issue/PR. | **Generation Fencing:** Lease renewal requires updating the DB generation counter. The node isolated from the DB fails closed immediately. |
|
||||
| **Stale Node Recovery** | Node A recovers after a long pause and executes a queued mutation. | **Lease Expiry & TTL Fencing:** Transactions verify that `expires_at > NOW()` within the atomic SQLite transaction boundaries. |
|
||||
| **Database Connection Loss** | Node loses access to shared control-plane DB substrate. | **Strict Fail-Closed:** The node immediately marks all task capabilities as `blocked` and rejects mutation tools until DB connectivity is re-established. |
|
||||
|
||||
---
|
||||
|
||||
## 5. Phased Implementation Milestones
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
M1[Milestone 1: Shared Control-Plane DB Schema & Resume Tokens] --> M2[Milestone 2: Idempotent Mutation Layer]
|
||||
M2 --> M3[Milestone 3: Health Routing & Standby Failover]
|
||||
M3 --> M4[Milestone 4: Active-Active Rolling Deployment & Auto-Drain]
|
||||
```
|
||||
|
||||
### Milestone 1: Shared Control-Plane DB Schema & Resume Tokens (Post-#655)
|
||||
* Extend [#613] Control-Plane DB schema to store multi-instance node heartbeat records and session resume tokens.
|
||||
* Enable session lookup across instances via `session_id`.
|
||||
|
||||
### Milestone 2: Idempotent Mutation Layer & Lease Fencing
|
||||
* Add mandatory `idempotency_key` tracking to all Gitea mutation tools.
|
||||
* Implement monotonic lease fencing counters in `gitea_allocate_next_work` and `gitea_adopt_workflow_lease`.
|
||||
|
||||
### Milestone 3: Health-Based Routing & Active-Passive Standby
|
||||
* Introduce lightweight proxy/router capable of checking node health endpoints.
|
||||
* Implement active-standby failover where standby node automatically assumes work if active node fails health checks.
|
||||
|
||||
### Milestone 4: Active-Active Horizontal Deployment & Rolling Upgrade Automation
|
||||
* Enable true active-active multi-instance execution.
|
||||
* Integrate automated zero-downtime rolling upgrades coordinated with `gitea_request_mcp_restart` maintenance drain.
|
||||
|
||||
---
|
||||
|
||||
## 6. Observability & Audit Requirements
|
||||
|
||||
High-availability control plane operations must expose clear telemetry and audit trails:
|
||||
|
||||
* **Node Registry Telemetry:** Active nodes, version numbers, uptime, and heartbeat timestamps reported via `gitea_get_runtime_context`.
|
||||
* **Lease Fencing Metrics:** Tracking lease acquire latency, fence rejection counts, and lease handoff durations.
|
||||
* **Failover & Re-route Audit Logs:** Durable logging of session migrations between nodes, drain initiation, and process retirement events.
|
||||
|
||||
---
|
||||
|
||||
## 7. Tradeoffs & Accepted Risks
|
||||
|
||||
* **Increased Architectural Complexity:** Moving from a single process to a multi-instance control plane requires robust DB locking, proxy routing, and migration governance.
|
||||
* **Database Dependency:** The control-plane database substrate becomes a critical shared dependency for multi-node deployments. High availability for the underlying SQLite file system / DB must be guaranteed.
|
||||
@@ -80,6 +80,8 @@ status, onboarding checklist state, and the fail-closed error payloads (#635).
|
||||
| `/sessions` | Runtime and session view (#641) — health + inventory sessions/namespaces/worktrees |
|
||||
| `/api/sessions` | JSON export for the runtime/session view |
|
||||
| `/api/v1/sessions` | Versioned alias of `/api/sessions` |
|
||||
| `/gitea` | Gitea issue↔PR linkage console (#645) — both directions, with the evidence for each edge |
|
||||
| `/api/v1/gitea/linkage` | JSON linkage export; `502` when the read could not be answered |
|
||||
| `/inventory` | Phase 1 shell stub — unified inventory (backed by #636) |
|
||||
| `/timeline` | Phase 1 shell stub — workflow event timeline |
|
||||
| `/policy` | Phase 1 shell stub — capability/role policy placeholder |
|
||||
@@ -327,6 +329,68 @@ Honesty rules specific to this view:
|
||||
The write-time redactor is a narrow denylist and is not relied on. The field
|
||||
itself is kept — it is the `#630` evidence naming which daemon was killed.
|
||||
|
||||
## Gitea issue/PR linkage (#645)
|
||||
|
||||
`/gitea` is the Phase 3 read-only linkage console: which PR carries which issue,
|
||||
which issues are claimed by more than one PR, and what the latest Canonical
|
||||
Thread Handoff on a thread said. Gitea remains the source of truth — this
|
||||
surface reads it and never writes to it. There is no issue/PR editor, no review,
|
||||
and no merge control.
|
||||
|
||||
Query parameters (all optional):
|
||||
|
||||
| Parameter | Meaning |
|
||||
|-----------|---------|
|
||||
| `project` | Registry project id to scope the read (default: first registry entry) |
|
||||
| `state` | `open` (default) or `all`; `all` widens the window to merged/closed items, where a landed edge lives |
|
||||
| `issue=N` / `pr=N` | Focus one thread and load *its* latest canonical handoff |
|
||||
|
||||
`GET /api/v1/gitea/linkage` returns the same model as JSON
|
||||
(`schema_version: 1`). It answers `502` when the read could not be answered, so
|
||||
an automated consumer cannot mistake a fail-closed payload for "no links exist".
|
||||
The HTML page always answers `200` and renders the reason instead — an operator
|
||||
view must show why a read failed rather than withhold the page.
|
||||
|
||||
### How an edge is found
|
||||
|
||||
Each edge carries the evidence that produced it, strongest first:
|
||||
|
||||
| Evidence | Meaning |
|
||||
|----------|---------|
|
||||
| `closes_keyword` | The PR title or body declares `closes/fixes/resolves #N`. Gitea itself acts on this keyword. |
|
||||
| `branch_marker` | The PR head branch carries the canonical `(fix\|feat\|docs\|chore)/issue-N-…` marker minted by the issue lock. |
|
||||
| `body_reference` | The PR body mentions `#N` with no closing keyword. A mention is not a claim to close. |
|
||||
|
||||
Only closing and branch-marker edges populate the **issue → PR** direction: a
|
||||
bare mention is a cross-link, and counting it as ownership would invent
|
||||
contested issues out of ordinary references. The mention stays visible on the
|
||||
**PR → issue** side, labelled as such. A PR whose two strongest edges tie is
|
||||
flagged `ambiguous`; an issue claimed by two PRs is flagged `contested`.
|
||||
|
||||
### Honesty rules specific to this view
|
||||
|
||||
* **A partial read never reads as an absence.** Linkage is a claim about the
|
||||
loaded window only. When pagination did not complete, every empty edge cell
|
||||
renders `none found (partial inventory)` rather than `none`, and the JSON
|
||||
carries `inventory_complete: false` plus per-row `links_authoritative: false`.
|
||||
* **A failed read renders no table at all.** Missing credentials, an unknown
|
||||
project, or a fetch error produce `ok: false` with a reason. An empty linkage
|
||||
table would assert that no issue is linked to any PR, which such a read is not
|
||||
in a position to claim.
|
||||
* **Handoffs are loaded, never assumed.** CTH comments are thread-scoped, so
|
||||
only the focused issue or PR has its comments fetched. Every other row reports
|
||||
`not_loaded` with the reason; a thread whose comments *were* loaded and carried
|
||||
no CTH says exactly that. A comment-source failure degrades the handoff alone —
|
||||
the linkage tables still render.
|
||||
* **Unrecognised handoff headings are reported, not republished.** A `## CTH:`
|
||||
heading outside `CTH_TYPES` renders as `unrecognized`.
|
||||
* **Redaction precedes display.** Titles, labels, handoff fields, and error
|
||||
reasons pass through `webui.console_redaction` before serialization, and the
|
||||
page HTML-escapes everything it renders.
|
||||
* **Deep links are opt-in.** A link out to the Gitea web UI appears only when
|
||||
`GITEA_MCP_REVEAL_ENDPOINTS=1` is set server-side, matching how the MCP tools
|
||||
gate URL exposure. Item numbers stay usable without it.
|
||||
|
||||
## System-health dashboard (#639)
|
||||
|
||||
`/system-health` renders the same snapshot the `/api/v1/system/health` API
|
||||
|
||||
@@ -0,0 +1,511 @@
|
||||
"""Tests for the Gitea issue↔PR linkage console (#645, Phase 3).
|
||||
|
||||
Covers the acceptance criteria of the issue:
|
||||
|
||||
* AC1 — issue↔PR linkage is visible for the selected project/repo, in both
|
||||
directions, with the evidence that produced each edge.
|
||||
* AC2 — the latest canonical handoff (CTH) is summarized for a focused thread.
|
||||
* AC3 — an external Gitea link appears only under the admin reveal opt-in.
|
||||
* AC4 — every case is driven by mocked Gitea payloads; no network.
|
||||
|
||||
Plus the invariants this console must not violate: a partial or failed read is
|
||||
never rendered as "no link exists", an unfetched thread is never rendered as
|
||||
"no handoff", redaction happens before display, and the surface stays read-only.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from tests.webui_testclient import TestClient
|
||||
|
||||
from canonical_thread_handoff import format_cth_body
|
||||
from webui.app import create_app
|
||||
from webui.linkage_loader import (
|
||||
EVIDENCE_BRANCH,
|
||||
EVIDENCE_CLOSES,
|
||||
EVIDENCE_REFERENCE,
|
||||
HANDOFF_LOADED,
|
||||
HANDOFF_NOT_LOADED,
|
||||
HANDOFF_UNAVAILABLE,
|
||||
LinkageSnapshot,
|
||||
load_linkage_snapshot,
|
||||
resolve_linkage,
|
||||
resolve_pr_links,
|
||||
snapshot_to_dict,
|
||||
summarize_handoff,
|
||||
)
|
||||
from webui.linkage_views import render_linkage_page
|
||||
from webui.nav import nav_hrefs
|
||||
from webui.queue_loader import PaginationMeta
|
||||
|
||||
|
||||
def _pagination(*, complete: bool = True, count: int = 0) -> PaginationMeta:
|
||||
return PaginationMeta(
|
||||
page=1,
|
||||
per_page=50,
|
||||
returned_count=count,
|
||||
has_more=not complete,
|
||||
is_final_page=complete,
|
||||
inventory_complete=complete,
|
||||
pages_fetched=1,
|
||||
)
|
||||
|
||||
|
||||
def _pr(
|
||||
number: int,
|
||||
*,
|
||||
title: str = "",
|
||||
body: str = "",
|
||||
head: str = "",
|
||||
state: str = "open",
|
||||
labels: tuple[str, ...] = (),
|
||||
) -> dict:
|
||||
return {
|
||||
"number": number,
|
||||
"title": title or f"pr {number}",
|
||||
"body": body,
|
||||
"state": state,
|
||||
"head": {"ref": head},
|
||||
"labels": [{"name": name} for name in labels],
|
||||
}
|
||||
|
||||
|
||||
def _issue(
|
||||
number: int,
|
||||
*,
|
||||
title: str = "",
|
||||
state: str = "open",
|
||||
labels: tuple[str, ...] = (),
|
||||
) -> dict:
|
||||
return {
|
||||
"number": number,
|
||||
"title": title or f"issue {number}",
|
||||
"state": state,
|
||||
"labels": [{"name": name} for name in labels],
|
||||
}
|
||||
|
||||
|
||||
def _fetcher(items: list[dict], *, complete: bool = True):
|
||||
def _fetch(*_args, **_kwargs):
|
||||
return items, _pagination(complete=complete, count=len(items))
|
||||
|
||||
return _fetch
|
||||
|
||||
|
||||
def _load(
|
||||
issues: list[dict],
|
||||
prs: list[dict],
|
||||
*,
|
||||
complete: bool = True,
|
||||
**kwargs,
|
||||
) -> LinkageSnapshot:
|
||||
return load_linkage_snapshot(
|
||||
fetch_prs=_fetcher(prs, complete=complete),
|
||||
fetch_issues=_fetcher(issues, complete=complete),
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
def _cth(comment_id: int, *, created_at: str, status: str, next_owner: str) -> dict:
|
||||
return {
|
||||
"id": comment_id,
|
||||
"created_at": created_at,
|
||||
"user": {"login": "jcwalker3"},
|
||||
"body": format_cth_body(
|
||||
cth_type="Author Handoff",
|
||||
status=status,
|
||||
next_owner=next_owner,
|
||||
current_blocker="none",
|
||||
decision="implemented",
|
||||
proof="full suite green",
|
||||
next_action="review PR",
|
||||
ready_to_paste_prompt="Review PR #902 now.",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
class TestLinkageEvidence(unittest.TestCase):
|
||||
"""AC1 — every edge records how it was found, and keeps all candidates."""
|
||||
|
||||
def test_closes_keyword_in_body_is_strongest_evidence(self):
|
||||
links = resolve_pr_links(_pr(902, body="Closes #643"))
|
||||
self.assertEqual([link.issue_number for link in links], [643])
|
||||
self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
|
||||
self.assertTrue(links[0].closes)
|
||||
|
||||
def test_closes_keyword_in_title_counts(self):
|
||||
links = resolve_pr_links(_pr(902, title="feat(webui): preview (Closes #643)"))
|
||||
self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
|
||||
|
||||
def test_canonical_branch_marker_links_without_a_keyword(self):
|
||||
links = resolve_pr_links(_pr(902, head="feat/issue-643-request-preview"))
|
||||
self.assertEqual([link.issue_number for link in links], [643])
|
||||
self.assertEqual(links[0].evidence, (EVIDENCE_BRANCH,))
|
||||
self.assertFalse(links[0].closes)
|
||||
|
||||
def test_non_canonical_branch_is_not_treated_as_a_marker(self):
|
||||
self.assertEqual(resolve_pr_links(_pr(902, head="issue-643-preview")), ())
|
||||
|
||||
def test_bare_mention_is_recorded_as_the_weakest_evidence(self):
|
||||
links = resolve_pr_links(_pr(902, body="context in #643"))
|
||||
self.assertEqual(links[0].evidence, (EVIDENCE_REFERENCE,))
|
||||
self.assertFalse(links[0].closes)
|
||||
|
||||
def test_several_evidence_kinds_merge_onto_one_edge(self):
|
||||
links = resolve_pr_links(
|
||||
_pr(902, body="Closes #643 — see #643", head="feat/issue-643-preview")
|
||||
)
|
||||
self.assertEqual(len(links), 1)
|
||||
self.assertEqual(
|
||||
links[0].evidence,
|
||||
(EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE),
|
||||
)
|
||||
|
||||
def test_stronger_evidence_sorts_first(self):
|
||||
links = resolve_pr_links(_pr(902, body="Closes #643, related #700"))
|
||||
self.assertEqual([link.issue_number for link in links], [643, 700])
|
||||
|
||||
def test_self_reference_is_not_linkage(self):
|
||||
links = resolve_pr_links(_pr(902, body="supersedes #902"))
|
||||
self.assertEqual(links, ())
|
||||
|
||||
def test_every_candidate_is_kept_never_collapsed_to_a_guess(self):
|
||||
links = resolve_pr_links(_pr(902, body="Closes #643\nCloses #644"))
|
||||
self.assertEqual([link.issue_number for link in links], [643, 644])
|
||||
|
||||
|
||||
class TestLinkageIndex(unittest.TestCase):
|
||||
def test_issue_direction_ignores_mention_only_edges(self):
|
||||
index = resolve_linkage([_pr(902, body="context in #643")])
|
||||
self.assertIsNone(index.issue_prs.get(643))
|
||||
self.assertEqual(index.pr_links[902][0].evidence, (EVIDENCE_REFERENCE,))
|
||||
|
||||
def test_contested_issue_is_reported_when_two_prs_claim_it(self):
|
||||
index = resolve_linkage(
|
||||
[_pr(902, body="Closes #643"), _pr(903, head="feat/issue-643-again")]
|
||||
)
|
||||
self.assertEqual(index.contested_issues(), (643,))
|
||||
self.assertEqual(index.issue_prs[643], (902, 903))
|
||||
|
||||
def test_single_claim_is_not_contested(self):
|
||||
index = resolve_linkage([_pr(902, body="Closes #643")])
|
||||
self.assertEqual(index.contested_issues(), ())
|
||||
|
||||
def test_ambiguous_when_two_issues_tie_at_the_strongest_evidence(self):
|
||||
index = resolve_linkage([_pr(902, body="Closes #643\nCloses #644")])
|
||||
self.assertTrue(index.ambiguous(902))
|
||||
|
||||
def test_weaker_candidate_alongside_a_stronger_one_is_not_ambiguous(self):
|
||||
index = resolve_linkage([_pr(902, body="Closes #643, see #700")])
|
||||
self.assertFalse(index.ambiguous(902))
|
||||
self.assertEqual(index.primary_issue(902).issue_number, 643)
|
||||
|
||||
def test_malformed_pr_row_is_skipped_not_raised_on(self):
|
||||
index = resolve_linkage([{"title": "no number"}, _pr(902, body="Closes #643")])
|
||||
self.assertEqual(sorted(index.pr_links), [902])
|
||||
|
||||
|
||||
class TestLinkageSnapshot(unittest.TestCase):
|
||||
"""AC1 — linkage is visible per project/repo, in both directions."""
|
||||
|
||||
def test_both_directions_are_populated(self):
|
||||
snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
|
||||
self.assertTrue(snapshot.ok)
|
||||
self.assertEqual([node.number for node in snapshot.issues], [643])
|
||||
self.assertEqual(snapshot.issues[0].linked_prs, (902,))
|
||||
self.assertEqual(snapshot.prs[0].links[0].issue_number, 643)
|
||||
|
||||
def test_repo_scope_comes_from_the_registry_project(self):
|
||||
snapshot = _load([], [])
|
||||
self.assertIn("/", snapshot.repo_label)
|
||||
self.assertTrue(snapshot.project_id)
|
||||
|
||||
def test_unknown_project_fails_closed_with_a_reason(self):
|
||||
snapshot = _load([_issue(643)], [], project_id="no-such-project")
|
||||
self.assertFalse(snapshot.ok)
|
||||
self.assertIn("not found in registry", snapshot.fetch_error)
|
||||
self.assertEqual(snapshot.issues, ())
|
||||
|
||||
def test_orphan_pr_is_identifiable(self):
|
||||
snapshot = _load([], [_pr(902), _pr(903, body="Closes #643")])
|
||||
self.assertEqual([node.number for node in snapshot.orphan_prs], [902])
|
||||
|
||||
def test_state_scope_defaults_to_open_and_is_reported(self):
|
||||
self.assertEqual(_load([], []).state_scope, "open")
|
||||
self.assertEqual(_load([], [], state="all").state_scope, "all")
|
||||
|
||||
def test_unsupported_state_falls_back_to_open(self):
|
||||
self.assertEqual(_load([], [], state="../etc").state_scope, "open")
|
||||
|
||||
def test_state_is_passed_through_to_the_fetchers(self):
|
||||
seen: list[str] = []
|
||||
|
||||
def _fetch(*_args, **kwargs):
|
||||
seen.append(kwargs.get("state", ""))
|
||||
return [], _pagination()
|
||||
|
||||
load_linkage_snapshot(state="all", fetch_prs=_fetch, fetch_issues=_fetch)
|
||||
self.assertEqual(seen, ["all", "all"])
|
||||
|
||||
|
||||
class TestPartialInventoryIsNotAnAbsenceClaim(unittest.TestCase):
|
||||
"""An empty edge list from a partial read must never read as 'no link'."""
|
||||
|
||||
def test_incomplete_pagination_marks_links_non_authoritative(self):
|
||||
snapshot = _load([_issue(643)], [], complete=False)
|
||||
self.assertFalse(snapshot.inventory_complete)
|
||||
self.assertFalse(snapshot.issues[0].links_authoritative)
|
||||
|
||||
def test_complete_pagination_marks_links_authoritative(self):
|
||||
snapshot = _load([_issue(643)], [], complete=True)
|
||||
self.assertTrue(snapshot.inventory_complete)
|
||||
self.assertTrue(snapshot.issues[0].links_authoritative)
|
||||
|
||||
def test_partial_window_renders_a_qualified_empty_cell(self):
|
||||
html = render_linkage_page(_load([_issue(643)], [], complete=False))
|
||||
self.assertIn("none found (partial inventory)", html)
|
||||
|
||||
def test_complete_window_renders_a_plain_none(self):
|
||||
html = render_linkage_page(_load([_issue(643)], [], complete=True))
|
||||
self.assertNotIn("partial inventory", html)
|
||||
self.assertIn(">none<", html)
|
||||
|
||||
def test_missing_credentials_fail_closed_without_a_table(self):
|
||||
with mock.patch(
|
||||
"webui.linkage_loader._offline_test_mode", return_value=False
|
||||
), mock.patch("webui.linkage_loader.get_auth_header", return_value=""):
|
||||
snapshot = load_linkage_snapshot()
|
||||
self.assertFalse(snapshot.ok)
|
||||
self.assertIn("credentials unavailable", snapshot.fetch_error)
|
||||
html = render_linkage_page(snapshot)
|
||||
self.assertIn("Linkage unavailable", html)
|
||||
self.assertNotIn("Issues → pull requests", html)
|
||||
|
||||
def test_fetch_failure_is_reported_not_raised(self):
|
||||
def _boom(*_args, **_kwargs):
|
||||
raise RuntimeError("gitea 502")
|
||||
|
||||
snapshot = load_linkage_snapshot(fetch_prs=_boom, fetch_issues=_boom)
|
||||
self.assertFalse(snapshot.ok)
|
||||
self.assertIn("Gitea fetch failed", snapshot.fetch_error)
|
||||
|
||||
|
||||
class TestHandoffSummary(unittest.TestCase):
|
||||
"""AC2 — the latest canonical handoff is summarized for a focused thread."""
|
||||
|
||||
def test_latest_cth_wins(self):
|
||||
summary = summarize_handoff([
|
||||
_cth(1, created_at="2026-07-24T10:00:00Z", status="in progress",
|
||||
next_owner="author"),
|
||||
_cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
|
||||
next_owner="reviewer"),
|
||||
])
|
||||
self.assertEqual(summary.comment_id, 2)
|
||||
self.assertEqual(summary.status, "PR-open")
|
||||
self.assertEqual(summary.next_owner, "reviewer")
|
||||
self.assertTrue(summary.cth_type_known)
|
||||
|
||||
def test_thread_without_a_cth_summarizes_to_none(self):
|
||||
self.assertIsNone(summarize_handoff([{"id": 1, "body": "ordinary comment"}]))
|
||||
|
||||
def test_unknown_heading_is_reported_not_republished(self):
|
||||
summary = summarize_handoff([
|
||||
{
|
||||
"id": 5,
|
||||
"created_at": "2026-07-25T10:00:00Z",
|
||||
"user": {"login": "someone"},
|
||||
"body": "<!-- cth:v1 -->\n## CTH: Totally Made Up\n\nStatus: odd\n",
|
||||
}
|
||||
])
|
||||
self.assertFalse(summary.cth_type_known)
|
||||
self.assertEqual(summary.cth_type, "unrecognized")
|
||||
self.assertNotIn("Totally Made Up", json.dumps(summary.to_dict()))
|
||||
|
||||
def test_focused_pr_loads_its_handoff(self):
|
||||
snapshot = _load(
|
||||
[_issue(643)],
|
||||
[_pr(902, body="Closes #643")],
|
||||
pr=902,
|
||||
comment_source=lambda kind, number: [
|
||||
_cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
|
||||
next_owner="reviewer")
|
||||
],
|
||||
)
|
||||
self.assertEqual(snapshot.handoff_status.state, HANDOFF_LOADED)
|
||||
self.assertEqual(snapshot.focus, ("pr", 902))
|
||||
self.assertEqual(snapshot.prs[0].handoff.status, "PR-open")
|
||||
|
||||
def test_unfocused_rows_report_not_loaded_never_none(self):
|
||||
snapshot = _load(
|
||||
[_issue(643)],
|
||||
[_pr(902, body="Closes #643"), _pr(903)],
|
||||
pr=902,
|
||||
comment_source=lambda kind, number: [],
|
||||
)
|
||||
other = next(node for node in snapshot.prs if node.number == 903)
|
||||
self.assertIsNone(other.handoff)
|
||||
self.assertEqual(other.handoff_status.state, HANDOFF_NOT_LOADED)
|
||||
self.assertIn("not loaded", render_linkage_page(snapshot))
|
||||
|
||||
def test_no_focus_means_no_thread_is_claimed_handoff_free(self):
|
||||
snapshot = _load([_issue(643)], [])
|
||||
self.assertEqual(snapshot.handoff_status.state, HANDOFF_NOT_LOADED)
|
||||
self.assertIn("thread-scoped", snapshot.handoff_status.reason)
|
||||
|
||||
def test_comment_source_failure_degrades_only_the_handoff(self):
|
||||
def _boom(_kind, _number):
|
||||
raise RuntimeError("comments 500")
|
||||
|
||||
snapshot = _load(
|
||||
[_issue(643)], [_pr(902, body="Closes #643")], pr=902, comment_source=_boom
|
||||
)
|
||||
self.assertTrue(snapshot.ok)
|
||||
self.assertEqual(snapshot.handoff_status.state, HANDOFF_UNAVAILABLE)
|
||||
self.assertEqual(snapshot.issues[0].linked_prs, (902,))
|
||||
self.assertIn("unavailable", render_linkage_page(snapshot))
|
||||
|
||||
def test_loaded_thread_with_no_cth_says_so_explicitly(self):
|
||||
snapshot = _load(
|
||||
[_issue(643)],
|
||||
[_pr(902, body="Closes #643")],
|
||||
pr=902,
|
||||
comment_source=lambda kind, number: [{"id": 1, "body": "hi"}],
|
||||
)
|
||||
self.assertIn(
|
||||
"no Canonical Thread Handoff comment found", render_linkage_page(snapshot)
|
||||
)
|
||||
|
||||
|
||||
class TestDeepLinks(unittest.TestCase):
|
||||
"""AC3 — an external Gitea link is emitted only when permitted."""
|
||||
|
||||
def test_deep_links_are_withheld_by_default(self):
|
||||
with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": ""}):
|
||||
snapshot = _load([_issue(643)], [])
|
||||
html = render_linkage_page(snapshot)
|
||||
self.assertFalse(snapshot.deep_links_enabled)
|
||||
self.assertIsNone(snapshot.issues[0].deep_link)
|
||||
self.assertIn("Gitea deep links are withheld", html)
|
||||
|
||||
def test_reveal_opt_in_emits_the_link(self):
|
||||
with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}):
|
||||
snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
|
||||
html = render_linkage_page(snapshot)
|
||||
self.assertTrue(snapshot.deep_links_enabled)
|
||||
self.assertIn("/issues/643", snapshot.issues[0].deep_link)
|
||||
self.assertIn("/pulls/902", snapshot.prs[0].deep_link)
|
||||
self.assertIn(f'href="{snapshot.issues[0].deep_link}"', html)
|
||||
|
||||
|
||||
class TestRedactionBoundary(unittest.TestCase):
|
||||
def test_secret_shaped_title_is_redacted_before_display(self):
|
||||
snapshot = _load(
|
||||
[_issue(643, title="token=ghp_thisisnotarealsecretvalue0001")], []
|
||||
)
|
||||
payload = json.dumps(snapshot_to_dict(snapshot))
|
||||
self.assertNotIn("ghp_thisisnotarealsecretvalue0001", payload)
|
||||
self.assertNotIn(
|
||||
"ghp_thisisnotarealsecretvalue0001", render_linkage_page(snapshot)
|
||||
)
|
||||
|
||||
def test_handoff_fields_are_redacted(self):
|
||||
comment = _cth(
|
||||
2, created_at="2026-07-25T10:00:00Z", status="ok", next_owner="reviewer"
|
||||
)
|
||||
comment["body"] += "\nDecision: password=hunter2hunter2\n"
|
||||
snapshot = _load(
|
||||
[_issue(643)],
|
||||
[_pr(902, body="Closes #643")],
|
||||
pr=902,
|
||||
comment_source=lambda kind, number: [comment],
|
||||
)
|
||||
self.assertNotIn("hunter2hunter2", json.dumps(snapshot_to_dict(snapshot)))
|
||||
self.assertNotIn("hunter2hunter2", render_linkage_page(snapshot))
|
||||
|
||||
def test_html_escapes_markup_in_a_title(self):
|
||||
snapshot = _load([_issue(643, title="<script>alert(1)</script>")], [])
|
||||
html = render_linkage_page(snapshot)
|
||||
self.assertNotIn("<script>alert(1)</script>", html)
|
||||
self.assertIn("<script>", html)
|
||||
|
||||
|
||||
class TestLinkageRoutes(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.snapshot = _load(
|
||||
[_issue(643, labels=("status:ready",))],
|
||||
[_pr(902, body="Closes #643", labels=("status:pr-open",))],
|
||||
)
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_page_renders_both_tables(self):
|
||||
with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
|
||||
response = self.client.get("/gitea")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Issues → pull requests", response.text)
|
||||
self.assertIn("Pull requests → issues", response.text)
|
||||
self.assertIn("#643", response.text)
|
||||
|
||||
def test_api_exports_the_same_model(self):
|
||||
with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
|
||||
response = self.client.get("/api/v1/gitea/linkage")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
payload = response.json()
|
||||
self.assertTrue(payload["ok"])
|
||||
self.assertEqual(payload["issues"][0]["linked_prs"], [902])
|
||||
self.assertEqual(payload["prs"][0]["links"][0]["issue_number"], 643)
|
||||
self.assertEqual(payload["schema_version"], 1)
|
||||
|
||||
def test_api_declares_the_evidence_vocabulary(self):
|
||||
with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
|
||||
payload = self.client.get("/api/v1/gitea/linkage").json()
|
||||
names = {entry["name"] for entry in payload["evidence_kinds"]}
|
||||
self.assertEqual(names, {EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE})
|
||||
|
||||
def test_api_fails_closed_with_a_non_200_when_the_read_failed(self):
|
||||
failed = _load([], [], project_id="no-such-project")
|
||||
with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
|
||||
response = self.client.get("/api/v1/gitea/linkage")
|
||||
self.assertEqual(response.status_code, 502)
|
||||
self.assertFalse(response.json()["ok"])
|
||||
|
||||
def test_page_still_renders_when_the_read_failed(self):
|
||||
failed = _load([], [], project_id="no-such-project")
|
||||
with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
|
||||
response = self.client.get("/gitea")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("Linkage unavailable", response.text)
|
||||
|
||||
def test_query_parameters_reach_the_loader(self):
|
||||
with mock.patch(
|
||||
"webui.app.load_linkage_snapshot", return_value=self.snapshot
|
||||
) as loader:
|
||||
self.client.get("/gitea?project=gitea-tools&state=all&pr=902")
|
||||
loader.assert_called_once()
|
||||
args, kwargs = loader.call_args
|
||||
self.assertEqual(args[0], "gitea-tools")
|
||||
self.assertEqual(kwargs["state"], "all")
|
||||
self.assertEqual(kwargs["pr"], 902)
|
||||
self.assertIsNone(kwargs["issue"])
|
||||
|
||||
def test_surface_stays_read_only(self):
|
||||
for path in ("/gitea", "/api/v1/gitea/linkage"):
|
||||
with self.subTest(path=path):
|
||||
self.assertEqual(self.client.post(path).status_code, 405)
|
||||
|
||||
def test_nav_exposes_the_linkage_page_as_live(self):
|
||||
self.assertIn("/gitea", nav_hrefs())
|
||||
home = self.client.get("/").text
|
||||
self.assertIn('href="/gitea"', home)
|
||||
self.assertIn(">Gitea<", home)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -53,6 +53,11 @@ from webui.session_loader import (
|
||||
snapshot_to_dict as session_view_snapshot_to_dict,
|
||||
)
|
||||
from webui.session_views import render_sessions_page
|
||||
from webui.linkage_loader import (
|
||||
load_linkage_snapshot,
|
||||
snapshot_to_dict as linkage_snapshot_to_dict,
|
||||
)
|
||||
from webui.linkage_views import render_linkage_page
|
||||
from webui.inventory import (
|
||||
SECTION_NAMES as _INVENTORY_SECTIONS,
|
||||
load_inventory_snapshot,
|
||||
@@ -342,6 +347,41 @@ async def api_sessions(_request: Request) -> JSONResponse:
|
||||
return JSONResponse(session_view_snapshot_to_dict(load_session_view_snapshot()))
|
||||
|
||||
|
||||
def _linkage_snapshot(request: Request):
|
||||
"""Load one linkage snapshot from the request's scope and focus parameters."""
|
||||
return load_linkage_snapshot(
|
||||
request.query_params.get("project") or None,
|
||||
state=request.query_params.get("state"),
|
||||
issue=_query_int(request, "issue"),
|
||||
pr=_query_int(request, "pr"),
|
||||
)
|
||||
|
||||
|
||||
async def gitea_linkage(request: Request) -> HTMLResponse:
|
||||
"""Gitea issue↔PR linkage console (#645) — read-only.
|
||||
|
||||
Always 200, including on a failed read: this is an operator view, and it
|
||||
must render *why* linkage could not be loaded rather than withhold the page.
|
||||
The snapshot itself carries ``ok=False`` and the page refuses to draw a
|
||||
linkage table it cannot stand behind.
|
||||
"""
|
||||
return HTMLResponse(render_linkage_page(_linkage_snapshot(request)))
|
||||
|
||||
|
||||
async def api_v1_gitea_linkage(request: Request) -> JSONResponse:
|
||||
"""JSON export of the issue↔PR linkage model (#645).
|
||||
|
||||
Unlike the HTML view, the API answers with 502 when the snapshot could not
|
||||
be loaded, so an automated consumer cannot read a fail-closed payload as a
|
||||
successful "no links exist" result.
|
||||
"""
|
||||
snapshot = _linkage_snapshot(request)
|
||||
return JSONResponse(
|
||||
linkage_snapshot_to_dict(snapshot),
|
||||
status_code=200 if snapshot.ok else 502,
|
||||
)
|
||||
|
||||
|
||||
async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
|
||||
if request.method == "GET":
|
||||
return "", None
|
||||
@@ -785,6 +825,8 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
|
||||
Route("/api/sessions", api_sessions, methods=["GET"]),
|
||||
Route("/api/v1/sessions", api_sessions, methods=["GET"]),
|
||||
Route("/api/v1/timeline", api_v1_timeline, methods=["GET"]),
|
||||
Route("/gitea", gitea_linkage, methods=["GET"]),
|
||||
Route("/api/v1/gitea/linkage", api_v1_gitea_linkage, methods=["GET"]),
|
||||
Route("/analytics", analytics, methods=["GET"]),
|
||||
Route("/api/analytics", api_v1_analytics, methods=["GET"]),
|
||||
Route("/api/v1/analytics", api_v1_analytics, methods=["GET"]),
|
||||
|
||||
@@ -0,0 +1,771 @@
|
||||
"""Gitea issue↔PR linkage model for the console (#645, Phase 3).
|
||||
|
||||
Operators lose context between an issue and the PR that closes it: which PR
|
||||
carries which issue, whether two PRs claim the same issue, and what the latest
|
||||
canonical handoff on that thread said. The evidence exists in Gitea, but only
|
||||
as free text scattered across PR titles, bodies, and branch names.
|
||||
|
||||
This module resolves that linkage into one read-only model:
|
||||
|
||||
* :func:`resolve_linkage` is a pure function from raw Gitea issue/PR payloads to
|
||||
a :class:`LinkageIndex`. It records *how* each edge was found (a ``Closes #N``
|
||||
keyword, the canonical ``feat/issue-N-…`` branch marker, or a bare ``#N``
|
||||
body reference) and never collapses several candidates into one silent guess.
|
||||
* :func:`load_linkage_snapshot` scopes that index to a registry project and
|
||||
optionally attaches the latest Canonical Thread Handoff (CTH) summary for one
|
||||
focused issue or PR.
|
||||
|
||||
Design rules, matching the rest of the console:
|
||||
|
||||
- **Read-only.** Gitea is read through the shared authenticated helpers. No
|
||||
endpoint here mutates anything, and no write action is registered.
|
||||
- **Qualified absence.** Linkage is a claim about a *loaded* window of Gitea.
|
||||
When pagination did not complete, when credentials were unavailable, or when
|
||||
only open items were fetched, the snapshot says so and every "no linked PR"
|
||||
is marked non-authoritative. An empty edge list from a partial read is not
|
||||
evidence that no link exists.
|
||||
- **Handoff is loaded, never assumed.** CTH comments are thread-scoped, so they
|
||||
are fetched only for an explicitly focused issue or PR. Every other row
|
||||
reports ``not_loaded`` rather than rendering as "no handoff".
|
||||
- **Redaction at the boundary.** Titles, labels, handoff fields, and error
|
||||
reasons are free text from Gitea and cross :mod:`webui.console_redaction`
|
||||
before they leave this module.
|
||||
- **Deep links are opt-in.** A link to the Gitea web UI is emitted only under
|
||||
the ``GITEA_MCP_REVEAL_ENDPOINTS`` admin opt-in, exactly as the MCP tools
|
||||
gate their own URL exposure.
|
||||
|
||||
Non-goals (from the issue): no issue/PR editor, no browser review or merge, no
|
||||
reimplementation of Gitea search.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable, Iterable, Sequence
|
||||
|
||||
from gitea_auth import api_fetch_page, get_auth_header, gitea_url, repo_api_url
|
||||
|
||||
from webui import console_redaction
|
||||
from webui.project_registry import ProjectRecord, load_registry
|
||||
from webui.queue_loader import (
|
||||
PaginationMeta,
|
||||
_fetch_issues,
|
||||
_fetch_prs,
|
||||
_host_from_url,
|
||||
)
|
||||
|
||||
#: Version of the serialized linkage contract. Bump on any breaking change.
|
||||
LINKAGE_SCHEMA_VERSION = 1
|
||||
|
||||
# --- Linkage evidence -------------------------------------------------------
|
||||
# Ordered strongest to weakest. The strength ordering is what makes an
|
||||
# ambiguous PR detectable: two candidates at the same strength are a genuine
|
||||
# ambiguity, while a weaker candidate alongside a stronger one is not.
|
||||
EVIDENCE_CLOSES = "closes_keyword"
|
||||
EVIDENCE_BRANCH = "branch_marker"
|
||||
EVIDENCE_REFERENCE = "body_reference"
|
||||
|
||||
EVIDENCE_ORDER: tuple[str, ...] = (
|
||||
EVIDENCE_CLOSES,
|
||||
EVIDENCE_BRANCH,
|
||||
EVIDENCE_REFERENCE,
|
||||
)
|
||||
_EVIDENCE_RANK = {name: rank for rank, name in enumerate(EVIDENCE_ORDER)}
|
||||
|
||||
EVIDENCE_DESCRIPTIONS: dict[str, str] = {
|
||||
EVIDENCE_CLOSES: (
|
||||
"the PR title or body declares 'closes/fixes/resolves #N' — Gitea itself "
|
||||
"acts on this keyword, so it is the strongest available evidence"
|
||||
),
|
||||
EVIDENCE_BRANCH: (
|
||||
"the PR head branch carries the canonical issue marker "
|
||||
"'(fix|feat|docs|chore)/issue-N-…' minted by the issue lock"
|
||||
),
|
||||
EVIDENCE_REFERENCE: (
|
||||
"the PR body mentions '#N' without a closing keyword; a mention is not "
|
||||
"a claim that the PR closes that issue"
|
||||
),
|
||||
}
|
||||
|
||||
_CLOSES_RE = re.compile(r"(?:closes|fixes|resolves)\s+#(\d+)", re.IGNORECASE)
|
||||
_REFERENCE_RE = re.compile(r"#(\d+)")
|
||||
_BRANCH_MARKER_RE = re.compile(
|
||||
r"^(?:fix|feat|docs|chore)/issue-(\d+)(?:[-/]|$)", re.IGNORECASE
|
||||
)
|
||||
|
||||
# Handoff-source states. ``not_loaded`` is deliberately distinct from "none
|
||||
# found": a row whose comments were never fetched proves nothing about whether
|
||||
# a handoff exists on that thread.
|
||||
HANDOFF_NOT_LOADED = "not_loaded"
|
||||
HANDOFF_LOADED = "loaded"
|
||||
HANDOFF_UNAVAILABLE = "unavailable"
|
||||
|
||||
# Which item states were fetched. Linkage claims are scoped to this window.
|
||||
STATE_OPEN = "open"
|
||||
STATE_ALL = "all"
|
||||
_SUPPORTED_STATES = (STATE_OPEN, STATE_ALL)
|
||||
|
||||
|
||||
def _redact(value: Any) -> Any:
|
||||
"""Redact one free-text field, failing closed to the placeholder."""
|
||||
if value is None:
|
||||
return None
|
||||
return console_redaction.redact_text(str(value))
|
||||
|
||||
|
||||
def deep_links_enabled(env: dict[str, str] | None = None) -> bool:
|
||||
"""Whether Gitea web-UI deep links may be emitted (admin/debug opt-in)."""
|
||||
source = env if env is not None else os.environ
|
||||
return (source.get("GITEA_MCP_REVEAL_ENDPOINTS") or "").strip().lower() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
"on",
|
||||
}
|
||||
|
||||
|
||||
def _deep_link(host: str, org: str, repo: str, kind: str, number: int) -> str | None:
|
||||
"""Build a Gitea web link for one item, or None when reveal is not enabled."""
|
||||
if not deep_links_enabled() or not (host and org and repo):
|
||||
return None
|
||||
segment = "pulls" if kind == "pr" else "issues"
|
||||
try:
|
||||
return gitea_url(host, f"/{org}/{repo}/{segment}/{int(number)}")
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
# --- Pure linkage resolution -------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class IssueLink:
|
||||
"""One resolved edge from a PR to an issue, with the evidence that found it."""
|
||||
|
||||
issue_number: int
|
||||
evidence: tuple[str, ...]
|
||||
|
||||
@property
|
||||
def strength(self) -> int:
|
||||
"""Rank of the strongest evidence backing this edge (lower is stronger)."""
|
||||
return min(
|
||||
(_EVIDENCE_RANK.get(name, len(EVIDENCE_ORDER)) for name in self.evidence),
|
||||
default=len(EVIDENCE_ORDER),
|
||||
)
|
||||
|
||||
@property
|
||||
def closes(self) -> bool:
|
||||
"""True only when the PR *declares* it closes the issue."""
|
||||
return EVIDENCE_CLOSES in self.evidence
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"issue_number": self.issue_number,
|
||||
"evidence": list(self.evidence),
|
||||
"closes": self.closes,
|
||||
}
|
||||
|
||||
|
||||
def _sorted_links(links: Iterable[IssueLink]) -> tuple[IssueLink, ...]:
|
||||
return tuple(sorted(links, key=lambda link: (link.strength, link.issue_number)))
|
||||
|
||||
|
||||
def resolve_pr_links(pr: dict[str, Any]) -> tuple[IssueLink, ...]:
|
||||
"""Resolve every issue a PR points at, strongest evidence first.
|
||||
|
||||
Every candidate is kept. Collapsing to a single "linked issue" is what makes
|
||||
a mislinked or double-claimed PR invisible, so the caller decides what to do
|
||||
with several candidates rather than being handed one guess.
|
||||
"""
|
||||
found: dict[int, set[str]] = {}
|
||||
|
||||
def _add(number: Any, evidence: str) -> None:
|
||||
try:
|
||||
issue_number = int(number)
|
||||
except (TypeError, ValueError):
|
||||
return
|
||||
if issue_number <= 0:
|
||||
return
|
||||
found.setdefault(issue_number, set()).add(evidence)
|
||||
|
||||
title = str(pr.get("title") or "")
|
||||
body = str(pr.get("body") or "")
|
||||
for text in (title, body):
|
||||
for match in _CLOSES_RE.finditer(text):
|
||||
_add(match.group(1), EVIDENCE_CLOSES)
|
||||
|
||||
head_ref = str((pr.get("head") or {}).get("ref") or "")
|
||||
branch_match = _BRANCH_MARKER_RE.match(head_ref.strip())
|
||||
if branch_match:
|
||||
_add(branch_match.group(1), EVIDENCE_BRANCH)
|
||||
|
||||
# The ``#N`` inside "Closes #N" is the *same* textual occurrence as the
|
||||
# closing keyword, not a second, independent mention. Blanking the closing
|
||||
# phrases first keeps "mention" meaning what the legend says it means: a
|
||||
# reference the PR made without claiming to close anything.
|
||||
for match in _REFERENCE_RE.finditer(_CLOSES_RE.sub(" ", body)):
|
||||
_add(match.group(1), EVIDENCE_REFERENCE)
|
||||
|
||||
# A PR's own number appearing in its body is self-reference, not linkage.
|
||||
try:
|
||||
found.pop(int(pr.get("number")), None)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
|
||||
return _sorted_links(
|
||||
IssueLink(
|
||||
issue_number=number,
|
||||
evidence=tuple(name for name in EVIDENCE_ORDER if name in evidence),
|
||||
)
|
||||
for number, evidence in found.items()
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LinkageIndex:
|
||||
"""Resolved linkage over one loaded window of issues and PRs."""
|
||||
|
||||
pr_links: dict[int, tuple[IssueLink, ...]]
|
||||
issue_prs: dict[int, tuple[int, ...]]
|
||||
|
||||
def primary_issue(self, pr_number: int) -> IssueLink | None:
|
||||
"""The strongest edge for a PR, or None when it points at no issue."""
|
||||
links = self.pr_links.get(int(pr_number)) or ()
|
||||
return links[0] if links else None
|
||||
|
||||
def ambiguous(self, pr_number: int) -> bool:
|
||||
"""True when two or more issues tie at the PR's strongest evidence."""
|
||||
links = self.pr_links.get(int(pr_number)) or ()
|
||||
if len(links) < 2:
|
||||
return False
|
||||
best = links[0].strength
|
||||
return sum(1 for link in links if link.strength == best) > 1
|
||||
|
||||
def contested_issues(self) -> tuple[int, ...]:
|
||||
"""Issues claimed by more than one PR in the loaded window."""
|
||||
return tuple(
|
||||
number for number, prs in sorted(self.issue_prs.items()) if len(prs) > 1
|
||||
)
|
||||
|
||||
|
||||
def resolve_linkage(prs: Sequence[dict[str, Any]]) -> LinkageIndex:
|
||||
"""Build the bidirectional linkage index for a loaded window of PRs.
|
||||
|
||||
Only *closing* and *branch-marker* edges populate the issue→PR direction: a
|
||||
bare ``#N`` mention is a reference, and treating it as "this PR is the work
|
||||
for issue N" would invent contested issues out of ordinary cross-links. The
|
||||
weaker edge stays visible on the PR→issue side, where it is labelled.
|
||||
"""
|
||||
pr_links: dict[int, tuple[IssueLink, ...]] = {}
|
||||
issue_prs: dict[int, list[int]] = {}
|
||||
for pr in prs or []:
|
||||
try:
|
||||
pr_number = int(pr["number"])
|
||||
except (KeyError, TypeError, ValueError):
|
||||
continue
|
||||
links = resolve_pr_links(pr)
|
||||
pr_links[pr_number] = links
|
||||
for link in links:
|
||||
if link.evidence == (EVIDENCE_REFERENCE,):
|
||||
continue
|
||||
bucket = issue_prs.setdefault(link.issue_number, [])
|
||||
if pr_number not in bucket:
|
||||
bucket.append(pr_number)
|
||||
return LinkageIndex(
|
||||
pr_links=pr_links,
|
||||
issue_prs={number: tuple(sorted(items)) for number, items in issue_prs.items()},
|
||||
)
|
||||
|
||||
|
||||
# --- Canonical handoff summary ----------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HandoffSummary:
|
||||
"""The latest CTH comment on one thread, redacted for display."""
|
||||
|
||||
comment_id: int | None
|
||||
created_at: str | None
|
||||
author: str | None
|
||||
cth_type: str
|
||||
cth_type_known: bool
|
||||
status: str | None
|
||||
next_owner: str | None
|
||||
current_blocker: str | None
|
||||
decision: str | None
|
||||
next_action: str | None
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"comment_id": self.comment_id,
|
||||
"created_at": self.created_at,
|
||||
"author": self.author,
|
||||
"cth_type": self.cth_type,
|
||||
"cth_type_known": self.cth_type_known,
|
||||
"status": self.status,
|
||||
"next_owner": self.next_owner,
|
||||
"current_blocker": self.current_blocker,
|
||||
"decision": self.decision,
|
||||
"next_action": self.next_action,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HandoffStatus:
|
||||
"""Why a thread's handoff summary is present, absent, or unknown."""
|
||||
|
||||
state: str
|
||||
reason: str | None = None
|
||||
target: str | None = None
|
||||
|
||||
@property
|
||||
def loaded(self) -> bool:
|
||||
return self.state == HANDOFF_LOADED
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {"state": self.state, "reason": self.reason, "target": self.target}
|
||||
|
||||
|
||||
def summarize_handoff(comments: Sequence[dict[str, Any]]) -> HandoffSummary | None:
|
||||
"""Summarize the newest CTH comment in *comments*, or None when there is none.
|
||||
|
||||
Every field is redacted before it is returned: a handoff body is operator
|
||||
free text that regularly quotes commands, and it is rendered verbatim on the
|
||||
page this feeds.
|
||||
"""
|
||||
from canonical_thread_handoff import find_latest_cth, is_known_cth_type
|
||||
|
||||
try:
|
||||
latest = find_latest_cth(list(comments or []))
|
||||
except Exception:
|
||||
return None
|
||||
if not latest:
|
||||
return None
|
||||
fields = latest.get("fields") or {}
|
||||
cth_type = str(latest.get("cth_type") or "").strip()
|
||||
known = is_known_cth_type(cth_type)
|
||||
try:
|
||||
comment_id: int | None = int(latest.get("comment_id"))
|
||||
except (TypeError, ValueError):
|
||||
comment_id = None
|
||||
return HandoffSummary(
|
||||
comment_id=comment_id,
|
||||
created_at=_redact(latest.get("created_at")),
|
||||
author=_redact(latest.get("author")),
|
||||
# An unrecognised heading is reported as such rather than republished:
|
||||
# the heading is free text, and CTH_TYPES is the only authority for what
|
||||
# a handoff type may be.
|
||||
cth_type=cth_type if known else "unrecognized",
|
||||
cth_type_known=known,
|
||||
status=_redact(fields.get("status")),
|
||||
next_owner=_redact(fields.get("next owner")),
|
||||
current_blocker=_redact(fields.get("current blocker")),
|
||||
decision=_redact(fields.get("decision")),
|
||||
next_action=_redact(fields.get("next action")),
|
||||
)
|
||||
|
||||
|
||||
CommentSource = Callable[[str, int], list[dict[str, Any]]]
|
||||
|
||||
|
||||
def build_comment_source(host: str, org: str, repo: str) -> CommentSource | None:
|
||||
"""Build an authenticated ``(kind, number) -> comments`` fetcher, or None.
|
||||
|
||||
Returns None when the console is running in offline test mode or when no
|
||||
credential is available for *host*, so the caller reports the handoff source
|
||||
as unavailable instead of as an empty thread.
|
||||
"""
|
||||
if _offline_test_mode() or not (host and org and repo):
|
||||
return None
|
||||
auth = get_auth_header(host)
|
||||
if not auth:
|
||||
return None
|
||||
|
||||
def _fetch(kind: str, number: int) -> list[dict[str, Any]]:
|
||||
segment = "pulls" if kind == "pr" else "issues"
|
||||
url = f"{repo_api_url(host, org, repo)}/{segment}/{int(number)}/comments"
|
||||
comments: list[dict[str, Any]] = []
|
||||
page = 1
|
||||
while page <= 20:
|
||||
raw, meta = api_fetch_page(url, auth, page=page, limit=50)
|
||||
comments.extend(raw)
|
||||
if bool(meta["is_final_page"]):
|
||||
break
|
||||
page += 1
|
||||
return comments
|
||||
|
||||
return _fetch
|
||||
|
||||
|
||||
# --- Snapshot ----------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LinkageNode:
|
||||
"""One issue or PR row with its resolved links and display metadata."""
|
||||
|
||||
kind: str
|
||||
number: int
|
||||
title: str
|
||||
state: str
|
||||
labels: tuple[str, ...] = ()
|
||||
links: tuple[IssueLink, ...] = ()
|
||||
linked_prs: tuple[int, ...] = ()
|
||||
ambiguous: bool = False
|
||||
contested: bool = False
|
||||
deep_link: str | None = None
|
||||
handoff: HandoffSummary | None = None
|
||||
handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
|
||||
links_authoritative: bool = True
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"kind": self.kind,
|
||||
"number": self.number,
|
||||
"title": self.title,
|
||||
"state": self.state,
|
||||
"labels": list(self.labels),
|
||||
"links": [link.to_dict() for link in self.links],
|
||||
"linked_prs": list(self.linked_prs),
|
||||
"ambiguous": self.ambiguous,
|
||||
"contested": self.contested,
|
||||
"deep_link": self.deep_link,
|
||||
"links_authoritative": self.links_authoritative,
|
||||
"handoff": self.handoff.to_dict() if self.handoff else None,
|
||||
"handoff_status": self.handoff_status.to_dict(),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LinkageSnapshot:
|
||||
"""One answered linkage query over a scoped window of a Gitea repo."""
|
||||
|
||||
ok: bool
|
||||
project_id: str
|
||||
repo_label: str
|
||||
host: str
|
||||
state_scope: str
|
||||
issues: tuple[LinkageNode, ...] = ()
|
||||
prs: tuple[LinkageNode, ...] = ()
|
||||
contested_issues: tuple[int, ...] = ()
|
||||
focus: tuple[str, int] | None = None
|
||||
inventory_complete: bool = False
|
||||
deep_links_enabled: bool = False
|
||||
handoff_status: HandoffStatus = HandoffStatus(HANDOFF_NOT_LOADED)
|
||||
fetch_error: str | None = None
|
||||
|
||||
@property
|
||||
def orphan_prs(self) -> tuple[LinkageNode, ...]:
|
||||
"""PRs in the loaded window that point at no issue at all."""
|
||||
return tuple(node for node in self.prs if not node.links)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"ok": self.ok,
|
||||
"schema_version": LINKAGE_SCHEMA_VERSION,
|
||||
"project_id": self.project_id,
|
||||
"repo": self.repo_label,
|
||||
"state_scope": self.state_scope,
|
||||
"inventory_complete": self.inventory_complete,
|
||||
"deep_links_enabled": self.deep_links_enabled,
|
||||
"focus": (
|
||||
None
|
||||
if self.focus is None
|
||||
else {"kind": self.focus[0], "number": self.focus[1]}
|
||||
),
|
||||
"handoff_source": self.handoff_status.to_dict(),
|
||||
"fetch_error": self.fetch_error,
|
||||
"contested_issues": list(self.contested_issues),
|
||||
"issues": [node.to_dict() for node in self.issues],
|
||||
"prs": [node.to_dict() for node in self.prs],
|
||||
"evidence_kinds": [
|
||||
{"name": name, "description": EVIDENCE_DESCRIPTIONS[name]}
|
||||
for name in EVIDENCE_ORDER
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def snapshot_to_dict(snapshot: LinkageSnapshot) -> dict[str, Any]:
|
||||
"""JSON-serializable export for ``/api/v1/gitea/linkage``."""
|
||||
return snapshot.to_dict()
|
||||
|
||||
|
||||
def _offline_test_mode() -> bool:
|
||||
return (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip().lower() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
}
|
||||
|
||||
|
||||
def _labels_of(item: dict[str, Any]) -> tuple[str, ...]:
|
||||
return tuple(
|
||||
str(_redact(label.get("name")))
|
||||
for label in (item.get("labels") or [])
|
||||
if label.get("name")
|
||||
)
|
||||
|
||||
|
||||
def _failed_snapshot(
|
||||
*,
|
||||
project_id: str,
|
||||
repo_label: str,
|
||||
host: str,
|
||||
state_scope: str,
|
||||
reason: str,
|
||||
) -> LinkageSnapshot:
|
||||
"""A read that could not be answered. Never an empty-and-healthy snapshot."""
|
||||
return LinkageSnapshot(
|
||||
ok=False,
|
||||
project_id=project_id,
|
||||
repo_label=repo_label,
|
||||
host=host,
|
||||
state_scope=state_scope,
|
||||
inventory_complete=False,
|
||||
deep_links_enabled=deep_links_enabled(),
|
||||
handoff_status=HandoffStatus(
|
||||
HANDOFF_UNAVAILABLE, reason="linkage inventory could not be loaded"
|
||||
),
|
||||
fetch_error=str(_redact(reason)),
|
||||
)
|
||||
|
||||
|
||||
def _resolve_project(project_id: str | None) -> ProjectRecord | None:
|
||||
registry = load_registry()
|
||||
if project_id:
|
||||
for entry in registry.projects:
|
||||
if entry.id == project_id:
|
||||
return entry
|
||||
return None
|
||||
return registry.projects[0] if registry.projects else None
|
||||
|
||||
|
||||
def _normalize_state(state: str | None) -> str:
|
||||
text = (state or STATE_OPEN).strip().lower()
|
||||
return text if text in _SUPPORTED_STATES else STATE_OPEN
|
||||
|
||||
|
||||
def load_linkage_snapshot(
|
||||
project_id: str | None = None,
|
||||
*,
|
||||
state: str | None = None,
|
||||
issue: int | None = None,
|
||||
pr: int | None = None,
|
||||
fetch_prs: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
|
||||
fetch_issues: Callable[..., tuple[list[dict], PaginationMeta]] | None = None,
|
||||
comment_source: CommentSource | None = None,
|
||||
) -> LinkageSnapshot:
|
||||
"""Load issue↔PR linkage for a registry project.
|
||||
|
||||
``issue``/``pr`` focus one thread: the focused row is the only one whose
|
||||
Canonical Thread Handoff comments are fetched, because handoff comments are
|
||||
thread-scoped and loading them for a whole queue would be one request per
|
||||
row. Every unfocused row reports its handoff as ``not_loaded``.
|
||||
"""
|
||||
state_scope = _normalize_state(state)
|
||||
try:
|
||||
project = _resolve_project(project_id)
|
||||
except Exception as exc: # registry invalid — fail closed with the reason
|
||||
return _failed_snapshot(
|
||||
project_id=project_id or "",
|
||||
repo_label="",
|
||||
host="",
|
||||
state_scope=state_scope,
|
||||
reason=f"project registry unavailable: {exc}",
|
||||
)
|
||||
|
||||
if project is None:
|
||||
return _failed_snapshot(
|
||||
project_id=project_id or "",
|
||||
repo_label="",
|
||||
host="",
|
||||
state_scope=state_scope,
|
||||
reason=(
|
||||
f"project {project_id!r} not found in registry"
|
||||
if project_id
|
||||
else "no projects registered"
|
||||
),
|
||||
)
|
||||
|
||||
host = _host_from_url(project.remote_host)
|
||||
repo_label = f"{project.gitea_owner}/{project.repo_name}"
|
||||
offline_test = _offline_test_mode()
|
||||
|
||||
def _empty_fetch(*_args, **_kwargs):
|
||||
return [], PaginationMeta(
|
||||
page=1,
|
||||
per_page=50,
|
||||
returned_count=0,
|
||||
has_more=False,
|
||||
is_final_page=True,
|
||||
# An offline stub loaded nothing; claiming a complete inventory here
|
||||
# would let the page assert that no issue has a linked PR.
|
||||
inventory_complete=False,
|
||||
pages_fetched=0,
|
||||
)
|
||||
|
||||
pr_fetch = fetch_prs or (_empty_fetch if offline_test else _fetch_prs)
|
||||
issue_fetch = fetch_issues or (_empty_fetch if offline_test else _fetch_issues)
|
||||
using_live_fetch = not offline_test and (fetch_prs is None or fetch_issues is None)
|
||||
auth = get_auth_header(host) if using_live_fetch else "test-auth"
|
||||
if using_live_fetch and not auth:
|
||||
return _failed_snapshot(
|
||||
project_id=project.id,
|
||||
repo_label=repo_label,
|
||||
host=host,
|
||||
state_scope=state_scope,
|
||||
reason=(
|
||||
f"Gitea credentials unavailable for {host}; linkage cannot be "
|
||||
"loaded (fail closed — not rendering an empty linkage table)"
|
||||
),
|
||||
)
|
||||
|
||||
try:
|
||||
raw_prs, pr_pagination = pr_fetch(
|
||||
host, project.gitea_owner, project.repo_name, auth, state=state_scope
|
||||
)
|
||||
raw_issues, issue_pagination = issue_fetch(
|
||||
host, project.gitea_owner, project.repo_name, auth, state=state_scope
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 — operator-visible fetch failure
|
||||
return _failed_snapshot(
|
||||
project_id=project.id,
|
||||
repo_label=repo_label,
|
||||
host=host,
|
||||
state_scope=state_scope,
|
||||
reason=f"Gitea fetch failed: {exc}",
|
||||
)
|
||||
|
||||
inventory_complete = bool(
|
||||
getattr(pr_pagination, "inventory_complete", False)
|
||||
and getattr(issue_pagination, "inventory_complete", False)
|
||||
)
|
||||
|
||||
index = resolve_linkage(raw_prs)
|
||||
contested = index.contested_issues()
|
||||
|
||||
focus: tuple[str, int] | None = None
|
||||
if pr is not None:
|
||||
focus = ("pr", int(pr))
|
||||
elif issue is not None:
|
||||
focus = ("issue", int(issue))
|
||||
|
||||
unfocused_reason = (
|
||||
"canonical handoff comments are thread-scoped; focus one issue or PR "
|
||||
"to load its latest handoff"
|
||||
)
|
||||
handoff_status = HandoffStatus(HANDOFF_NOT_LOADED, reason=unfocused_reason)
|
||||
focus_handoff: HandoffSummary | None = None
|
||||
if focus is not None:
|
||||
source = comment_source
|
||||
if source is None and not offline_test:
|
||||
source = build_comment_source(host, project.gitea_owner, project.repo_name)
|
||||
target = f"{focus[0]}#{focus[1]}"
|
||||
if source is None:
|
||||
handoff_status = HandoffStatus(
|
||||
HANDOFF_UNAVAILABLE,
|
||||
reason="no authenticated comment source available for this read",
|
||||
target=target,
|
||||
)
|
||||
else:
|
||||
try:
|
||||
focus_handoff = summarize_handoff(source(focus[0], focus[1]) or [])
|
||||
handoff_status = HandoffStatus(HANDOFF_LOADED, target=target)
|
||||
except Exception as exc: # fail soft: degrade this source only
|
||||
handoff_status = HandoffStatus(
|
||||
HANDOFF_UNAVAILABLE,
|
||||
reason=str(_redact(f"handoff fetch failed: {exc}")),
|
||||
target=target,
|
||||
)
|
||||
|
||||
def _node_handoff(
|
||||
kind: str, number: int
|
||||
) -> tuple[HandoffSummary | None, HandoffStatus]:
|
||||
"""Attach the handoff only to the focused row; qualify every other row."""
|
||||
if focus == (kind, number):
|
||||
return (focus_handoff, handoff_status)
|
||||
return (
|
||||
None,
|
||||
HandoffStatus(
|
||||
HANDOFF_NOT_LOADED,
|
||||
reason=unfocused_reason if focus is None else "not the focused thread",
|
||||
),
|
||||
)
|
||||
|
||||
def _number_of(raw: dict[str, Any]) -> int | None:
|
||||
try:
|
||||
return int(raw["number"])
|
||||
except (KeyError, TypeError, ValueError):
|
||||
return None
|
||||
|
||||
def _sort_key(raw: dict[str, Any]) -> int:
|
||||
number = _number_of(raw)
|
||||
return -1 if number is None else number
|
||||
|
||||
issue_nodes: list[LinkageNode] = []
|
||||
for raw in sorted(raw_issues or [], key=_sort_key, reverse=True):
|
||||
number = _number_of(raw)
|
||||
if number is None:
|
||||
continue
|
||||
node_handoff, node_status = _node_handoff("issue", number)
|
||||
linked_prs = index.issue_prs.get(number, ())
|
||||
issue_nodes.append(
|
||||
LinkageNode(
|
||||
kind="issue",
|
||||
number=number,
|
||||
title=str(_redact(raw.get("title")) or ""),
|
||||
state=str(raw.get("state") or ""),
|
||||
labels=_labels_of(raw),
|
||||
linked_prs=linked_prs,
|
||||
contested=len(linked_prs) > 1,
|
||||
deep_link=_deep_link(
|
||||
host, project.gitea_owner, project.repo_name, "issue", number
|
||||
),
|
||||
handoff=node_handoff,
|
||||
handoff_status=node_status,
|
||||
links_authoritative=inventory_complete,
|
||||
)
|
||||
)
|
||||
|
||||
pr_nodes: list[LinkageNode] = []
|
||||
for raw in sorted(raw_prs or [], key=_sort_key, reverse=True):
|
||||
number = _number_of(raw)
|
||||
if number is None:
|
||||
continue
|
||||
node_handoff, node_status = _node_handoff("pr", number)
|
||||
links = index.pr_links.get(number, ())
|
||||
pr_nodes.append(
|
||||
LinkageNode(
|
||||
kind="pr",
|
||||
number=number,
|
||||
title=str(_redact(raw.get("title")) or ""),
|
||||
state=str(raw.get("state") or ""),
|
||||
labels=_labels_of(raw),
|
||||
links=links,
|
||||
ambiguous=index.ambiguous(number),
|
||||
contested=any(link.issue_number in contested for link in links),
|
||||
deep_link=_deep_link(
|
||||
host, project.gitea_owner, project.repo_name, "pr", number
|
||||
),
|
||||
handoff=node_handoff,
|
||||
handoff_status=node_status,
|
||||
links_authoritative=inventory_complete,
|
||||
)
|
||||
)
|
||||
|
||||
return LinkageSnapshot(
|
||||
ok=True,
|
||||
project_id=project.id,
|
||||
repo_label=repo_label,
|
||||
host=host,
|
||||
state_scope=state_scope,
|
||||
issues=tuple(issue_nodes),
|
||||
prs=tuple(pr_nodes),
|
||||
contested_issues=contested,
|
||||
focus=focus,
|
||||
inventory_complete=inventory_complete,
|
||||
deep_links_enabled=deep_links_enabled(),
|
||||
handoff_status=handoff_status,
|
||||
)
|
||||
@@ -0,0 +1,364 @@
|
||||
"""HTML views for the Gitea issue↔PR linkage console (#645, Phase 3).
|
||||
|
||||
Read-only renderer over :mod:`webui.linkage_loader`. The page's job is to make
|
||||
three things impossible to misread:
|
||||
|
||||
* **why** an edge exists — every link carries its evidence badge, so a bare
|
||||
``#N`` mention never looks like a closing claim;
|
||||
* **what was not loaded** — a partial inventory, an unfocused thread, or an
|
||||
unavailable handoff source renders as an explicit qualifier, never as an
|
||||
affirmative "none";
|
||||
* **that nothing here mutates** — there is no review, merge, or edit control,
|
||||
and the deep link out to Gitea appears only under the admin reveal opt-in.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from html import escape
|
||||
from typing import Sequence
|
||||
|
||||
from webui.layout import render_page
|
||||
from webui.linkage_loader import (
|
||||
EVIDENCE_BRANCH,
|
||||
EVIDENCE_CLOSES,
|
||||
EVIDENCE_DESCRIPTIONS,
|
||||
EVIDENCE_ORDER,
|
||||
EVIDENCE_REFERENCE,
|
||||
HANDOFF_LOADED,
|
||||
HANDOFF_NOT_LOADED,
|
||||
HandoffSummary,
|
||||
LinkageNode,
|
||||
LinkageSnapshot,
|
||||
)
|
||||
|
||||
_EVIDENCE_CSS = {
|
||||
EVIDENCE_CLOSES: "badge-health-ok",
|
||||
EVIDENCE_BRANCH: "badge-health-skipped",
|
||||
EVIDENCE_REFERENCE: "badge-health-unproven",
|
||||
}
|
||||
|
||||
_EVIDENCE_LABEL = {
|
||||
EVIDENCE_CLOSES: "closes",
|
||||
EVIDENCE_BRANCH: "branch",
|
||||
EVIDENCE_REFERENCE: "mention",
|
||||
}
|
||||
|
||||
|
||||
def _badge(text: str, css: str) -> str:
|
||||
return f'<span class="badge {css}">{escape(text)}</span>'
|
||||
|
||||
|
||||
def _labels(names: Sequence[str]) -> str:
|
||||
if not names:
|
||||
return '<span class="muted">—</span>'
|
||||
return " ".join(_badge(name, "badge-health-skipped") for name in names)
|
||||
|
||||
|
||||
def _ref(node: LinkageNode) -> str:
|
||||
"""Render an item reference, hyperlinked only when deep links are revealed."""
|
||||
label = f"#{node.number}"
|
||||
if node.deep_link:
|
||||
return f'<a href="{escape(node.deep_link)}"><code>{escape(label)}</code></a>'
|
||||
return f"<code>{escape(label)}</code>"
|
||||
|
||||
|
||||
def _scope_card(snapshot: LinkageSnapshot) -> str:
|
||||
focus = (
|
||||
"none"
|
||||
if snapshot.focus is None
|
||||
else f"{snapshot.focus[0]}#{snapshot.focus[1]}"
|
||||
)
|
||||
completeness = (
|
||||
_badge("complete", "badge-health-ok")
|
||||
if snapshot.inventory_complete
|
||||
else _badge("partial", "badge-health-degraded")
|
||||
)
|
||||
links_note = (
|
||||
"Every linkage edge below is a claim about this loaded window only."
|
||||
if snapshot.inventory_complete
|
||||
else (
|
||||
"Pagination did not complete for this window, so an empty link list "
|
||||
"means <em>none found in what was loaded</em> — not that no link exists."
|
||||
)
|
||||
)
|
||||
deep_links = (
|
||||
_badge("enabled", "badge-health-ok")
|
||||
if snapshot.deep_links_enabled
|
||||
else _badge("hidden", "badge-health-skipped")
|
||||
)
|
||||
return f"""<div class="health-card">
|
||||
<h3>Scope</h3>
|
||||
<table class="detail">
|
||||
<tr><th>Project</th><td><code>{escape(snapshot.project_id or "—")}</code></td></tr>
|
||||
<tr><th>Repository</th><td><code>{escape(snapshot.repo_label or "—")}</code></td></tr>
|
||||
<tr><th>Item state</th><td><code>{escape(snapshot.state_scope)}</code></td></tr>
|
||||
<tr><th>Focused thread</th><td><code>{escape(focus)}</code></td></tr>
|
||||
<tr><th>Inventory</th><td>{completeness}</td></tr>
|
||||
<tr><th>Gitea deep links</th><td>{deep_links}</td></tr>
|
||||
</table>
|
||||
<p class="muted">{links_note}</p>
|
||||
</div>"""
|
||||
|
||||
|
||||
def _error_card(snapshot: LinkageSnapshot) -> str:
|
||||
if snapshot.ok and not snapshot.fetch_error:
|
||||
return ""
|
||||
return (
|
||||
'<div class="health-card health-stale"><strong>Linkage unavailable:</strong> '
|
||||
f"{escape(snapshot.fetch_error or 'the linkage read did not complete')}. "
|
||||
"No linkage table is rendered: an empty table would read as "
|
||||
"<em>no issue is linked to any PR</em>, which this read cannot claim."
|
||||
"</div>"
|
||||
)
|
||||
|
||||
|
||||
def _contested_card(snapshot: LinkageSnapshot) -> str:
|
||||
if not snapshot.contested_issues:
|
||||
return ""
|
||||
refs = ", ".join(f"<code>#{number}</code>" for number in snapshot.contested_issues)
|
||||
return (
|
||||
'<div class="health-card health-stale">'
|
||||
f"<strong>Contested issues:</strong> {refs}. More than one PR in this "
|
||||
"window claims each of these — duplicate work or a superseded PR. "
|
||||
"Resolution stays in Gitea and the workflow; this console only reports it."
|
||||
"</div>"
|
||||
)
|
||||
|
||||
|
||||
def _evidence_badges(evidence: Sequence[str]) -> str:
|
||||
return " ".join(
|
||||
_badge(
|
||||
_EVIDENCE_LABEL.get(name, name),
|
||||
_EVIDENCE_CSS.get(name, "badge-health-skipped"),
|
||||
)
|
||||
for name in EVIDENCE_ORDER
|
||||
if name in evidence
|
||||
)
|
||||
|
||||
|
||||
def _handoff_inline(handoff: HandoffSummary) -> str:
|
||||
type_css = "badge-health-ok" if handoff.cth_type_known else "badge-health-degraded"
|
||||
return (
|
||||
f'{_badge(handoff.cth_type or "—", type_css)}'
|
||||
f'<div class="muted" style="font-size:0.82rem;">'
|
||||
f'{escape(handoff.status or "—")} → {escape(handoff.next_owner or "—")}</div>'
|
||||
)
|
||||
|
||||
|
||||
def _handoff_cell(node: LinkageNode) -> str:
|
||||
"""Render the handoff column, distinguishing 'none found' from 'not loaded'."""
|
||||
status = node.handoff_status
|
||||
if status.state == HANDOFF_LOADED:
|
||||
if node.handoff is None:
|
||||
return '<span class="muted">no canonical handoff on this thread</span>'
|
||||
return _handoff_inline(node.handoff)
|
||||
if status.state == HANDOFF_NOT_LOADED:
|
||||
return (
|
||||
f'{_badge("not loaded", "badge-health-skipped")}'
|
||||
f'<div class="muted" style="font-size:0.82rem;">'
|
||||
f'{escape(status.reason or "")}</div>'
|
||||
)
|
||||
return (
|
||||
f'{_badge("unavailable", "badge-health-degraded")}'
|
||||
f'<div class="muted" style="font-size:0.82rem;">'
|
||||
f'{escape(status.reason or "")}</div>'
|
||||
)
|
||||
|
||||
|
||||
def _issue_rows(snapshot: LinkageSnapshot) -> str:
|
||||
rows = []
|
||||
for node in snapshot.issues:
|
||||
if node.linked_prs:
|
||||
linked = ", ".join(f"<code>#{number}</code>" for number in node.linked_prs)
|
||||
if node.contested:
|
||||
linked += " " + _badge("contested", "badge-blocked")
|
||||
elif node.links_authoritative:
|
||||
linked = '<span class="muted">none</span>'
|
||||
else:
|
||||
# The distinction an operator needs: nothing found in a window that
|
||||
# was not fully loaded is not the same as nothing existing.
|
||||
linked = '<span class="muted">none found (partial inventory)</span>'
|
||||
rows.append(
|
||||
"<tr>"
|
||||
f"<td>{_ref(node)}</td>"
|
||||
f"<td>{escape(node.title)}</td>"
|
||||
f"<td>{escape(node.state or '—')}</td>"
|
||||
f"<td>{_labels(node.labels)}</td>"
|
||||
f"<td>{linked}</td>"
|
||||
f"<td>{_handoff_cell(node)}</td>"
|
||||
"</tr>"
|
||||
)
|
||||
if not rows:
|
||||
return '<tr><td colspan="6" class="muted">No issues in the loaded window.</td></tr>'
|
||||
return "".join(rows)
|
||||
|
||||
|
||||
def _pr_rows(snapshot: LinkageSnapshot) -> str:
|
||||
rows = []
|
||||
for node in snapshot.prs:
|
||||
if node.links:
|
||||
linked = "".join(
|
||||
f"<div><code>#{link.issue_number}</code> "
|
||||
f"{_evidence_badges(link.evidence)}</div>"
|
||||
for link in node.links
|
||||
)
|
||||
if node.ambiguous:
|
||||
linked += _badge("ambiguous", "badge-blocked")
|
||||
if node.contested:
|
||||
linked += " " + _badge("contested", "badge-blocked")
|
||||
elif node.links_authoritative:
|
||||
linked = '<span class="muted">no issue reference</span>'
|
||||
else:
|
||||
linked = '<span class="muted">none found (partial inventory)</span>'
|
||||
rows.append(
|
||||
"<tr>"
|
||||
f"<td>{_ref(node)}</td>"
|
||||
f"<td>{escape(node.title)}</td>"
|
||||
f"<td>{escape(node.state or '—')}</td>"
|
||||
f"<td>{_labels(node.labels)}</td>"
|
||||
f"<td>{linked}</td>"
|
||||
f"<td>{_handoff_cell(node)}</td>"
|
||||
"</tr>"
|
||||
)
|
||||
if not rows:
|
||||
return (
|
||||
'<tr><td colspan="6" class="muted">No pull requests in the loaded '
|
||||
"window.</td></tr>"
|
||||
)
|
||||
return "".join(rows)
|
||||
|
||||
|
||||
def _focus_card(snapshot: LinkageSnapshot) -> str:
|
||||
"""Render the focused thread's latest canonical handoff, when one was loaded."""
|
||||
if snapshot.focus is None:
|
||||
return f"""<div class="prompt-card">
|
||||
<h3>Canonical handoff</h3>
|
||||
<p class="muted">{escape(snapshot.handoff_status.reason or "")}
|
||||
Add <code>?issue=N</code> or <code>?pr=N</code> to load the latest
|
||||
Canonical Thread Handoff for one thread.</p>
|
||||
</div>"""
|
||||
|
||||
kind, number = snapshot.focus
|
||||
target = f"{kind} #{number}"
|
||||
if not snapshot.handoff_status.loaded:
|
||||
return f"""<div class="prompt-card">
|
||||
<h3>Canonical handoff — {escape(target)}</h3>
|
||||
<p class="muted">{_badge("unavailable", "badge-health-degraded")}
|
||||
{escape(snapshot.handoff_status.reason or "handoff source did not run")}.
|
||||
This is not evidence that the thread carries no handoff.</p>
|
||||
</div>"""
|
||||
|
||||
handoff = next(
|
||||
(
|
||||
node.handoff
|
||||
for node in (snapshot.issues + snapshot.prs)
|
||||
if node.kind == kind and node.number == number and node.handoff
|
||||
),
|
||||
None,
|
||||
)
|
||||
if handoff is None:
|
||||
return f"""<div class="prompt-card">
|
||||
<h3>Canonical handoff — {escape(target)}</h3>
|
||||
<p class="muted">Comments loaded; no Canonical Thread Handoff comment found on
|
||||
this thread.</p>
|
||||
</div>"""
|
||||
|
||||
type_css = "badge-health-ok" if handoff.cth_type_known else "badge-health-degraded"
|
||||
unknown_note = (
|
||||
""
|
||||
if handoff.cth_type_known
|
||||
else (
|
||||
'<p class="muted">The comment\'s heading is not a declared CTH type, '
|
||||
"so it is reported as unrecognized rather than republished.</p>"
|
||||
)
|
||||
)
|
||||
return f"""<div class="prompt-card">
|
||||
<h3>Canonical handoff — {escape(target)}</h3>
|
||||
<p class="meta">{_badge(handoff.cth_type or "—", type_css)}
|
||||
by <code>{escape(handoff.author or "unknown")}</code>
|
||||
at <code>{escape(handoff.created_at or "unknown")}</code></p>
|
||||
{unknown_note}
|
||||
<table class="detail">
|
||||
<tr><th>Status</th><td>{escape(handoff.status or "—")}</td></tr>
|
||||
<tr><th>Next owner</th><td>{escape(handoff.next_owner or "—")}</td></tr>
|
||||
<tr><th>Current blocker</th><td>{escape(handoff.current_blocker or "—")}</td></tr>
|
||||
<tr><th>Decision</th><td>{escape(handoff.decision or "—")}</td></tr>
|
||||
<tr><th>Next action</th><td>{escape(handoff.next_action or "—")}</td></tr>
|
||||
</table>
|
||||
<p class="muted">Full event history:
|
||||
<a href="/api/v1/timeline?{escape(kind)}={number}"><code>/api/v1/timeline</code></a></p>
|
||||
</div>"""
|
||||
|
||||
|
||||
def _legend_card(snapshot: LinkageSnapshot) -> str:
|
||||
items = "".join(
|
||||
f"<li>{_badge(_EVIDENCE_LABEL[name], _EVIDENCE_CSS[name])} — "
|
||||
f"{escape(EVIDENCE_DESCRIPTIONS[name])}</li>"
|
||||
for name in EVIDENCE_ORDER
|
||||
)
|
||||
reveal_note = (
|
||||
"Gitea deep links are shown because the "
|
||||
"<code>GITEA_MCP_REVEAL_ENDPOINTS</code> admin opt-in is set."
|
||||
if snapshot.deep_links_enabled
|
||||
else (
|
||||
"Gitea deep links are withheld. Set "
|
||||
"<code>GITEA_MCP_REVEAL_ENDPOINTS=1</code> server-side to reveal "
|
||||
"them; item numbers stay usable without them."
|
||||
)
|
||||
)
|
||||
return f"""<div class="prompt-card">
|
||||
<h3>How an edge was found</h3>
|
||||
<ul class="reasons">{items}</ul>
|
||||
<p class="muted">{reveal_note}</p>
|
||||
<p class="muted">Read-only surface: no issue or PR editing, no review, and no merge.
|
||||
JSON export: <a href="/api/v1/gitea/linkage"><code>/api/v1/gitea/linkage</code></a></p>
|
||||
</div>"""
|
||||
|
||||
|
||||
def render_linkage_page(snapshot: LinkageSnapshot) -> str:
|
||||
"""Render the full HTML page for the Gitea linkage console."""
|
||||
if not snapshot.ok:
|
||||
return render_page(
|
||||
title="Gitea linkage",
|
||||
body_html=f"""<h2>Gitea issue and PR linkage</h2>
|
||||
<p class="meta">Phase 3 read-only linkage console (#645).</p>
|
||||
{_error_card(snapshot)}
|
||||
{_scope_card(snapshot)}""",
|
||||
)
|
||||
|
||||
body = f"""<h2>Gitea issue and PR linkage</h2>
|
||||
<p class="meta">Phase 3 read-only linkage console (#645). Gitea remains the source of
|
||||
truth; this page reads it and never writes to it.</p>
|
||||
{_scope_card(snapshot)}
|
||||
{_contested_card(snapshot)}
|
||||
|
||||
<div class="prompt-card">
|
||||
<h3>Issues → pull requests</h3>
|
||||
<table class="registry">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Issue</th><th>Title</th><th>State</th><th>Labels</th>
|
||||
<th>Linked PRs</th><th>Latest handoff</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>{_issue_rows(snapshot)}</tbody>
|
||||
</table>
|
||||
</div>
|
||||
|
||||
<div class="prompt-card">
|
||||
<h3>Pull requests → issues</h3>
|
||||
<table class="registry">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>PR</th><th>Title</th><th>State</th><th>Labels</th>
|
||||
<th>Linked issues</th><th>Latest handoff</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>{_pr_rows(snapshot)}</tbody>
|
||||
</table>
|
||||
</div>
|
||||
|
||||
{_focus_card(snapshot)}
|
||||
{_legend_card(snapshot)}
|
||||
"""
|
||||
return render_page(title="Gitea linkage", body_html=body)
|
||||
+5
-1
@@ -6,7 +6,8 @@ destination is a GET view or a Phase 1 placeholder. No mutation links.
|
||||
|
||||
Nav groups follow the #631 Phase 1 information architecture: Health, Traffic,
|
||||
Runtime/Sessions, Projects, Inventory, Timeline, Policy (placeholder), and
|
||||
Insights (placeholder). Later-phase surfaces are declared as ``stub`` items and
|
||||
Insights (placeholder), joined by the Phase 3 Gitea linkage group (#645).
|
||||
Later-phase surfaces are declared as ``stub`` items and
|
||||
backed by ``STUB_PAGES`` so their nav links resolve to a graceful placeholder
|
||||
instead of a 404.
|
||||
"""
|
||||
@@ -60,6 +61,9 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
|
||||
NavGroup("Timeline", (
|
||||
NavItem("/timeline", "Timeline", "stub"),
|
||||
)),
|
||||
NavGroup("Gitea", (
|
||||
NavItem("/gitea", "Issue/PR linkage"),
|
||||
)),
|
||||
NavGroup("Policy", (
|
||||
NavItem("/policy", "Policy", "stub"),
|
||||
NavItem("/prompts", "Prompts"),
|
||||
|
||||
+27
-2
@@ -211,6 +211,19 @@ def _pagination_from_pages(
|
||||
)
|
||||
|
||||
|
||||
_SUPPORTED_FETCH_STATES = ("open", "closed", "all")
|
||||
|
||||
|
||||
def _safe_state(state: str | None) -> str:
|
||||
"""Constrain a caller-supplied item state before it reaches a query string.
|
||||
|
||||
The value is interpolated into the Gitea URL, so an unrecognised state falls
|
||||
back to ``open`` rather than being passed through.
|
||||
"""
|
||||
text = (state or "open").strip().lower()
|
||||
return text if text in _SUPPORTED_FETCH_STATES else "open"
|
||||
|
||||
|
||||
def _fetch_prs(
|
||||
host: str,
|
||||
org: str,
|
||||
@@ -218,8 +231,15 @@ def _fetch_prs(
|
||||
auth: str,
|
||||
*,
|
||||
per_page: int = 50,
|
||||
state: str = "open",
|
||||
) -> tuple[list[dict], PaginationMeta]:
|
||||
url = f"{repo_api_url(host, org, repo)}/pulls?state=open"
|
||||
"""Fetch PRs in *state* (``open``, ``closed``, or ``all``).
|
||||
|
||||
The queue dashboard only ever wants the open window, so ``open`` stays the
|
||||
default. The linkage console (#645) widens it, because a landed issue↔PR
|
||||
edge lives on a merged PR.
|
||||
"""
|
||||
url = f"{repo_api_url(host, org, repo)}/pulls?state={_safe_state(state)}"
|
||||
all_raw: list[dict] = []
|
||||
pages_fetched = 0
|
||||
is_final = False
|
||||
@@ -248,8 +268,13 @@ def _fetch_issues(
|
||||
auth: str,
|
||||
*,
|
||||
per_page: int = 50,
|
||||
state: str = "open",
|
||||
) -> tuple[list[dict], PaginationMeta]:
|
||||
url = f"{repo_api_url(host, org, repo)}/issues?state=open&type=issues"
|
||||
"""Fetch issues in *state* (``open``, ``closed``, or ``all``); see :func:`_fetch_prs`."""
|
||||
url = (
|
||||
f"{repo_api_url(host, org, repo)}/issues"
|
||||
f"?state={_safe_state(state)}&type=issues"
|
||||
)
|
||||
all_raw: list[dict] = []
|
||||
page = 1
|
||||
pages_fetched = 0
|
||||
|
||||
Reference in New Issue
Block a user