From 371fae19ac453a08af085826575c2ab14453203f Mon Sep 17 00:00:00 2001 From: ExoHayvan <18trevor3695@gmail.com> Date: Thu, 16 Jul 2026 05:25:49 -0400 Subject: [PATCH 1/2] chore: complete phase-1 step-144 --- src/aethermesh_core/local_audit_event.py | 70 ++++++++++++ src/aethermesh_core/runtime_service.py | 130 +++++++++++++++++++++++ tests/test_local_audit_event.py | 2 +- tests/test_runtime_service_api_cli.py | 120 ++++++++++++++++++++- 4 files changed, 318 insertions(+), 4 deletions(-) diff --git a/src/aethermesh_core/local_audit_event.py b/src/aethermesh_core/local_audit_event.py index 6819628..0fb44df 100644 --- a/src/aethermesh_core/local_audit_event.py +++ b/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", + ) diff --git a/src/aethermesh_core/runtime_service.py b/src/aethermesh_core/runtime_service.py index adbb816..b026d9e 100644 --- a/src/aethermesh_core/runtime_service.py +++ b/src/aethermesh_core/runtime_service.py @@ -2491,6 +2491,27 @@ 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, + output_artifact_refs=[result_ref] if succeeded else [], + 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 +2609,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) diff --git a/tests/test_local_audit_event.py b/tests/test_local_audit_event.py index e4bcca0..a66ca36 100644 --- a/tests/test_local_audit_event.py +++ b/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: diff --git a/tests/test_runtime_service_api_cli.py b/tests/test_runtime_service_api_cli.py index 659a8eb..edd32db 100644 --- a/tests/test_runtime_service_api_cli.py +++ b/tests/test_runtime_service_api_cli.py @@ -789,6 +789,112 @@ 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"], []) + 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 +938,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") From e2d2dc626a1bbd4e7d18a1ba4a2f9ee40eedeee2 Mon Sep 17 00:00:00 2001 From: ExoHayvan <18trevor3695@gmail.com> Date: Thu, 16 Jul 2026 05:42:58 -0400 Subject: [PATCH 2/2] fix: address automated review feedback --- src/aethermesh_core/runtime_service.py | 5 ++++- tests/test_runtime_service_api_cli.py | 5 ++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/src/aethermesh_core/runtime_service.py b/src/aethermesh_core/runtime_service.py index b026d9e..eca9538 100644 --- a/src/aethermesh_core/runtime_service.py +++ b/src/aethermesh_core/runtime_service.py @@ -2508,7 +2508,10 @@ def execute_submitted_local_job( ), validation_receipt_ref=receipt_ref, validator_node_id=worker_node_id, - output_artifact_refs=[result_ref] if succeeded else [], + # 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, ) diff --git a/tests/test_runtime_service_api_cli.py b/tests/test_runtime_service_api_cli.py index edd32db..f8eda3d 100644 --- a/tests/test_runtime_service_api_cli.py +++ b/tests/test_runtime_service_api_cli.py @@ -877,7 +877,10 @@ def test_execution_finish_audits_terminal_provenance_once(self) -> None: failure_event["validation_receipt_ref"], failed_status["validation"]["receipt_ref"], ) - self.assertEqual(failure_event["output_artifact_refs"], []) + 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"