gabriel / musehub public
musehub_issues.py python
1,166 lines 39.8 KB
Raw
sha256:4d42a346263e7cbbd152c147f3e6f24576f4b4440df9249ffb9fbcf9db699fcb feat: populate url in create_issue and create_proposal responses Sonnet 4.6 minor ⚠ breaking 35 days ago
1 """MuseHub issue persistence adapter — single point of DB access for issues.
2
3 This module is the ONLY place that touches the ``musehub_issues``,
4 ``musehub_issue_comments``, and ``musehub_issue_events`` tables.
5 Route handlers delegate here; no business logic lives in routes.
6
7 Boundary rules:
8 - Must NOT import state stores, SSE queues, or LLM clients.
9 - May import ORM models from musehub.db domain-specific modules.
10 - May import Pydantic response models from musehub.models.musehub.
11 - May import musehub.core.genesis for genesis ID computation.
12 """
13
14 import logging
15 from datetime import datetime, timezone
16
17 from sqlalchemy import any_, func, literal, select, text, union_all
18 from sqlalchemy.ext.asyncio import AsyncSession
19
20 from typing import TypedDict
21
22 from musehub.core.genesis import compute_comment_id, compute_issue_event_id, compute_issue_id
23 from musehub.db.musehub_release_models import MusehubRelease
24 from musehub.db.musehub_repo_models import MusehubCommit, MusehubCommitRef
25 from musehub.db.musehub_social_models import MusehubIssue, MusehubIssueComment, MusehubIssueEvent, MusehubProposal
26 from musehub.types.json_types import JSONObject
27
28
29 class _ProposalMatch(TypedDict):
30 proposal_id: str
31 proposal_number: int
32 title: str
33 state: str
34 from_branch: str
35 to_branch: str
36 match_reason: str
37
38
39 type _SeenMap = dict[str, _ProposalMatch]
40 type _CommitMap = dict[str, MusehubCommit]
41
42
43 class IssueReleaseContext(TypedDict):
44 no_commits: bool
45 anchor_commits: list[JSONObject]
46 all_landed: bool
47 any_landed: bool
48 pending_count: int
49 next_tag: str | None
50 latest_release_tag: str | None
51 from musehub.models.musehub import (
52 IssueCommentListResponse,
53 IssueCommentResponse,
54 IssueListResponse,
55 IssueResponse,
56 )
57
58 logger = logging.getLogger(__name__)
59
60
61 def _to_issue_response(row: MusehubIssue, comment_count: int = 0, *, url_prefix: str = "") -> IssueResponse:
62 """Convert a DB row to an IssueResponse wire model.
63
64 ``comment_count`` must be passed in by the caller — it requires a separate
65 aggregate query so we avoid N+1 loads on list endpoints.
66
67 ``url_prefix`` should be ``"{base_url}/{owner}/{repo_slug}"`` — the canonical
68 URL for the issue is then ``{url_prefix}/issues/{number}``.
69 """
70 return IssueResponse(
71 issue_id=row.issue_id,
72 number=row.number,
73 url=f"{url_prefix}/issues/{row.number}" if url_prefix else "",
74 title=row.title,
75 body=row.body,
76 state=row.state,
77 labels=list(row.labels or []),
78 symbol_anchors=list(row.symbol_anchors or []),
79 commit_anchors=list(row.commit_anchors or []),
80 author=row.author,
81 assignee=row.assignee,
82 agent_id=row.agent_id or "",
83 model_id=row.model_id or "",
84 created_at=row.created_at,
85 updated_at=row.updated_at,
86 comment_count=comment_count,
87 )
88
89
90 def _to_comment_response(row: MusehubIssueComment) -> IssueCommentResponse:
91 """Convert a DB comment row to the wire representation."""
92 return IssueCommentResponse(
93 comment_id=row.comment_id,
94 issue_id=row.issue_id,
95 author=row.author,
96 body=row.body,
97 parent_id=row.parent_id,
98 is_deleted=row.is_deleted,
99 created_at=row.created_at,
100 updated_at=row.updated_at,
101 )
102
103
104 async def _next_issue_number(session: AsyncSession, repo_id: str) -> int:
105 """Return the next sequential issue number for the given repo (1-based)."""
106 stmt = select(func.max(MusehubIssue.number)).where(
107 MusehubIssue.repo_id == repo_id
108 )
109 current_max: int | None = (await session.execute(stmt)).scalar_one_or_none()
110 return (current_max or 0) + 1
111
112
113 async def _count_comments(session: AsyncSession, issue_id: str) -> int:
114 """Return the non-deleted comment count for a single issue."""
115 stmt = select(func.count(MusehubIssueComment.comment_id)).where(
116 MusehubIssueComment.issue_id == issue_id,
117 MusehubIssueComment.is_deleted.is_(False),
118 )
119 count: int = (await session.execute(stmt)).scalar_one()
120 return count
121
122
123 async def create_issue(
124 session: AsyncSession,
125 *,
126 repo_id: str,
127 title: str,
128 body: str,
129 labels: list[str],
130 author: str = "",
131 author_identity_id: str = "",
132 symbol_anchors: list[str] | None = None,
133 commit_anchors: list[str] | None = None,
134 agent_id: str = "",
135 model_id: str = "",
136 url_prefix: str = "",
137 ) -> IssueResponse:
138 """Persist a new issue in ``open`` state and return its wire representation.
139
140 ``author`` is the MSign handle for display; ``author_identity_id`` is the
141 genesis-addressed identity ID used to compute the canonical ``issue_id``.
142 Automatically emits an ``opened`` event for the activity timeline.
143 """
144 number = await _next_issue_number(session, repo_id)
145 created_at = datetime.now(timezone.utc)
146 issue_id = compute_issue_id(repo_id, author_identity_id, created_at.isoformat())
147 issue = MusehubIssue(
148 issue_id=issue_id,
149 repo_id=repo_id,
150 number=number,
151 title=title,
152 body=body,
153 state="open",
154 labels=labels,
155 symbol_anchors=symbol_anchors or [],
156 commit_anchors=commit_anchors or [],
157 author=author,
158 agent_id=agent_id,
159 model_id=model_id,
160 created_at=created_at,
161 )
162 session.add(issue)
163 await session.flush()
164 await session.refresh(issue)
165 # Emit the "opened" event for the activity timeline.
166 session.add(MusehubIssueEvent(
167 event_id=compute_issue_event_id(issue.issue_id, "opened", author, issue.created_at.isoformat()),
168 issue_id=issue.issue_id,
169 repo_id=repo_id,
170 actor=author,
171 event_type="opened",
172 payload={},
173 created_at=issue.created_at,
174 ))
175 await session.flush()
176 logger.info("✅ Created issue #%d for repo %s: %s", number, repo_id, title)
177 return _to_issue_response(issue, url_prefix=url_prefix)
178
179
180 async def list_issues(
181 session: AsyncSession,
182 repo_id: str,
183 *,
184 state: str = "open",
185 label: str | None = None,
186 cursor: str | None = None,
187 limit: int = 20,
188 url_prefix: str = "",
189 ) -> IssueListResponse:
190 """Return issues for a repo with cursor-based keyset pagination.
191
192 ``state`` may be ``"open"``, ``"closed"``, or ``"all"``.
193 ``label`` filters to issues whose labels list contains the given string.
194 Results are ordered by issue number ascending.
195
196 ``cursor`` is the last seen issue number (opaque to callers — pass
197 ``nextCursor`` from a previous response verbatim). Omit to start from
198 the beginning. ``total`` reflects all matching issues regardless of
199 the current page so UIs can show "showing N of M" without paginating
200 through everything.
201 """
202 conditions = [MusehubIssue.repo_id == repo_id]
203 if state != "all":
204 conditions.append(MusehubIssue.state == state)
205 if label is not None:
206 conditions.append(literal(label) == any_(MusehubIssue.labels))
207
208 count_stmt = select(func.count(MusehubIssue.issue_id)).where(*conditions)
209 total: int = (await session.execute(count_stmt)).scalar_one()
210
211 data_conditions = list(conditions)
212 if cursor is not None:
213 data_conditions.append(MusehubIssue.number > int(cursor))
214
215 rows = list(
216 (
217 await session.execute(
218 select(MusehubIssue)
219 .where(*data_conditions)
220 .order_by(MusehubIssue.number)
221 .limit(limit + 1)
222 )
223 ).scalars()
224 )
225
226 next_cursor: str | None = None
227 if len(rows) == limit + 1:
228 next_cursor = str(rows[limit - 1].number)
229 rows = rows[:limit]
230
231 results: list[IssueResponse] = []
232 for r in rows:
233 comment_count = await _count_comments(session, r.issue_id)
234 results.append(_to_issue_response(r, comment_count, url_prefix=url_prefix))
235
236 return IssueListResponse(issues=results, total=total, next_cursor=next_cursor)
237
238
239 async def get_issue(
240 session: AsyncSession,
241 repo_id: str,
242 issue_number: int,
243 *,
244 url_prefix: str = "",
245 ) -> IssueResponse | None:
246 """Return a single issue by its per-repo number, or None if not found."""
247 stmt = (
248 select(MusehubIssue)
249 .where(
250 MusehubIssue.repo_id == repo_id,
251 MusehubIssue.number == issue_number,
252 )
253 )
254 row = (await session.execute(stmt)).scalar_one_or_none()
255 if row is None:
256 return None
257 count = await _count_comments(session, row.issue_id)
258 return _to_issue_response(row, count, url_prefix=url_prefix)
259
260
261 async def close_issue(
262 session: AsyncSession,
263 repo_id: str,
264 issue_number: int,
265 actor: str = "",
266 ) -> IssueResponse | None:
267 """Set the issue state to ``closed``. Returns None if the issue does not exist.
268
269 Idempotent: if the issue is already closed no event is emitted and no
270 webhook is fired — the current state is returned unchanged. This prevents
271 duplicate timeline entries and spurious notifications when the caller
272 retries or the issue was already closed by another actor.
273 """
274 stmt = (
275 select(MusehubIssue)
276 .where(
277 MusehubIssue.repo_id == repo_id,
278 MusehubIssue.number == issue_number,
279 )
280 )
281 row = (await session.execute(stmt)).scalar_one_or_none()
282 if row is None:
283 return None
284 count = await _count_comments(session, row.issue_id)
285 if row.state == "closed":
286 return _to_issue_response(row, count)
287 row.state = "closed"
288 await session.flush()
289 await session.refresh(row)
290 _closed_at = row.updated_at or _utc_now()
291 session.add(MusehubIssueEvent(
292 event_id=compute_issue_event_id(row.issue_id, "closed", actor, _closed_at.isoformat()),
293 issue_id=row.issue_id,
294 repo_id=repo_id,
295 actor=actor,
296 event_type="closed",
297 payload={},
298 created_at=_closed_at,
299 ))
300 await session.flush()
301 logger.info("✅ Closed issue #%d for repo %s", issue_number, repo_id)
302 return _to_issue_response(row, count)
303
304
305 async def reopen_issue(
306 session: AsyncSession,
307 repo_id: str,
308 issue_number: int,
309 actor: str = "",
310 ) -> IssueResponse | None:
311 """Set the issue state back to ``open``. Returns None if the issue does not exist.
312
313 Idempotent: if the issue is already open no event is emitted and no
314 webhook is fired — the current state is returned unchanged. This prevents
315 duplicate timeline entries and spurious notifications when the caller
316 retries or the issue was already reopened by another actor.
317 """
318 stmt = (
319 select(MusehubIssue)
320 .where(
321 MusehubIssue.repo_id == repo_id,
322 MusehubIssue.number == issue_number,
323 )
324 )
325 row = (await session.execute(stmt)).scalar_one_or_none()
326 if row is None:
327 return None
328 count = await _count_comments(session, row.issue_id)
329 if row.state == "open":
330 return _to_issue_response(row, count)
331 row.state = "open"
332 await session.flush()
333 await session.refresh(row)
334 _reopened_at = row.updated_at or _utc_now()
335 session.add(MusehubIssueEvent(
336 event_id=compute_issue_event_id(row.issue_id, "reopened", actor, _reopened_at.isoformat()),
337 issue_id=row.issue_id,
338 repo_id=repo_id,
339 actor=actor,
340 event_type="reopened",
341 payload={},
342 created_at=_reopened_at,
343 ))
344 await session.flush()
345 logger.info("✅ Reopened issue #%d for repo %s", issue_number, repo_id)
346 return _to_issue_response(row, count)
347
348
349 async def update_issue(
350 session: AsyncSession,
351 repo_id: str,
352 issue_number: int,
353 *,
354 title: str | None = None,
355 body: str | None = None,
356 labels: list[str] | None = None,
357 symbol_anchors: list[str] | None = None,
358 commit_anchors: list[str] | None = None,
359 agent_id: str | None = None,
360 model_id: str | None = None,
361 ) -> IssueResponse | None:
362 """Partially update an issue's fields.
363
364 Only non-None arguments are applied. Returns None if the issue is not found.
365 """
366 stmt = (
367 select(MusehubIssue)
368 .where(
369 MusehubIssue.repo_id == repo_id,
370 MusehubIssue.number == issue_number,
371 )
372 )
373 row = (await session.execute(stmt)).scalar_one_or_none()
374 if row is None:
375 return None
376 if title is not None:
377 row.title = title
378 if body is not None:
379 row.body = body
380 if labels is not None:
381 row.labels = labels
382 if symbol_anchors is not None:
383 row.symbol_anchors = symbol_anchors
384 if commit_anchors is not None:
385 row.commit_anchors = commit_anchors
386 if agent_id is not None:
387 row.agent_id = agent_id
388 if model_id is not None:
389 row.model_id = model_id
390 await session.flush()
391 await session.refresh(row)
392 count = await _count_comments(session, row.issue_id)
393 return _to_issue_response(row, count)
394
395
396 async def assign_issue(
397 session: AsyncSession,
398 repo_id: str,
399 issue_number: int,
400 *,
401 assignee: str | None,
402 ) -> IssueResponse | None:
403 """Set or clear the assignee on an issue.
404
405 Pass ``assignee=None`` to unassign. Returns None if the issue is not found.
406 """
407 stmt = (
408 select(MusehubIssue)
409 .where(
410 MusehubIssue.repo_id == repo_id,
411 MusehubIssue.number == issue_number,
412 )
413 )
414 row = (await session.execute(stmt)).scalar_one_or_none()
415 if row is None:
416 return None
417 row.assignee = assignee
418 await session.flush()
419 await session.refresh(row)
420 logger.info(
421 "✅ %s issue #%d for repo %s",
422 f"Assigned {assignee} to" if assignee else "Unassigned",
423 issue_number,
424 repo_id,
425 )
426 count = await _count_comments(session, row.issue_id)
427 return _to_issue_response(row, count)
428
429
430 async def assign_labels(
431 session: AsyncSession,
432 repo_id: str,
433 issue_number: int,
434 *,
435 labels: list[str],
436 ) -> IssueResponse | None:
437 """Replace the label list on an issue with the provided labels.
438
439 Returns None if the issue is not found.
440 The replacement is total — callers must merge old and new labels themselves
441 when they only want to append.
442 """
443 stmt = (
444 select(MusehubIssue)
445 .where(
446 MusehubIssue.repo_id == repo_id,
447 MusehubIssue.number == issue_number,
448 )
449 )
450 row = (await session.execute(stmt)).scalar_one_or_none()
451 if row is None:
452 return None
453 row.labels = labels
454 await session.flush()
455 await session.refresh(row)
456 logger.info("✅ Assigned labels %r to issue #%d for repo %s", labels, issue_number, repo_id)
457 count = await _count_comments(session, row.issue_id)
458 return _to_issue_response(row, count)
459
460
461 async def remove_label(
462 session: AsyncSession,
463 repo_id: str,
464 issue_number: int,
465 *,
466 label: str,
467 ) -> IssueResponse | None:
468 """Remove a single label from an issue's label list.
469
470 Silently no-ops when the label is not present (idempotent).
471 Returns None if the issue is not found.
472 """
473 stmt = (
474 select(MusehubIssue)
475 .where(
476 MusehubIssue.repo_id == repo_id,
477 MusehubIssue.number == issue_number,
478 )
479 )
480 row = (await session.execute(stmt)).scalar_one_or_none()
481 if row is None:
482 return None
483 current: list[str] = list(row.labels or [])
484 row.labels = [lbl for lbl in current if lbl != label]
485 await session.flush()
486 await session.refresh(row)
487 logger.info("✅ Removed label %r from issue #%d for repo %s", label, issue_number, repo_id)
488 count = await _count_comments(session, row.issue_id)
489 return _to_issue_response(row, count)
490
491
492 # ── Issue comment operations ───────────────────────────────────────────────────
493
494
495 async def create_comment(
496 session: AsyncSession,
497 *,
498 issue_id: str,
499 repo_id: str,
500 body: str,
501 author: str,
502 author_identity_id: str = "",
503 parent_id: str | None = None,
504 ) -> IssueCommentResponse:
505 """Create a new comment on an issue."""
506 if parent_id is not None:
507 parent_stmt = select(MusehubIssueComment).where(
508 MusehubIssueComment.comment_id == parent_id,
509 MusehubIssueComment.issue_id == issue_id,
510 )
511 parent_row = (await session.execute(parent_stmt)).scalar_one_or_none()
512 if parent_row is None:
513 raise ValueError(f"Parent comment {parent_id!r} not found on issue {issue_id!r}")
514
515 created_at = datetime.now(timezone.utc)
516 comment_id = compute_comment_id(issue_id, author_identity_id, created_at.isoformat())
517 comment = MusehubIssueComment(
518 comment_id=comment_id,
519 issue_id=issue_id,
520 repo_id=repo_id,
521 author=author,
522 body=body,
523 parent_id=parent_id,
524 created_at=created_at,
525 )
526 session.add(comment)
527 await session.flush()
528 await session.refresh(comment)
529 logger.info("✅ Created comment %s on issue %s by %s", comment.comment_id, issue_id, author)
530 return _to_comment_response(comment)
531
532
533 async def list_comments(
534 session: AsyncSession,
535 issue_id: str,
536 *,
537 include_deleted: bool = False,
538 cursor: str | None = None,
539 limit: int = 20,
540 ) -> IssueCommentListResponse:
541 """Return comments on an issue with cursor-based keyset pagination.
542
543 Comments are returned in chronological order (oldest first).
544 Deleted comments are omitted by default; pass ``include_deleted=True``
545 to include them (e.g. for moderation views).
546
547 ``cursor`` is the ISO 8601 ``created_at`` of the last seen comment
548 (opaque to callers — pass ``nextCursor`` from a previous response
549 verbatim). Omit to start from the beginning.
550 """
551 conditions = [MusehubIssueComment.issue_id == issue_id]
552 if not include_deleted:
553 conditions.append(MusehubIssueComment.is_deleted.is_(False))
554
555 count_stmt = select(func.count(MusehubIssueComment.comment_id)).where(*conditions)
556 total: int = (await session.execute(count_stmt)).scalar_one()
557
558 data_conditions = list(conditions)
559 if cursor is not None:
560 data_conditions.append(
561 MusehubIssueComment.created_at > datetime.fromisoformat(cursor)
562 )
563
564 rows = list(
565 (
566 await session.execute(
567 select(MusehubIssueComment)
568 .where(*data_conditions)
569 .order_by(MusehubIssueComment.created_at)
570 .limit(limit + 1)
571 )
572 ).scalars()
573 )
574
575 next_cursor: str | None = None
576 if len(rows) == limit + 1:
577 next_cursor = rows[limit - 1].created_at.isoformat()
578 rows = rows[:limit]
579
580 comments = [_to_comment_response(r) for r in rows]
581 return IssueCommentListResponse(comments=comments, total=total, next_cursor=next_cursor)
582
583
584 async def get_timeline(
585 session: AsyncSession,
586 issue_id: str,
587 ) -> list[JSONObject]:
588 """Return a unified chronological activity timeline for an issue.
589
590 Unions ``musehub_issue_events`` (typed events: opened, closed, labeled, …)
591 with ``musehub_issue_comments`` (backwards-compatible comment rows), sorted
592 by ``created_at`` ascending. Each entry carries a ``kind`` field so the
593 template can render the appropriate icon and layout.
594
595 ``kind`` values: ``"event"`` | ``"comment"``
596 """
597 events_stmt = (
598 select(MusehubIssueEvent)
599 .where(MusehubIssueEvent.issue_id == issue_id)
600 .order_by(MusehubIssueEvent.created_at)
601 )
602 comments_stmt = (
603 select(MusehubIssueComment)
604 .where(
605 MusehubIssueComment.issue_id == issue_id,
606 MusehubIssueComment.is_deleted.is_(False),
607 )
608 .order_by(MusehubIssueComment.created_at)
609 )
610
611 events = (await session.execute(events_stmt)).scalars().all()
612 comments = (await session.execute(comments_stmt)).scalars().all()
613
614 timeline: list[JSONObject] = []
615 for ev in events:
616 timeline.append({
617 "kind": "event",
618 "event_id": ev.event_id,
619 "event_type": ev.event_type,
620 "actor": ev.actor,
621 "payload": ev.payload or {},
622 "created_at": ev.created_at.isoformat(),
623 })
624 for c in comments:
625 timeline.append({
626 "kind": "comment",
627 "comment_id": c.comment_id,
628 "author": c.author,
629 "body": c.body,
630 "parent_id": c.parent_id,
631 "created_at": c.created_at.isoformat(),
632 })
633
634 timeline.sort(key=lambda x: str(x.get("created_at") or ""))
635 return timeline
636
637
638 async def delete_comment(
639 session: AsyncSession,
640 comment_id: str,
641 issue_id: str,
642 ) -> bool:
643 """Soft-delete a comment. Returns True if the comment existed and was deleted."""
644 stmt = select(MusehubIssueComment).where(
645 MusehubIssueComment.comment_id == comment_id,
646 MusehubIssueComment.issue_id == issue_id,
647 )
648 row = (await session.execute(stmt)).scalar_one_or_none()
649 if row is None:
650 return False
651 row.is_deleted = True
652 await session.flush()
653 logger.info("✅ Soft-deleted comment %s", comment_id)
654 return True
655
656
657
658
659 # ── Signal 1: commit graph containment — Muse-native proposal linking ─────────
660
661
662 async def find_proposals_by_commit_graph(
663 session: AsyncSession,
664 repo_id: str,
665 commit_anchors: list[str],
666 max_depth: int = 20,
667 ) -> list[_ProposalMatch]:
668 """Find proposals linked to an issue via the VCS commit graph.
669
670 Signal 1: commit graph containment.
671
672 **Merged proposals**: BFS ancestor walk from ``merge_commit_id`` through
673 ``parent_ids`` (JSON array on each commit row). If any ancestor commit ID
674 matches a commit anchor on the issue, the proposal is linked.
675
676 **Open proposals**: Any commit stored under ``from_branch`` whose ID matches
677 a commit anchor indicates the fix is in flight and not yet merged.
678
679 Short-form anchors (fewer than 64 hex chars) are resolved via prefix match
680 so that ``8b624581`` matches ``8b624581c3f2...`` in the DB.
681
682 Returns a list of proposal dicts, each containing:
683 proposal_id, proposal_number, title, state,
684 from_branch, to_branch, match_reason="commit_graph"
685 """
686 if not commit_anchors:
687 return []
688
689 # ── Step 1: resolve short anchors to full commit IDs ──────────────────────
690 where_clauses: list[str] = []
691 resolve_params: JSONObject = {"repo_id": repo_id}
692 for i, anchor in enumerate(commit_anchors):
693 key = f"anc_{i}"
694 if len(anchor) < 64:
695 where_clauses.append(f"c.commit_id LIKE :{key} || '%'")
696 else:
697 where_clauses.append(f"c.commit_id = :{key}")
698 resolve_params[key] = anchor
699
700 resolve_sql = text(
701 f"SELECT c.commit_id FROM musehub_commits c"
702 f" JOIN musehub_commit_refs r ON r.commit_id = c.commit_id"
703 f" WHERE r.repo_id = :repo_id AND ({' OR '.join(where_clauses)})"
704 )
705 resolved_rows = (await session.execute(resolve_sql, resolve_params)).fetchall()
706 if not resolved_rows:
707 return []
708 resolved_ids: list[str] = [row[0] for row in resolved_rows]
709
710 seen: _SeenMap = {}
711
712 # ── Step 2: merged proposals — recursive ancestor walk ────────────────────
713 #
714 # The CTE starts from the parents of each merge_commit_id and walks up the
715 # ancestor graph up to max_depth hops, tracking which proposal each walk
716 # originated from. ``COALESCE(..., ARRAY[]::text[])`` guards against NULL
717 # when the merge commit is not stored in musehub_commits yet.
718 merged_sql = text("""
719 WITH RECURSIVE ancestors(proposal_id, commit_id, depth) AS (
720 SELECT
721 p.proposal_id,
722 elem.v AS commit_id,
723 1 AS depth
724 FROM musehub_proposals p
725 CROSS JOIN LATERAL unnest(
726 COALESCE(
727 (SELECT mc.parent_ids
728 FROM musehub_commits mc
729 WHERE mc.commit_id = p.merge_commit_id
730 LIMIT 1),
731 ARRAY[]::text[]
732 )
733 ) AS elem(v)
734 WHERE p.repo_id = :repo_id
735 AND p.state = 'merged'
736 AND p.merge_commit_id IS NOT NULL
737
738 UNION ALL
739
740 SELECT
741 a.proposal_id,
742 elem.v AS commit_id,
743 a.depth + 1
744 FROM ancestors a
745 JOIN musehub_commits mc
746 ON mc.commit_id = a.commit_id
747 CROSS JOIN LATERAL unnest(mc.parent_ids) AS elem(v)
748 WHERE a.depth < :max_depth
749 )
750 SELECT DISTINCT
751 p.proposal_id,
752 p.proposal_number,
753 p.title,
754 p.state,
755 p.from_branch,
756 p.to_branch
757 FROM ancestors anc
758 JOIN musehub_proposals p ON p.proposal_id = anc.proposal_id
759 WHERE anc.commit_id = ANY(:resolved_ids)
760 LIMIT 10
761 """)
762 merged_rows = (await session.execute(merged_sql, {
763 "repo_id": repo_id,
764 "max_depth": max_depth,
765 "resolved_ids": resolved_ids,
766 })).fetchall()
767 for row in merged_rows:
768 pid = str(row[0])
769 seen[pid] = {
770 "proposal_id": pid,
771 "proposal_number": row[1],
772 "title": row[2],
773 "state": row[3],
774 "from_branch": row[4],
775 "to_branch": row[5],
776 "match_reason": "commit_graph",
777 }
778
779 # ── Step 3: open proposals — branch membership check ──────────────────────
780 #
781 # If a commit anchor lives on the proposal's from_branch, the fix is in
782 # flight and the proposal is linked even though it hasn't merged yet.
783 open_sql = text("""
784 SELECT DISTINCT
785 p.proposal_id,
786 p.proposal_number,
787 p.title,
788 p.state,
789 p.from_branch,
790 p.to_branch
791 FROM musehub_proposals p
792 JOIN musehub_commit_refs cr
793 ON cr.repo_id = :repo_id
794 AND cr.commit_id = ANY(:resolved_ids)
795 JOIN musehub_commits mc
796 ON mc.commit_id = cr.commit_id
797 AND mc.branch = p.from_branch
798 WHERE p.repo_id = :repo_id
799 AND p.state = 'open'
800 LIMIT 10
801 """)
802 open_rows = (await session.execute(open_sql, {
803 "repo_id": repo_id,
804 "resolved_ids": resolved_ids,
805 })).fetchall()
806 for row in open_rows:
807 pid = str(row[0])
808 if pid not in seen:
809 seen[pid] = {
810 "proposal_id": pid,
811 "proposal_number": row[1],
812 "title": row[2],
813 "state": row[3],
814 "from_branch": row[4],
815 "to_branch": row[5],
816 "match_reason": "commit_graph",
817 }
818
819 return list(seen.values())[:10]
820
821
822 # ── Signal 3: branch-to-issue commit reachability ─────────────────────────────
823
824
825 async def find_proposals_by_branch_reachability(
826 session: AsyncSession,
827 repo_id: str,
828 commit_anchors: list[str],
829 max_depth: int = 20,
830 ) -> list[_ProposalMatch]:
831 """Find open proposals whose from_branch contains anchor commits not yet in to_branch.
832
833 Signal 3: branch-to-issue commit reachability.
834
835 A proposal's ``from_branch`` and ``to_branch`` define a merge path. We walk
836 the commit graph reachable from the ``from_branch`` HEAD and subtract all
837 commits reachable from the ``to_branch`` HEAD. The remaining "exclusive" set
838 represents work in flight — not yet integrated. If any of the issue's commit
839 anchors appears in that exclusive set, the proposal is actively working toward
840 resolving this issue.
841
842 Only open proposals are checked: merged proposals have their ``from_branch``
843 deleted at merge time, so the branch HEAD pointer no longer exists.
844
845 Short-form anchors (< 64 chars) are resolved to full commit IDs via prefix
846 match before the graph walk.
847
848 Returns a list of proposal dicts with ``match_reason="branch_reachability"``.
849 """
850 if not commit_anchors:
851 return []
852
853 # ── Step 1: resolve short anchors to full commit IDs ──────────────────────
854 where_clauses: list[str] = []
855 resolve_params: JSONObject = {"repo_id": repo_id}
856 for i, anchor in enumerate(commit_anchors):
857 key = f"anc_{i}"
858 if len(anchor) < 64:
859 where_clauses.append(f"c.commit_id LIKE :{key} || '%'")
860 else:
861 where_clauses.append(f"c.commit_id = :{key}")
862 resolve_params[key] = anchor
863
864 resolve_sql = text(
865 f"SELECT c.commit_id FROM musehub_commits c"
866 f" JOIN musehub_commit_refs r ON r.commit_id = c.commit_id"
867 f" WHERE r.repo_id = :repo_id AND ({' OR '.join(where_clauses)})"
868 )
869 resolved_rows = (await session.execute(resolve_sql, resolve_params)).fetchall()
870 if not resolved_rows:
871 return []
872 resolved_ids: list[str] = [row[0] for row in resolved_rows]
873
874 # ── Step 2: batched graph walk for all open proposals ─────────────────────
875 #
876 # Two parallel recursive CTEs share the same WITH RECURSIVE block:
877 #
878 # from_walk — ancestors reachable from each proposal's from_branch HEAD
879 # to_walk — ancestors reachable from each proposal's to_branch HEAD
880 #
881 # The final SELECT finds proposals where an anchor appears in from_walk but
882 # NOT in to_walk (i.e. the commit is on the feature branch but hasn't been
883 # integrated into the target branch yet).
884 reachability_sql = text("""
885 WITH RECURSIVE
886 from_walk(proposal_id, commit_id, depth) AS (
887 SELECT p.proposal_id, CAST(b.head_commit_id AS TEXT), 0
888 FROM musehub_proposals p
889 JOIN musehub_branches b
890 ON b.repo_id = p.repo_id
891 AND b.name = p.from_branch
892 WHERE p.repo_id = :repo_id
893 AND p.state = 'open'
894 AND b.head_commit_id IS NOT NULL
895
896 UNION ALL
897
898 SELECT fw.proposal_id, elem.v, fw.depth + 1
899 FROM from_walk fw
900 JOIN musehub_commits mc
901 ON mc.commit_id = fw.commit_id
902 CROSS JOIN LATERAL unnest(mc.parent_ids) AS elem(v)
903 WHERE fw.depth < :max_depth
904 ),
905 to_walk(proposal_id, commit_id, depth) AS (
906 SELECT p.proposal_id, CAST(b.head_commit_id AS TEXT), 0
907 FROM musehub_proposals p
908 JOIN musehub_branches b
909 ON b.repo_id = p.repo_id
910 AND b.name = p.to_branch
911 WHERE p.repo_id = :repo_id
912 AND p.state = 'open'
913 AND b.head_commit_id IS NOT NULL
914
915 UNION ALL
916
917 SELECT tw.proposal_id, elem.v, tw.depth + 1
918 FROM to_walk tw
919 JOIN musehub_commits mc
920 ON mc.commit_id = tw.commit_id
921 CROSS JOIN LATERAL unnest(mc.parent_ids) AS elem(v)
922 WHERE tw.depth < :max_depth
923 )
924 SELECT DISTINCT
925 p.proposal_id,
926 p.proposal_number,
927 p.title,
928 p.state,
929 p.from_branch,
930 p.to_branch
931 FROM from_walk fw
932 JOIN musehub_proposals p ON p.proposal_id = fw.proposal_id
933 WHERE fw.commit_id = ANY(:resolved_ids)
934 AND NOT EXISTS (
935 SELECT 1 FROM to_walk tw
936 WHERE tw.proposal_id = fw.proposal_id
937 AND tw.commit_id = fw.commit_id
938 )
939 LIMIT 10
940 """)
941 rows = (await session.execute(reachability_sql, {
942 "repo_id": repo_id,
943 "max_depth": max_depth,
944 "resolved_ids": resolved_ids,
945 })).fetchall()
946
947 return [
948 {
949 "proposal_id": str(row[0]),
950 "proposal_number": row[1],
951 "title": row[2],
952 "state": row[3],
953 "from_branch": row[4],
954 "to_branch": row[5],
955 "match_reason": "branch_reachability",
956 }
957 for row in rows
958 ]
959
960
961 # ── Signal 2: symbol anchor overlap ───────────────────────────────────────────
962
963
964 async def find_proposals_by_symbol_overlap(
965 session: AsyncSession,
966 repo_id: str,
967 symbol_anchors: list[str],
968 ) -> list[_ProposalMatch]:
969 """Find proposals whose ``touched_symbols`` intersect the issue's ``symbol_anchors``.
970
971 Signal 2: symbol anchor overlap.
972
973 When a proposal is created or merged, ``touched_symbols`` is populated from
974 the ``structured_delta`` column of every commit on ``from_branch``. A proposal is
975 linked to an issue when at least one of the issue's ``symbol_anchors`` appears
976 in the proposal's ``touched_symbols`` — meaning that proposal's commits
977 literally touched the exact symbol the issue is about.
978
979 Returns a list of proposal dicts, each containing:
980 proposal_id, proposal_number, title, state,
981 from_branch, to_branch, match_reason="symbol_overlap"
982 """
983 if not symbol_anchors:
984 return []
985
986 # Fetch all proposals for this repo that have a non-empty touched_symbols.
987 # We filter in Python because JSON array containment queries are dialect-
988 # specific and the anchor list is small enough that Python intersection is fast.
989 stmt = (
990 select(MusehubProposal)
991 .where(
992 MusehubProposal.repo_id == repo_id,
993 )
994 .order_by(MusehubProposal.created_at.desc())
995 .limit(200) # cap: repos with thousands of proposals are an edge case
996 )
997 rows = (await session.execute(stmt)).scalars().all()
998
999 anchor_set = set(symbol_anchors)
1000 results: list[_ProposalMatch] = []
1001 for row in rows:
1002 touched: list[str] = list(row.touched_symbols or [])
1003 if anchor_set.intersection(touched):
1004 results.append({
1005 "proposal_id": row.proposal_id,
1006 "proposal_number": row.proposal_number,
1007 "title": row.title,
1008 "state": row.state,
1009 "from_branch": row.from_branch,
1010 "to_branch": row.to_branch,
1011 "match_reason": "symbol_overlap",
1012 })
1013 if len(results) >= 10:
1014 break
1015
1016 return results
1017
1018
1019 # ── Release context (Muse-native issue → release tracking) ────────────────────
1020
1021
1022 async def get_issue_release_context(
1023 session: AsyncSession,
1024 repo_id: str,
1025 commit_anchors: list[str],
1026 ) -> IssueReleaseContext:
1027 """Derive Muse-native release context for an issue from the VCS graph.
1028
1029 Returns rich per-commit data so the UI can show real commit hashes,
1030 messages, authors, and which release each commit landed in — not a
1031 manually-entered field, but a live read of the VCS graph.
1032
1033 A commit anchor is "in" a release when the anchor commit's timestamp is at
1034 or before the release's pinned commit timestamp (linear history ordering).
1035
1036 Returns a dict with keys:
1037 no_commits – True when the issue has no commit_anchors at all
1038 anchor_commits – list of {short, full, message, author, date, branch,
1039 landed_in: [{tag, release_id, channel}]}
1040 One entry per resolved anchor commit, in timestamp order.
1041 all_landed – True when every anchor commit is in at least one release
1042 any_landed – True when at least one anchor commit is in a release
1043 pending_count – count of anchor commits not yet in any release
1044 next_tag – proposed next release tag (bump patch on latest semver)
1045 e.g. "v0.3.1" — derived from VCS history, never typed
1046 latest_release_tag – tag of the most recent release, or None
1047 """
1048 if not commit_anchors:
1049 return {
1050 "no_commits": True,
1051 "anchor_commits": [],
1052 "all_landed": False,
1053 "any_landed": False,
1054 "pending_count": 0,
1055 "next_tag": None,
1056 "latest_release_tag": None,
1057 }
1058
1059 from sqlalchemy import or_
1060 conditions = [
1061 MusehubCommit.commit_id.startswith(cid) if len(cid) < 64
1062 else MusehubCommit.commit_id == cid
1063 for cid in commit_anchors
1064 ]
1065 anchor_rows = (await session.execute(
1066 select(MusehubCommit)
1067 .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id)
1068 .where(
1069 MusehubCommitRef.repo_id == repo_id,
1070 or_(*conditions),
1071 )
1072 )).scalars().all()
1073
1074 # Fetch all releases ordered oldest → newest.
1075 release_rows = (await session.execute(
1076 select(MusehubRelease)
1077 .where(MusehubRelease.repo_id == repo_id)
1078 .order_by(
1079 MusehubRelease.semver_major,
1080 MusehubRelease.semver_minor,
1081 MusehubRelease.semver_patch,
1082 )
1083 )).scalars().all()
1084
1085 # Build a map: release commit_id → release timestamp for containment check.
1086 rel_commit_ids = [r.commit_id for r in release_rows if r.commit_id]
1087 rel_commit_rows: _CommitMap = {}
1088 if rel_commit_ids:
1089 rows = (await session.execute(
1090 select(MusehubCommit)
1091 .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id)
1092 .where(
1093 MusehubCommitRef.repo_id == repo_id,
1094 MusehubCommit.commit_id.in_(rel_commit_ids),
1095 )
1096 )).scalars().all()
1097 rel_commit_rows = {r.commit_id: r for r in rows}
1098
1099 # Determine next proposed tag from latest release semver.
1100 # If no releases exist yet, propose v0.1.0 as the first release.
1101 latest_release: "MusehubRelease | None" = release_rows[-1] if release_rows else None
1102 latest_release_tag: str | None = latest_release.tag if latest_release else None
1103 if latest_release and latest_release.semver_major is not None:
1104 patch = (latest_release.semver_patch or 0) + 1
1105 next_tag: str = f"v{latest_release.semver_major}.{latest_release.semver_minor or 0}.{patch}"
1106 else:
1107 next_tag = "v0.1.0"
1108
1109 # Build per-anchor-commit entries.
1110 anchor_commits: list[JSONObject] = []
1111 for row in sorted(anchor_rows, key=lambda r: (r.timestamp or "")):
1112 # Determine which releases contain this commit.
1113 landed_in: list[JSONObject] = []
1114 for rel in release_rows:
1115 if not rel.commit_id:
1116 continue
1117 rel_commit = rel_commit_rows.get(rel.commit_id)
1118 if (rel_commit and rel_commit.timestamp and row.timestamp
1119 and rel_commit.timestamp >= row.timestamp):
1120 landed_in.append({
1121 "tag": rel.tag,
1122 "release_id": rel.release_id,
1123 "channel": rel.channel or "stable",
1124 })
1125
1126 # First line of commit message, capped at 72 chars.
1127 first_line = (row.message or "").split("\n")[0].strip()
1128 if len(first_line) > 72:
1129 first_line = f"{first_line[:69]}…"
1130
1131 anchor_commits.append({
1132 "short": row.commit_id,
1133 "full": row.commit_id,
1134 "message": first_line,
1135 "author": row.author or "unknown",
1136 "date": row.timestamp.strftime("%Y-%m-%d") if row.timestamp else "",
1137 "branch": row.branch or "",
1138 "landed_in": landed_in,
1139 })
1140
1141 if not anchor_commits:
1142 # Anchor refs recorded but commits not in DB yet — still show the refs.
1143 for ref in commit_anchors:
1144 anchor_commits.append({
1145 "short": ref,
1146 "full": ref,
1147 "message": "",
1148 "author": "",
1149 "date": "",
1150 "branch": "",
1151 "landed_in": [],
1152 })
1153
1154 pending_count = sum(1 for ac in anchor_commits if not ac["landed_in"])
1155 all_landed = pending_count == 0
1156 any_landed = any(ac["landed_in"] for ac in anchor_commits)
1157
1158 return {
1159 "no_commits": False,
1160 "anchor_commits": anchor_commits,
1161 "all_landed": all_landed,
1162 "any_landed": any_landed,
1163 "pending_count": pending_count,
1164 "next_tag": next_tag,
1165 "latest_release_tag": latest_release_tag,
1166 }
File History 2 commits
sha256:4d42a346263e7cbbd152c147f3e6f24576f4b4440df9249ffb9fbcf9db699fcb feat: populate url in create_issue and create_proposal responses Sonnet 4.6 minor 35 days ago
sha256:3707eba7ad42cadedf18c8b9c534d839b88cfd1c30924c3c5a3edc74e1d809de feat: add url field to mist, issue, and proposal list/read … Sonnet 4.6 minor 35 days ago