open-swe/agent/utils/reviewer_outcomes.py

"""Persist reviewer finding outcomes to a LangSmith dataset.

Every time a published finding is resolved (a later commit fixed it),
dismissed (false positive / human pushback), or gets a 👍/👎 reaction, we
upsert one labelled example into a single LangSmith dataset. The ``analyzer``
graph reads these per-repo to refine its review-style prompts: confirmed
findings teach what to hunt for, dismissed ones teach what to skip.

Writes are best-effort and deterministic (one example id per
``finding_id`` + ``label_source``) so re-processing the same transition
updates in place instead of duplicating.
"""

from __future__ import annotations

import logging
import os
import uuid
from typing import Any

from langsmith import Client as LangSmithClient

logger = logging.getLogger(__name__)

OUTCOMES_DATASET_NAME = os.environ.get("REVIEWER_OUTCOMES_DATASET", "openswe-reviewer-outcomes")

TRUE_POSITIVE = "true_positive"
FALSE_POSITIVE = "false_positive"

_DIFF_HUNK_MAX_CHARS = 4000


def outcome_from_status(
    status: str,
    *,
    first_seen_sha: str | None,
    head_sha: str | None,
) -> tuple[str, str] | None:
    """Map a finding status transition to ``(label, label_source)``.

    ``resolved`` is a positive signal: we flagged something and it was closed.
    When a commit landed between first sighting and resolution we tag it
    ``resolved_by_commit`` (a fix shipped); otherwise ``resolved_same_sha``.
    """
    if status == "resolved":
        if head_sha and first_seen_sha and head_sha != first_seen_sha:
            return TRUE_POSITIVE, "resolved_by_commit"
        return TRUE_POSITIVE, "resolved_same_sha"
    if status == "dismissed":
        return FALSE_POSITIVE, "dismissed"
    return None


def outcome_from_score(score: float | None, *, source: str) -> tuple[str, str] | None:
    """Map a reaction score (1.0 👍 / 0.0 👎) to ``(label, label_source)``."""
    if score is None:
        return None
    if score >= 1.0:
        return TRUE_POSITIVE, f"{source}_thumbs_up"
    return FALSE_POSITIVE, f"{source}_thumbs_down"


def _outcomes_client() -> LangSmithClient | None:
    """Build a single LangSmith client, preferring the prod tenant."""
    prod_key = os.environ.get("LANGSMITH_API_KEY_PROD")
    if prod_key:
        api_url = os.environ.get("LANGSMITH_ENDPOINT_PROD") or os.environ.get(
            "LANGSMITH_ENDPOINT", "https://api.smith.langchain.com"
        )
        return LangSmithClient(api_key=prod_key, api_url=api_url)
    api_key = os.environ.get("LANGSMITH_API_KEY") or os.environ.get("LANGCHAIN_API_KEY")
    if not api_key:
        return None
    api_url = os.environ.get("LANGSMITH_ENDPOINT", "https://api.smith.langchain.com")
    return LangSmithClient(api_key=api_key, api_url=api_url)


def _ensure_dataset(client: LangSmithClient) -> Any:
    existing = next((d for d in client.list_datasets(dataset_name=OUTCOMES_DATASET_NAME)), None)
    if existing is not None:
        return existing.id
    ds = client.create_dataset(
        dataset_name=OUTCOMES_DATASET_NAME,
        description=(
            "Open SWE reviewer finding outcomes (resolved / dismissed / 👍👎) "
            "captured in production for per-repo continual learning."
        ),
    )
    return ds.id


def _example_id(repo: str, finding_id: str, label_source: str) -> uuid.UUID:
    return uuid.uuid5(uuid.NAMESPACE_URL, f"finding-outcome:{repo}:{finding_id}:{label_source}")


def _truncate(value: Any, limit: int) -> Any:
    if isinstance(value, str) and len(value) > limit:
        return value[:limit]
    return value


def _create_or_update_example(
    client: LangSmithClient,
    *,
    dataset_id: Any,
    example_id: uuid.UUID,
    inputs: dict[str, Any],
    outputs: dict[str, Any],
    metadata: dict[str, Any],
) -> None:
    try:
        client.create_example(
            inputs=inputs,
            outputs=outputs,
            metadata=metadata,
            dataset_id=dataset_id,
            example_id=example_id,
        )
    except Exception:  # noqa: BLE001 — example already exists; update in place
        client.update_example(
            example_id=example_id,
            inputs=inputs,
            outputs=outputs,
            metadata=metadata,
        )


def upsert_finding_outcome(
    finding: dict[str, Any],
    *,
    label: str,
    label_source: str,
    repo: str,
    pr_number: int | None = None,
    pr_url: str = "",
    base_sha: str = "",
    head_sha: str = "",
    run_id: str | None = None,
    thread_id: str | None = None,
) -> bool:
    """Upsert one finding-level outcome example. Best-effort; never raises."""
    finding_id = str(finding.get("id") or "")
    if not finding_id or not repo:
        return False
    client = _outcomes_client()
    if client is None:
        logger.debug("No LangSmith client configured; skipping outcome example")
        return False
    try:
        dataset_id = _ensure_dataset(client)
        inputs = {
            "repo": repo,
            "pr_number": pr_number,
            "pr_url": pr_url,
            "file": finding.get("file"),
            "start_line": finding.get("start_line"),
            "end_line": finding.get("end_line"),
            "side": finding.get("side", "RIGHT"),
            "diff_hunk": _truncate(finding.get("diff_hunk"), _DIFF_HUNK_MAX_CHARS),
            "base_sha": base_sha,
            "head_sha": head_sha,
        }
        outputs = {
            "label": label,
            "label_source": label_source,
            "finding": {
                "title": finding.get("title"),
                "description": _truncate(finding.get("description"), _DIFF_HUNK_MAX_CHARS),
                "severity": finding.get("severity"),
                "confidence": finding.get("confidence"),
                "category": finding.get("category"),
            },
            "resolution_note": finding.get("resolution_note"),
        }
        metadata = {
            "granularity": "finding",
            "repo": repo,
            "finding_id": finding_id,
            "label": label,
            "label_source": label_source,
            "run_id": run_id or finding.get("github_review_run_id"),
            "thread_id": thread_id,
            "first_seen_sha": finding.get("first_seen_sha"),
        }
        _create_or_update_example(
            client,
            dataset_id=dataset_id,
            example_id=_example_id(repo, finding_id, label_source),
            inputs=inputs,
            outputs=outputs,
            metadata=metadata,
        )
        return True
    except Exception:  # noqa: BLE001 — dataset writes must never break a review
        logger.exception("Failed to upsert reviewer finding outcome for %s", finding_id)
        return False


def upsert_run_outcome(
    *,
    label: str,
    label_source: str,
    run_id: str,
    repo: str | None = None,
    extra: dict[str, Any] | None = None,
) -> bool:
    """Upsert a coarse run-level outcome (e.g. a Slack 👍/👎 on a review).

    These have no finding/diff anchor, so they are tagged
    ``granularity="run"`` and ignored by the per-repo analyzer reader.
    """
    if not run_id:
        return False
    client = _outcomes_client()
    if client is None:
        return False
    try:
        dataset_id = _ensure_dataset(client)
        inputs = {"repo": repo, "run_id": run_id, **(extra or {})}
        outputs = {"label": label, "label_source": label_source}
        metadata = {
            "granularity": "run",
            "repo": repo,
            "run_id": run_id,
            "label": label,
            "label_source": label_source,
        }
        _create_or_update_example(
            client,
            dataset_id=dataset_id,
            example_id=uuid.uuid5(uuid.NAMESPACE_URL, f"run-outcome:{run_id}:{label_source}"),
            inputs=inputs,
            outputs=outputs,
            metadata=metadata,
        )
        return True
    except Exception:  # noqa: BLE001
        logger.exception("Failed to upsert run outcome for run %s", run_id)
        return False


def repo_full_name_from_config(configurable: dict[str, Any]) -> str:
    repo = configurable.get("repo") if isinstance(configurable, dict) else None
    if isinstance(repo, dict) and repo.get("owner") and repo.get("name"):
        return f"{repo['owner']}/{repo['name']}"
    return ""


def emit_finding_status_outcome(
    finding: dict[str, Any],
    status: str,
    *,
    configurable: dict[str, Any],
    thread_id: str | None = None,
) -> bool:
    """Map a resolve/dismiss transition to an outcome example and upsert it.

    Reads repo / PR / SHA context from the reviewer run ``configurable``.
    Best-effort: a no-op when the status isn't terminal or repo is unknown.
    """
    head_sha = str(configurable.get("head_sha") or "")
    mapping = outcome_from_status(
        status, first_seen_sha=finding.get("first_seen_sha"), head_sha=head_sha
    )
    if mapping is None:
        return False
    repo = repo_full_name_from_config(configurable)
    if not repo:
        return False
    label, label_source = mapping
    pr_number = configurable.get("pr_number")
    return upsert_finding_outcome(
        finding,
        label=label,
        label_source=label_source,
        repo=repo,
        pr_number=pr_number if isinstance(pr_number, int) else None,
        pr_url=str(configurable.get("pr_url") or ""),
        base_sha=str(configurable.get("base_sha") or ""),
        head_sha=head_sha,
        thread_id=thread_id,
    )


def read_outcomes_for_repo(repo: str, *, limit: int = 100) -> dict[str, list[dict[str, Any]]]:
    """Return confirmed (true-positive) and dismissed (false-positive) findings
    for ``repo`` from the outcomes dataset. Best-effort; returns empty on error."""
    confirmed: list[dict[str, Any]] = []
    dismissed: list[dict[str, Any]] = []
    client = _outcomes_client()
    if client is None or not repo:
        return {"confirmed": confirmed, "dismissed": dismissed}
    try:
        existing = next((d for d in client.list_datasets(dataset_name=OUTCOMES_DATASET_NAME)), None)
        if existing is None:
            return {"confirmed": confirmed, "dismissed": dismissed}
        for example in client.list_examples(dataset_id=existing.id):
            metadata = getattr(example, "metadata", None) or {}
            if metadata.get("granularity") != "finding" or metadata.get("repo") != repo:
                continue
            outputs = getattr(example, "outputs", None) or {}
            inputs = getattr(example, "inputs", None) or {}
            finding = outputs.get("finding") or {}
            row = {
                "file": inputs.get("file"),
                "title": finding.get("title"),
                "description": finding.get("description"),
                "severity": finding.get("severity"),
                "category": finding.get("category"),
                "label_source": outputs.get("label_source"),
                "diff_hunk": inputs.get("diff_hunk"),
                "resolution_note": outputs.get("resolution_note"),
            }
            if outputs.get("label") == TRUE_POSITIVE:
                confirmed.append(row)
            elif outputs.get("label") == FALSE_POSITIVE:
                dismissed.append(row)
            if len(confirmed) + len(dismissed) >= limit:
                break
    except Exception:  # noqa: BLE001
        logger.exception("Failed to read outcomes for repo %s", repo)
    return {"confirmed": confirmed, "dismissed": dismissed}