From 1705d6eb23ccc8c7caf9f78ac487302824ef53c5 Mon Sep 17 00:00:00 2001 From: ExoHayvan <18trevor3695@gmail.com> Date: Thu, 16 Jul 2026 00:57:38 -0400 Subject: [PATCH 1/2] chore: complete phase-1 step-142 --- docs/local-audit-event-schema.md | 6 +- src/aethermesh_core/local_audit_event.py | 45 +++++++++++++ src/aethermesh_core/runtime_service.py | 61 ++++++++++++++++++ tests/test_local_audit_event.py | 26 +++++++- tests/test_runtime_service_api_cli.py | 80 ++++++++++++++++++++++++ 5 files changed, 216 insertions(+), 2 deletions(-) diff --git a/docs/local-audit-event-schema.md b/docs/local-audit-event-schema.md index 68cffd9..e950154 100644 --- a/docs/local-audit-event-schema.md +++ b/docs/local-audit-event-schema.md @@ -18,7 +18,7 @@ Store one event per UTF-8 newline-delimited JSON (JSONL) line, normally under th | `creator_node_id` | non-empty string or `null` | Original creator node when known; `null` explicitly records that it is not known. | | `local_run_id` | non-empty string | Local invocation/run that produced the event. | -Supported event types are `node_initialized`, `manifest_created`, `work_submitted`, `validation_attempted`, `validation_result`, `lineage_linked`, `contribution_record_updated`, `capability_advertised`, and `node.shutdown`. +Supported event types are `node_initialized`, `manifest_created`, `work_submitted`, `job_submitted`, `validation_attempted`, `validation_result`, `lineage_linked`, `contribution_record_updated`, `capability_advertised`, and `node.shutdown`. ## Optional references @@ -37,6 +37,10 @@ Omit optional fields when they do not apply. Their absence is valid and has no e Hashes and signatures are optional. This format neither creates nor requires them unless existing local validation already produces them. +## Job submission events + +A `job_submitted` event is appended to `data/audit/job-submissions.jsonl` before a local submission is queued. It requires `job_id`, `local_node_id`, `manifest_ref`, a `manifest_hash` in `hashes`, `lineage_refs`, `validation_expectation`, `contribution_attribution`, and `attribution_metadata_hash`. The attribution map preserves the job and creator IDs plus a digest of the submitted metadata. The metadata body and input payload are deliberately not copied into the audit log, preventing prompts, credentials, and other large or private request data from being exposed there. + ## Capability advertisement events A `capability_advertised` event records one validated local startup-manifest claim after startup artifacts have been created. It requires `capability_advertisement_action` (`created`, `refreshed`, or `replaced`), `node_id`, `capability_id`, `manifest_ref`, `manifest_digest`, `advertisement_payload_digest`, `validation_status`, `validation_receipt_refs`, `lineage_refs`, and a string-only `contribution_attribution` map. The event stores digests and safe relative references rather than the full advertisement payload, so local runtime details are not copied into the audit log. diff --git a/src/aethermesh_core/local_audit_event.py b/src/aethermesh_core/local_audit_event.py index cd0eb64..249f232 100644 --- a/src/aethermesh_core/local_audit_event.py +++ b/src/aethermesh_core/local_audit_event.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import os from collections.abc import Mapping from datetime import datetime from pathlib import Path, PureWindowsPath @@ -14,6 +15,7 @@ "node_initialized", "manifest_created", "work_submitted", + "job_submitted", "validation_attempted", "validation_result", "lineage_linked", @@ -59,6 +61,10 @@ "advertisement_payload_digest", "validation_receipt_refs", "contribution_attribution", + "job_id", + "local_node_id", + "validation_expectation", + "attribution_metadata_hash", } ) @@ -134,6 +140,8 @@ def validate_local_audit_event(event: Mapping[str, Any]) -> dict[str, Any]: _validate_shutdown_event(document) if document["event_type"] == "capability_advertised": _validate_capability_advertisement_event(document) + if document["event_type"] == "job_submitted": + _validate_job_submission_event(document) return document @@ -147,6 +155,8 @@ def append_local_audit_event( audit_path.parent.mkdir(parents=True, exist_ok=True) with audit_path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(document, sort_keys=True, separators=(",", ":")) + "\n") + handle.flush() + os.fsync(handle.fileno()) return document @@ -272,3 +282,38 @@ def _validate_capability_advertisement_event(document: Mapping[str, Any]) -> Non document["contribution_attribution"], "capability_advertised.contribution_attribution", ) + + +def _validate_job_submission_event(document: Mapping[str, Any]) -> None: + """Require compact, non-payload evidence for one accepted local submission.""" + + required_fields = ( + "job_id", + "local_node_id", + "manifest_ref", + "hashes", + "lineage_refs", + "validation_expectation", + "contribution_attribution", + "attribution_metadata_hash", + ) + missing = [field for field in required_fields if field not in document] + if missing: + raise LocalAuditEventError( + f"job_submitted event is missing required context: {', '.join(missing)}" + ) + for field in ( + "job_id", + "local_node_id", + "validation_expectation", + "attribution_metadata_hash", + ): + _require_text(document[field], f"job_submitted.{field}") + if document["actor_node_id"] != document["local_node_id"]: + raise LocalAuditEventError( + "job_submitted.local_node_id must match actor_node_id" + ) + _require_text(document["creator_node_id"], "job_submitted.creator_node_id") + _require_text_mapping( + document["contribution_attribution"], "job_submitted.contribution_attribution" + ) diff --git a/src/aethermesh_core/runtime_service.py b/src/aethermesh_core/runtime_service.py index a59f8e6..0d8d4f5 100644 --- a/src/aethermesh_core/runtime_service.py +++ b/src/aethermesh_core/runtime_service.py @@ -37,6 +37,10 @@ validate_job_result_document, ) from aethermesh_core.local_json_helpers import canonical_json_hash +from aethermesh_core.local_audit_event import ( + LocalAuditEventError, + append_local_audit_event, +) from aethermesh_core.models import Job, JobResult, NodeIdentity from aethermesh_core.result_hash import canonical_result_document_hash from aethermesh_core.runner import LocalRunner, run_local_job @@ -999,6 +1003,51 @@ def submit_local_job(self, request: dict[str, Any]) -> dict[str, Any]: return self._existing_local_submission( manifest_path, job_id, submission_fingerprint ) + try: + local_node_id = self._local_node_id_for_submission() + attribution_metadata_hash = canonical_json_hash( + attribution_metadata, prefix="sha256:" + ) + append_local_audit_event( + self.paths.data_dir / "audit" / "job-submissions.jsonl", + { + "schema_version": 1, + "event_id": f"local-audit-{job_id}-submitted", + "timestamp": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), + "event_type": "job_submitted", + "actor_node_id": local_node_id, + "local_node_id": local_node_id, + "creator_node_id": creator_node_id, + "local_run_id": job_id, + "job_id": job_id, + "work_id": job_id, + "manifest_id": job_id, + "manifest_ref": manifest_ref, + "hashes": { + "manifest_hash": canonical_json_hash( + json.loads(manifest_path.read_text(encoding="utf-8")), + prefix="sha256:", + ) + }, + "lineage_refs": lineage_parent_refs, + "validation_expectation": validation_mode, + "validation_status": "pending", + "contribution_attribution": { + "job_id": job_id, + "creator_node_id": creator_node_id, + "metadata_hash": attribution_metadata_hash, + }, + "attribution_metadata_hash": attribution_metadata_hash, + }, + ) + except (LocalAuditEventError, OSError, ValueError) as exc: + try: + manifest_path.unlink() + except OSError: + pass + raise RuntimeServiceError( + f"could not write local job submission audit event: {exc}" + ) from exc record: dict[str, Any] = { "version": LOCAL_JOB_SUBMISSION_SCHEMA_VERSION, "job_id": job_id, @@ -1033,6 +1082,18 @@ def submit_local_job(self, request: dict[str, Any]) -> dict[str, Any]: "network_mode": "local-only-no-p2p", } + def _local_node_id_for_submission(self) -> str: + """Load the persistent local identity needed for a submission audit record.""" + + if not self.paths.config_path.exists(): + self.initialize_local_node_data() + local_node_id = _config_node_id(self.load_config()) + if local_node_id is None: + raise RuntimeServiceError( + "local node identity is unavailable for job audit" + ) + return local_node_id + def _existing_local_submission( self, manifest_path: Path, job_id: str, submission_fingerprint: str ) -> dict[str, Any]: diff --git a/tests/test_local_audit_event.py b/tests/test_local_audit_event.py index 1e8f34d..e2d04da 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), 9) + self.assertEqual(len(AUDIT_EVENT_TYPES), 10) self.assertIsNone(event["creator_node_id"]) def test_appends_parseable_jsonl_events_with_local_references(self) -> None: @@ -84,6 +84,30 @@ def test_rejects_invalid_optional_references(self) -> None: with self.assertRaisesRegex(LocalAuditEventError, message): validate_local_audit_event({**_minimal_event(), **optional_fields}) + def test_job_submitted_requires_compact_submission_evidence(self) -> None: + event = { + **_minimal_event(), + "event_type": "job_submitted", + "actor_node_id": "local-node-a", + "creator_node_id": "creator-node-a", + "job_id": "local-job-a", + "local_node_id": "local-node-a", + "manifest_ref": "data/job-submissions/local-job-a.json", + "hashes": {"manifest_hash": "sha256:manifest"}, + "lineage_refs": ["data/lineage/parent.json"], + "validation_expectation": "deterministic-local", + "contribution_attribution": { + "job_id": "local-job-a", + "creator_node_id": "creator-node-a", + "metadata_hash": "sha256:metadata", + }, + "attribution_metadata_hash": "sha256:metadata", + } + + self.assertEqual(validate_local_audit_event(event), event) + with self.assertRaisesRegex(LocalAuditEventError, "local_node_id"): + validate_local_audit_event({**event, "local_node_id": "other-node"}) + def _minimal_event() -> dict[str, object]: return { diff --git a/tests/test_runtime_service_api_cli.py b/tests/test_runtime_service_api_cli.py index 84ccd3e..decdd7f 100644 --- a/tests/test_runtime_service_api_cli.py +++ b/tests/test_runtime_service_api_cli.py @@ -671,6 +671,86 @@ def test_local_job_submission_persists_provenance_without_execution(self) -> Non self.assertNotEqual(response["job_id"], second["job_id"]) self.assertFalse((Path(temp_dir) / "data" / "receipts").exists()) + def test_local_job_submission_appends_complete_audit_events_before_queueing( + 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": "private"}, + }, + "creator_node_id": "creator-local-a", + "requested_validation_mode": "deterministic-local", + "lineage_parent_refs": ["data/prior-job.json"], + "attribution_metadata": {"project": "prototype"}, + } + + first = service.submit_local_job(request) + second = service.submit_local_job(request) + audit_path = root / "data" / "audit" / "job-submissions.jsonl" + events = [ + 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(event["event_type"], "job_submitted") + self.assertTrue(event["local_node_id"]) + self.assertEqual(event["creator_node_id"], "creator-local-a") + self.assertEqual(event["job_id"], submission["job_id"]) + self.assertEqual(event["manifest_ref"], submission["manifest_ref"]) + self.assertEqual(event["lineage_refs"], ["data/prior-job.json"]) + self.assertEqual(event["validation_expectation"], "deterministic-local") + self.assertEqual( + event["contribution_attribution"]["creator_node_id"], + "creator-local-a", + ) + self.assertIn("metadata_hash", event["contribution_attribution"]) + self.assertNotIn("private", json.dumps(event)) + + def test_local_job_is_not_accepted_when_submission_audit_write_fails(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + service = NodeRuntimeService.from_home(root) + job_id = "local-job-0123456789abcdef0123456789abcdef" + request = { + "schema_version": 1, + "job_id": job_id, + "job_type": "echo", + "requested_capability": {"identifier": "work.echo"}, + "input_payload": { + "payload_type": "json", + "content": {"message": "hello"}, + }, + "creator_node_id": "creator-local-a", + "requested_validation_mode": "deterministic-local", + "lineage_parent_refs": [], + "attribution_metadata": {}, + } + + with patch( + "aethermesh_core.runtime_service.append_local_audit_event", + side_effect=OSError("read-only audit log"), + ): + with self.assertRaisesRegex( + RuntimeServiceError, + "could not write local job submission audit event", + ): + service.submit_local_job(request) + + self.assertFalse((root / "data" / "job-status" / f"{job_id}.json").exists()) + self.assertFalse( + (root / "data" / "job-submissions" / f"{job_id}.json").exists() + ) + def test_supplied_local_job_id_retries_idempotently_and_rejects_conflicts( self, ) -> None: From b3b0d1365ef4cf3ecc621a11681b865be754bdd8 Mon Sep 17 00:00:00 2001 From: ExoHayvan <18trevor3695@gmail.com> Date: Thu, 16 Jul 2026 01:49:25 -0400 Subject: [PATCH 2/2] fix: address automated review feedback --- src/aethermesh_core/runtime_service.py | 2 +- tests/test_runtime_service_api_cli.py | 38 ++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/src/aethermesh_core/runtime_service.py b/src/aethermesh_core/runtime_service.py index 0d8d4f5..4cb9e20 100644 --- a/src/aethermesh_core/runtime_service.py +++ b/src/aethermesh_core/runtime_service.py @@ -1040,7 +1040,7 @@ def submit_local_job(self, request: dict[str, Any]) -> dict[str, Any]: "attribution_metadata_hash": attribution_metadata_hash, }, ) - except (LocalAuditEventError, OSError, ValueError) as exc: + except (LocalAuditEventError, OSError, RuntimeServiceError, ValueError) as exc: try: manifest_path.unlink() except OSError: diff --git a/tests/test_runtime_service_api_cli.py b/tests/test_runtime_service_api_cli.py index decdd7f..722ea52 100644 --- a/tests/test_runtime_service_api_cli.py +++ b/tests/test_runtime_service_api_cli.py @@ -751,6 +751,44 @@ def test_local_job_is_not_accepted_when_submission_audit_write_fails(self) -> No (root / "data" / "job-submissions" / f"{job_id}.json").exists() ) + def test_local_job_manifest_is_removed_when_audit_identity_is_unavailable( + self, + ) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + service = NodeRuntimeService.from_home(root) + job_id = "local-job-0123456789abcdef0123456789abcdef" + request = { + "schema_version": 1, + "job_id": job_id, + "job_type": "echo", + "requested_capability": {"identifier": "work.echo"}, + "input_payload": { + "payload_type": "json", + "content": {"message": "hello"}, + }, + "creator_node_id": "creator-local-a", + "requested_validation_mode": "deterministic-local", + "lineage_parent_refs": [], + "attribution_metadata": {}, + } + + with patch.object( + service, + "_local_node_id_for_submission", + side_effect=RuntimeServiceError("identity unavailable"), + ): + with self.assertRaisesRegex( + RuntimeServiceError, + "could not write local job submission audit event", + ): + service.submit_local_job(request) + + self.assertFalse((root / "data" / "job-status" / f"{job_id}.json").exists()) + self.assertFalse( + (root / "data" / "job-submissions" / f"{job_id}.json").exists() + ) + def test_supplied_local_job_id_retries_idempotently_and_rejects_conflicts( self, ) -> None: