chore: 147 - Log contribution ledger update

docs/local-audit-event-schema.md

@@ -51,6 +51,12 @@ Startup appends one event per capability advertisement. Normal repeat startup is
 
 A `validation_receipt_created` event is appended to `data/audit/validation-receipt-creations.jsonl` only after its local validation receipt is durably created. It requires the work and manifest references, receipt ID and reference, accepted or rejected validation result, validator node and validator name, lineage references, and available contribution attribution. The lineage references include the submitted-work manifest and any parent references already present in that manifest. A rejected receipt remains explicitly `rejected`; this log does not imply a successful validation, production security, or network consensus. If this append fails, receipt creation fails clearly rather than reporting the job as complete.
 
+## Contribution ledger update events
+
+`record_validated_contribution` appends exactly one `contribution_record_updated` event for each local attribution attempt beside its contribution journal, unless an explicit audit path is supplied. The event records the creator as known, the contributing node as the actor, work ID, validation receipt ID, available manifest ID, lineage parent IDs, and contribution record ID. Its deterministic `event_id`/`local_run_id` is derived from compact identifiers and the local outcome, making repeated attempts traceable without copying a prompt, result payload, credentials, or other sensitive data.
+
+`validation_status` is deliberately limited to `recorded`, `already_recorded`, `rejected`, or `validation_failed`. These are local evidence states only; they do not assert consensus, finality, reputation, or a reward value. A rejected receipt or invalid evidence still receives an audit event, while only an accepted passed receipt can enter the contribution journal.
+
 ## Examples
 
 A minimal initialization event intentionally has no optional references:

src/aethermesh_core/contribution_record.py

@@ -20,6 +20,7 @@
     JobResultSchemaError,
     validate_job_result_document,
 )
+from aethermesh_core.local_audit_event import append_local_audit_event
 from aethermesh_core.local_json_helpers import canonical_json_hash
 from aethermesh_core.validation_receipt_schema import (
     ValidationReceiptSchemaError,
@@ -220,56 +221,149 @@ def record_validated_contribution(
     journal_path: Path,
     *,
     clock: Callable[[], datetime] = lambda: datetime.now(UTC),
+    audit_path: Path | None = None,
 ) -> dict[str, Any]:
-    """Append passed receipt-backed attribution once per work manifest."""
+    """Append receipt-backed attribution and one local audit event per attempt."""
 
-    contribution = validate_local_contribution_record(document, local_root)
-    receipt_ref = cast(str, contribution["validation"]["validation_receipt_ref"])
-    receipt = validate_validation_receipt_document(
-        _read_local_json(local_root, receipt_ref, "validation receipt")
+    resolved_audit_path = audit_path or journal_path.with_name(
+        "contribution-ledger-updates.jsonl"
     )
-    if receipt["status"] != "accepted" or receipt["validation_status"] != "pass":
-        raise ContributionRecordError(
-            "contribution recording requires an accepted passed validation receipt"
+    rejected_evidence = False
+    audit_manifest_id: str | None = None
+    try:
+        contribution = validate_local_contribution_record(document, local_root)
+        receipt_ref = cast(str, contribution["validation"]["validation_receipt_ref"])
+        receipt = validate_validation_receipt_document(
+            _read_local_json(local_root, receipt_ref, "validation receipt")
+        )
+        audit_manifest_id = receipt["manifest_id"]
+        if receipt["status"] != "accepted" or receipt["validation_status"] != "pass":
+            rejected_evidence = True
+            raise ContributionRecordError(
+                "contribution recording requires an accepted passed validation receipt"
+            )
+        work_manifest_ref = cast(
+            str, contribution["manifest_links"]["work_manifest_ref"]
+        )
+        work_manifest = cast(
+            dict[str, Any],
+            _read_local_json(local_root, work_manifest_ref, "work manifest"),
+        )
+        if (
+            canonical_json_hash(work_manifest, prefix="sha256:")
+            != receipt["manifest_id"]
+        ):
+            raise ContributionRecordError(
+                "validation receipt manifest_id does not match work manifest"
+            )
+
+        with _contribution_journal_lock(journal_path):
+            entries = _load_contribution_journal(journal_path)
+            manifest_id = receipt["manifest_id"]
+            for existing_entry in entries:
+                if existing_entry["manifest_id"] == manifest_id:
+                    _append_ledger_update_audit(
+                        resolved_audit_path,
+                        contribution,
+                        manifest_id,
+                        clock,
+                        "already_recorded",
+                    )
+                    return existing_entry
+
+            entry: dict[str, Any] = {
+                "entry_version": LOCAL_CONTRIBUTION_JOURNAL_ENTRY_VERSION,
+                "recorded_at": _utc_timestamp(clock()),
+                "manifest_id": manifest_id,
+                "creator_node_id": contribution["creator_node_id"],
+                "contributor_node_id": contribution["contributor_node_id"],
+                "validator_node_id": receipt["validator_id"],
+                "validation_receipt_id": contribution["validation_receipt_id"],
+                "lineage_parent_contribution_ids": contribution["lineage"][
+                    "parent_contribution_ids"
+                ],
+                "contribution": contribution,
+                "prior_entry_hash": entries[-1]["entry_hash"] if entries else None,
+            }
+            entry["entry_hash"] = _journal_entry_hash(entry)
+            try:
+                with journal_path.open("a", encoding="utf-8") as handle:
+                    handle.write(json.dumps(entry, sort_keys=True))
+                    handle.write("\n")
+            except OSError as exc:
+                raise ContributionRecordError(
+                    "contribution journal is unwritable"
+                ) from exc
+        _append_ledger_update_audit(
+            resolved_audit_path, contribution, manifest_id, clock, "recorded"
         )
-    work_manifest_ref = cast(str, contribution["manifest_links"]["work_manifest_ref"])
-    work_manifest = cast(
-        dict[str, Any], _read_local_json(local_root, work_manifest_ref, "work manifest")
-    )
-    if canonical_json_hash(work_manifest, prefix="sha256:") != receipt["manifest_id"]:
-        raise ContributionRecordError(
-            "validation receipt manifest_id does not match work manifest"
-        )
-
-    with _contribution_journal_lock(journal_path):
-        entries = _load_contribution_journal(journal_path)
-        manifest_id = receipt["manifest_id"]
-        for existing_entry in entries:
-            if existing_entry["manifest_id"] == manifest_id:
-                return existing_entry
-
-        entry: dict[str, Any] = {
-            "entry_version": LOCAL_CONTRIBUTION_JOURNAL_ENTRY_VERSION,
-            "recorded_at": _utc_timestamp(clock()),
-            "manifest_id": manifest_id,
-            "creator_node_id": contribution["creator_node_id"],
-            "contributor_node_id": contribution["contributor_node_id"],
-            "validator_node_id": receipt["validator_id"],
-            "validation_receipt_id": contribution["validation_receipt_id"],
-            "lineage_parent_contribution_ids": contribution["lineage"][
-                "parent_contribution_ids"
-            ],
-            "contribution": contribution,
-            "prior_entry_hash": entries[-1]["entry_hash"] if entries else None,
-        }
-        entry["entry_hash"] = _journal_entry_hash(entry)
-        try:
-            with journal_path.open("a", encoding="utf-8") as handle:
-                handle.write(json.dumps(entry, sort_keys=True))
-                handle.write("\n")
-        except OSError as exc:
-            raise ContributionRecordError("contribution journal is unwritable") from exc
         return entry
+    except ContributionRecordError:
+        outcome = "rejected" if rejected_evidence else "validation_failed"
+        _append_ledger_update_audit(
+            resolved_audit_path, document, audit_manifest_id, clock, outcome
+        )
+        raise
+
+
+def _append_ledger_update_audit(
+    audit_path: Path,
+    document: object,
+    manifest_id: str | None,
+    clock: Callable[[], datetime],
+    outcome: str,
+) -> None:
+    """Append compact local evidence, never raw work payloads or credentials."""
+
+    # Only copy attribution from a schema-valid record. This helper also handles
+    # malformed input, whose arbitrary strings must not become durable audit data.
+    try:
+        contribution = validate_contribution_record(document)
+    except ContributionRecordError:
+        contribution = {}
+    contributor = contribution.get("contributor_node_id")
+    if not isinstance(contributor, str) or not contributor.strip():
+        contributor = "local-ledger"
+    creator = contribution.get("creator_node_id")
+    if not isinstance(creator, str) or not creator.strip():
+        creator = None
+    update_identity = {
+        field: value if isinstance(value, str) and value.strip() else None
+        for field in ("record_id", "job_id", "validation_receipt_id")
+        for value in (contribution.get(field),)
+    }
+    update_identity["outcome"] = outcome
+    update_id = "ledger-update-" + canonical_json_hash(update_identity).removeprefix(
+        "sha256:"
+    )
+    event: dict[str, Any] = {
+        "schema_version": 1,
+        "event_id": update_id,
+        "timestamp": _utc_timestamp(clock()),
+        "event_type": "contribution_record_updated",
+        "actor_node_id": contributor,
+        "creator_node_id": creator,
+        "local_run_id": update_id,
+        "validation_status": outcome,
+    }
+    if contribution:
+        event.update(
+            {
+                "work_id": contribution["job_id"],
+                "validation_receipt_id": contribution["validation_receipt_id"],
+                "manifest_ref": contribution["manifest_links"]["work_manifest_ref"],
+                "validation_receipt_ref": contribution["validation"][
+                    "validation_receipt_ref"
+                ],
+                "lineage_parent_ids": contribution["lineage"][
+                    "parent_contribution_ids"
+                ],
+                "contribution_attribution_ids": [contribution["record_id"]],
+            }
+        )
+    if manifest_id is not None:
+        event["manifest_id"] = manifest_id
+    append_local_audit_event(audit_path, event)
 
 
 def new_unvalidated_validation() -> dict[str, Any]:

src/aethermesh_core/local_audit_event.py

@@ -167,6 +167,8 @@ def validate_local_audit_event(event: Mapping[str, Any]) -> dict[str, Any]:
         _validate_result_reported_event(document)
     if document["event_type"] == "validation_receipt_created":
         _validate_validation_receipt_created_event(document)
+    if document["event_type"] == "contribution_record_updated":
+        _validate_contribution_record_updated_event(document)
     return document
 
 
@@ -309,6 +311,24 @@ def _validate_capability_advertisement_event(document: Mapping[str, Any]) -> Non
     )
 
 
+def _validate_contribution_record_updated_event(document: Mapping[str, Any]) -> None:
+    """Limit ledger updates to the documented local evidence outcomes."""
+
+    if "validation_status" not in document:
+        raise LocalAuditEventError(
+            "contribution_record_updated event is missing validation_status"
+        )
+    if document["validation_status"] not in {
+        "recorded",
+        "already_recorded",
+        "rejected",
+        "validation_failed",
+    }:
+        raise LocalAuditEventError(
+            "contribution_record_updated.validation_status is unsupported"
+        )
+
+
 def _validate_job_submission_event(document: Mapping[str, Any]) -> None:
     """Require compact, non-payload evidence for one accepted local submission."""
 

tests/test_contribution_record.py

@@ -146,6 +146,34 @@ def test_records_one_accepted_contribution_with_exact_evidence_and_attribution(
             self.assertEqual(
                 len(journal_path.read_text(encoding="utf-8").splitlines()), 1
             )
+            audit_entries = [
+                json.loads(line)
+                for line in (journal_path.parent / "contribution-ledger-updates.jsonl")
+                .read_text(encoding="utf-8")
+                .splitlines()
+            ]
+            self.assertEqual(len(audit_entries), 1)
+            audit = audit_entries[0]
+            self.assertEqual(audit["validation_status"], "recorded")
+            self.assertEqual(audit["creator_node_id"], "node.local-creator")
+            self.assertEqual(audit["actor_node_id"], "node.local-worker")
+            self.assertEqual(audit["manifest_id"], entry["manifest_id"])
+            self.assertEqual(
+                audit["manifest_ref"],
+                "examples/job-envelopes/minimal-local-echo.json",
+            )
+            self.assertEqual(
+                audit["validation_receipt_ref"],
+                "examples/validation-receipts/local-echo-pass.json",
+            )
+            self.assertEqual(
+                audit["validation_receipt_id"], entry["validation_receipt_id"]
+            )
+            self.assertEqual(audit["lineage_parent_ids"], ["contribution.parent-001"])
+            self.assertEqual(
+                audit["contribution_attribution_ids"], [accepted["record_id"]]
+            )
+            self.assertEqual(audit["event_id"], audit["local_run_id"])
 
     def test_failed_or_missing_validation_records_nothing(self) -> None:
         with TemporaryDirectory() as directory:
@@ -160,6 +188,7 @@ def test_failed_or_missing_validation_records_nothing(self) -> None:
             missing = copy.deepcopy(self.minimal)
             missing["validation"]["validation_receipt_ref"] = None
             missing["manifest_links"]["validation_manifest_ref"] = None
+            missing["manifest_links"]["work_manifest_ref"] = None
             with self.assertRaisesRegex(ContributionRecordError, "receipt reference"):
                 record_validated_contribution(missing, self.root, journal_path)
             self.assertFalse(journal_path.exists())
@@ -178,6 +207,66 @@ def test_failed_or_missing_validation_records_nothing(self) -> None:
             ):
                 record_validated_contribution(malformed, self.root, journal_path)
             self.assertFalse(journal_path.exists())
+            audit_entries = [
+                json.loads(line)
+                for line in (Path(directory) / "contribution-ledger-updates.jsonl")
+                .read_text(encoding="utf-8")
+                .splitlines()
+            ]
+            self.assertEqual(
+                [entry["validation_status"] for entry in audit_entries],
+                ["rejected", "validation_failed", "validation_failed"],
+            )
+            self.assertEqual(
+                audit_entries[0]["validation_receipt_id"],
+                self.failed["validation_receipt_id"],
+            )
+            self.assertEqual(
+                audit_entries[0]["manifest_id"],
+                "sha256:" + "d" * 64,
+            )
+            with self.assertRaisesRegex(ContributionRecordError, "missing"):
+                record_validated_contribution({}, self.root, journal_path)
+            fallback_audit = json.loads(
+                (Path(directory) / "contribution-ledger-updates.jsonl")
+                .read_text(encoding="utf-8")
+                .splitlines()[-1]
+            )
+            self.assertEqual(fallback_audit["actor_node_id"], "local-ledger")
+            self.assertIsNone(fallback_audit["creator_node_id"])
+            self.assertEqual(fallback_audit["validation_status"], "validation_failed")
+
+            non_json = {
+                "record_id": {"not-json"},
+                "creator_node_id": "secret-creator",
+                "contributor_node_id": "secret-contributor",
+                "job_id": "secret-job",
+                "validation_receipt_id": "secret-receipt",
+            }
+            with self.assertRaisesRegex(ContributionRecordError, "missing"):
+                record_validated_contribution(non_json, self.root, journal_path)
+            non_json_audit = json.loads(
+                (Path(directory) / "contribution-ledger-updates.jsonl")
+                .read_text(encoding="utf-8")
+                .splitlines()[-1]
+            )
+            self.assertEqual(non_json_audit["validation_status"], "validation_failed")
+            self.assertEqual(non_json_audit["actor_node_id"], "local-ledger")
+            self.assertIsNone(non_json_audit["creator_node_id"])
+            self.assertNotIn("work_id", non_json_audit)
+            self.assertNotIn("validation_receipt_id", non_json_audit)
+
+            unsafe_rejected = copy.deepcopy(self.failed)
+            unsafe_rejected["validation"]["validation_receipt_ref"] = "../../secret"
+            with self.assertRaisesRegex(ContributionRecordError, "safe local"):
+                record_validated_contribution(unsafe_rejected, self.root, journal_path)
+            unsafe_audit = json.loads(
+                (Path(directory) / "contribution-ledger-updates.jsonl")
+                .read_text(encoding="utf-8")
+                .splitlines()[-1]
+            )
+            self.assertEqual(unsafe_audit["validation_status"], "validation_failed")
+            self.assertNotIn("validation_receipt_ref", unsafe_audit)
 
     def test_duplicate_validated_manifest_is_not_recorded_twice(self) -> None:
         with TemporaryDirectory() as directory:
@@ -191,6 +280,15 @@ def test_duplicate_validated_manifest_is_not_recorded_twice(self) -> None:
             self.assertEqual(
                 len(journal_path.read_text(encoding="utf-8").splitlines()), 1
             )
+            audit_entries = [
+                json.loads(line)
+                for line in (Path(directory) / "contribution-ledger-updates.jsonl")
+                .read_text(encoding="utf-8")
+                .splitlines()
+            ]
+            self.assertEqual(len(audit_entries), 2)
+            self.assertEqual(audit_entries[1]["validation_status"], "already_recorded")
+            self.assertEqual(audit_entries[1]["manifest_id"], first["manifest_id"])
 
             prior = json.loads(journal_path.read_text(encoding="utf-8"))
             prior["manifest_id"] = "sha256:" + "b" * 64