Confirmation

Some operations are allowed and still deserve a question first.

Gantry Confirmation v0 is not an authentication or authorization mechanism.

Policy handles authorization. Confirmation records that the host application indicated user consent before execution — nothing more. Gantry never claims that a particular authenticated person approved anything, which is why there is no approved_by field anywhere in this model.

policy = gantry.Policy(
    name="prod-data-agents",
    rules=[
        gantry.allow.materialize(
            sources=["raw.*"],
            destinations=["prod.*"],
            require_confirmation=True,
            confirmation_code="PRODUCTION_WRITE",
            confirmation_message="This will write to production data.",
        ),
    ],
)

run = await materialize(sql)

if run.status is gantry.RunStatus.AWAITING_CONFIRMATION:
    print(run.confirmation.message)
    run = await gantry.runs.confirm(run.id)  # or gantry.runs.decline(run.id)

Nothing has happened while a run is parked

AWAITING_CONFIRMATION is neither a refusal nor an execution. The run is durable, the policy decision is recorded, and no engine has been asked to do anything. It resumes as the same run — the same immutable proposal — if the host confirms.

Host-side control

gantry.runs.confirm async

confirm(
    run_id: str,
    *,
    metadata: Mapping[str, object] | None = None,
) -> Run

The host says the user agreed. Resume the run and return what it became.

Idempotent and single-shot: the transition out of AWAITING_CONFIRMATION is a compare-and-set in the store, so of any number of concurrent or repeated confirmations exactly one releases the work and the rest return the run as it now is.

This records that the host supplied confirmation. It does not record that an authenticated person approved anything — Gantry cannot know that, and v0 does not claim it.

Source code in gantry/runs/service.py

async def confirm(run_id: str, *, metadata: Mapping[str, object] | None = None) -> Run:
    """The host says the user agreed. Resume the run and return what it became.

    Idempotent and single-shot: the transition out of `AWAITING_CONFIRMATION` is
    a compare-and-set in the store, so of any number of concurrent or repeated
    confirmations exactly one releases the work and the rest return the run as
    it now is.

    This records that the host supplied confirmation. It does not record that an
    authenticated person approved anything — Gantry cannot know that, and v0
    does not claim it.
    """
    run = _require(run_id)
    if run.status is not RunStatus.AWAITING_CONFIRMATION:
        return _already(run, ConfirmationStatus.CONFIRMED)

    parked = _pending.peek(run_id)
    if parked is None:
        raise ConfirmationError(
            f"run {run_id} is awaiting confirmation but cannot be resumed in this process; "
            "v0 resumes a parked run only where it was proposed"
        )
    _check_proposal(run, parked)

    record = _record(run, run_id).confirmed(metadata=metadata)
    # Durable before anything external happens, and atomic so that only one
    # caller is ever the reason work starts.
    if not _compare_and_set(
        run_id,
        RunStatus.AWAITING_CONFIRMATION,
        run.advanced(RunStatus.RUNNING, confirmation=record),
    ):
        return _already(_require(run_id), ConfirmationStatus.CONFIRMED)

    claimed = _pending.take(run_id)
    if claimed is None:  # pragma: no cover - another task took it between peek and take
        return _require(run_id)
    confirmed = _require(run_id)
    claimed.recorder.resumed(confirmed)
    return await claimed.resume()

gantry.runs.decline async

decline(
    run_id: str,
    *,
    metadata: Mapping[str, object] | None = None,
) -> Run

The host says no. The run becomes terminal and nothing external runs.

Source code in gantry/runs/service.py

async def decline(run_id: str, *, metadata: Mapping[str, object] | None = None) -> Run:
    """The host says no. The run becomes terminal and nothing external runs."""
    run = _require(run_id)
    if run.status is not RunStatus.AWAITING_CONFIRMATION:
        return _already(run, ConfirmationStatus.DECLINED)

    record = _record(run, run_id).declined(metadata=metadata)
    declined = run.advanced(RunStatus.CONFIRMATION_DECLINED, confirmation=record)
    if not _compare_and_set(run_id, RunStatus.AWAITING_CONFIRMATION, declined):
        return _already(_require(run_id), ConfirmationStatus.DECLINED)
    _pending.drop(run_id)
    return declined

gantry.runs.awaiting

awaiting() -> tuple[str, ...]

Run ids parked in this process, for a host that wants to list them.

Source code in gantry/runs/service.py

def awaiting() -> tuple[str, ...]:
    """Run ids parked in this process, for a host that wants to list them."""
    with _pending.lock:
        return tuple(_pending.entries)

gantry.runs.service.ConfirmationError

Bases: RuntimeError

A confirmation that cannot be honoured, refused rather than approximated.

Source code in gantry/runs/service.py

class ConfirmationError(RuntimeError):
    """A confirmation that cannot be honoured, refused rather than approximated."""

What was asked

gantry.ConfirmationRequirement dataclass

Whether to ask, and everything to say when asking.

Several rules can each want confirmation; they collapse into one requirement carrying several reasons rather than into several prompts. Being asked twice about one operation teaches a user to click through.

Source code in gantry/confirmation/model.py

@dataclass(frozen=True, slots=True)
class ConfirmationRequirement:
    """Whether to ask, and everything to say when asking.

    Several rules can each want confirmation; they collapse into one
    requirement carrying several reasons rather than into several prompts. Being
    asked twice about one operation teaches a user to click through.
    """

    required: bool = False
    reasons: tuple[ConfirmationReason, ...] = ()

    def __post_init__(self) -> None:
        object.__setattr__(self, "reasons", tuple(self.reasons))
        if self.required and not self.reasons:
            raise ValueError("a required confirmation must carry at least one reason")

    @property
    def message(self) -> str:
        """Every reason in one sentence, for a host that wants a single line."""
        return " ".join(reason.message for reason in self.reasons)

    @property
    def codes(self) -> tuple[str, ...]:
        return tuple(dict.fromkeys(reason.code.value for reason in self.reasons))

    def as_dict(self) -> dict[str, object]:
        return {
            "required": self.required,
            "reasons": [reason.as_dict() for reason in self.reasons],
        }

message property

message: str

Every reason in one sentence, for a host that wants a single line.

gantry.ConfirmationReason dataclass

One reason the host should ask before this runs.

Source code in gantry/confirmation/model.py

@dataclass(frozen=True, slots=True)
class ConfirmationReason:
    """One reason the host should ask before this runs."""

    code: ConfirmationReasonCode
    message: str
    rule: str | None = None
    details: Mapping[str, object] | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "code", ConfirmationReasonCode(self.code))
        if not self.message.strip():
            raise ValueError("a confirmation reason must say something to the user")

    def as_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {"code": self.code.value, "message": self.message}
        if self.rule is not None:
            payload["rule"] = self.rule
        if self.details:
            payload["details"] = dict(self.details)
        return payload

gantry.ConfirmationReasonCode

Bases: StrEnum

Why the host should ask, in machine-readable form.

Small on purpose. A code is for the host to decide how loudly to ask — a production write may warrant a different prompt from an expensive query — and a vocabulary nobody can enumerate cannot be used for that.

Source code in gantry/confirmation/status.py

class ConfirmationReasonCode(StrEnum):
    """Why the host should ask, in machine-readable form.

    Small on purpose. A code is for the host to decide how loudly to ask — a
    production write may warrant a different prompt from an expensive query —
    and a vocabulary nobody can enumerate cannot be used for that.
    """

    SENSITIVE_DESTINATION = "SENSITIVE_DESTINATION"
    PRODUCTION_WRITE = "PRODUCTION_WRITE"
    DESTRUCTIVE_OPERATION = "DESTRUCTIVE_OPERATION"
    HIGH_COST = "HIGH_COST"
    LONG_RUNNING_JOB = "LONG_RUNNING_JOB"
    CUSTOM_POLICY_REQUIREMENT = "CUSTOM_POLICY_REQUIREMENT"

What was answered

gantry.ConfirmationRecord dataclass

What was asked, what was answered, and which proposal it was about.

proposal_hash is the binding that makes the record mean anything: a confirmation is for one immutable proposal, so confirming one statement can never license running a different one. A run whose proposal changes needs a new run and a fresh confirmation.

Source code in gantry/confirmation/model.py

@dataclass(frozen=True, slots=True)
class ConfirmationRecord:
    """What was asked, what was answered, and which proposal it was about.

    `proposal_hash` is the binding that makes the record mean anything: a
    confirmation is for one immutable proposal, so confirming one statement can
    never license running a different one. A run whose proposal changes needs a
    new run and a fresh confirmation.
    """

    status: ConfirmationStatus
    run_id: str
    proposal_hash: str | None = None
    reasons: tuple[ConfirmationReason, ...] = ()
    required_at: datetime | None = None
    confirmed_at: datetime | None = None
    declined_at: datetime | None = None
    metadata: Mapping[str, object] = field(default_factory=dict)

    def __post_init__(self) -> None:
        object.__setattr__(self, "status", ConfirmationStatus(self.status))
        object.__setattr__(self, "reasons", tuple(self.reasons))
        oversized = [key for key, value in self.metadata.items() if len(str(value)) > 1024]
        if oversized:
            raise ValueError(
                f"confirmation metadata values must stay small; too long: "
                f"{', '.join(sorted(oversized))}"
            )

    @classmethod
    def required(
        cls,
        run_id: str,
        requirement: ConfirmationRequirement,
        *,
        proposal_hash: str | None = None,
    ) -> ConfirmationRecord:
        return cls(
            status=ConfirmationStatus.REQUIRED,
            run_id=run_id,
            proposal_hash=proposal_hash,
            reasons=requirement.reasons,
            required_at=datetime.now(UTC),
        )

    @property
    def message(self) -> str:
        """What to show the user."""
        return " ".join(reason.message for reason in self.reasons)

    @property
    def codes(self) -> tuple[str, ...]:
        return tuple(dict.fromkeys(reason.code.value for reason in self.reasons))

    def confirmed(self, *, metadata: Mapping[str, object] | None = None) -> ConfirmationRecord:
        from dataclasses import replace

        return replace(
            self,
            status=ConfirmationStatus.CONFIRMED,
            confirmed_at=datetime.now(UTC),
            metadata={**self.metadata, **(metadata or {})},
        )

    def declined(self, *, metadata: Mapping[str, object] | None = None) -> ConfirmationRecord:
        from dataclasses import replace

        return replace(
            self,
            status=ConfirmationStatus.DECLINED,
            declined_at=datetime.now(UTC),
            metadata={**self.metadata, **(metadata or {})},
        )

    def as_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {
            "status": self.status.value,
            "run_id": self.run_id,
            "proposal_hash": self.proposal_hash,
            "reasons": [reason.as_dict() for reason in self.reasons],
            "required_at": None if self.required_at is None else self.required_at.isoformat(),
            "confirmed_at": None if self.confirmed_at is None else self.confirmed_at.isoformat(),
            "declined_at": None if self.declined_at is None else self.declined_at.isoformat(),
        }
        if self.metadata:
            payload["metadata"] = {str(key): value for key, value in self.metadata.items()}
        return payload

message property

message: str

What to show the user.

gantry.ConfirmationStatus

Bases: StrEnum

Whether a run needed confirmation, and what the host said.

NOT_REQUIRED is a real answer rather than an absence: a run that never needed asking about reads differently from one still waiting, and both have to be distinguishable months later.

Source code in gantry/confirmation/status.py

class ConfirmationStatus(StrEnum):
    """Whether a run needed confirmation, and what the host said.

    `NOT_REQUIRED` is a real answer rather than an absence: a run that never
    needed asking about reads differently from one still waiting, and both have
    to be distinguishable months later.
    """

    NOT_REQUIRED = "not_required"
    """No rule asked for confirmation. Execution proceeded on policy alone."""

    REQUIRED = "required"
    """Policy allowed the operation and asked that the user be told first."""

    CONFIRMED = "confirmed"
    """The host application said the user confirmed. Not an authenticated approval."""

    DECLINED = "declined"
    """The host said no. Terminal, and nothing external ran."""

    @property
    def pending(self) -> bool:
        return self is ConfirmationStatus.REQUIRED

NOT_REQUIRED class-attribute instance-attribute

NOT_REQUIRED = 'not_required'

No rule asked for confirmation. Execution proceeded on policy alone.

REQUIRED class-attribute instance-attribute

REQUIRED = 'required'

Policy allowed the operation and asked that the user be told first.

CONFIRMED class-attribute instance-attribute

CONFIRMED = 'confirmed'

The host application said the user confirmed. Not an authenticated approval.

DECLINED class-attribute instance-attribute

DECLINED = 'declined'

The host said no. Terminal, and nothing external ran.

The transition

gantry.runs.compare_and_set

compare_and_set(
    run_id: str, expected: RunStatus, updated: Run
) -> bool

Transition a run if it is still in expected, atomically.

Used by confirmation, where only one caller may be the one that lets work start. A store that predates this method cannot make the guarantee, so the failure is explicit rather than a silently weaker transition.

Source code in gantry/runs/store.py

def compare_and_set(run_id: str, expected: RunStatus, updated: Run) -> bool:
    """Transition a run if it is still in `expected`, atomically.

    Used by confirmation, where only one caller may be the one that lets work
    start. A store that predates this method cannot make the guarantee, so the
    failure is explicit rather than a silently weaker transition.
    """
    transition = getattr(_default, "compare_and_set", None)
    if transition is None:
        raise RunPersistenceError(
            f"{type(_default).__name__} cannot transition a run atomically; "
            "confirmation needs a store that implements compare_and_set"
        )
    try:
        return bool(transition(run_id, expected, updated))
    except Exception as error:
        raise RunPersistenceError(f"could not transition run {run_id}: {error}") from error