Skip to content

Results and outputs

What an operation hands back. The question these types exist to keep separate is "did it run" versus "may the answer be believed" — status answers the first, verification the second.

gantry.sql.SQLResult dataclass

The outcome of a governed SQL call: rows, references, and evidence.

Check ok rather than the absence of a failure. inline holds rows that came back with the result; uri is the first engine-owned output that is not inline, for results too large to pass through Gantry.

Source code in gantry/sql/result.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@dataclass(frozen=True, slots=True)
class SQLResult:
    """The outcome of a governed SQL call: rows, references, and evidence.

    Check `ok` rather than the absence of a failure. `inline` holds rows that
    came back with the result; `uri` is the first engine-owned output that is
    not inline, for results too large to pass through Gantry.
    """

    status: ResultStatus
    handle: ExecutionHandle | None = None
    inline: InlineRows | None = None
    outputs: tuple[OutputRef, ...] = ()
    metrics: ExecutionMetrics = field(default_factory=ExecutionMetrics)
    verification: VerificationResult | None = None
    failure: Failure | None = None
    evidence: EvidenceBundle | None = None
    """What Gantry observed while deciding — the same shape a materialization emits."""

    @property
    def ok(self) -> bool:
        return self.status is ResultStatus.ACCEPTED

    @property
    def uri(self) -> str | None:
        """Return the first engine-owned output URI, excluding inline rows."""

        return next(
            (output.uri for output in self.outputs if output.kind is not OutputKind.INLINE),
            None,
        )

evidence class-attribute instance-attribute 

evidence: EvidenceBundle | None = None

What Gantry observed while deciding — the same shape a materialization emits.

uri property 

uri: str | None

Return the first engine-owned output URI, excluding inline rows.

gantry.Result dataclass

The terminal outcome of governed work: status, outputs, and evidence.

Check is_accepted rather than the absence of a failure — an engine success whose verification failed is not accepted. admission and verification are kept so the result carries its own justification: what was allowed, and what was checked afterwards.

Source code in gantry/result.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@dataclass(frozen=True, slots=True)
class Result:
    """The terminal outcome of governed work: status, outputs, and evidence.

    Check `is_accepted` rather than the absence of a failure — an engine
    success whose verification failed is not accepted. `admission` and
    `verification` are kept so the result carries its own justification: what
    was allowed, and what was checked afterwards.
    """

    status: ResultStatus
    handle: ExecutionHandle | None = None
    execution: Execution | None = None
    outputs: tuple[OutputRef, ...] = ()
    metrics: ExecutionMetrics = field(default_factory=ExecutionMetrics)
    verification: VerificationResult | None = None
    failure: Failure | None = None
    admission: AdmissionDecision | None = None

    @property
    def is_accepted(self) -> bool:
        return self.status is ResultStatus.ACCEPTED

    @classmethod
    def rejected(cls, admission: AdmissionDecision) -> Result:
        failure = (
            admission.reasons[0]
            if admission.reasons
            else Failure(
                kind=FailureKind.POLICY_REJECTED,
                retryable=False,
                message="execution was not admitted",
            )
        )
        return cls(status=ResultStatus.REJECTED, failure=failure, admission=admission)

    @classmethod
    def terminal_failure(
        cls,
        status: ResultStatus,
        failure: Failure,
        *,
        handle: ExecutionHandle | None = None,
        execution: Execution | None = None,
        admission: AdmissionDecision | None = None,
    ) -> Result:
        return cls(
            status=status,
            handle=handle,
            execution=execution,
            metrics=execution.metrics if execution is not None else ExecutionMetrics(),
            failure=failure,
            admission=admission,
        )

    @classmethod
    def from_execution(
        cls,
        *,
        execution: Execution,
        engine_result: ExecutionResult,
        verification: VerificationResult,
        admission: AdmissionDecision,
    ) -> Result:
        if not verification.ok:
            # A check the provider could not evaluate is reported as such. Both
            # outcomes reject — an unmeasured bound is not a bound — but they
            # call for different remedies, and only one of them is about the
            # data.
            unsupported = verification.unsupported_checks
            kind = (
                FailureKind.UNSUPPORTED_VERIFICATION
                if unsupported
                else FailureKind.VERIFICATION_FAILED
            )
            reasons = unsupported or verification.failed_checks
            message = next(
                (check.message for check in reasons if check.message),
                f"{reasons[0].name} could not be evaluated"
                if unsupported
                else "verification failed",
            )
            return cls(
                status=ResultStatus.VERIFICATION_FAILED,
                handle=execution.handle,
                execution=execution,
                outputs=engine_result.outputs,
                metrics=engine_result.metrics,
                verification=verification,
                failure=Failure(
                    kind=kind,
                    retryable=False,
                    message=message,
                ),
                admission=admission,
            )
        return cls(
            status=ResultStatus.ACCEPTED,
            handle=execution.handle,
            execution=execution,
            outputs=engine_result.outputs,
            metrics=engine_result.metrics,
            verification=verification,
            admission=admission,
        )

gantry.ResultStatus

Bases: StrEnum

The five ways work ends, separating "ran" from "can be believed".

REJECTED never reached the engine. FAILED and CANCELLED ran and did not finish. VERIFICATION_FAILED finished and is not trustworthy. UNKNOWN means Gantry cannot say, which is not the same as failure.

Source code in gantry/result.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class ResultStatus(StrEnum):
    """The five ways work ends, separating "ran" from "can be believed".

    `REJECTED` never reached the engine. `FAILED` and `CANCELLED` ran and did
    not finish. `VERIFICATION_FAILED` finished and is not trustworthy.
    `UNKNOWN` means Gantry cannot say, which is not the same as failure.
    """

    ACCEPTED = "ACCEPTED"
    REJECTED = "REJECTED"
    FAILED = "FAILED"
    CANCELLED = "CANCELLED"
    VERIFICATION_FAILED = "VERIFICATION_FAILED"
    VERIFICATION_UNSUPPORTED = "VERIFICATION_UNSUPPORTED"
    VERIFICATION_CONFLICT = "VERIFICATION_CONFLICT"
    UNKNOWN = "UNKNOWN"

Rows and references

An adapter returns rows inline when they are bounded, and a reference when the engine produced something too large to inline. Which you get depends on the provider.

gantry.sql.InlineRows dataclass

Rows returned with the result, already bounded by the policy.

truncated says the engine had more rows than max_rows allowed, so a caller can tell a complete small answer from a clipped large one. Construction rejects rows whose width does not match columns.

Source code in gantry/sql/output.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@dataclass(frozen=True, slots=True)
class InlineRows:
    """Rows returned with the result, already bounded by the policy.

    `truncated` says the engine had more rows than `max_rows` allowed, so a
    caller can tell a complete small answer from a clipped large one.
    Construction rejects rows whose width does not match `columns`.
    """

    columns: tuple[str, ...]
    rows: tuple[tuple[object, ...], ...]
    truncated: bool = False

    def __post_init__(self) -> None:
        width = len(self.columns)
        if any(len(row) != width for row in self.rows):
            raise ValueError("inline row width must match the column count")

gantry.OutputRef dataclass

A reference to something an execution produced.

The uri is interpreted per kind and must be non-empty. Gantry returns references rather than data so that a large result does not have to pass through the process governing it.

Source code in gantry/output.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@dataclass(frozen=True, slots=True)
class OutputRef:
    """A reference to something an execution produced.

    The `uri` is interpreted per `kind` and must be non-empty. Gantry returns
    references rather than data so that a large result does not have to pass
    through the process governing it.
    """

    kind: OutputKind
    uri: str
    metadata: Mapping[str, object] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.uri.strip():
            raise ValueError("output reference URI must not be empty")

gantry.OutputKind

Bases: StrEnum

Where an execution's output lives, which decides how to read it.

INLINE means the rows came back with the result; everything else is a reference to somewhere the engine wrote.

Source code in gantry/output.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class OutputKind(StrEnum):
    """Where an execution's output lives, which decides how to read it.

    `INLINE` means the rows came back with the result; everything else is a
    reference to somewhere the engine wrote.
    """

    INLINE = "INLINE"
    TABLE = "TABLE"
    DATASET = "DATASET"
    FILE = "FILE"
    OBJECT = "OBJECT"
    STREAM = "STREAM"
    CUSTOM = "CUSTOM"

Metrics

gantry.ExecutionMetrics dataclass

Engine-reported counters, portable across adapters and all optional.

Every field is None when the engine does not report it, which is deliberately distinct from zero: "no rows were written" and "this engine does not count rows" lead to different conclusions. native keeps the engine's own payload for anything this model does not model.

Source code in gantry/metrics.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@dataclass(frozen=True, slots=True)
class ExecutionMetrics:
    """Engine-reported counters, portable across adapters and all optional.

    Every field is `None` when the engine does not report it, which is
    deliberately distinct from zero: "no rows were written" and "this engine
    does not count rows" lead to different conclusions. `native` keeps the
    engine's own payload for anything this model does not model.
    """

    rows_read: int | None = None
    rows_written: int | None = None
    bytes_read: int | None = None
    bytes_written: int | None = None
    runtime_seconds: float | None = None
    estimated_cost_usd: float | None = None
    worker_count: int | None = None
    native: Mapping[str, object] = field(default_factory=dict)

Verification

gantry.VerificationResult dataclass

Whether the result may be believed, with the checks that decided it.

A failing verification turns an engine success into ResultStatus.VERIFICATION_FAILED: the job ran, and the answer is still not usable. Build one with passed() or failed().

Source code in gantry/verifier.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@dataclass(frozen=True, slots=True, init=False)
class VerificationResult:
    """Whether the result may be believed, with the checks that decided it.

    A failing verification turns an engine success into
    `ResultStatus.VERIFICATION_FAILED`: the job ran, and the answer is still
    not usable. Build one with `passed()` or `failed()`.
    """

    ok: bool
    checks: tuple[CheckResult, ...] = ()
    metadata: Mapping[str, object] = field(default_factory=dict)

    def __init__(
        self,
        ok: bool | None = None,
        checks: tuple[CheckResult, ...] = (),
        metadata: Mapping[str, object] | None = None,
        *,
        passed: bool | None = None,
    ) -> None:
        """Accept ``passed=`` from v0.2 while retaining ``ok=`` compatibility."""
        resolved = passed if passed is not None else ok
        if resolved is None:
            raise TypeError("passed is required")
        if passed is not None and ok is not None and passed != ok:
            raise ValueError("passed and ok disagree")
        object.__setattr__(self, "ok", resolved)
        object.__setattr__(self, "checks", tuple(checks))
        object.__setattr__(self, "metadata", {} if metadata is None else metadata)

    @property
    def failed_checks(self) -> tuple[CheckResult, ...]:
        """The checks that decided against acceptance, for a caller to act on."""
        return tuple(check for check in self.checks if not check.ok)

    @property
    def unsupported_checks(self) -> tuple[CheckResult, ...]:
        """Checks the provider could not evaluate.

        Separate from `failed_checks` because the remedy is different: a failed
        check means fix the data or the query, an unsupported one means this
        provider cannot answer the question and something has to change about
        the policy or the backend.
        """
        return tuple(check for check in self.checks if not check.supported)

    @classmethod
    def passed(cls, *checks: CheckResult) -> VerificationResult:
        return cls(ok=True, checks=checks)

    @classmethod
    def failed(
        cls,
        message: str,
        *,
        name: str = "verification",
        expected: object | None = None,
        actual: object | None = None,
    ) -> VerificationResult:
        return cls(
            ok=False,
            checks=(
                CheckResult(
                    name=name,
                    ok=False,
                    expected=expected,
                    actual=actual,
                    message=message,
                ),
            ),
        )

failed_checks property 

failed_checks: tuple[CheckResult, ...]

The checks that decided against acceptance, for a caller to act on.

unsupported_checks property 

unsupported_checks: tuple[CheckResult, ...]

Checks the provider could not evaluate.

Separate from failed_checks because the remedy is different: a failed check means fix the data or the query, an unsupported one means this provider cannot answer the question and something has to change about the policy or the backend.

gantry.CheckResult dataclass

One verification check and what it observed.

expected and actual are recorded even when the check passes, so an accepted result still carries the evidence for why it was accepted.

Source code in gantry/verifier.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@dataclass(frozen=True, slots=True, init=False)
class CheckResult:
    """One verification check and what it observed.

    `expected` and `actual` are recorded even when the check passes, so an
    accepted result still carries the evidence for why it was accepted.
    """

    name: str
    ok: bool
    expected: object | None = None
    actual: object | None = None
    message: str | None = None
    metadata: Mapping[str, object] = field(default_factory=dict)
    supported: bool = True
    """Whether the provider could evaluate this check at all.

    A check that could not be evaluated fails, because a bound nobody measured
    is not a bound. But "the destination had no rows" and "I could not count
    the rows" are different facts, and collapsing them hides the second — which
    is a gap in the provider, not in the data.
    """

    source: CheckSource | str | None = None
    """Who supplied the requirement: ``trusted`` or ``agent``.

    Strings remain accepted for evidence written before provenance existed.
    New governed operations assign :class:`CheckSource` themselves; it is not
    accepted in agent verification input.
    """

    evidence_refs: tuple[str, ...] = ()
    """Bounded references to observations supporting the decision."""

    def __init__(
        self,
        name: str | None = None,
        ok: bool | None = None,
        expected: object | None = None,
        actual: object | None = None,
        message: str | None = None,
        metadata: Mapping[str, object] | None = None,
        supported: bool = True,
        source: CheckSource | str | None = None,
        evidence_refs: tuple[str, ...] = (),
        *,
        check: str | None = None,
        passed: bool | None = None,
        observed: object | None = None,
    ) -> None:
        """Accept both the v0.2 vocabulary and the pre-v0.2 field names."""
        resolved_name = check if check is not None else name
        resolved_ok = passed if passed is not None else ok
        if resolved_name is None:
            raise TypeError("check is required")
        if resolved_ok is None:
            raise TypeError("passed is required")
        if check is not None and name is not None and check != name:
            raise ValueError("check and name disagree")
        if passed is not None and ok is not None and passed != ok:
            raise ValueError("passed and ok disagree")
        if observed is not None and actual is not None and observed != actual:
            raise ValueError("observed and actual disagree")
        object.__setattr__(self, "name", resolved_name)
        object.__setattr__(self, "ok", resolved_ok)
        object.__setattr__(self, "expected", expected)
        object.__setattr__(self, "actual", observed if observed is not None else actual)
        object.__setattr__(self, "message", message)
        object.__setattr__(self, "metadata", {} if metadata is None else metadata)
        object.__setattr__(self, "supported", supported)
        object.__setattr__(self, "source", source)
        object.__setattr__(self, "evidence_refs", tuple(evidence_refs))

    @property
    def check(self) -> str:
        """Specification spelling for :attr:`name`."""
        return self.name

    @property
    def passed(self) -> bool:
        """Specification spelling for :attr:`ok`."""
        return self.ok

    @property
    def observed(self) -> object | None:
        """Specification spelling for :attr:`actual`."""
        return self.actual

supported class-attribute instance-attribute 

supported: bool = True

Whether the provider could evaluate this check at all.

A check that could not be evaluated fails, because a bound nobody measured is not a bound. But "the destination had no rows" and "I could not count the rows" are different facts, and collapsing them hides the second — which is a gap in the provider, not in the data.

source class-attribute instance-attribute 

source: CheckSource | str | None = None

Who supplied the requirement: trusted or agent.

Strings remain accepted for evidence written before provenance existed. New governed operations assign :class:CheckSource themselves; it is not accepted in agent verification input.

evidence_refs class-attribute instance-attribute 

evidence_refs: tuple[str, ...] = ()

Bounded references to observations supporting the decision.

check property 

check: str

Specification spelling for :attr:name.

passed property 

passed: bool

Specification spelling for :attr:ok.

observed property 

observed: object | None

Specification spelling for :attr:actual.

gantry.Verifier

Bases: Protocol

A check run after the engine succeeds, deciding whether to accept.

Implement verify and pass instances to wait or run. It receives what was proposed (artifact, context) and what happened (execution, result), so it can compare the two rather than trusting either.

Source code in gantry/verifier.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
@runtime_checkable
class Verifier(Protocol):
    """A check run after the engine succeeds, deciding whether to accept.

    Implement `verify` and pass instances to `wait` or `run`. It receives what
    was proposed (`artifact`, `context`) and what happened (`execution`,
    `result`), so it can compare the two rather than trusting either.
    """

    async def verify(
        self,
        *,
        artifact: Artifact,
        context: Context,
        execution: Execution,
        result: ExecutionResult,
    ) -> VerificationResult: ...