aaronrene / overseer-kit public
handlers.py python
210 lines 7.8 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 22 hours ago
1 """Closed GET-only API handlers (§HGD.5)."""
2
3 from __future__ import annotations
4
5 from typing import Any
6 from urllib.parse import parse_qs
7
8 from tools.hosted_dashboard.adapters.github import GitHubAdapters
9 from tools.hosted_dashboard.config import HostedDashboardConfig
10 from tools.hosted_dashboard.discovery import build_org_summary
11 from tools.hosted_dashboard.envelope import (
12 ApiEnvelope,
13 build_meta,
14 failure,
15 health_success,
16 success,
17 )
18 from tools.hosted_dashboard.http_client import UpstreamError
19 from tools.hosted_dashboard.parsers import parse_document_derived_gates
20 from tools.hosted_dashboard.validators import unknown_query_keys, valid_owner_repo_segment
21
22 API_GET_ROUTES = frozenset(
23 {
24 "/api/health",
25 "/api/org/summary",
26 # repo routes matched separately
27 }
28 )
29
30 TRACK_Q_ACT_PREFIXES = (
31 "/api/review/",
32 "/api/governance-sync",
33 "/api/ledger/",
34 "/api/honesty-status",
35 "/api/init",
36 "/api/sync",
37 )
38
39
40 def is_track_q_act_path(path: str) -> bool:
41 return any(path == p or path.startswith(p) for p in TRACK_Q_ACT_PREFIXES)
42
43
44 class DashboardService:
45 """Business logic for closed read surface."""
46
47 def __init__(self, adapters: GitHubAdapters, config: HostedDashboardConfig) -> None:
48 self._adapters = adapters
49 self._config = config
50
51 def health(self) -> ApiEnvelope:
52 return health_success()
53
54 def org_summary(self, query: dict[str, list[str]]) -> ApiEnvelope:
55 unknown = unknown_query_keys(query)
56 if unknown:
57 return failure(error="unknown_query", http_status=400)
58 repos = build_org_summary(self._adapters, self._config)
59 result = {
60 "repos": [
61 {
62 "owner": r.owner,
63 "name": r.name,
64 "full_name": r.full_name,
65 "default_branch": r.default_branch,
66 "eligibility": r.eligibility,
67 "marker_present": r.marker_present,
68 }
69 for r in repos
70 ]
71 }
72 return success(
73 result,
74 meta=build_meta(source_id="github_meta", ref="n/a"),
75 )
76
77 def roadmap(self, owner: str, repo: str, query: dict[str, list[str]]) -> ApiEnvelope:
78 return self._doc_endpoint(owner, repo, query, kind="roadmap")
79
80 def handover(self, owner: str, repo: str, query: dict[str, list[str]]) -> ApiEnvelope:
81 return self._doc_endpoint(owner, repo, query, kind="handover")
82
83 def gates(self, owner: str, repo: str, query: dict[str, list[str]]) -> ApiEnvelope:
84 bad = self._validate_owner_repo_query(owner, repo, query)
85 if bad is not None:
86 return bad
87 try:
88 meta = self._adapters.get_repo_meta(owner, repo)
89 marker = self._adapters.fetch_marker_summary(owner, repo, ref=meta.default_branch)
90 roadmap_path, handover_path = self._adapters.doc_paths_from_marker(marker)
91 roadmap_text = None
92 handover_text = None
93 try:
94 roadmap_text = self._adapters.fetch_file(
95 owner, repo, roadmap_path, ref=meta.default_branch
96 ).text
97 except UpstreamError:
98 pass
99 try:
100 handover_text = self._adapters.fetch_file(
101 owner, repo, handover_path, ref=meta.default_branch
102 ).text
103 except UpstreamError:
104 pass
105 derived = parse_document_derived_gates(
106 roadmap_text=roadmap_text,
107 handover_text=handover_text,
108 )
109 advisory = self._adapters.advisory_checks(owner, repo, ref=meta.default_branch)
110 result = {
111 "document_derived": {
112 "ok": derived.ok,
113 "error": derived.error,
114 "phases": derived.phases,
115 "pending_gates_excerpt": derived.pending_gates_excerpt,
116 },
117 "advisory_checks": advisory,
118 }
119 return success(
120 result,
121 meta=build_meta(source_id="github_contents", ref=meta.default_branch),
122 )
123 except UpstreamError as exc:
124 return self._upstream_failure(exc)
125
126 def config_marker(self, owner: str, repo: str, query: dict[str, list[str]]) -> ApiEnvelope:
127 bad = self._validate_owner_repo_query(owner, repo, query)
128 if bad is not None:
129 return bad
130 try:
131 meta = self._adapters.get_repo_meta(owner, repo)
132 marker = self._adapters.fetch_marker_summary(owner, repo, ref=meta.default_branch)
133 result = {
134 "present": marker.present,
135 "roadmap_path": marker.roadmap_path,
136 "handover_path": marker.handover_path,
137 "vcs_regime": marker.vcs_regime,
138 }
139 # Explicitly ensure raw_text is never included.
140 assert "raw_text" not in result
141 return success(
142 result,
143 meta=build_meta(source_id="github_contents", ref=meta.default_branch),
144 )
145 except UpstreamError as exc:
146 return self._upstream_failure(exc)
147
148 def _doc_endpoint(
149 self, owner: str, repo: str, query: dict[str, list[str]], *, kind: str
150 ) -> ApiEnvelope:
151 bad = self._validate_owner_repo_query(owner, repo, query)
152 if bad is not None:
153 return bad
154 try:
155 meta = self._adapters.get_repo_meta(owner, repo)
156 marker = self._adapters.fetch_marker_summary(owner, repo, ref=meta.default_branch)
157 roadmap_path, handover_path = self._adapters.doc_paths_from_marker(marker)
158 path = roadmap_path if kind == "roadmap" else handover_path
159 file = self._adapters.fetch_file(owner, repo, path, ref=meta.default_branch)
160 result = {
161 "path": file.path,
162 "text": file.text,
163 "sha256": file.sha256,
164 }
165 return success(
166 result,
167 meta=build_meta(
168 source_id=file.source_id,
169 ref=file.ref,
170 content_sha256=file.sha256,
171 ),
172 )
173 except UpstreamError as exc:
174 return self._upstream_failure(exc)
175
176 def _validate_owner_repo_query(
177 self, owner: str, repo: str, query: dict[str, list[str]]
178 ) -> ApiEnvelope | None:
179 if not valid_owner_repo_segment(owner) or not valid_owner_repo_segment(repo):
180 return failure(error="invalid_path", http_status=400)
181 unknown = unknown_query_keys(query)
182 if unknown:
183 return failure(error="unknown_query", http_status=400)
184 return None
185
186 @staticmethod
187 def _upstream_failure(exc: UpstreamError) -> ApiEnvelope:
188 if exc.token == "not_found":
189 return failure(error="not_found", http_status=404)
190 if exc.token == "upstream_unauthorized":
191 return failure(error="upstream_unauthorized", http_status=502)
192 if exc.token == "upstream_rate_limited":
193 return failure(error="upstream_rate_limited", http_status=502)
194 if exc.token == "upstream_host_refused":
195 return failure(error="upstream_host_refused", http_status=403)
196 return failure(error=exc.token, http_status=502)
197
198
199 def match_repo_route(path: str) -> tuple[str, str, str] | None:
200 """Match ``/api/repos/{owner}/{repo}/{action}`` → (owner, repo, action)."""
201 parts = path.strip("/").split("/")
202 # api, repos, owner, repo, action
203 if len(parts) != 5:
204 return None
205 if parts[0] != "api" or parts[1] != "repos":
206 return None
207 action = parts[4]
208 if action not in {"roadmap", "handover", "gates", "config-marker"}:
209 return None
210 return parts[2], parts[3], action
File History 1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 22 hours ago