Fix durable result receipts

.github/workflows/ingest.yml

@@ -24,7 +24,7 @@ jobs:
       - uses: actions/setup-python@v5
         with:
           python-version: '3.12'
-      - name: Drain labelled issue inbox
+      - name: Stage labelled issue inbox
         env:
           GH_TOKEN: ${{ github.token }}
         run: |
@@ -36,26 +36,36 @@ jobs:
             number=$(jq -r .number .inbox/issue.json)
             account=$(jq -r .author.login .inbox/issue.json)
             jq -r .body .inbox/issue.json > .inbox/body.md
-            if python scripts/extract_envelope.py --body .inbox/body.md --output .inbox/envelope.json && python scripts/ingest.py --root . --account "$account" --input .inbox/envelope.json > .inbox/outcome.txt 2>&1; then
-              printf 'Processed result batch:\n\n```text\n' > .inbox/comment.md
-              cat .inbox/outcome.txt >> .inbox/comment.md
-              printf '\n```\n' >> .inbox/comment.md
+            : > .inbox/outcome.txt
+            if python scripts/extract_envelope.py --body .inbox/body.md --output .inbox/envelope.json >> .inbox/outcome.txt 2>&1 && python scripts/ingest.py --root . --account "$account" --input .inbox/envelope.json >> .inbox/outcome.txt 2>&1; then
+              printf 'Processed result batch:\n\n```text\n' > ".inbox/comment-$number.md"
+              cat .inbox/outcome.txt >> ".inbox/comment-$number.md"
+              printf '\n```\n' >> ".inbox/comment-$number.md"
             else
-              printf 'Rejected result batch before individual records could be read:\n\n```text\n' > .inbox/comment.md
-              cat .inbox/outcome.txt >> .inbox/comment.md
-              printf '\n```\n' >> .inbox/comment.md
+              printf 'Rejected result batch before individual records could be read:\n\n```text\n' > ".inbox/comment-$number.md"
+              cat .inbox/outcome.txt >> ".inbox/comment-$number.md"
+              printf '\n```\n' >> ".inbox/comment-$number.md"
             fi
-            gh issue comment "$number" --repo "$GITHUB_REPOSITORY" --body-file .inbox/comment.md
-            gh issue close "$number" --repo "$GITHUB_REPOSITORY"
+            printf '%s\n' "$number" >> .inbox/processed-issues
           done < .inbox/issues
       - name: Regenerate projections and commit accepted facts
         run: |
           python scripts/aggregate.py --root .
           git config user.name 'github-actions[bot]'
           git config user.email '41898282+github-actions[bot]@users.noreply.github.com'
-          rm -rf .inbox
           for path in results leaderboard matchmaking clients.json site/data; do
             [ -e "$path" ] && git add -- "$path"
           done
           git diff --cached --quiet || git commit -m 'chore: ingest Rumble result batch'
           git push
+      - name: Publish durable receipts
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          if [ -f .inbox/processed-issues ]; then
+            while read -r number; do
+              gh issue comment "$number" --repo "$GITHUB_REPOSITORY" --body-file ".inbox/comment-$number.md"
+              gh issue close "$number" --repo "$GITHUB_REPOSITORY"
+            done < .inbox/processed-issues
+          fi
+          rm -rf .inbox

CONTRIBUTING.md

@@ -10,7 +10,7 @@ Create an issue titled `[result] <client-id> <UTC timestamp>`, apply the `result
 
 Each participant supplies cataloged `name` and `version`, `isTeam`, a 1224-system `rank`, `totalScore`, `survival`, `lastSurvivorBonus`, `bulletDamage`, `bulletKillBonus`, `ramDamage`, `ramKillBonus`, `firstPlaces`, `secondPlaces`, and `thirdPlaces`. Scores and place counts are non-negative signed 32-bit integers. `isTeam`, result-entry count, rank placement, and place-count totals must match the selected ranked game type.
 
-The submitting issue account and `clientId` must be registered. Every bot must be active in `catalog.json`; the hourly synchronization workflow copies the reviewed Rumble bot catalog declared by that file. The drain workflow validates each record independently, keeps valid records when neighboring records are rejected, and closes every processed issue with a receipt line for every record.
+The submitting issue account and `clientId` must be registered. Every bot must be active in `catalog.json`; the hourly synchronization workflow copies the reviewed Rumble bot catalog declared by that file. The drain workflow validates each record independently, keeps valid records when neighboring records are rejected, pushes accepted facts, and then closes every processed issue with a receipt line for every record. Retrying an identical retained result returns the same successful outcome without creating another fact; reusing a battle ID for different content is rejected.
 
 ## Rules
 

README.md

@@ -12,7 +12,7 @@ Run the focused suite with `python -m unittest discover -s tests -v`. It uses on
 
 Before a client can submit ranked results, its forge account must be registered through a reviewed pull request adding `clients/<account>.json`. The client then creates an issue labelled `result-submission`, with a `[result]` title and exactly one fenced JSON batch envelope. See [CONTRIBUTING.md](CONTRIBUTING.md) for the contract and limits.
 
-Issue bodies are transport receipts, never durable storage. The only authoritative accepted result is a content-addressed JSON fact under `results/raw/`; projections are disposable and reproducible.
+Issue bodies are transport, never durable storage. The only authoritative accepted result is a content-addressed JSON fact under `results/raw/`; projections are disposable and reproducible. Successful receipt comments are published only after accepted facts are pushed, and identical retries are acknowledged idempotently.
 
 ## Forking and operations
 

scripts/ingest.py

@@ -6,8 +6,8 @@
 import argparse
 from pathlib import Path
 
-from common import read_json, write_json
-from validate import AcceptedResult, ValidationError, ValidationOutcome, validate_envelope
+from common import content_hash, read_json, write_json
+from validate import SCHEMA_VERSION, AcceptedResult, ValidationError, ValidationOutcome, validate_envelope
 
 
 def raw_path(root: Path, completed_at: str, digest: str) -> Path:
@@ -18,16 +18,53 @@ def raw_path(root: Path, completed_at: str, digest: str) -> Path:
     return root / "results" / "raw" / date[0] / date[1] / f"{digest}.json"
 
 
-def retained_values(root: Path, field: str) -> set[str]:
-    """Return one field's values from every retained raw fact and rollup record."""
-    result: set[str] = set()
+def retained_results(root: Path) -> dict[str, dict]:
+    """Return retained facts indexed by battle ID."""
+    result: dict[str, dict] = {}
     for path in (root / "results" / "raw").rglob("*.json") if (root / "results" / "raw").exists() else []:
-        result.add(str(read_json(path).get(field)))
+        record = read_json(path)
+        result[str(record.get("battleId"))] = record
     for path in (root / "results" / "rollups").rglob("*.json") if (root / "results" / "rollups").exists() else []:
-        result.update(str(record.get(field)) for record in read_json(path).get("results", []))
+        for record in read_json(path).get("results", []):
+            result[str(record.get("battleId"))] = record
     return result
 
 
+def validate_with_idempotent_retries(root: Path, envelope: object, *, account: str,
+                                     retained: dict[str, dict]) -> list[ValidationOutcome]:
+    """Validate new records while recovering successes for identical retained facts."""
+    if not isinstance(envelope, dict) or not isinstance(envelope.get("results"), list):
+        return validate_envelope(root, envelope, account=account)
+    recovered: dict[int, ValidationOutcome] = {}
+    pending_records: list[object] = []
+    pending_indexes: list[int] = []
+    recoverable_envelope = envelope.get("schemaVersion") == SCHEMA_VERSION and 0 < len(envelope["results"]) <= 60
+    for index, record in enumerate(envelope["results"]):
+        battle_id = str(record.get("battleId")) if isinstance(record, dict) and record.get("battleId") is not None else None
+        existing = retained.get(battle_id) if battle_id is not None else None
+        client = record.get("client") if isinstance(record, dict) else None
+        identical = (
+            recoverable_envelope
+            and existing is not None
+            and existing.get("submittedBy") == account
+            and content_hash(record) == existing.get("payloadHash")
+            and isinstance(client, dict)
+            and envelope.get("clientId") == client.get("id")
+            and envelope.get("clientVersion") == client.get("version")
+        )
+        if identical:
+            accepted = AcceptedResult(existing, content_hash(existing))
+            recovered[index] = ValidationOutcome(index=index, battle_id=battle_id, accepted=accepted)
+        else:
+            pending_indexes.append(index)
+            pending_records.append(record)
+    if pending_records:
+        pending_envelope = envelope | {"results": pending_records}
+        for index, outcome in zip(pending_indexes, validate_envelope(root, pending_envelope, account=account)):
+            recovered[index] = ValidationOutcome(index=index, battle_id=outcome.battle_id, accepted=outcome.accepted, error=outcome.error)
+    return [recovered[index] for index in range(len(envelope["results"]))]
+
+
 def format_outcome(outcome: ValidationOutcome) -> str:
     """Render one stable receipt line for a submitted record."""
     label = outcome.battle_id or f"result[{outcome.index}]"
@@ -36,29 +73,34 @@ def format_outcome(outcome: ValidationOutcome) -> str:
 
 def ingest(root: Path, envelope: object, *, account: str) -> list[ValidationOutcome]:
     """Append every independently accepted record and return all receipt outcomes."""
-    outcomes = validate_envelope(root, envelope, account=account)
-    battle_ids = retained_values(root, "battleId")
-    payload_hashes = retained_values(root, "payloadHash")
+    retained = retained_results(root)
+    outcomes = validate_with_idempotent_retries(root, envelope, account=account, retained=retained)
+    payload_hashes = {str(record.get("payloadHash")) for record in retained.values()}
     persisted: list[ValidationOutcome] = []
     for outcome in outcomes:
         if outcome.accepted is None:
             persisted.append(outcome)
             continue
         item: AcceptedResult = outcome.accepted
         battle_id = str(item.record["battleId"])
-        if battle_id in battle_ids:
+        payload_hash = str(item.record["payloadHash"])
+        existing = retained.get(battle_id)
+        if existing is item.record:
+            persisted.append(outcome)
+            continue
+        if battle_id in retained:
             persisted.append(ValidationOutcome(index=outcome.index, battle_id=battle_id, error="duplicate battleId already retained"))
             continue
-        if item.record["payloadHash"] in payload_hashes:
+        if payload_hash in payload_hashes:
             persisted.append(ValidationOutcome(index=outcome.index, battle_id=battle_id, error="duplicate payload hash already retained"))
             continue
         path = raw_path(root, str(item.record["completedAt"]), item.digest)
         if path.exists():
             persisted.append(ValidationOutcome(index=outcome.index, battle_id=battle_id, error="duplicate payload hash already retained"))
             continue
         write_json(path, item.record)
-        battle_ids.add(battle_id)
-        payload_hashes.add(str(item.record["payloadHash"]))
+        retained[battle_id] = item.record
+        payload_hashes.add(payload_hash)
         persisted.append(outcome)
     return persisted
 

tests/test_rumble_data.py

@@ -75,17 +75,48 @@ def testRDA001_IntegrationPositive_valid_records_survive_invalid_batch_neighbors
         self.assertIn("rank", outcomes[1].error)
         self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json"))))
 
-    def testRDA002_IntegrationNegative_structural_and_duplicate_records_never_persist(self) -> None:
+    def testRDA002_IntegrationNegative_structural_records_never_persist(self) -> None:
         malformed = self.envelope()["results"][0]
         malformed.pop("rounds")
         outcome = ingest(self.root, {**self.envelope(), "results": [malformed]}, account="alice")[0]
         self.assertIsNone(outcome.accepted)
         self.assertIn("rounds", outcome.error)
         self.assertFalse((self.root / "results/raw").exists())
+
+    def testRDA001_IntegrationPositive_identical_retry_is_idempotently_accepted(self) -> None:
+        self.assertTrue(ingest(self.root, self.envelope(), account="alice")[0].accepted)
+        retry = ingest(self.root, self.envelope(), account="alice")[0]
+        self.assertTrue(retry.accepted)
+        self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json"))))
+
+    def testRDA001_IntegrationPositive_identical_retry_survives_current_pin_change(self) -> None:
         self.assertTrue(ingest(self.root, self.envelope(), account="alice")[0].accepted)
-        duplicate = ingest(self.root, self.envelope(), account="alice")[0]
+        self.write("engine.json", {"schemaVersion": 1, "behaviorVersion": 2, "gameTypes": {"1v1": {"rounds": 35, "battlefield": [800, 600], "participants": 2}}})
+        retry = ingest(self.root, self.envelope(), account="alice")[0]
+        self.assertTrue(retry.accepted)
+        self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json"))))
+
+    def testRDA002_IntegrationNegative_conflicting_battle_id_is_rejected(self) -> None:
+        self.assertTrue(ingest(self.root, self.envelope(), account="alice")[0].accepted)
+        conflicting = self.envelope()
+        conflicting["results"][0]["completedAt"] = "2026-08-02T12:01:00Z"
+        duplicate = ingest(self.root, conflicting, account="alice")[0]
         self.assertIsNone(duplicate.accepted)
         self.assertIn("duplicate battleId", duplicate.error)
+        self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json"))))
+
+    def testRDA002_IntegrationNegative_duplicate_within_one_batch_is_rejected(self) -> None:
+        envelope = self.envelope()
+        envelope["results"].append(dict(envelope["results"][0]))
+        outcomes = ingest(self.root, envelope, account="alice")
+        self.assertTrue(outcomes[0].accepted)
+        self.assertIsNone(outcomes[1].accepted)
+        self.assertIn("duplicate battleId", outcomes[1].error)
+        self.assertEqual(1, len(list((self.root / "results/raw").rglob("*.json"))))
+
+    def testArch_successful_receipts_follow_fact_publication(self) -> None:
+        workflow = (ROOT / ".github/workflows/ingest.yml").read_text(encoding="utf-8")
+        self.assertLess(workflow.index("git push"), workflow.index("gh issue comment"))
 
     def testRDA002_IntegrationNegative_rejects_each_documented_structural_violation(self) -> None:
         invalid_records = []