revisions

def semantic_diff(old_revision: ArtifactRevision, new_revision: ArtifactRevision) -> set[str]:
    """Return the set of artifact facets whose content changed between two revisions (AC-3-2).

    Compares the two revisions' per-facet content hashes
    (:attr:`~strands_teams.models.ArtifactRevision.facets`). A facet is *changed*
    when its hash differs between the revisions, when it exists in only one of them
    (added or removed), and unchanged when both hashes are equal. The comparison is
    over hashes, not raw content, so it is exact and cheap.

    Args:
        old_revision: The earlier revision.
        new_revision: The later revision.

    Returns:
        The set of changed facet names. Empty when the two revisions carry
        identical facet content.
    """
    old_facets = old_revision.facets
    new_facets = new_revision.facets
    changed: set[str] = set()
    for name in old_facets.keys() | new_facets.keys():
        if old_facets.get(name) != new_facets.get(name):
            changed.add(name)
    return changed

def invalidates_review(
    old_binding: ArtifactRevision,
    new_revision: ArtifactRevision,
    review: ReviewBinding | Any,
) -> ReviewVerdict:
    """Decide whether ``new_revision`` invalidates a review bound to ``old_binding`` (AC-3-3..5).

    Runs :func:`semantic_diff` between the revisions and intersects the changed
    facets with the facets the review covers:

    - **invalidated** when the intersection is non-empty — a facet the review
      examined changed, so the review must be re-opened (AC-3-3).
    - **valid** when the intersection is empty — the review's facets are untouched,
      so it is carried forward and reused rather than recomputed (AC-3-4). A review
      that covers no specific facets (empty ``reviewed_facets``) is treated as
      covering the WHOLE artifact, so any changed facet invalidates it.
    - **error** when ``new_revision``'s lineage does not supersede the review's
      bound revision — the verdict cannot be trusted, so it is reported as an error
      rather than silently treated as valid or invalidated (AC-3-5).

    Args:
        old_binding: The revision the review was produced against.
        new_revision: The candidate new revision.
        review: A revision-bound review — a :class:`Review` or a Phase 1
            :class:`~strands_teams.models.Finding` (consumed structurally).

    Returns:
        A :class:`ReviewVerdict` with the status, reason, and the changed /
        intersecting facet sets.
    """
    bound_revision, covered = _review_binding(review)

    if not _lineage_supersedes(old_binding, new_revision, bound_revision):
        return ReviewVerdict(
            status="error",
            reason=(
                f"new_revision {new_revision.revision_id!r} does not supersede the review's bound revision "
                f"{bound_revision or old_binding.revision_id!r} (lineage: supersedes="
                f"{new_revision.supersedes}, parents={new_revision.parents})"
            ),
        )

    changed = semantic_diff(old_binding, new_revision)
    # Empty covered-facet set means the review covers the whole artifact: any change touches it.
    intersecting = sorted(changed) if not covered else sorted(changed & set(covered))

    if intersecting:
        return ReviewVerdict(
            status="invalidated",
            reason=f"changed facets {intersecting} intersect the review's covered facets",
            changed_facets=sorted(changed),
            intersecting_facets=intersecting,
        )
    return ReviewVerdict(
        status="valid",
        reason="no changed facet intersects the review's covered facets — reuse the review",
        changed_facets=sorted(changed),
        intersecting_facets=[],
    )

def reconcile_reviews(
    old_binding: ArtifactRevision,
    new_revision: ArtifactRevision,
    reviews: Iterable[ReviewBinding | Any],
) -> ReviewReconciliation:
    """Partition bound reviews into carried-forward, re-opened, and errored (AC-3-6).

    When a re-run promotes a new revision of a logical artifact, every bound review
    that :func:`invalidates_review` reports as still valid is carried forward
    (reused), only the invalidated reviews are re-opened, and any lineage-mismatch
    review is surfaced as an error (AC-3-5, AC-3-6).

    Args:
        old_binding: The revision the reviews were produced against.
        new_revision: The newly promoted revision.
        reviews: The bound reviews to reconcile.

    Returns:
        A :class:`ReviewReconciliation` grouping the verdicts.
    """
    carried: list[ReviewVerdict] = []
    reopened: list[ReviewVerdict] = []
    errored: list[ReviewVerdict] = []
    for review in reviews:
        verdict = invalidates_review(old_binding, new_revision, review)
        if verdict.status == "valid":
            carried.append(verdict)
        elif verdict.status == "invalidated":
            reopened.append(verdict)
        else:
            errored.append(verdict)
    return ReviewReconciliation(carried_forward=carried, reopened=reopened, errored=errored)

@tool(context=True)
def promote_artifact(
    logical_name: str,
    expected_current_revision: str,
    proposed_revision: dict[str, Any],
    tool_context: ToolContext,
) -> str:
    """Promote a new revision of a logical artifact by optimistic compare-and-set.

    Makes ``proposed_revision`` the current revision of ``logical_name`` ONLY when
    ``expected_current_revision`` still matches the alias's stored current revision
    — a durable optimistic lock. If another promotion changed the current revision
    since you read it, this returns a ``status="stale"`` error and leaves the alias
    unchanged; re-read the current revision and retry. Pass the sentinel ``"∅"`` as
    ``expected_current_revision`` for the FIRST promotion of a logical name.

    The proposed revision's content is content-addressed: supply its content as
    ``proposed_revision={"facets": {"<facet>": "<content>", ...}}`` (preferred, so
    facet-level review invalidation works) or ``{"body": "<content>"}``; the
    revision id and content hash are computed from that content. On success the
    prior current revision is recorded in the new revision's ``supersedes`` lineage.

    Args:
        logical_name: The logical artifact to promote (e.g. ``"BUNDLE.auth"``).
        expected_current_revision: The revision id you believe is current, or the
            sentinel ``"∅"`` for the first promotion.
        proposed_revision: The new revision's content as a JSON dict — ``facets``
            (a ``{name: content}`` map) and/or ``body`` (a single string), plus
            optional ``artifact_type``, ``produced_by``, ``parents``,
            ``classification``.
        tool_context: Framework-injected run context (invocation state).

    Returns:
        On success, a JSON object with ``logical_name``, the new current
        ``current_revision`` id, and the ``superseded_revision`` id (the sentinel
        when this was the first revision). On a failed compare-and-set,
        ``{"status": "stale", "message": ..., "expected": ..., "actual": ...}`` so
        you can re-read and retry.
    """
    layout = _layout_of(tool_context)
    try:
        proposed = _coerce_proposed_revision(logical_name, proposed_revision)
    except (PydanticValidationError, ValueError, TypeError) as e:
        return json.dumps({"status": "error", "message": f"invalid proposed_revision: {e}"})

    try:
        stored = layout.promote_revision(logical_name, expected_current_revision, proposed)
    except StaleArtifactError as e:
        logger.info(f"promote_artifact({logical_name}): stale CAS — expected {e.expected!r}, actual {e.actual!r}")
        return json.dumps(
            {
                "status": "stale",
                "message": str(e),
                "logical_name": logical_name,
                "expected": e.expected,
                "actual": e.actual,
            },
        )
    except OSError as e:
        return json.dumps({"status": "error", "message": f"promotion failed: {e}"})

    superseded = expected_current_revision if expected_current_revision != NO_CURRENT_REVISION else NO_CURRENT_REVISION
    result = PromotionResult(
        logical_name=logical_name,
        current_revision=stored.revision_id,
        superseded_revision=superseded,
    )
    return result.model_dump_json()