gabriel / musehub public
coord.py python
261 lines 10.0 KB
Raw
sha256:20b27796ed512236119feef48b4577a8fd9ded05a15267dd4e6da23604ad1f1a docs: update Suggested Ownership Split table to reflect ful… Sonnet 5 27 days ago
1 """MuseHub coordination bus API routes.
2
3 Three endpoints, all scoped under ``/{owner}/{slug}/coord/``:
4
5 ``POST /{owner}/{slug}/coord/push``
6 Agents push local coordination records to the hub. Auth required.
7 Rate-limited at 60/minute per IP (coordination traffic is lighter than
8 wire protocol pushes but should still be bounded).
9
10 ``POST /{owner}/{slug}/coord/pull``
11 Agents pull coordination records since a cursor ID. Auth required for
12 private repos; public repos allow unauthenticated pulls.
13 Rate-limited at 300/minute per IP.
14
15 ``GET /{owner}/{slug}/coord/watch``
16 SSE stream of new coordination records as they arrive. Long-lived
17 connection — clients reconnect with ``?since_id=<cursor>`` on disconnect.
18 Auth follows the same public/private visibility rules as pull.
19 Rate-limited at 30/minute per IP (each connect is cheap but connections
20 are long-lived; the limit prevents rapid reconnect loops).
21
22 Security
23 --------
24 - ``owner`` and ``slug`` are resolved to a repo row (404 for unknown repos,
25 404 for private repos the caller cannot read).
26 - ``record_id`` values are validated by the Pydantic model before reaching
27 the service layer.
28 - ``kind`` values are constrained to the known set by the Pydantic model.
29 - Auth tokens are validated by ``require_valid_token`` / ``optional_token``
30 from the shared auth dependency module.
31 - ANSI injection via ``run_id`` or ``kind`` in log lines is prevented by the
32 standard logging sanitization in the transport layer.
33
34 Exit codes (HTTP status)
35 ------------------------
36 200 OK — push/pull succeeded.
37 400 Validation error (bad kind, bad record ID, record count out of range).
38 401 Unauthenticated — push always requires auth; pull/watch require auth
39 for private repos.
40 404 Repo not found or private repo invisible to caller.
41 429 Rate limit exceeded.
42 500 Server error.
43 """
44
45 import logging
46 from typing import Annotated
47
48 from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
49 from fastapi.responses import StreamingResponse
50 from sqlalchemy.ext.asyncio import AsyncSession
51
52 from musehub.api.validation import SlugParam
53 from musehub.auth.dependencies import optional_token, require_valid_token, TokenClaims
54 from musehub.db.musehub_repo_models import MusehubRepo
55 from musehub.db.database import get_db
56 from musehub.models.coord import (
57 CoordPollRequest,
58 CoordPullResponse,
59 CoordPushRequest,
60 CoordPushResponse,
61 _VALID_KINDS,
62 )
63 from musehub.mcp.sse import SSE_CONTENT_TYPE
64 from musehub.rate_limits import limiter
65 from musehub.services.musehub_coord import coord_pull, coord_push, coord_watch_stream
66 from musehub.services.musehub_repository import get_repo_row_by_owner_slug
67
68 logger = logging.getLogger(__name__)
69
70 router = APIRouter(tags=["Coordination Bus"])
71
72 # ── Rate limits ────────────────────────────────────────────────────────────────
73 _COORD_PUSH_LIMIT = "60/minute"
74 _COORD_PULL_LIMIT = "300/minute"
75 _COORD_WATCH_LIMIT = "30/minute"
76
77 # ── Helpers ────────────────────────────────────────────────────────────────────
78
79 async def _resolve_repo(
80 session: AsyncSession,
81 owner: SlugParam,
82 slug: SlugParam,
83 ) -> MusehubRepo:
84 """Resolve owner/slug → repo row or raise 404."""
85 repo = await get_repo_row_by_owner_slug(session, owner, slug)
86 if repo is None:
87 raise HTTPException(
88 status_code=status.HTTP_404_NOT_FOUND,
89 detail=f"repo '{owner}/{slug}' not found",
90 )
91 return repo
92
93 def _assert_readable(repo: MusehubRepo, claims: TokenClaims | None) -> None:
94 """Raise 404 if the repo is private and the caller is not the owner."""
95 if repo.visibility == "public":
96 return
97 caller_id: str | None = claims.identity_id if claims else None
98 if caller_id != repo.owner_user_id:
99 raise HTTPException(
100 status_code=status.HTTP_404_NOT_FOUND,
101 detail="repo not found",
102 )
103
104 def _assert_writable(repo: MusehubRepo, claims: TokenClaims) -> None:
105 """Raise 403 if the authenticated caller is not the repo owner.
106
107 Push requires ownership. Collaborator write access can be added here
108 when the collaborators table is extended with coord permissions.
109 """
110 caller_id: str | None = claims.identity_id
111 if caller_id != repo.owner_user_id:
112 raise HTTPException(
113 status_code=status.HTTP_403_FORBIDDEN,
114 detail="only the repo owner may push coordination records",
115 )
116
117 # ── Push ───────────────────────────────────────────────────────────────────────
118
119 @router.post(
120 "/{owner}/{slug}/coord/push",
121 response_model=CoordPushResponse,
122 status_code=status.HTTP_200_OK,
123 summary="Push coordination records to the hub.",
124 )
125 @limiter.limit(_COORD_PUSH_LIMIT)
126 async def push_coord(
127 request: Request,
128 owner: SlugParam,
129 slug: SlugParam,
130 body: CoordPushRequest,
131 claims: Annotated[TokenClaims, Depends(require_valid_token)],
132 session: Annotated[AsyncSession, Depends(get_db)],
133 ) -> CoordPushResponse:
134 """Push a batch of coordination records to MuseHub.
135
136 Idempotent: re-pushing the same ``record_id`` for the same kind is
137 silently skipped (counted as ``skipped`` in the response). ``heartbeat``
138 records are upserted instead.
139
140 Args:
141 request: FastAPI request (used by the rate limiter).
142 owner: Repo owner username.
143 slug: Repo URL slug.
144 body: 1–500 coordination records.
145 claims: Authenticated caller's MSign context.
146 session: Database session.
147
148 Returns:
149 :class:`~musehub.models.coord.CoordPushResponse`.
150 """
151 repo = await _resolve_repo(session, owner, slug)
152 _assert_writable(repo, claims)
153 return await coord_push(session, repo.repo_id, body)
154
155 # ── Pull ───────────────────────────────────────────────────────────────────────
156
157 @router.post(
158 "/{owner}/{slug}/coord/pull",
159 response_model=CoordPullResponse,
160 status_code=status.HTTP_200_OK,
161 summary="Pull coordination records from the hub.",
162 )
163 @limiter.limit(_COORD_PULL_LIMIT)
164 async def pull_coord(
165 request: Request,
166 owner: SlugParam,
167 slug: SlugParam,
168 body: CoordPollRequest,
169 claims: Annotated[TokenClaims | None, Depends(optional_token)],
170 session: Annotated[AsyncSession, Depends(get_db)],
171 ) -> CoordPullResponse:
172 """Pull coordination records since a cursor ID.
173
174 Public repos allow unauthenticated pulls. Private repos return 404 for
175 unauthenticated callers (same visibility semantics as all other hub reads).
176
177 Args:
178 request: FastAPI request (used by the rate limiter).
179 owner: Repo owner username.
180 slug: Repo URL slug.
181 body: Pull parameters (``since_id``, optional ``kinds`` filter, ``limit``).
182 claims: Optional MSign context (required for private repos).
183 session: Database session.
184
185 Returns:
186 :class:`~musehub.models.coord.CoordPullResponse`.
187 """
188 repo = await _resolve_repo(session, owner, slug)
189 _assert_readable(repo, claims)
190 return await coord_pull(session, repo.repo_id, body)
191
192 # ── Watch (SSE) ────────────────────────────────────────────────────────────────
193
194 @router.get(
195 "/{owner}/{slug}/coord/watch",
196 summary="Watch coordination records as an SSE stream.",
197 )
198 @limiter.limit(_COORD_WATCH_LIMIT)
199 async def watch_coord(
200 request: Request,
201 owner: SlugParam,
202 slug: SlugParam,
203 claims: Annotated[TokenClaims | None, Depends(optional_token)],
204 session: Annotated[AsyncSession, Depends(get_db)],
205 since_id: int = Query(default=0, ge=0, description="Return records with id > since_id."),
206 kinds: list[str] = Query(
207 default=[],
208 description=f"Filter by kind. Valid values: {sorted(_VALID_KINDS)}.",
209 ),
210 ) -> StreamingResponse:
211 """Stream new coordination records as Server-Sent Events.
212
213 The client receives one ``event: coord_record`` SSE event per new record.
214 The event ``id`` field is set to the record's auto-increment ``id`` — clients
215 should pass it back as ``?since_id=<id>`` on reconnection to resume without
216 missing events.
217
218 A ``": heartbeat"`` comment is injected every 15 seconds of silence to keep
219 HTTP proxies alive.
220
221 Args:
222 request: FastAPI request (used by rate limiter).
223 owner: Repo owner username.
224 slug: Repo URL slug.
225 claims: Optional MSign context.
226 session: Database session (used only for repo resolution).
227 since_id: Only stream records with ``id > since_id``.
228 kinds: Filter by kind.
229
230 Returns:
231 :class:`~fastapi.responses.StreamingResponse` with ``text/event-stream`` media type.
232 """
233 # Validate kinds query param values.
234 from musehub.models.coord import _VALID_KINDS as VALID
235 for k in kinds:
236 if k not in VALID:
237 raise HTTPException(
238 status_code=status.HTTP_400_BAD_REQUEST,
239 detail=f"invalid kind {k!r}; must be one of {sorted(VALID)}",
240 )
241
242 repo = await _resolve_repo(session, owner, slug)
243 _assert_readable(repo, claims)
244
245 async def _stream() -> "AsyncIterator[str]":
246 async for chunk in coord_watch_stream(
247 repo_id=repo.repo_id,
248 since_id=since_id,
249 kinds=list(kinds),
250 get_session=get_db,
251 ):
252 yield chunk
253
254 return StreamingResponse(
255 _stream(),
256 media_type=SSE_CONTENT_TYPE,
257 headers={
258 "Cache-Control": "no-cache",
259 "X-Accel-Buffering": "no",
260 },
261 )
File History 3 commits
sha256:20b27796ed512236119feef48b4577a8fd9ded05a15267dd4e6da23604ad1f1a docs: update Suggested Ownership Split table to reflect ful… Sonnet 5 27 days ago
sha256:8ea2f75226d5834770ccaddbaa6efda1cc337446fd481bb385de8bec4555ae79 docs: Section 9 CI pipeline — confirm job queue is idempote… Sonnet 5 27 days ago
sha256:80e1a60a39562f6e616aaafbb27487f03abbec7d8ffc6164302b7a0c0bfc63ee docs: check off Section 0 items verified in inventory doc, … Sonnet 5 27 days ago