Run pull-lab boot and LTP tests on the supplied kernel

dockerfiles/aws/test.dockerfile

@@ -15,6 +15,7 @@ WORKDIR /app
 COPY src/kernel_ci_cloud_labs/__init__.py        /app/kernel_ci_cloud_labs/__init__.py
 COPY src/kernel_ci_cloud_labs/core/__init__.py   /app/kernel_ci_cloud_labs/core/__init__.py
 COPY src/kernel_ci_cloud_labs/core/log_scrub.py  /app/kernel_ci_cloud_labs/core/log_scrub.py
+COPY src/kernel_ci_cloud_labs/kernel_test_results.py /app/kernel_ci_cloud_labs/kernel_test_results.py
 COPY src/kernel_ci_cloud_labs/launch_vm.py       /app/launch_vm.py
 COPY src/kernel_ci_cloud_labs/debug_aws_setup.py /app/debug_aws_setup.py
 

src/kernel_ci_cloud_labs/auth/aws_auth.py

@@ -289,7 +289,8 @@ def _make_client(service):
                 logger.debug("Execution role ARN: %s", task_config.get("execution_role_arn"))
 
                 task_manager = AWSTaskDefinitionManager(ecs_client, {})
-                task_manager.ensure_exists(task_config["family"], task_config)
+                task_arn, _ = task_manager.ensure_exists(task_config["family"], task_config)
+                self.config["ecs"]["task_definition"]["arn"] = task_arn
 
                 # Store network manager for later use
                 from kernel_ci_cloud_labs.auth.aws_network_manager import (
@@ -401,12 +402,7 @@ def _build_and_push_docker_image(self, repo_uri: str) -> str:
             try:
                 ecr_client.describe_images(repositoryName=repo_name, imageIds=[{"imageTag": image_tag}])
                 logger.info("✓ Image already exists: %s", image_uri)
-
-                # Update task definition config with existing image
-                if self.config.get("ecs", {}).get("task_definition"):
-                    self.config["ecs"]["task_definition"]["image"] = image_uri
-
-                return image_uri
+                return self._use_image_digest(ecr_client, repo_name, repo_uri, image_tag)
             except ecr_client.exceptions.ImageNotFoundException:
                 logger.info("Image not found, building from %s", dockerfile)
         else:
@@ -450,9 +446,20 @@ def _build_and_push_docker_image(self, repo_uri: str) -> str:
         subprocess.run(["docker", "push", image_uri], check=True)
 
         logger.info("✓ Image pushed: %s", image_uri)
+        return self._use_image_digest(ecr_client, repo_name, repo_uri, image_tag)
+
+    def _use_image_digest(self, ecr_client, repo_name: str, repo_uri: str, image_tag: str) -> str:
+        """Point the task definition at the tag's current immutable digest.
 
-        # Update task definition config with new image
+        A mutable tag such as ``latest`` would let ECS start whatever image
+        the tag names at launch time, and the task-definition drift check
+        could never see the runner change. A digest reference changes with
+        the image, so a new runner registers a new revision.
+        """
+        details = ecr_client.describe_images(repositoryName=repo_name, imageIds=[{"imageTag": image_tag}])
+        digest = details["imageDetails"][0]["imageDigest"]
+        image_uri = f"{repo_uri}@{digest}"
+        logger.info("Runner image pinned: %s:%s -> %s", repo_uri, image_tag, image_uri)
         if self.config.get("ecs", {}).get("task_definition"):
             self.config["ecs"]["task_definition"]["image"] = image_uri
-
         return image_uri

src/kernel_ci_cloud_labs/auth/aws_task_definition_manager.py

@@ -6,13 +6,60 @@
 
 
 from typing import Any, Dict
+from botocore.exceptions import ClientError
 
 from kernel_ci_cloud_labs.core.base_resource_manager import BaseResourceManager
 
 
+def _contains(actual, wanted):
+    """True when *actual* holds everything in *wanted*.
+
+    describe_task_definition normalises what was registered: it adds defaults
+    such as ``logConfiguration.secretOptions: []`` and empty container lists.
+    Only the fields this manager sets are compared, recursively, so those
+    additions do not register a new revision on every run.
+    """
+    if isinstance(wanted, dict):
+        return isinstance(actual, dict) and all(_contains(actual.get(key), value) for key, value in wanted.items())
+    if isinstance(wanted, list):
+        return (
+            isinstance(actual, list)
+            and len(actual) == len(wanted)
+            and all(_contains(item, value) for item, value in zip(actual, wanted))
+        )
+    return actual == wanted
+
+
 class AWSTaskDefinitionManager(BaseResourceManager):
     """Manages ECS task definitions"""
 
+    def ensure_exists(self, resource_name, resource_config=None, force_recreate=False):
+        """Register a revision when the executable contract changes.
+
+        Finding an ACTIVE family is insufficient: it may still reference the
+        previous runner image. Existing revisions/tasks are never deregistered.
+        """
+        config = resource_config or self.config.get(resource_name, {})
+        try:
+            current = self.client.describe_task_definition(taskDefinition=resource_name)["taskDefinition"]
+        except ClientError as error:
+            if error.response.get("Error", {}).get("Code") != "ClientException":
+                raise
+            return self.create(resource_name, config), True
+        if current.get("status") == "ACTIVE" and not force_recreate:
+            if not config:
+                return current["taskDefinitionArn"], False
+            desired = self._definition(resource_name, config)
+            fields = ("cpu", "memory", "networkMode", "executionRoleArn", "taskRoleArn")
+            containers = current.get("containerDefinitions", [])
+            expected = desired["containerDefinitions"]
+            matches = all(current.get(key) == desired.get(key) for key in fields)
+            matches = matches and len(containers) == len(expected)
+            matches = matches and all(_contains(actual, wanted) for actual, wanted in zip(containers, expected))
+            if matches:
+                return current["taskDefinitionArn"], False
+        return self.create(resource_name, config), True
+
     def check_exists(self, resource_name: str) -> bool:
         """Check if task definition exists"""
         try:
@@ -25,6 +72,12 @@ def check_exists(self, resource_name: str) -> bool:
     def create(self, resource_name: str, resource_config: Dict[str, Any]) -> str:
         """Create task definition with CloudWatch logs"""
 
+        response = self.client.register_task_definition(**self._definition(resource_name, resource_config))
+        return response["taskDefinition"]["taskDefinitionArn"]
+
+    @staticmethod
+    def _definition(resource_name, resource_config):
+
         # Build container definition
         container_def = {
             "name": resource_config.get("container_name", "app"),
@@ -50,7 +103,7 @@ def create(self, resource_name: str, resource_config: Dict[str, Any]) -> str:
         }
 
         # Register task definition
-        response = self.client.register_task_definition(
+        return dict(
             family=resource_name,
             networkMode="awsvpc",
             requiresCompatibilities=["FARGATE"],
@@ -61,8 +114,6 @@ def create(self, resource_name: str, resource_config: Dict[str, Any]) -> str:
             containerDefinitions=[container_def],
         )
 
-        return response["taskDefinition"]["taskDefinitionArn"]
-
     def get_identifier(self, resource_name: str) -> str:
         """Get task definition ARN"""
         response = self.client.describe_task_definition(taskDefinition=resource_name)

src/kernel_ci_cloud_labs/core/pipeline.py

@@ -13,13 +13,41 @@
 from kernel_ci_cloud_labs.auth.aws_cloudwatch_manager import AWSCloudWatchManager
 from kernel_ci_cloud_labs.core.artifacts import collect_run_artifacts
 from kernel_ci_cloud_labs.core.logging_config import create_run_directory, get_logger
+from kernel_ci_cloud_labs.kernel_test_results import EVIDENCE_MARKER, requires_evidence
 
 logger = get_logger(__name__)
 
 
-def parse_vm_logs(vms_dir):
+def parse_evidence_verdicts(container_log):
+    """Read per-instance native-test verdicts emitted by launch_vm.py.
+
+    Returns ``{instance_id: verdict}``; a missing or unreadable log yields an
+    empty dict, which fails every evidence-requiring instance closed.
+    """
+    verdicts = {}
+    try:
+        content = Path(container_log).read_text(encoding="utf-8")
+    except OSError:
+        return verdicts
+    for line in content.splitlines():
+        if EVIDENCE_MARKER not in line:
+            continue
+        try:
+            verdict = json.loads(line.split(EVIDENCE_MARKER, 1)[1])
+        except ValueError:
+            continue
+        if isinstance(verdict, dict) and verdict.get("instance_id"):
+            verdicts[verdict["instance_id"]] = verdict
+    return verdicts
+
+
+def parse_vm_logs(vms_dir, evidence_tests=(), verdicts=None):
     """Parse VM log files and extract test statistics.
 
+    Instances of a test in *evidence_tests* pass only when *verdicts* (see
+    ``parse_evidence_verdicts``) holds an accepted verdict for that instance;
+    their entry also carries ``boot_status`` from the same verdict.
+
     Expects one ``vms/<instance_id>.log`` per VM (written from CloudWatch
     streams). Boot-console logs written alongside as
     ``vms/<instance_id>-console.log`` by ``core.artifacts.collect_run_artifacts``
@@ -69,7 +97,14 @@ def parse_vm_logs(vms_dir):
                         break
 
             # Check success/failure
-            if "Test execution completed: SUCCESS" in content:
+            passed = "Test execution completed: SUCCESS" in content
+            entry = {"test": test_name, "instance_id": instance_id}
+            if test_name in evidence_tests:
+                verdict = (verdicts or {}).get(instance_id) or {}
+                passed = passed and verdict.get("result") == "PASS"
+                if verdict.get("boot_requested"):
+                    entry["boot_status"] = "PASS" if verdict.get("boot") == "PASS" else "FAIL"
+            if passed:
                 stats["successful"] += 1
                 status = "PASS"
             else:
@@ -79,9 +114,8 @@ def parse_vm_logs(vms_dir):
                     stats["failed_tests"][test_name] = []
                 stats["failed_tests"][test_name].append(instance_id)
 
-            stats["instances"].append(
-                {"test": test_name, "instance_id": instance_id, "status": status}
-            )
+            entry["status"] = status
+            stats["instances"].append(entry)
 
         except Exception as e:
             logger.warning("Could not parse VM log %s: %s", log_file, e)
@@ -184,6 +218,7 @@ def create_summary(
     expected_vm_count=None,
     s3_context=None,
     container_failure_log_url=None,
+    evidence_tests=(),
 ):
     """Create summary.json with VM statistics.
 
@@ -197,6 +232,8 @@ def create_summary(
             it exited non-zero before launching any VM. The KCIDB submitter
             uses this as ``tests[*].log_url`` for the synthetic Infrastructure
             row so users have a link to the actual failure reason.
+        evidence_tests: Test names whose VM success also requires a verified
+            evidence verdict in ``container.log`` (native pull-lab tests).
     """
     end_time = time.time()
     total_runtime = end_time - start_time
@@ -205,7 +242,8 @@ def create_summary(
 
     # Parse VM logs
     vms_dir = Path(run_dir) / "vms"
-    vm_stats = parse_vm_logs(vms_dir)
+    verdicts = parse_evidence_verdicts(Path(run_dir) / "container.log") if evidence_tests else {}
+    vm_stats = parse_vm_logs(vms_dir, evidence_tests=evidence_tests, verdicts=verdicts)
 
     # Determine status based on expected vs actual
     status = "success"
@@ -350,6 +388,7 @@ def run_pipeline(
         test_id = None
         test_names = set()  # Collect unique test names
         expected_vm_count = 0  # Calculate expected VM count
+        evidence_tests = set()  # Tests that must also pass structured verification
         if hasattr(provider, "config") and provider.config:
             test_config = provider.config.get("test_config", {})
             test_id = test_config.get("test_id")
@@ -358,6 +397,8 @@ def run_pipeline(
             vms = test_config.get("vms", [])
             for vm in vms:
                 test_value = vm.get("test")
+                if requires_evidence(vm.get("test_params")):
+                    evidence_tests.update(test_value if isinstance(test_value, list) else [test_value])
                 # Handle list of tests
                 if isinstance(test_value, list):
                     for test_name in test_value:
@@ -749,4 +790,5 @@ def run_pipeline(
             if "container_failure_log_url" in locals()
             else None
         ),
+        evidence_tests=evidence_tests if "evidence_tests" in locals() else (),
     )