Quality Pipeline¶

Audience: developers who want to understand how Bernstein decides whether agent output is good enough to merge.

Overview¶

After every agent finishes a task, Bernstein runs the janitor, which combines two complementary verification surfaces. The first is a structured-signal verifier that evaluates declarative completion signals attached to the task (file exists, test passes, regex match). The second is the gate pipeline - a configurable sequence of build/lint/type/test/security gates that runs on the actual diff. Only when both layers agree does the task move toward merge.

If verification fails, Bernstein doesn't just block - it feeds the failure back into the cascade-router via record_and_escalate(), which retries the task on a more capable model in the same chain. An optional cross-model verifier runs the diff past a second model from a different provider for A/B-style review; this layer is shipped but disabled by default. The result is a deterministic, programmable quality gate where escalation cost is bounded by the cascade order and observable through .sdd/metrics/.

The Janitor¶

Source: src/bernstein/core/quality/janitor.py. The janitor is the post-completion verification entry point. Inputs: a Task and the worktree path. Outputs: a JanitorResult per signal evaluated.

The janitor evaluates each CompletionSignal declared on the task (janitor.py:48-77):

A test_passes command is resolved against the tree before it runs. The path in the signal is written when the task is planned, not read off the suite, so the two drift: a plan mirroring src/bernstein/core/security/ asks for tests/unit/core/security/test_policy.py while the file lives at tests/unit/test_policy.py. pytest then exits during collection without running anything, and the task is rejected over a path the agent never chose. _resolve_test_path_command() rewrites the command onto the file that exists, but only when exactly one file under the declared root carries that basename. A basename absent everywhere means the test was never written, and an ambiguous one means the janitor cannot tell which was meant; both keep the original path so the command fails honestly.

Which evaluator runs is decided by the task's declared artifact kind. verify_task_completion() (core/tasks/artifact_completion.py) is the seam the completion paths call: a code_diff task falls through to verify_task() below, unchanged; an artifact-mode task is evaluated against the artifact it produced and completes on a signed lineage receipt rather than a commit. See ../operations/artifacts.md.

verify_task() (janitor.py:80-97) reduces all signals to a single pass/fail and a list of failure descriptions. The async run_janitor() entry point (janitor.py:171-260) is what the orchestrator calls after each agent completes. It mixes synchronous signal evaluation with async LLM judges (judge_task(), janitor.py:462), enforces a per-judge CompletionBudget, and emits one JanitorResult per evaluated task.

Two LLM-mediated paths exist for ambiguous verification:

• llm_review - synchronous, runs once per signal, expects a yes/no verdict against a rubric (janitor.py:_check_llm_review).
• llm_judge - async with retry. JUDGE_MODEL = "anthropic/ claude-sonnet-4-20250514", JUDGE_MAX_TOKENS = 1024, JUDGE_CONFIDENCE_THRESHOLD = 0.7; below the threshold, results are flagged for human review (janitor.py:36-44). The judge prompt template lives in prompts/judge.md.

The janitor never blocks merge by itself - it produces results that the orchestrator interprets. A failed janitor verification is the first escalation trigger consulted by the cascade-router (see Pipeline → cascade-router escalation below).

Gates¶

Source: src/bernstein/core/quality/gate_pipeline.py, src/bernstein/core/quality/quality_gates.py, src/bernstein/core/quality/gate_runner.py, src/bernstein/core/quality/gate_plugins.py.

A gate is a discrete check with a unique name, a required/optional flag, and an execution condition. Gates run on the diff after every agent completion, in the order the configured pipeline lists them.

The full set of recognised built-in gate names lives in gate_pipeline.py:VALID_GATE_NAMES (:16-41). The default pipeline, synthesised when quality_gates.pipeline is not explicitly set, is build_default_pipeline() in gate_pipeline.py:164-170, driven by the table at gate_pipeline.py:137-161. Each entry is (config_flag, gate_name, required, condition).

Default required gates (only those whose quality_gates.<flag>: true):

A failing required gate hard-blocks merge. A failing optional gate is reported but does not block.

behavior_probe is the only gate that executes the changed code against inputs the worker did not choose (behavior_probe.py). It walks the diff with ast, turns each parameter annotation into a fixed list of boundary values (empty/singleton/large collection, zero/negative/large int, empty/blank/long string, None where the annotation admits it) and runs one probe per subprocess in the worktree.

Its claim is crash-level, not semantic. A probe is red only when the callable raises an exception its own docstring does not document, returns a value whose type contradicts its return annotation, or does not return inside the per-callable budget. The gate never asserts what a callable should have computed - that judgement stays with the review lane.

Derivation uses no model and no randomness: the boundary vocabulary and the ordering of argument combinations are fixed, so the same files at the same content yield a byte-identical probe set. Its SHA-256 (probe_set_hash), the per-probe outcomes and the minimal failing input are attached to the gate result as metadata["probe_receipt"], so a red verdict names the exact call that produced it. Callables the deriver could not turn into inputs are listed in the receipt with a reason code (missing-annotation, unsupported-annotation, receiver-required, async-callable, variadic-parameter, keyword-only-parameter, unparsable-module, cap-exceeded) - absence is recorded, never silent. Runtime is bounded by behavior_probe_per_callable_timeout_s, behavior_probe_gate_timeout_s and the two probe caps.

Gate conditions (gate_pipeline.py:42) gate execution by what changed: always, python_changed, tests_changed, any_changed, deps_changed. The legacy condition string changed_files.any('.py') is normalised to python_changed (gate_pipeline.py:74-81).

Adding a custom gate¶

Custom gates plug in through the bernstein.gates entry-point group (gate_plugins.py:107-120) or via a Python file dropped into .bernstein/gates/*.py (gate_plugins.py:87-105). Both modes load classes that subclass GatePlugin (gate_plugins.py:20-46):

from pathlib import Path
from bernstein.core.quality.gate_plugins import GatePlugin
from bernstein.core.quality.gate_runner import GateResult


class NoFooGate(GatePlugin):
    @property
    def name(self) -> str:
        return "no_foo"

    @property
    def required(self) -> bool:
        return True

    @property
    def condition(self) -> str:
        return "any_changed"

    def run(
        self,
        changed_files: list[str],
        run_dir: Path,
        task_title: str,
        task_description: str,
    ) -> GateResult:
        offending = [f for f in changed_files if "foo" in Path(f).read_text()]
        passed = not offending
        return GateResult(
            name=self.name,
            status="pass" if passed else "fail",
            required=self.required,
            blocked=not passed,
            cached=False,
            duration_ms=0,
            details=f"Found 'foo' in {offending}" if offending else "Clean",
        )

Register via pyproject.toml:

[project.entry-points."bernstein.gates"]
no_foo = "my_pkg.gates:NoFooGate"

The plugin name must not collide with a built-in (gate_plugins.py:81-82). Names are validated and duplicates raise. File-based plugins under .bernstein/gates/ are loaded for ad-hoc project-local checks; they have the same lifecycle but are not packaged.

Cross-model verifier¶

Source: src/bernstein/core/quality/cross_model_verifier.py. This is the "writer != reviewer" layer: after an agent finishes, the diff is sent to a different model (a cheap one from a different provider) with a focused code-review prompt.

The default reviewer mapping (cross_model_verifier.py:37-43):

CrossModelVerifierConfig (:84-106) is enabled=True as a class default, but the orchestrator config wires it off by default - operators must enable it explicitly via quality_gates.cross_model.enabled: true.

The reviewer is asked for one of two verdicts (:120-123):

• approve - diff is fine.
• request_changes - diff has issues. When block_on_issues=True (default), this prevents merge and creates a fix task; otherwise findings are logged only.

For higher-stakes deployments, voting_config: VotingConfig lets you elect multiple reviewer models and apply quorum logic (cross_model_verifier.py:106). A single reviewer is the default QUORUM(1,1) behaviour.

Cost controls baked into the module (:29-34): diff truncated at 12,000 chars, response capped at 512 tokens, provider="openrouter". With default reviewers this is in the cents-per-task range.

Pipeline → cascade-router escalation¶

The whole pipeline exists to feed information back into the cascade- router so weaker-but-cheaper models can be tried first. The escalation contract is in src/bernstein/core/routing/cascade_router.py.

After the orchestrator records a completed attempt, it calls CascadeRouter.record_and_escalate(chain_id, task, attempt, janitor_passed=..., output=...) (cascade_router.py:386-478). The function consults _should_escalate() (:639-673) in this order:

1. Hard task failure - attempt.success=False with no other context → escalate (:655-657).
2. Janitor verification failure - janitor_passed=False → escalate (:660-661). This is the wire from janitor results into model escalation.
3. Low-confidence regex on agent output - detect_low_confidence() scans the last 2,000 chars for phrases like "I'm not sure", "partial implementation", "TODO: escalat" (:371-384, _LOW_CONFIDENCE_PATTERN). When matched, escalate.
4. Explicit failure flag - attempt.success=False after the above checks (:670-671).

If any trigger fires, the cascade list (_cascade_for_task()) is consulted: tasks step sonnet → opus. The earlier haiku tier was retired, so standard and high-stakes tasks now share the same chain. When the current model is already at the top, escalation gives up.

The bandit (EpsilonGreedyBandit from core/cost/cost.py) is updated on every observation (cascade_router.py). On the next call to select() for a fresh task, the router proactively skips a tier when observations >= MIN_OBSERVATIONS and success_rate < QUALITY_THRESHOLD - i.e. the bandit learns "sonnet rarely works for role=qa, start at opus."

Chain reports persist to .sdd/metrics/cascade_chains.jsonl (save_chain(), :518-539). Each line lists every attempt with {model, cost_usd, latency_s, success, escalated, escalation_reason}, the final model, total cost, and saved_vs_direct_opus_usd.

The full cross-adapter (rather than intra-Claude) story - what happens on rate-limit / timeout / API error - lives in core/routing/cascade.py:CascadeFallbackManager. Both surfaces are documented end-to-end in Model routing.

Configuration¶

All knobs live under quality_gates.* in bernstein.yaml. The dataclass that defines them is QualityGatesConfig (core/quality/quality_gates.py:135-265). Highlights:

quality_gates:
  enabled: true                  # master switch
  lint: true
  lint_command: "ruff check ."
  type_check: false
  type_check_command: "pyright"
  tests: false
  test_command: "uv run python scripts/run_tests.py -x"
  timeout_s: 120
  base_ref: "main"               # base for incremental diff
  cache_enabled: true            # reuse gate results when diff is unchanged
  allow_bypass: false            # whether the CLI can skip gates

  pii_scan: true
  dlp_scan: true
  security_scan: false
  coverage_delta: false
  complexity_check: false
  dead_code_check: false
  comment_quality_check: false
  import_cycle_check: false
  merge_conflict_check: false
  mutation_testing: false
  dep_audit: false
  benchmark:
    enabled: false

  intent_verification:
    enabled: false               # LLM-based "did this satisfy intent?"
    model: "google/gemini-flash-1.5"
    block_on_no: true

  cross_model:                   # cross-model verifier (writer != reviewer)
    enabled: false

When pipeline: is omitted, Bernstein synthesises one from the booleans above. To override the order (or insert custom gates), declare an explicit pipeline:

quality_gates:
  pipeline:
    - { name: "lint",       required: true,  condition: "always" }
    - { name: "type_check", required: true,  condition: "python_changed" }
    - { name: "tests",      required: true,  condition: "python_changed" }
    - { name: "no_foo",     required: true,  condition: "any_changed" }  # custom

Observability¶

Quality endpoints (FastAPI, all in core/routes/quality.py and file_health.py):

On-disk artefacts:

• .sdd/metrics/quality_gates.jsonl - one line per gate execution (quality_gates.py:1148).
• .sdd/metrics/cascade_chains.jsonl - one line per cascade chain completion (cascade_router.py:533).
• .sdd/metrics/tasks.jsonl - task lifecycle events used by behaviour anomaly detection.

Trend reads, file-health rollups, and budget forecasts all stream from these JSONL files; tail them directly when the API is unavailable. See Observability overview for how these signals integrate with Prometheus, Grafana, and SLOs.

Code pointers¶