From 2c0145ee0f4103e9041557a291ec258b4a223f4c Mon Sep 17 00:00:00 2001 From: ExoHayvan <18trevor3695@gmail.com> Date: Thu, 16 Jul 2026 07:47:35 -0400 Subject: [PATCH 1/4] chore: complete phase-1 step-147 --- docs/local-audit-event-schema.md | 6 + src/aethermesh_core/contribution_record.py | 184 ++++++++++++++++----- tests/test_contribution_record.py | 53 ++++++ 3 files changed, 198 insertions(+), 45 deletions(-) diff --git a/docs/local-audit-event-schema.md b/docs/local-audit-event-schema.md index 6f0e94d..feb8209 100644 --- a/docs/local-audit-event-schema.md +++ b/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: diff --git a/src/aethermesh_core/contribution_record.py b/src/aethermesh_core/contribution_record.py index 1f87356..eb7b8a0 100644 --- a/src/aethermesh_core/contribution_record.py +++ b/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" + 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") + ) + if receipt["status"] != "accepted" or receipt["validation_status"] != "pass": + 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, + None, + 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: + validation = document.get("validation") if isinstance(document, dict) else None + outcome = ( + "rejected" + if isinstance(validation, dict) and validation.get("status") == "invalid" + else "validation_failed" + ) + _append_ledger_update_audit(resolved_audit_path, document, None, 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.""" + + contribution = document if isinstance(document, dict) else {} + contributor = contribution.get("contributor_node_id") + if not isinstance(contributor, str) or not contributor: + contributor = "local-ledger" + creator = contribution.get("creator_node_id") + if not isinstance(creator, str) or not creator: + creator = None + update_id = "ledger-update-" + canonical_json_hash( + { + "record_id": contribution.get("record_id"), + "job_id": contribution.get("job_id"), + "validation_receipt_id": contribution.get("validation_receipt_id"), + "outcome": outcome, + } + ).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, + } + for field in ("job_id", "validation_receipt_id"): + value = contribution.get(field) + if isinstance(value, str) and value: + event["work_id" if field == "job_id" else field] = value + if manifest_id is not None: + event["manifest_id"] = manifest_id + manifest_links = contribution.get("manifest_links") + if isinstance(manifest_links, dict): + manifest_ref = manifest_links.get("work_manifest_ref") + if isinstance(manifest_ref, str) and manifest_ref: + event["manifest_ref"] = manifest_ref + validation = contribution.get("validation") + if isinstance(validation, dict): + receipt_ref = validation.get("validation_receipt_ref") + if isinstance(receipt_ref, str) and receipt_ref: + event["validation_receipt_ref"] = receipt_ref + lineage = contribution.get("lineage") + if isinstance(lineage, dict) and isinstance( + lineage.get("parent_contribution_ids"), list + ): + event["lineage_parent_ids"] = lineage["parent_contribution_ids"] + record_id = contribution.get("record_id") + if isinstance(record_id, str) and record_id: + event["contribution_attribution_ids"] = [record_id] + append_local_audit_event(audit_path, event) def new_unvalidated_validation() -> dict[str, Any]: diff --git a/tests/test_contribution_record.py b/tests/test_contribution_record.py index f486a8e..98d05c6 100644 --- a/tests/test_contribution_record.py +++ b/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,30 @@ 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"], + ) + 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") def test_duplicate_validated_manifest_is_not_recorded_twice(self) -> None: with TemporaryDirectory() as directory: From cf0e86eff94bc1997414f1cd342a002d7bb54be1 Mon Sep 17 00:00:00 2001 From: ExoHayvan <18trevor3695@gmail.com> Date: Thu, 16 Jul 2026 08:20:39 -0400 Subject: [PATCH 2/4] fix: address automated review feedback --- src/aethermesh_core/contribution_record.py | 38 ++++++++++++++-------- tests/test_contribution_record.py | 12 +++++++ 2 files changed, 36 insertions(+), 14 deletions(-) diff --git a/src/aethermesh_core/contribution_record.py b/src/aethermesh_core/contribution_record.py index eb7b8a0..25d502c 100644 --- a/src/aethermesh_core/contribution_record.py +++ b/src/aethermesh_core/contribution_record.py @@ -20,7 +20,10 @@ JobResultSchemaError, validate_job_result_document, ) -from aethermesh_core.local_audit_event import append_local_audit_event +from aethermesh_core.local_audit_event import ( + LocalAuditEventError, + append_local_audit_event, +) from aethermesh_core.local_json_helpers import canonical_json_hash from aethermesh_core.validation_receipt_schema import ( ValidationReceiptSchemaError, @@ -228,6 +231,7 @@ def record_validated_contribution( resolved_audit_path = audit_path or journal_path.with_name( "contribution-ledger-updates.jsonl" ) + rejected_evidence = False try: contribution = validate_local_contribution_record(document, local_root) receipt_ref = cast(str, contribution["validation"]["validation_receipt_ref"]) @@ -235,6 +239,7 @@ def record_validated_contribution( _read_local_json(local_root, receipt_ref, "validation receipt") ) if receipt["status"] != "accepted" or receipt["validation_status"] != "pass": + rejected_evidence = True raise ContributionRecordError( "contribution recording requires an accepted passed validation receipt" ) @@ -295,12 +300,7 @@ def record_validated_contribution( ) return entry except ContributionRecordError: - validation = document.get("validation") if isinstance(document, dict) else None - outcome = ( - "rejected" - if isinstance(validation, dict) and validation.get("status") == "invalid" - else "validation_failed" - ) + outcome = "rejected" if rejected_evidence else "validation_failed" _append_ledger_update_audit(resolved_audit_path, document, None, clock, outcome) raise @@ -316,10 +316,10 @@ def _append_ledger_update_audit( contribution = document if isinstance(document, dict) else {} contributor = contribution.get("contributor_node_id") - if not isinstance(contributor, str) or not contributor: + 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: + if not isinstance(creator, str) or not creator.strip(): creator = None update_id = "ledger-update-" + canonical_json_hash( { @@ -341,19 +341,19 @@ def _append_ledger_update_audit( } for field in ("job_id", "validation_receipt_id"): value = contribution.get(field) - if isinstance(value, str) and value: + if isinstance(value, str) and value.strip(): event["work_id" if field == "job_id" else field] = value if manifest_id is not None: event["manifest_id"] = manifest_id manifest_links = contribution.get("manifest_links") if isinstance(manifest_links, dict): manifest_ref = manifest_links.get("work_manifest_ref") - if isinstance(manifest_ref, str) and manifest_ref: + if isinstance(manifest_ref, str) and manifest_ref.strip(): event["manifest_ref"] = manifest_ref validation = contribution.get("validation") if isinstance(validation, dict): receipt_ref = validation.get("validation_receipt_ref") - if isinstance(receipt_ref, str) and receipt_ref: + if isinstance(receipt_ref, str) and receipt_ref.strip(): event["validation_receipt_ref"] = receipt_ref lineage = contribution.get("lineage") if isinstance(lineage, dict) and isinstance( @@ -361,9 +361,19 @@ def _append_ledger_update_audit( ): event["lineage_parent_ids"] = lineage["parent_contribution_ids"] record_id = contribution.get("record_id") - if isinstance(record_id, str) and record_id: + if isinstance(record_id, str) and record_id.strip(): event["contribution_attribution_ids"] = [record_id] - append_local_audit_event(audit_path, event) + try: + append_local_audit_event(audit_path, event) + except LocalAuditEventError: + # Invalid contribution evidence must not make its compact failure event invalid. + for field in ( + "manifest_ref", + "validation_receipt_ref", + "lineage_parent_ids", + ): + event.pop(field, None) + append_local_audit_event(audit_path, event) def new_unvalidated_validation() -> dict[str, Any]: diff --git a/tests/test_contribution_record.py b/tests/test_contribution_record.py index 98d05c6..1c1475d 100644 --- a/tests/test_contribution_record.py +++ b/tests/test_contribution_record.py @@ -232,6 +232,18 @@ def test_failed_or_missing_validation_records_nothing(self) -> None: self.assertIsNone(fallback_audit["creator_node_id"]) self.assertEqual(fallback_audit["validation_status"], "validation_failed") + 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: journal_path = Path(directory) / "contributions.jsonl" From 6a4be460cc8f2d6907c22aa412a6902aab458bd3 Mon Sep 17 00:00:00 2001 From: ExoHayvan <18trevor3695@gmail.com> Date: Thu, 16 Jul 2026 08:31:31 -0400 Subject: [PATCH 3/4] fix: address automated review feedback --- src/aethermesh_core/contribution_record.py | 25 +++++++++++++--------- tests/test_contribution_record.py | 23 ++++++++++++++++++++ 2 files changed, 38 insertions(+), 10 deletions(-) diff --git a/src/aethermesh_core/contribution_record.py b/src/aethermesh_core/contribution_record.py index 25d502c..a892cc1 100644 --- a/src/aethermesh_core/contribution_record.py +++ b/src/aethermesh_core/contribution_record.py @@ -232,12 +232,14 @@ def record_validated_contribution( "contribution-ledger-updates.jsonl" ) 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( @@ -266,7 +268,7 @@ def record_validated_contribution( _append_ledger_update_audit( resolved_audit_path, contribution, - None, + manifest_id, clock, "already_recorded", ) @@ -301,7 +303,9 @@ def record_validated_contribution( return entry except ContributionRecordError: outcome = "rejected" if rejected_evidence else "validation_failed" - _append_ledger_update_audit(resolved_audit_path, document, None, clock, outcome) + _append_ledger_update_audit( + resolved_audit_path, document, audit_manifest_id, clock, outcome + ) raise @@ -321,14 +325,15 @@ def _append_ledger_update_audit( creator = contribution.get("creator_node_id") if not isinstance(creator, str) or not creator.strip(): creator = None - update_id = "ledger-update-" + canonical_json_hash( - { - "record_id": contribution.get("record_id"), - "job_id": contribution.get("job_id"), - "validation_receipt_id": contribution.get("validation_receipt_id"), - "outcome": outcome, - } - ).removeprefix("sha256:") + 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, diff --git a/tests/test_contribution_record.py b/tests/test_contribution_record.py index 1c1475d..78e809f 100644 --- a/tests/test_contribution_record.py +++ b/tests/test_contribution_record.py @@ -221,6 +221,10 @@ def test_failed_or_missing_validation_records_nothing(self) -> None: 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( @@ -232,6 +236,16 @@ def test_failed_or_missing_validation_records_nothing(self) -> None: self.assertIsNone(fallback_audit["creator_node_id"]) self.assertEqual(fallback_audit["validation_status"], "validation_failed") + non_json = {"record_id": {"not-json"}} + 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") + unsafe_rejected = copy.deepcopy(self.failed) unsafe_rejected["validation"]["validation_receipt_ref"] = "../../secret" with self.assertRaisesRegex(ContributionRecordError, "safe local"): @@ -256,6 +270,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 From b1f305e025b3c3b4fb7a5affced9ac5454e7efba Mon Sep 17 00:00:00 2001 From: ExoHayvan <18trevor3695@gmail.com> Date: Thu, 16 Jul 2026 08:47:24 -0400 Subject: [PATCH 4/4] fix: address automated review feedback --- src/aethermesh_core/contribution_record.py | 61 ++++++++-------------- src/aethermesh_core/local_audit_event.py | 20 +++++++ tests/test_contribution_record.py | 12 ++++- tests/test_local_audit_event.py | 18 +++++++ 4 files changed, 72 insertions(+), 39 deletions(-) diff --git a/src/aethermesh_core/contribution_record.py b/src/aethermesh_core/contribution_record.py index a892cc1..4eeaeae 100644 --- a/src/aethermesh_core/contribution_record.py +++ b/src/aethermesh_core/contribution_record.py @@ -20,10 +20,7 @@ JobResultSchemaError, validate_job_result_document, ) -from aethermesh_core.local_audit_event import ( - LocalAuditEventError, - append_local_audit_event, -) +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, @@ -318,7 +315,12 @@ def _append_ledger_update_audit( ) -> None: """Append compact local evidence, never raw work payloads or credentials.""" - contribution = document if isinstance(document, dict) else {} + # 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" @@ -344,41 +346,24 @@ def _append_ledger_update_audit( "local_run_id": update_id, "validation_status": outcome, } - for field in ("job_id", "validation_receipt_id"): - value = contribution.get(field) - if isinstance(value, str) and value.strip(): - event["work_id" if field == "job_id" else field] = value + 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 - manifest_links = contribution.get("manifest_links") - if isinstance(manifest_links, dict): - manifest_ref = manifest_links.get("work_manifest_ref") - if isinstance(manifest_ref, str) and manifest_ref.strip(): - event["manifest_ref"] = manifest_ref - validation = contribution.get("validation") - if isinstance(validation, dict): - receipt_ref = validation.get("validation_receipt_ref") - if isinstance(receipt_ref, str) and receipt_ref.strip(): - event["validation_receipt_ref"] = receipt_ref - lineage = contribution.get("lineage") - if isinstance(lineage, dict) and isinstance( - lineage.get("parent_contribution_ids"), list - ): - event["lineage_parent_ids"] = lineage["parent_contribution_ids"] - record_id = contribution.get("record_id") - if isinstance(record_id, str) and record_id.strip(): - event["contribution_attribution_ids"] = [record_id] - try: - append_local_audit_event(audit_path, event) - except LocalAuditEventError: - # Invalid contribution evidence must not make its compact failure event invalid. - for field in ( - "manifest_ref", - "validation_receipt_ref", - "lineage_parent_ids", - ): - event.pop(field, None) - append_local_audit_event(audit_path, event) + append_local_audit_event(audit_path, event) def new_unvalidated_validation() -> dict[str, Any]: diff --git a/src/aethermesh_core/local_audit_event.py b/src/aethermesh_core/local_audit_event.py index db8c3f7..13b5360 100644 --- a/src/aethermesh_core/local_audit_event.py +++ b/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.""" diff --git a/tests/test_contribution_record.py b/tests/test_contribution_record.py index 78e809f..46087ff 100644 --- a/tests/test_contribution_record.py +++ b/tests/test_contribution_record.py @@ -236,7 +236,13 @@ def test_failed_or_missing_validation_records_nothing(self) -> None: self.assertIsNone(fallback_audit["creator_node_id"]) self.assertEqual(fallback_audit["validation_status"], "validation_failed") - non_json = {"record_id": {"not-json"}} + 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( @@ -245,6 +251,10 @@ def test_failed_or_missing_validation_records_nothing(self) -> None: .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" diff --git a/tests/test_local_audit_event.py b/tests/test_local_audit_event.py index 49e05b8..e774274 100644 --- a/tests/test_local_audit_event.py +++ b/tests/test_local_audit_event.py @@ -84,6 +84,24 @@ def test_rejects_invalid_optional_references(self) -> None: with self.assertRaisesRegex(LocalAuditEventError, message): validate_local_audit_event({**_minimal_event(), **optional_fields}) + def test_contribution_update_requires_a_supported_outcome(self) -> None: + event = { + **_minimal_event(), + "event_type": "contribution_record_updated", + "validation_status": "recorded", + } + self.assertEqual(validate_local_audit_event(event), event) + with self.assertRaisesRegex(LocalAuditEventError, "missing validation_status"): + validate_local_audit_event( + { + key: value + for key, value in event.items() + if key != "validation_status" + } + ) + with self.assertRaisesRegex(LocalAuditEventError, "unsupported"): + validate_local_audit_event({**event, "validation_status": "rewarded"}) + def test_job_submitted_requires_compact_submission_evidence(self) -> None: event = { **_minimal_event(),