chore: 144 - Log job execution finish

src/aethermesh_core/local_audit_event.py

@@ -21,6 +21,7 @@
         "lineage_linked",
         "contribution_record_updated",
         "job.execution.started",
+        "job.execution.finished",
         "capability_advertised",
         "node.shutdown",
     }
@@ -67,6 +68,14 @@
         "validation_expectation",
         "attribution_metadata_hash",
         "executing_node_id",
+        "execution_id",
+        "worker_node_id",
+        "terminal_status",
+        "finished_at",
+        "duration_ms",
+        "validator_node_id",
+        "output_artifact_refs",
+        "error_summary",
     }
 )
 
@@ -146,6 +155,8 @@ def validate_local_audit_event(event: Mapping[str, Any]) -> dict[str, Any]:
         _validate_job_submission_event(document)
     if document["event_type"] == "job.execution.started":
         _validate_job_execution_started_event(document)
+    if document["event_type"] == "job.execution.finished":
+        _validate_job_execution_finished_event(document)
     return document
 
 
@@ -358,3 +369,62 @@ def _validate_job_execution_started_event(document: Mapping[str, Any]) -> None:
         document["contribution_attribution"],
         "job.execution.started.contribution_attribution",
     )
+
+
+def _validate_job_execution_finished_event(document: Mapping[str, Any]) -> None:
+    """Require terminal local execution evidence without rewriting start entries."""
+
+    required_fields = (
+        "job_id",
+        "execution_id",
+        "worker_node_id",
+        "manifest_id",
+        "manifest_ref",
+        "lineage_refs",
+        "terminal_status",
+        "finished_at",
+        "duration_ms",
+        "validation_receipt_ref",
+        "validator_node_id",
+        "output_artifact_refs",
+        "error_summary",
+        "contribution_attribution",
+    )
+    missing = [field for field in required_fields if field not in document]
+    if missing:
+        raise LocalAuditEventError(
+            "job.execution.finished event is missing required context: "
+            f"{', '.join(missing)}"
+        )
+    for field in ("job_id", "execution_id", "manifest_id", "terminal_status"):
+        _require_text(document[field], f"job.execution.finished.{field}")
+    if document["terminal_status"] not in {
+        "completed",
+        "failed",
+        "cancelled",
+        "validation-failed",
+    }:
+        raise LocalAuditEventError(
+            "job.execution.finished.terminal_status is unsupported"
+        )
+    _require_timestamp(document["finished_at"])
+    if (
+        not isinstance(document["duration_ms"], int)
+        or isinstance(document["duration_ms"], bool)
+        or document["duration_ms"] < 0
+    ):
+        raise LocalAuditEventError(
+            "job.execution.finished.duration_ms must be a non-negative integer"
+        )
+    for field in ("worker_node_id", "validator_node_id", "error_summary"):
+        if document[field] is not None:
+            _require_text(document[field], f"job.execution.finished.{field}")
+    _require_local_paths(
+        document["output_artifact_refs"],
+        "job.execution.finished.output_artifact_refs",
+    )
+    _require_text(document["creator_node_id"], "job.execution.finished.creator_node_id")
+    _require_text_mapping(
+        document["contribution_attribution"],
+        "job.execution.finished.contribution_attribution",
+    )

src/aethermesh_core/runtime_service.py

@@ -2491,6 +2491,30 @@ def execute_submitted_local_job(
         error = result.error or (
             None if succeeded else f"validation failed: {validation.reason}"
         )
+        self._append_job_execution_finished_audit_event(
+            job_id=job_id,
+            worker_node_id=worker_node_id,
+            manifest=manifest,
+            manifest_ref=manifest_ref,
+            manifest_id=manifest_id,
+            finished_at=executor_finished_at,
+            duration_ms=_duration_ms(executor_started_at, executor_finished_at),
+            terminal_status=(
+                "completed"
+                if succeeded
+                else "validation-failed"
+                if result.status == "completed"
+                else "failed"
+            ),
+            validation_receipt_ref=receipt_ref,
+            validator_node_id=worker_node_id,
+            # The result artifact is written for both successful and failed
+            # executions, and remains useful terminal evidence when validation
+            # rejects an otherwise completed result.
+            output_artifact_refs=[result_ref],
+            error_summary=None if succeeded else _result_summary(error),
+            contribution_attribution=contribution_attribution,
+        )
         self._transition_local_job_state(
             job_id,
             "succeeded" if succeeded else "failed",
@@ -2588,12 +2612,121 @@ def _append_job_execution_started_audit_event(
                 f"could not write local job execution start audit event: {exc}"
             ) from exc
 
+    def _append_job_execution_finished_audit_event(
+        self,
+        *,
+        job_id: str,
+        worker_node_id: str | None,
+        manifest: dict[str, Any],
+        manifest_ref: str,
+        manifest_id: str,
+        finished_at: str,
+        duration_ms: int,
+        terminal_status: str,
+        validation_receipt_ref: str | None,
+        validator_node_id: str | None,
+        output_artifact_refs: list[str],
+        error_summary: str | None,
+        contribution_attribution: dict[str, Any],
+    ) -> None:
+        """Append one terminal local execution receipt before persisting terminal state."""
+
+        creator_node_id = manifest.get("creator_node_id")
+        lineage = manifest.get("lineage")
+        if not isinstance(creator_node_id, str) or not creator_node_id.strip():
+            raise RuntimeServiceError(
+                "job submission manifest creator_node_id is required for execution audit"
+            )
+        if not _provenance_matches_job(
+            lineage, contribution_attribution, job_id, creator_node_id
+        ):
+            raise RuntimeServiceError(
+                "job execution finish lineage or contribution attribution is invalid"
+            )
+        lineage = cast(dict[str, Any], lineage)
+        attribution_metadata = contribution_attribution.get("metadata")
+        if not isinstance(attribution_metadata, dict):
+            raise RuntimeServiceError(
+                "job execution finish contribution attribution metadata is invalid"
+            )
+        metadata_hash = canonical_json_hash(attribution_metadata, prefix="sha256:")
+        audit_attribution = {
+            "job_id": job_id,
+            "creator_node_id": creator_node_id,
+            "worker_node_id": worker_node_id or "not-assigned",
+            "validator_node_id": validator_node_id or "not-used",
+            "metadata_hash": metadata_hash,
+        }
+        actor_node_id = worker_node_id or self._local_node_id_for_submission()
+        audit_finished_at = datetime.fromisoformat(
+            finished_at.removesuffix("Z") + "+00:00"
+        ).strftime("%Y-%m-%dT%H:%M:%SZ")
+        audit_path = self.paths.data_dir / "audit" / "job-executions.jsonl"
+        event_id = f"local-audit-{job_id}-execution-finished"
+        try:
+            if audit_path.exists():
+                for line in audit_path.read_text(encoding="utf-8").splitlines():
+                    if json.loads(line).get("event_id") == event_id:
+                        return
+            append_local_audit_event(
+                audit_path,
+                {
+                    "schema_version": 1,
+                    "event_id": f"local-audit-{job_id}-execution-finished",
+                    "timestamp": audit_finished_at,
+                    "event_type": "job.execution.finished",
+                    "actor_node_id": actor_node_id,
+                    "creator_node_id": creator_node_id,
+                    "local_run_id": job_id,
+                    "job_id": job_id,
+                    "work_id": job_id,
+                    "execution_id": f"local-execution-{job_id}",
+                    "worker_node_id": worker_node_id,
+                    "manifest_id": manifest_id,
+                    "manifest_ref": manifest_ref,
+                    "lineage_refs": lineage["parent_refs"],
+                    "terminal_status": terminal_status,
+                    "finished_at": audit_finished_at,
+                    "duration_ms": duration_ms,
+                    "validation_receipt_ref": validation_receipt_ref,
+                    "validator_node_id": validator_node_id,
+                    "output_artifact_refs": output_artifact_refs,
+                    "error_summary": error_summary,
+                    "contribution_attribution_ids": [f"local-contribution-{job_id}"],
+                    "contribution_attribution": audit_attribution,
+                },
+            )
+        except (LocalAuditEventError, OSError, ValueError) as exc:
+            raise RuntimeServiceError(
+                f"could not write local job execution finish audit event: {exc}"
+            ) from exc
+
     def cancel_submitted_local_job(self, job_id: str) -> dict[str, Any]:
         """Cancel an unstarted local job without altering its manifest evidence."""
 
         before = self.get_local_job_status(job_id)
         if before["status"] == "not_found":
             raise RuntimeServiceError("local job not found")
+        manifest = self._load_local_job_document(
+            self.paths.data_dir / "job-submissions" / f"{job_id}.json",
+            "job submission manifest",
+        )
+        finished_at = _utc_timestamp()
+        self._append_job_execution_finished_audit_event(
+            job_id=job_id,
+            worker_node_id=before["worker_node_id"],
+            manifest=manifest,
+            manifest_ref=before["manifest_ref"],
+            manifest_id=canonical_json_hash(manifest, prefix="sha256:"),
+            finished_at=finished_at,
+            duration_ms=0,
+            terminal_status="cancelled",
+            validation_receipt_ref=None,
+            validator_node_id=None,
+            output_artifact_refs=[],
+            error_summary="local job cancelled before completion",
+            contribution_attribution=before["contribution_attribution"],
+        )
         self._transition_local_job_state(job_id, "canceled")
         self._append_event(f"canceled local job submission {job_id}")
         return self.get_local_job_status(job_id)

tests/test_local_audit_event.py

@@ -16,7 +16,7 @@ def test_validates_required_fields_and_allows_missing_optional_fields(self) -> N
         event = _minimal_event()
 
         self.assertEqual(validate_local_audit_event(event), event)
-        self.assertEqual(len(AUDIT_EVENT_TYPES), 11)
+        self.assertEqual(len(AUDIT_EVENT_TYPES), 12)
         self.assertIsNone(event["creator_node_id"])
 
     def test_appends_parseable_jsonl_events_with_local_references(self) -> None:

tests/test_runtime_service_api_cli.py

@@ -789,6 +789,115 @@ def test_local_job_manifest_is_removed_when_audit_identity_is_unavailable(
                 (root / "data" / "job-submissions" / f"{job_id}.json").exists()
             )
 
+    def test_execution_finish_audits_terminal_provenance_once(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            service = NodeRuntimeService.from_home(root)
+            request = {
+                "schema_version": 1,
+                "job_type": "echo",
+                "requested_capability": {"identifier": "work.echo"},
+                "input_payload": {
+                    "payload_type": "json",
+                    "content": {"message": "done"},
+                },
+                "creator_node_id": "creator-local-a",
+                "requested_validation_mode": "deterministic-local",
+                "lineage_parent_refs": ["data/prior-job.json"],
+                "attribution_metadata": {"project": "prototype"},
+            }
+            successful = service.submit_local_job(request)
+            completed = service.execute_submitted_local_job(
+                successful["job_id"], "worker-local-a"
+            )
+            failed = service.submit_local_job(
+                {
+                    **request,
+                    "job_type": "text_stats",
+                    "requested_capability": {"identifier": "work.text_stats"},
+                }
+            )
+            failed_status = service.execute_submitted_local_job(
+                failed["job_id"], "worker-local-b"
+            )
+            audit_path = root / "data" / "audit" / "job-executions.jsonl"
+            before_status_checks = audit_path.read_text(encoding="utf-8")
+            service.get_local_job_status(successful["job_id"])
+            service.get_local_job_status(failed["job_id"])
+            self.assertEqual(
+                audit_path.read_text(encoding="utf-8"), before_status_checks
+            )
+
+            finish_events = [
+                json.loads(line)
+                for line in audit_path.read_text(encoding="utf-8").splitlines()
+                if json.loads(line)["event_type"] == "job.execution.finished"
+            ]
+            self.assertEqual(len(finish_events), 2)
+            success_event, failure_event = finish_events
+            self.assertEqual(success_event["terminal_status"], "completed")
+            self.assertEqual(success_event["job_id"], successful["job_id"])
+            self.assertEqual(
+                success_event["execution_id"],
+                f"local-execution-{successful['job_id']}",
+            )
+            self.assertEqual(success_event["creator_node_id"], "creator-local-a")
+            self.assertEqual(success_event["worker_node_id"], "worker-local-a")
+            self.assertEqual(success_event["manifest_ref"], successful["manifest_ref"])
+            self.assertEqual(success_event["lineage_refs"], ["data/prior-job.json"])
+            self.assertEqual(
+                success_event["validation_receipt_ref"],
+                completed["validation"]["receipt_ref"],
+            )
+            self.assertEqual(success_event["validator_node_id"], "worker-local-a")
+            self.assertEqual(
+                success_event["contribution_attribution"],
+                {
+                    "job_id": successful["job_id"],
+                    "creator_node_id": "creator-local-a",
+                    "worker_node_id": "worker-local-a",
+                    "validator_node_id": "worker-local-a",
+                    "metadata_hash": success_event["contribution_attribution"][
+                        "metadata_hash"
+                    ],
+                },
+            )
+            self.assertEqual(
+                success_event["contribution_attribution_ids"],
+                [f"local-contribution-{successful['job_id']}"],
+            )
+            self.assertEqual(
+                success_event["output_artifact_refs"], [completed["result"]["ref"]]
+            )
+            self.assertIsNone(success_event["error_summary"])
+            self.assertEqual(failure_event["terminal_status"], "failed")
+            self.assertEqual(failure_event["job_id"], failed["job_id"])
+            self.assertEqual(failure_event["worker_node_id"], "worker-local-b")
+            self.assertEqual(
+                failure_event["validation_receipt_ref"],
+                failed_status["validation"]["receipt_ref"],
+            )
+            self.assertEqual(
+                failure_event["output_artifact_refs"],
+                [failed_status["result"]["ref"]],
+            )
+            self.assertEqual(failure_event["error_summary"], failed_status["error"])
+
+            cancel_root = root / "cancelled-runtime"
+            cancel_service = NodeRuntimeService.from_home(cancel_root)
+            cancelled = cancel_service.submit_local_job(request)
+            cancel_service.cancel_submitted_local_job(cancelled["job_id"])
+            cancellation_events = [
+                json.loads(line)
+                for line in (cancel_root / "data" / "audit" / "job-executions.jsonl")
+                .read_text(encoding="utf-8")
+                .splitlines()
+                if json.loads(line)["event_type"] == "job.execution.finished"
+            ]
+            self.assertEqual(len(cancellation_events), 1)
+            self.assertEqual(cancellation_events[0]["terminal_status"], "cancelled")
+            self.assertIsNone(cancellation_events[0]["validation_receipt_ref"])
+
     def test_execution_start_audits_precede_work_and_preserve_provenance(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
             root = Path(temp_dir)
@@ -832,9 +941,17 @@ def run_after_start(*args: Any, **kwargs: Any) -> JobResult:
                 json.loads(line)
                 for line in audit_path.read_text(encoding="utf-8").splitlines()
             ]
-            self.assertEqual(len(events), 2)
-            self.assertNotEqual(events[0]["event_id"], events[1]["event_id"])
-            for event, submission in zip(events, (first, second), strict=True):
+            self.assertEqual(len(events), 4)
+            started_events = [
+                event
+                for event in events
+                if event["event_type"] == "job.execution.started"
+            ]
+            self.assertEqual(len(started_events), 2)
+            self.assertNotEqual(
+                started_events[0]["event_id"], started_events[1]["event_id"]
+            )
+            for event, submission in zip(started_events, (first, second), strict=True):
                 self.assertEqual(event["event_type"], "job.execution.started")
                 self.assertEqual(event["job_id"], submission["job_id"])
                 self.assertEqual(event["creator_node_id"], "creator-local-a")