feat(connectors): draft check runs can run, re-run, skip or stop one check

Draft runs and pair dry runs take per-check actions: check_ids (run or
re-run only these checks), skip_check_ids and start_check_id (Start Now).
A skipped optional check is skipped (skipped_by_user). A skipped required
check is stopped (stopped_by_user): it is no pass and holds Create. Its
dependents are "Not run". Creation reuses an optional skip and always runs
a required check. Invalid actions give INVALID_INPUT.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KFMazQbuG8ui3jxBxm47sC
This commit is contained in:
Claude committed 2026-10-17 03:15:14 +00:00
1 parent aaf49caab5
commit b15f7e0d80
11 files changed
+769 -48

No files matched your search

@@ -523,6 +523,18 @@ same draft key supersedes an older one, and the cancel endpoint does the same
without a successor: the task stops before its next check. The plan endpoint
(`/admin/connector-checks/plan`) lists the same checks without a credential or a
run, so the form can tell which checks exist and which are required first.
A draft run or a pair dry run (`/admin/cc-pair/{cc_pair_id}/edit/checks`) also
takes per-check actions (`draft_runs.py:DraftCheckActions`). `check_ids` runs
only those checks, without their cached results. Every other check keeps its
cached result or its result in the latest run of the draft key, and runs only
when it has none, for example after a dependency that now runs.
`skip_check_ids` do not run: an optional check is `SKIPPED`
(`not_run_reason: skipped_by_user`), and a required one is `STOPPED`
(`stopped_by_user`), which is no pass and holds Create. Their dependents are
"Not run" (`dependency_skipped`). Creation reuses a cached optional skip and
always runs a stopped required check. `start_check_id` runs that check and the
checks it needs first. An unknown check id, or a check to run and to skip, is
`INVALID_INPUT`. Stored runs take no actions.
A passing check may return *findings* (`capability_checks/models.py:CheckFinding`):
structured information about what the credential and config reach, such as a
@@ -339,7 +339,9 @@ def run_draft_capability_checks_task(
try:
# Inside the try, so a failure here marks the run FAILED_TO_RUN rather
# than leaving it RUNNING until its lease runs out.
pending = pending_in_run_order(snapshot, get_capability_checks(snapshot.source))
pending = pending_in_run_order(
snapshot, get_capability_checks(snapshot.source), run.first_check_id
)
pending_result_keys = list(
dict.fromkeys(run.result_cache_keys[check.check_id] for check in pending)
)
@@ -379,6 +381,7 @@ def run_draft_capability_checks_task(
timeout_cap_seconds=DRAFT_CHECK_TIMEOUT_SECONDS,
prior_results=settled_results(snapshot),
form_run=True,
first_check_id=run.first_check_id,
)
snapshot.status = DraftRunStatus.COMPLETED
save_draft_run(run)
@@ -5,9 +5,11 @@ For a source with named checks, this replaces the legacy blocking validation
(``validate_connector_settings`` and ``validate_perm_sync``). It runs the
source's checks for the pairing's access type and stores the full report. A
fresh result of a draft run on the same form (for an edit, of a dry run on the
same pair and proposed state) is reused, so the check does not run again. Only
a required check that failed within the blocking budget blocks the pairing:
INCOMPLETE is transient, and skipped checks do not apply. A check that is
same pair and proposed state) is reused, so the check does not run again. An
optional check the admin skipped on that form is reused as skipped; a required
check always runs, also when the admin stopped it. Only a required check that
failed within the blocking budget blocks the pairing: INCOMPLETE is transient,
and skipped checks do not apply. A check that is
still running at the end of the budget does not block; a background run
produces its result.
"""
@@ -133,7 +135,10 @@ def _fresh_draft_results(
) -> dict[str, CachedDraftResult]:
"""The cached draft results that a validation reuses. A cached FAILED
result is left out, so the check runs again: the source may have been
fixed since, and a client without a draft run cannot ask for a re-run."""
fixed since, and a client without a draft run cannot ask for a re-run.
The admin's skip is reused only for an optional check: a skip never lets
a required check pass unrun."""
required_ids = {check.check_id for check in checks if check.required}
return {
check_id: cached
for check_id, cached in _cached_draft_results(
@@ -145,6 +150,7 @@ def _fresh_draft_results(
cc_pair_id=cc_pair_id,
).items()
if cached.state != DraftCheckStateKind.FAILED
and not (cached.skipped_by_user and check_id in required_ids)
}
@@ -87,6 +87,8 @@ _CONFIG_INDEPENDENT_HASH = "config_independent"
_NEEDS_CONFIG_MESSAGE = "Needs connector settings."
_NEEDS_COMPLETE_CONFIG_MESSAGE = "Needs complete connector settings."
_SKIPPED_BY_USER_LINE = "Skipped by you"
_STOPPED_BY_USER_LINE = "Not run: stopped by user"
class DraftCheckStateKind(str, Enum):
@@ -98,6 +100,8 @@ class DraftCheckStateKind(str, Enum):
# Legacy: INCOMPLETE in runs started before it had its own value.
INDETERMINATE = "indeterminate"
SKIPPED = "skipped"
# A required check the admin skipped: it holds Create until it runs.
STOPPED = "stopped"
# A required field is missing or invalid; the check runs once it is valid.
WAITING = "waiting"
# The access type or the check's ``applies`` excludes it.
@@ -110,6 +114,7 @@ _STATE_BY_STATUS: dict[CapabilityCheckStatus, DraftCheckStateKind] = {
CapabilityCheckStatus.INCOMPLETE: DraftCheckStateKind.INCOMPLETE,
CapabilityCheckStatus.INDETERMINATE: DraftCheckStateKind.INDETERMINATE,
CapabilityCheckStatus.SKIPPED: DraftCheckStateKind.SKIPPED,
CapabilityCheckStatus.STOPPED: DraftCheckStateKind.STOPPED,
}
_STATUS_BY_STATE = {state: status for status, state in _STATE_BY_STATUS.items()}
_UNFINISHED_STATES = frozenset(
@@ -154,6 +159,21 @@ class DraftRerunMode(str, Enum):
ALL = "all"
class DraftCheckActions(BaseModel):
"""The per-check actions of a draft run request: Run Check and Re-run
Check (``check_ids``), Skip (``skip_check_ids``) and Start Now
(``start_check_id``)."""
# Only these checks run, ignoring their cached results. Every other check
# keeps its result from the cache or from the latest run of the draft key,
# and runs only when it has none.
check_ids: list[str] | None = None
# These do not run: an optional check is SKIPPED, a required one STOPPED.
skip_check_ids: list[str] = []
# Runs next: this check and the checks it depends on go first.
start_check_id: str | None = None
class DraftCheckState(BaseModel):
check_id: str
# The row the check shows on (``CapabilityCheck.row_id``). None in runs
@@ -279,8 +299,11 @@ class StoredDraftRun(BaseModel):
# Set for a dry run of an existing pair. Kept out of the snapshot, so the
# create-form endpoint's response does not change.
pair_scope: DraftRunPairScope | None = None
# check_id to its result cache key, for the checks the run task executes.
# check_id to its result cache key, for every check that could run when
# the run started, also one filled from the cache or skipped.
result_cache_keys: dict[str, str]
# Start Now: the run task runs this check, and what it needs, first.
first_check_id: str | None = None
# While RUNNING: the time by which the task writes the run again. A RUNNING
# run read after it has no live task.
lease_expires_at: datetime | None = None
@@ -298,6 +321,8 @@ class CachedDraftResult(BaseModel):
error_detail: CheckErrorDetail | None = None
summary: str | None = None
description: str | None = None
# Set only for an optional check the admin skipped, so creation honors it.
not_run_reason: CheckNotRunReason | None = None
findings: list[CheckFinding] = []
advisories: list[CheckAdvisory] = []
@@ -306,6 +331,10 @@ class CachedDraftResult(BaseModel):
def _known_findings(cls, findings: Any) -> Any:
return parse_known_findings(findings)
@property
def skipped_by_user(self) -> bool:
return self.not_run_reason == CheckNotRunReason.SKIPPED_BY_USER
def decide_draft_check_state(
check: CapabilityCheck[Any], context: CapabilityCheckContext
@@ -556,6 +585,7 @@ def cached_check_result(
error_detail=cached.error_detail,
summary=cached.summary,
description=cached.description,
not_run_reason=cached.not_run_reason,
is_fallback=check.is_fallback,
remediation=check.remediation,
docs_link=check.docs_link,
@@ -597,14 +627,83 @@ def apply_cached_result(
check_state.error_detail = cached.error_detail
check_state.summary = cached.summary
check_state.description = cached.description
check_state.not_run_reason = cached.not_run_reason
check_state.duration_ms = cached.duration_ms
check_state.findings = cached.findings if check_findings_enabled() else []
check_state.advisories = cached.advisories
check_state.from_cache = True
def skip_draft_check(check_state: DraftCheckState) -> None:
"""Marks a check the admin skipped: an optional check is SKIPPED, a
required one STOPPED, so it still holds Create."""
if check_state.required:
check_state.state = DraftCheckStateKind.STOPPED
check_state.not_run_reason = CheckNotRunReason.STOPPED_BY_USER
line = _STOPPED_BY_USER_LINE
else:
check_state.state = DraftCheckStateKind.SKIPPED
check_state.not_run_reason = CheckNotRunReason.SKIPPED_BY_USER
line = _SKIPPED_BY_USER_LINE
check_state.message = line
check_state.summary = line
def cache_user_skip(key: str, check_state: DraftCheckState) -> None:
"""Caches the admin's skip of an optional check, so creation on the same
form does not run it. A stopped required check is never cached: creation
always runs it."""
if check_state.required:
return
cached = CachedDraftResult(
state=check_state.state,
message=check_state.message,
duration_ms=None,
summary=check_state.summary,
not_run_reason=check_state.not_run_reason,
)
get_cache_backend().set(
key, cached.model_dump_json(), ex=DRAFT_RESULT_CACHE_TTL_SECONDS
)
# Results a run keeps from the latest run of its draft key. A check that a
# gate or the admin stopped has no result of its own: it runs again, so it
# sees the new result of its dependency, or the request's skips decide it.
_KEPT_STATES = frozenset(
{
DraftCheckStateKind.PASSED,
DraftCheckStateKind.FAILED,
DraftCheckStateKind.INCOMPLETE,
DraftCheckStateKind.INDETERMINATE,
DraftCheckStateKind.SKIPPED,
}
)
def kept_draft_state(
prior: StoredDraftRun | None, check_state: DraftCheckState, cache_key: str
) -> DraftCheckState | None:
"""The check's result in ``prior``, the latest run of the draft key, for a
run that does not run the check again. None when ``prior`` ran it on
another credential or form, or holds no result of its own for it."""
if prior is None or prior.result_cache_keys.get(check_state.check_id) != cache_key:
return None
for prior_state in prior.snapshot.checks:
if (
prior_state.check_id == check_state.check_id
and prior_state.capability == check_state.capability
and prior_state.state in _KEPT_STATES
and prior_state.not_run_reason in (None, CheckNotRunReason.NOTHING_IN_SCOPE)
):
return prior_state.model_copy(update={"from_cache": True})
return None
def pending_in_run_order(
snapshot: DraftCheckRunSnapshot, checks: Sequence[CapabilityCheck[Any]]
snapshot: DraftCheckRunSnapshot,
checks: Sequence[CapabilityCheck[Any]],
first_check_id: str | None = None,
) -> list[DraftCheckState]:
"""The run's PENDING checks in the order the runner executes them
(``run_order``). ``checks`` are the source's registered checks."""
@@ -615,7 +714,7 @@ def pending_in_run_order(
}
return [
pending[key]
for check in run_order(checks)
for check in run_order(checks, first_check_id)
if (key := (check.check_id, check.capability)) in pending
]
@@ -665,6 +764,12 @@ def load_draft_run(run_id: UUID) -> StoredDraftRun | None:
return StoredDraftRun.model_validate_json(raw) if raw is not None else None
def load_latest_draft_run(user_id: UUID, draft_key: str) -> StoredDraftRun | None:
"""The user's latest run of the draft key, while it has not expired."""
latest = get_cache_backend().get(_latest_run_key(user_id, draft_key))
return load_draft_run(UUID(latest.decode())) if latest is not None else None
def draft_run_start_lock(user_id: UUID, draft_key: str) -> CacheLock:
"""Serializes the save and the latest-run update of concurrent starts for
one draft key, so the run saved last is also the latest."""
@@ -180,11 +180,18 @@ class CapabilityCheckStatus(str, Enum):
# The check needs state (connector instance or config) that was not
# supplied.
SKIPPED = "skipped"
# A required check the admin skipped on a form: it did not run, and it
# holds Create. Only draft runs and pair dry runs hold it.
STOPPED = "stopped"
INCOMPLETE_STATUSES: frozenset[CapabilityCheckStatus] = frozenset(
{CapabilityCheckStatus.INCOMPLETE, CapabilityCheckStatus.INDETERMINATE}
)
# Statuses of a check that did not run. Neither is a pass.
NOT_RUN_STATUSES: frozenset[CapabilityCheckStatus] = frozenset(
{CapabilityCheckStatus.SKIPPED, CapabilityCheckStatus.STOPPED}
)
class CheckErrorKind(str, Enum):
@@ -337,7 +344,8 @@ def check_error_detail(
class CheckNotRunReason(str, Enum):
"""Why a SKIPPED check did not run, when a gate of the run stopped it."""
"""Why a SKIPPED or STOPPED check did not run: a gate of the run, or the
admin."""
# A check it depends on FAILED.
DEPENDENCY_FAILED = "dependency_failed"
@@ -350,6 +358,15 @@ class CheckNotRunReason(str, Enum):
RATE_LIMITED = "rate_limited"
# The check found nothing in scope to test (``NothingToCheckError``).
NOTHING_IN_SCOPE = "nothing_in_scope"
# The admin skipped the check on a form: an optional check is SKIPPED, a
# required one STOPPED.
SKIPPED_BY_USER = "skipped_by_user"
STOPPED_BY_USER = "stopped_by_user"
USER_NOT_RUN_REASONS: frozenset[CheckNotRunReason] = frozenset(
{CheckNotRunReason.SKIPPED_BY_USER, CheckNotRunReason.STOPPED_BY_USER}
)
# Kinds that leave the check INCOMPLETE: they prove nothing about the
@@ -723,10 +740,10 @@ def aggregate_capability_verdict(
Pure function, evaluated in strict precedence order. Required checks gate
the verdict: a required FAILED is FAILED, and a required INCOMPLETE or
SKIPPED leaves the capability's core unverified, so no pass-ish claim may be
made. Non-required failures and incomplete checks only downgrade to
PASSED_WITH_WARNINGS (partial capability); non-required skips do not
downgrade at all. Checks that do not apply are left out. An empty result
SKIPPED (or STOPPED) leaves the capability's core unverified, so no
pass-ish claim may be made. Non-required failures and incomplete checks
only downgrade to PASSED_WITH_WARNINGS (partial capability); non-required
skips do not downgrade at all. Checks that do not apply are left out. An empty result
list aggregates to SKIPPED: a capability is never PASSED on the basis of
having checked nothing.
"""
@@ -744,10 +761,7 @@ def aggregate_capability_verdict(
result.status in INCOMPLETE_STATUSES and result.required for result in results
):
return CapabilityVerdict.INDETERMINATE
if any(
result.status == CapabilityCheckStatus.SKIPPED and result.required
for result in results
):
if any(result.status in NOT_RUN_STATUSES and result.required for result in results):
return CapabilityVerdict.SKIPPED
if any(
result.status == CapabilityCheckStatus.FAILED
@@ -755,7 +769,7 @@ def aggregate_capability_verdict(
for result in results
):
return CapabilityVerdict.PASSED_WITH_WARNINGS
if all(result.status == CapabilityCheckStatus.SKIPPED for result in results):
if all(result.status in NOT_RUN_STATUSES for result in results):
return CapabilityVerdict.SKIPPED
return CapabilityVerdict.PASSED
@@ -412,17 +412,46 @@ def capability_check_run_stale_after(source: DocumentSource) -> timedelta:
return timedelta(seconds=2 * capability_check_run_ceiling_seconds(source))
def run_order(checks: Sequence[CapabilityCheck]) -> list[CapabilityCheck]:
def _with_dependencies(checks: Sequence[CapabilityCheck], check_id: str) -> set[str]:
"""``check_id`` and every check of ``checks`` it depends on, directly or
through other checks."""
depends_on_by_id: dict[str, tuple[str, ...]] = {
check.check_id: check.depends_on for check in checks
}
found: set[str] = set()
stack: list[str] = [check_id]
while stack:
current = stack.pop()
if current in found or current not in depends_on_by_id:
continue
found.add(current)
stack.extend(depends_on_by_id[current])
return found
def run_order(
checks: Sequence[CapabilityCheck], first_check_id: str | None = None
) -> list[CapabilityCheck]:
"""The order the runner executes ``checks``: validation checks before
discovery checks, and every check after the checks of ``checks`` it
depends on; otherwise their given order.
``first_check_id`` (Start Now) moves that check, and the checks it needs,
ahead of the others of its kind. A discovery check still runs after
validation.
Raises:
ValueError: The dependencies among ``checks`` form a cycle.
"""
remaining: list[CapabilityCheck] = [
check for check in checks if not check.discovery
] + [check for check in checks if check.discovery]
first_ids: set[str] = (
_with_dependencies(checks, first_check_id)
if first_check_id is not None
else set()
)
# A stable sort, so the given order holds within each group.
remaining: list[CapabilityCheck] = sorted(
checks, key=lambda check: (check.discovery, check.check_id not in first_ids)
)
present_ids: set[str] = {check.check_id for check in checks}
ordered: list[CapabilityCheck] = []
ordered_ids: set[str] = set()
@@ -458,7 +487,13 @@ _NOT_RUN_LINE_BY_REASON: dict[CheckNotRunReason, str] = {
CheckNotRunReason.DEPENDENCY_INCOMPLETE: "Not run: {name} incomplete",
CheckNotRunReason.DEPENDENCY_SKIPPED: "Not run: {name} did not run",
}
_NOTHING_IN_SCOPE_LINE = "Not run: {name} found nothing to test"
# The line of a check whose dependency did not run for a reason of its own
# (not a gate of the run). The reason stays the dependency's.
_DEPENDENCY_NOT_RUN_LINE_BY_REASON: dict[CheckNotRunReason, str] = {
CheckNotRunReason.NOTHING_IN_SCOPE: "Not run: {name} found nothing to test",
CheckNotRunReason.SKIPPED_BY_USER: "Not run: {name} skipped",
CheckNotRunReason.STOPPED_BY_USER: "Not run: {name} stopped",
}
def _dependency_not_run_outcome(
@@ -474,16 +509,17 @@ def _dependency_not_run_outcome(
dependency = results_by_check_id.get(dependency_id)
if dependency is None or not dependency.applicable:
continue
if dependency.not_run_reason == CheckNotRunReason.NOTHING_IN_SCOPE:
line = _NOTHING_IN_SCOPE_LINE.format(name=dependency.display_name)
return _CheckOutcome(
status=CapabilityCheckStatus.SKIPPED,
message=line,
summary=line,
not_run_reason=CheckNotRunReason.DEPENDENCY_SKIPPED,
blocked_by=dependency_id,
)
if dependency.not_run_reason is not None:
if (dependency_reason := dependency.not_run_reason) is not None:
own_line = _DEPENDENCY_NOT_RUN_LINE_BY_REASON.get(dependency_reason)
if own_line is not None:
line = own_line.format(name=dependency.display_name)
return _CheckOutcome(
status=CapabilityCheckStatus.SKIPPED,
message=line,
summary=line,
not_run_reason=CheckNotRunReason.DEPENDENCY_SKIPPED,
blocked_by=dependency_id,
)
return _CheckOutcome(
status=CapabilityCheckStatus.SKIPPED,
message=dependency.message,
@@ -547,9 +583,10 @@ def run_capability_checks(
timeout_cap_seconds: float | None = None,
prior_results: Sequence[CapabilityCheckResult] = (),
form_run: bool = False,
first_check_id: str | None = None,
) -> list[CapabilityCheckResult]:
"""Runs checks sequentially, in ``run_order``, and maps their outcomes to
statuses.
"""Runs checks sequentially, in ``run_order`` (with ``first_check_id``),
and maps their outcomes to statuses.
Sequential on purpose: source APIs can rate limit aggressively, so
concurrent probes against the same credential are counterproductive. Check
@@ -615,7 +652,7 @@ def run_capability_checks(
# Discovery checks run after validation, and only when it found the
# credential usable: they exist to show what a working credential reaches.
for check in run_order(checks):
for check in run_order(checks, first_check_id):
if check.discovery:
discovery_outcome: _CheckOutcome | None = _discovery_skip_outcome(
[*prior_results, *results], check_findings_enabled()
@@ -808,6 +845,7 @@ def generate_capability_report(
timeout_cap_seconds: float | None = None,
prior_results: Sequence[CapabilityCheckResult] = (),
form_run: bool = False,
first_check_id: str | None = None,
) -> CredentialCapabilityReport:
"""Runs every capability check for a credential and packages a report.
@@ -829,8 +867,8 @@ def generate_capability_report(
connector only when a selected check needs the instance.
``timeout_cap_seconds`` caps every hang guard of the run: each check's and
the connector instantiation's. ``prior_results`` and ``form_run`` are
forwarded to ``run_capability_checks``.
the connector instantiation's. ``prior_results``, ``form_run`` and
``first_check_id`` are forwarded to ``run_capability_checks``.
"""
source = source or credential.source
checks = get_capability_checks(source)
@@ -932,6 +970,7 @@ def generate_capability_report(
timeout_cap_seconds=timeout_cap_seconds,
prior_results=prior_results,
form_run=form_run,
first_check_id=first_check_id,
)
return CredentialCapabilityReport(
credential_id=credential.id,
@@ -19,6 +19,7 @@ from onyx.configs.constants import (
)
from onyx.connectors.capability_checks.draft_runs import (
DRAFT_RUN_QUEUE_EXPIRY_SECONDS,
DraftCheckActions,
DraftCheckPlan,
DraftCheckRunSnapshot,
DraftCheckState,
@@ -28,6 +29,7 @@ from onyx.connectors.capability_checks.draft_runs import (
DraftRunStatus,
StoredDraftRun,
apply_cached_result,
cache_user_skip,
cc_pair_draft_key,
clear_latest_draft_run,
decide_draft_check_state,
@@ -35,9 +37,12 @@ from onyx.connectors.capability_checks.draft_runs import (
draft_run_start_lock,
get_cached_draft_result,
is_superseded,
kept_draft_state,
load_draft_run,
load_latest_draft_run,
save_draft_run,
set_latest_draft_run,
skip_draft_check,
wait_for_dependencies,
)
from onyx.connectors.capability_checks.form_state import (
@@ -312,6 +317,49 @@ def plan_draft_capability_checks(
)
def validate_draft_check_actions(
actions: DraftCheckActions,
checks: list[CapabilityCheck[Any]],
rerun: DraftRerunMode,
) -> None:
"""Rejects actions that name a check the source does not have, or that
contradict each other.
Raises:
OnyxError: INVALID_INPUT, with the reason.
"""
run_ids = set(actions.check_ids or ())
skip_ids = set(actions.skip_check_ids)
named = run_ids | skip_ids
if actions.start_check_id is not None:
named.add(actions.start_check_id)
unknown = named - {check.check_id for check in checks}
if unknown:
raise OnyxError(
OnyxErrorCode.INVALID_INPUT,
f"Unknown capability checks: {', '.join(sorted(unknown))}.",
)
if actions.check_ids is not None and not run_ids:
raise OnyxError(
OnyxErrorCode.INVALID_INPUT, "check_ids must name at least one check."
)
if both := run_ids & skip_ids:
raise OnyxError(
OnyxErrorCode.INVALID_INPUT,
f"Checks to run and to skip at once: {', '.join(sorted(both))}.",
)
if actions.start_check_id in skip_ids:
raise OnyxError(
OnyxErrorCode.INVALID_INPUT,
f"The check to start now is skipped: {actions.start_check_id}.",
)
if actions.check_ids is not None and rerun != DraftRerunMode.NONE:
raise OnyxError(
OnyxErrorCode.INVALID_INPUT,
"check_ids picks the checks to run again; send it without rerun.",
)
def start_draft_capability_check_run(
db_session: Session,
*,
@@ -323,6 +371,7 @@ def start_draft_capability_check_run(
draft_key: str,
form_values: dict[str, Any],
rerun: DraftRerunMode = DraftRerunMode.NONE,
actions: DraftCheckActions | None = None,
pair_scope: DraftRunPairScope | None = None,
) -> DraftCheckRunSnapshot:
"""Decides every check's draft state, fills results from the cache, and
@@ -336,13 +385,19 @@ def start_draft_capability_check_run(
result is run again, so a fix made at the source since the last run shows;
with ALL, every check runs again. Fresh results still go to the cache.
``actions`` are the admin's per-check actions (``DraftCheckActions``);
None runs every check. A skip applies to a check that could run. The skip of an optional check is
cached, so creation on the same form honors it.
Raises:
CapabilityRunEnqueueError: The broker did not accept the task. The run
is stored as FAILED_TO_RUN.
OnyxError: The user may not use the credential with this config, or a
concurrent start of the same draft key held the start lock for too
long.
OnyxError: The actions are not valid, the user may not use the
credential with this config, or a concurrent start of the same
draft key held the start lock for too long.
"""
actions = actions or DraftCheckActions()
validate_draft_check_actions(actions, get_capability_checks(source), rerun)
user_id = user.id
plan = _plan_draft(
config_is_complete=pair_scope is not None,
@@ -370,6 +425,12 @@ def start_draft_capability_check_run(
else None
),
)
run_ids: set[str] | None = (
set(actions.check_ids) if actions.check_ids is not None else None
)
skip_ids = set(actions.skip_check_ids)
# A run of chosen checks keeps the other results of the latest run.
prior = load_latest_draft_run(user_id, draft_key) if run_ids is not None else None
checks: list[DraftCheckState] = []
result_cache_keys: dict[str, str] = {}
for check, check_state in plan.checks:
@@ -387,16 +448,29 @@ def start_draft_capability_check_run(
),
cc_pair_id=pair_scope.cc_pair_id if pair_scope is not None else None,
)
result_cache_keys[check.check_id] = cache_key
if check.check_id in skip_ids:
skip_draft_check(check_state)
cache_user_skip(cache_key, check_state)
continue
if run_ids is not None and check.check_id in run_ids:
continue
cached = (
None if rerun == DraftRerunMode.ALL else get_cached_draft_result(cache_key)
)
# A cached skip holds only while the request still skips the check.
if cached is not None and cached.skipped_by_user:
cached = None
if cached is not None and not (
rerun == DraftRerunMode.FAILED
and cached.state == DraftCheckStateKind.FAILED
):
apply_cached_result(check_state, cached)
else:
result_cache_keys[check.check_id] = cache_key
continue
if run_ids is not None and (
kept := kept_draft_state(prior, check_state, cache_key)
):
checks[-1] = kept
has_pending = any(check.state == DraftCheckStateKind.PENDING for check in checks)
run = StoredDraftRun(
@@ -413,6 +487,7 @@ def start_draft_capability_check_run(
checks=checks,
),
result_cache_keys=result_cache_keys,
first_check_id=actions.start_check_id,
pair_scope=pair_scope,
)
if has_pending:
@@ -470,6 +545,7 @@ def start_cc_pair_draft_check_run(
access_type: AccessType,
connector_specific_config: dict[str, Any],
rerun: DraftRerunMode = DraftRerunMode.NONE,
actions: DraftCheckActions | None = None,
) -> DraftCheckRunSnapshot:
"""Starts a dry run of the capability checks for a proposed state of an
existing pair. Like a create-form draft run, it never writes a stored
@@ -507,6 +583,7 @@ def start_cc_pair_draft_check_run(
draft_key=cc_pair_draft_key(cc_pair_id),
form_values=connector_specific_config,
rerun=rerun,
actions=actions,
pair_scope=DraftRunPairScope(
cc_pair_id=cc_pair_id,
connector_id=connector.id,
@@ -338,6 +338,7 @@ def start_cc_pair_edit_checks(
access_type=request.access_type,
connector_specific_config=request.connector_specific_config,
rerun=request.rerun,
actions=request,
)
except CapabilityRunEnqueueError as e:
raise OnyxError(
@@ -6,7 +6,10 @@ from uuid import UUID
from pydantic import BaseModel, ConfigDict
from onyx.connectors.capability_checks.draft_runs import DraftRerunMode
from onyx.connectors.capability_checks.draft_runs import (
DraftCheckActions,
DraftRerunMode,
)
from onyx.connectors.capability_checks.findings_setting import (
hide_findings_if_disabled,
)
@@ -89,7 +92,7 @@ class ConnectorEditApplyResponse(BaseModel):
reactivated: bool
class ConnectorEditChecksRequest(BaseModel):
class ConnectorEditChecksRequest(DraftCheckActions):
model_config = ConfigDict(extra="forbid")
connector_specific_config: dict[str, Any]
@@ -20,6 +20,7 @@ from onyx.configs.constants import (
DocumentSource,
)
from onyx.connectors.capability_checks.draft_runs import (
DraftCheckActions,
DraftCheckPlan,
DraftCheckRunSnapshot,
DraftRerunMode,
@@ -541,8 +542,9 @@ def plan_draft_checks(
)
class DraftCheckRunRequest(BaseModel):
"""Body of the draft run endpoint: an unsaved connector form."""
class DraftCheckRunRequest(DraftCheckActions):
"""Body of the draft run endpoint: an unsaved connector form, and the
admin's per-check actions."""
model_config = ConfigDict(extra="forbid")
@@ -581,7 +583,8 @@ def start_draft_check_run(
Checks that cannot run yet resolve at once (waiting, not applicable), and
fresh cached results fill others; one task runs the rest. The caller polls
the GET with the returned ``run_id``. Never writes a stored report.
the GET with the returned ``run_id``. Never writes a stored report. The
request's per-check actions run, re-run, skip or start one check first.
"""
mapping = CONNECTOR_CLASS_MAP.get(request.source)
if mapping is None:
@@ -607,6 +610,7 @@ def start_draft_check_run(
draft_key=request.draft_key,
form_values=request.form_state,
rerun=request.rerun,
actions=request,
)
except CapabilityRunEnqueueError as e:
raise OnyxError(
@@ -0,0 +1,457 @@
"""Per-check actions of draft runs: Run Check and Re-run Check
(``check_ids``), Skip (``skip_check_ids``) and Start Now
(``start_check_id``). Covers the request rules, the stopped and skipped
states, the results a run keeps, the run order, the verdict, and creation's
reuse of the admin's skips."""
from contextlib import nullcontext
from datetime import datetime, timezone
from typing import Any
from unittest.mock import MagicMock
from uuid import uuid4
import pytest
from onyx.background.celery.tasks.capability_checks import (
tasks as capability_check_tasks,
)
from onyx.configs.constants import DocumentSource
from onyx.connectors.capability_checks import creation, draft_runs, runner
from onyx.connectors.capability_checks.draft_runs import (
DraftCheckActions,
DraftCheckRunSnapshot,
DraftCheckState,
DraftCheckStateKind,
DraftRerunMode,
DraftRunStatus,
get_cached_draft_result,
load_draft_run,
)
from onyx.connectors.capability_checks.models import (
CapabilityCheck,
CapabilityCheckContext,
CapabilityCheckResult,
CapabilityCheckStatus,
CapabilityVerdict,
CheckNotRunReason,
CheckOutput,
CredentialCapability,
aggregate_capability_verdict,
)
from onyx.connectors.capability_checks.runner import run_order
from onyx.connectors.exceptions import UnexpectedValidationError
from onyx.connectors.slack.config import SlackConnectorConfig
from onyx.db.enums import AccessType
from onyx.db.models import Credential, User
from onyx.error_handling.error_codes import OnyxErrorCode
from onyx.error_handling.exceptions import OnyxError
from onyx.server.documents import capability_check_runs
from onyx.server.documents.capability_check_runs import (
start_draft_capability_check_run,
)
from tests.unit.fakes import FakeCache
_SIGN_IN = "sign_in"
_LIST = "list_items"
_READ = "read_items"
_EXTRA = "read_extra"
_FIND = "find_options"
_DRAFT_KEY = "form-1"
class _Check(CapabilityCheck):
def __init__(
self,
check_id: str,
runs: list[str],
*,
required: bool = True,
depends_on: tuple[str, ...] = (),
discovery: bool = False,
error: Exception | None = None,
) -> None:
super().__init__(
capability=CredentialCapability.INDEXING,
check_id=check_id,
display_name=f"Run {check_id}",
required=required,
requires_connector_instance=False,
depends_on=depends_on,
discovery=discovery,
)
self._runs = runs
self.error = error
def run(self, context: CapabilityCheckContext) -> CheckOutput | None: # noqa: ARG002
self._runs.append(self.check_id)
if self.error is not None:
raise self.error
return None
class _Harness:
"""A draft run of a Slack form with fake checks, on a fake cache, with
the run task called in process."""
def __init__(self, monkeypatch: pytest.MonkeyPatch) -> None:
self.runs: list[str] = []
self.checks: list[CapabilityCheck[Any]] = [
_Check(_SIGN_IN, self.runs),
_Check(_LIST, self.runs, depends_on=(_SIGN_IN,)),
_Check(_READ, self.runs, depends_on=(_LIST,)),
_Check(_EXTRA, self.runs, required=False, depends_on=(_SIGN_IN,)),
_Check(_FIND, self.runs, discovery=True),
]
self.cache = FakeCache()
self.send_task = MagicMock()
self.user = MagicMock(spec=User)
self.user.id = uuid4()
self.credential = Credential(
id=7,
source=DocumentSource.SLACK,
credential_json=None,
time_updated=datetime(2026, 10, 1, tzinfo=timezone.utc),
)
monkeypatch.setattr(draft_runs, "get_cache_backend", lambda: self.cache)
monkeypatch.setattr(
capability_check_runs.client_app, "send_task", self.send_task
)
monkeypatch.setattr(
capability_check_runs, "assert_user_can_use_credential", MagicMock()
)
for module in (capability_check_runs, runner, capability_check_tasks):
monkeypatch.setattr(
module, "get_capability_checks", lambda _source: self.checks
)
monkeypatch.setattr(creation, "get_capability_checks", lambda _: self.checks)
monkeypatch.setattr(runner, "get_source_operations_class", lambda _: None)
monkeypatch.setattr(
capability_check_tasks,
"get_session_with_current_tenant",
lambda: nullcontext(MagicMock()),
)
monkeypatch.setattr(
capability_check_tasks,
"fetch_credential_by_id",
lambda _id, _session: self.credential,
)
def check(self, check_id: str) -> _Check:
found = next(check for check in self.checks if check.check_id == check_id)
assert isinstance(found, _Check)
return found
def start(
self,
actions: DraftCheckActions | None = None,
rerun: DraftRerunMode = DraftRerunMode.NONE,
) -> DraftCheckRunSnapshot:
self.send_task.reset_mock()
return start_draft_capability_check_run(
MagicMock(),
user=self.user,
credential=self.credential,
source=DocumentSource.SLACK,
config_class=SlackConnectorConfig,
access_type=AccessType.PUBLIC,
draft_key=_DRAFT_KEY,
form_values={},
rerun=rerun,
actions=actions,
)
def run(
self,
actions: DraftCheckActions | None = None,
rerun: DraftRerunMode = DraftRerunMode.NONE,
) -> DraftCheckRunSnapshot:
"""Starts a run, runs its task, and returns the finished snapshot."""
self.runs.clear()
snapshot = self.start(actions, rerun)
if self.send_task.called:
capability_check_tasks.run_draft_capability_checks_task(
**self.send_task.call_args.kwargs["kwargs"]
)
run = load_draft_run(snapshot.run_id)
assert run is not None
return run.snapshot
def _state(snapshot: DraftCheckRunSnapshot, check_id: str) -> DraftCheckState:
return next(check for check in snapshot.checks if check.check_id == check_id)
@pytest.fixture
def harness(monkeypatch: pytest.MonkeyPatch) -> _Harness:
return _Harness(monkeypatch)
def test_plain_run_runs_every_check_in_run_order(harness: _Harness) -> None:
snapshot = harness.run()
assert snapshot.status == DraftRunStatus.COMPLETED
assert harness.runs == [_SIGN_IN, _LIST, _READ, _EXTRA, _FIND]
assert {check.state for check in snapshot.checks} == {DraftCheckStateKind.PASSED}
def test_skipped_required_check_is_stopped_and_its_dependents_do_not_run(
harness: _Harness,
) -> None:
snapshot = harness.run(DraftCheckActions(skip_check_ids=[_LIST]))
stopped = _state(snapshot, _LIST)
assert stopped.state == DraftCheckStateKind.STOPPED
assert stopped.not_run_reason == CheckNotRunReason.STOPPED_BY_USER
assert stopped.summary == "Not run: stopped by user"
dependent = _state(snapshot, _READ)
assert dependent.state == DraftCheckStateKind.SKIPPED
assert dependent.not_run_reason == CheckNotRunReason.DEPENDENCY_SKIPPED
assert dependent.blocked_by == _LIST
assert dependent.summary == f"Not run: Run {_LIST} stopped"
assert harness.runs == [_SIGN_IN, _EXTRA, _FIND]
# A stopped required check is never cached, so creation runs it.
run = load_draft_run(snapshot.run_id)
assert run is not None
assert get_cached_draft_result(run.result_cache_keys[_LIST]) is None
def test_skipped_optional_check_is_skipped_by_user_and_cached(
harness: _Harness,
) -> None:
snapshot = harness.run(DraftCheckActions(skip_check_ids=[_EXTRA]))
skipped = _state(snapshot, _EXTRA)
assert skipped.state == DraftCheckStateKind.SKIPPED
assert skipped.not_run_reason == CheckNotRunReason.SKIPPED_BY_USER
assert skipped.summary == "Skipped by you"
assert _EXTRA not in harness.runs
run = load_draft_run(snapshot.run_id)
assert run is not None
cached = get_cached_draft_result(run.result_cache_keys[_EXTRA])
assert cached is not None and cached.skipped_by_user
def test_a_cached_skip_holds_only_while_the_request_skips_the_check(
harness: _Harness,
) -> None:
harness.run(DraftCheckActions(skip_check_ids=[_EXTRA]))
snapshot = harness.run()
assert _EXTRA in harness.runs
assert _state(snapshot, _EXTRA).state == DraftCheckStateKind.PASSED
def test_check_ids_reruns_only_the_chosen_checks(harness: _Harness) -> None:
harness.check(_EXTRA).error = UnexpectedValidationError("flaky")
first = harness.run()
assert _state(first, _EXTRA).state == DraftCheckStateKind.INCOMPLETE
harness.check(_EXTRA).error = None
snapshot = harness.run(DraftCheckActions(check_ids=[_SIGN_IN]))
# Re-run ignores the cached pass of the chosen check.
assert harness.runs == [_SIGN_IN]
assert _state(snapshot, _SIGN_IN).from_cache is False
# A result the cache does not hold is kept from the latest run.
kept = _state(snapshot, _EXTRA)
assert kept.state == DraftCheckStateKind.INCOMPLETE
assert kept.from_cache is True
assert _state(snapshot, _READ).state == DraftCheckStateKind.PASSED
def test_run_check_on_a_stopped_check_runs_it_and_its_dependents(
harness: _Harness,
) -> None:
harness.run(DraftCheckActions(skip_check_ids=[_LIST]))
snapshot = harness.run(DraftCheckActions(check_ids=[_LIST]))
# The dependent's "Not run" was no result of its own, so it runs too.
assert harness.runs == [_LIST, _READ]
assert _state(snapshot, _LIST).state == DraftCheckStateKind.PASSED
assert _state(snapshot, _READ).state == DraftCheckStateKind.PASSED
def test_start_now_runs_the_check_and_what_it_needs_first(
harness: _Harness,
) -> None:
snapshot = harness.run(DraftCheckActions(start_check_id=_EXTRA))
assert harness.runs == [_SIGN_IN, _EXTRA, _LIST, _READ, _FIND]
assert snapshot.status == DraftRunStatus.COMPLETED
def test_start_now_on_a_check_after_a_skip_does_not_run_it(
harness: _Harness,
) -> None:
snapshot = harness.run(
DraftCheckActions(skip_check_ids=[_LIST], start_check_id=_READ)
)
assert harness.runs == [_SIGN_IN, _EXTRA, _FIND]
assert _state(snapshot, _READ).blocked_by == _LIST
@pytest.mark.parametrize(
"actions,rerun",
[
(DraftCheckActions(check_ids=["nope"]), DraftRerunMode.NONE),
(DraftCheckActions(skip_check_ids=["nope"]), DraftRerunMode.NONE),
(DraftCheckActions(start_check_id="nope"), DraftRerunMode.NONE),
(DraftCheckActions(check_ids=[]), DraftRerunMode.NONE),
(
DraftCheckActions(check_ids=[_LIST], skip_check_ids=[_LIST]),
DraftRerunMode.NONE,
),
(
DraftCheckActions(skip_check_ids=[_LIST], start_check_id=_LIST),
DraftRerunMode.NONE,
),
(DraftCheckActions(check_ids=[_LIST]), DraftRerunMode.FAILED),
],
)
def test_invalid_actions_are_rejected_before_anything_starts(
harness: _Harness, actions: DraftCheckActions, rerun: DraftRerunMode
) -> None:
with pytest.raises(OnyxError) as raised:
harness.start(actions, rerun)
assert raised.value.error_code == OnyxErrorCode.INVALID_INPUT
assert raised.value.status_code == 400
assert not harness.send_task.called
assert harness.cache.store == {}
def test_run_order_start_now_keeps_discovery_after_validation() -> None:
runs: list[str] = []
checks: list[CapabilityCheck] = [
_Check(_SIGN_IN, runs),
_Check(_LIST, runs),
_Check(_FIND, runs, discovery=True, depends_on=(_SIGN_IN,)),
]
ordered = [check.check_id for check in run_order(checks, _FIND)]
assert ordered == [_SIGN_IN, _LIST, _FIND]
def _result(
status: CapabilityCheckStatus,
*,
required: bool,
reason: CheckNotRunReason | None = None,
) -> CapabilityCheckResult:
return CapabilityCheckResult(
capability=CredentialCapability.INDEXING,
check_id=str(uuid4()),
display_name="Check",
required=required,
status=status,
not_run_reason=reason,
)
def test_a_stopped_required_check_is_no_pass() -> None:
verdict = aggregate_capability_verdict(
True,
[
_result(CapabilityCheckStatus.PASSED, required=True),
_result(
CapabilityCheckStatus.STOPPED,
required=True,
reason=CheckNotRunReason.STOPPED_BY_USER,
),
],
)
assert verdict == CapabilityVerdict.SKIPPED
def test_a_skipped_optional_check_does_not_downgrade_the_verdict() -> None:
verdict = aggregate_capability_verdict(
True,
[
_result(CapabilityCheckStatus.PASSED, required=True),
_result(
CapabilityCheckStatus.SKIPPED,
required=False,
reason=CheckNotRunReason.SKIPPED_BY_USER,
),
],
)
assert verdict == CapabilityVerdict.PASSED
def test_creation_reuses_an_optional_skip_and_runs_a_required_one(
harness: _Harness, monkeypatch: pytest.MonkeyPatch
) -> None:
snapshot = harness.run(DraftCheckActions(skip_check_ids=[_EXTRA, _LIST]))
run = load_draft_run(snapshot.run_id)
assert run is not None
# A skip cached before the check became required must not let it pass.
draft_runs.cache_user_skip(
run.result_cache_keys[_READ],
_state(snapshot, _EXTRA).model_copy(update={"check_id": _READ}),
)
ran: list[str] = []
def run_within_budget(
check_ids: list[str], **_kwargs: Any
) -> tuple[list[CapabilityCheckResult], frozenset[str]]:
ran.extend(check_ids)
return [
_result(CapabilityCheckStatus.PASSED, required=True).model_copy(
update={"check_id": check_id}
)
for check_id in check_ids
], frozenset()
monkeypatch.setattr(creation, "_run_checks_within_budget", run_within_budget)
named_run = creation.run_named_checks_within_budget(
connector_id=None,
cc_pair_id=None,
source=DocumentSource.SLACK,
input_type=None,
connector_specific_config={},
credential=harness.credential,
access_type=AccessType.PUBLIC,
)
assert _EXTRA not in ran
assert _LIST in ran and _READ in ran
extra = next(r for r in named_run.finished_results if r.check_id == _EXTRA)
assert extra.status == CapabilityCheckStatus.SKIPPED
assert extra.not_run_reason == CheckNotRunReason.SKIPPED_BY_USER
assert not named_run.failed_required_results
def test_a_chain_after_a_skipped_check_names_the_skipped_check() -> None:
runs: list[str] = []
checks: list[CapabilityCheck] = [
_Check(_READ, runs, depends_on=(_LIST,)),
_Check(_EXTRA, runs, depends_on=(_READ,)),
]
skipped = _result(
CapabilityCheckStatus.SKIPPED,
required=False,
reason=CheckNotRunReason.SKIPPED_BY_USER,
).model_copy(update={"check_id": _LIST, "display_name": "List items"})
results = runner.run_capability_checks(
checks,
CapabilityCheckContext(source=DocumentSource.SLACK, credential_json={}),
prior_results=[skipped],
)
assert runs == []
assert [result.summary for result in results] == [
"Not run: List items skipped",
"Not run: List items skipped",
]
assert {result.blocked_by for result in results} == {_LIST}
assert {result.not_run_reason for result in results} == {
CheckNotRunReason.DEPENDENCY_SKIPPED
}