spanforge.sdk.gate — CI/CD Gate Pipeline

Module: spanforge.sdk.gate
Added in: 2.0.7 (Phase 8 — CI/CD Gate Pipeline)
Import: from spanforge.sdk import sf_gate or from spanforge.sdk.gate import SFGateClient

The gate module provides the SFGateClient singleton and the GateRunner YAML engine. It evaluates quality gates (schema validation, dependency security, secrets scanning, performance regression, and hallucination checks) against AI system artifacts and emits structured GateArtifact records with durable storage.

Quick example

from spanforge.sdk import sf_gate
from spanforge.sdk import GateVerdict

# Evaluate a single gate
result = sf_gate.evaluate(
    gate_id="schema-check",
    payload={"schema_version": "2.0", "source": "my-app@1.0.0"},
)
if result.verdict == GateVerdict.PASS:
    print(f"Gate passed in {result.duration_ms:.0f} ms")
else:
    raise RuntimeError(f"Gate {result.gate_id} {result.verdict.value}: {result.metrics}")

Singleton

spanforge.sdk.sf_gate is a module-level SFGateClient instance constructed from environment variables. For most use-cases, import and use the singleton:

from spanforge.sdk import sf_gate

status = sf_gate.get_status()
print(status.gate_count, status.artifact_count)

SFGateClient

class SFGateClient:
    def evaluate(
        self,
        gate_id: str,
        payload: dict,
        *,
        project_id: str | None = None,
    ) -> GateEvaluationResult: ...

    def evaluate_prri(
        self,
        prri_score: float,
        *,
        project_id: str | None = None,
        framework: str = "default",
        policy_file: str | None = None,
        dimension_breakdown: dict[str, float] | None = None,
    ) -> PRRIResult: ...

    def run_pipeline(
        self,
        gate_config_path: str,
        *,
        context: dict | None = None,
    ) -> GateRunResult: ...

    def get_artifact(self, gate_id: str) -> GateArtifact | None: ...

    def list_artifacts(self, *, project_id: str | None = None) -> list[GateArtifact]: ...

    def purge_artifacts(self, *, older_than_days: int = 90) -> int: ...

    def get_status(self) -> GateStatusInfo: ...

    def configure(self, config: dict) -> None: ...

evaluate(gate_id, payload, *, project_id) → GateEvaluationResult

Evaluate a single named gate against payload.

evaluate_async() — async variant

Added in: 2.0.14

async def evaluate_async(
    self,
    gate_id: str,
    payload: dict,
    *,
    project_id: str = "",
    pipeline_id: str = "",
) -> GateEvaluationResult

Non-blocking async variant of evaluate(). Delegates to the synchronous method via asyncio.get_event_loop().run_in_executor() so it never blocks the event loop. Ideal for async CI/CD runners and FastAPI-based pipelines.

Returns GateEvaluationResult

Example:

import asyncio
from spanforge.sdk import sf_gate

result = asyncio.run(
    sf_gate.evaluate_async("schema-check", {"schema_version": "2.0"})
)
assert result.verdict.value == "pass"

The evaluation applies the gate logic registered under gate_id (schema validation, secrets scan, dependency audit, performance regression check, or hallucination verifier). The result is written to the artifact store.

Parameters

Returns GateEvaluationResult

Raises

Example

from spanforge.sdk import sf_gate, GateVerdict

result = sf_gate.evaluate("dependency-audit", {"package": "requests==2.28.0"})
assert result.verdict in (GateVerdict.PASS, GateVerdict.WARN)
print(result.metrics)         # {"cve_count": 0, "severity_max": "NONE"}
print(result.artifact_url)    # ".sf-gate/artifacts/dependency-audit-<ulid>.json"

evaluate_prri(prri_score, *, project_id, framework, policy_file, dimension_breakdown) → PRRIResult

Evaluate a Pre-Release Readiness Index (PRRI) score against policy thresholds.

Scores at or above SPANFORGE_GATE_PRRI_RED_THRESHOLD (default 70) receive a RED verdict, which blocks release when on_fail="block". Scores between 30 and 69 receive AMBER (warn). Scores below 30 receive GREEN (pass).

Parameters

Returns PRRIResult

Raises SFGateEvaluationError — policy file not found, or prri_score outside [0, 100].

Example

result = sf_gate.evaluate_prri(
    65.0,
    project_id="my-agent",
    dimension_breakdown={"hallucination": 0.65, "bias": 0.50},
)
print(result.verdict)    # PRRIVerdict.AMBER
print(result.allow)      # True  (AMBER warns but does not block by default)

run_pipeline(gate_config_path, *, context) → GateRunResult

Parse and execute a YAML gate pipeline file.

The YAML file defines an ordered list of gates. Each gate has an id, type, optional pass_condition, on_fail policy (block, warn, or report), and optional parallel flag. Gates with on_fail: block that evaluate to FAIL raise SFGatePipelineError.

Parameters

Returns GateRunResult — aggregate result with per-gate GateArtifact list.

Raises

Example

from spanforge.sdk import sf_gate

run = sf_gate.run_pipeline(
    "examples/gates/sf-gate.yaml",
    context={"project_id": "my-agent", "env": "staging"},
)
print(run.passed, run.failed, run.warned)
for artifact in run.artifacts:
    print(f"  {artifact.gate_id}: {artifact.verdict.value}")

Gate YAML format

gates:
  - id: schema-validation
    name: "Schema Validation"
    type: schema_validation
    on_fail: block
    artifact: true

  - id: secrets-scan
    name: "Secrets Scan"
    type: secrets_scan
    on_fail: block
    artifact: true

  - id: perf-regression
    name: "Performance Regression"
    type: performance_regression
    pass_condition: "p99_latency_ms < 2000"
    on_fail: warn
    artifact: true

  - id: prri-check
    name: "PRRI Readiness Gate"
    type: halluccheck_prri
    framework: default
    on_fail: block
    artifact: true

  - id: trust-gate
    name: "Trust Gate"
    type: halluccheck_trust
    on_fail: block
    artifact: true

Gate types

Gate YAML fields

get_artifact(gate_id) → GateArtifact | None

Retrieve the most recent stored artifact for a gate by gate_id.

Returns None if no artifact is found.

artifact = sf_gate.get_artifact("schema-validation")
if artifact:
    print(artifact.verdict, artifact.timestamp)

list_artifacts(*, project_id) → list[GateArtifact]

List all stored artifacts, optionally filtered to a project_id.

Returns artifacts sorted most-recent-first.

artifacts = sf_gate.list_artifacts(project_id="my-agent")
for a in artifacts:
    print(f"{a.gate_id}: {a.verdict.value} @ {a.timestamp}")

purge_artifacts(*, older_than_days) → int

Delete artifacts older than older_than_days. Returns the count of files removed from the artifact store.

removed = sf_gate.purge_artifacts(older_than_days=30)
print(f"Purged {removed} stale artifact(s)")

get_status() → GateStatusInfo

Return a live status snapshot of the gate subsystem.

status = sf_gate.get_status()
print(status.status)         # "ok"
print(status.gate_count)     # number of registered gate types
print(status.artifact_count) # total artifacts on disk
print(status.healthy)        # True

configure(config) → None

Override gate settings at runtime. Any keys not present keep their current (env-var-sourced or default) values.

sf_gate.configure({
    "artifact_dir": "/var/spanforge/gate-artifacts",
    "artifact_retention_days": 30,
    "prri_red_threshold": 60,
    "hri_critical_threshold": 0.03,
})

Types

All types are exported from spanforge.sdk:

from spanforge.sdk import (
    GateVerdict,
    PRRIVerdict,
    GateArtifact,
    GateEvaluationResult,
    PRRIResult,
    TrustGateResult,
    GateStatusInfo,
)

GateVerdict

class GateVerdict(str, Enum):
    PASS    = "PASS"
    FAIL    = "FAIL"
    WARN    = "WARN"
    SKIPPED = "SKIPPED"

PRRIVerdict

class PRRIVerdict(str, Enum):
    GREEN = "GREEN"   # score < 30 — release ready
    AMBER = "AMBER"   # 30 ≤ score < 70 — warn
    RED   = "RED"     # score ≥ 70 — block

GateArtifact

Immutable record written to the artifact store after gate evaluation.

GateEvaluationResult

Returned by evaluate().

PRRIResult

Returned by evaluate_prri().

TrustGateResult

Returned internally by the halluccheck_trust executor; also accessible via get_artifact() after a trust gate run.

GateStatusInfo

Returned by get_status().

Exceptions

All exceptions are exported from spanforge.sdk:

from spanforge.sdk import (
    SFGateError,
    SFGateEvaluationError,
    SFGatePipelineError,
    SFGateTrustFailedError,
    SFGateSchemaError,
)

Environment variables

Test coverage

The gate module has dedicated subprocess-level mock tests for every built-in executor (tests/test_coverage_gaps.py :: TestGateExecutorSubprocessMocks).

Additional gate tests in tests/test_sf_gate.py:

• TestInferVerdictBranches (18 tests) — exercises every _infer_verdict path.
• TestPostEvaluateHooksSideEffects (7 tests) — hook execution, failure isolation, ordering, and metrics access.
• TestGateExecutorHelpers — _evaluate_pass_condition and _substitute_template edge cases.

See also

• Gate Pipeline user guide — Getting started, YAML examples, CI/CD integration
• alert — SFAlertClient (Phase 7) triggered by SFGateTrustFailedError
• secrets — Secrets scanner underlying the secrets_scan gate type
• pii — PII scanner underlying the trust gate PII window check