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