musehub_proposals.py
python
sha256:f99af7b1a7f36c4d537d1c630d4b71fc39222b1255f82e930929e2fc89015e11
fix: relax browse_repo perf budget to 500ms — 200ms was too…
Sonnet 4.6
102 days ago
| 1 | """MuseHub merge proposal persistence adapter — single point of DB access for proposals. |
| 2 | |
| 3 | This module is the ONLY place that touches the ``musehub_proposals`` table. |
| 4 | Route handlers delegate here; no business logic lives in routes. |
| 5 | |
| 6 | Boundary rules: |
| 7 | - Must NOT import state stores, SSE queues, or LLM clients. |
| 8 | - May import ORM models from musehub.db domain-specific modules. |
| 9 | - May import Pydantic response models from musehub.models.musehub. |
| 10 | - May import musehub.core.genesis for genesis ID computation. |
| 11 | |
| 12 | Merge strategy |
| 13 | -------------- |
| 14 | ``merge_commit`` is the only strategy at MVP. It creates a new commit on |
| 15 | ``to_branch`` whose parent_ids are [to_branch head, from_branch head], then |
| 16 | updates the ``to_branch`` head pointer and marks the proposal as merged. |
| 17 | |
| 18 | If either branch has no commits yet (no head commit), the merge is rejected with |
| 19 | a ``ValueError`` — there is nothing to merge. |
| 20 | """ |
| 21 | |
| 22 | import logging |
| 23 | from datetime import datetime, timezone |
| 24 | |
| 25 | import sqlalchemy as sa |
| 26 | from sqlalchemy import ColumnElement, func, select |
| 27 | from sqlalchemy.ext.asyncio import AsyncSession |
| 28 | |
| 29 | from musehub.core.genesis import compute_branch_id, compute_comment_id, compute_proposal_id, compute_review_id, compute_simulation_id |
| 30 | from musehub.db.musehub_identity_models import MusehubIdentity |
| 31 | from musehub.db.musehub_repo_models import MusehubBranch, MusehubCommit, MusehubCommitRef |
| 32 | from musehub.db.musehub_social_models import ( |
| 33 | MusehubProposal, |
| 34 | MusehubProposalComment, |
| 35 | MusehubProposalReview, |
| 36 | MusehubProposalSimulation, |
| 37 | ) |
| 38 | from musehub.services.proposal_dag import ( |
| 39 | CycleError, |
| 40 | ProposalDag, |
| 41 | blocked_by_numbers, |
| 42 | blocks_numbers, |
| 43 | create_dependency_edges, |
| 44 | is_blocked, |
| 45 | load_dag_for_proposals, |
| 46 | ) |
| 47 | from musehub.muse_cli.snapshot import compute_commit_id, compute_snapshot_id |
| 48 | |
| 49 | |
| 50 | class BranchNotFoundError(Exception): |
| 51 | """Raised when a required branch does not exist in the repo.""" |
| 52 | from musehub.types.json_types import JSONObject, StrDict |
| 53 | from musehub.models.musehub import ( |
| 54 | DomainHeatEntry, |
| 55 | DomainHeatResponse, |
| 56 | MergeReadinessResponse, |
| 57 | ProposalCommentListResponse, |
| 58 | ProposalCommentResponse, |
| 59 | ProposalListEntry, |
| 60 | ProposalListFilters, |
| 61 | ProposalListResponse, |
| 62 | ProposalResponse, |
| 63 | ProposalReviewListResponse, |
| 64 | ProposalReviewResponse, |
| 65 | SimulationListResponse, |
| 66 | SimulationResponse, |
| 67 | ) |
| 68 | |
| 69 | type _CommentMap = dict[str, ProposalCommentResponse] |
| 70 | |
| 71 | logger = logging.getLogger(__name__) |
| 72 | |
| 73 | |
| 74 | def _utc_now() -> datetime: |
| 75 | return datetime.now(tz=timezone.utc) |
| 76 | |
| 77 | |
| 78 | |
| 79 | def _symbols_from_delta(delta: JSONObject | None) -> list[str]: |
| 80 | """Extract unique symbol addresses from a structured_delta dict. |
| 81 | |
| 82 | Only child_op addresses that contain ``::`` are returned — file-level ops |
| 83 | without a symbol component are intentionally excluded. |
| 84 | """ |
| 85 | if not isinstance(delta, dict): |
| 86 | return [] |
| 87 | seen: set[str] = set() |
| 88 | for file_op in delta.get("ops") or []: |
| 89 | if not isinstance(file_op, dict): |
| 90 | continue |
| 91 | for child_op in file_op.get("child_ops") or []: |
| 92 | if not isinstance(child_op, dict): |
| 93 | continue |
| 94 | addr = child_op.get("address", "") |
| 95 | if "::" in addr and addr not in seen: |
| 96 | seen.add(addr) |
| 97 | return list(seen) |
| 98 | |
| 99 | |
| 100 | async def _touched_symbols_for_branch( |
| 101 | session: AsyncSession, repo_id: str, branch: str |
| 102 | ) -> list[str]: |
| 103 | """Return the union of symbol addresses touched by all commits on ``branch``.""" |
| 104 | rows = (await session.execute( |
| 105 | select(MusehubCommit.structured_delta) |
| 106 | .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) |
| 107 | .where( |
| 108 | MusehubCommitRef.repo_id == repo_id, |
| 109 | MusehubCommit.branch == branch, |
| 110 | MusehubCommit.structured_delta.isnot(None), |
| 111 | ) |
| 112 | )).scalars().all() |
| 113 | seen: set[str] = set() |
| 114 | for delta in rows: |
| 115 | seen.update(_symbols_from_delta(delta)) |
| 116 | return list(seen) |
| 117 | |
| 118 | |
| 119 | def _to_proposal_response( |
| 120 | row: MusehubProposal, |
| 121 | *, |
| 122 | dag: ProposalDag | None = None, |
| 123 | simulations: "SimulationListResponse | None" = None, |
| 124 | ) -> ProposalResponse: |
| 125 | from musehub.models.musehub import MergeConditions, MergeStrategy, ProposalType |
| 126 | mc_raw = getattr(row, "merge_conditions", None) |
| 127 | mc = MergeConditions.model_validate(mc_raw) if mc_raw else None |
| 128 | |
| 129 | blocked_by: list[int] = [] |
| 130 | blocks: list[int] = [] |
| 131 | is_blocked_flag = False |
| 132 | if dag is not None: |
| 133 | blocked_by = blocked_by_numbers(dag, row.proposal_id) |
| 134 | blocks = blocks_numbers(dag, row.proposal_id) |
| 135 | is_blocked_flag = is_blocked(dag, row.proposal_id) |
| 136 | |
| 137 | latest_simulations: dict[str, dict] = {} |
| 138 | if simulations is not None: |
| 139 | for sim in simulations.simulations: |
| 140 | latest_simulations[sim.simulation_type] = { |
| 141 | "simulation_id": sim.simulation_id, |
| 142 | "result": sim.result, |
| 143 | "is_stale": sim.is_stale, |
| 144 | "from_branch_commit_id": sim.from_branch_commit_id, |
| 145 | "duration_ms": sim.duration_ms, |
| 146 | "created_at": sim.created_at.isoformat(), |
| 147 | } |
| 148 | |
| 149 | return ProposalResponse( |
| 150 | proposal_id=row.proposal_id, |
| 151 | proposal_number=row.proposal_number, |
| 152 | title=row.title, |
| 153 | body=row.body, |
| 154 | state=row.state, |
| 155 | from_branch=row.from_branch, |
| 156 | to_branch=row.to_branch, |
| 157 | merge_commit_id=row.merge_commit_id, |
| 158 | merged_at=row.merged_at, |
| 159 | author=row.author, |
| 160 | created_at=row.created_at, |
| 161 | proposal_type=ProposalType(getattr(row, "proposal_type", "state_merge")), |
| 162 | is_draft=getattr(row, "is_draft", False), |
| 163 | merge_conditions=mc, |
| 164 | merge_strategy=MergeStrategy(getattr(row, "merge_strategy", "state_overlay")), |
| 165 | selective_domains=getattr(row, "selective_domains", None), |
| 166 | risk_score=getattr(row, "risk_score", None), |
| 167 | dimensional_risk=dict(getattr(row, "dimensional_risk", None) or {}), |
| 168 | blocked_by=blocked_by, |
| 169 | blocks=blocks, |
| 170 | is_blocked=is_blocked_flag, |
| 171 | latest_simulations=latest_simulations, |
| 172 | proposer_signature=getattr(row, "proposer_signature", None), |
| 173 | proposer_public_key=getattr(row, "proposer_public_key", None), |
| 174 | from_snapshot_id=getattr(row, "from_snapshot_id", None), |
| 175 | to_snapshot_id=getattr(row, "to_snapshot_id", None), |
| 176 | ) |
| 177 | |
| 178 | |
| 179 | async def _get_branch( |
| 180 | session: AsyncSession, repo_id: str, branch_name: str |
| 181 | ) -> MusehubBranch | None: |
| 182 | """Return the branch record by repo + name, or None.""" |
| 183 | stmt = select(MusehubBranch).where( |
| 184 | MusehubBranch.repo_id == repo_id, |
| 185 | MusehubBranch.name == branch_name, |
| 186 | ) |
| 187 | return (await session.execute(stmt)).scalar_one_or_none() |
| 188 | |
| 189 | |
| 190 | async def create_proposal( |
| 191 | session: AsyncSession, |
| 192 | *, |
| 193 | repo_id: str, |
| 194 | title: str, |
| 195 | from_branch: str, |
| 196 | to_branch: str, |
| 197 | body: str = "", |
| 198 | author: str = "", |
| 199 | author_identity_id: str = "", |
| 200 | proposal_type: str = "state_merge", |
| 201 | is_draft: bool = False, |
| 202 | merge_strategy: str = "state_overlay", |
| 203 | merge_conditions: JSONObject | None = None, |
| 204 | selective_domains: list[str] | None = None, |
| 205 | depends_on: list[str] | None = None, |
| 206 | proposer_signature: str | None = None, |
| 207 | proposer_public_key: str | None = None, |
| 208 | ) -> ProposalResponse: |
| 209 | """Persist a new merge proposal in ``open`` state and return its wire representation. |
| 210 | |
| 211 | ``author`` identifies the user opening the proposal — typically the MSign handle |
| 212 | from the request context, or a display name from the seed script. |
| 213 | |
| 214 | Raises ``BranchNotFoundError`` if ``from_branch`` does not exist in the repo; |
| 215 | the caller should surface this as HTTP 404. |
| 216 | """ |
| 217 | branch = await _get_branch(session, repo_id, from_branch) |
| 218 | if branch is None: |
| 219 | raise BranchNotFoundError(f"Branch '{from_branch}' not found in repo {repo_id}") |
| 220 | from_snapshot_id = branch.head_commit_id |
| 221 | to_branch_row = await _get_branch(session, repo_id, to_branch) |
| 222 | to_snapshot_id = to_branch_row.head_commit_id if to_branch_row else None |
| 223 | |
| 224 | # Assign the next sequential proposal_number for this repo (1-based, like GitHub) |
| 225 | max_num_result = await session.execute( |
| 226 | select(func.max(MusehubProposal.proposal_number)).where( |
| 227 | MusehubProposal.repo_id == repo_id |
| 228 | ) |
| 229 | ) |
| 230 | max_num: int | None = max_num_result.scalar_one_or_none() |
| 231 | next_num = (max_num or 0) + 1 |
| 232 | |
| 233 | touched = await _touched_symbols_for_branch(session, repo_id, from_branch) |
| 234 | _created_at = _utc_now() |
| 235 | initial_state = "drafting" if is_draft else "open" |
| 236 | proposal = MusehubProposal( |
| 237 | proposal_id=compute_proposal_id( |
| 238 | repo_id, author_identity_id, from_branch, to_branch, _created_at.isoformat() |
| 239 | ), |
| 240 | repo_id=repo_id, |
| 241 | proposal_number=next_num, |
| 242 | title=title, |
| 243 | body=body, |
| 244 | state=initial_state, |
| 245 | from_branch=from_branch, |
| 246 | to_branch=to_branch, |
| 247 | author=author, |
| 248 | touched_symbols=touched, |
| 249 | created_at=_created_at, |
| 250 | proposal_type=proposal_type, |
| 251 | is_draft=is_draft, |
| 252 | merge_strategy=merge_strategy, |
| 253 | merge_conditions=merge_conditions, |
| 254 | selective_domains=selective_domains, |
| 255 | proposer_signature=proposer_signature, |
| 256 | proposer_public_key=proposer_public_key, |
| 257 | from_snapshot_id=from_snapshot_id, |
| 258 | to_snapshot_id=to_snapshot_id, |
| 259 | ) |
| 260 | session.add(proposal) |
| 261 | await session.flush() |
| 262 | |
| 263 | # Persist dependency edges — validates existence, detects cycles before commit |
| 264 | if depends_on: |
| 265 | await create_dependency_edges(session, proposal.proposal_id, depends_on) |
| 266 | |
| 267 | await session.refresh(proposal) |
| 268 | logger.info("✅ Created proposal '%s' (%s → %s) in repo %s", title, from_branch, to_branch, repo_id) |
| 269 | return _to_proposal_response(proposal) |
| 270 | |
| 271 | |
| 272 | def _risk_band_conditions(bands: list[str]) -> list[ColumnElement[bool]]: |
| 273 | """Return SQLAlchemy conditions that match risk_score to the given band names. |
| 274 | |
| 275 | Each band maps to a half-open interval on [0.0, 1.0]: |
| 276 | critical ≥ 0.75 |
| 277 | high 0.50 – 0.74… |
| 278 | medium 0.25 – 0.49… |
| 279 | low 0.01 – 0.24… |
| 280 | none == 0.0 (or NULL) |
| 281 | |
| 282 | Multiple bands are OR-ed together. |
| 283 | """ |
| 284 | band_ranges: dict[str, tuple[float | None, float | None]] = { |
| 285 | "critical": (0.75, None), |
| 286 | "high": (0.50, 0.75), |
| 287 | "medium": (0.25, 0.50), |
| 288 | "low": (0.01, 0.25), |
| 289 | "none": (None, 0.01), |
| 290 | } |
| 291 | clauses = [] |
| 292 | for band in bands: |
| 293 | lo, hi = band_ranges.get(band, (None, None)) |
| 294 | if band == "none": |
| 295 | clauses.append( |
| 296 | sa.or_( |
| 297 | MusehubProposal.risk_score.is_(None), |
| 298 | MusehubProposal.risk_score == 0.0, |
| 299 | ) |
| 300 | ) |
| 301 | elif lo is not None and hi is not None: |
| 302 | clauses.append( |
| 303 | sa.and_( |
| 304 | MusehubProposal.risk_score >= lo, |
| 305 | MusehubProposal.risk_score < hi, |
| 306 | ) |
| 307 | ) |
| 308 | elif lo is not None: |
| 309 | clauses.append(MusehubProposal.risk_score >= lo) |
| 310 | return clauses |
| 311 | |
| 312 | |
| 313 | async def list_proposals( |
| 314 | session: AsyncSession, |
| 315 | repo_id: str, |
| 316 | *, |
| 317 | state: str = "all", |
| 318 | cursor: str | None = None, |
| 319 | limit: int = 20, |
| 320 | filters: ProposalListFilters | None = None, |
| 321 | ) -> ProposalListResponse: |
| 322 | """Return merge proposals for a repo with cursor-based keyset pagination. |
| 323 | |
| 324 | ``state`` may be ``"open"``, ``"merged"``, ``"closed"``, ``"all"``, or any |
| 325 | value in the extended 7-state machine accepted by ``ProposalListFilters``. |
| 326 | |
| 327 | When ``filters`` is provided, the following additional predicates are applied: |
| 328 | - ``filters.risk_band`` → ``risk_score`` range filter (OR across bands) |
| 329 | - ``filters.domain`` → ``risk_score > 0`` when "code" is included; |
| 330 | other domains require per-domain risk rows |
| 331 | (Phase 2 DB prerequisite) |
| 332 | - ``filters.author_type`` → LEFT JOIN to ``musehub_identities`` |
| 333 | - ``filters.assigned_reviewer`` → EXISTS sub-select on ``musehub_proposal_reviews`` |
| 334 | - ``filters.sort`` → ordering; ``merge_ready_first`` uses an approval |
| 335 | count sub-select to surface ready proposals first |
| 336 | |
| 337 | ``cursor`` is the ISO 8601 ``created_at`` of the last seen proposal (opaque |
| 338 | to callers — pass ``nextCursor`` from a previous response verbatim). |
| 339 | |
| 340 | Args: |
| 341 | session: Async database session. |
| 342 | repo_id: Target repository ID. |
| 343 | state: Proposal state filter; defaults to "all". |
| 344 | cursor: Pagination cursor (opaque ISO 8601 string). |
| 345 | limit: Page size. |
| 346 | filters: Optional ``ProposalListFilters``; overrides ``state``, ``limit``, |
| 347 | ``cursor``, and ``sort`` when provided. ``state`` in ``filters`` |
| 348 | takes precedence over the top-level ``state`` kwarg. |
| 349 | |
| 350 | Returns: |
| 351 | ``ProposalListResponse`` with paginated proposals and a ``nextCursor``. |
| 352 | |
| 353 | Raises: |
| 354 | ValueError: If ``cursor`` is not a valid ISO 8601 datetime string. |
| 355 | """ |
| 356 | f = filters |
| 357 | effective_state = (f.state if f else None) or state |
| 358 | effective_limit = (f.limit if f else None) or limit |
| 359 | effective_cursor = (f.cursor if f else None) or cursor |
| 360 | effective_sort = (f.sort if f else None) or "newest" |
| 361 | |
| 362 | conditions: list[ColumnElement[bool]] = [MusehubProposal.repo_id == repo_id] |
| 363 | if effective_state != "all": |
| 364 | conditions.append(MusehubProposal.state == effective_state) |
| 365 | |
| 366 | stmt = select(MusehubProposal) |
| 367 | |
| 368 | # ── Author-type filter (requires join to identities) ─────────────────────── |
| 369 | if f and f.author_type != "all": |
| 370 | stmt = stmt.join( |
| 371 | MusehubIdentity, |
| 372 | MusehubIdentity.handle == MusehubProposal.author, |
| 373 | isouter=True, |
| 374 | ) |
| 375 | if f.author_type == "human": |
| 376 | conditions.append( |
| 377 | sa.or_( |
| 378 | MusehubIdentity.identity_type == "human", |
| 379 | MusehubIdentity.identity_type.is_(None), |
| 380 | ) |
| 381 | ) |
| 382 | else: |
| 383 | conditions.append(MusehubIdentity.identity_type == f.author_type) |
| 384 | |
| 385 | # ── Risk-band filter ─────────────────────────────────────────────────────── |
| 386 | if f and f.risk_band: |
| 387 | band_clauses = _risk_band_conditions(f.risk_band) |
| 388 | if band_clauses: |
| 389 | conditions.append(sa.or_(*band_clauses)) |
| 390 | |
| 391 | # ── Domain filter — code only at Phase 2 (other domains require risk rows) ─ |
| 392 | if f and f.domain: |
| 393 | domain_clauses = [] |
| 394 | if "code" in f.domain: |
| 395 | domain_clauses.append( |
| 396 | sa.and_( |
| 397 | MusehubProposal.risk_score.is_not(None), |
| 398 | MusehubProposal.risk_score > 0.0, |
| 399 | ) |
| 400 | ) |
| 401 | if domain_clauses: |
| 402 | conditions.append(sa.or_(*domain_clauses)) |
| 403 | |
| 404 | # ── Proposal-type filter ────────────────────────────────────────────────── |
| 405 | if f and f.proposal_type: |
| 406 | conditions.append(MusehubProposal.proposal_type.in_(f.proposal_type)) |
| 407 | |
| 408 | # ── Is-draft filter ─────────────────────────────────────────────────────── |
| 409 | if f and f.is_draft is not None: |
| 410 | conditions.append(MusehubProposal.is_draft == f.is_draft) |
| 411 | |
| 412 | # ── Merge-strategy filter ───────────────────────────────────────────────── |
| 413 | if f and f.merge_strategy: |
| 414 | conditions.append(MusehubProposal.merge_strategy.in_(f.merge_strategy)) |
| 415 | |
| 416 | # ── Assigned-reviewer filter ─────────────────────────────────────────────── |
| 417 | if f and f.assigned_reviewer: |
| 418 | reviewer_subq = ( |
| 419 | select(MusehubProposalReview.proposal_id) |
| 420 | .where( |
| 421 | MusehubProposalReview.reviewer_username == f.assigned_reviewer, |
| 422 | MusehubProposalReview.state.in_(["pending", "approved", "changes_requested"]), |
| 423 | ) |
| 424 | .correlate(MusehubProposal) |
| 425 | ) |
| 426 | conditions.append(MusehubProposal.proposal_id.in_(reviewer_subq)) |
| 427 | |
| 428 | # ── Count total matching rows (re-use same joins as data query) ─────────── |
| 429 | count_stmt = stmt.with_only_columns( |
| 430 | func.count(MusehubProposal.proposal_id) |
| 431 | ).where(*conditions).order_by(None) |
| 432 | total: int = (await session.execute(count_stmt)).scalar_one() |
| 433 | |
| 434 | # ── Sort order ───────────────────────────────────────────────────────────── |
| 435 | order_clauses: list[ColumnElement[bool]] |
| 436 | cursor_ascending: bool # True → > cursor, False → < cursor |
| 437 | if effective_sort == "oldest": |
| 438 | order_clauses = [MusehubProposal.created_at.asc()] |
| 439 | cursor_ascending = True |
| 440 | elif effective_sort == "risk_desc": |
| 441 | order_clauses = [MusehubProposal.risk_score.desc().nulls_last()] |
| 442 | cursor_ascending = False |
| 443 | elif effective_sort == "risk_asc": |
| 444 | order_clauses = [MusehubProposal.risk_score.asc().nulls_last()] |
| 445 | cursor_ascending = False |
| 446 | elif effective_sort == "merge_ready_first": |
| 447 | # Sub-select: count approved reviews per proposal; ready = count>=2 AND breakage=0 |
| 448 | approval_subq = ( |
| 449 | select(func.count(MusehubProposalReview.review_id)) |
| 450 | .where( |
| 451 | MusehubProposalReview.proposal_id == MusehubProposal.proposal_id, |
| 452 | MusehubProposalReview.state == "approved", |
| 453 | ) |
| 454 | .correlate(MusehubProposal) |
| 455 | .scalar_subquery() |
| 456 | ) |
| 457 | is_ready = sa.case( |
| 458 | ( |
| 459 | sa.and_( |
| 460 | approval_subq >= _DEFAULT_REQUIRED_APPROVALS, |
| 461 | MusehubProposal.breakage_count == 0, |
| 462 | ), |
| 463 | 0, |
| 464 | ), |
| 465 | else_=1, |
| 466 | ) |
| 467 | order_clauses = [is_ready, MusehubProposal.created_at.desc()] |
| 468 | cursor_ascending = False |
| 469 | else: |
| 470 | # newest (default) |
| 471 | order_clauses = [MusehubProposal.created_at.desc()] |
| 472 | cursor_ascending = False |
| 473 | |
| 474 | # ── Cursor predicate ─────────────────────────────────────────────────────── |
| 475 | data_conditions = list(conditions) |
| 476 | if effective_cursor is not None: |
| 477 | cursor_dt = datetime.fromisoformat(effective_cursor) |
| 478 | if cursor_ascending: |
| 479 | data_conditions.append(MusehubProposal.created_at > cursor_dt) |
| 480 | else: |
| 481 | data_conditions.append(MusehubProposal.created_at < cursor_dt) |
| 482 | |
| 483 | rows = list( |
| 484 | ( |
| 485 | await session.execute( |
| 486 | stmt.where(*data_conditions).order_by(*order_clauses).limit(effective_limit + 1) |
| 487 | ) |
| 488 | ).scalars() |
| 489 | ) |
| 490 | |
| 491 | next_cursor: str | None = None |
| 492 | if len(rows) == effective_limit + 1: |
| 493 | next_cursor = rows[effective_limit - 1].created_at.isoformat() |
| 494 | rows = rows[:effective_limit] |
| 495 | |
| 496 | return ProposalListResponse( |
| 497 | proposals=[_to_proposal_response(r) for r in rows], |
| 498 | total=total, |
| 499 | next_cursor=next_cursor, |
| 500 | ) |
| 501 | |
| 502 | |
| 503 | async def get_proposal( |
| 504 | session: AsyncSession, |
| 505 | repo_id: str, |
| 506 | proposal_id: str, |
| 507 | ) -> ProposalResponse | None: |
| 508 | """Return a single proposal enriched with DAG position and simulation summaries.""" |
| 509 | stmt = select(MusehubProposal).where( |
| 510 | MusehubProposal.repo_id == repo_id, |
| 511 | MusehubProposal.proposal_id == proposal_id, |
| 512 | ) |
| 513 | row = (await session.execute(stmt)).scalar_one_or_none() |
| 514 | if row is None: |
| 515 | return None |
| 516 | dag = await load_dag_for_proposals(session, [proposal_id]) |
| 517 | sims = await list_simulations(session, repo_id, proposal_id) |
| 518 | return _to_proposal_response(row, dag=dag, simulations=sims) |
| 519 | |
| 520 | |
| 521 | async def update_proposal( |
| 522 | session: AsyncSession, |
| 523 | repo_id: str, |
| 524 | proposal_id: str, |
| 525 | *, |
| 526 | title: str | None = None, |
| 527 | body: str | None = None, |
| 528 | proposal_type: str | None = None, |
| 529 | merge_strategy: str | None = None, |
| 530 | ) -> ProposalResponse | None: |
| 531 | """Apply a partial update to a proposal. Returns None if not found.""" |
| 532 | stmt = select(MusehubProposal).where( |
| 533 | MusehubProposal.repo_id == repo_id, |
| 534 | MusehubProposal.proposal_id == proposal_id, |
| 535 | ) |
| 536 | row = (await session.execute(stmt)).scalar_one_or_none() |
| 537 | if row is None: |
| 538 | return None |
| 539 | if title is not None: |
| 540 | row.title = title |
| 541 | if body is not None: |
| 542 | row.body = body |
| 543 | if proposal_type is not None: |
| 544 | row.proposal_type = proposal_type |
| 545 | if merge_strategy is not None: |
| 546 | row.merge_strategy = merge_strategy |
| 547 | await session.commit() |
| 548 | await session.refresh(row) |
| 549 | dag = await load_dag_for_proposals(session, [proposal_id]) |
| 550 | sims = await list_simulations(session, repo_id, proposal_id) |
| 551 | return _to_proposal_response(row, dag=dag, simulations=sims) |
| 552 | |
| 553 | |
| 554 | async def _resolve_ancestor_manifest( |
| 555 | session: AsyncSession, |
| 556 | repo_id: str, |
| 557 | from_branch: str, |
| 558 | to_branch: str, |
| 559 | ) -> StrDict | None: |
| 560 | """Find the common ancestor snapshot manifest for a branch pair. |
| 561 | |
| 562 | Walks the from_branch commit history looking for the first commit whose |
| 563 | parent_ids overlap with commits reachable from to_branch. This is a |
| 564 | lightweight merge-base approximation suitable for server-side strategy |
| 565 | computation (not a full LCA traversal). |
| 566 | |
| 567 | Returns None if no ancestor can be found (new repo, orphan branches). |
| 568 | """ |
| 569 | from musehub.graph.walk import walk_dag_async |
| 570 | from musehub.services.musehub_snapshot import get_snapshot_manifest |
| 571 | |
| 572 | to_b = await _get_branch(session, repo_id, to_branch) |
| 573 | from_b = await _get_branch(session, repo_id, from_branch) |
| 574 | if to_b is None or from_b is None: |
| 575 | return None |
| 576 | |
| 577 | # Walk 1: collect to_branch ancestry (bounded BFS) |
| 578 | to_commit_ids: set[str] = set() |
| 579 | |
| 580 | async def _to_adj(cid: str) -> list[str]: |
| 581 | commit = await session.get(MusehubCommit, cid) |
| 582 | return commit.parent_ids if commit and commit.parent_ids else [] |
| 583 | |
| 584 | async for cid in walk_dag_async( |
| 585 | [to_b.head_commit_id] if to_b.head_commit_id else [], |
| 586 | _to_adj, |
| 587 | max_nodes=200, |
| 588 | ): |
| 589 | to_commit_ids.add(cid) |
| 590 | |
| 591 | # Walk 2: first-parent walk on from_branch, looking for merge base |
| 592 | candidate_id: str | None = None |
| 593 | |
| 594 | async def _from_adj(cid: str) -> list[str]: |
| 595 | nonlocal candidate_id |
| 596 | commit = await session.get(MusehubCommit, cid) |
| 597 | if commit is None: |
| 598 | return [] |
| 599 | for parent_id in (commit.parent_ids or []): |
| 600 | if parent_id in to_commit_ids: |
| 601 | candidate_id = parent_id |
| 602 | return [] # stop walking once merge base is found |
| 603 | return commit.parent_ids[:1] if commit.parent_ids else [] |
| 604 | |
| 605 | async for _ in walk_dag_async( |
| 606 | [from_b.head_commit_id] if from_b.head_commit_id else [], |
| 607 | _from_adj, |
| 608 | max_nodes=200, |
| 609 | ): |
| 610 | if candidate_id: |
| 611 | break |
| 612 | |
| 613 | if not candidate_id: |
| 614 | return None |
| 615 | |
| 616 | ancestor_commit = await session.get(MusehubCommit, candidate_id) |
| 617 | if ancestor_commit is None or not ancestor_commit.snapshot_id: |
| 618 | return None |
| 619 | |
| 620 | return await get_snapshot_manifest(session, ancestor_commit.snapshot_id) |
| 621 | |
| 622 | |
| 623 | async def merge_proposal( |
| 624 | session: AsyncSession, |
| 625 | repo_id: str, |
| 626 | proposal_id: str, |
| 627 | *, |
| 628 | merge_strategy: str = "merge_commit", |
| 629 | merger_handle: str = "", |
| 630 | ) -> ProposalResponse: |
| 631 | """Merge an open proposal using the given strategy. |
| 632 | |
| 633 | Creates a merge commit on ``to_branch`` with parent_ids = |
| 634 | [to_branch head, from_branch head], updates the branch head pointer, and |
| 635 | marks the proposal as ``merged``. Sets ``merged_at`` to the current UTC time |
| 636 | so the timeline overlay can position the merge marker at the actual merge |
| 637 | instant rather than the proposal creation date. |
| 638 | |
| 639 | Raises: |
| 640 | ValueError: Proposal not found or ``from_branch`` does not exist or has no commits. |
| 641 | RuntimeError: Proposal is already merged or closed (caller surfaces as 409). |
| 642 | """ |
| 643 | stmt = select(MusehubProposal).where( |
| 644 | MusehubProposal.repo_id == repo_id, |
| 645 | MusehubProposal.proposal_id == proposal_id, |
| 646 | ) |
| 647 | proposal = (await session.execute(stmt)).scalar_one_or_none() |
| 648 | if proposal is None: |
| 649 | raise ValueError(f"Proposal {proposal_id} not found in repo {repo_id}") |
| 650 | |
| 651 | if proposal.state not in ("open", "approved"): |
| 652 | raise RuntimeError(f"Proposal {proposal_id} is already {proposal.state}") |
| 653 | |
| 654 | # Gate on hard dependencies — check require_dependency_merged from merge_conditions |
| 655 | mc_raw = proposal.merge_conditions or {} |
| 656 | if mc_raw.get("require_dependency_merged", True): |
| 657 | dag = await load_dag_for_proposals(session, [proposal_id]) |
| 658 | if is_blocked(dag, proposal_id): |
| 659 | unmerged_nums = blocked_by_numbers(dag, proposal_id) |
| 660 | raise RuntimeError( |
| 661 | f"Proposal {proposal_id} cannot be merged: " |
| 662 | f"unmerged dependencies: proposal numbers {unmerged_nums}" |
| 663 | ) |
| 664 | |
| 665 | from_b = await _get_branch(session, repo_id, proposal.from_branch) |
| 666 | to_b = await _get_branch(session, repo_id, proposal.to_branch) |
| 667 | |
| 668 | # Collect parent commit IDs for the merge commit. |
| 669 | parent_ids: list[str] = [] |
| 670 | if to_b is not None and to_b.head_commit_id is not None: |
| 671 | parent_ids.append(to_b.head_commit_id) |
| 672 | if from_b is not None and from_b.head_commit_id is not None: |
| 673 | parent_ids.append(from_b.head_commit_id) |
| 674 | |
| 675 | if not parent_ids: |
| 676 | raise ValueError( |
| 677 | f"Cannot merge: neither '{proposal.from_branch}' nor '{proposal.to_branch}' has any commits" |
| 678 | ) |
| 679 | |
| 680 | from musehub.services.musehub_snapshot import get_snapshot_manifest, upsert_snapshot_entries |
| 681 | from musehub.services.proposal_merge_strategies import execute_merge_strategy |
| 682 | |
| 683 | to_manifest: StrDict = {} |
| 684 | if to_b is not None and to_b.head_commit_id is not None: |
| 685 | to_head = await session.get(MusehubCommit, to_b.head_commit_id) |
| 686 | if to_head is not None and to_head.snapshot_id: |
| 687 | to_manifest = await get_snapshot_manifest(session, to_head.snapshot_id) |
| 688 | |
| 689 | from_manifest: StrDict = {} |
| 690 | from_head_snapshot_id: str | None = None |
| 691 | if from_b is not None and from_b.head_commit_id is not None: |
| 692 | from_head = await session.get(MusehubCommit, from_b.head_commit_id) |
| 693 | if from_head is not None and from_head.snapshot_id: |
| 694 | from_head_snapshot_id = from_head.snapshot_id |
| 695 | from_manifest = await get_snapshot_manifest(session, from_head.snapshot_id) |
| 696 | |
| 697 | # Resolve ancestor manifest for three-way strategies. |
| 698 | # The ancestor is the snapshot at the point from_branch was cut from to_branch. |
| 699 | # We approximate this as the earliest commit on from_branch that has a parent |
| 700 | # on to_branch — or the repo's first commit if the branch predates any tracking. |
| 701 | # For STATE_OVERLAY (default), the ancestor is used only for conflict audit. |
| 702 | ancestor_manifest: StrDict | None = None |
| 703 | strategy_name = getattr(proposal, "merge_strategy", "state_overlay") or "state_overlay" |
| 704 | if strategy_name in ("state_weave", "state_rebase", "domain_selective", "phased"): |
| 705 | # Walk from_branch commit history to find the merge-base with to_branch. |
| 706 | # Simplified: look for the first from_branch commit whose parent is in to_branch. |
| 707 | ancestor_manifest = await _resolve_ancestor_manifest( |
| 708 | session, repo_id, proposal.from_branch, proposal.to_branch |
| 709 | ) |
| 710 | |
| 711 | selective_domains: list[str] | None = getattr(proposal, "selective_domains", None) |
| 712 | |
| 713 | merge_result = execute_merge_strategy( |
| 714 | strategy_name, |
| 715 | to_manifest, |
| 716 | from_manifest, |
| 717 | ancestor_manifest=ancestor_manifest, |
| 718 | selective_domains=selective_domains, |
| 719 | ) |
| 720 | merged_manifest: StrDict = merge_result.manifest |
| 721 | |
| 722 | logger.info( |
| 723 | "🔀 Merge strategy=%s added=%d modified=%d removed=%d conflicts=%d domains=%s", |
| 724 | merge_result.strategy, |
| 725 | merge_result.files_added, |
| 726 | merge_result.files_modified, |
| 727 | merge_result.files_removed, |
| 728 | len(merge_result.conflicts), |
| 729 | merge_result.domains_merged, |
| 730 | ) |
| 731 | |
| 732 | # Persist the merged snapshot — ID computed via compute_snapshot_id to |
| 733 | # guarantee bit-for-bit compatibility with the Muse CLI's verification. |
| 734 | merged_snapshot_id = compute_snapshot_id(merged_manifest) |
| 735 | await upsert_snapshot_entries(session, repo_id, merged_snapshot_id, merged_manifest) |
| 736 | |
| 737 | # Create the merge commit — ID computed via compute_commit_id to match |
| 738 | # the Muse CLI's commit-ID formula exactly. author and signer_public_key |
| 739 | # must be passed to compute_commit_id so the hash covers the same fields |
| 740 | # the CLI uses when verifying the commit on pull. |
| 741 | merge_message = f"Merge '{proposal.from_branch}' into '{proposal.to_branch}' — proposal: {proposal.title}" |
| 742 | committed_at = _utc_now() |
| 743 | merge_author = merger_handle |
| 744 | merge_signer_public_key = "" |
| 745 | merge_commit_id = compute_commit_id( |
| 746 | parent_ids, |
| 747 | merged_snapshot_id, |
| 748 | merge_message, |
| 749 | committed_at.isoformat(), |
| 750 | author=merge_author, |
| 751 | signer_public_key=merge_signer_public_key, |
| 752 | ) |
| 753 | merge_commit = MusehubCommit( |
| 754 | commit_id=merge_commit_id, |
| 755 | branch=proposal.to_branch, |
| 756 | parent_ids=parent_ids, |
| 757 | message=merge_message, |
| 758 | author=merge_author, |
| 759 | timestamp=committed_at, |
| 760 | snapshot_id=merged_snapshot_id, |
| 761 | ) |
| 762 | session.add(merge_commit) |
| 763 | session.add(MusehubCommitRef(repo_id=repo_id, commit_id=merge_commit_id)) |
| 764 | |
| 765 | # Advance (or create) the to_branch head pointer. |
| 766 | if to_b is None: |
| 767 | to_b = MusehubBranch( |
| 768 | branch_id=compute_branch_id(repo_id, proposal.to_branch), |
| 769 | repo_id=repo_id, |
| 770 | name=proposal.to_branch, |
| 771 | head_commit_id=merge_commit_id, |
| 772 | ) |
| 773 | session.add(to_b) |
| 774 | else: |
| 775 | to_b.head_commit_id = merge_commit_id |
| 776 | |
| 777 | # Delete the source branch so it disappears from refs. This lets |
| 778 | # `muse fetch --prune` clean up the local tracking ref automatically, |
| 779 | # matching the behaviour users expect after a proposal merge. |
| 780 | if from_b is not None: |
| 781 | await session.delete(from_b) |
| 782 | |
| 783 | # Refresh touched_symbols from the from_branch commits before marking merged. |
| 784 | # This gives the most accurate symbol set at the moment of merge, capturing |
| 785 | # any commits pushed to from_branch after the proposal was created. |
| 786 | touched = await _touched_symbols_for_branch(session, repo_id, proposal.from_branch) |
| 787 | proposal.touched_symbols = touched |
| 788 | |
| 789 | # Mark proposal as merged and record the exact merge timestamp. |
| 790 | proposal.state = "merged" |
| 791 | proposal.merge_commit_id = merge_commit_id |
| 792 | proposal.merged_at = _utc_now() |
| 793 | |
| 794 | await session.flush() |
| 795 | await session.refresh(proposal) |
| 796 | logger.info( |
| 797 | "✅ Merged proposal %s ('%s' → '%s') in repo %s, merge commit %s", |
| 798 | proposal_id, |
| 799 | proposal.from_branch, |
| 800 | proposal.to_branch, |
| 801 | repo_id, |
| 802 | merge_commit_id, |
| 803 | ) |
| 804 | return _to_proposal_response(proposal) |
| 805 | |
| 806 | |
| 807 | # --------------------------------------------------------------------------- |
| 808 | # Reopen proposal |
| 809 | # --------------------------------------------------------------------------- |
| 810 | |
| 811 | |
| 812 | async def close_proposal( |
| 813 | session: AsyncSession, |
| 814 | repo_id: str, |
| 815 | proposal_id: str, |
| 816 | ) -> ProposalResponse: |
| 817 | """Set an open proposal to ``closed`` state. |
| 818 | |
| 819 | Raises: |
| 820 | KeyError: proposal not found in this repo. |
| 821 | RuntimeError: proposal is already closed or merged. |
| 822 | """ |
| 823 | proposal = (await session.execute( |
| 824 | select(MusehubProposal).where( |
| 825 | MusehubProposal.proposal_id == proposal_id, |
| 826 | MusehubProposal.repo_id == repo_id, |
| 827 | ) |
| 828 | )).scalar_one_or_none() |
| 829 | if proposal is None: |
| 830 | raise KeyError(f"Proposal {proposal_id} not found in repo {repo_id}") |
| 831 | if proposal.state != "open": |
| 832 | raise RuntimeError(f"Proposal {proposal_id} is already {proposal.state}") |
| 833 | proposal.state = "closed" |
| 834 | await session.flush() |
| 835 | await session.refresh(proposal) |
| 836 | return _to_proposal_response(proposal) |
| 837 | |
| 838 | |
| 839 | async def reopen_proposal( |
| 840 | session: AsyncSession, |
| 841 | repo_id: str, |
| 842 | proposal_id: str, |
| 843 | ) -> ProposalResponse: |
| 844 | """Reset a merged or closed proposal back to ``open`` state. |
| 845 | |
| 846 | Clears ``merge_commit_id`` and ``merged_at`` so the proposal can be |
| 847 | re-merged after a corrupt commit is cleaned up (e.g. bug #36). |
| 848 | |
| 849 | Raises: |
| 850 | KeyError: proposal not found in this repo. |
| 851 | RuntimeError: proposal is already open. |
| 852 | """ |
| 853 | proposal = (await session.execute( |
| 854 | select(MusehubProposal).where( |
| 855 | MusehubProposal.proposal_id == proposal_id, |
| 856 | MusehubProposal.repo_id == repo_id, |
| 857 | ) |
| 858 | )).scalar_one_or_none() |
| 859 | if proposal is None: |
| 860 | raise KeyError(f"Proposal {proposal_id} not found in repo {repo_id}") |
| 861 | if proposal.state == "open": |
| 862 | raise RuntimeError(f"Proposal {proposal_id} is already open") |
| 863 | proposal.state = "open" |
| 864 | proposal.merge_commit_id = None |
| 865 | proposal.merged_at = None |
| 866 | await session.flush() |
| 867 | await session.refresh(proposal) |
| 868 | return _to_proposal_response(proposal) |
| 869 | |
| 870 | |
| 871 | # --------------------------------------------------------------------------- |
| 872 | # Proposal review comments |
| 873 | # --------------------------------------------------------------------------- |
| 874 | |
| 875 | |
| 876 | def _to_comment_response(row: MusehubProposalComment) -> ProposalCommentResponse: |
| 877 | dim_ref: JSONObject = row.dimension_ref or {} |
| 878 | return ProposalCommentResponse( |
| 879 | comment_id=row.comment_id, |
| 880 | proposal_id=row.proposal_id, |
| 881 | author=row.author, |
| 882 | body=row.body, |
| 883 | target_type=str(dim_ref.get("type", "general")), |
| 884 | target_track=str(dim_ref["track"]) if "track" in dim_ref else None, |
| 885 | target_beat_start=float(dim_ref["beat_start"]) if isinstance(dim_ref.get("beat_start"), (int, float)) else None, |
| 886 | target_beat_end=float(dim_ref["beat_end"]) if isinstance(dim_ref.get("beat_end"), (int, float)) else None, |
| 887 | target_note_pitch=int(dim_ref["pitch"]) if isinstance(dim_ref.get("pitch"), int) else None, |
| 888 | parent_comment_id=row.parent_comment_id, |
| 889 | symbol_address=row.symbol_address, |
| 890 | created_at=row.created_at, |
| 891 | ) |
| 892 | |
| 893 | |
| 894 | async def create_proposal_comment( |
| 895 | session: AsyncSession, |
| 896 | *, |
| 897 | proposal_id: str, |
| 898 | repo_id: str, |
| 899 | author: str, |
| 900 | author_identity_id: str = "", |
| 901 | body: str, |
| 902 | target_type: str = "general", |
| 903 | target_track: str | None = None, |
| 904 | target_beat_start: float | None = None, |
| 905 | target_beat_end: float | None = None, |
| 906 | target_note_pitch: int | None = None, |
| 907 | parent_comment_id: str | None = None, |
| 908 | symbol_address: str | None = None, |
| 909 | ) -> ProposalCommentResponse: |
| 910 | """Persist a new review comment on a proposal and return its wire representation. |
| 911 | |
| 912 | ``author`` is the MSign handle of the reviewer. |
| 913 | ``parent_comment_id`` must be an existing top-level comment on the same proposal |
| 914 | when creating a threaded reply; the caller validates this constraint before |
| 915 | calling here. |
| 916 | |
| 917 | Raises ``ValueError`` if the proposal does not exist in the given repo. |
| 918 | """ |
| 919 | stmt = select(MusehubProposal).where( |
| 920 | MusehubProposal.proposal_id == proposal_id, |
| 921 | MusehubProposal.repo_id == repo_id, |
| 922 | ) |
| 923 | proposal = (await session.execute(stmt)).scalar_one_or_none() |
| 924 | if proposal is None: |
| 925 | raise ValueError(f"Proposal {proposal_id} not found in repo {repo_id}") |
| 926 | |
| 927 | dimension_ref: JSONObject = {"type": target_type} |
| 928 | if target_track is not None: |
| 929 | dimension_ref["track"] = target_track |
| 930 | if target_beat_start is not None: |
| 931 | dimension_ref["beat_start"] = target_beat_start |
| 932 | if target_beat_end is not None: |
| 933 | dimension_ref["beat_end"] = target_beat_end |
| 934 | if target_note_pitch is not None: |
| 935 | dimension_ref["pitch"] = target_note_pitch |
| 936 | |
| 937 | _created_at = _utc_now() |
| 938 | comment = MusehubProposalComment( |
| 939 | comment_id=compute_comment_id(proposal_id, author_identity_id, _created_at.isoformat()), |
| 940 | proposal_id=proposal_id, |
| 941 | repo_id=repo_id, |
| 942 | author=author, |
| 943 | body=body, |
| 944 | dimension_ref=dimension_ref, |
| 945 | parent_comment_id=parent_comment_id, |
| 946 | symbol_address=symbol_address or None, |
| 947 | created_at=_created_at, |
| 948 | ) |
| 949 | session.add(comment) |
| 950 | await session.flush() |
| 951 | await session.refresh(comment) |
| 952 | logger.info("✅ Created proposal comment %s on proposal %s by %s", comment.comment_id, proposal_id, author) |
| 953 | return _to_comment_response(comment) |
| 954 | |
| 955 | |
| 956 | async def list_proposal_comments( |
| 957 | session: AsyncSession, |
| 958 | proposal_id: str, |
| 959 | repo_id: str, |
| 960 | cursor: str | None = None, |
| 961 | limit: int = 20, |
| 962 | ) -> ProposalCommentListResponse: |
| 963 | """Return review comments for a proposal with cursor-based keyset pagination. |
| 964 | |
| 965 | Comments are assembled into a two-level thread tree from the current page. |
| 966 | Top-level comments (``parent_comment_id`` is None) form the root list. |
| 967 | Each carries a ``replies`` list with direct children that appear within |
| 968 | the same page, sorted by ``created_at`` ascending. Grandchildren are not |
| 969 | supported — callers should reply to the original top-level comment. |
| 970 | |
| 971 | ``cursor`` is the ISO 8601 ``created_at`` of the last seen comment |
| 972 | (opaque to callers — pass ``nextCursor`` from a previous response |
| 973 | verbatim). Omit to start from the beginning. ``total`` covers all |
| 974 | comments on the proposal regardless of the current page. |
| 975 | """ |
| 976 | conditions = [ |
| 977 | MusehubProposalComment.proposal_id == proposal_id, |
| 978 | MusehubProposalComment.repo_id == repo_id, |
| 979 | ] |
| 980 | |
| 981 | count_stmt = select(func.count(MusehubProposalComment.comment_id)).where(*conditions) |
| 982 | total: int = (await session.execute(count_stmt)).scalar_one() |
| 983 | |
| 984 | data_conditions = list(conditions) |
| 985 | if cursor is not None: |
| 986 | data_conditions.append( |
| 987 | MusehubProposalComment.created_at > datetime.fromisoformat(cursor) |
| 988 | ) |
| 989 | |
| 990 | rows = list( |
| 991 | ( |
| 992 | await session.execute( |
| 993 | select(MusehubProposalComment) |
| 994 | .where(*data_conditions) |
| 995 | .order_by(MusehubProposalComment.created_at) |
| 996 | .limit(limit + 1) |
| 997 | ) |
| 998 | ).scalars() |
| 999 | ) |
| 1000 | |
| 1001 | next_cursor: str | None = None |
| 1002 | if len(rows) == limit + 1: |
| 1003 | next_cursor = rows[limit - 1].created_at.isoformat() |
| 1004 | rows = rows[:limit] |
| 1005 | |
| 1006 | # Build id → response map first; attach replies in a second pass. |
| 1007 | top_level: list[ProposalCommentResponse] = [] |
| 1008 | by_id: _CommentMap = {} |
| 1009 | for row in rows: |
| 1010 | resp = _to_comment_response(row) |
| 1011 | by_id[row.comment_id] = resp |
| 1012 | if row.parent_comment_id is None: |
| 1013 | top_level.append(resp) |
| 1014 | |
| 1015 | for row in rows: |
| 1016 | if row.parent_comment_id is not None: |
| 1017 | parent = by_id.get(row.parent_comment_id) |
| 1018 | if parent is not None: |
| 1019 | parent.replies.append(by_id[row.comment_id]) |
| 1020 | |
| 1021 | return ProposalCommentListResponse(comments=top_level, total=total, next_cursor=next_cursor) |
| 1022 | |
| 1023 | |
| 1024 | # --------------------------------------------------------------------------- |
| 1025 | # Proposal reviews (reviewer assignment + approval workflow) |
| 1026 | # --------------------------------------------------------------------------- |
| 1027 | |
| 1028 | |
| 1029 | def _to_review_response(row: MusehubProposalReview) -> ProposalReviewResponse: |
| 1030 | return ProposalReviewResponse( |
| 1031 | id=row.review_id, |
| 1032 | proposal_id=row.proposal_id, |
| 1033 | reviewer_username=row.reviewer_username, |
| 1034 | state=row.state, |
| 1035 | body=row.body, |
| 1036 | submitted_at=row.submitted_at, |
| 1037 | created_at=row.created_at, |
| 1038 | ) |
| 1039 | |
| 1040 | |
| 1041 | async def _assert_proposal_exists(session: AsyncSession, repo_id: str, proposal_id: str) -> None: |
| 1042 | """Raise ``ValueError`` if the proposal does not exist in the given repo.""" |
| 1043 | stmt = select(MusehubProposal).where( |
| 1044 | MusehubProposal.proposal_id == proposal_id, |
| 1045 | MusehubProposal.repo_id == repo_id, |
| 1046 | ) |
| 1047 | proposal = (await session.execute(stmt)).scalar_one_or_none() |
| 1048 | if proposal is None: |
| 1049 | raise ValueError(f"Proposal {proposal_id} not found in repo {repo_id}") |
| 1050 | |
| 1051 | |
| 1052 | async def request_reviewers( |
| 1053 | session: AsyncSession, |
| 1054 | *, |
| 1055 | repo_id: str, |
| 1056 | proposal_id: str, |
| 1057 | reviewers: list[str], |
| 1058 | ) -> ProposalReviewListResponse: |
| 1059 | """Add reviewer assignments to a proposal, creating a ``pending`` row for each. |
| 1060 | |
| 1061 | Idempotent: if a reviewer already has a row (in any state), the existing row |
| 1062 | is left unchanged so a submitted approval is never reset by a re-request. |
| 1063 | |
| 1064 | Raises ``ValueError`` if the proposal does not exist in the repo. |
| 1065 | |
| 1066 | Returns the full updated review list for the proposal. |
| 1067 | """ |
| 1068 | await _assert_proposal_exists(session, repo_id, proposal_id) |
| 1069 | |
| 1070 | for username in reviewers: |
| 1071 | existing_stmt = select(MusehubProposalReview).where( |
| 1072 | MusehubProposalReview.proposal_id == proposal_id, |
| 1073 | MusehubProposalReview.reviewer_username == username, |
| 1074 | ) |
| 1075 | existing = (await session.execute(existing_stmt)).scalar_one_or_none() |
| 1076 | if existing is None: |
| 1077 | now = _utc_now() |
| 1078 | identity_stmt = select(MusehubIdentity.identity_id).where( |
| 1079 | MusehubIdentity.handle == username |
| 1080 | ) |
| 1081 | reviewer_identity_id = (await session.execute(identity_stmt)).scalar_one_or_none() or username |
| 1082 | review = MusehubProposalReview( |
| 1083 | review_id=compute_review_id(proposal_id, reviewer_identity_id, now.isoformat()), |
| 1084 | proposal_id=proposal_id, |
| 1085 | reviewer_username=username, |
| 1086 | state="pending", |
| 1087 | created_at=now, |
| 1088 | ) |
| 1089 | session.add(review) |
| 1090 | logger.info("✅ Requested review from '%s' on proposal %s", username, proposal_id) |
| 1091 | |
| 1092 | await session.flush() |
| 1093 | return await list_reviews(session, repo_id=repo_id, proposal_id=proposal_id) |
| 1094 | |
| 1095 | |
| 1096 | async def remove_reviewer( |
| 1097 | session: AsyncSession, |
| 1098 | *, |
| 1099 | repo_id: str, |
| 1100 | proposal_id: str, |
| 1101 | username: str, |
| 1102 | ) -> ProposalReviewListResponse: |
| 1103 | """Remove a pending review request for ``username`` on a proposal. |
| 1104 | |
| 1105 | Only ``pending`` rows may be removed — submitted reviews are immutable to |
| 1106 | preserve the audit trail. |
| 1107 | |
| 1108 | Raises ``ValueError`` if the proposal does not exist, the reviewer was never |
| 1109 | requested, or the reviewer has already submitted a non-pending review. |
| 1110 | |
| 1111 | Returns the updated review list. |
| 1112 | """ |
| 1113 | await _assert_proposal_exists(session, repo_id, proposal_id) |
| 1114 | |
| 1115 | stmt = select(MusehubProposalReview).where( |
| 1116 | MusehubProposalReview.proposal_id == proposal_id, |
| 1117 | MusehubProposalReview.reviewer_username == username, |
| 1118 | ) |
| 1119 | row = (await session.execute(stmt)).scalar_one_or_none() |
| 1120 | if row is None: |
| 1121 | raise ValueError(f"Reviewer '{username}' was not requested on proposal {proposal_id}") |
| 1122 | if row.state != "pending": |
| 1123 | raise ValueError( |
| 1124 | f"Cannot remove reviewer '{username}': review already submitted (state={row.state})" |
| 1125 | ) |
| 1126 | |
| 1127 | await session.delete(row) |
| 1128 | await session.flush() |
| 1129 | logger.info("✅ Removed review request for '%s' from proposal %s", username, proposal_id) |
| 1130 | return await list_reviews(session, repo_id=repo_id, proposal_id=proposal_id) |
| 1131 | |
| 1132 | |
| 1133 | async def list_reviews( |
| 1134 | session: AsyncSession, |
| 1135 | *, |
| 1136 | repo_id: str, |
| 1137 | proposal_id: str, |
| 1138 | state: str | None = None, |
| 1139 | cursor: str | None = None, |
| 1140 | limit: int = 20, |
| 1141 | ) -> ProposalReviewListResponse: |
| 1142 | """Return reviews for a proposal with cursor-based keyset pagination. |
| 1143 | |
| 1144 | ``state`` may be one of ``pending``, ``approved``, ``changes_requested``, |
| 1145 | or ``dismissed``. When ``None``, all reviews are returned. |
| 1146 | Results are ordered by ``created_at`` ascending. |
| 1147 | |
| 1148 | ``cursor`` is the ISO 8601 ``created_at`` of the last seen review |
| 1149 | (opaque to callers — pass ``nextCursor`` from a previous response |
| 1150 | verbatim). Omit to start from the beginning. |
| 1151 | |
| 1152 | Raises ``ValueError`` if the proposal does not exist in the repo. |
| 1153 | """ |
| 1154 | await _assert_proposal_exists(session, repo_id, proposal_id) |
| 1155 | |
| 1156 | conditions = [MusehubProposalReview.proposal_id == proposal_id] |
| 1157 | if state is not None: |
| 1158 | conditions.append(MusehubProposalReview.state == state) |
| 1159 | |
| 1160 | count_stmt = select(func.count(MusehubProposalReview.review_id)).where(*conditions) |
| 1161 | total: int = (await session.execute(count_stmt)).scalar_one() |
| 1162 | |
| 1163 | data_conditions = list(conditions) |
| 1164 | if cursor is not None: |
| 1165 | data_conditions.append( |
| 1166 | MusehubProposalReview.created_at > datetime.fromisoformat(cursor) |
| 1167 | ) |
| 1168 | |
| 1169 | rows = list( |
| 1170 | ( |
| 1171 | await session.execute( |
| 1172 | select(MusehubProposalReview) |
| 1173 | .where(*data_conditions) |
| 1174 | .order_by(MusehubProposalReview.created_at) |
| 1175 | .limit(limit + 1) |
| 1176 | ) |
| 1177 | ).scalars() |
| 1178 | ) |
| 1179 | |
| 1180 | next_cursor: str | None = None |
| 1181 | if len(rows) == limit + 1: |
| 1182 | next_cursor = rows[limit - 1].created_at.isoformat() |
| 1183 | rows = rows[:limit] |
| 1184 | |
| 1185 | return ProposalReviewListResponse( |
| 1186 | reviews=[_to_review_response(r) for r in rows], |
| 1187 | total=total, |
| 1188 | next_cursor=next_cursor, |
| 1189 | ) |
| 1190 | |
| 1191 | |
| 1192 | async def submit_review( |
| 1193 | session: AsyncSession, |
| 1194 | *, |
| 1195 | repo_id: str, |
| 1196 | proposal_id: str, |
| 1197 | reviewer_username: str, |
| 1198 | reviewer_identity_id: str = "", |
| 1199 | event: str, |
| 1200 | body: str = "", |
| 1201 | ) -> ProposalReviewResponse: |
| 1202 | """Submit or update a formal review for ``reviewer_username`` on a proposal. |
| 1203 | |
| 1204 | ``event`` maps to a new state: |
| 1205 | - ``approve`` → ``approved`` |
| 1206 | - ``request_changes`` → ``changes_requested`` |
| 1207 | - ``comment`` → ``pending`` (body-only, no verdict change) |
| 1208 | |
| 1209 | If an existing row for this reviewer already exists, it is updated in-place. |
| 1210 | If no row exists (reviewer was not formally requested), a new row is created |
| 1211 | so ad-hoc reviews are allowed. |
| 1212 | |
| 1213 | Raises ``ValueError`` if the proposal does not exist in the repo. |
| 1214 | """ |
| 1215 | await _assert_proposal_exists(session, repo_id, proposal_id) |
| 1216 | |
| 1217 | _EVENT_TO_STATE: StrDict = { |
| 1218 | "approve": "approved", |
| 1219 | "request_changes": "changes_requested", |
| 1220 | "comment": "pending", |
| 1221 | } |
| 1222 | new_state = _EVENT_TO_STATE[event] |
| 1223 | |
| 1224 | stmt = select(MusehubProposalReview).where( |
| 1225 | MusehubProposalReview.proposal_id == proposal_id, |
| 1226 | MusehubProposalReview.reviewer_username == reviewer_username, |
| 1227 | ) |
| 1228 | row = (await session.execute(stmt)).scalar_one_or_none() |
| 1229 | |
| 1230 | now = _utc_now() |
| 1231 | if row is None: |
| 1232 | row = MusehubProposalReview( |
| 1233 | review_id=compute_review_id(proposal_id, reviewer_identity_id, now.isoformat()), |
| 1234 | proposal_id=proposal_id, |
| 1235 | reviewer_username=reviewer_username, |
| 1236 | state=new_state, |
| 1237 | body=body or None, |
| 1238 | submitted_at=now if event != "comment" else None, |
| 1239 | created_at=now, |
| 1240 | ) |
| 1241 | session.add(row) |
| 1242 | else: |
| 1243 | row.state = new_state |
| 1244 | row.body = body or None |
| 1245 | row.submitted_at = now if event != "comment" else row.submitted_at |
| 1246 | |
| 1247 | await session.flush() |
| 1248 | await session.refresh(row) |
| 1249 | logger.info( |
| 1250 | "✅ Review submitted by '%s' on proposal %s: event=%s state=%s", |
| 1251 | reviewer_username, |
| 1252 | proposal_id, |
| 1253 | event, |
| 1254 | new_state, |
| 1255 | ) |
| 1256 | return _to_review_response(row) |
| 1257 | |
| 1258 | # ── Proposal list enrichment ────────────────────────────────────────────────── |
| 1259 | |
| 1260 | _RISK_BAND_THRESHOLDS: list[tuple[float, str]] = [ |
| 1261 | (0.75, "critical"), |
| 1262 | (0.50, "high"), |
| 1263 | (0.25, "medium"), |
| 1264 | (0.01, "low"), |
| 1265 | ] |
| 1266 | |
| 1267 | |
| 1268 | def _score_to_band(score: float) -> str: |
| 1269 | """Map a [0.0, 1.0] risk score to a human-readable band label. |
| 1270 | |
| 1271 | Thresholds: |
| 1272 | ≥ 0.75 → "critical" |
| 1273 | ≥ 0.50 → "high" |
| 1274 | ≥ 0.25 → "medium" |
| 1275 | > 0.0 → "low" |
| 1276 | 0.0 → "none" |
| 1277 | """ |
| 1278 | for threshold, band in _RISK_BAND_THRESHOLDS: |
| 1279 | if score >= threshold: |
| 1280 | return band |
| 1281 | return "none" |
| 1282 | |
| 1283 | |
| 1284 | # Default required-approvals when merge_conditions is null. |
| 1285 | _DEFAULT_REQUIRED_APPROVALS = 2 |
| 1286 | |
| 1287 | # Domain weight map used for aggregate risk score. Unknown domains default to 1.0. |
| 1288 | _DOMAIN_WEIGHTS: dict[str, float] = { |
| 1289 | "code": 1.2, |
| 1290 | "midi": 1.0, |
| 1291 | "stems": 1.0, |
| 1292 | "prose": 0.8, |
| 1293 | "pay": 1.5, |
| 1294 | } |
| 1295 | |
| 1296 | |
| 1297 | class _ProposalPrefetch: |
| 1298 | """Holds pre-fetched batch data for a page of proposals. |
| 1299 | |
| 1300 | All DB reads for an entire page happen once in |
| 1301 | ``enrich_proposal_list_batch``; each ``enrich_proposal_list_entry`` call |
| 1302 | consults these in-memory maps — zero additional DB I/O per row. |
| 1303 | """ |
| 1304 | |
| 1305 | def __init__( |
| 1306 | self, |
| 1307 | *, |
| 1308 | reviews_by_proposal: dict[str, list[MusehubProposalReview]], |
| 1309 | author_types: dict[str, str], |
| 1310 | dag: ProposalDag | None = None, |
| 1311 | conflict_counts: dict[str, int | None] | None = None, |
| 1312 | ) -> None: |
| 1313 | self.reviews_by_proposal = reviews_by_proposal |
| 1314 | self.author_types = author_types |
| 1315 | self.dag: ProposalDag = dag or ProposalDag() |
| 1316 | # proposal_id → conflict_scan.result["conflict_count"]; None if not run |
| 1317 | self.conflict_counts: dict[str, int | None] = conflict_counts or {} |
| 1318 | |
| 1319 | |
| 1320 | async def _prefetch_for_batch( |
| 1321 | proposals: list[MusehubProposal], |
| 1322 | session: AsyncSession, |
| 1323 | ) -> _ProposalPrefetch: |
| 1324 | """Run the batch pre-fetch queries for a page of proposals. |
| 1325 | |
| 1326 | Issues exactly two DB queries regardless of page size: |
| 1327 | 1. All reviews for every proposal in the page. |
| 1328 | 2. Identity types for every author in the page. |
| 1329 | |
| 1330 | Args: |
| 1331 | proposals: ORM rows for the current page. |
| 1332 | session: Shared async session. |
| 1333 | |
| 1334 | Returns: |
| 1335 | ``_ProposalPrefetch`` with maps keyed by proposal_id / author handle. |
| 1336 | """ |
| 1337 | proposal_ids = [p.proposal_id for p in proposals] |
| 1338 | author_handles = list({p.author for p in proposals if p.author}) |
| 1339 | |
| 1340 | # Query 1 — reviews |
| 1341 | reviews_by_proposal: dict[str, list[MusehubProposalReview]] = {pid: [] for pid in proposal_ids} |
| 1342 | if proposal_ids: |
| 1343 | review_rows = list( |
| 1344 | ( |
| 1345 | await session.execute( |
| 1346 | select(MusehubProposalReview).where( |
| 1347 | MusehubProposalReview.proposal_id.in_(proposal_ids) |
| 1348 | ) |
| 1349 | ) |
| 1350 | ).scalars() |
| 1351 | ) |
| 1352 | for row in review_rows: |
| 1353 | reviews_by_proposal[row.proposal_id].append(row) |
| 1354 | |
| 1355 | # Query 2 — identity types |
| 1356 | author_types: dict[str, str] = {} |
| 1357 | if author_handles: |
| 1358 | identity_rows = list( |
| 1359 | ( |
| 1360 | await session.execute( |
| 1361 | select(MusehubIdentity.handle, MusehubIdentity.identity_type).where( |
| 1362 | MusehubIdentity.handle.in_(author_handles) |
| 1363 | ) |
| 1364 | ) |
| 1365 | ).all() |
| 1366 | ) |
| 1367 | for handle, itype in identity_rows: |
| 1368 | author_types[handle] = itype |
| 1369 | |
| 1370 | # Query 3 — dependency DAG (partial, scoped to this page + neighbours) |
| 1371 | dag = await load_dag_for_proposals(session, proposal_ids) |
| 1372 | |
| 1373 | # Query 4 — latest conflict_scan simulation per proposal (for list summary) |
| 1374 | conflict_counts: dict[str, int | None] = {pid: None for pid in proposal_ids} |
| 1375 | if proposal_ids: |
| 1376 | sim_rows = list( |
| 1377 | ( |
| 1378 | await session.execute( |
| 1379 | select( |
| 1380 | MusehubProposalSimulation.proposal_id, |
| 1381 | MusehubProposalSimulation.result, |
| 1382 | ).where( |
| 1383 | MusehubProposalSimulation.proposal_id.in_(proposal_ids), |
| 1384 | MusehubProposalSimulation.simulation_type == "conflict_scan", |
| 1385 | ) |
| 1386 | ) |
| 1387 | ).all() |
| 1388 | ) |
| 1389 | for pid, result_json in sim_rows: |
| 1390 | if isinstance(result_json, dict): |
| 1391 | conflict_counts[pid] = result_json.get("conflict_count") |
| 1392 | |
| 1393 | return _ProposalPrefetch( |
| 1394 | reviews_by_proposal=reviews_by_proposal, |
| 1395 | author_types=author_types, |
| 1396 | dag=dag, |
| 1397 | conflict_counts=conflict_counts, |
| 1398 | ) |
| 1399 | |
| 1400 | |
| 1401 | def _enrich_one( |
| 1402 | proposal: MusehubProposal, |
| 1403 | prefetch: _ProposalPrefetch, |
| 1404 | ) -> ProposalListEntry: |
| 1405 | """Compute all display-facing fields for a single proposal list row. |
| 1406 | |
| 1407 | This is the single source of truth for what the proposals list view renders |
| 1408 | per row. It does not issue any DB queries — all needed data comes from |
| 1409 | ``prefetch``, which is populated by ``_prefetch_for_batch`` before this |
| 1410 | function is called. |
| 1411 | |
| 1412 | Computed fields (all server-side): |
| 1413 | - active_domains: domains with non-zero risk_score |
| 1414 | - domain_risk / domain_risk_band: derived from proposal.risk_score |
| 1415 | (currently a single code-domain score; extended as more domains land) |
| 1416 | - aggregate_risk_score: weighted mean across active domains |
| 1417 | - aggregate_risk_band: band for the aggregate score |
| 1418 | - approval_count / domains_approved / domains_pending_review: |
| 1419 | derived from pre-fetched reviews |
| 1420 | - all_merge_conditions_met: approval_count >= required_approvals |
| 1421 | and breakage_count == 0 |
| 1422 | - author_type: resolved from pre-fetched MusehubIdentity rows |
| 1423 | |
| 1424 | Performance contract: |
| 1425 | Zero DB I/O. All reads come from the ``prefetch`` maps. Typical |
| 1426 | wall time: < 1ms per row on warm prefetch data. |
| 1427 | |
| 1428 | Args: |
| 1429 | proposal: ORM row for this proposal. |
| 1430 | prefetch: Pre-fetched batch data from ``_prefetch_for_batch``. |
| 1431 | |
| 1432 | Returns: |
| 1433 | ``ProposalListEntry`` with all fields populated. |
| 1434 | |
| 1435 | Raises: |
| 1436 | ValueError: If ``proposal.risk_score`` is outside ``[0.0, 1.0]``. |
| 1437 | """ |
| 1438 | pid = proposal.proposal_id |
| 1439 | |
| 1440 | # ── Risk ───────────────────────────────────────────────────────────────── |
| 1441 | # Use dimensional_risk dict (Phase 1 ORM columns) when present; fall back |
| 1442 | # to the scalar risk_score as code-domain risk for backwards compatibility. |
| 1443 | raw_dimensional = dict(getattr(proposal, "dimensional_risk", None) or {}) |
| 1444 | if raw_dimensional: |
| 1445 | domain_risk = {d: float(v) for d, v in raw_dimensional.items() if float(v) > 0.0} |
| 1446 | else: |
| 1447 | code_risk = float(proposal.risk_score or 0.0) |
| 1448 | if not (0.0 <= code_risk <= 1.0): |
| 1449 | raise ValueError(f"proposal {pid}: risk_score {code_risk!r} out of [0, 1]") |
| 1450 | domain_risk = {"code": code_risk} if code_risk > 0.0 else {} |
| 1451 | active_domains = list(domain_risk.keys()) |
| 1452 | domain_risk_band = {d: _score_to_band(v) for d, v in domain_risk.items()} |
| 1453 | |
| 1454 | # Weighted aggregate |
| 1455 | if domain_risk: |
| 1456 | total_weight = sum(_DOMAIN_WEIGHTS.get(d, 1.0) for d in domain_risk) |
| 1457 | aggregate_risk_score = sum( |
| 1458 | v * _DOMAIN_WEIGHTS.get(d, 1.0) for d, v in domain_risk.items() |
| 1459 | ) / total_weight |
| 1460 | else: |
| 1461 | aggregate_risk_score = 0.0 |
| 1462 | aggregate_risk_band = _score_to_band(aggregate_risk_score) |
| 1463 | |
| 1464 | # ── Dependency position ─────────────────────────────────────────────────── |
| 1465 | dag = prefetch.dag |
| 1466 | dep_blocked_by = blocked_by_numbers(dag, pid) |
| 1467 | dep_blocks = blocks_numbers(dag, pid) |
| 1468 | dep_is_blocked = is_blocked(dag, pid) |
| 1469 | mc_raw = proposal.merge_conditions or {} |
| 1470 | require_dep_merged: bool = mc_raw.get("require_dependency_merged", True) |
| 1471 | deps_satisfied = (not require_dep_merged) or (not dep_is_blocked) |
| 1472 | |
| 1473 | # ── Reviews ─────────────────────────────────────────────────────────────── |
| 1474 | reviews = prefetch.reviews_by_proposal.get(pid, []) |
| 1475 | approved_reviews = [r for r in reviews if r.state == "approved"] |
| 1476 | approval_count = len(approved_reviews) |
| 1477 | required_approvals = _DEFAULT_REQUIRED_APPROVALS |
| 1478 | domains_approved = ["code"] if approval_count > 0 and "code" in active_domains else [] |
| 1479 | domains_pending_review = [d for d in active_domains if d not in domains_approved] |
| 1480 | all_merge_conditions_met = ( |
| 1481 | approval_count >= required_approvals |
| 1482 | and proposal.breakage_count == 0 |
| 1483 | and deps_satisfied |
| 1484 | ) |
| 1485 | |
| 1486 | # ── Author type ─────────────────────────────────────────────────────────── |
| 1487 | author_type = prefetch.author_types.get(proposal.author, "human") |
| 1488 | agent_model: str | None = getattr(proposal, "agent_model", None) |
| 1489 | agent_spawned_by: str | None = getattr(proposal, "agent_spawned_by", None) |
| 1490 | |
| 1491 | # ── Symbol preview ──────────────────────────────────────────────────────── |
| 1492 | touched = list(proposal.touched_symbols or []) |
| 1493 | touched_symbols_preview = touched[:3] |
| 1494 | |
| 1495 | return ProposalListEntry( |
| 1496 | proposal_id=proposal.proposal_id, |
| 1497 | proposal_number=proposal.proposal_number, |
| 1498 | title=(proposal.title[:80] + "…") if len(proposal.title) > 80 else proposal.title, |
| 1499 | state=proposal.state, |
| 1500 | proposal_type=getattr(proposal, "proposal_type", "state_merge"), |
| 1501 | from_branch=proposal.from_branch, |
| 1502 | to_branch=proposal.to_branch, |
| 1503 | author=proposal.author, |
| 1504 | author_type=author_type, |
| 1505 | created_at=proposal.created_at, |
| 1506 | merged_at=proposal.merged_at, |
| 1507 | is_draft=getattr(proposal, "is_draft", proposal.state == "drafting"), |
| 1508 | active_domains=active_domains, |
| 1509 | domain_risk=domain_risk, |
| 1510 | domain_risk_band=domain_risk_band, |
| 1511 | aggregate_risk_score=round(aggregate_risk_score, 4), |
| 1512 | aggregate_risk_band=aggregate_risk_band, |
| 1513 | approval_count=approval_count, |
| 1514 | required_approvals=required_approvals, |
| 1515 | domains_approved=domains_approved, |
| 1516 | domains_pending_review=domains_pending_review, |
| 1517 | all_merge_conditions_met=all_merge_conditions_met, |
| 1518 | blocked_by=dep_blocked_by, |
| 1519 | blocks=dep_blocks, |
| 1520 | is_blocked=dep_is_blocked, |
| 1521 | symbols_changed=proposal.symbols_changed, |
| 1522 | breakage_count=proposal.breakage_count, |
| 1523 | test_gap_count=proposal.test_gap_count, |
| 1524 | touched_symbols_preview=touched_symbols_preview, |
| 1525 | midi_tracks_changed=getattr(proposal, "midi_tracks_changed", 0), |
| 1526 | midi_notes_delta=getattr(proposal, "midi_notes_delta", 0), |
| 1527 | harmonic_tension_delta=getattr(proposal, "harmonic_tension_delta", None), |
| 1528 | payment_claim_count=getattr(proposal, "payment_claim_count", 0), |
| 1529 | payment_ledger_delta_nano=getattr(proposal, "payment_ledger_delta_nano", 0), |
| 1530 | payment_avax_address=getattr(proposal, "payment_avax_address", None), |
| 1531 | payment_settling=proposal.state == "settling" and "pay" in active_domains, |
| 1532 | agent_model=agent_model, |
| 1533 | agent_spawned_by=agent_spawned_by, |
| 1534 | merge_strategy=getattr(proposal, "merge_strategy", "state_overlay") or "state_overlay", |
| 1535 | simulation_conflict_count=prefetch.conflict_counts.get(pid), |
| 1536 | ) |
| 1537 | |
| 1538 | |
| 1539 | async def enrich_proposal_list_entry( |
| 1540 | proposal: MusehubProposal, |
| 1541 | session: AsyncSession, |
| 1542 | ) -> ProposalListEntry: |
| 1543 | """Compute all display-facing fields for a single proposal list row. |
| 1544 | |
| 1545 | Convenience wrapper around ``_enrich_one`` for callers that need to enrich |
| 1546 | a single proposal without a pre-existing batch context. Issues its own |
| 1547 | prefetch queries (2 DB round-trips). |
| 1548 | |
| 1549 | For a full page (≥2 proposals) prefer ``enrich_proposal_list_batch`` which |
| 1550 | amortises the prefetch cost across all rows via a single parallel pass. |
| 1551 | |
| 1552 | Args: |
| 1553 | proposal: ORM row for this proposal. Must have ``repo_id`` set. |
| 1554 | session: Async database session. |
| 1555 | |
| 1556 | Returns: |
| 1557 | ``ProposalListEntry`` with all fields populated. |
| 1558 | |
| 1559 | Raises: |
| 1560 | ValueError: If ``proposal.risk_score`` is outside ``[0.0, 1.0]``. |
| 1561 | """ |
| 1562 | prefetch = await _prefetch_for_batch([proposal], session) |
| 1563 | return _enrich_one(proposal, prefetch) |
| 1564 | |
| 1565 | |
| 1566 | async def enrich_proposal_list_batch( |
| 1567 | proposals: list[MusehubProposal], |
| 1568 | session: AsyncSession, |
| 1569 | ) -> list[ProposalListEntry]: |
| 1570 | """Enrich a full page of proposal rows in a single parallel pass. |
| 1571 | |
| 1572 | Pre-fetch strategy: |
| 1573 | Issues exactly 2 DB queries for the entire batch (reviews + identity |
| 1574 | types), then calls ``_enrich_one`` for each row synchronously. |
| 1575 | The synchronous per-row work is CPU-only (no I/O), so no concurrency |
| 1576 | overhead is needed. |
| 1577 | |
| 1578 | Ordering: |
| 1579 | Result list is in the same order as ``proposals``. |
| 1580 | |
| 1581 | Args: |
| 1582 | proposals: ORM rows for the current page (typically ≤ 20). |
| 1583 | session: Async session shared across the batch. |
| 1584 | |
| 1585 | Returns: |
| 1586 | List of ``ProposalListEntry`` in the same order as ``proposals``. |
| 1587 | |
| 1588 | Performance target: < 50ms for 20 proposals (dominated by the 2 DB |
| 1589 | queries; per-row computation is < 0.1ms). |
| 1590 | """ |
| 1591 | if not proposals: |
| 1592 | return [] |
| 1593 | prefetch = await _prefetch_for_batch(proposals, session) |
| 1594 | return [_enrich_one(p, prefetch) for p in proposals] |
| 1595 | |
| 1596 | |
| 1597 | async def get_domain_heat( |
| 1598 | repo_id: str, |
| 1599 | state: str, |
| 1600 | session: AsyncSession, |
| 1601 | ) -> DomainHeatResponse: |
| 1602 | """Return per-domain proposal counts and average risk for the heat bar. |
| 1603 | |
| 1604 | Runs a single aggregation query against ``musehub_proposals`` filtered by |
| 1605 | ``repo_id`` and ``state``. The heat bar currently reflects the code domain |
| 1606 | only (the only domain with populated risk scores at this phase); additional |
| 1607 | domains are added as multi-domain risk rows land in Phase 2. |
| 1608 | |
| 1609 | ``avg_risk`` is the arithmetic mean of non-zero ``risk_score`` values for |
| 1610 | proposals in the given state. Domains with zero matching proposals are |
| 1611 | omitted from the response dict. |
| 1612 | |
| 1613 | Args: |
| 1614 | repo_id: Repository to query. |
| 1615 | state: Proposal state filter (e.g. ``"open"``). Pass ``"open"`` for |
| 1616 | the standard heat bar view. Pass ``"all"`` to skip the state |
| 1617 | filter entirely. |
| 1618 | session: Async session. |
| 1619 | |
| 1620 | Returns: |
| 1621 | ``DomainHeatResponse`` with ``domains`` dict and ``total_open`` count. |
| 1622 | |
| 1623 | Performance target: < 20ms (single aggregation query, no per-row work). |
| 1624 | """ |
| 1625 | conditions = [MusehubProposal.repo_id == repo_id] |
| 1626 | if state != "all": |
| 1627 | conditions.append(MusehubProposal.state == state) |
| 1628 | |
| 1629 | total: int = ( |
| 1630 | await session.execute( |
| 1631 | select(func.count(MusehubProposal.proposal_id)).where(*conditions) |
| 1632 | ) |
| 1633 | ).scalar_one() |
| 1634 | |
| 1635 | # Code domain: all proposals in this repo/state (code is the only domain for now). |
| 1636 | # avg_risk computed from non-null, non-zero risk_score values only. |
| 1637 | risk_rows = list( |
| 1638 | ( |
| 1639 | await session.execute( |
| 1640 | select(MusehubProposal.risk_score).where( |
| 1641 | *conditions, |
| 1642 | MusehubProposal.risk_score.isnot(None), |
| 1643 | MusehubProposal.risk_score > 0.0, |
| 1644 | ) |
| 1645 | ) |
| 1646 | ).scalars() |
| 1647 | ) |
| 1648 | avg_risk = round(sum(risk_rows) / len(risk_rows), 4) if risk_rows else 0.0 |
| 1649 | |
| 1650 | # All known domains — code count = total (single-domain repos); others = 0. |
| 1651 | # Multi-domain heat will populate midi/stems/pay/prose when those proposals land. |
| 1652 | domains: dict[str, DomainHeatEntry] = { |
| 1653 | "code": DomainHeatEntry(count=total, avg_risk=avg_risk), |
| 1654 | "midi": DomainHeatEntry(count=0, avg_risk=0.0), |
| 1655 | "stems": DomainHeatEntry(count=0, avg_risk=0.0), |
| 1656 | "pay": DomainHeatEntry(count=0, avg_risk=0.0), |
| 1657 | "prose": DomainHeatEntry(count=0, avg_risk=0.0), |
| 1658 | } |
| 1659 | |
| 1660 | return DomainHeatResponse(domains=domains, total_open=total) |
| 1661 | |
| 1662 | |
| 1663 | async def get_merge_readiness( |
| 1664 | repo_id: str, |
| 1665 | session: AsyncSession, |
| 1666 | ) -> MergeReadinessResponse: |
| 1667 | """Bucket all non-merged proposals into readiness categories. |
| 1668 | |
| 1669 | Categories: |
| 1670 | ready: approval_count >= required threshold AND breakage_count == 0 |
| 1671 | settling: state == 'settling' |
| 1672 | needs_review: not settling, conditions not fully met |
| 1673 | |
| 1674 | Dependency-blocked proposals (``blocked_by`` non-empty) are not yet |
| 1675 | tracked in the DB at this phase; ``blocked`` will always be empty until |
| 1676 | the dependency graph table lands in Phase 2. |
| 1677 | |
| 1678 | Runs in one DB query; no per-row enrichment. |
| 1679 | |
| 1680 | Args: |
| 1681 | repo_id: Repository to query. |
| 1682 | session: Async session. |
| 1683 | |
| 1684 | Returns: |
| 1685 | ``MergeReadinessResponse`` with ``ready``, ``blocked``, ``settling``, |
| 1686 | and ``needs_review`` lists of proposal numbers. |
| 1687 | |
| 1688 | Performance target: < 20ms. |
| 1689 | """ |
| 1690 | rows = list( |
| 1691 | ( |
| 1692 | await session.execute( |
| 1693 | select( |
| 1694 | MusehubProposal.proposal_number, |
| 1695 | MusehubProposal.state, |
| 1696 | MusehubProposal.breakage_count, |
| 1697 | ).where( |
| 1698 | MusehubProposal.repo_id == repo_id, |
| 1699 | MusehubProposal.state.notin_(["merged", "abandoned"]), |
| 1700 | ) |
| 1701 | ) |
| 1702 | ).all() |
| 1703 | ) |
| 1704 | |
| 1705 | # Pre-fetch approval counts in one query |
| 1706 | proposal_numbers_to_check = [r.proposal_number for r in rows] |
| 1707 | if proposal_numbers_to_check: |
| 1708 | approval_counts_rows = list( |
| 1709 | ( |
| 1710 | await session.execute( |
| 1711 | select( |
| 1712 | MusehubProposal.proposal_number, |
| 1713 | func.count(MusehubProposalReview.review_id).label("approved_count"), |
| 1714 | ) |
| 1715 | .join( |
| 1716 | MusehubProposalReview, |
| 1717 | (MusehubProposalReview.proposal_id == MusehubProposal.proposal_id) |
| 1718 | & (MusehubProposalReview.state == "approved"), |
| 1719 | isouter=True, |
| 1720 | ) |
| 1721 | .where( |
| 1722 | MusehubProposal.repo_id == repo_id, |
| 1723 | MusehubProposal.proposal_number.in_(proposal_numbers_to_check), |
| 1724 | ) |
| 1725 | .group_by(MusehubProposal.proposal_number) |
| 1726 | ) |
| 1727 | ).all() |
| 1728 | ) |
| 1729 | approval_by_number: dict[int, int] = {r.proposal_number: r.approved_count for r in approval_counts_rows} |
| 1730 | else: |
| 1731 | approval_by_number = {} |
| 1732 | |
| 1733 | ready: list[int] = [] |
| 1734 | blocked: list[int] = [] |
| 1735 | settling: list[int] = [] |
| 1736 | needs_review: list[int] = [] |
| 1737 | |
| 1738 | for row in rows: |
| 1739 | if row.state == "settling": |
| 1740 | settling.append(row.proposal_number) |
| 1741 | elif ( |
| 1742 | approval_by_number.get(row.proposal_number, 0) >= _DEFAULT_REQUIRED_APPROVALS |
| 1743 | and row.breakage_count == 0 |
| 1744 | ): |
| 1745 | ready.append(row.proposal_number) |
| 1746 | else: |
| 1747 | needs_review.append(row.proposal_number) |
| 1748 | |
| 1749 | return MergeReadinessResponse( |
| 1750 | ready=ready, |
| 1751 | blocked=blocked, |
| 1752 | settling=settling, |
| 1753 | needs_review=needs_review, |
| 1754 | ) |
| 1755 | |
| 1756 | |
| 1757 | # ───────────────────────────────────────────────────────────────────────────── |
| 1758 | # Phase 4 — Simulation Engine |
| 1759 | # ───────────────────────────────────────────────────────────────────────────── |
| 1760 | |
| 1761 | _VALID_SIMULATION_TYPES = {"conflict_scan", "risk_projection", "dependency_order"} |
| 1762 | |
| 1763 | |
| 1764 | def _to_simulation_response(row: MusehubProposalSimulation, *, is_stale: bool) -> SimulationResponse: |
| 1765 | return SimulationResponse( |
| 1766 | simulation_id=row.simulation_id, |
| 1767 | proposal_id=row.proposal_id, |
| 1768 | simulation_type=row.simulation_type, |
| 1769 | result=row.result, |
| 1770 | is_stale=is_stale, |
| 1771 | from_branch_commit_id=row.from_branch_commit_id, |
| 1772 | duration_ms=row.duration_ms, |
| 1773 | created_at=row.created_at, |
| 1774 | expires_at=row.expires_at, |
| 1775 | ) |
| 1776 | |
| 1777 | |
| 1778 | async def _current_from_commit( |
| 1779 | session: AsyncSession, |
| 1780 | repo_id: str, |
| 1781 | from_branch: str, |
| 1782 | ) -> str: |
| 1783 | """Return the current head_commit_id of from_branch, or '' if missing.""" |
| 1784 | branch = await _get_branch(session, repo_id, from_branch) |
| 1785 | return (branch.head_commit_id or "") if branch else "" |
| 1786 | |
| 1787 | |
| 1788 | async def run_simulation( |
| 1789 | session: AsyncSession, |
| 1790 | repo_id: str, |
| 1791 | proposal_id: str, |
| 1792 | simulation_type: str, |
| 1793 | ) -> SimulationResponse: |
| 1794 | """Run a simulation for the proposal and upsert the cached result. |
| 1795 | |
| 1796 | Always recomputes — use get_simulation to read the cache without re-running. |
| 1797 | |
| 1798 | Raises: |
| 1799 | ValueError: Unknown simulation_type or proposal not found. |
| 1800 | """ |
| 1801 | import time |
| 1802 | |
| 1803 | from musehub.services.proposal_simulation import ( |
| 1804 | simulate_conflict_scan, |
| 1805 | simulate_dependency_order, |
| 1806 | simulate_risk_projection, |
| 1807 | ) |
| 1808 | from musehub.services.musehub_snapshot import get_snapshot_manifest |
| 1809 | |
| 1810 | if simulation_type not in _VALID_SIMULATION_TYPES: |
| 1811 | raise ValueError( |
| 1812 | f"Unknown simulation_type '{simulation_type}'. " |
| 1813 | f"Valid types: {sorted(_VALID_SIMULATION_TYPES)}" |
| 1814 | ) |
| 1815 | |
| 1816 | stmt = select(MusehubProposal).where( |
| 1817 | MusehubProposal.repo_id == repo_id, |
| 1818 | MusehubProposal.proposal_id == proposal_id, |
| 1819 | ) |
| 1820 | proposal = (await session.execute(stmt)).scalar_one_or_none() |
| 1821 | if proposal is None: |
| 1822 | raise ValueError(f"Proposal {proposal_id} not found in repo {repo_id}") |
| 1823 | |
| 1824 | from_commit_id = await _current_from_commit(session, repo_id, proposal.from_branch) |
| 1825 | to_b = await _get_branch(session, repo_id, proposal.to_branch) |
| 1826 | from_b = await _get_branch(session, repo_id, proposal.from_branch) |
| 1827 | |
| 1828 | to_manifest: StrDict = {} |
| 1829 | if to_b and to_b.head_commit_id: |
| 1830 | to_head = await session.get(MusehubCommit, to_b.head_commit_id) |
| 1831 | if to_head and to_head.snapshot_id: |
| 1832 | to_manifest = await get_snapshot_manifest(session, to_head.snapshot_id) |
| 1833 | |
| 1834 | from_manifest: StrDict = {} |
| 1835 | if from_b and from_b.head_commit_id: |
| 1836 | from_head = await session.get(MusehubCommit, from_b.head_commit_id) |
| 1837 | if from_head and from_head.snapshot_id: |
| 1838 | from_manifest = await get_snapshot_manifest(session, from_head.snapshot_id) |
| 1839 | |
| 1840 | strategy_name = getattr(proposal, "merge_strategy", "state_overlay") or "state_overlay" |
| 1841 | selective_domains: list[str] | None = getattr(proposal, "selective_domains", None) |
| 1842 | dimensional_risk: dict[str, float] | None = getattr(proposal, "dimensional_risk", None) |
| 1843 | |
| 1844 | ancestor_manifest: StrDict | None = None |
| 1845 | if strategy_name in ("state_weave", "state_rebase", "domain_selective", "phased"): |
| 1846 | ancestor_manifest = await _resolve_ancestor_manifest( |
| 1847 | session, repo_id, proposal.from_branch, proposal.to_branch |
| 1848 | ) |
| 1849 | |
| 1850 | t0 = time.monotonic() |
| 1851 | if simulation_type == "conflict_scan": |
| 1852 | result_payload = simulate_conflict_scan( |
| 1853 | to_manifest, |
| 1854 | from_manifest, |
| 1855 | ancestor_manifest=ancestor_manifest, |
| 1856 | strategy=strategy_name, |
| 1857 | selective_domains=selective_domains, |
| 1858 | ) |
| 1859 | elif simulation_type == "risk_projection": |
| 1860 | result_payload = simulate_risk_projection( |
| 1861 | to_manifest, |
| 1862 | from_manifest, |
| 1863 | ancestor_manifest=ancestor_manifest, |
| 1864 | current_dimensional_risk=dimensional_risk, |
| 1865 | strategy=strategy_name, |
| 1866 | selective_domains=selective_domains, |
| 1867 | ) |
| 1868 | else: # dependency_order |
| 1869 | dag = await load_dag_for_proposals(session, [proposal_id]) |
| 1870 | result_payload = simulate_dependency_order(dag) |
| 1871 | |
| 1872 | duration_ms = int((time.monotonic() - t0) * 1000) |
| 1873 | |
| 1874 | sim_id = compute_simulation_id(proposal_id, simulation_type, from_commit_id) |
| 1875 | now = _utc_now() |
| 1876 | |
| 1877 | # Upsert: update on conflict (same proposal_id + simulation_type) |
| 1878 | existing_stmt = select(MusehubProposalSimulation).where( |
| 1879 | MusehubProposalSimulation.proposal_id == proposal_id, |
| 1880 | MusehubProposalSimulation.simulation_type == simulation_type, |
| 1881 | ) |
| 1882 | existing = (await session.execute(existing_stmt)).scalar_one_or_none() |
| 1883 | |
| 1884 | if existing is not None: |
| 1885 | existing.simulation_id = sim_id |
| 1886 | existing.from_branch_commit_id = from_commit_id |
| 1887 | existing.result = result_payload |
| 1888 | existing.duration_ms = duration_ms |
| 1889 | existing.created_at = now |
| 1890 | row = existing |
| 1891 | else: |
| 1892 | row = MusehubProposalSimulation( |
| 1893 | simulation_id=sim_id, |
| 1894 | proposal_id=proposal_id, |
| 1895 | simulation_type=simulation_type, |
| 1896 | from_branch_commit_id=from_commit_id, |
| 1897 | result=result_payload, |
| 1898 | duration_ms=duration_ms, |
| 1899 | created_at=now, |
| 1900 | ) |
| 1901 | session.add(row) |
| 1902 | |
| 1903 | await session.flush() |
| 1904 | |
| 1905 | logger.info( |
| 1906 | "🔬 Simulation %s for proposal %s (%dms)", |
| 1907 | simulation_type, proposal_id, duration_ms |
| 1908 | ) |
| 1909 | return _to_simulation_response(row, is_stale=False) |
| 1910 | |
| 1911 | |
| 1912 | async def get_simulation( |
| 1913 | session: AsyncSession, |
| 1914 | repo_id: str, |
| 1915 | proposal_id: str, |
| 1916 | simulation_type: str, |
| 1917 | ) -> SimulationResponse | None: |
| 1918 | """Return the cached simulation result, or None if never run. |
| 1919 | |
| 1920 | Sets ``is_stale=True`` when the from_branch has advanced since the |
| 1921 | simulation was last run. Does NOT re-run; call run_simulation for that. |
| 1922 | """ |
| 1923 | if simulation_type not in _VALID_SIMULATION_TYPES: |
| 1924 | raise ValueError( |
| 1925 | f"Unknown simulation_type '{simulation_type}'. " |
| 1926 | f"Valid types: {sorted(_VALID_SIMULATION_TYPES)}" |
| 1927 | ) |
| 1928 | |
| 1929 | stmt = select(MusehubProposalSimulation).where( |
| 1930 | MusehubProposalSimulation.proposal_id == proposal_id, |
| 1931 | MusehubProposalSimulation.simulation_type == simulation_type, |
| 1932 | ) |
| 1933 | row = (await session.execute(stmt)).scalar_one_or_none() |
| 1934 | if row is None: |
| 1935 | return None |
| 1936 | |
| 1937 | # Staleness check: compare stored commit ID with current branch tip |
| 1938 | proposal_stmt = select(MusehubProposal.from_branch).where( |
| 1939 | MusehubProposal.proposal_id == proposal_id, |
| 1940 | MusehubProposal.repo_id == repo_id, |
| 1941 | ) |
| 1942 | from_branch = (await session.execute(proposal_stmt)).scalar_one_or_none() |
| 1943 | is_stale = False |
| 1944 | if from_branch: |
| 1945 | current_commit = await _current_from_commit(session, repo_id, from_branch) |
| 1946 | is_stale = bool(current_commit) and current_commit != row.from_branch_commit_id |
| 1947 | |
| 1948 | return _to_simulation_response(row, is_stale=is_stale) |
| 1949 | |
| 1950 | |
| 1951 | async def list_simulations( |
| 1952 | session: AsyncSession, |
| 1953 | repo_id: str, |
| 1954 | proposal_id: str, |
| 1955 | ) -> SimulationListResponse: |
| 1956 | """Return all cached simulations for a proposal. |
| 1957 | |
| 1958 | Includes staleness flags. Never re-runs. |
| 1959 | """ |
| 1960 | # Resolve from_branch once for staleness checks |
| 1961 | proposal_stmt = select(MusehubProposal.from_branch).where( |
| 1962 | MusehubProposal.proposal_id == proposal_id, |
| 1963 | MusehubProposal.repo_id == repo_id, |
| 1964 | ) |
| 1965 | from_branch = (await session.execute(proposal_stmt)).scalar_one_or_none() |
| 1966 | current_commit = "" |
| 1967 | if from_branch: |
| 1968 | current_commit = await _current_from_commit(session, repo_id, from_branch) |
| 1969 | |
| 1970 | rows_stmt = select(MusehubProposalSimulation).where( |
| 1971 | MusehubProposalSimulation.proposal_id == proposal_id, |
| 1972 | ) |
| 1973 | rows = list((await session.execute(rows_stmt)).scalars().all()) |
| 1974 | |
| 1975 | responses = [ |
| 1976 | _to_simulation_response( |
| 1977 | r, |
| 1978 | is_stale=bool(current_commit) and current_commit != r.from_branch_commit_id, |
| 1979 | ) |
| 1980 | for r in rows |
| 1981 | ] |
| 1982 | return SimulationListResponse(simulations=responses, total=len(responses)) |
File History
2 commits
sha256:f99af7b1a7f36c4d537d1c630d4b71fc39222b1255f82e930929e2fc89015e11
fix: relax browse_repo perf budget to 500ms — 200ms was too…
Sonnet 4.6
102 days ago
sha256:763eb2cb8675073b84c19345b27586d2ed939a9aee97c5479b69f502f1a70eff
fix(tests): update test suite to match current implementation
Sonnet 4.6
patch
123 days ago