fix: align V1 result ingestion contract

.github/workflows/ingest.yml

@@ -37,11 +37,11 @@ jobs:
             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 'Accepted result batch:\n\n```text\n' > .inbox/comment.md
+              printf 'Processed result batch:\n\n```text\n' > .inbox/comment.md
               cat .inbox/outcome.txt >> .inbox/comment.md
               printf '\n```\n' >> .inbox/comment.md
             else
-              printf 'Rejected result batch:\n\n```text\n' > .inbox/comment.md
+              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
             fi

CONTRIBUTING.md

@@ -6,9 +6,11 @@ Open a pull request that adds `clients/<your-forge-account>.json`. Its `account`
 
 ## Submit a ranked batch
 
-Create an issue titled `[result] <client-id> <UTC timestamp>`, apply the `result-submission` label, and place exactly one JSON envelope in a fenced `json` block. The envelope has `schemaVersion: 1`, `clientId`, `clientVersion`, and 1–60 `results`. Each result supplies a UUID `battleId`, `completedAt`, the pinned `behaviorVersion`, a supported game type, and one participant record per required bot with `name`, `version`, and non-negative `totalScore`.
+Create an issue titled `[result] <client-id> <UTC timestamp>`, apply the `result-submission` label, and place exactly one JSON envelope in a fenced `json` block. The envelope has `schemaVersion: 1`, `clientId`, `clientVersion`, and 1–60 `results`. Each result supplies a UUID `battleId`, `completedAt`, nested `client.id` and `client.version` matching the envelope, nested `engine.behaviorVersion` matching the engine pin, game type, pinned rounds and arena dimensions, and the complete Battle Runner participant result model.
 
-The submitting issue account and `clientId` must be registered. Every bot must be active in `catalog.json`; the catalog is synchronized from the reviewed Rumble bot catalog by maintainers. The drain workflow closes every processed issue with accepted and rejected record diagnostics.
+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 catalog is synchronized from the reviewed Rumble bot catalog by maintainers. 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.
 
 ## Rules
 

GOVERNANCE.md

@@ -8,7 +8,7 @@ Keep facts immutable. To exclude a disputed result, add its `battleId` to `exclu
 
 ## Ingestion
 
-The workflow is triggered by labelled issues and a modest schedule. It serializes runs, validates every issue batch, commits all accepted facts in one commit, regenerates projections, comments on the issue, and closes it. If scheduled workflows are disabled after inactivity, re-enable the workflow; an incoming labelled issue also wakes the system.
+The workflow is triggered by labelled issues and a modest schedule. It serializes runs, validates every record in an issue batch independently, commits all accepted facts in one commit, regenerates projections, comments with one receipt line per record, and closes the issue. Aggregation applies the current registrations, bans, disqualified bots, and exclusions to immutable facts, so later moderation immediately changes projections without rewriting history. If scheduled workflows are disabled after inactivity, re-enable the workflow; an incoming labelled issue also wakes the system.
 
 ## Compaction and fork drill
 

scripts/aggregate.py

@@ -14,6 +14,34 @@
 TARGET_SAMPLES_PER_PAIRING = 6
 
 
+def registered_clients(root: Path) -> dict[str, set[str]]:
+    """Return current client registrations indexed by forge account."""
+    registrations: dict[str, set[str]] = {}
+    for path in repository_files(root, "clients"):
+        registration = read_json(path)
+        account, client_ids = registration.get("account"), registration.get("clientIds")
+        if isinstance(account, str) and isinstance(client_ids, list) and all(isinstance(client_id, str) and client_id for client_id in client_ids):
+            registrations[account] = set(client_ids)
+    return registrations
+
+
+def record_client_id(record: dict[str, Any]) -> str | None:
+    """Return a record's V1 nested client identifier."""
+    client = record.get("client")
+    return client.get("id") if isinstance(client, dict) and isinstance(client.get("id"), str) else None
+
+
+def eligible_fact(record: dict[str, Any], *, registrations: dict[str, set[str]], banned_accounts: set[str], disqualified_bots: set[tuple[str, str]], exclusions: set[str]) -> bool:
+    """Select a fact against every current moderation and registration input."""
+    account = record.get("submittedBy")
+    if not isinstance(account, str) or account in banned_accounts or record.get("battleId") in exclusions:
+        return False
+    client_id = record_client_id(record)
+    if client_id not in registrations.get(account, set()):
+        return False
+    return not any(identity(participant) in disqualified_bots for participant in record.get("participants", []) if isinstance(participant, dict))
+
+
 def facts(root: Path) -> list[dict[str, Any]]:
     """Load raw facts and compacted rollups in a deterministic order."""
     records: list[dict[str, Any]] = []
@@ -23,7 +51,11 @@ def facts(root: Path) -> list[dict[str, Any]]:
         rollup = read_json(path)
         records.extend(rollup.get("results", []))
     exclusions = set(read_json(root / "exclusions.json").get("battleIds", []))
-    return sorted((record for record in records if record.get("battleId") not in exclusions), key=lambda item: (str(item.get("completedAt")), str(item.get("payloadHash"))))
+    bans = read_json(root / "bans.json")
+    banned_accounts = set(bans.get("bannedAccounts", []))
+    disqualified_bots = {(str(item.get("name")), str(item.get("version"))) for item in bans.get("disqualifiedBots", [])}
+    registrations = registered_clients(root)
+    return sorted((record for record in records if eligible_fact(record, registrations=registrations, banned_accounts=banned_accounts, disqualified_bots=disqualified_bots, exclusions=exclusions)), key=lambda item: (str(item.get("completedAt")), str(item.get("payloadHash"))))
 
 
 def active_catalog(root: Path) -> list[dict[str, Any]]:
@@ -39,7 +71,7 @@ def identity(participant: dict[str, Any]) -> tuple[str, str]:
 def aggregate_game_type(records: list[dict[str, Any]], catalog: list[dict[str, Any]], game_type: str, behavior_version: int) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
     """Produce leaderboard, pairings, and matchmaking advice for one game type."""
     eligible = {(str(bot["name"]), str(bot["version"])): bot for bot in catalog}
-    relevant = [record for record in records if record.get("gameType") == game_type and record.get("behaviorVersion") == behavior_version]
+    relevant = [record for record in records if record.get("gameType") == game_type and record.get("engine", {}).get("behaviorVersion") == behavior_version]
     shares: dict[tuple[str, str], dict[tuple[tuple[str, str], ...], list[float]]] = defaultdict(lambda: defaultdict(list))
     pairing_counts: dict[tuple[tuple[str, str], ...], int] = defaultdict(int)
     for record in relevant:
@@ -50,7 +82,7 @@ def aggregate_game_type(records: list[dict[str, Any]], catalog: list[dict[str, A
         for participant in participants:
             bot = identity(participant)
             if bot in eligible:
-                shares[bot][bots].append(float(participant["totalScore"]) / total)
+                shares[bot][bots].append(float(participant["totalScore"]) / total if total else 0.0)
     entries = []
     for bot, catalog_entry in eligible.items():
         bot_pairings = shares.get(bot, {})
@@ -92,7 +124,9 @@ def aggregate(root: Path) -> None:
             write_json(root / "site" / "data" / "bots" / f"{entry['name']}-{entry['version']}.json", {"schemaVersion": 1, "projectionId": leaderboard["projectionId"], "gameType": game_type, "entry": entry})
     client_totals: dict[str, int] = defaultdict(int)
     for record in records:
-        client_totals[str(record.get("clientId"))] += 1
+        client_id = record_client_id(record)
+        if client_id is not None:
+            client_totals[client_id] += 1
     write_json(root / "clients.json", {"schemaVersion": 1, "clients": [{"clientId": client_id, "battles": battles} for client_id, battles in sorted(client_totals.items())]})
     write_json(root / "site" / "data" / "clients.json", {"schemaVersion": 1, "clients": [{"clientId": client_id, "battles": battles} for client_id, battles in sorted(client_totals.items())]})
 

scripts/ingest.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 
 from common import read_json, write_json
-from validate import ValidationError, validate_envelope
+from validate import AcceptedResult, ValidationError, ValidationOutcome, validate_envelope
 
 
 def raw_path(root: Path, completed_at: str, digest: str) -> Path:
@@ -18,31 +18,49 @@ def raw_path(root: Path, completed_at: str, digest: str) -> Path:
     return root / "results" / "raw" / date[0] / date[1] / f"{digest}.json"
 
 
-def existing_battle_ids(root: Path) -> set[str]:
-    """Return battle IDs already retained as raw facts."""
+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()
     for path in (root / "results" / "raw").rglob("*.json") if (root / "results" / "raw").exists() else []:
-        result.add(str(read_json(path).get("battleId")))
+        result.add(str(read_json(path).get(field)))
     for path in (root / "results" / "rollups").rglob("*.json") if (root / "results" / "rollups").exists() else []:
-        result.update(str(record.get("battleId")) for record in read_json(path).get("results", []))
+        result.update(str(record.get(field)) for record in read_json(path).get("results", []))
     return result
 
 
-def ingest(root: Path, envelope: object, *, account: str) -> list[str]:
-    """Append accepted records once and return stable outcome messages."""
-    accepted = validate_envelope(root, envelope, account=account)
-    seen = existing_battle_ids(root)
-    duplicates = [item.record["battleId"] for item in accepted if item.record["battleId"] in seen]
-    if duplicates:
-        raise ValidationError(f"duplicate battleId already retained: {', '.join(sorted(duplicates))}")
-    paths: list[str] = []
-    for item in accepted:
+def format_outcome(outcome: ValidationOutcome) -> str:
+    """Render one stable receipt line for a submitted record."""
+    label = outcome.battle_id or f"result[{outcome.index}]"
+    return f"{label}: {'accepted' if outcome.accepted else f'rejected: {outcome.error}'}"
+
+
+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")
+    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:
+            persisted.append(ValidationOutcome(index=outcome.index, battle_id=battle_id, error="duplicate battleId already retained"))
+            continue
+        if item.record["payloadHash"] 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():
-            raise ValidationError(f"duplicate payload hash: {item.digest}")
+            persisted.append(ValidationOutcome(index=outcome.index, battle_id=battle_id, error="duplicate payload hash already retained"))
+            continue
         write_json(path, item.record)
-        paths.append(path.relative_to(root).as_posix())
-    return paths
+        battle_ids.add(battle_id)
+        payload_hashes.add(str(item.record["payloadHash"]))
+        persisted.append(outcome)
+    return persisted
 
 
 def main() -> int:
@@ -53,11 +71,11 @@ def main() -> int:
     parser.add_argument("--input", type=Path, required=True)
     arguments = parser.parse_args()
     try:
-        paths = ingest(arguments.root.resolve(), read_json(arguments.input), account=arguments.account)
+        outcomes = ingest(arguments.root.resolve(), read_json(arguments.input), account=arguments.account)
     except (OSError, ValueError, ValidationError) as error:
         print(f"ingestion failed: {error}")
         return 1
-    print("accepted:\n" + "\n".join(paths))
+    print("\n".join(format_outcome(outcome) for outcome in outcomes))
     return 0