chore: 142 - Log job submission

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.

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

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, RuntimeServiceError, 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]:

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 {

tests/test_runtime_service_api_cli.py

@@ -671,6 +671,124 @@ 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_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: