chore: 143 - Log job execution start

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",
+    )

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."""
 

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 {

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: