feat(gateway): run a fast quality gate at i_am_done, before QA

The developer's i_am_done submit now runs the project's fast quality gate
(lint + typecheck) in the developer's workspace and blocks the transition to
awaiting_qa if it's red, returning the failing output as the remediate hint —
so a red gate is caught at the dev's desk instead of in QA review or CI. The
slow test suite intentionally stays on CI. The gate is fail-open on
infrastructure errors (missing workspace/toolchain never blocks a submit) and a
no-op for projects that configure no lint/typecheck commands. Developer prompt
updated.
This commit is contained in:
Renn F
2026-06-14 04:54:05 +02:00
parent 2db11c7833
commit 481c5fe17b
5 changed files with 298 additions and 2 deletions
@@ -1503,12 +1503,40 @@ class Choreographer:
return await self._reject_i_am_done(ctx, rejection)
if rejection := await self._ensure_branch_pushed(ctx):
return await self._reject_i_am_done(ctx, rejection)
if rejection := await self._check_quality_gate(ctx):
return await self._reject_i_am_done(ctx, rejection)
# Pre-gateway parity: persist per-criterion
# status now that all gates have passed. The write runs AFTER the
# verdict so it cannot change i_am_done's rejection behavior.
await self._write_criteria_status(ctx.agent_id, ctx.task_id, ctx.task)
return None
async def _check_quality_gate(self, ctx: _IAmDoneContext) -> Envelope | None:
"""Run the project's fast quality gate (lint + typecheck) in the dev's
workspace before the task reaches QA, so a red gate is caught at the
dev's desk instead of in QA review or CI. The full test suite stays on
CI. Fail-open: a gate-infrastructure error (missing workspace or
toolchain) is logged and never blocks the submit; only an actual check
failure blocks.
"""
try:
result = await self.git.run_pre_submit_quality_gate(ctx.agent_id, ctx.task)
except Exception as exc:
logger.warning(
"quality_gate_skipped", task_id=str(ctx.task_id), error=str(exc)
)
return None
if result.passed:
return None
return Envelope.invalid_state(
message=f"quality gate failed before QA — {result.summary}",
remediate=(
"Fix these in your workspace, commit, and call i_am_done again "
"— QA reviews working code, not a red gate:\n\n" + result.output_excerpt
),
context_briefing=ctx.briefing,
)
async def _ensure_branch_pushed(self, ctx: _IAmDoneContext) -> Envelope | None:
"""Push the task branch to origin before it reaches awaiting_qa.
+90
View File
@@ -0,0 +1,90 @@
"""Run a project's fast quality gate in a developer's workspace.
Invoked by the developer's ``i_am_done`` submit so a red gate is caught at the
dev's desk instead of surfacing in QA review or CI. The gate runs the project's
configured *non-mutating* fast checks (lint, typecheck); the slow test suite
intentionally stays on CI. The gate is fail-open on infrastructure errors (a
missing workspace or absent toolchain never blocks a submit) and fail-closed on
an actual check failure.
"""
from __future__ import annotations
import asyncio
from dataclasses import dataclass
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pathlib import Path
# A single check can be slow (mypy on a large tree); cap it generously but
# never hang the submit verb forever.
_GATE_TIMEOUT_SECONDS = 600
# Cap the remediate excerpt so a huge lint dump doesn't bloat the envelope.
_OUTPUT_EXCERPT_CHARS = 2000
@dataclass(frozen=True)
class GateResult:
"""Outcome of a pre-submit quality gate run."""
passed: bool
skipped: bool = False
failures: tuple[str, ...] = ()
output: str = ""
@property
def summary(self) -> str:
if self.skipped:
return "no quality commands configured (skipped)"
if self.passed:
return "all checks passed"
return f"failed: {', '.join(self.failures)}"
@property
def output_excerpt(self) -> str:
"""The tail of the combined output (where errors usually are)."""
return self.output[-_OUTPUT_EXCERPT_CHARS:]
async def run_quality_commands(
workspace: Path, commands: list[tuple[str, str]]
) -> GateResult:
"""Run each ``(name, command)`` in ``workspace`` and aggregate the result.
Every command runs (we don't stop at the first failure) so the developer
sees all gate failures in one shot. A non-zero exit is a failure; its
combined stdout+stderr is captured for the remediate hint.
"""
if not commands:
return GateResult(passed=True, skipped=True)
failures: list[str] = []
chunks: list[str] = []
for name, command in commands:
return_code, out = await _run_one(workspace, command)
chunks.append(f"$ {command}\n{out.strip()}")
if return_code != 0:
failures.append(name)
return GateResult(
passed=not failures,
failures=tuple(failures),
output="\n\n".join(chunks),
)
async def _run_one(workspace: Path, command: str) -> tuple[int, str]:
"""Run one operator-configured command string in the workspace shell."""
proc = await asyncio.create_subprocess_shell(
command,
cwd=str(workspace),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
)
try:
stdout, _ = await asyncio.wait_for(
proc.communicate(), timeout=_GATE_TIMEOUT_SECONDS
)
except TimeoutError:
proc.kill()
return 124, f"command timed out after {_GATE_TIMEOUT_SECONDS}s"
return proc.returncode or 0, stdout.decode("utf-8", errors="replace")
+34
View File
@@ -48,6 +48,7 @@ from roboco.services.base import (
UnauthorizedError,
ValidationError,
)
from roboco.services.gateway.quality_gate import GateResult, run_quality_commands
from roboco.services.project import get_project_service
from roboco.services.task import TaskService, get_task_service
from roboco.services.work_session import get_work_session_service
@@ -1980,6 +1981,39 @@ class GitService(BaseService):
return None
return await project_service.get(project_ids[0])
@staticmethod
def _fast_gate_commands(project: Any) -> list[tuple[str, str]]:
"""The project's non-mutating fast-gate commands (lint + typecheck).
Format and the test suite are intentionally excluded: format mutates
files, and the slow test run stays on CI.
"""
candidates = (
("lint", getattr(project, "lint_command", None)),
("typecheck", getattr(project, "typecheck_command", None)),
)
return [(name, cmd) for name, cmd in candidates if cmd]
async def run_pre_submit_quality_gate(
self, actor_agent_id: UUID, task: Any
) -> GateResult:
"""Run the project's fast quality gate in the developer's workspace.
Resolves the task's project and the developer's workspace clone, then
runs the configured lint + typecheck commands there. Returns a skipped
pass when the project configures no fast-gate commands (so projects that
opt out are never blocked). Raises only on workspace-resolution failure;
the caller treats any such failure as fail-open.
"""
project = await self._project_for_task(task)
if project is None:
return GateResult(passed=True, skipped=True)
commands = self._fast_gate_commands(project)
if not commands:
return GateResult(passed=True, skipped=True)
workspace = await self.get_workspace(project.slug, actor_agent_id)
return await run_quality_commands(workspace, commands)
async def _project_slug_for_branch(self, branch_name: str) -> str | None:
"""Resolve project slug via the task that owns the branch."""
task = await self._task_for_branch(branch_name)