"""MuseHub merge proposal persistence adapter — single point of DB access for proposals. This module is the ONLY place that touches the ``musehub_proposals`` table. Route handlers delegate here; no business logic lives in routes. Boundary rules: - Must NOT import state stores, SSE queues, or LLM clients. - May import ORM models from musehub.db domain-specific modules. - May import Pydantic response models from musehub.models.musehub. - May import musehub.core.genesis for genesis ID computation. Merge strategy -------------- ``merge_commit`` is the only strategy at MVP. It creates a new commit on ``to_branch`` whose parent_ids are [to_branch head, from_branch head], then updates the ``to_branch`` head pointer and marks the proposal as merged. If either branch has no commits yet (no head commit), the merge is rejected with a ``ValueError`` — there is nothing to merge. """ import logging from collections.abc import Awaitable, Callable from datetime import datetime, timezone import sqlalchemy as sa from sqlalchemy import ColumnElement, func, select from sqlalchemy.ext.asyncio import AsyncSession from musehub.core.genesis import compute_branch_id, compute_comment_id, compute_proposal_id, compute_review_id, compute_simulation_id from musehub.db.musehub_identity_models import MusehubIdentity from musehub.db.musehub_repo_models import MusehubBranch, MusehubCommit, MusehubCommitGraph, MusehubCommitRef from musehub.db.musehub_social_models import ( MusehubProposal, MusehubProposalComment, MusehubProposalReview, MusehubProposalSimulation, ) from musehub.services.proposal_dag import ( CycleError, ProposalDag, blocked_by_numbers, blocks_numbers, create_dependency_edges, is_blocked, load_dag_for_proposals, ) from musehub.muse_cli.snapshot import compute_commit_id, compute_snapshot_id class BranchNotFoundError(Exception): """Raised when a required branch does not exist in the repo.""" from musehub.types.json_types import JSONObject, StrDict from musehub.models.musehub import ( DomainHeatEntry, DomainHeatResponse, MergeReadinessResponse, MergeResultEmbed, ProposalCommentListResponse, ProposalCommentResponse, ProposalListEntry, ProposalListFilters, ProposalListResponse, ProposalResponse, ProposalReviewListResponse, ProposalReviewResponse, SimulationListResponse, SimulationResponse, ) type _CommentMap = dict[str, ProposalCommentResponse] logger = logging.getLogger(__name__) def _utc_now() -> datetime: return datetime.now(tz=timezone.utc) def _symbols_from_delta(delta: JSONObject | None) -> list[str]: """Extract unique symbol addresses from a structured_delta dict. Only child_op addresses that contain ``::`` are returned — file-level ops without a symbol component are intentionally excluded. """ if not isinstance(delta, dict): return [] seen: set[str] = set() for file_op in delta.get("ops") or []: if not isinstance(file_op, dict): continue for child_op in file_op.get("child_ops") or []: if not isinstance(child_op, dict): continue addr = child_op.get("address", "") if "::" in addr and addr not in seen: seen.add(addr) return list(seen) async def _touched_symbols_for_branch( session: AsyncSession, repo_id: str, branch: str ) -> list[str]: """Return the union of symbol addresses touched by all commits on ``branch``.""" rows = (await session.execute( select(MusehubCommit.structured_delta) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where( MusehubCommitRef.repo_id == repo_id, MusehubCommit.branch == branch, MusehubCommit.structured_delta.isnot(None), ) )).scalars().all() seen: set[str] = set() for delta in rows: seen.update(_symbols_from_delta(delta)) return list(seen) def _to_proposal_response( row: MusehubProposal, *, dag: ProposalDag | None = None, simulations: "SimulationListResponse | None" = None, merge_result: MergeResultEmbed | None = None, url_prefix: str = "", ) -> ProposalResponse: from musehub.models.musehub import MergeConditions, MergeStrategy, ProposalType mc_raw = getattr(row, "merge_conditions", None) mc = MergeConditions.model_validate(mc_raw) if mc_raw else None blocked_by: list[int] = [] blocks: list[int] = [] is_blocked_flag = False if dag is not None: blocked_by = blocked_by_numbers(dag, row.proposal_id) blocks = blocks_numbers(dag, row.proposal_id) is_blocked_flag = is_blocked(dag, row.proposal_id) latest_simulations: dict[str, dict] = {} if simulations is not None: for sim in simulations.simulations: latest_simulations[sim.simulation_type] = { "simulation_id": sim.simulation_id, "result": sim.result, "is_stale": sim.is_stale, "from_branch_commit_id": sim.from_branch_commit_id, "duration_ms": sim.duration_ms, "created_at": sim.created_at.isoformat(), } return ProposalResponse( proposal_id=row.proposal_id, proposal_number=row.proposal_number, url=f"{url_prefix}/proposals/{row.proposal_id}" if url_prefix else "", title=row.title, body=row.body, state=row.state, from_branch=row.from_branch, to_branch=row.to_branch, merge_commit_id=row.merge_commit_id, merged_at=row.merged_at, author=row.author, created_at=row.created_at, proposal_type=ProposalType(getattr(row, "proposal_type", "state_merge")), is_draft=getattr(row, "is_draft", False), merge_conditions=mc, merge_strategy=MergeStrategy(getattr(row, "merge_strategy", "overlay")), selective_domains=getattr(row, "selective_domains", None), risk_score=getattr(row, "risk_score", None), dimensional_risk=dict(getattr(row, "dimensional_risk", None) or {}), blocked_by=blocked_by, blocks=blocks, is_blocked=is_blocked_flag, latest_simulations=latest_simulations, proposer_signature=getattr(row, "proposer_signature", None), proposer_public_key=getattr(row, "proposer_public_key", None), from_commit_id=getattr(row, "from_commit_id", None), to_commit_id=getattr(row, "to_commit_id", None), from_snapshot_id=getattr(row, "from_snapshot_id", None), to_snapshot_id=getattr(row, "to_snapshot_id", None), merge_result=merge_result, ) async def _get_branch( session: AsyncSession, repo_id: str, branch_name: str ) -> MusehubBranch | None: """Return the branch record by repo + name, or None.""" stmt = select(MusehubBranch).where( MusehubBranch.repo_id == repo_id, MusehubBranch.name == branch_name, ) return (await session.execute(stmt)).scalar_one_or_none() async def create_proposal( session: AsyncSession, *, repo_id: str, title: str, from_branch: str, to_branch: str, body: str = "", author: str = "", author_identity_id: str = "", proposal_type: str = "state_merge", is_draft: bool = False, merge_strategy: str = "overlay", merge_conditions: JSONObject | None = None, selective_domains: list[str] | None = None, depends_on: list[str] | None = None, proposer_signature: str | None = None, proposer_public_key: str | None = None, url_prefix: str = "", ) -> ProposalResponse: """Persist a new merge proposal in ``open`` state and return its wire representation. ``author`` identifies the user opening the proposal — typically the MSign handle from the request context, or a display name from the seed script. Raises ``BranchNotFoundError`` if ``from_branch`` does not exist in the repo; the caller should surface this as HTTP 404. """ branch = await _get_branch(session, repo_id, from_branch) if branch is None: raise BranchNotFoundError(f"Branch '{from_branch}' not found in repo {repo_id}") from_commit_id = branch.head_commit_id to_branch_row = await _get_branch(session, repo_id, to_branch) to_commit_id = to_branch_row.head_commit_id if to_branch_row else None # from_snapshot_id/to_snapshot_id hold the real Snapshot (manifest) ID that # each anchor commit points to -- not the commit ID itself (see musehub#144 # review; #0049 named these columns "snapshot" but had always stored the # commit ID). Look each one up; leave null if the commit row is somehow # absent rather than guessing. from_snapshot_id: str | None = None if from_commit_id is not None: from_commit_row = await session.get(MusehubCommit, from_commit_id) from_snapshot_id = from_commit_row.snapshot_id if from_commit_row else None to_snapshot_id: str | None = None if to_commit_id is not None: to_commit_row = await session.get(MusehubCommit, to_commit_id) to_snapshot_id = to_commit_row.snapshot_id if to_commit_row else None # Assign the next sequential proposal_number for this repo (1-based, like GitHub) max_num_result = await session.execute( select(func.max(MusehubProposal.proposal_number)).where( MusehubProposal.repo_id == repo_id ) ) max_num: int | None = max_num_result.scalar_one_or_none() next_num = (max_num or 0) + 1 touched = await _touched_symbols_for_branch(session, repo_id, from_branch) _created_at = _utc_now() initial_state = "drafting" if is_draft else "open" proposal = MusehubProposal( proposal_id=compute_proposal_id( repo_id, author_identity_id, from_branch, to_branch, _created_at.isoformat() ), repo_id=repo_id, proposal_number=next_num, title=title, body=body, state=initial_state, from_branch=from_branch, to_branch=to_branch, author=author, touched_symbols=touched, created_at=_created_at, proposal_type=proposal_type, is_draft=is_draft, merge_strategy=merge_strategy, merge_conditions=merge_conditions, selective_domains=selective_domains, proposer_signature=proposer_signature, proposer_public_key=proposer_public_key, from_commit_id=from_commit_id, to_commit_id=to_commit_id, from_snapshot_id=from_snapshot_id, to_snapshot_id=to_snapshot_id, ) session.add(proposal) await session.flush() # Persist dependency edges — validates existence, detects cycles before commit if depends_on: await create_dependency_edges(session, proposal.proposal_id, depends_on) await session.refresh(proposal) logger.info("✅ Created proposal '%s' (%s → %s) in repo %s", title, from_branch, to_branch, repo_id) return _to_proposal_response(proposal, url_prefix=url_prefix) def _risk_band_conditions(bands: list[str]) -> list[ColumnElement[bool]]: """Return SQLAlchemy conditions that match risk_score to the given band names. Each band maps to a half-open interval on [0.0, 1.0]: critical ≥ 0.75 high 0.50 – 0.74… medium 0.25 – 0.49… low 0.01 – 0.24… none == 0.0 (or NULL) Multiple bands are OR-ed together. """ band_ranges: dict[str, tuple[float | None, float | None]] = { "critical": (0.75, None), "high": (0.50, 0.75), "medium": (0.25, 0.50), "low": (0.01, 0.25), "none": (None, 0.01), } clauses = [] for band in bands: lo, hi = band_ranges.get(band, (None, None)) if band == "none": clauses.append( sa.or_( MusehubProposal.risk_score.is_(None), MusehubProposal.risk_score == 0.0, ) ) elif lo is not None and hi is not None: clauses.append( sa.and_( MusehubProposal.risk_score >= lo, MusehubProposal.risk_score < hi, ) ) elif lo is not None: clauses.append(MusehubProposal.risk_score >= lo) return clauses async def list_proposals( session: AsyncSession, repo_id: str, *, state: str = "all", cursor: str | None = None, limit: int = 20, filters: ProposalListFilters | None = None, url_prefix: str = "", ) -> ProposalListResponse: """Return merge proposals for a repo with cursor-based keyset pagination. ``state`` may be ``"open"``, ``"merged"``, ``"closed"``, ``"all"``, or any value in the extended 7-state machine accepted by ``ProposalListFilters``. When ``filters`` is provided, the following additional predicates are applied: - ``filters.risk_band`` → ``risk_score`` range filter (OR across bands) - ``filters.domain`` → ``risk_score > 0`` when "code" is included; other domains require per-domain risk rows (Phase 2 DB prerequisite) - ``filters.author_type`` → LEFT JOIN to ``musehub_identities`` - ``filters.assigned_reviewer`` → EXISTS sub-select on ``musehub_proposal_reviews`` - ``filters.sort`` → ordering; ``merge_ready_first`` uses an approval count sub-select to surface ready proposals first ``cursor`` is the ISO 8601 ``created_at`` of the last seen proposal (opaque to callers — pass ``nextCursor`` from a previous response verbatim). Args: session: Async database session. repo_id: Target repository ID. state: Proposal state filter; defaults to "all". cursor: Pagination cursor (opaque ISO 8601 string). limit: Page size. filters: Optional ``ProposalListFilters``; overrides ``state``, ``limit``, ``cursor``, and ``sort`` when provided. ``state`` in ``filters`` takes precedence over the top-level ``state`` kwarg. Returns: ``ProposalListResponse`` with paginated proposals and a ``nextCursor``. Raises: ValueError: If ``cursor`` is not a valid ISO 8601 datetime string. """ f = filters effective_state = (f.state if f else None) or state effective_limit = (f.limit if f else None) or limit effective_cursor = (f.cursor if f else None) or cursor effective_sort = (f.sort if f else None) or "newest" conditions: list[ColumnElement[bool]] = [MusehubProposal.repo_id == repo_id] if effective_state != "all": conditions.append(MusehubProposal.state == effective_state) stmt = select(MusehubProposal) # ── Author-type filter (requires join to identities) ─────────────────────── if f and f.author_type != "all": stmt = stmt.join( MusehubIdentity, MusehubIdentity.handle == MusehubProposal.author, isouter=True, ) if f.author_type == "human": conditions.append( sa.or_( MusehubIdentity.identity_type == "human", MusehubIdentity.identity_type.is_(None), ) ) else: conditions.append(MusehubIdentity.identity_type == f.author_type) # ── Risk-band filter ─────────────────────────────────────────────────────── if f and f.risk_band: band_clauses = _risk_band_conditions(f.risk_band) if band_clauses: conditions.append(sa.or_(*band_clauses)) # ── Domain filter — code only at Phase 2 (other domains require risk rows) ─ if f and f.domain: domain_clauses = [] if "code" in f.domain: domain_clauses.append( sa.and_( MusehubProposal.risk_score.is_not(None), MusehubProposal.risk_score > 0.0, ) ) if domain_clauses: conditions.append(sa.or_(*domain_clauses)) # ── Proposal-type filter ────────────────────────────────────────────────── if f and f.proposal_type: conditions.append(MusehubProposal.proposal_type.in_(f.proposal_type)) # ── Is-draft filter ─────────────────────────────────────────────────────── if f and f.is_draft is not None: conditions.append(MusehubProposal.is_draft == f.is_draft) # ── Merge-strategy filter ───────────────────────────────────────────────── if f and f.merge_strategy: conditions.append(MusehubProposal.merge_strategy.in_(f.merge_strategy)) # ── Assigned-reviewer filter ─────────────────────────────────────────────── if f and f.assigned_reviewer: reviewer_subq = ( select(MusehubProposalReview.proposal_id) .where( MusehubProposalReview.reviewer_username == f.assigned_reviewer, MusehubProposalReview.state.in_(["pending", "approved", "changes_requested"]), ) .correlate(MusehubProposal) ) conditions.append(MusehubProposal.proposal_id.in_(reviewer_subq)) # ── Count total matching rows (re-use same joins as data query) ─────────── count_stmt = stmt.with_only_columns( func.count(MusehubProposal.proposal_id) ).where(*conditions).order_by(None) total: int = (await session.execute(count_stmt)).scalar_one() # ── Sort order ───────────────────────────────────────────────────────────── order_clauses: list[ColumnElement[bool]] cursor_ascending: bool # True → > cursor, False → < cursor if effective_sort == "oldest": order_clauses = [MusehubProposal.created_at.asc()] cursor_ascending = True elif effective_sort == "risk_desc": order_clauses = [MusehubProposal.risk_score.desc().nulls_last()] cursor_ascending = False elif effective_sort == "risk_asc": order_clauses = [MusehubProposal.risk_score.asc().nulls_last()] cursor_ascending = False elif effective_sort == "merge_ready_first": # Sub-select: count approved reviews per proposal; ready = count>=2 AND breakage=0 approval_subq = ( select(func.count(MusehubProposalReview.review_id)) .where( MusehubProposalReview.proposal_id == MusehubProposal.proposal_id, MusehubProposalReview.state == "approved", ) .correlate(MusehubProposal) .scalar_subquery() ) is_ready = sa.case( ( sa.and_( approval_subq >= _DEFAULT_REQUIRED_APPROVALS, MusehubProposal.breakage_count == 0, ), 0, ), else_=1, ) order_clauses = [is_ready, MusehubProposal.created_at.desc()] cursor_ascending = False else: # newest (default) order_clauses = [MusehubProposal.created_at.desc()] cursor_ascending = False # ── Cursor predicate ─────────────────────────────────────────────────────── data_conditions = list(conditions) if effective_cursor is not None: cursor_dt = datetime.fromisoformat(effective_cursor) if cursor_ascending: data_conditions.append(MusehubProposal.created_at > cursor_dt) else: data_conditions.append(MusehubProposal.created_at < cursor_dt) rows = list( ( await session.execute( stmt.where(*data_conditions).order_by(*order_clauses).limit(effective_limit + 1) ) ).scalars() ) next_cursor: str | None = None if len(rows) == effective_limit + 1: next_cursor = rows[effective_limit - 1].created_at.isoformat() rows = rows[:effective_limit] return ProposalListResponse( proposals=[_to_proposal_response(r, url_prefix=url_prefix) for r in rows], total=total, next_cursor=next_cursor, ) async def get_proposal( session: AsyncSession, repo_id: str, proposal_id: str, *, url_prefix: str = "", ) -> ProposalResponse | None: """Return a single proposal enriched with DAG position and simulation summaries.""" stmt = select(MusehubProposal).where( MusehubProposal.repo_id == repo_id, MusehubProposal.proposal_id == proposal_id, ) row = (await session.execute(stmt)).scalar_one_or_none() if row is None: return None dag = await load_dag_for_proposals(session, [proposal_id]) sims = await list_simulations(session, repo_id, proposal_id) return _to_proposal_response(row, dag=dag, simulations=sims, url_prefix=url_prefix) async def update_proposal( session: AsyncSession, repo_id: str, proposal_id: str, *, title: str | None = None, body: str | None = None, proposal_type: str | None = None, merge_strategy: str | None = None, ) -> ProposalResponse | None: """Apply a partial update to a proposal. Returns None if not found.""" stmt = select(MusehubProposal).where( MusehubProposal.repo_id == repo_id, MusehubProposal.proposal_id == proposal_id, ) row = (await session.execute(stmt)).scalar_one_or_none() if row is None: return None if title is not None: row.title = title if body is not None: row.body = body if proposal_type is not None: row.proposal_type = proposal_type if merge_strategy is not None: row.merge_strategy = merge_strategy await session.commit() await session.refresh(row) dag = await load_dag_for_proposals(session, [proposal_id]) sims = await list_simulations(session, repo_id, proposal_id) return _to_proposal_response(row, dag=dag, simulations=sims) async def _resolve_ancestor_manifest( session: AsyncSession, repo_id: str, from_branch: str, to_branch: str, ) -> StrDict | None: """Find the common ancestor snapshot manifest for a branch pair. Walks the from_branch commit history looking for the first commit whose parent_ids overlap with commits reachable from to_branch. This is a lightweight merge-base approximation suitable for server-side strategy computation (not a full LCA traversal). Returns None if no ancestor can be found (new repo, orphan branches). """ from musehub.graph.walk import walk_dag_async from musehub.services.musehub_snapshot import get_snapshot_manifest to_b = await _get_branch(session, repo_id, to_branch) from_b = await _get_branch(session, repo_id, from_branch) if to_b is None or from_b is None: return None # Walk 1: collect to_branch ancestry (bounded BFS) to_commit_ids: set[str] = set() async def _to_adj(cid: str) -> list[str]: commit = await session.get(MusehubCommit, cid) return commit.parent_ids if commit and commit.parent_ids else [] async for cid in walk_dag_async( [to_b.head_commit_id] if to_b.head_commit_id else [], _to_adj, max_nodes=200, ): to_commit_ids.add(cid) # Walk 2: first-parent walk on from_branch, looking for merge base candidate_id: str | None = None async def _from_adj(cid: str) -> list[str]: nonlocal candidate_id commit = await session.get(MusehubCommit, cid) if commit is None: return [] for parent_id in (commit.parent_ids or []): if parent_id in to_commit_ids: candidate_id = parent_id return [] # stop walking once merge base is found return commit.parent_ids[:1] if commit.parent_ids else [] async for _ in walk_dag_async( [from_b.head_commit_id] if from_b.head_commit_id else [], _from_adj, max_nodes=200, ): if candidate_id: break if not candidate_id: return None ancestor_commit = await session.get(MusehubCommit, candidate_id) if ancestor_commit is None or not ancestor_commit.snapshot_id: return None return await get_snapshot_manifest(session, ancestor_commit.snapshot_id) async def _rebase_commits( session: AsyncSession, repo_id: str, proposal: "MusehubProposal", merger_handle: str, to_b: "MusehubBranch | None", from_b: "MusehubBranch | None", to_manifest: StrDict, compute_commit_id_fn: Callable[..., str], upsert_snapshot_entries_fn: Callable[..., Awaitable[None]], ) -> str: """Replay each from_branch commit individually onto to_branch (linear history). Algorithm: 1. Collect from_branch commits not reachable from to_branch, oldest-first. 2. For each commit in order: a. Compute its file delta vs its parent snapshot. b. Apply that delta to the running rebased state (starts = to_manifest). c. Upsert the new snapshot. d. Create a new MusehubCommit with parent = previous replayed commit. 3. Return the commit_id of the tip (last replayed commit). """ from musehub.services.musehub_snapshot import get_snapshot_manifest to_head_cid = to_b.head_commit_id if to_b else None # Collect from_branch commits not on to_branch (walk parent_ids BFS). from_head_cid = from_b.head_commit_id if from_b else None if not from_head_cid: raise ValueError(f"from_branch '{proposal.from_branch}' has no commits") to_ancestors: set[str] = set() if to_head_cid: _q: list[str] = [to_head_cid] while _q: _cid = _q.pop() if _cid in to_ancestors: continue to_ancestors.add(_cid) _row = await session.get(MusehubCommit, _cid) if _row: _q.extend(_row.parent_ids or []) # BFS from from_head, collecting commits not in to_ancestors. from_commits_unordered: list[MusehubCommit] = [] _seen: set[str] = set() _frontier = [from_head_cid] while _frontier: _cid = _frontier.pop(0) if _cid in _seen or _cid in to_ancestors: continue _seen.add(_cid) _c = await session.get(MusehubCommit, _cid) if _c is None: continue from_commits_unordered.append(_c) _frontier.extend(p for p in (_c.parent_ids or []) if p not in _seen) # Topological sort: oldest-first (Kahn's algorithm on parent_ids subset). cid_set = {c.commit_id for c in from_commits_unordered} in_degree: dict[str, int] = {c.commit_id: 0 for c in from_commits_unordered} children: dict[str, list[MusehubCommit]] = {c.commit_id: [] for c in from_commits_unordered} for c in from_commits_unordered: for p in (c.parent_ids or []): if p in cid_set: children[p].append(c) in_degree[c.commit_id] += 1 queue = [c for c in from_commits_unordered if in_degree[c.commit_id] == 0] from_commits: list[MusehubCommit] = [] while queue: c = queue.pop(0) from_commits.append(c) for child in children[c.commit_id]: in_degree[child.commit_id] -= 1 if in_degree[child.commit_id] == 0: queue.append(child) # Look up generation of to_head for commit graph. _to_gen = 0 if to_head_cid: _tg_row = await session.get(MusehubCommitGraph, to_head_cid) if _tg_row: _to_gen = _tg_row.generation # Replay each commit. rebased_state: StrDict = dict(to_manifest) current_parent_id: str = to_head_cid or "" current_gen = _to_gen tip_cid = current_parent_id for orig in from_commits: # Compute file delta: orig_snapshot vs orig's parent snapshot. orig_manifest = await get_snapshot_manifest(session, orig.snapshot_id) if orig.snapshot_id else {} parent_manifest: StrDict = {} for p in (orig.parent_ids or []): _p_row = await session.get(MusehubCommit, p) if _p_row and _p_row.snapshot_id: parent_manifest = await get_snapshot_manifest(session, _p_row.snapshot_id) break # Apply delta to rebased_state. for path, oid in orig_manifest.items(): rebased_state[path] = oid for path in list(parent_manifest.keys()): if path not in orig_manifest: rebased_state.pop(path, None) # Upsert new snapshot for this rebased state. rebased_snap_id = compute_snapshot_id(rebased_state) await upsert_snapshot_entries_fn(session, repo_id, rebased_snap_id, rebased_state) # Create the replayed commit. committed_at = _utc_now() new_parent_ids = [current_parent_id] if current_parent_id else [] new_cid = compute_commit_id_fn( new_parent_ids, rebased_snap_id, orig.message, committed_at.isoformat(), author=orig.author or merger_handle, signer_public_key="", ) session.add(MusehubCommit( commit_id=new_cid, branch=proposal.to_branch, parent_ids=new_parent_ids, message=orig.message, author=orig.author or merger_handle, timestamp=committed_at, snapshot_id=rebased_snap_id, structured_delta=orig.structured_delta, agent_id=orig.agent_id or "", model_id=orig.model_id or "", )) session.add(MusehubCommitRef(repo_id=repo_id, commit_id=new_cid)) current_gen += 1 session.add(MusehubCommitGraph( commit_id=new_cid, parent_ids=new_parent_ids, generation=current_gen, snapshot_id=rebased_snap_id, )) current_parent_id = new_cid tip_cid = new_cid if not tip_cid: raise ValueError("rebase produced no commits — from_branch has no unique commits") return tip_cid async def merge_proposal( session: AsyncSession, repo_id: str, proposal_id: str, *, merge_strategy: str = "merge_commit", merger_handle: str = "", commit_history: str = "merge", ) -> ProposalResponse: """Merge an open proposal. Args: merge_strategy: Content/snapshot merge strategy — how file manifests combine. ``overlay`` (default), ``weave``, ``replay``, ``selective``. ``overlay`` (default), ``weave``, ``replay``, ``selective``. commit_history: VCS commit graph style. ``merge`` (default) — new commit with two parents [to_head, from_head]. ``squash`` — one new commit, parent = [to_head] only. ``rebase`` — replay each from_branch commit linearly (TODO). Raises: ValueError: Proposal not found or branch has no commits. RuntimeError: Proposal is already merged or closed (surfaces as 409). """ stmt = select(MusehubProposal).where( MusehubProposal.repo_id == repo_id, MusehubProposal.proposal_id == proposal_id, ) proposal = (await session.execute(stmt)).scalar_one_or_none() if proposal is None: raise ValueError(f"Proposal {proposal_id} not found in repo {repo_id}") if proposal.state not in ("open", "approved"): raise RuntimeError(f"Proposal {proposal_id} is already {proposal.state}") # Gate on hard dependencies — check require_dependency_merged from merge_conditions mc_raw = proposal.merge_conditions or {} if mc_raw.get("require_dependency_merged", True): dag = await load_dag_for_proposals(session, [proposal_id]) if is_blocked(dag, proposal_id): unmerged_nums = blocked_by_numbers(dag, proposal_id) raise RuntimeError( f"Proposal {proposal_id} cannot be merged: " f"unmerged dependencies: proposal numbers {unmerged_nums}" ) from_b = await _get_branch(session, repo_id, proposal.from_branch) to_b = await _get_branch(session, repo_id, proposal.to_branch) # Collect parent commit IDs for the merge commit. parent_ids: list[str] = [] if to_b is not None and to_b.head_commit_id is not None: parent_ids.append(to_b.head_commit_id) if from_b is not None and from_b.head_commit_id is not None: parent_ids.append(from_b.head_commit_id) if not parent_ids: raise ValueError( f"Cannot merge: neither '{proposal.from_branch}' nor '{proposal.to_branch}' has any commits" ) # Squash: drop from_head from parent_ids so the result is linear. if commit_history == "squash" and to_b is not None and to_b.head_commit_id is not None: parent_ids = [to_b.head_commit_id] from musehub.services.musehub_snapshot import get_snapshot_manifest, upsert_snapshot_entries from musehub.services.proposal_merge_strategies import execute_merge_strategy to_manifest: StrDict = {} if to_b is not None and to_b.head_commit_id is not None: to_head = await session.get(MusehubCommit, to_b.head_commit_id) if to_head is not None and to_head.snapshot_id: to_manifest = await get_snapshot_manifest(session, to_head.snapshot_id) from_manifest: StrDict = {} from_head_snapshot_id: str | None = None if from_b is not None and from_b.head_commit_id is not None: from_head = await session.get(MusehubCommit, from_b.head_commit_id) if from_head is not None and from_head.snapshot_id: from_head_snapshot_id = from_head.snapshot_id from_manifest = await get_snapshot_manifest(session, from_head.snapshot_id) # Resolve ancestor manifest for three-way strategies. # The ancestor is the snapshot at the point from_branch was cut from to_branch. # We approximate this as the earliest commit on from_branch that has a parent # on to_branch — or the repo's first commit if the branch predates any tracking. # For OVERLAY (default), the ancestor is used only for conflict audit. ancestor_manifest: StrDict | None = None strategy_name = getattr(proposal, "merge_strategy", "overlay") or "overlay" # Caller can override via merge_strategy parameter (passed as the kwarg). if merge_strategy and merge_strategy != "merge_commit": strategy_name = merge_strategy if strategy_name in ("weave", "replay", "selective", "phased"): # Walk from_branch commit history to find the merge-base with to_branch. # Simplified: look for the first from_branch commit whose parent is in to_branch. ancestor_manifest = await _resolve_ancestor_manifest( session, repo_id, proposal.from_branch, proposal.to_branch ) selective_domains: list[str] | None = getattr(proposal, "selective_domains", None) merge_result = execute_merge_strategy( strategy_name, to_manifest, from_manifest, ancestor_manifest=ancestor_manifest, selective_domains=selective_domains, ) merged_manifest: StrDict = merge_result.manifest logger.info( "🔀 Merge strategy=%s added=%d modified=%d removed=%d conflicts=%d domains=%s", merge_result.strategy, merge_result.files_added, merge_result.files_modified, merge_result.files_removed, len(merge_result.conflicts), merge_result.domains_merged, ) # ── Commit creation — varies by commit_history style ──────────────── merge_commit_id: str if commit_history == "rebase": # Replay each from_branch commit individually onto to_branch. # Walk from_branch ancestors oldest-first, stopping at to_branch head. merge_commit_id = await _rebase_commits( session, repo_id, proposal, merger_handle, to_b, from_b, to_manifest, compute_commit_id, upsert_snapshot_entries, ) else: # merge and squash: create a single new commit. merged_snapshot_id = compute_snapshot_id(merged_manifest) await upsert_snapshot_entries(session, repo_id, merged_snapshot_id, merged_manifest) merge_message = f"Merge '{proposal.from_branch}' into '{proposal.to_branch}' — proposal: {proposal.title}" committed_at = _utc_now() merge_commit_id = compute_commit_id( parent_ids, merged_snapshot_id, merge_message, committed_at.isoformat(), author=merger_handle, signer_public_key="", ) session.add(MusehubCommit( commit_id=merge_commit_id, branch=proposal.to_branch, parent_ids=parent_ids, message=merge_message, author=merger_handle, timestamp=committed_at, snapshot_id=merged_snapshot_id, )) session.add(MusehubCommitRef(repo_id=repo_id, commit_id=merge_commit_id)) _parent_gen_q = await session.execute( select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) .where(MusehubCommitGraph.commit_id.in_(parent_ids)) ) _parent_gens = [row[1] for row in _parent_gen_q.all()] _merge_generation = (max(_parent_gens) + 1) if _parent_gens else 0 session.add(MusehubCommitGraph( commit_id=merge_commit_id, parent_ids=parent_ids, generation=_merge_generation, snapshot_id=merged_snapshot_id, )) # Advance (or create) the to_branch head pointer. if to_b is None: to_b = MusehubBranch( branch_id=compute_branch_id(repo_id, proposal.to_branch), repo_id=repo_id, name=proposal.to_branch, head_commit_id=merge_commit_id, ) session.add(to_b) else: to_b.head_commit_id = merge_commit_id # Delete the source branch so it disappears from refs. This lets # `muse fetch --prune` clean up the local tracking ref automatically, # matching the behaviour users expect after a proposal merge. if from_b is not None: await session.delete(from_b) # Refresh touched_symbols from the from_branch commits before marking merged. # This gives the most accurate symbol set at the moment of merge, capturing # any commits pushed to from_branch after the proposal was created. touched = await _touched_symbols_for_branch(session, repo_id, proposal.from_branch) proposal.touched_symbols = touched # Mark proposal as merged and record the exact merge timestamp. proposal.state = "merged" proposal.merge_commit_id = merge_commit_id proposal.merged_at = _utc_now() await session.flush() await session.refresh(proposal) logger.info( "✅ Merged proposal %s ('%s' → '%s') in repo %s, merge commit %s", proposal_id, proposal.from_branch, proposal.to_branch, repo_id, merge_commit_id, ) embed = MergeResultEmbed( status="merged", commit_id=merge_commit_id, strategy=merge_result.strategy, on_conflict=None, history=commit_history, conflicts=[c.path for c in merge_result.conflicts], files_changed={ "added": merge_result.files_added, "modified": merge_result.files_modified, "deleted": merge_result.files_removed, }, semver_impact="", ) return _to_proposal_response(proposal, merge_result=embed) # --------------------------------------------------------------------------- # Reopen proposal # --------------------------------------------------------------------------- async def close_proposal( session: AsyncSession, repo_id: str, proposal_id: str, ) -> ProposalResponse: """Set an open proposal to ``closed`` state. Raises: KeyError: proposal not found in this repo. RuntimeError: proposal is already closed or merged. """ proposal = (await session.execute( select(MusehubProposal).where( MusehubProposal.proposal_id == proposal_id, MusehubProposal.repo_id == repo_id, ) )).scalar_one_or_none() if proposal is None: raise KeyError(f"Proposal {proposal_id} not found in repo {repo_id}") if proposal.state != "open": raise RuntimeError(f"Proposal {proposal_id} is already {proposal.state}") proposal.state = "closed" await session.flush() await session.refresh(proposal) return _to_proposal_response(proposal) async def reopen_proposal( session: AsyncSession, repo_id: str, proposal_id: str, ) -> ProposalResponse: """Reset a merged or closed proposal back to ``open`` state. Clears ``merge_commit_id`` and ``merged_at`` so the proposal can be re-merged after a corrupt commit is cleaned up (e.g. bug #36). Raises: KeyError: proposal not found in this repo. RuntimeError: proposal is already open. """ proposal = (await session.execute( select(MusehubProposal).where( MusehubProposal.proposal_id == proposal_id, MusehubProposal.repo_id == repo_id, ) )).scalar_one_or_none() if proposal is None: raise KeyError(f"Proposal {proposal_id} not found in repo {repo_id}") if proposal.state == "open": raise RuntimeError(f"Proposal {proposal_id} is already open") proposal.state = "open" proposal.merge_commit_id = None proposal.merged_at = None await session.flush() await session.refresh(proposal) return _to_proposal_response(proposal) # --------------------------------------------------------------------------- # Proposal review comments # --------------------------------------------------------------------------- def _to_comment_response(row: MusehubProposalComment) -> ProposalCommentResponse: dim_ref: JSONObject = row.dimension_ref or {} return ProposalCommentResponse( comment_id=row.comment_id, proposal_id=row.proposal_id, author=row.author, author_user_id=getattr(row, "author_user_id", None), agent_id=getattr(row, "agent_id", None) or None, model_id=getattr(row, "model_id", None) or None, body=row.body, target_type=str(dim_ref.get("type", "general")), target_track=str(dim_ref["track"]) if "track" in dim_ref else None, target_beat_start=float(dim_ref["beat_start"]) if isinstance(dim_ref.get("beat_start"), (int, float)) else None, target_beat_end=float(dim_ref["beat_end"]) if isinstance(dim_ref.get("beat_end"), (int, float)) else None, target_note_pitch=int(dim_ref["pitch"]) if isinstance(dim_ref.get("pitch"), int) else None, parent_comment_id=row.parent_comment_id, symbol_address=row.symbol_address, created_at=row.created_at, updated_at=getattr(row, "updated_at", None), is_deleted=bool(getattr(row, "is_deleted", False)), ) async def create_proposal_comment( session: AsyncSession, *, proposal_id: str, repo_id: str, author: str, author_identity_id: str = "", body: str, target_type: str = "general", target_track: str | None = None, target_beat_start: float | None = None, target_beat_end: float | None = None, target_note_pitch: int | None = None, parent_comment_id: str | None = None, symbol_address: str | None = None, ) -> ProposalCommentResponse: """Persist a new review comment on a proposal and return its wire representation. ``author`` is the MSign handle of the reviewer. ``parent_comment_id`` must be an existing top-level comment on the same proposal when creating a threaded reply; the caller validates this constraint before calling here. Raises ``ValueError`` if the proposal does not exist in the given repo. """ stmt = select(MusehubProposal).where( MusehubProposal.proposal_id == proposal_id, MusehubProposal.repo_id == repo_id, ) proposal = (await session.execute(stmt)).scalar_one_or_none() if proposal is None: raise ValueError(f"Proposal {proposal_id} not found in repo {repo_id}") dimension_ref: JSONObject = {"type": target_type} if target_track is not None: dimension_ref["track"] = target_track if target_beat_start is not None: dimension_ref["beat_start"] = target_beat_start if target_beat_end is not None: dimension_ref["beat_end"] = target_beat_end if target_note_pitch is not None: dimension_ref["pitch"] = target_note_pitch _created_at = _utc_now() comment = MusehubProposalComment( comment_id=compute_comment_id(proposal_id, author_identity_id, _created_at.isoformat()), proposal_id=proposal_id, repo_id=repo_id, author=author, body=body, dimension_ref=dimension_ref, parent_comment_id=parent_comment_id, symbol_address=symbol_address or None, created_at=_created_at, ) session.add(comment) await session.flush() await session.refresh(comment) logger.info("✅ Created proposal comment %s on proposal %s by %s", comment.comment_id, proposal_id, author) return _to_comment_response(comment) async def backfill_comment_author_user_ids( session: AsyncSession, repo_id: str | None = None, batch: int = 500, ) -> int: """Populate author_user_id on proposal comments that have none. Joins musehub_proposal_comments → musehub_identities on handle (author) and writes the identity_id. Idempotent — rows already populated are skipped. Returns count of rows updated. """ from musehub.db.musehub_identity_models import MusehubIdentity where = [MusehubProposalComment.author_user_id.is_(None)] if repo_id: where.append(MusehubProposalComment.repo_id == repo_id) rows = (await session.execute( select(MusehubProposalComment).where(*where).limit(batch) )).scalars().all() if not rows: return 0 handles = {r.author for r in rows} identity_rows = (await session.execute( select(MusehubIdentity.handle, MusehubIdentity.identity_id) .where(MusehubIdentity.handle.in_(handles)) )).all() handle_to_id = {h: iid for h, iid in identity_rows} updated = 0 for row in rows: uid = handle_to_id.get(row.author) if uid: row.author_user_id = uid updated += 1 if updated: await session.flush() return updated async def list_proposal_comments( session: AsyncSession, proposal_id: str, repo_id: str, cursor: str | None = None, limit: int = 20, include_deleted: bool = False, ) -> ProposalCommentListResponse: """Return review comments for a proposal with cursor-based keyset pagination. Comments are assembled into a two-level thread tree from the current page. Top-level comments (``parent_comment_id`` is None) form the root list. Each carries a ``replies`` list with direct children that appear within the same page, sorted by ``created_at`` ascending. Grandchildren are not supported — callers should reply to the original top-level comment. ``include_deleted`` surfaces soft-deleted comments (``is_deleted=True``) alongside live ones — callers must independently verify the caller has write access before setting this, matching the audit-log intent of soft-delete rather than genuine removal. ``cursor`` is the ISO 8601 ``created_at`` of the last seen comment (opaque to callers — pass ``nextCursor`` from a previous response verbatim). Omit to start from the beginning. ``total`` covers all comments on the proposal regardless of the current page. """ conditions = [ MusehubProposalComment.proposal_id == proposal_id, MusehubProposalComment.repo_id == repo_id, ] if not include_deleted: conditions.append(MusehubProposalComment.is_deleted.is_(False)) count_stmt = select(func.count(MusehubProposalComment.comment_id)).where(*conditions) total: int = (await session.execute(count_stmt)).scalar_one() data_conditions = list(conditions) if cursor is not None: data_conditions.append( MusehubProposalComment.created_at > datetime.fromisoformat(cursor) ) rows = list( ( await session.execute( select(MusehubProposalComment) .where(*data_conditions) .order_by(MusehubProposalComment.created_at) .limit(limit + 1) ) ).scalars() ) next_cursor: str | None = None if len(rows) == limit + 1: next_cursor = rows[limit - 1].created_at.isoformat() rows = rows[:limit] # Build id → response map first; attach replies in a second pass. top_level: list[ProposalCommentResponse] = [] by_id: _CommentMap = {} for row in rows: resp = _to_comment_response(row) by_id[row.comment_id] = resp if row.parent_comment_id is None: top_level.append(resp) for row in rows: if row.parent_comment_id is not None: parent = by_id.get(row.parent_comment_id) if parent is not None: parent.replies.append(by_id[row.comment_id]) return ProposalCommentListResponse(comments=top_level, total=total, next_cursor=next_cursor) async def delete_comment( session: AsyncSession, comment_id: str, proposal_id: str, *, hard: bool = False, ) -> bool: """Delete a proposal comment. Returns True if the comment existed and was deleted. Soft-delete (default) sets ``is_deleted`` -- mirrors ``musehub_issues.delete_comment``, the reference implementation for this idiom (musehub#141/#143). ``hard=True`` actually removes the row; callers must independently verify the caller has admin-level access before setting this, per #141's framework-wide "--hard gated to owner/admin" convention. """ stmt = select(MusehubProposalComment).where( MusehubProposalComment.comment_id == comment_id, MusehubProposalComment.proposal_id == proposal_id, ) row = (await session.execute(stmt)).scalar_one_or_none() if row is None: return False if hard: await session.delete(row) logger.info("✅ Hard-deleted proposal comment %s", comment_id) else: row.is_deleted = True await session.flush() logger.info("✅ Soft-deleted proposal comment %s", comment_id) return True # --------------------------------------------------------------------------- # Proposal reviews (reviewer assignment + approval workflow) # --------------------------------------------------------------------------- def _to_review_response(row: MusehubProposalReview) -> ProposalReviewResponse: return ProposalReviewResponse( id=row.review_id, proposal_id=row.proposal_id, reviewer_username=row.reviewer_username, state=row.state, body=row.body, submitted_at=row.submitted_at, created_at=row.created_at, ) async def _assert_proposal_exists(session: AsyncSession, repo_id: str, proposal_id: str) -> None: """Raise ``ValueError`` if the proposal does not exist in the given repo.""" stmt = select(MusehubProposal).where( MusehubProposal.proposal_id == proposal_id, MusehubProposal.repo_id == repo_id, ) proposal = (await session.execute(stmt)).scalar_one_or_none() if proposal is None: raise ValueError(f"Proposal {proposal_id} not found in repo {repo_id}") async def request_reviewers( session: AsyncSession, *, repo_id: str, proposal_id: str, reviewers: list[str], ) -> ProposalReviewListResponse: """Add reviewer assignments to a proposal, creating a ``pending`` row for each. Idempotent: if a reviewer already has a row (in any state), the existing row is left unchanged so a submitted approval is never reset by a re-request. Raises ``ValueError`` if the proposal does not exist in the repo. Returns the full updated review list for the proposal. """ await _assert_proposal_exists(session, repo_id, proposal_id) for username in reviewers: existing_stmt = select(MusehubProposalReview).where( MusehubProposalReview.proposal_id == proposal_id, MusehubProposalReview.reviewer_username == username, ) existing = (await session.execute(existing_stmt)).scalar_one_or_none() if existing is None: now = _utc_now() identity_stmt = select(MusehubIdentity.identity_id).where( MusehubIdentity.handle == username ) reviewer_identity_id = (await session.execute(identity_stmt)).scalar_one_or_none() or username review = MusehubProposalReview( review_id=compute_review_id(proposal_id, reviewer_identity_id, now.isoformat()), proposal_id=proposal_id, reviewer_username=username, state="pending", created_at=now, ) session.add(review) logger.info("✅ Requested review from '%s' on proposal %s", username, proposal_id) await session.flush() return await list_reviews(session, repo_id=repo_id, proposal_id=proposal_id) async def remove_reviewer( session: AsyncSession, *, repo_id: str, proposal_id: str, username: str, ) -> ProposalReviewListResponse: """Remove a pending review request for ``username`` on a proposal. Only ``pending`` rows may be removed — submitted reviews are immutable to preserve the audit trail. Raises ``ValueError`` if the proposal does not exist, the reviewer was never requested, or the reviewer has already submitted a non-pending review. Returns the updated review list. """ await _assert_proposal_exists(session, repo_id, proposal_id) stmt = select(MusehubProposalReview).where( MusehubProposalReview.proposal_id == proposal_id, MusehubProposalReview.reviewer_username == username, ) row = (await session.execute(stmt)).scalar_one_or_none() if row is None: raise ValueError(f"Reviewer '{username}' was not requested on proposal {proposal_id}") if row.state != "pending": raise ValueError( f"Cannot remove reviewer '{username}': review already submitted (state={row.state})" ) await session.delete(row) await session.flush() logger.info("✅ Removed review request for '%s' from proposal %s", username, proposal_id) return await list_reviews(session, repo_id=repo_id, proposal_id=proposal_id) async def list_reviews( session: AsyncSession, *, repo_id: str, proposal_id: str, state: str | None = None, cursor: str | None = None, limit: int = 20, ) -> ProposalReviewListResponse: """Return reviews for a proposal with cursor-based keyset pagination. ``state`` may be one of ``pending``, ``approved``, ``changes_requested``, or ``dismissed``. When ``None``, all reviews are returned. Results are ordered by ``created_at`` ascending. ``cursor`` is the ISO 8601 ``created_at`` of the last seen review (opaque to callers — pass ``nextCursor`` from a previous response verbatim). Omit to start from the beginning. Raises ``ValueError`` if the proposal does not exist in the repo. """ await _assert_proposal_exists(session, repo_id, proposal_id) conditions = [MusehubProposalReview.proposal_id == proposal_id] if state is not None: conditions.append(MusehubProposalReview.state == state) count_stmt = select(func.count(MusehubProposalReview.review_id)).where(*conditions) total: int = (await session.execute(count_stmt)).scalar_one() data_conditions = list(conditions) if cursor is not None: data_conditions.append( MusehubProposalReview.created_at > datetime.fromisoformat(cursor) ) rows = list( ( await session.execute( select(MusehubProposalReview) .where(*data_conditions) .order_by(MusehubProposalReview.created_at) .limit(limit + 1) ) ).scalars() ) next_cursor: str | None = None if len(rows) == limit + 1: next_cursor = rows[limit - 1].created_at.isoformat() rows = rows[:limit] return ProposalReviewListResponse( reviews=[_to_review_response(r) for r in rows], total=total, next_cursor=next_cursor, ) async def submit_review( session: AsyncSession, *, repo_id: str, proposal_id: str, reviewer_username: str, reviewer_identity_id: str = "", verdict: str, body: str = "", ) -> ProposalReviewResponse: """Submit or update a formal review verdict for ``reviewer_username`` on a proposal. ``verdict`` maps to a new state: - ``approve`` → ``approved`` - ``request_changes`` → ``changes_requested`` If an existing row for this reviewer already exists, it is updated in-place, so changing from approve → request_changes (or vice versa) works by resubmitting. Ad-hoc reviews (no prior reviewer request) are also allowed. Raises ``ValueError`` if the proposal does not exist in the repo. """ await _assert_proposal_exists(session, repo_id, proposal_id) _VERDICT_TO_STATE: StrDict = { "approve": "approved", "request_changes": "changes_requested", } if verdict not in _VERDICT_TO_STATE: raise ValueError(f"Invalid verdict '{verdict}'. Must be approve or request_changes.") new_state = _VERDICT_TO_STATE[verdict] stmt = select(MusehubProposalReview).where( MusehubProposalReview.proposal_id == proposal_id, MusehubProposalReview.reviewer_username == reviewer_username, ) row = (await session.execute(stmt)).scalar_one_or_none() now = _utc_now() if row is None: row = MusehubProposalReview( review_id=compute_review_id(proposal_id, reviewer_identity_id, now.isoformat()), proposal_id=proposal_id, reviewer_username=reviewer_username, state=new_state, body=body or None, submitted_at=now, created_at=now, ) session.add(row) else: row.state = new_state row.body = body or None row.submitted_at = now await session.flush() await session.refresh(row) logger.info( "✅ Review submitted by '%s' on proposal %s: verdict=%s state=%s", reviewer_username, proposal_id, verdict, new_state, ) return _to_review_response(row) # ── Proposal list enrichment ────────────────────────────────────────────────── _RISK_BAND_THRESHOLDS: list[tuple[float, str]] = [ (0.75, "critical"), (0.50, "high"), (0.25, "medium"), (0.01, "low"), ] def _score_to_band(score: float) -> str: """Map a [0.0, 1.0] risk score to a human-readable band label. Thresholds: ≥ 0.75 → "critical" ≥ 0.50 → "high" ≥ 0.25 → "medium" > 0.0 → "low" 0.0 → "none" """ for threshold, band in _RISK_BAND_THRESHOLDS: if score >= threshold: return band return "none" # Default required-approvals when merge_conditions is null. _DEFAULT_REQUIRED_APPROVALS = 2 # Domain weight map used for aggregate risk score. Unknown domains default to 1.0. _DOMAIN_WEIGHTS: dict[str, float] = { "code": 1.2, "midi": 1.0, "stems": 1.0, "pay": 1.5, } class _ProposalPrefetch: """Holds pre-fetched batch data for a page of proposals. All DB reads for an entire page happen once in ``enrich_proposal_list_batch``; each ``enrich_proposal_list_entry`` call consults these in-memory maps — zero additional DB I/O per row. """ def __init__( self, *, reviews_by_proposal: dict[str, list[MusehubProposalReview]], author_types: dict[str, str], dag: ProposalDag | None = None, conflict_counts: dict[str, int | None] | None = None, ) -> None: self.reviews_by_proposal = reviews_by_proposal self.author_types = author_types self.dag: ProposalDag = dag or ProposalDag() # proposal_id → conflict_scan.result["conflict_count"]; None if not run self.conflict_counts: dict[str, int | None] = conflict_counts or {} async def _prefetch_for_batch( proposals: list[MusehubProposal], session: AsyncSession, ) -> _ProposalPrefetch: """Run the batch pre-fetch queries for a page of proposals. Issues exactly two DB queries regardless of page size: 1. All reviews for every proposal in the page. 2. Identity types for every author in the page. Args: proposals: ORM rows for the current page. session: Shared async session. Returns: ``_ProposalPrefetch`` with maps keyed by proposal_id / author handle. """ proposal_ids = [p.proposal_id for p in proposals] author_handles = list({p.author for p in proposals if p.author}) # Query 1 — reviews reviews_by_proposal: dict[str, list[MusehubProposalReview]] = {pid: [] for pid in proposal_ids} if proposal_ids: review_rows = list( ( await session.execute( select(MusehubProposalReview).where( MusehubProposalReview.proposal_id.in_(proposal_ids) ) ) ).scalars() ) for row in review_rows: reviews_by_proposal[row.proposal_id].append(row) # Query 2 — identity types author_types: dict[str, str] = {} if author_handles: identity_rows = list( ( await session.execute( select(MusehubIdentity.handle, MusehubIdentity.identity_type).where( MusehubIdentity.handle.in_(author_handles) ) ) ).all() ) for handle, itype in identity_rows: author_types[handle] = itype # Query 3 — dependency DAG (partial, scoped to this page + neighbours) dag = await load_dag_for_proposals(session, proposal_ids) # Query 4 — latest conflict_scan simulation per proposal (for list summary) conflict_counts: dict[str, int | None] = {pid: None for pid in proposal_ids} if proposal_ids: sim_rows = list( ( await session.execute( select( MusehubProposalSimulation.proposal_id, MusehubProposalSimulation.result, ).where( MusehubProposalSimulation.proposal_id.in_(proposal_ids), MusehubProposalSimulation.simulation_type == "conflict_scan", ) ) ).all() ) for pid, result_json in sim_rows: if isinstance(result_json, dict): conflict_counts[pid] = result_json.get("conflict_count") return _ProposalPrefetch( reviews_by_proposal=reviews_by_proposal, author_types=author_types, dag=dag, conflict_counts=conflict_counts, ) def _enrich_one( proposal: MusehubProposal, prefetch: _ProposalPrefetch, ) -> ProposalListEntry: """Compute all display-facing fields for a single proposal list row. This is the single source of truth for what the proposals list view renders per row. It does not issue any DB queries — all needed data comes from ``prefetch``, which is populated by ``_prefetch_for_batch`` before this function is called. Computed fields (all server-side): - active_domains: domains with non-zero risk_score - domain_risk / domain_risk_band: derived from proposal.risk_score (currently a single code-domain score; extended as more domains land) - aggregate_risk_score: weighted mean across active domains - aggregate_risk_band: band for the aggregate score - approval_count / domains_approved / domains_pending_review: derived from pre-fetched reviews - all_merge_conditions_met: approval_count >= required_approvals and breakage_count == 0 - author_type: resolved from pre-fetched MusehubIdentity rows Performance contract: Zero DB I/O. All reads come from the ``prefetch`` maps. Typical wall time: < 1ms per row on warm prefetch data. Args: proposal: ORM row for this proposal. prefetch: Pre-fetched batch data from ``_prefetch_for_batch``. Returns: ``ProposalListEntry`` with all fields populated. Raises: ValueError: If ``proposal.risk_score`` is outside ``[0.0, 1.0]``. """ pid = proposal.proposal_id # ── Risk ───────────────────────────────────────────────────────────────── # Use dimensional_risk dict (Phase 1 ORM columns) when present; fall back # to the scalar risk_score as code-domain risk for backwards compatibility. raw_dimensional = dict(getattr(proposal, "dimensional_risk", None) or {}) if raw_dimensional: domain_risk = {d: float(v) for d, v in raw_dimensional.items() if float(v) > 0.0} else: code_risk = float(proposal.risk_score or 0.0) if not (0.0 <= code_risk <= 1.0): raise ValueError(f"proposal {pid}: risk_score {code_risk!r} out of [0, 1]") domain_risk = {"code": code_risk} if code_risk > 0.0 else {} active_domains = list(domain_risk.keys()) domain_risk_band = {d: _score_to_band(v) for d, v in domain_risk.items()} # Weighted aggregate if domain_risk: total_weight = sum(_DOMAIN_WEIGHTS.get(d, 1.0) for d in domain_risk) aggregate_risk_score = sum( v * _DOMAIN_WEIGHTS.get(d, 1.0) for d, v in domain_risk.items() ) / total_weight else: aggregate_risk_score = 0.0 aggregate_risk_band = _score_to_band(aggregate_risk_score) # ── Dependency position ─────────────────────────────────────────────────── dag = prefetch.dag dep_blocked_by = blocked_by_numbers(dag, pid) dep_blocks = blocks_numbers(dag, pid) dep_is_blocked = is_blocked(dag, pid) mc_raw = proposal.merge_conditions or {} require_dep_merged: bool = mc_raw.get("require_dependency_merged", True) deps_satisfied = (not require_dep_merged) or (not dep_is_blocked) # ── Reviews ─────────────────────────────────────────────────────────────── reviews = prefetch.reviews_by_proposal.get(pid, []) approved_reviews = [r for r in reviews if r.state == "approved"] approval_count = len(approved_reviews) required_approvals = _DEFAULT_REQUIRED_APPROVALS domains_approved = ["code"] if approval_count > 0 and "code" in active_domains else [] domains_pending_review = [d for d in active_domains if d not in domains_approved] all_merge_conditions_met = ( approval_count >= required_approvals and proposal.breakage_count == 0 and deps_satisfied ) # ── Author type ─────────────────────────────────────────────────────────── author_type = prefetch.author_types.get(proposal.author, "human") agent_model: str | None = getattr(proposal, "agent_model", None) agent_spawned_by: str | None = getattr(proposal, "agent_spawned_by", None) # ── Symbol preview ──────────────────────────────────────────────────────── touched = list(proposal.touched_symbols or []) touched_symbols_preview = touched[:3] return ProposalListEntry( proposal_id=proposal.proposal_id, proposal_number=proposal.proposal_number, title=(proposal.title[:80] + "…") if len(proposal.title) > 80 else proposal.title, state=proposal.state, proposal_type=getattr(proposal, "proposal_type", "state_merge"), from_branch=proposal.from_branch, to_branch=proposal.to_branch, author=proposal.author, author_type=author_type, created_at=proposal.created_at, merged_at=proposal.merged_at, is_draft=getattr(proposal, "is_draft", proposal.state == "drafting"), active_domains=active_domains, domain_risk=domain_risk, domain_risk_band=domain_risk_band, aggregate_risk_score=round(aggregate_risk_score, 4), aggregate_risk_band=aggregate_risk_band, approval_count=approval_count, required_approvals=required_approvals, domains_approved=domains_approved, domains_pending_review=domains_pending_review, all_merge_conditions_met=all_merge_conditions_met, blocked_by=dep_blocked_by, blocks=dep_blocks, is_blocked=dep_is_blocked, symbols_changed=proposal.symbols_changed, breakage_count=proposal.breakage_count, test_gap_count=proposal.test_gap_count, touched_symbols_preview=touched_symbols_preview, midi_tracks_changed=getattr(proposal, "midi_tracks_changed", 0), midi_notes_delta=getattr(proposal, "midi_notes_delta", 0), harmonic_tension_delta=getattr(proposal, "harmonic_tension_delta", None), payment_claim_count=getattr(proposal, "payment_claim_count", 0), payment_ledger_delta_nano=getattr(proposal, "payment_ledger_delta_nano", 0), payment_avax_address=getattr(proposal, "payment_avax_address", None), payment_settling=proposal.state == "settling" and "pay" in active_domains, agent_model=agent_model, agent_spawned_by=agent_spawned_by, merge_strategy=getattr(proposal, "merge_strategy", "overlay") or "overlay", simulation_conflict_count=prefetch.conflict_counts.get(pid), ) async def enrich_proposal_list_entry( proposal: MusehubProposal, session: AsyncSession, ) -> ProposalListEntry: """Compute all display-facing fields for a single proposal list row. Convenience wrapper around ``_enrich_one`` for callers that need to enrich a single proposal without a pre-existing batch context. Issues its own prefetch queries (2 DB round-trips). For a full page (≥2 proposals) prefer ``enrich_proposal_list_batch`` which amortises the prefetch cost across all rows via a single parallel pass. Args: proposal: ORM row for this proposal. Must have ``repo_id`` set. session: Async database session. Returns: ``ProposalListEntry`` with all fields populated. Raises: ValueError: If ``proposal.risk_score`` is outside ``[0.0, 1.0]``. """ prefetch = await _prefetch_for_batch([proposal], session) return _enrich_one(proposal, prefetch) async def enrich_proposal_list_batch( proposals: list[MusehubProposal], session: AsyncSession, ) -> list[ProposalListEntry]: """Enrich a full page of proposal rows in a single parallel pass. Pre-fetch strategy: Issues exactly 2 DB queries for the entire batch (reviews + identity types), then calls ``_enrich_one`` for each row synchronously. The synchronous per-row work is CPU-only (no I/O), so no concurrency overhead is needed. Ordering: Result list is in the same order as ``proposals``. Args: proposals: ORM rows for the current page (typically ≤ 20). session: Async session shared across the batch. Returns: List of ``ProposalListEntry`` in the same order as ``proposals``. Performance target: < 50ms for 20 proposals (dominated by the 2 DB queries; per-row computation is < 0.1ms). """ if not proposals: return [] prefetch = await _prefetch_for_batch(proposals, session) return [_enrich_one(p, prefetch) for p in proposals] async def get_domain_heat( repo_id: str, state: str, session: AsyncSession, ) -> DomainHeatResponse: """Return per-domain proposal counts and average risk for the heat bar. Runs a single aggregation query against ``musehub_proposals`` filtered by ``repo_id`` and ``state``. The heat bar currently reflects the code domain only (the only domain with populated risk scores at this phase); additional domains are added as multi-domain risk rows land in Phase 2. ``avg_risk`` is the arithmetic mean of non-zero ``risk_score`` values for proposals in the given state. Domains with zero matching proposals are omitted from the response dict. Args: repo_id: Repository to query. state: Proposal state filter (e.g. ``"open"``). Pass ``"open"`` for the standard heat bar view. Pass ``"all"`` to skip the state filter entirely. session: Async session. Returns: ``DomainHeatResponse`` with ``domains`` dict and ``total_open`` count. Performance target: < 20ms (single aggregation query, no per-row work). """ conditions = [MusehubProposal.repo_id == repo_id] if state != "all": conditions.append(MusehubProposal.state == state) total: int = ( await session.execute( select(func.count(MusehubProposal.proposal_id)).where(*conditions) ) ).scalar_one() # Code domain: all proposals in this repo/state (code is the only domain for now). # avg_risk computed from non-null, non-zero risk_score values only. risk_rows = list( ( await session.execute( select(MusehubProposal.risk_score).where( *conditions, MusehubProposal.risk_score.isnot(None), MusehubProposal.risk_score > 0.0, ) ) ).scalars() ) avg_risk = round(sum(risk_rows) / len(risk_rows), 4) if risk_rows else 0.0 # All known domains — code count = total (single-domain repos); others = 0. # Multi-domain heat will populate midi/stems/pay when those proposals land. domains: dict[str, DomainHeatEntry] = { "code": DomainHeatEntry(count=total, avg_risk=avg_risk), "midi": DomainHeatEntry(count=0, avg_risk=0.0), "stems": DomainHeatEntry(count=0, avg_risk=0.0), "pay": DomainHeatEntry(count=0, avg_risk=0.0), } return DomainHeatResponse(domains=domains, total_open=total) async def get_merge_readiness( repo_id: str, session: AsyncSession, ) -> MergeReadinessResponse: """Bucket all non-merged proposals into readiness categories. Categories: ready: approval_count >= required threshold AND breakage_count == 0 settling: state == 'settling' needs_review: not settling, conditions not fully met Dependency-blocked proposals (``blocked_by`` non-empty) are not yet tracked in the DB at this phase; ``blocked`` will always be empty until the dependency graph table lands in Phase 2. Runs in one DB query; no per-row enrichment. Args: repo_id: Repository to query. session: Async session. Returns: ``MergeReadinessResponse`` with ``ready``, ``blocked``, ``settling``, and ``needs_review`` lists of proposal numbers. Performance target: < 20ms. """ rows = list( ( await session.execute( select( MusehubProposal.proposal_number, MusehubProposal.state, MusehubProposal.breakage_count, ).where( MusehubProposal.repo_id == repo_id, MusehubProposal.state.notin_(["merged", "abandoned"]), ) ) ).all() ) # Pre-fetch approval counts in one query proposal_numbers_to_check = [r.proposal_number for r in rows] if proposal_numbers_to_check: approval_counts_rows = list( ( await session.execute( select( MusehubProposal.proposal_number, func.count(MusehubProposalReview.review_id).label("approved_count"), ) .join( MusehubProposalReview, (MusehubProposalReview.proposal_id == MusehubProposal.proposal_id) & (MusehubProposalReview.state == "approved"), isouter=True, ) .where( MusehubProposal.repo_id == repo_id, MusehubProposal.proposal_number.in_(proposal_numbers_to_check), ) .group_by(MusehubProposal.proposal_number) ) ).all() ) approval_by_number: dict[int, int] = {r.proposal_number: r.approved_count for r in approval_counts_rows} else: approval_by_number = {} ready: list[int] = [] blocked: list[int] = [] settling: list[int] = [] needs_review: list[int] = [] for row in rows: if row.state == "settling": settling.append(row.proposal_number) elif ( approval_by_number.get(row.proposal_number, 0) >= _DEFAULT_REQUIRED_APPROVALS and row.breakage_count == 0 ): ready.append(row.proposal_number) else: needs_review.append(row.proposal_number) return MergeReadinessResponse( ready=ready, blocked=blocked, settling=settling, needs_review=needs_review, ) # ───────────────────────────────────────────────────────────────────────────── # Phase 4 — Simulation Engine # ───────────────────────────────────────────────────────────────────────────── _VALID_SIMULATION_TYPES = {"conflict_scan", "risk_projection", "dependency_order"} def _to_simulation_response(row: MusehubProposalSimulation, *, is_stale: bool) -> SimulationResponse: return SimulationResponse( simulation_id=row.simulation_id, proposal_id=row.proposal_id, simulation_type=row.simulation_type, result=row.result, is_stale=is_stale, from_branch_commit_id=row.from_branch_commit_id, duration_ms=row.duration_ms, created_at=row.created_at, expires_at=row.expires_at, ) async def _current_from_commit( session: AsyncSession, repo_id: str, from_branch: str, ) -> str: """Return the current head_commit_id of from_branch, or '' if missing.""" branch = await _get_branch(session, repo_id, from_branch) return (branch.head_commit_id or "") if branch else "" async def run_simulation( session: AsyncSession, repo_id: str, proposal_id: str, simulation_type: str, ) -> SimulationResponse: """Run a simulation for the proposal and upsert the cached result. Always recomputes — use get_simulation to read the cache without re-running. Raises: ValueError: Unknown simulation_type or proposal not found. """ import time from musehub.services.proposal_simulation import ( simulate_conflict_scan, simulate_dependency_order, simulate_risk_projection, ) from musehub.services.musehub_snapshot import get_snapshot_manifest if simulation_type not in _VALID_SIMULATION_TYPES: raise ValueError( f"Unknown simulation_type '{simulation_type}'. " f"Valid types: {sorted(_VALID_SIMULATION_TYPES)}" ) stmt = select(MusehubProposal).where( MusehubProposal.repo_id == repo_id, MusehubProposal.proposal_id == proposal_id, ) proposal = (await session.execute(stmt)).scalar_one_or_none() if proposal is None: raise ValueError(f"Proposal {proposal_id} not found in repo {repo_id}") from_commit_id = await _current_from_commit(session, repo_id, proposal.from_branch) to_b = await _get_branch(session, repo_id, proposal.to_branch) from_b = await _get_branch(session, repo_id, proposal.from_branch) to_manifest: StrDict = {} if to_b and to_b.head_commit_id: to_head = await session.get(MusehubCommit, to_b.head_commit_id) if to_head and to_head.snapshot_id: to_manifest = await get_snapshot_manifest(session, to_head.snapshot_id) from_manifest: StrDict = {} if from_b and from_b.head_commit_id: from_head = await session.get(MusehubCommit, from_b.head_commit_id) if from_head and from_head.snapshot_id: from_manifest = await get_snapshot_manifest(session, from_head.snapshot_id) strategy_name = getattr(proposal, "merge_strategy", "overlay") or "overlay" selective_domains: list[str] | None = getattr(proposal, "selective_domains", None) dimensional_risk: dict[str, float] | None = getattr(proposal, "dimensional_risk", None) ancestor_manifest: StrDict | None = None if strategy_name in ("weave", "replay", "selective", "phased"): ancestor_manifest = await _resolve_ancestor_manifest( session, repo_id, proposal.from_branch, proposal.to_branch ) t0 = time.monotonic() if simulation_type == "conflict_scan": result_payload = simulate_conflict_scan( to_manifest, from_manifest, ancestor_manifest=ancestor_manifest, strategy=strategy_name, selective_domains=selective_domains, ) elif simulation_type == "risk_projection": result_payload = simulate_risk_projection( to_manifest, from_manifest, ancestor_manifest=ancestor_manifest, current_dimensional_risk=dimensional_risk, strategy=strategy_name, selective_domains=selective_domains, ) else: # dependency_order dag = await load_dag_for_proposals(session, [proposal_id]) result_payload = simulate_dependency_order(dag) duration_ms = int((time.monotonic() - t0) * 1000) sim_id = compute_simulation_id(proposal_id, simulation_type, from_commit_id) now = _utc_now() # Upsert: update on conflict (same proposal_id + simulation_type) existing_stmt = select(MusehubProposalSimulation).where( MusehubProposalSimulation.proposal_id == proposal_id, MusehubProposalSimulation.simulation_type == simulation_type, ) existing = (await session.execute(existing_stmt)).scalar_one_or_none() if existing is not None: existing.simulation_id = sim_id existing.from_branch_commit_id = from_commit_id existing.result = result_payload existing.duration_ms = duration_ms existing.created_at = now row = existing else: row = MusehubProposalSimulation( simulation_id=sim_id, proposal_id=proposal_id, simulation_type=simulation_type, from_branch_commit_id=from_commit_id, result=result_payload, duration_ms=duration_ms, created_at=now, ) session.add(row) await session.flush() logger.info( "🔬 Simulation %s for proposal %s (%dms)", simulation_type, proposal_id, duration_ms ) return _to_simulation_response(row, is_stale=False) async def get_simulation( session: AsyncSession, repo_id: str, proposal_id: str, simulation_type: str, ) -> SimulationResponse | None: """Return the cached simulation result, or None if never run. Sets ``is_stale=True`` when the from_branch has advanced since the simulation was last run. Does NOT re-run; call run_simulation for that. """ if simulation_type not in _VALID_SIMULATION_TYPES: raise ValueError( f"Unknown simulation_type '{simulation_type}'. " f"Valid types: {sorted(_VALID_SIMULATION_TYPES)}" ) stmt = select(MusehubProposalSimulation).where( MusehubProposalSimulation.proposal_id == proposal_id, MusehubProposalSimulation.simulation_type == simulation_type, ) row = (await session.execute(stmt)).scalar_one_or_none() if row is None: return None # Staleness check: compare stored commit ID with current branch tip proposal_stmt = select(MusehubProposal.from_branch).where( MusehubProposal.proposal_id == proposal_id, MusehubProposal.repo_id == repo_id, ) from_branch = (await session.execute(proposal_stmt)).scalar_one_or_none() is_stale = False if from_branch: current_commit = await _current_from_commit(session, repo_id, from_branch) is_stale = bool(current_commit) and current_commit != row.from_branch_commit_id return _to_simulation_response(row, is_stale=is_stale) async def list_simulations( session: AsyncSession, repo_id: str, proposal_id: str, ) -> SimulationListResponse: """Return all cached simulations for a proposal. Includes staleness flags. Never re-runs. """ # Resolve from_branch once for staleness checks proposal_stmt = select(MusehubProposal.from_branch).where( MusehubProposal.proposal_id == proposal_id, MusehubProposal.repo_id == repo_id, ) from_branch = (await session.execute(proposal_stmt)).scalar_one_or_none() current_commit = "" if from_branch: current_commit = await _current_from_commit(session, repo_id, from_branch) rows_stmt = select(MusehubProposalSimulation).where( MusehubProposalSimulation.proposal_id == proposal_id, ) rows = list((await session.execute(rows_stmt)).scalars().all()) responses = [ _to_simulation_response( r, is_stale=bool(current_commit) and current_commit != r.from_branch_commit_id, ) for r in rows ] return SimulationListResponse(simulations=responses, total=len(responses))