diff --git a/src/aethermesh_core/local_audit_event.py b/src/aethermesh_core/local_audit_event.py index 249f232..6819628 100644 --- a/src/aethermesh_core/local_audit_event.py +++ b/src/aethermesh_core/local_audit_event.py @@ -20,6 +20,7 @@ "validation_result", "lineage_linked", "contribution_record_updated", + "job.execution.started", "capability_advertised", "node.shutdown", } @@ -65,6 +66,7 @@ "local_node_id", "validation_expectation", "attribution_metadata_hash", + "executing_node_id", } ) @@ -142,6 +144,8 @@ def validate_local_audit_event(event: Mapping[str, Any]) -> dict[str, Any]: _validate_capability_advertisement_event(document) if document["event_type"] == "job_submitted": _validate_job_submission_event(document) + if document["event_type"] == "job.execution.started": + _validate_job_execution_started_event(document) return document @@ -317,3 +321,40 @@ def _validate_job_submission_event(document: Mapping[str, Any]) -> None: _require_text_mapping( document["contribution_attribution"], "job_submitted.contribution_attribution" ) + + +def _validate_job_execution_started_event(document: Mapping[str, Any]) -> None: + """Require compact provenance before a locally submitted job can run.""" + + required_fields = ( + "job_id", + "executing_node_id", + "manifest_id", + "manifest_ref", + "hashes", + "lineage_refs", + "contribution_attribution", + "attribution_metadata_hash", + ) + missing = [field for field in required_fields if field not in document] + if missing: + raise LocalAuditEventError( + "job.execution.started event is missing required context: " + f"{', '.join(missing)}" + ) + for field in ( + "job_id", + "executing_node_id", + "manifest_id", + "attribution_metadata_hash", + ): + _require_text(document[field], f"job.execution.started.{field}") + if document["actor_node_id"] != document["executing_node_id"]: + raise LocalAuditEventError( + "job.execution.started.executing_node_id must match actor_node_id" + ) + _require_text(document["creator_node_id"], "job.execution.started.creator_node_id") + _require_text_mapping( + document["contribution_attribution"], + "job.execution.started.contribution_attribution", + ) diff --git a/src/aethermesh_core/runtime_service.py b/src/aethermesh_core/runtime_service.py index 4cb9e20..adbb816 100644 --- a/src/aethermesh_core/runtime_service.py +++ b/src/aethermesh_core/runtime_service.py @@ -2258,14 +2258,6 @@ def execute_submitted_local_job( raise RuntimeServiceError("local job is not queued") if not isinstance(worker_node_id, str) or not worker_node_id.strip(): raise RuntimeServiceError("worker_node_id must be a non-empty string") - self._transition_local_job_state( - job_id, - "running", - updates={ - "worker_node_id": worker_node_id, - "executor_node_id": worker_node_id, - }, - ) manifest = self._load_local_job_document( self.paths.data_dir / "job-submissions" / f"{job_id}.json", "job submission manifest", @@ -2289,6 +2281,22 @@ def execute_submitted_local_job( raise RuntimeServiceError( "job submission manifest input_payload hash is invalid" ) + self._append_job_execution_started_audit_event( + job_id=job_id, + worker_node_id=worker_node_id, + manifest=manifest, + manifest_ref=f"data/job-submissions/{job_id}.json", + manifest_hash=canonical_json_hash(manifest, prefix="sha256:"), + payload_hash=payload_hash, + ) + self._transition_local_job_state( + job_id, + "running", + updates={ + "worker_node_id": worker_node_id, + "executor_node_id": worker_node_id, + }, + ) job = Job( job_id=job_id, job_type=job_data["job_type"], @@ -2518,6 +2526,68 @@ def execute_submitted_local_job( self._append_event(f"executed local job submission {job_id}") return self.get_local_job_status(job_id) + def _append_job_execution_started_audit_event( + self, + *, + job_id: str, + worker_node_id: str, + manifest: dict[str, Any], + manifest_ref: str, + manifest_hash: str, + payload_hash: str, + ) -> None: + """Append the durable local start receipt before invoking the work runner.""" + + creator_node_id = manifest.get("creator_node_id") + lineage = manifest.get("lineage") + attribution = manifest.get("contribution_attribution") + 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, attribution, job_id, creator_node_id): + raise RuntimeServiceError( + "job submission manifest lineage or contribution attribution is invalid " + "for execution audit" + ) + lineage = cast(dict[str, Any], lineage) + attribution = cast(dict[str, Any], attribution) + metadata_hash = canonical_json_hash(attribution["metadata"], prefix="sha256:") + try: + append_local_audit_event( + self.paths.data_dir / "audit" / "job-executions.jsonl", + { + "schema_version": 1, + "event_id": f"local-audit-{job_id}-execution-started", + "timestamp": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), + "event_type": "job.execution.started", + "actor_node_id": worker_node_id, + "creator_node_id": creator_node_id, + "local_run_id": job_id, + "job_id": job_id, + "work_id": job_id, + "executing_node_id": worker_node_id, + "manifest_id": job_id, + "manifest_ref": manifest_ref, + "hashes": { + "manifest_hash": manifest_hash, + "input_payload_hash": payload_hash, + }, + "lineage_refs": lineage["parent_refs"], + "contribution_attribution": { + "job_id": job_id, + "creator_node_id": creator_node_id, + "executing_node_id": worker_node_id, + "metadata_hash": metadata_hash, + }, + "attribution_metadata_hash": metadata_hash, + }, + ) + except (LocalAuditEventError, OSError, ValueError) as exc: + raise RuntimeServiceError( + f"could not write local job execution start 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.""" diff --git a/tests/test_local_audit_event.py b/tests/test_local_audit_event.py index e2d04da..e4bcca0 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), 10) + self.assertEqual(len(AUDIT_EVENT_TYPES), 11) self.assertIsNone(event["creator_node_id"]) def test_appends_parseable_jsonl_events_with_local_references(self) -> None: @@ -108,6 +108,38 @@ def test_job_submitted_requires_compact_submission_evidence(self) -> None: with self.assertRaisesRegex(LocalAuditEventError, "local_node_id"): validate_local_audit_event({**event, "local_node_id": "other-node"}) + def test_execution_started_requires_compact_execution_provenance(self) -> None: + event = { + **_minimal_event(), + "event_type": "job.execution.started", + "actor_node_id": "worker-node-a", + "creator_node_id": "creator-node-a", + "job_id": "local-job-a", + "executing_node_id": "worker-node-a", + "manifest_id": "local-job-a", + "manifest_ref": "data/job-submissions/local-job-a.json", + "hashes": { + "manifest_hash": "sha256:manifest", + "input_payload_hash": "sha256:input", + }, + "lineage_refs": ["data/lineage/parent.json"], + "contribution_attribution": { + "job_id": "local-job-a", + "creator_node_id": "creator-node-a", + "executing_node_id": "worker-node-a", + "metadata_hash": "sha256:metadata", + }, + "attribution_metadata_hash": "sha256:metadata", + } + + self.assertEqual(validate_local_audit_event(event), event) + with self.assertRaisesRegex(LocalAuditEventError, "manifest_id"): + validate_local_audit_event( + {key: value for key, value in event.items() if key != "manifest_id"} + ) + with self.assertRaisesRegex(LocalAuditEventError, "must match actor_node_id"): + validate_local_audit_event({**event, "executing_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 722ea52..659a8eb 100644 --- a/tests/test_runtime_service_api_cli.py +++ b/tests/test_runtime_service_api_cli.py @@ -789,6 +789,137 @@ def test_local_job_manifest_is_removed_when_audit_identity_is_unavailable( (root / "data" / "job-submissions" / f"{job_id}.json").exists() ) + def test_execution_start_audits_precede_work_and_preserve_provenance(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 execution payload"}, + }, + "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-executions.jsonl" + runner = run_local_job + + def run_after_start(*args: Any, **kwargs: Any) -> JobResult: + events = [ + json.loads(line) + for line in audit_path.read_text(encoding="utf-8").splitlines() + ] + self.assertEqual(len(events), 1) + self.assertFalse( + (root / "data" / "job-results" / f"{first['job_id']}.json").exists() + ) + return runner(*args, **kwargs) + + with patch( + "aethermesh_core.runtime_service.run_local_job", run_after_start + ): + service.execute_submitted_local_job(first["job_id"], "worker-local-a") + service.execute_submitted_local_job(second["job_id"], "worker-local-a") + + 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.execution.started") + self.assertEqual(event["job_id"], submission["job_id"]) + self.assertEqual(event["creator_node_id"], "creator-local-a") + self.assertEqual(event["executing_node_id"], "worker-local-a") + self.assertEqual(event["manifest_id"], submission["job_id"]) + self.assertEqual(event["manifest_ref"], submission["manifest_ref"]) + self.assertEqual(event["lineage_refs"], ["data/prior-job.json"]) + self.assertEqual( + event["contribution_attribution"]["creator_node_id"], + "creator-local-a", + ) + self.assertIn("input_payload_hash", event["hashes"]) + self.assertNotIn("private execution payload", json.dumps(event)) + + def test_execution_start_audit_failure_keeps_job_queued_and_blocks_work( + self, + ) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + service = NodeRuntimeService.from_home(root) + submission = service.submit_local_job( + { + "schema_version": 1, + "job_type": "echo", + "requested_capability": {"identifier": "work.echo"}, + "input_payload": {"payload_type": "json", "content": {}}, + "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"), + ), + patch("aethermesh_core.runtime_service.run_local_job") as runner, + ): + with self.assertRaisesRegex( + RuntimeServiceError, + "could not write local job execution start audit event", + ): + service.execute_submitted_local_job( + submission["job_id"], "worker-local-a" + ) + runner.assert_not_called() + self.assertEqual( + service.get_local_job_status(submission["job_id"])["status"], "queued" + ) + + def test_execution_start_requires_manifest_creator_identifier(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + service = NodeRuntimeService.from_home(root) + submission = service.submit_local_job( + { + "schema_version": 1, + "job_type": "echo", + "requested_capability": {"identifier": "work.echo"}, + "input_payload": {"payload_type": "json", "content": {}}, + "creator_node_id": "creator-local-a", + "requested_validation_mode": "deterministic-local", + "lineage_parent_refs": [], + "attribution_metadata": {}, + } + ) + manifest_path = root / submission["manifest_ref"] + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["creator_node_id"] = "" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + + with self.assertRaisesRegex( + RuntimeServiceError, "creator_node_id is required" + ): + service.execute_submitted_local_job( + submission["job_id"], "worker-local-a" + ) + self.assertFalse( + (root / "data" / "audit" / "job-executions.jsonl").exists() + ) + self.assertEqual( + service.get_local_job_status(submission["job_id"])["status"], "queued" + ) + def test_supplied_local_job_id_retries_idempotently_and_rejects_conflicts( self, ) -> None: