merge(master): resolve PR #905 conflicts with requests/linkage

Keep notifications (#648) routes and nav alongside master requests (#643)
and other base updates.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

webui/app.py

@@ -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,
@@ -77,6 +82,8 @@ from webui.notifications import (
     snapshot_to_dict as notifications_snapshot_to_dict,
 )
 from webui.notification_views import render_notifications_page
+from webui import request_service
+from webui.request_views import render_requests_page
 
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -347,6 +354,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
@@ -760,6 +802,108 @@ async def api_notifications(request: Request) -> JSONResponse:
     data = notifications_snapshot_to_dict(snap)
     return JSONResponse(data)
 
+def _default_request_scope() -> dict[str, str]:
+    """Resolve remote/org/repo from the project registry for request forms.
+
+    Returns an empty mapping when the registry cannot be read, which makes
+    ``parse_request`` reject a request that did not name its own scope rather
+    than letting it default to some other repository.
+    """
+    from webui.queue_loader import _host_from_url  # host normalisation helper
+
+    registry, error = _load_project_registry()
+    if error is not None or not registry.projects:
+        return {}
+    project = registry.projects[0]
+    host = _host_from_url(project.remote_host)
+    return {
+        "remote": _derive_remote(host),
+        "org": project.gitea_owner or "",
+        "repo": project.repo_name or "",
+    }
+
+
+async def _request_payload(request: Request) -> dict[str, object]:
+    """Read a request body as JSON or form-encoded. Never raises."""
+    content_type = (request.headers.get("content-type") or "").lower()
+    if "application/json" in content_type:
+        try:
+            body = await request.json()
+        except Exception:
+            return {}
+        return dict(body) if isinstance(body, dict) else {}
+    try:
+        form = await request.form()
+    except Exception:
+        return {}
+    return {key: form[key] for key in form}
+
+
+async def requests_page(request: Request) -> HTMLResponse:
+    """Operator request form and intent preview (#643).
+
+    POST here only ever *previews*. Initiation is a separate confirmed call to
+    ``/api/v1/requests/apply`` so that submitting this form cannot reserve
+    work as a side effect.
+    """
+    submitted: dict[str, object] = {}
+    preview = None
+    error = None
+    if request.method == "POST":
+        submitted = await _request_payload(request)
+        work_request, error = request_service.parse_request(
+            submitted, default_scope=_default_request_scope()
+        )
+        if work_request is not None:
+            preview = request_service.preview_request(
+                work_request,
+                principal=resolve_principal(headers=dict(request.headers)),
+            )
+    return HTMLResponse(
+        render_requests_page(
+            preview=preview, error=error, submitted=submitted
+        )
+    )
+
+
+async def api_v1_request_preview(request: Request) -> JSONResponse:
+    """Dry-run authorization and intent preview for a work request (#643)."""
+    payload = await _request_payload(request)
+    work_request, error = request_service.parse_request(
+        payload, default_scope=_default_request_scope()
+    )
+    if work_request is None:
+        return JSONResponse(error.to_dict(), status_code=400)
+    preview = request_service.preview_request(
+        work_request,
+        principal=resolve_principal(headers=dict(request.headers)),
+    )
+    return JSONResponse(
+        preview.to_dict(), status_code=200 if preview.authorized else 403
+    )
+
+
+async def api_v1_request_apply(request: Request) -> JSONResponse:
+    """Initiate a previewed work request through the allocator (#643).
+
+    Fail-closed at every step: unauthorized, unconfirmed, not-next-safe, and
+    already-claimed all return without attempting an assignment.
+    """
+    payload = await _request_payload(request)
+    work_request, error = request_service.parse_request(
+        payload, default_scope=_default_request_scope()
+    )
+    if work_request is None:
+        return JSONResponse(error.to_dict(), status_code=400)
+    confirm = _truthy_flag(str(payload.get("confirm") or ""))
+    result = request_service.apply_request(
+        work_request,
+        principal=resolve_principal(headers=dict(request.headers)),
+        confirm=confirm,
+    )
+    status = int(result.pop("status_code", 403))
+    return JSONResponse(result, status_code=status)
+
 
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
     path = request.url.path
@@ -810,6 +954,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"]),
@@ -831,6 +977,17 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 api_action_attempt,
                 methods=["POST"],
             ),
+            Route("/requests", requests_page, methods=["GET", "POST"]),
+            Route(
+                "/api/v1/requests/preview",
+                api_v1_request_preview,
+                methods=["POST"],
+            ),
+            Route(
+                "/api/v1/requests/apply",
+                api_v1_request_apply,
+                methods=["POST"],
+            ),
             Route("/api/leases", api_leases, methods=["GET"]),
             Route("/api/v1/inventory", api_inventory, methods=["GET"]),
             Route(