chore: 178 - Include manifest reference in contribution records when applicable

src/aethermesh_core/cli.py

@@ -712,6 +712,7 @@ def run_local_batch(
                 validation_valid=validation.valid,
                 validation_reason=validation.reason,
                 job_type=job.job_type,
+                manifest_ref=manifest_path,
             )
         save_ledger_document(ledger_path, ledger, extra_fields)
         result["ledger_path"] = ledger_path

src/aethermesh_core/dispatch.py

@@ -118,6 +118,7 @@ def dispatch_local_batch(
                 "job_type": job.job_type,
                 "payload": dict(job.payload),
                 "node_id": assignment.node_id,
+                "manifest_ref": manifest_path,
             },
             correlation_id=job.job_id,
         )

src/aethermesh_core/ledger.py

@@ -33,13 +33,16 @@ class ContributionRecord:
     result_hash: str | None = None
     version_metadata_ref: str | None = None
     created_at: str | None = None
+    manifest_ref: str | None = None
 
     def to_dict(self) -> dict[str, Any]:
         """Serialize the record into a JSON-compatible dictionary."""
 
         document = asdict(self)
         if self.version_metadata_ref is None:
             document.pop("version_metadata_ref")
+        if self.manifest_ref is None:
+            document.pop("manifest_ref")
         if self.created_at is None:
             document.pop("created_at")
         return document
@@ -80,6 +83,8 @@ def from_dict(cls, payload: dict[str, Any]) -> "ContributionRecord":
             raise LedgerPersistenceError(
                 "ledger record field 'version_metadata_ref' must be a string or null"
             )
+        manifest_ref = payload.get("manifest_ref")
+        _require_optional_manifest_ref(manifest_ref)
         created_at = payload.get("created_at")
         if created_at is not None:
             _require_utc_timestamp("created_at", created_at)
@@ -95,6 +100,7 @@ def from_dict(cls, payload: dict[str, Any]) -> "ContributionRecord":
             result_hash=result_hash,
             version_metadata_ref=version_ref,
             created_at=created_at,
+            manifest_ref=manifest_ref,
         )
 
 
@@ -134,14 +140,19 @@ def record(
         validation_reason: str | None = None,
         job_type: str | None = None,
         version_metadata_ref: str | None = None,
+        manifest_ref: str | None = None,
     ) -> ContributionRecord:
         """Record one result and return its local contribution record.
 
         Only completed results with positive integer contribution units add to
         totals. Failed, zero, and negative-unit results are preserved for local
         auditability but contribute zero units.
+
+        ``manifest_ref`` is recorded only when the caller has known manifest
+        provenance; it may be a stable local reference or content hash.
         """
 
+        _require_optional_manifest_ref(manifest_ref)
         units = _accounted_units(result)
         message = (
             result.error if result.error is not None else _string_output(result.output)
@@ -158,6 +169,7 @@ def record(
             result_hash=canonical_result_hash(result),
             version_metadata_ref=version_metadata_ref,
             created_at=_utc_timestamp(self._clock()),
+            manifest_ref=manifest_ref,
         )
         self._records.append(record)
         return record
@@ -338,6 +350,13 @@ def _require_record_field(
         )
 
 
+def _require_optional_manifest_ref(value: object) -> None:
+    if value is not None and (not isinstance(value, str) or not value.strip()):
+        raise LedgerPersistenceError(
+            "ledger record field 'manifest_ref' must be a non-empty string or null"
+        )
+
+
 def _require_result_hash(field_name: str, value: object) -> None:
     if not isinstance(value, str) or len(value) != 64:
         raise LedgerPersistenceError(

src/aethermesh_core/node_service.py

@@ -107,6 +107,12 @@ def process_inbox(self) -> InboxProcessResult:
 
     def _process_assignment(self, message: MeshMessage) -> ProcessedAssignment:
         job = _job_from_assignment_payload(message)
+        raw_manifest_ref = message.payload.get("manifest_ref")
+        manifest_ref = (
+            raw_manifest_ref
+            if isinstance(raw_manifest_ref, str) and raw_manifest_ref
+            else None
+        )
         result = self.runner.run(job)
         validation = validate_job_result(
             job, result, expected_node_id=self.identity.node_id
@@ -122,6 +128,7 @@ def _process_assignment(self, message: MeshMessage) -> ProcessedAssignment:
             validation_reason=validation.reason,
             job_type=job.job_type,
             version_metadata_ref=self.version_metadata_ref,
+            manifest_ref=manifest_ref,
         )
 
         result_message = self._send_message(

tests/test_cli.py

@@ -1016,10 +1016,12 @@ def test_run_local_batch_persists_json_ledger_and_accumulates(self) -> None:
         self.assertEqual(persisted["records"][0]["validation_valid"], True)
         self.assertEqual(persisted["records"][0]["validation_reason"], "ok")
         self.assertEqual(persisted["records"][0]["job_type"], "echo")
+        self.assertEqual(persisted["records"][0]["manifest_ref"], str(manifest_path))
         self.assertEqual(persisted["records"][1]["node_id"], "local-node-b")
         self.assertEqual(persisted["records"][1]["validation_valid"], True)
         self.assertEqual(persisted["records"][1]["validation_reason"], "ok")
         self.assertEqual(persisted["records"][1]["job_type"], "text_stats")
+        self.assertEqual(persisted["records"][1]["manifest_ref"], str(manifest_path))
 
     def test_run_local_batch_preserves_existing_ledger_metadata(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:

tests/test_dispatch.py

@@ -64,18 +64,21 @@ def test_dispatch_emits_heartbeats_and_assignments_without_execution_messages(
                     "job_type": "echo",
                     "payload": {"message": "one"},
                     "node_id": "node-a",
+                    "manifest_ref": "examples/local-batch.json",
                 },
                 {
                     "job_id": "stats-1",
                     "job_type": "text_stats",
                     "payload": {"text": "hello mesh"},
                     "node_id": "node-a",
+                    "manifest_ref": "examples/local-batch.json",
                 },
                 {
                     "job_id": "echo-2",
                     "job_type": "echo",
                     "payload": {"message": "two"},
                     "node_id": "node-c",
+                    "manifest_ref": "examples/local-batch.json",
                 },
             ],
         )

tests/test_ledger.py

@@ -174,11 +174,16 @@ def test_contribution_record_round_trips_through_json_dict(self) -> None:
             validation_reason="ok",
             job_type="echo",
             result_hash="a" * 64,
+            manifest_ref="examples/job-envelopes/minimal-local-echo.json",
         )
 
         decoded = ContributionRecord.from_dict(json.loads(json.dumps(record.to_dict())))
 
         self.assertEqual(decoded, record)
+        self.assertEqual(
+            decoded.manifest_ref,
+            "examples/job-envelopes/minimal-local-echo.json",
+        )
 
     def test_contribution_record_rejects_invalid_result_hash(self) -> None:
         valid = ContributionRecord(
@@ -197,6 +202,28 @@ def test_contribution_record_rejects_invalid_result_hash(self) -> None:
                 ):
                     ContributionRecord.from_dict(payload)
 
+    def test_contribution_record_rejects_empty_manifest_reference(self) -> None:
+        payload = ContributionRecord(
+            node_id="node-a",
+            job_id="job-1",
+            status="completed",
+            contribution_units=1,
+        ).to_dict()
+
+        with self.assertRaisesRegex(LedgerPersistenceError, "manifest_ref"):
+            ContributionRecord.from_dict({**payload, "manifest_ref": ""})
+
+    def test_record_rejects_empty_manifest_reference_before_persistence(self) -> None:
+        ledger = ContributionLedger()
+
+        with self.assertRaisesRegex(LedgerPersistenceError, "manifest_ref"):
+            ledger.record(
+                JobResult("job-1", "node-a", "completed", "hello mesh", None, 1),
+                manifest_ref="",
+            )
+
+        self.assertEqual(ledger.to_document()["records"], [])
+
     def test_legacy_contribution_record_without_validation_metadata_loads(self) -> None:
         payload = {
             "node_id": "node-a",
@@ -217,6 +244,7 @@ def test_legacy_contribution_record_without_validation_metadata_loads(self) -> N
         self.assertIsNone(record.validation_reason)
         self.assertIsNone(record.job_type)
         self.assertIsNone(record.result_hash)
+        self.assertIsNone(record.manifest_ref)
 
     def test_record_persists_validation_metadata_when_supplied(self) -> None:
         ledger = ContributionLedger(
@@ -257,7 +285,7 @@ def test_record_persists_validation_metadata_when_supplied(self) -> None:
             },
         )
 
-    def test_timestamp_preserves_existing_audit_references(self) -> None:
+    def test_record_preserves_explicit_audit_references(self) -> None:
         ledger = ContributionLedger(
             clock=lambda: datetime(2026, 7, 14, 14, 8, 7, tzinfo=UTC)
         )
@@ -268,15 +296,27 @@ def test_timestamp_preserves_existing_audit_references(self) -> None:
             validation_reason="ok",
             job_type="echo",
             version_metadata_ref="version-metadata.json",
+            manifest_ref="manifests/local-job-1.json",
         )
 
         self.assertEqual(record.node_id, "node-a")
         self.assertEqual(record.job_id, "job-1")
         self.assertEqual(record.validation_valid, True)
         self.assertEqual(record.validation_reason, "ok")
         self.assertEqual(record.version_metadata_ref, "version-metadata.json")
+        self.assertEqual(record.manifest_ref, "manifests/local-job-1.json")
         self.assertEqual(record.created_at, "2026-07-14T14:08:07Z")
 
+    def test_non_manifest_record_omits_manifest_reference(self) -> None:
+        ledger = ContributionLedger()
+
+        record = ledger.record(
+            JobResult("job-1", "node-a", "completed", "hello mesh", None, 1)
+        )
+
+        self.assertIsNone(record.manifest_ref)
+        self.assertNotIn("manifest_ref", record.to_dict())
+
     def test_timestamp_rejects_non_utc_or_malformed_persisted_values(self) -> None:
         record = ContributionRecord(
             "node-a", "job-1", "completed", 1, created_at="2026-07-14T14:08:07Z"

tests/test_node_service.py

@@ -21,6 +21,7 @@ def test_successful_echo_assignment_is_processed_from_inbox(self) -> None:
                 "job_id": "echo-1",
                 "job_type": "echo",
                 "payload": {"message": "hello mesh"},
+                "manifest_ref": "manifests/local-batch.json",
             },
         )
         bus.send(assignment)
@@ -37,6 +38,10 @@ def test_successful_echo_assignment_is_processed_from_inbox(self) -> None:
         self.assertEqual(processed.result.output, "hello mesh")
         self.assertTrue(processed.validation.valid)
         self.assertEqual(processed.contribution_record.contribution_units, 1)
+        self.assertEqual(
+            processed.contribution_record.manifest_ref,
+            "manifests/local-batch.json",
+        )
         self.assertEqual(ledger.summary_for_node("node-a").total_contribution_units, 1)
         self.assertEqual(
             [message.message_type for message in processed.emitted_messages],