preflight gating

This commit is contained in:
2026-05-07 15:52:13 +02:00
parent 4f8d8a1f52
commit 18299b03aa
7 changed files with 157 additions and 5 deletions

View File

@@ -310,6 +310,9 @@ Resolves an assessment profile into an executable run plan:
- isolation and timeout policy,
- artifact retention policy.
At execution time, a failing preflight blocks downstream check groups for the
same extension so expensive or misleading harness steps are not invoked.
### Runner Bridge
Executes or coordinates extension checks.

View File

@@ -182,6 +182,11 @@ package `artifact_manifest`.
If a Python runner raises an exception, the core converts that failure into
`infrastructure_error` evidence so the assessment package remains complete.
Preflight runners are gates. If an extension preflight returns `fail`, `blocked`,
or `infrastructure_error`, downstream check groups for that extension are not
executed; they receive `blocked` evidence with `blocked_reason:
preflight_failed`.
## Result Statuses
Initial statuses:

View File

@@ -109,6 +109,8 @@ Progress:
and parseable JSON repository metadata through the guide-board runner bridge.
- The preflight runner preserves raw response metadata and body artifacts for
assessment-package fingerprinting.
- Failed CMIS preflight now blocks downstream OpenCMIS TCK groups instead of
invoking the Java/Maven wrapper against an invalid target.
- Capability flag normalization remains to be expanded after a live target sample
is captured.

View File

@@ -17,7 +17,7 @@
"tck_invocation_not_configured"
],
"expected": true,
"reason": "The local bootstrap can plan CMIS TCK groups before Java/Maven and the final Apache Chemistry invocation are configured.",
"reason": "When preflight passes, the local bootstrap can plan CMIS TCK groups before Java/Maven and the final Apache Chemistry invocation are configured.",
"status": "active"
}
]

View File

@@ -27,10 +27,7 @@ def run_assessment(
run_dir = output_dir or root / "runs" / run_id
created_at = _now()
evidence = [
_evidence_for_step(root, run_dir, run_id, plan, step)
for step in plan["ordered_steps"]
]
evidence = _execute_steps(root, run_dir, run_id, plan)
for item in evidence:
assert_valid(item, "evidence-item")
@@ -83,6 +80,63 @@ def run_assessment(
}
def _execute_steps(
root: Path,
run_dir: Path,
run_id: str,
plan: dict[str, Any],
) -> list[dict[str, Any]]:
evidence: list[dict[str, Any]] = []
preflight_blocks: dict[str, dict[str, Any]] = {}
for step in plan["ordered_steps"]:
extension_id = step["extension_id"]
if step["kind"] == "check_group" and extension_id in preflight_blocks:
item = _blocked_by_preflight_evidence(run_id, plan, step, preflight_blocks[extension_id])
else:
item = _evidence_for_step(root, run_dir, run_id, plan, step)
evidence.append(item)
if step["kind"] == "preflight" and _blocks_downstream(item):
preflight_blocks[extension_id] = item
return evidence
def _blocked_by_preflight_evidence(
run_id: str,
plan: dict[str, Any],
step: dict[str, Any],
preflight: dict[str, Any],
) -> dict[str, Any]:
now = _now()
runner_ref = step.get("runner_ref")
return {
"id": f"evidence:{step['id']}",
"run_id": run_id,
"extension_id": step["extension_id"],
"check_id": step["id"],
"subject_ref": plan["target_profile_snapshot"]["id"],
"result": "blocked",
"observations": [
"Check group was not executed because extension preflight did not pass."
],
"facts": {
"step_kind": step["kind"],
"runner_ref": runner_ref,
"blocked_reason": "preflight_failed",
"preflight_evidence_ref": preflight["id"],
"preflight_result": preflight["result"],
},
"requirement_refs": _requirement_refs(plan, step),
"artifact_refs": [],
"started_at": now,
"completed_at": now,
}
def _blocks_downstream(evidence: dict[str, Any]) -> bool:
return evidence["result"] in {"fail", "blocked", "infrastructure_error"}
def _evidence_for_step(
root: Path,
run_dir: Path,
@@ -169,6 +223,7 @@ def _expected_for_item(item: dict[str, Any]) -> bool:
return blocked_reason in {
"missing_command",
"missing_dependency",
"preflight_failed",
"tck_invocation_not_configured",
}
@@ -179,6 +234,8 @@ def _remediation_for_item(item: dict[str, Any]) -> str:
blocked_reason = item.get("facts", {}).get("blocked_reason")
if blocked_reason == "missing_dependency":
return "Install the missing runner dependencies and rerun the assessment."
if blocked_reason == "preflight_failed":
return "Fix the preflight failure and rerun downstream checks."
if blocked_reason == "tck_invocation_not_configured":
return "Configure the final harness invocation, group mapping, and raw artifact capture."
return "Implement or configure the declared extension runner."

View File

@@ -344,6 +344,89 @@ class CoreArchitectureTests(unittest.TestCase):
thread.join(timeout=5)
server.server_close()
def test_preflight_failure_blocks_downstream_checks(self) -> None:
with TemporaryDirectory() as temporary_directory:
temp_root = Path(temporary_directory)
target_path = temp_root / "target.json"
assessment_path = temp_root / "assessment.json"
target_path.write_text(
json.dumps(
{
"id": "local-cmis-preflight-failure",
"subject_type": "cmis-browser-binding-endpoint",
"subject_name": "Local CMIS Preflight Failure",
"environment": "test",
"scope": ["preflight", "tck-wrapper"],
"endpoints": [
{
"id": "browser-binding",
"url": "http://127.0.0.1:9/cmis/browser",
"binding": "cmis-browser",
}
],
"artifacts": [],
"credentials_ref": None,
"declared_capabilities": ["cmis.repository-info"],
"known_gaps": [],
}
),
encoding="utf-8",
)
assessment_path.write_text(
json.dumps(
{
"id": "local-cmis-preflight-gate",
"framework_refs": ["cmis.browser-binding.compatibility.v1"],
"extension_refs": ["open-cmis-tck"],
"target_profile_ref": "local-cmis-preflight-failure",
"selected_check_groups": {
"open-cmis-tck": ["repository-type"]
},
"expectations_ref": None,
"waivers_ref": None,
"output_policy": {
"report_formats": ["json", "markdown"],
"artifact_retention": "summary-only",
},
"retention_policy": {
"summary_days": 365,
"raw_artifact_days": 0,
},
"runtime_policy": {
"offline": False,
"timeout_seconds": 1,
},
}
),
encoding="utf-8",
)
result = run_assessment(
ROOT,
target_path,
assessment_path,
temp_root / "run",
)
run_dir = Path(result["run_dir"])
evidence = json.loads(
(run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8")
)["evidence"]
findings = json.loads(
(run_dir / "normalized" / "findings.json").read_text(encoding="utf-8")
)["findings"]
self.assertEqual(result["status"], "infrastructure_error")
self.assertEqual(evidence[0]["result"], "infrastructure_error")
self.assertEqual(evidence[1]["result"], "blocked")
self.assertEqual(evidence[1]["facts"]["blocked_reason"], "preflight_failed")
self.assertEqual(
evidence[1]["facts"]["preflight_evidence_ref"],
evidence[0]["id"],
)
self.assertFalse((run_dir / "artifacts" / "runner-contexts").exists())
self.assertEqual(findings[1]["classification"], "preflight_failed")
self.assertTrue(findings[1]["expected"])
class _CmisHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None:

View File

@@ -190,6 +190,8 @@ Acceptance:
runner-emitted raw artifacts.
- The baseline executor applies expectation and waiver policy refs from
assessment profiles and reports policy summary counts.
- Failed extension preflight evidence gates downstream check groups so later
runners are not invoked against an invalid target posture.
## D1.7 - Extension SDK Skeleton