diff --git a/AGENTS.md b/AGENTS.md index 7059f57f..430f5498 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,7 +29,7 @@ make format # ruff format + ruff check --fix |---|---|---| | `agent` | `agent.server:get_agent` | Main coding agent (Slack/Linear/GitHub-triggered). | | `reviewer` | `agent.reviewer:get_reviewer_agent` | Read-only PR reviewer. Findings model + `publish_review`. | -| `review_style_analyzer` | `agent.review_style_analyzer:get_review_style_analyzer` | Learns per-repo reviewer style from historical PRs. | +| `analyzer` | `agent.analyzer:get_analyzer` | Learns per-repo reviewer style from historical PRs and this reviewer's own finding outcomes. | The FastAPI app is `agent.webapp:app`. @@ -39,7 +39,7 @@ The FastAPI app is `agent.webapp:app`. - **`agent/server.py` → `get_agent(config)`** — main graph factory. Called per-thread. Resolves the GitHub token, gets-or-creates the sandbox for the thread, resolves the team/profile/per-thread model + effort, then constructs a fresh `create_deep_agent(...)` with the curated tool list and middleware stack. The agent itself is stateless — all per-thread state lives in the sandbox + thread metadata. - **`agent/reviewer.py` → `get_reviewer_agent(config)`** — reviewer graph factory. Shares `ensure_sandbox_for_thread` with the main agent but wires a reviewer-only toolset (`add_finding`, `update_finding`, `list_findings`, `publish_review`, `web_search`, `fetch_url`, `http_request`) and a different system prompt that pins the single-evolving-findings model and the diff-anchored bar for filing a finding. Read-only: no commit/push/PR-opening tools. -- **`agent/review_style_analyzer.py` → `get_review_style_analyzer(config)`** — small graph that reads historical PR reviews from a repo and emits a per-repo style prompt via the `save_review_style_prompt` tool. Output is consumed by the reviewer as a "repository-specific review style" appendix. +- **`agent/analyzer.py` → `get_analyzer(config)`** — small graph that emits a per-repo style prompt via the `save_review_style_prompt` tool, consumed by the reviewer as a "repository-specific review style" appendix. It runs in one of two modes (`analyzer_mode` in `configurable`): **bootstrap** (cold-start: crawl historical PR reviews) and **continual** (nightly: refine using this reviewer's own finding outcomes via `read_finding_outcomes`). Each mode's procedure lives in a deepagents **skill** (`agent/skills/bootstrap-repo-analysis/`, `agent/skills/continual-learning/`) served as virtual files via a `CompositeBackend` `/skills/` route + `StateBackend` (seeded into the run's `files` channel by the launcher — never written to the sandbox). Launchers and the per-repo nightly cron live in `agent/dashboard/review_style_jobs.py` and `agent/dashboard/analyzer_cron.py`; the cron is registered when bootstrap completes. - **`agent/webapp.py`** — custom FastAPI routes mounted alongside the LangGraph server. Webhooks land here (GitHub, Linear, Slack). Each webhook resolves a deterministic `thread_id` (so follow-up messages route to the same agent run) and triggers/streams a run via the `langgraph_sdk` client. Also auto-reviews PRs on `opened` / `ready_for_review` events when the repo+author opt in. - **`agent/dashboard/`** — `router` mounted under the FastAPI app at startup (`app.include_router(dashboard_router)`). Owns GitHub OAuth, per-user profiles, admin endpoints, team defaults, enabled-repo lists, review-style management, and the Agents chat thread API used by the UI in `ui/`. @@ -48,7 +48,7 @@ The FastAPI app is `agent.webapp:app`. `SANDBOX_BACKENDS` (in `agent/utils/sandbox_state.py`) is an in-process dict keyed by `thread_id`. Thread metadata persists `sandbox_id` across processes. `ensure_sandbox_for_thread` handles four cases: 1. Sandbox cached in memory → ping it (`echo ok`); recreate on `SandboxClientError`. Healthy reused sandboxes also get a GitHub-proxy refresh (recreate on failure). -2. Metadata says `__creating__` and no cache → poll until ready (`_wait_for_sandbox_id`). +2. Metadata says `__creating__` and no cache → reset stale metadata so a fresh sandbox can be created. 3. No sandbox at all → set `__creating__` sentinel, create one, persist the real id. 4. Metadata has an id but no cache → reconnect; fall back to recreate on failure. diff --git a/CLAUDE.md b/CLAUDE.md index 6065a271..f70c7bd3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -29,7 +29,7 @@ make format # ruff format + ruff check --fix |---|---|---| | `agent` | `agent.server:get_agent` | Main coding agent (Slack/Linear/GitHub-triggered). | | `reviewer` | `agent.reviewer:get_reviewer_agent` | Read-only PR reviewer. Findings model + `publish_review`. | -| `review_style_analyzer` | `agent.review_style_analyzer:get_review_style_analyzer` | Learns per-repo reviewer style from historical PRs. | +| `analyzer` | `agent.analyzer:get_analyzer` | Learns per-repo reviewer style from historical PRs and this reviewer's own finding outcomes. | The FastAPI app is `agent.webapp:app`. @@ -39,7 +39,7 @@ The FastAPI app is `agent.webapp:app`. - **`agent/server.py` → `get_agent(config)`** — main graph factory. Called per-thread. Resolves the GitHub token, gets-or-creates the sandbox for the thread, resolves the team/profile/per-thread model + effort, then constructs a fresh `create_deep_agent(...)` with the curated tool list and middleware stack. The agent itself is stateless — all per-thread state lives in the sandbox + thread metadata. - **`agent/reviewer.py` → `get_reviewer_agent(config)`** — reviewer graph factory. Shares `ensure_sandbox_for_thread` with the main agent but wires a reviewer-only toolset (`add_finding`, `update_finding`, `list_findings`, `publish_review`, `web_search`, `fetch_url`, `http_request`) and a different system prompt that pins the single-evolving-findings model and the diff-anchored bar for filing a finding. Read-only: no commit/push/PR-opening tools. -- **`agent/review_style_analyzer.py` → `get_review_style_analyzer(config)`** — small graph that reads historical PR reviews from a repo and emits a per-repo style prompt via the `save_review_style_prompt` tool. Output is consumed by the reviewer as a "repository-specific review style" appendix. +- **`agent/analyzer.py` → `get_analyzer(config)`** — small graph that emits a per-repo style prompt via the `save_review_style_prompt` tool, consumed by the reviewer as a "repository-specific review style" appendix. It runs in one of two modes (`analyzer_mode` in `configurable`): **bootstrap** (cold-start: crawl historical PR reviews) and **continual** (nightly: refine using this reviewer's own finding outcomes via `read_finding_outcomes`). Each mode's procedure lives in a deepagents **skill** (`agent/skills/bootstrap-repo-analysis/`, `agent/skills/continual-learning/`) served as virtual files via a `CompositeBackend` `/skills/` route + `StateBackend` (seeded into the run's `files` channel by the launcher — never written to the sandbox). Launchers and the per-repo nightly cron live in `agent/dashboard/review_style_jobs.py` and `agent/dashboard/analyzer_cron.py`; the cron is registered when bootstrap completes. - **`agent/webapp.py`** — custom FastAPI routes mounted alongside the LangGraph server. Webhooks land here (GitHub, Linear, Slack). Each webhook resolves a deterministic `thread_id` (so follow-up messages route to the same agent run) and triggers/streams a run via the `langgraph_sdk` client. Also auto-reviews PRs on `opened` / `ready_for_review` events when the repo+author opt in. - **`agent/dashboard/`** — `router` mounted under the FastAPI app at startup (`app.include_router(dashboard_router)`). Owns GitHub OAuth, per-user profiles, admin endpoints, team defaults, enabled-repo lists, review-style management, and the Agents chat thread API used by the UI in `ui/`. diff --git a/agent/review_style_analyzer.py b/agent/analyzer.py similarity index 64% rename from agent/review_style_analyzer.py rename to agent/analyzer.py index 77a2c83a..d89d264f 100644 --- a/agent/review_style_analyzer.py +++ b/agent/analyzer.py @@ -1,4 +1,8 @@ -"""Review style analyzer graph. +"""Analyzer graph. + +Learns a per-repo review-style prompt for the reviewer agent. It mines +historical human PR review feedback and this reviewer's own past finding +outcomes (resolved / dismissed / 👍👎) to teach what this team flags and skips. Uses the same sandbox + ``gh`` pattern as the reviewer agent. The dashboard user's OAuth token is injected into the LangSmith GitHub proxy so ``gh`` works @@ -20,7 +24,9 @@ warnings.filterwarnings("ignore", module="langchain_core._api.deprecation") warnings.filterwarnings("ignore", message=".*Pydantic V1.*", category=UserWarning) from deepagents import create_deep_agent +from deepagents.backends.composite import CompositeBackend from deepagents.backends.protocol import SandboxBackendProtocol +from deepagents.backends.state import StateBackend from langchain.agents.middleware import ModelCallLimitMiddleware from .integrations.langsmith import _configure_github_proxy @@ -33,7 +39,10 @@ from .server import ( ensure_sandbox_for_thread, graph_loaded_for_execution, ) +from .tools.read_finding_outcomes import read_finding_outcomes from .tools.save_review_style import save_review_style_prompt +from .utils.analyzer_skills import SKILLS_ROUTE, skill_path_for_mode +from .utils.github_app import get_github_app_installation_token from .utils.model import DEFAULT_LLM_REASONING, make_model, provider_model_kwargs from .utils.sandbox_paths import aresolve_sandbox_work_dir from .utils.sandbox_state import unwrap_sandbox_backend @@ -42,55 +51,28 @@ logger = logging.getLogger(__name__) STYLE_ANALYZER_MODEL_CALL_LIMIT = 80 +# The per-mode procedure lives in the bundled SKILL.md playbooks (agent/skills/). +# This base prompt only orients the agent and points it at the right skill. STYLE_ANALYZER_PROMPT = """You are a code-review style analyst for `{repo_owner}/{repo_name}`. Sandbox: `{working_dir}`. Use the shell (``execute``) to run GitHub commands. +**Always invoke gh as:** `GH_TOKEN=dummy gh `. -**Always invoke gh as:** `GH_TOKEN=dummy gh ` +Your job is to produce/refine the per-repo review-style prompt and persist it with +`save_review_style_prompt`. -# How to research (required) +# Run mode: {mode} -Browse historical **merged** PR review feedback until you have catalogued at least -**8 substantive human** review comments (not bots). Suggested commands: +Read and follow the playbook for this mode, then proceed: -``` -GH_TOKEN=dummy gh pr list --repo {repo_owner}/{repo_name} --state merged --limit 30 -GH_TOKEN=dummy gh api repos/{repo_owner}/{repo_name}/pulls//reviews -GH_TOKEN=dummy gh api repos/{repo_owner}/{repo_name}/pulls//comments -GH_TOKEN=dummy gh api repos/{repo_owner}/{repo_name}/issues//comments -``` + read_file("{skill_path}", limit=1000) -If the first batch is sparse, increase `--limit` or walk older PR numbers. Skip -`[bot]` accounts and obvious automation (codecov, dependabot, etc.). - -Identify the top ~5 human reviewers by volume and note phrasing, severity, and -what they ignore. - -# When you may call `save_review_style_prompt` - -Only after real research. Your `custom_prompt` (400–1200 words) must teach our -reviewer agent this repo's norms: - -- What the team routinely flags vs skips (paraphrased patterns, not invented quotes) -- Severity calibration -- Tone and test expectations -- Repo-specific conventions -- Anti-patterns reviewers here avoid - -`analysis_summary`: 2–4 sentences for the dashboard. -Pass `prs_sampled`, `reviews_sampled`, and `top_reviewers` (comma-separated logins). - -Do **not** save a generic guide after one or two commands. Only after ~25+ merged -PRs with zero human feedback may you save a short conservative guide and say so in -`analysis_summary`. +Do not improvise the procedure — the skill is authoritative for how to gather +evidence and what to save. # Alignment with our reviewer agent {reviewer_themes} - -# Optional preloaded samples - -The user message may include pre-collected samples — verify and extend with ``gh``. """ @@ -104,7 +86,7 @@ async def _configure_sandbox_github_proxy( await asyncio.to_thread(_configure_github_proxy, backend.id, github_token) -async def get_review_style_analyzer(config: RunnableConfig) -> Pregel: +async def get_analyzer(config: RunnableConfig) -> Pregel: thread_id = config["configurable"].get("thread_id") config["recursion_limit"] = DEFAULT_RECURSION_LIMIT @@ -118,10 +100,20 @@ async def get_review_style_analyzer(config: RunnableConfig) -> Pregel: full_name = str(configurable.get("review_style_full_name") or "owner/repo") owner, _, name = full_name.partition("/") samples_text = str(configurable.get("review_style_samples_text") or "") + mode = str(configurable.get("analyzer_mode") or "bootstrap") + github_token = configurable.get("review_style_github_token") + if not (isinstance(github_token, str) and github_token): + # Nightly continual runs have no fresh dashboard OAuth token; fall back to + # the GitHub App installation token so `gh` still works through the proxy. + github_token = await get_github_app_installation_token() if isinstance(github_token, str) and github_token: await _configure_sandbox_github_proxy(sandbox_backend, github_token) + # Skills are served from a virtual StateBackend route; gh/clone/execute stay on + # the sandbox. SKILL.md files are seeded into the `files` channel at invoke time. + backend = CompositeBackend(default=sandbox_backend, routes={SKILLS_ROUTE: StateBackend()}) + model_id = DEFAULT_LLM_MODEL_ID model_kwargs = provider_model_kwargs( model_id, @@ -134,21 +126,19 @@ async def get_review_style_analyzer(config: RunnableConfig) -> Pregel: repo_owner=owner or "", repo_name=name or "", working_dir=work_dir, + mode=mode, + skill_path=skill_path_for_mode(mode), reviewer_themes=REVIEWER_STYLE_THEMES.strip(), ) - user_context = ( - f"Repository: `{full_name}`\n\n" - f"{samples_text}\n\n" - "Research review style with `GH_TOKEN=dummy gh ...` via execute, then call " - "`save_review_style_prompt` once you have enough evidence." - ) + user_context = f"Repository: `{full_name}`\n\n{samples_text}".strip() system_prompt = f"{system_prompt}\n\n{user_context}" return create_deep_agent( model=make_model(model_id, **model_kwargs), system_prompt=system_prompt, - tools=[save_review_style_prompt], - backend=sandbox_backend, + tools=[save_review_style_prompt, read_finding_outcomes], + backend=backend, + skills=[SKILLS_ROUTE], middleware=[ SanitizeToolInputsMiddleware(), ModelCallLimitMiddleware( diff --git a/agent/dashboard/analyzer_cron.py b/agent/dashboard/analyzer_cron.py new file mode 100644 index 00000000..2b8743eb --- /dev/null +++ b/agent/dashboard/analyzer_cron.py @@ -0,0 +1,70 @@ +"""Per-repo nightly continual-learning crons for the analyzer. + +When a repo's bootstrap analysis completes we register one daily LangGraph cron +that fires a continual-learning run for that repo. Runs are threadless (a fresh +thread + sandbox each night) and authenticate via the GitHub App installation +token resolved inside ``get_analyzer`` (the cron carries no fresh user token). +""" + +from __future__ import annotations + +import hashlib +import logging + +from .review_style_jobs import ( + _client, + build_continual_run_configurable, + build_continual_run_input, +) +from .review_styles import get_review_style, update_review_style + +logger = logging.getLogger(__name__) + +_ASSISTANT_ID = "analyzer" + + +def _daily_schedule(full_name: str) -> str: + """Daily cron expression, staggered per repo to avoid a thundering herd.""" + digest = int(hashlib.sha256(full_name.encode()).hexdigest(), 16) + minute = digest % 60 + hour = 5 + (digest // 60) % 4 # 05:00–08:59 UTC + return f"{minute} {hour} * * *" + + +async def ensure_continual_cron(full_name: str) -> str | None: + """Idempotently register the per-repo nightly continual-learning cron.""" + record = await get_review_style(full_name) + existing = record.get("continual_cron_id") if record else None + if isinstance(existing, str) and existing: + return existing + + try: + cron = await _client().crons.create( + _ASSISTANT_ID, + schedule=_daily_schedule(full_name), + input=build_continual_run_input(full_name), + config={"configurable": build_continual_run_configurable(full_name)}, + metadata={"kind": "analyzer_continual", "repo": full_name}, + ) + except Exception: + logger.exception("Failed to create continual cron for %s", full_name) + return None + + cron_id = cron.get("cron_id") if isinstance(cron, dict) else getattr(cron, "cron_id", None) + if isinstance(cron_id, str) and cron_id: + await update_review_style(full_name, {"continual_cron_id": cron_id}) + return cron_id + return None + + +async def remove_continual_cron(full_name: str) -> None: + """Delete the per-repo continual-learning cron, if one is registered.""" + record = await get_review_style(full_name) + cron_id = record.get("continual_cron_id") if record else None + if not (isinstance(cron_id, str) and cron_id): + return + try: + await _client().crons.delete(cron_id) + except Exception: + logger.debug("Could not delete continual cron %s for %s", cron_id, full_name, exc_info=True) + await update_review_style(full_name, {"continual_cron_id": None}) diff --git a/agent/dashboard/review_style_jobs.py b/agent/dashboard/review_style_jobs.py index cb9e5605..6c801467 100644 --- a/agent/dashboard/review_style_jobs.py +++ b/agent/dashboard/review_style_jobs.py @@ -13,6 +13,7 @@ from ..review_style_collector import ( format_samples_for_analyzer, generate_review_style_thread_id, ) +from ..utils.analyzer_skills import build_skill_files from .review_styles import ( get_review_style, has_saved_prompt, @@ -24,7 +25,7 @@ from .review_styles import ( logger = logging.getLogger(__name__) -_ASSISTANT_ID = "review_style_analyzer" +_ASSISTANT_ID = "analyzer" def _client(): @@ -35,13 +36,48 @@ def _client(): return get_client() -async def start_review_style_analysis( +def build_continual_run_input(full_name: str) -> dict[str, Any]: + """Run input for a continual-learning analyzer run (shared with the cron).""" + return { + "messages": [ + { + "role": "user", + "content": ( + f"Refine the review-style prompt for `{full_name}` using this " + "reviewer's recorded finding outcomes. Follow the continual-learning " + "skill, then save the refined prompt." + ), + } + ], + "files": build_skill_files(), + } + + +def build_continual_run_configurable(full_name: str) -> dict[str, Any]: + """Configurable for a continual-learning analyzer run (shared with the cron). + + Includes an explicit ``thread_id`` so the run is anchored to the repo's + deterministic analyzer thread. The nightly cron is threadless, so without + this ``get_analyzer`` would early-return an empty agent (no thread_id) and + the run would no-op. Reusing the deterministic id keys the sandbox + thread + metadata to the repo; the threadless run carries no message history, so it + does not accumulate across nights. + """ + owner, repo = full_name.split("/", 1) + return { + "thread_id": generate_review_style_thread_id(owner, repo), + "review_style_full_name": full_name, + "analyzer_mode": "continual", + } + + +async def start_bootstrap_analysis( full_name: str, *, github_token: str, created_by: str, ) -> dict[str, Any]: - """Collect samples, persist metadata, and start the analyzer graph.""" + """Collect samples, persist metadata, and start a bootstrap analyzer run.""" owner, repo = full_name.split("/", 1) try: samples = await collect_review_samples(github_token, owner, repo) @@ -67,6 +103,7 @@ async def start_review_style_analysis( "review_style_top_reviewers": samples.top_reviewers, "review_style_prs_sampled": samples.prs_scanned, "review_style_reviews_sampled": samples.reviews_scanned, + "analyzer_mode": "bootstrap", } if not samples.samples: logger.info( @@ -93,12 +130,14 @@ async def start_review_style_analysis( { "role": "user", "content": ( - f"Analyze review style for `{full_name}`. Browse merged PR " - "review feedback with `GH_TOKEN=dummy gh` until you have enough " - "human examples, then save the repository-specific prompt." + f"Analyze review style for `{full_name}`. Follow the " + "bootstrap-repo-analysis skill: browse merged PR review feedback " + "with `GH_TOKEN=dummy gh` until you have enough human examples, " + "then save the repository-specific prompt." ), } - ] + ], + "files": build_skill_files(), }, config={"configurable": configurable}, if_not_exists="create", @@ -120,6 +159,33 @@ async def start_review_style_analysis( } +async def start_continual_run( + full_name: str, + *, + created_by: str = "manual", +) -> dict[str, Any]: + """Start an immediate continual-learning run (outcome-driven refinement).""" + configurable = build_continual_run_configurable(full_name) + thread_id = configurable["thread_id"] + try: + run = await _client().runs.create( + thread_id, + _ASSISTANT_ID, + input=build_continual_run_input(full_name), + config={"configurable": configurable}, + if_not_exists="create", + ) + run_id = run.get("run_id") if isinstance(run, dict) else getattr(run, "run_id", None) + return await update_review_style( + full_name, + {"analysis_run_id": run_id, "created_by": created_by}, + ) + except Exception: + logger.exception("Failed to start continual analyzer run for %s", full_name) + record = await get_review_style(full_name) + return record or {"full_name": full_name, "status": "failed", "error": "run start failed"} + + async def sync_review_style_run_status(full_name: str) -> dict[str, Any]: """Refresh store status from the latest analyzer run when still running.""" record = await get_review_style(full_name) diff --git a/agent/dashboard/review_styles.py b/agent/dashboard/review_styles.py index 0a0e759e..31630aab 100644 --- a/agent/dashboard/review_styles.py +++ b/agent/dashboard/review_styles.py @@ -89,6 +89,7 @@ def _default_record(full_name: str, created_by: str) -> dict[str, Any]: "reviews_sampled": 0, "analysis_thread_id": None, "analysis_run_id": None, + "continual_cron_id": None, "error": None, "created_by": created_by, "created_at": _now_iso(), diff --git a/agent/dashboard/routes.py b/agent/dashboard/routes.py index b6dabc9a..899dcb73 100644 --- a/agent/dashboard/routes.py +++ b/agent/dashboard/routes.py @@ -13,6 +13,7 @@ from fastapi.responses import RedirectResponse, Response, StreamingResponse from pydantic import BaseModel from .admin import is_admin +from .analyzer_cron import remove_continual_cron from .enabled_repos import ( list_enabled_review_repos, set_review_repo_enabled, @@ -43,7 +44,7 @@ from .profiles import ( ) from .review_style_jobs import ( cancel_review_style_analysis, - start_review_style_analysis, + start_bootstrap_analysis, sync_review_style_run_status, ) from .review_styles import ( @@ -539,7 +540,7 @@ async def api_analyze_review_style( record = await sync_review_style_run_status(full_name) if record.get("status") == "running": raise HTTPException(409, "analysis already running") - return await start_review_style_analysis( + return await start_bootstrap_analysis( full_name, github_token=token, created_by=session["sub"], @@ -571,6 +572,7 @@ async def api_delete_review_style( raise HTTPException(404, "review style not found") if record.get("status") == "running": await cancel_review_style_analysis(full_name) + await remove_continual_cron(full_name) await delete_review_style(full_name) return Response(status_code=204) diff --git a/agent/server.py b/agent/server.py index 1bac01df..9887cb3a 100644 --- a/agent/server.py +++ b/agent/server.py @@ -5,6 +5,7 @@ # ruff: noqa: E402 import logging import os +import time import warnings from typing import Any @@ -191,15 +192,12 @@ async def _recreate_sandbox(thread_id: str) -> SandboxBackendProtocol: (with proxy auth configured), swapping the per-thread proxy target. The agent is responsible for cloning repos via tools. """ - await client.threads.update( - thread_id=thread_id, - metadata={"sandbox_id": SANDBOX_CREATING}, - ) + await client.threads.update(thread_id=thread_id, metadata=_creating_metadata()) try: sandbox_backend = set_sandbox_backend(thread_id, await _create_sandbox_with_proxy()) except Exception: logger.exception("Failed to recreate sandbox after connection failure") - await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": None}) + await client.threads.update(thread_id=thread_id, metadata=_RESET_METADATA) raise return sandbox_backend @@ -226,25 +224,42 @@ async def check_or_recreate_sandbox( return sandbox_backend -async def _wait_for_sandbox_id(thread_id: str) -> str: - """Wait for sandbox_id to be set in thread metadata. +def _creating_metadata() -> dict[str, Any]: + """Metadata that claims the cross-process creation lock with a timestamp.""" + return {"sandbox_id": SANDBOX_CREATING, "sandbox_creating_at": time.time()} - Polls thread metadata until sandbox_id is set to a real value - (not the creating sentinel). - Raises: - TimeoutError: If sandbox creation takes too long +_RESET_METADATA: dict[str, Any] = {"sandbox_id": None, "sandbox_creating_at": None} + + +async def _resolve_creating_sentinel(thread_id: str) -> str | None: + """Resolve a ``__creating__`` sentinel seen with no cached backend. + + The sentinel is a cross-process lock: another worker may still be creating + the sandbox. Poll live thread metadata until it resolves to a real id. Only + when the sentinel is older than ``SANDBOX_CREATION_TIMEOUT`` (e.g. the + creating worker was restarted) is it treated as stale: metadata is reset and + ``None`` is returned so the caller creates a fresh sandbox. A sentinel with + no timestamp (written before this field existed) is also treated as stale. """ - elapsed = 0.0 - while elapsed < SANDBOX_CREATION_TIMEOUT: - sandbox_id = await get_sandbox_id_from_metadata(thread_id) - if sandbox_id is not None and sandbox_id != SANDBOX_CREATING: - return sandbox_id - await asyncio.sleep(SANDBOX_POLL_INTERVAL) - elapsed += SANDBOX_POLL_INTERVAL + while True: + thread = await client.threads.get(thread_id) + metadata = thread.get("metadata", {}) if isinstance(thread, dict) else {} + sandbox_id = metadata.get("sandbox_id") if isinstance(metadata, dict) else None - msg = f"Timeout waiting for sandbox creation for thread {thread_id}" - raise TimeoutError(msg) + if sandbox_id != SANDBOX_CREATING: + return sandbox_id if isinstance(sandbox_id, str) else None + + creating_at = metadata.get("sandbox_creating_at") if isinstance(metadata, dict) else None + age = time.time() - creating_at if isinstance(creating_at, (int, float)) else None + if age is None or age > SANDBOX_CREATION_TIMEOUT: + logger.warning( + "Resetting stale SANDBOX_CREATING for thread %s (age=%s)", thread_id, age + ) + await client.threads.update(thread_id=thread_id, metadata=_RESET_METADATA) + return None + + await asyncio.sleep(SANDBOX_POLL_INTERVAL) def graph_loaded_for_execution(config: RunnableConfig) -> bool: @@ -262,7 +277,8 @@ async def ensure_sandbox_for_thread(thread_id: str) -> SandboxBackendProtocol: Implements the four-state lifecycle described in AGENTS.md: 1. Cached in memory → ping; recreate on ``SandboxClientError``. - 2. Metadata says ``__creating__`` and no cache → poll until ready. + 2. Metadata says ``__creating__`` and no cache → wait for the creating + worker; only reset if the sentinel is proven stale (timestamp/timeout). 3. No sandbox at all → create one and persist the id. 4. Metadata has an id but no cache → reconnect; recreate on failure. @@ -275,7 +291,7 @@ async def ensure_sandbox_for_thread(thread_id: str) -> SandboxBackendProtocol: if sandbox_id == SANDBOX_CREATING and not sandbox_backend: logger.info("Sandbox creation in progress for thread %s, waiting...", thread_id) - sandbox_id = await _wait_for_sandbox_id(thread_id) + sandbox_id = await _resolve_creating_sentinel(thread_id) if sandbox_backend: logger.info("Using cached sandbox backend for thread %s", thread_id) @@ -285,14 +301,14 @@ async def ensure_sandbox_for_thread(thread_id: str) -> SandboxBackendProtocol: sandbox_backend = await _refresh_github_proxy_or_recreate(sandbox_backend, thread_id) elif sandbox_id is None: logger.info("Creating new sandbox for thread %s", thread_id) - await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": SANDBOX_CREATING}) + await client.threads.update(thread_id=thread_id, metadata=_creating_metadata()) try: sandbox_backend = await _create_sandbox_with_proxy() logger.info("Sandbox created: %s", sandbox_backend.id) except Exception: logger.exception("Failed to create sandbox") try: - await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": None}) + await client.threads.update(thread_id=thread_id, metadata=_RESET_METADATA) except Exception: logger.exception("Failed to reset sandbox_id metadata") raise @@ -303,15 +319,13 @@ async def ensure_sandbox_for_thread(thread_id: str) -> SandboxBackendProtocol: sandbox_backend = await asyncio.to_thread(create_sandbox, sandbox_id) except Exception: logger.warning("Failed to connect to existing sandbox %s, creating new one", sandbox_id) - await client.threads.update( - thread_id=thread_id, metadata={"sandbox_id": SANDBOX_CREATING} - ) + await client.threads.update(thread_id=thread_id, metadata=_creating_metadata()) try: sandbox_backend = await _create_sandbox_with_proxy() created_replacement_sandbox = True except Exception: logger.exception("Failed to create replacement sandbox") - await client.threads.update(thread_id=thread_id, metadata={"sandbox_id": None}) + await client.threads.update(thread_id=thread_id, metadata=_RESET_METADATA) raise if not created_replacement_sandbox: original_sandbox_id = sandbox_backend.id diff --git a/agent/skills/bootstrap-repo-analysis/SKILL.md b/agent/skills/bootstrap-repo-analysis/SKILL.md new file mode 100644 index 00000000..20d45e7f --- /dev/null +++ b/agent/skills/bootstrap-repo-analysis/SKILL.md @@ -0,0 +1,61 @@ +--- +name: bootstrap-repo-analysis +description: First-time analysis of a repository with no prior reviewer outcomes. Crawl historical merged-PR review feedback with the gh CLI (plus any preloaded samples), extract the team's review norms, and synthesize the initial per-repo review-style prompt. Use this for a cold-start repo; use continual-learning instead once the reviewer has accumulated finding outcomes. +--- + +# Bootstrap repo analysis + +You are writing the **first** review-style prompt for the repository named in the +system prompt. There is no outcomes history yet, so your signal comes entirely from +the repo's own historical PR review feedback. Do not call `read_finding_outcomes` in +this mode — it will be empty. + +Always invoke gh as: `GH_TOKEN=dummy gh `. + +## 1. Research (required) + +Browse historical **merged** PR review feedback until you have catalogued at least +**8 substantive human** review comments (skip `[bot]` accounts and obvious automation +like codecov / dependabot). Useful commands: + +``` +GH_TOKEN=dummy gh pr list --repo / --state merged --limit 30 +GH_TOKEN=dummy gh api repos///pulls//reviews +GH_TOKEN=dummy gh api repos///pulls//comments +GH_TOKEN=dummy gh api repos///issues//comments +``` + +If the first batch is sparse, raise `--limit` or walk older PR numbers. The user +message may include **preloaded samples** — verify and extend them with `gh`, don't +just trust them. + +Identify the top ~5 human reviewers by volume and note their phrasing, what severity +they assign, and what they routinely ignore. + +## 2. Extract concrete, repo-specific patterns + +The highest-value content is a **bug taxonomy tied to this repo's stack** — concrete +"hunt for X" rules a maintainer would catch on first read — plus a calibrated "do not +flag" list. Pair each pattern with the failure mode and, where you saw it, the kind of +diff that triggered it. Avoid generic advice that would apply to any repo. + +Cover: +- What the team routinely flags vs. skips (paraphrased patterns, not invented quotes) +- Severity calibration tied to user-visible / runtime consequence +- Tone and test expectations +- Repo-specific conventions (frameworks, repository/data-access boundaries, naming) +- Anti-patterns the reviewers here deliberately avoid + +Stay aligned with the reviewer-agent themes in the system prompt (high-signal, +diff-anchored defects — not nits). + +## 3. Save + +Only after real research, call `save_review_style_prompt` once with: +- `custom_prompt`: 400–1200 words teaching the reviewer this repo's norms. +- `analysis_summary`: 2–4 sentences for the dashboard. +- `top_reviewers` (comma-separated logins), `prs_sampled`, `reviews_sampled`. + +Do **not** save a generic guide after one or two commands. Only after ~25+ merged PRs +with zero human feedback may you save a short, conservative guide — and say so in +`analysis_summary`. diff --git a/agent/skills/continual-learning/SKILL.md b/agent/skills/continual-learning/SKILL.md new file mode 100644 index 00000000..b3deb574 --- /dev/null +++ b/agent/skills/continual-learning/SKILL.md @@ -0,0 +1,44 @@ +--- +name: continual-learning +description: Nightly refinement of an existing per-repo review-style prompt using this reviewer's own finding outcomes. Read confirmed (resolved-by-commit / thumbs-up) and dismissed (thumbs-down) findings, promote the bug patterns the team actually fixes, demote the false-positive patterns, reconcile against the current prompt, and save the refined version. Use this once outcomes exist; use bootstrap-repo-analysis for a cold-start repo. +--- + +# Continual learning + +You are **refining** the existing review-style prompt for the repository named in the +system prompt, using outcomes the reviewer has accrued since the last run. The goal is +to raise recall (catch more real bugs) without hurting precision (stop repeating +dismissed ones). + +## 1. Read outcomes first + +Call `read_finding_outcomes` once. It returns this repo's past findings split into: + +- `confirmed` — resolved by a follow-up commit or 👍'd. These are **real** bug patterns + this team fixes. Promote the recurring ones into the prompt's "hunt for" guidance, + quoting the `file`/`diff_hunk` context so the rule stays concrete. +- `dismissed` — dismissed or 👎'd. These are **false-positive** patterns. Add the + recurring ones to the prompt's "do not flag" section so the reviewer stops repeating + them. + +Look for repetition, not one-offs. A single dismissed finding is noise; the same class +dismissed several times is a rule. + +## 2. Reconcile against the current prompt + +The current `custom_prompt` is the starting point — you are editing it, not rewriting +from scratch. Read it (it is summarized for you / available via the dashboard record). +Keep what still holds, strengthen rules the outcomes confirm, and remove or soften rules +the outcomes contradict. Optionally do a **light** `gh` top-up +(`GH_TOKEN=dummy gh ...`) to confirm a pattern, but outcomes are the primary signal — do +not re-run a full PR crawl. + +Stay aligned with the reviewer-agent themes in the system prompt. + +## 3. Save + +Call `save_review_style_prompt` once with the refined `custom_prompt` (400–1200 words), +an `analysis_summary` that names what changed this cycle (e.g. "promoted N-pattern after +3 confirmed fixes; dropped M-pattern after repeated dismissals"), and the +`top_reviewers` / counts you have. If outcomes were empty and nothing changed, say so in +`analysis_summary` and re-save the existing prompt unchanged rather than degrading it. diff --git a/agent/tools/read_finding_outcomes.py b/agent/tools/read_finding_outcomes.py new file mode 100644 index 00000000..2f1cc2d8 --- /dev/null +++ b/agent/tools/read_finding_outcomes.py @@ -0,0 +1,39 @@ +"""Tool: read this reviewer's past finding outcomes for the repo under analysis. + +Surfaces findings that were later confirmed (resolved by a commit / 👍) vs +dismissed (false positive / 👎), so the analyzer can promote the bug patterns +this team actually fixes and add the noisy ones to a skip-list. +""" + +from __future__ import annotations + +from typing import Any + +from langgraph.config import get_config + +from ..utils.reviewer_outcomes import read_outcomes_for_repo + + +def read_finding_outcomes(limit: int = 60) -> dict[str, Any]: + """Return confirmed and dismissed past findings for the repo being analyzed. + + Call this before synthesizing the style prompt. Use the ``confirmed`` + findings to reinforce what to hunt for and the ``dismissed`` findings to + build the "do not flag" list. + """ + config = get_config() + configurable = config.get("configurable") or {} + full_name = configurable.get("review_style_full_name") + if not isinstance(full_name, str) or "/" not in full_name: + return {"ok": False, "error": "no repo under analysis", "confirmed": [], "dismissed": []} + + outcomes = read_outcomes_for_repo(full_name, limit=limit) + confirmed = outcomes["confirmed"] + dismissed = outcomes["dismissed"] + return { + "ok": True, + "repo": full_name, + "counts": {"confirmed": len(confirmed), "dismissed": len(dismissed)}, + "confirmed": confirmed, + "dismissed": dismissed, + } diff --git a/agent/tools/resolve_finding_thread.py b/agent/tools/resolve_finding_thread.py index aed1927d..4ad4215c 100644 --- a/agent/tools/resolve_finding_thread.py +++ b/agent/tools/resolve_finding_thread.py @@ -21,6 +21,7 @@ from ..reviewer_publish import ( ) from ..reviewer_reconcile import reconcile_findings_with_review_threads from ..utils.github_token import get_github_token +from ..utils.reviewer_outcomes import emit_finding_status_outcome def _normalize_note(note: str | None) -> str | None: @@ -66,7 +67,7 @@ def resolve_finding_thread( if not token: return {"success": False, "error": "No GitHub token available"} - return asyncio.run( + result = asyncio.run( _resolve_finding_thread_async( finding_id=finding_id, status=status, @@ -77,6 +78,15 @@ def resolve_finding_thread( token=token, ) ) + if result.get("success") and isinstance(result.get("finding"), dict): + thread_id = configurable.get("thread_id") if isinstance(configurable, dict) else None + emit_finding_status_outcome( + result["finding"], + status, + configurable=configurable, + thread_id=thread_id if isinstance(thread_id, str) else None, + ) + return result async def _resolve_finding_thread_async( diff --git a/agent/tools/save_review_style.py b/agent/tools/save_review_style.py index 7939d2e9..5f12cd41 100644 --- a/agent/tools/save_review_style.py +++ b/agent/tools/save_review_style.py @@ -3,12 +3,30 @@ from __future__ import annotations import asyncio +import logging from typing import Any from langgraph.config import get_config +from ..dashboard.analyzer_cron import ensure_continual_cron from ..dashboard.review_styles import mark_analysis_completed, mark_analysis_failed +logger = logging.getLogger(__name__) + + +async def _complete_and_register(full_name: str, **completed_kwargs: Any) -> dict[str, Any]: + """Persist the prompt, then ensure the repo's nightly continual cron exists. + + Cron registration is idempotent, so continual runs completing later don't + re-register it; it just guarantees a cron once a prompt first exists. + """ + record = await mark_analysis_completed(full_name, **completed_kwargs) + try: + await ensure_continual_cron(full_name) + except Exception: + logger.exception("Failed to ensure continual cron for %s", full_name) + return record + def save_review_style_prompt( custom_prompt: str, @@ -41,7 +59,7 @@ def save_review_style_prompt( return {"ok": False, "error": "custom_prompt cannot be empty"} record = asyncio.run( - mark_analysis_completed( + _complete_and_register( full_name, custom_prompt=custom_prompt.strip(), analysis_summary=analysis_summary.strip(), diff --git a/agent/tools/update_finding.py b/agent/tools/update_finding.py index ad02f8b2..b6048c35 100644 --- a/agent/tools/update_finding.py +++ b/agent/tools/update_finding.py @@ -17,6 +17,7 @@ from ..reviewer_findings import ( normalize_finding_title, update_finding_fields, ) +from ..utils.reviewer_outcomes import emit_finding_status_outcome def _is_non_empty_str(value: Any) -> bool: @@ -150,6 +151,7 @@ def update_finding( if finding is None: return {"success": False, "error": f"No finding found with id {finding_id}"} + delegated_resolution = False repo_config = configurable.get("repo") if isinstance(configurable, dict) else None pr_number = configurable.get("pr_number") if isinstance(configurable, dict) else None can_resolve_github_thread = ( @@ -172,6 +174,7 @@ def update_finding( "error": "GitHub review thread resolution failed; finding was left open.", "github_resolution": resolve_result, } + delegated_resolution = True updates.pop("status", None) updates.pop("last_update_note", None) updates.pop("resolution_note", None) @@ -193,6 +196,8 @@ def update_finding( updated = asyncio.run(update_finding_fields(thread_id, finding_id, updates)) if updated is None: return {"success": False, "error": f"No finding found with id {finding_id}"} + if status in {"resolved", "dismissed"} and not delegated_resolution: + emit_finding_status_outcome(updated, status, configurable=configurable, thread_id=thread_id) result = {"success": True, "finding": updated} if suggestion_dropped: result["suggestion_dropped"] = True diff --git a/agent/utils/analyzer_skills.py b/agent/utils/analyzer_skills.py new file mode 100644 index 00000000..69ef019e --- /dev/null +++ b/agent/utils/analyzer_skills.py @@ -0,0 +1,49 @@ +"""Repo-bundled analyzer skills, served to the agent as virtual files. + +The two analyzer playbooks live as ``SKILL.md`` files under ``agent/skills/``. +They are surfaced to the deepagents ``SkillsMiddleware`` via a ``StateBackend`` +mounted at ``/skills/`` in a ``CompositeBackend`` — so the agent reads them with +``read_file`` without anything ever being written to the execution sandbox. + +The ``files`` channel is seeded at invoke time (see the launchers). Because +``CompositeBackend`` strips the ``/skills/`` route prefix before delegating to the +``StateBackend``, the seeded keys are the *stripped* paths (e.g. +``/bootstrap-repo-analysis/SKILL.md``), while the agent and ``SkillsMiddleware`` +address them under ``/skills/...``. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from deepagents.backends.utils import create_file_data + +SKILLS_DIR = Path(__file__).resolve().parent.parent / "skills" +SKILLS_ROUTE = "/skills/" + +BOOTSTRAP_SKILL = "bootstrap-repo-analysis" +CONTINUAL_SKILL = "continual-learning" + +ANALYZER_MODES = {"bootstrap": BOOTSTRAP_SKILL, "continual": CONTINUAL_SKILL} + + +def skill_path_for_mode(mode: str) -> str: + """Return the agent-facing ``/skills//SKILL.md`` path for a run mode.""" + skill = ANALYZER_MODES.get(mode, BOOTSTRAP_SKILL) + return f"{SKILLS_ROUTE}{skill}/SKILL.md" + + +def build_skill_files() -> dict[str, Any]: + """Return ``{stripped_path: FileData}`` for every bundled analyzer skill. + + Seed this into the run input's ``files`` so the ``/skills/`` StateBackend route + can serve them. Keys omit the ``/skills`` prefix (stripped by the composite + route); values are ``FileData`` v2 entries. + """ + files: dict[str, Any] = {} + for skill in ANALYZER_MODES.values(): + skill_md = SKILLS_DIR / skill / "SKILL.md" + text = skill_md.read_text(encoding="utf-8") + files[f"/{skill}/SKILL.md"] = create_file_data(text) + return files diff --git a/agent/utils/github_feedback.py b/agent/utils/github_feedback.py index d6e4f9b8..a57ed772 100644 --- a/agent/utils/github_feedback.py +++ b/agent/utils/github_feedback.py @@ -12,6 +12,7 @@ from langgraph_sdk.client import LangGraphClient from ..reviewer_findings import list_findings from .langsmith import create_langsmith_feedback, delete_langsmith_feedback +from .reviewer_outcomes import outcome_from_score, upsert_finding_outcome logger = logging.getLogger(__name__) @@ -221,6 +222,22 @@ async def process_github_reaction( comment=f"GitHub review reaction feedback from {user_login}", source_info={**source_info, "reactions": sorted(active_reactions)}, ) + outcome = outcome_from_score(score, source="github") + if outcome is not None: + label, label_source = outcome + await asyncio.to_thread( + upsert_finding_outcome, + finding, + label=label, + label_source=label_source, + repo=repo_key, + pr_number=pr_number, + pr_url=f"https://github.com/{repo_key}/pull/{pr_number}", + head_sha=str(finding.get("first_seen_sha") or ""), + run_id=run_id, + thread_id=thread_id, + ) + if success: await _mark_event_processed(langgraph_client, repo_key, delivery_id) diff --git a/agent/utils/reviewer_outcomes.py b/agent/utils/reviewer_outcomes.py new file mode 100644 index 00000000..1c94f8af --- /dev/null +++ b/agent/utils/reviewer_outcomes.py @@ -0,0 +1,322 @@ +"""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} diff --git a/agent/utils/slack_feedback.py b/agent/utils/slack_feedback.py index f4db65d2..16920c58 100644 --- a/agent/utils/slack_feedback.py +++ b/agent/utils/slack_feedback.py @@ -11,6 +11,8 @@ from langgraph_sdk import get_client from langgraph_sdk.client import LangGraphClient from .langsmith import create_langsmith_feedback, delete_langsmith_feedback +from .reviewer_outcomes import outcome_from_score as _outcome_from_score +from .reviewer_outcomes import upsert_run_outcome from .slack import lookup_slack_run_mapping logger = logging.getLogger(__name__) @@ -201,6 +203,17 @@ async def process_slack_reaction( source_info={**source_info, "reactions": sorted(active_reactions)}, ) + outcome = _outcome_from_score(score, source="slack") + if outcome is not None: + label, label_source = outcome + await asyncio.to_thread( + upsert_run_outcome, + label=label, + label_source=label_source, + run_id=run_id, + extra={"channel_id": channel_id}, + ) + if success: await _mark_event_processed(langgraph_client, channel_id, event_id) diff --git a/langgraph.json b/langgraph.json index 89e3efae..c8d4dccd 100644 --- a/langgraph.json +++ b/langgraph.json @@ -4,7 +4,7 @@ "graphs": { "agent": "agent.server:get_agent", "reviewer": "agent.reviewer:get_reviewer_agent", - "review_style_analyzer": "agent.review_style_analyzer:get_review_style_analyzer" + "analyzer": "agent.analyzer:get_analyzer" }, "dependencies": ["."], "http": { diff --git a/tests/test_analyzer_cron.py b/tests/test_analyzer_cron.py new file mode 100644 index 00000000..aa7d1154 --- /dev/null +++ b/tests/test_analyzer_cron.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from agent.dashboard import analyzer_cron + + +class _FakeCrons: + def __init__(self) -> None: + self.created: list[dict[str, Any]] = [] + self.deleted: list[str] = [] + + async def create(self, assistant_id: str, **kwargs: Any) -> dict[str, Any]: + self.created.append({"assistant_id": assistant_id, **kwargs}) + return {"cron_id": "cron_123"} + + async def delete(self, cron_id: str) -> None: + self.deleted.append(cron_id) + + +class _FakeClient: + def __init__(self) -> None: + self.crons = _FakeCrons() + + +@pytest.fixture +def fake_client(monkeypatch) -> _FakeClient: # noqa: ANN001 + client = _FakeClient() + monkeypatch.setattr(analyzer_cron, "_client", lambda: client) + return client + + +def _patch_record(monkeypatch, record: dict[str, Any] | None) -> dict[str, Any]: # noqa: ANN001 + updates: dict[str, Any] = {} + + async def fake_get(full_name: str) -> dict[str, Any] | None: + return record + + async def fake_update(full_name: str, patch: dict[str, Any]) -> dict[str, Any]: + updates.update(patch) + return {**(record or {}), **patch} + + monkeypatch.setattr(analyzer_cron, "get_review_style", fake_get) + monkeypatch.setattr(analyzer_cron, "update_review_style", fake_update) + return updates + + +async def test_ensure_continual_cron_creates_and_stores(monkeypatch, fake_client) -> None: # noqa: ANN001 + updates = _patch_record(monkeypatch, {"full_name": "o/r", "continual_cron_id": None}) + + cron_id = await analyzer_cron.ensure_continual_cron("o/r") + + assert cron_id == "cron_123" + assert len(fake_client.crons.created) == 1 + created = fake_client.crons.created[0] + assert created["assistant_id"] == "analyzer" + assert created["config"]["configurable"]["analyzer_mode"] == "continual" + # Must carry an explicit thread_id, else get_analyzer early-returns an empty + # agent and the nightly run no-ops. + assert created["config"]["configurable"].get("thread_id") + assert "/continual-learning/SKILL.md" in created["input"]["files"] + assert updates["continual_cron_id"] == "cron_123" + + +async def test_ensure_continual_cron_idempotent(monkeypatch, fake_client) -> None: # noqa: ANN001 + _patch_record(monkeypatch, {"full_name": "o/r", "continual_cron_id": "existing"}) + + cron_id = await analyzer_cron.ensure_continual_cron("o/r") + + assert cron_id == "existing" + assert fake_client.crons.created == [] + + +async def test_remove_continual_cron(monkeypatch, fake_client) -> None: # noqa: ANN001 + updates = _patch_record(monkeypatch, {"full_name": "o/r", "continual_cron_id": "cron_123"}) + + await analyzer_cron.remove_continual_cron("o/r") + + assert fake_client.crons.deleted == ["cron_123"] + assert updates["continual_cron_id"] is None + + +async def test_remove_continual_cron_noop_when_absent(monkeypatch, fake_client) -> None: # noqa: ANN001 + _patch_record(monkeypatch, {"full_name": "o/r", "continual_cron_id": None}) + + await analyzer_cron.remove_continual_cron("o/r") + + assert fake_client.crons.deleted == [] + + +def test_daily_schedule_is_stable_and_in_window() -> None: + sched = analyzer_cron._daily_schedule("o/r") + sched2 = analyzer_cron._daily_schedule("o/r") + assert sched == sched2 + minute, hour, *_ = sched.split() + assert 0 <= int(minute) <= 59 + assert 5 <= int(hour) <= 8 diff --git a/tests/test_analyzer_skills.py b/tests/test_analyzer_skills.py new file mode 100644 index 00000000..89d79cd5 --- /dev/null +++ b/tests/test_analyzer_skills.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import pathlib + +from deepagents.middleware.skills import _parse_skill_metadata + +from agent.dashboard.review_style_jobs import ( + build_continual_run_configurable, + build_continual_run_input, +) +from agent.utils.analyzer_skills import ( + ANALYZER_MODES, + SKILLS_DIR, + build_skill_files, + skill_path_for_mode, +) + + +def test_build_skill_files_stripped_keys_and_valid_file_data() -> None: + files = build_skill_files() + assert set(files) == { + "/bootstrap-repo-analysis/SKILL.md", + "/continual-learning/SKILL.md", + } + for entry in files.values(): + assert entry["encoding"] == "utf-8" + assert isinstance(entry["content"], str) and entry["content"].strip() + assert "created_at" in entry and "modified_at" in entry + + +def test_skill_path_for_mode() -> None: + assert skill_path_for_mode("bootstrap") == "/skills/bootstrap-repo-analysis/SKILL.md" + assert skill_path_for_mode("continual") == "/skills/continual-learning/SKILL.md" + # Unknown modes fall back to bootstrap. + assert skill_path_for_mode("whatever") == "/skills/bootstrap-repo-analysis/SKILL.md" + + +def test_bundled_skill_md_parse() -> None: + for skill in ANALYZER_MODES.values(): + path = SKILLS_DIR / skill / "SKILL.md" + meta = _parse_skill_metadata(path.read_text(), str(path), path.parent.name) + assert meta["name"] == skill + assert meta["description"].strip() + + +def test_skills_dir_resolves() -> None: + assert SKILLS_DIR.name == "skills" + assert (SKILLS_DIR / "bootstrap-repo-analysis" / "SKILL.md").exists() + assert isinstance(SKILLS_DIR, pathlib.Path) + + +def test_continual_run_payload_carries_mode_and_skill_files() -> None: + configurable = build_continual_run_configurable("o/r") + assert configurable["analyzer_mode"] == "continual" + assert configurable["review_style_full_name"] == "o/r" + assert configurable.get("thread_id") + + run_input = build_continual_run_input("o/r") + assert "/continual-learning/SKILL.md" in run_input["files"] + assert run_input["messages"][0]["role"] == "user" diff --git a/tests/test_reviewer_outcomes.py b/tests/test_reviewer_outcomes.py new file mode 100644 index 00000000..178260c9 --- /dev/null +++ b/tests/test_reviewer_outcomes.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +from typing import Any + +from agent.utils import reviewer_outcomes +from agent.utils.reviewer_outcomes import ( + FALSE_POSITIVE, + TRUE_POSITIVE, + emit_finding_status_outcome, + outcome_from_score, + outcome_from_status, + upsert_finding_outcome, +) + + +def test_outcome_from_status_resolved_by_commit() -> None: + assert outcome_from_status("resolved", first_seen_sha="aaa", head_sha="bbb") == ( + TRUE_POSITIVE, + "resolved_by_commit", + ) + + +def test_outcome_from_status_resolved_same_sha() -> None: + assert outcome_from_status("resolved", first_seen_sha="aaa", head_sha="aaa") == ( + TRUE_POSITIVE, + "resolved_same_sha", + ) + + +def test_outcome_from_status_dismissed() -> None: + assert outcome_from_status("dismissed", first_seen_sha="aaa", head_sha="bbb") == ( + FALSE_POSITIVE, + "dismissed", + ) + + +def test_outcome_from_status_open_is_none() -> None: + assert outcome_from_status("open", first_seen_sha="aaa", head_sha="bbb") is None + + +def test_outcome_from_score() -> None: + assert outcome_from_score(1.0, source="github") == (TRUE_POSITIVE, "github_thumbs_up") + assert outcome_from_score(0.0, source="github") == (FALSE_POSITIVE, "github_thumbs_down") + assert outcome_from_score(1.0, source="slack") == (TRUE_POSITIVE, "slack_thumbs_up") + assert outcome_from_score(None, source="github") is None + + +def test_example_id_is_deterministic() -> None: + a = reviewer_outcomes._example_id("o/r", "f_1", "dismissed") + b = reviewer_outcomes._example_id("o/r", "f_1", "dismissed") + c = reviewer_outcomes._example_id("o/r", "f_1", "resolved_by_commit") + assert a == b + assert a != c + + +class _FakeDataset: + id = "ds_123" + + +class _FakeClient: + def __init__(self) -> None: + self.created: list[dict[str, Any]] = [] + self.updated: list[dict[str, Any]] = [] + self.conflict_once = False + + def list_datasets(self, dataset_name: str): # noqa: ANN001 + return iter([_FakeDataset()]) + + def create_example(self, **kwargs: Any) -> None: + if self.conflict_once: + self.conflict_once = False + raise RuntimeError("already exists") + self.created.append(kwargs) + + def update_example(self, **kwargs: Any) -> None: + self.updated.append(kwargs) + + +def _finding() -> dict[str, Any]: + return { + "id": "f_abc", + "file": "app/x.rb", + "start_line": 10, + "end_line": 12, + "side": "RIGHT", + "diff_hunk": "@@ -1 +1 @@\n-foo\n+bar", + "title": "NoMethodError on nil", + "description": "body", + "severity": "high", + "confidence": "high", + "category": "correctness", + "first_seen_sha": "aaa", + "github_review_run_id": "run_1", + "resolution_note": "fixed in def", + } + + +def test_upsert_finding_outcome_builds_payload(monkeypatch) -> None: # noqa: ANN001 + fake = _FakeClient() + monkeypatch.setattr(reviewer_outcomes, "_outcomes_client", lambda: fake) + + ok = upsert_finding_outcome( + _finding(), + label=TRUE_POSITIVE, + label_source="resolved_by_commit", + repo="o/r", + pr_number=7, + pr_url="https://github.com/o/r/pull/7", + base_sha="aaa", + head_sha="bbb", + thread_id="t1", + ) + assert ok + assert len(fake.created) == 1 + call = fake.created[0] + assert call["dataset_id"] == "ds_123" + assert call["inputs"]["repo"] == "o/r" + assert call["inputs"]["file"] == "app/x.rb" + assert call["inputs"]["diff_hunk"].startswith("@@") + assert call["outputs"]["label"] == TRUE_POSITIVE + assert call["outputs"]["label_source"] == "resolved_by_commit" + assert call["outputs"]["finding"]["severity"] == "high" + assert call["metadata"]["granularity"] == "finding" + assert call["metadata"]["repo"] == "o/r" + assert call["metadata"]["run_id"] == "run_1" + + +def test_upsert_finding_outcome_updates_on_conflict(monkeypatch) -> None: # noqa: ANN001 + fake = _FakeClient() + fake.conflict_once = True + monkeypatch.setattr(reviewer_outcomes, "_outcomes_client", lambda: fake) + + ok = upsert_finding_outcome( + _finding(), label=FALSE_POSITIVE, label_source="dismissed", repo="o/r" + ) + assert ok + assert not fake.created + assert len(fake.updated) == 1 + + +def test_upsert_no_client_is_noop(monkeypatch) -> None: # noqa: ANN001 + monkeypatch.setattr(reviewer_outcomes, "_outcomes_client", lambda: None) + assert ( + upsert_finding_outcome(_finding(), label=TRUE_POSITIVE, label_source="x", repo="o/r") + is False + ) + + +def test_emit_finding_status_outcome_maps_and_calls(monkeypatch) -> None: # noqa: ANN001 + captured: dict[str, Any] = {} + + def fake_upsert(finding, **kwargs: Any) -> bool: # noqa: ANN001 + captured.update(kwargs) + captured["finding_id"] = finding["id"] + return True + + monkeypatch.setattr(reviewer_outcomes, "upsert_finding_outcome", fake_upsert) + + configurable = { + "repo": {"owner": "o", "name": "r"}, + "pr_number": 7, + "pr_url": "https://github.com/o/r/pull/7", + "base_sha": "aaa", + "head_sha": "bbb", + } + assert emit_finding_status_outcome( + _finding(), "resolved", configurable=configurable, thread_id="t1" + ) + assert captured["repo"] == "o/r" + assert captured["label"] == TRUE_POSITIVE + assert captured["label_source"] == "resolved_by_commit" + assert captured["pr_number"] == 7 + + +def test_emit_finding_status_outcome_no_repo_is_noop(monkeypatch) -> None: # noqa: ANN001 + monkeypatch.setattr(reviewer_outcomes, "upsert_finding_outcome", lambda *a, **k: pytest_fail()) + assert emit_finding_status_outcome(_finding(), "dismissed", configurable={}) is False + + +def pytest_fail() -> bool: + raise AssertionError("upsert should not be called without a repo") diff --git a/tests/test_stale_sandbox_creating.py b/tests/test_stale_sandbox_creating.py new file mode 100644 index 00000000..2c7ed57d --- /dev/null +++ b/tests/test_stale_sandbox_creating.py @@ -0,0 +1,99 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from agent.server import ( + SANDBOX_BACKENDS, + SANDBOX_CREATING, + SANDBOX_CREATION_TIMEOUT, + ensure_sandbox_for_thread, +) + + +@pytest.mark.asyncio +async def test_stale_sandbox_creating_without_cached_backend_resets_and_creates() -> None: + thread_id = "thread-stale-creating" + SANDBOX_BACKENDS.clear() + sandbox_backend = MagicMock() + sandbox_backend.id = "sandbox-new" + + stale_at = 0.0 # epoch 0 → far older than the timeout + thread = {"metadata": {"sandbox_id": SANDBOX_CREATING, "sandbox_creating_at": stale_at}} + + with ( + patch( + "agent.server.get_sandbox_id_from_metadata", + new_callable=AsyncMock, + return_value=SANDBOX_CREATING, + ), + patch("agent.server.client.threads.get", new_callable=AsyncMock, return_value=thread), + patch("agent.server._create_sandbox_with_proxy", new_callable=AsyncMock) as create_sandbox, + patch("agent.server._configure_git_identity", new_callable=AsyncMock), + patch("agent.server.client.threads.update", new_callable=AsyncMock) as update_thread, + ): + create_sandbox.return_value = sandbox_backend + + result = await ensure_sandbox_for_thread(thread_id) + + assert result.id == "sandbox-new" + create_sandbox.assert_awaited_once() + # Stale sentinel reset (id + timestamp cleared), then fresh creation claims it. + assert update_thread.await_args_list[0].kwargs == { + "thread_id": thread_id, + "metadata": {"sandbox_id": None, "sandbox_creating_at": None}, + } + assert update_thread.await_args_list[1].kwargs["metadata"]["sandbox_id"] == SANDBOX_CREATING + assert update_thread.await_args_list[2].kwargs == { + "thread_id": thread_id, + "metadata": {"sandbox_id": "sandbox-new"}, + } + + SANDBOX_BACKENDS.clear() + + +@pytest.mark.asyncio +async def test_fresh_sandbox_creating_waits_for_other_worker() -> None: + """A recent sentinel from another worker must be waited on, not overwritten.""" + thread_id = "thread-concurrent-creating" + SANDBOX_BACKENDS.clear() + existing_backend = MagicMock() + existing_backend.id = "sandbox-existing" + + import time + + fresh_at = time.time() # well within the timeout + threads = [ + {"metadata": {"sandbox_id": SANDBOX_CREATING, "sandbox_creating_at": fresh_at}}, + {"metadata": {"sandbox_id": "sandbox-existing", "sandbox_creating_at": fresh_at}}, + ] + + async def passthrough(sb, _thread_id): + return sb + + with ( + patch( + "agent.server.get_sandbox_id_from_metadata", + new_callable=AsyncMock, + return_value=SANDBOX_CREATING, + ), + patch("agent.server.client.threads.get", new_callable=AsyncMock, side_effect=threads), + patch("agent.server.asyncio.sleep", new_callable=AsyncMock), + patch("agent.server.create_sandbox", return_value=existing_backend) as connect_sandbox, + patch("agent.server._create_sandbox_with_proxy", new_callable=AsyncMock) as create_sandbox, + patch("agent.server.check_or_recreate_sandbox", side_effect=passthrough), + patch("agent.server._refresh_github_proxy_or_recreate", side_effect=passthrough), + patch("agent.server._configure_git_identity", new_callable=AsyncMock), + patch("agent.server.client.threads.update", new_callable=AsyncMock) as update_thread, + ): + result = await ensure_sandbox_for_thread(thread_id) + + assert result.id == "sandbox-existing" + # Connected to the worker's sandbox; no second sandbox was created. + connect_sandbox.assert_called_once_with("sandbox-existing") + create_sandbox.assert_not_awaited() + # The fresh sentinel was never reset. + for call in update_thread.await_args_list: + assert call.kwargs["metadata"] != {"sandbox_id": None, "sandbox_creating_at": None} + + assert SANDBOX_CREATION_TIMEOUT > 0 + SANDBOX_BACKENDS.clear()