feat: capture and export gear, cartridges and loadouts

Adds a third export alongside pull history and achievements: the cartridges,
modules, character levels and equipped console boards the game sends at login.

Shaped as added files plus a handful of one-line calls, so this repo can keep
merging upstream:

- decoder/gear.py: self-contained bit-level decoder over a list of datagrams.
- export/gear_export.py: the nte-gear-export envelope.
- live_capture/gear_collector.py: UDP flow accumulation, the mid-capture
  decode, the fallback scan, the console line and the export file.

Gear is reported during the capture rather than only at the stop. The burst
arrives in a rush and then stops, so "this flow went quiet" is the only
mid-stream marker available; a flow is retried only once it has grown, since
decoding per packet would be quadratic in flow size.

Flows are tried largest first and the first one that decodes wins - picking
the single biggest was wrong, history traffic can outweigh a short login
burst.

Verified against a live capture: 817 items (310 cartridges, 507 modules),
20 characters, 13 loadouts, every bucket matching the in-game counts exactly.
26 synthetic tests, no real capture used as a fixture; 149 total.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.github/workflows/update-mappings.yml

@@ -1,4 +1,4 @@
-name: Update reward mappings
+name: Update mappings
 
 on:
   workflow_dispatch:
@@ -14,7 +14,7 @@ permissions:
   pull-requests: write
 
 concurrency:
-  group: update-reward-mappings
+  group: update-mappings
   cancel-in-progress: false
 
 jobs:
@@ -23,7 +23,7 @@ jobs:
     runs-on: ubuntu-latest
 
     env:
-      AUTOMATION_BRANCH: dev/automated-mapping-update
+      AUTOMATION_BRANCH: dev/automated-mapping-update-${{ github.ref_name }}
       NTE_ASSETS_REF: ${{ inputs.source_ref }}
       GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 
@@ -32,7 +32,7 @@ jobs:
         uses: actions/checkout@v5
         with:
           fetch-depth: 0
-          ref: ${{ github.event.repository.default_branch }}
+          ref: ${{ github.ref_name }}
 
       - name: Get Python
         uses: actions/setup-python@v6
@@ -42,7 +42,7 @@ jobs:
       - name: Install project
         run: python -m pip install -e .
 
-      - name: Rebuild reward mappings
+      - name: Rebuild mappings
         run: python tools/update_mappings.py --source-ref "$NTE_ASSETS_REF" --apply
 
       - name: Run tests
@@ -91,7 +91,7 @@ jobs:
         if: steps.changes.outputs.changed == 'true'
         shell: bash
         env:
-          BASE_BRANCH: ${{ github.event.repository.default_branch }}
+          BASE_BRANCH: ${{ github.ref_name }}
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
         run: |
           python - <<'PY'
@@ -108,7 +108,7 @@ jobs:
               f"- {changes['additions']} additions\n"
               f"- {changes['updates']} updates\n"
               f"- {changes['deletions']} deletions\n\n"
-              "The workflow rebuilt all reward mappings, resolved English names through "
+              "The workflow rebuilt all reward and achievement mappings, resolved English text through "
               "`Localization/en/game.json`, and passed the complete test suite. Pool mapping "
               "files and UID inputs were not changed.\n\n"
               f"[Workflow run]({os.environ['RUN_URL']})\n"
@@ -125,12 +125,12 @@ jobs:
 
           if [ -n "$pr_number" ]; then
             gh pr edit "$pr_number" \
-              --title "chore: update reward mappings from NTE_Assets" \
+              --title "chore: update mappings from NTE_Assets" \
               --body-file build/mapping-update/pull-request-body.md
           else
             gh pr create \
               --head "$AUTOMATION_BRANCH" \
               --base "$BASE_BRANCH" \
-              --title "chore: update reward mappings from NTE_Assets" \
+              --title "chore: update mappings from NTE_Assets" \
               --body-file build/mapping-update/pull-request-body.md
           fi

README.md

@@ -4,7 +4,7 @@
 
 # NTE History Exporter
 
-Prototype CLI exporter for **Neverness to Everness** pull history — decodes your own game traffic into sanitized JSON ready for tracker import.
+Prototype CLI exporter for **Neverness to Everness** pull history and achievements — decodes your own game traffic into sanitized JSON.
 
 ![Python](https://img.shields.io/badge/python-3.10+-blue)
 ![Platform](https://img.shields.io/badge/platform-Windows%20%7C%20Linux%20%7C%20macOS-0078D6)
@@ -23,10 +23,17 @@ Prototype CLI exporter for **Neverness to Everness** pull history — decodes yo
 
 ## What It Does
 
-The exporter decodes Permanent Board, Limited Character Board, Arc Miracle Box, and Mystery Box history pages from captured UDP data, applies conservative timestamp-boundary handling, and writes sanitized JSON suitable for tracker import.
+The exporter decodes Permanent Board, Limited Character Board, Arc Miracle Box,
+and Mystery Box history pages from captured UDP data. It also captures tracked
+achievements during login, including completed and in-progress records. Pull
+history is written as tracker-ready JSON and achievements use a separate
+versioned JSON format.
 
 > [!NOTE]
-> The import JSON contains decoded history rows and the shareable NTE user UID when it can be detected. It does **not** export tokens, account IDs, role IDs, device IDs, server IPs, raw packets, cookies, session data, or other capture metadata.
+> Export JSON contains decoded history or achievement records and the shareable
+> NTE user UID when it can be detected. It does **not** export tokens, account
+> IDs, role IDs, device IDs, server IPs, raw packets, cookies, session data, or
+> other capture metadata.
 
 ## Requirements
 
@@ -90,7 +97,11 @@ From source:
 Or simply double-click **`run-exporter.cmd`**.
 
 > [!IMPORTANT]
-> For automatic user UID and account-server detection, launch the tool **before pressing Start on the game's main menu**. If you are already in game, history capture can still work; the tool will ask for missing account details before saving.
+> Start the exporter **before pressing Start on the game's main menu**. This is
+> required for achievement capture and enables automatic user UID and
+> account-server detection. If already in game, pull-history capture still
+> works, but achievements require returning to the main menu and logging in
+> again; missing account details are requested before saving.
 
 Once running, open any supported history board in game. The tool keeps listening until you press any key. After it prints and saves the results, press any key again to close the exporter. Exports are written under `exports\` as:
 
@@ -98,6 +109,7 @@ Once running, open any supported history board in game. The tool keeps listening
 - `<user_uid>_Limited_<date_time>.json`
 - `<user_uid>_Arc_<date_time>.json`
 - `<user_uid>_MysteryBox_<date_time>.json`
+- `<user_uid>_Achievements_<date_time>.json`
 
 If the user UID is not detected automatically, the console asks for it before saving. Leaving it blank saves as `unknown_<banner>_<date_time>.json`, but may prevent import on some trackers.
 
@@ -112,19 +124,22 @@ If a page response is missed, the exporter reports the missing page number while
 
 ### Achievement export
 
-Start the live exporter before launching the game. The completed achievement
+Start the live exporter before pressing Start on the game's main menu. The
-list is loaded during login, captured automatically, and written to
+tracked achievement state is loaded during login, captured automatically, and written to
 `exports/<user_uid>_Achievements_*.json` as a versioned achievement export with
 account metadata, completion totals, and records grouped by achievement category.
-Both completed and incomplete achievements are included with their current
+Both completed and in-progress achievements are included with their current
 progress, status, and completion time when available. Generated metadata adds
 the localized title and description, target, quality, and rewards from the
 current `NTE_Assets` achievement table. Newly captured IDs that are not yet in
 the bundled mapping still export normally; only their optional display metadata
 is omitted until the mappings are refreshed.
-Completed PlayStation trophy entries are retained and identified separately
+PlayStation trophy entries are retained and identified separately
 even though they are not shown in the in-game achievement list. The console
-reports the visible and PlayStation counts as soon as the list has been decoded.
+reports completed and in-progress totals for in-game and PlayStation records as
+soon as the list has been decoded. Every observed record without a completion
+timestamp is classified as `in_progress`, even when its numeric progress is
+zero; compound achievements can track hidden checklist progress this way.
 
 #### Linux/macOS
 
@@ -191,22 +206,22 @@ The exporter automatically includes the shareable NTE user UID when it appears i
 
 ## Mapping maintenance
 
-Reward metadata can be rebuilt directly from the latest NTE_Assets tables and
+Reward and achievement metadata can be rebuilt directly from the latest
-English translation files. The reward snapshot may change IDs while UID inputs
+NTE_Assets tables and English translation files. Mapping snapshots may change
-remain untouched:
+IDs while UID inputs remain untouched:
 
-Run **Update reward mappings** from the GitHub Actions tab to generate and test
+Run **Update mappings** from the GitHub Actions tab to generate and test
 the snapshot in a reviewable pull request, or run it locally:
 
 ```powershell
 python tools/update_mappings.py
 ```
 
-This rebuilds reviewable reward-map candidates directly from NTE_Assets under
+This rebuilds reviewable mapping candidates directly from NTE_Assets under
 `build/mapping-update/`; committed mappings are untouched unless `--apply` is
 explicitly supplied. The snapshot may include additions, updates, and removals,
 while pool mappings and UID inputs remain untouched. See
-[Reward mapping updates](docs/mapping-updates.md) for the source rules and
+[Mapping updates](docs/mapping-updates.md) for the source rules and
 review workflow.
 
 ## Privacy

docs/export-format.md

@@ -15,7 +15,7 @@ decoder-only offsets.
   "capture_source": "npcap",
   "exporter": {
     "name": "nte-history-exporter",
-    "version": "0.2.1"
+    "version": "0.3.0"
   },
   "banner": {
     "id": "Lottery_Permanent",
@@ -84,6 +84,12 @@ Current stable pool IDs:
 Avoid using `banner.name`, `reward_name`, `reward_type` or `reward_rank` as primary IDs. They
 are useful display fields, but may change when mapping files are updated.
 
+Known `reward_id` values are matched case-insensitively and emitted using the
+canonical casing from the bundled mapping. This prevents packet/asset casing
+differences from dropping display metadata or producing different downstream
+IDs. An unknown reward keeps the casing decoded from the packet. Reward IDs are
+not UID inputs, so canonicalizing their casing does not change pull UIDs.
+
 ## Stability notes
 
 Every JSON record is exported with a stable `uid`. Re-scanning deeper history can
@@ -176,6 +182,79 @@ Mystery Box:
 }
 ```
 
+## Achievement export
+
+Achievements are written separately as
+`<user_uid>_Achievements_<date_time>.json`. The document contains only records
+observed in the account's login response; definitions found only in the asset
+mapping are not synthesized into the export.
+
+```json
+{
+  "format": "nte-achievement-export",
+  "format_version": 1,
+  "game": "Neverness to Everness",
+  "source": "live_capture",
+  "capture_source": "npcap",
+  "exporter": {
+    "name": "nte-history-exporter",
+    "version": "0.3.0"
+  },
+  "scan": {
+    "in_game": {
+      "total_achievements": 367,
+      "completed_achievements": 310,
+      "in_progress_achievements": 57
+    },
+    "playstation": {
+      "total_achievements": 35,
+      "completed_achievements": 31,
+      "in_progress_achievements": 4
+    }
+  },
+  "user_uid": "218216016349",
+  "server_id": "23003",
+  "account_region": "EU",
+  "categories": {
+    "battle": [
+      {
+        "id": "Battle_25",
+        "platform": "in_game",
+        "status": "in_progress",
+        "progress": 20,
+        "completed": false,
+        "completed_at": null,
+        "name": "Devil Within II",
+        "description": "Trigger Hexed ×50.",
+        "quality": "high",
+        "target": 50,
+        "rewards": [
+          {"item_id": "Annulith", "amount": 10}
+        ]
+      }
+    ]
+  }
+}
+```
+
+Achievement fields:
+
+- `scan.in_game` and `scan.playstation` separate records visible in game from
+  PlayStation trophy records and report completed/in-progress totals.
+- `categories` groups records by the prefix of their captured ID. The category
+  is therefore not repeated on every record.
+- `status` is `completed` when completion ticks are present and `in_progress`
+  otherwise. An observed compound achievement can be in progress while its
+  numeric `progress` is zero because individual checklist state is server-side.
+- `completed_at` is a decoded UTC time for completed records and `null` for
+  in-progress records.
+- `name`, `description`, `quality`, `target`, and `rewards` are optional display
+  metadata from `mappings/achievements.json`.
+- A newly captured ID missing from the bundled mapping still exports its core
+  captured fields; only optional display metadata is omitted.
+- Asset definitions absent from the network response are not exported because
+  absence cannot distinguish unstarted, locked, or unavailable content.
+
 ## CSV diagnostics
 
 CSV exports are mainly for debugging the decoder. Tools should prefer JSON

docs/limitations.md

@@ -21,5 +21,13 @@ Known limitations:
 - The file adapter reads mitmproxy `.flows` captures for research and testing.
 - Npcap is Windows-only and is not redistributed with this project; Linux and macOS use their system libpcap.
 - Reward keys decode to their reward id string, so unknown rewards still export a usable `reward_id`; display names/ranks come from the mapping JSON files and should be expanded as new rewards appear.
+- Achievement exports contain only records observed in the login response.
+  Asset-only definitions are not synthesized because absence cannot distinguish
+  unstarted, locked, or unavailable content. Unknown captured achievement IDs
+  still export without optional display metadata.
+- Observed achievements have only `completed` and `in_progress` states. Some
+  compound achievements expose numeric progress `0` while the server tracks
+  hidden checklist progress, so every observed record without completion ticks
+  is classified as `in_progress`.
 
 Privacy guardrail: raw packet data must not be included in sanitized exports.

docs/mapping-updates.md

@@ -1,4 +1,4 @@
-# Reward mapping updates
+# Mapping updates
 
 `tools/update_mappings.py` rebuilds `arcs.json`, `characters.json`,
 `items.json`, and `achievements.json` directly from the current
@@ -19,15 +19,17 @@ game whose asset tables evolve over time.
 
 ### GitHub Actions
 
-Run **Update reward mappings** from the repository's Actions tab. The optional
+Run **Update mappings** from the repository's Actions tab. **Use workflow
-`source_ref` input accepts an NTE_Assets branch, tag, or commit and defaults to
+from** selects the exporter branch whose updater is run and the target branch
-`main`.
+for the resulting pull request. The `source_ref` input separately selects an
+NTE_Assets branch, tag, or commit and defaults to `main`.
 
 The workflow rebuilds the mappings, runs the complete test suite, uploads the
 JSON review report, and opens a pull request from
-`dev/automated-mapping-update`. Running it again updates the same open pull
+`dev/automated-mapping-update-<selected-branch>`. Running it again for the same
-request. If the authoritative snapshot has not changed, it reports zero changes
+selected branch updates that branch's open pull request. If the authoritative
-and does not create a commit or pull request.
+snapshot has not changed, it reports zero changes and does not create a commit
+or pull request.
 
 The repository setting **Allow GitHub Actions to create and approve pull
 requests** must be enabled for automatic pull-request creation. The workflow
@@ -68,7 +70,7 @@ python -m pytest -q
 ```
 
 For CI, `--check` exits with status 1 whenever the generated snapshot differs
-from the committed reward mappings. It still writes the staged artifacts.
+from the committed mappings. It still writes the staged artifacts.
 
 ## Source rules
 
@@ -80,6 +82,9 @@ from the committed reward mappings. It still writes the staged artifacts.
 - Other pull rewards are selected by `GachaIllustrate` and resolved to the
   inventory tables for quality and localization keys. Appearance-table IDs
   provide canonical casing for glider rewards.
+- Vehicle liveries are resolved from their illustration name and icon. Rows
+  whose icon identifies a different reward are treated as unreleased upstream
+  placeholders and skipped until the assets are corrected.
 - Mystery Box rewards are selected from every row in
   `DT_GashaponLotteryGlobal`. No event IDs are hard-coded, so future rotations
   are included automatically. Rewards resolve through the inventory, capital,
@@ -89,8 +94,8 @@ from the committed reward mappings. It still writes the staged artifacts.
   currencies, are categorized as items.
 - Names are resolved strictly through `Localization/en/game.json`. A missing or
   ambiguous key fails the update instead of falling back to a DT value.
-- Character-awakening and character-vehicle illustration entries are not
+- Character-awakening illustration entries are not independent pull rewards
-  independent pull rewards and are excluded.
+  and are excluded.
 - Orange, purple, and blue item qualities map to `S`, `A`, and `B`.
 
 ## UID compatibility boundary
@@ -100,7 +105,7 @@ reads or writes the permanent, limited, beginner, or Arc pool mapping files.
 Banner IDs, timestamps, record ordering, format version, and all UID inputs are
 therefore unchanged by mapping synchronization.
 
-Reward IDs and display metadata may change or disappear when the authoritative
+Reward and achievement IDs or display metadata may change or disappear when
-NTE_Assets snapshot changes. Existing exported history files remain unchanged;
+the authoritative NTE_Assets snapshot changes. Existing exports remain
-new exports describe rewards using the current asset snapshot. UID stability is
+unchanged; new exports use the current asset snapshot. UID stability is
-independent of reward display mappings.
+independent of display mappings.

docs/packet-format.md

@@ -1,6 +1,21 @@
 # Packet Format Notes
 
-This prototype supports separate Monopoly, Arc/Gashapon, and Mystery Box history decoders.
+This prototype supports separate Monopoly, Arc/Gashapon, Mystery Box history,
+and achievement decoders.
+
+## Achievements
+
+The account's achievement component arrives in an inbound TCP game-data block
+during login. The block is LZ4-decoded and contains `AchievementRecord` entries.
+Each record provides an achievement ID, an unsigned numeric progress value, and
+a .NET completion timestamp. A non-zero timestamp marks a completed record.
+Every observed record without completion ticks is treated as in progress;
+compound achievements can have server-side checklist progress while exposing a
+numeric value of zero.
+
+Only records present in the network response are exported. Display metadata is
+looked up case-insensitively in `mappings/achievements.json`, because captured
+and asset-table ID casing can differ. Missing metadata never prevents export.
 
 ## Decoder strategy
 
@@ -86,6 +101,8 @@ Decoded fields:
   Arc/Gashapon history uses the same scheme at `ASCII * 2`.
 - The decoder decodes the key to its id string and looks up display metadata
   in `mappings/arcs.json`, `mappings/characters.json`, and `mappings/items.json`.
+  Lookups are case-insensitive across Monopoly, Arc, and Mystery Box decoders;
+  known IDs are emitted using mapping-canonical casing.
   Achievement display metadata is generated separately in
   `mappings/achievements.json`.
   Unknown rewards still export their decoded id with empty name/rank.

mappings/arcs.json

@@ -10,8 +10,10 @@
   "fork_Butterfly": {"name": "Reality Refuge", "rank": "S"},
   "fork_Castle": {"name": "Call of the Twisted City", "rank": "A"},
   "fork_Crowbar": {"name": "Time Bandit", "rank": "A"},
+  "fork_DemonBlade": {"name": "Ravenous Blade", "rank": "S"},
   "fork_Door": {"name": "The Wrong Gate", "rank": "S"},
   "fork_dustbin": {"name": "Dangerous Game", "rank": "B"},
+  "fork_GoldRecord": {"name": "Voice of the Voyager", "rank": "S"},
   "fork_GoldWool": {"name": "What's Desired", "rank": "S"},
   "fork_jiaojuan": {"name": "Shiny Days", "rank": "A"},
   "fork_jingmotingyuan": {"name": "Camellia Society", "rank": "S"},

mappings/characters.json

@@ -9,6 +9,7 @@
   "1023": {"name": "Baicang", "rank": "S"},
   "1025": {"name": "Hathor", "rank": "S"},
   "1033": {"name": "Adler", "rank": "A"},
+  "1036": {"name": "Zankou", "rank": "S"},
   "1039": {"name": "Fadia", "rank": "S"},
   "1046": {"name": "Zero", "rank": "S"},
   "1051": {"name": "Zero", "rank": "S"},
@@ -18,7 +19,9 @@
   "1056": {"name": "Lacrimosa", "rank": "S"},
   "1070": {"name": "Aurelia", "rank": "A"},
   "1071": {"name": "Chaos", "rank": "S"},
+  "1072": {"name": "Linko", "rank": "S"},
   "1073": {"name": "Chiz", "rank": "S"},
   "1075": {"name": "Iroi", "rank": "S"},
-  "1076": {"name": "Shinku", "rank": "S"}
+  "1076": {"name": "Shinku", "rank": "S"},
+  "1091": {"name": "Lacrimosa", "rank": "S"}
 }

mappings/items.json

@@ -1,40 +1,51 @@
 {
+  "BussinessCard_MangHe_bsj": {"type": "cosmetic", "name": "Trackside Shadow", "rank": "S"},
+  "CharacterUpMaterial_lv3": {"type": "item", "name": "Elite Hunter Guide", "rank": "A"},
+  "CityAbility_UpMaterial": {"type": "item", "name": "Dreamless Seed", "rank": "B"},
   "Dice_ticket_01": {"type": "item", "name": "Warp Piece", "rank": "S"},
   "Dice_ticket_02": {"type": "item", "name": "Lost Piece", "rank": "A"},
   "Dicelimite": {"type": "item", "name": "Solid Dice", "rank": "S"},
   "DiceNormal": {"type": "item", "name": "Fabricated Dice", "rank": "S"},
+  "Fashion_1051_rpg_level3_Base": {"type": "cosmetic", "name": "Glory Days", "rank": "S"},
+  "Fashion_1051_rpg_level3_Dye1": {"type": "cosmetic", "name": "Glory Days - Braveheart", "rank": "S"},
+  "Fashion_1051_rpg_level3_Dye2": {"type": "cosmetic", "name": "Glory Days - Starfall", "rank": "S"},
+  "Fashion_1075_rpg_level2_Base": {"type": "cosmetic", "name": "Archmage", "rank": "S"},
   "Fashion_character_1004_01": {"type": "cosmetic", "name": "Gilded Rhapsody", "rank": "S"},
   "Fashion_character_1010": {"type": "cosmetic", "name": "Phoenix Kick", "rank": "S"},
+  "Fashion_character_1010_swimsuit": {"type": "cosmetic", "name": "Sunny Surf", "rank": "S"},
+  "Fashion_character_1019_swimsuit": {"type": "cosmetic", "name": "Saltwater Taffy", "rank": "S"},
+  "Fashion_character_1036_03": {"type": "cosmetic", "name": "Autumn Vignette", "rank": "S"},
+  "Fashion_character_1051_07": {"type": "cosmetic", "name": "Perfect Partners", "rank": "S"},
+  "Fashion_character_1051_swimsuit": {"type": "cosmetic", "name": "Cloudless Summer", "rank": "S"},
   "Fashion_character_1052_01": {"type": "cosmetic", "name": "Priceless Orchid", "rank": "S"},
   "Fashion_character_1071_01": {"type": "cosmetic", "name": "Clear Skies", "rank": "S"},
-  "Fashion_character_1075_01": {"type": "cosmetic", "name": "Summer Dream", "rank": "S"},
+  "Fashion_character_1072_01": {"type": "cosmetic", "name": "Surfing Among Stars", "rank": "S"},
   "Fashion_character_1076_01": {"type": "cosmetic", "name": "Student of Terrasea", "rank": "S"},
   "Fashion_Glide_1004": {"type": "cosmetic", "name": "Tomato Duo", "rank": "A"},
   "Fashion_Glide_1010": {"type": "cosmetic", "name": "Underboss-of-the-Underboss", "rank": "A"},
+  "Fashion_Glide_1036": {"type": "cosmetic", "name": "Scarlet Sash", "rank": "A"},
   "Fashion_Glide_1052": {"type": "cosmetic", "name": "Orchid Breeze", "rank": "A"},
   "Fashion_Glide_1071": {"type": "cosmetic", "name": "Skyrider", "rank": "A"},
-  "Fashion_glide_1075": {"type": "cosmetic", "name": "Skyrider", "rank": "A"},
+  "Fashion_Glide_1072": {"type": "cosmetic", "name": "Sheepcopter", "rank": "A"},
+  "Fashion_Glide_1075": {"type": "cosmetic", "name": "Sheepcopter", "rank": "A"},
   "Fashion_Glide_1076": {"type": "cosmetic", "name": "Overcast Canopy", "rank": "A"},
   "Fashion_vehicle_1004_V021": {"type": "cosmetic", "name": "Tomato Cruise - Livery", "rank": "S"},
   "Fashion_vehicle_1010_V008": {"type": "cosmetic", "name": "Tiger Incoming! - Livery", "rank": "S"},
+  "Fashion_vehicle_1036_V043": {"type": "cosmetic", "name": "Sandy Mirage - Livery", "rank": "S"},
   "Fashion_vehicle_1052_V024": {"type": "cosmetic", "name": "Autumn Haze - Livery", "rank": "S"},
   "Fashion_vehicle_1071_V010": {"type": "cosmetic", "name": "Hound Blitz - Livery", "rank": "S"},
-  "Fashion_vehicle_1075_V006": {"type": "cosmetic", "name": "Hidden Dragon - Livery", "rank": "S"},
+  "Fashion_vehicle_1075_V006": {"type": "cosmetic", "name": "Three's Company - Livery", "rank": "S"},
   "Fashion_vehicle_1076_V024": {"type": "cosmetic", "name": "Hidden Dragon - Livery", "rank": "S"},
   "Fashion_vehicleSkin_1052_V016": {"type": "cosmetic", "name": "Golden Breeze - Livery", "rank": "S"},
-  "vehicle039": {"type": "item", "name": "Draco", "rank": "S"},
-  "Fashion_1051_rpg_level3_Dye1": {"type": "cosmetic", "name": "Glory Days - Braveheart", "rank": "S"},
-  "Fashion_1051_rpg_level3_Dye2": {"type": "cosmetic", "name": "Glory Days - Starfall", "rank": "S"},
-  "Fashion_1051_rpg_level3_Base": {"type": "cosmetic", "name": "Glory Days", "rank": "S"},
-  "SpecialGift_ticket": {"type": "item", "name": "Floe Cinema Ticket", "rank": "A"},
   "Fons": {"type": "item", "name": "Fons", "rank": "A"},
-  "CharacterUpMaterial_lv3": {"type": "item", "name": "Elite Hunter Guide", "rank": "A"},
-  "Gold": {"type": "item", "name": "Beetle Coin", "rank": "B"},
-  "CityAbility_UpMaterial": {"type": "item", "name": "Dreamless Seed", "rank": "B"},
-  "SpecialGift_letter": {"type": "item", "name": "A Handwritten Letter", "rank": "S"},
-  "Frame_MangHe_shinku": {"type": "cosmetic", "name": "Ember Trail", "rank": "S"},
-  "Fashion_character_1051_07": {"type": "cosmetic", "name": "Perfect Partners", "rank": "S"},
-  "vehicle026": {"type": "item", "name": "Porsche 918 Spyder", "rank": "S"},
   "Frame_MangHe_bsj": {"type": "cosmetic", "name": "Heartbeat Moment", "rank": "S"},
-  "BussinessCard_MangHe_bsj": {"type": "cosmetic", "name": "Trackside Shadow", "rank": "S"}
+  "Frame_MangHe_shinku": {"type": "cosmetic", "name": "Ember Trail", "rank": "S"},
+  "Frame_MangHe_summer_01": {"type": "cosmetic", "name": "Endless Summer", "rank": "S"},
+  "Gold": {"type": "item", "name": "Beetle Coin", "rank": "B"},
+  "SpecialGift_letter": {"type": "item", "name": "A Handwritten Letter", "rank": "S"},
+  "SpecialGift_ticket": {"type": "item", "name": "Floe Cinema Ticket", "rank": "A"},
+  "vehicle026": {"type": "item", "name": "Porsche 918 Spyder", "rank": "S"},
+  "vehicle039": {"type": "item", "name": "Draco", "rank": "S"},
+  "Vehicle044": {"type": "item", "name": "Tide", "rank": "S"},
+  "weeklycloneboss_selectbox_02": {"type": "item", "name": "Anomaly Pilgrimage Material Selection Box", "rank": "S"}
 }

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "nte-history-exporter"
-version = "0.2.1"
+version = "0.4.0"
-description = "Cross-platform Neverness to Everness pull-history exporter."
+description = "Cross-platform Neverness to Everness pull-history and achievement exporter."
 requires-python = ">=3.10"
 dependencies = []
 

src/nte_history_exporter/__init__.py

@@ -1 +1 @@
-__version__ = "0.2.1"
+__version__ = "0.4.0"

src/nte_history_exporter/console.py

@@ -94,9 +94,13 @@ def print_live_instructions(local_ip: str, backend: str = "windows_raw", detail:
     print(style("    You can capture more than one board in the same session.", DIM))
     print()
     print(style("  Achievements", BOLD))
-    print("    Your completed achievements are captured automatically when")
+    print("    Your tracked achievements are captured automatically when")
     print("    you log in after following the step above.")
     print()
+    print(style("  Gear", BOLD))
+    print("    Your cartridges, modules and equipped loadouts are captured")
+    print("    automatically at login, in the same step as achievements.")
+    print()
     print(style("  Ready - use the game, then press any key here when finished.", BOLD, GREEN))
     print(rule())
 
@@ -106,12 +110,26 @@ def print_page_captured(label: str, page: int | None, *, recaptured: bool = Fals
     print(style("  + ", GREEN, BOLD) + label + style(f"  {action} {page}", DIM))
 
 
-def print_achievements_captured(total: int, playstation: int) -> None:
+def print_achievements_captured(
-    visible = total - playstation
+    in_game_completed: int,
+    in_game_in_progress: int,
+    playstation_completed: int,
+    playstation_in_progress: int,
+) -> None:
+    print(style("  + ", GREEN, BOLD) + "Achievements captured")
     print(
-        style("  + ", GREEN, BOLD)
+        style(
-        + "Achievements"
+            f"      In-game      {in_game_completed} completed, "
-        + style(f"  captured {visible} visible + {playstation} PlayStation", DIM)
+            f"{in_game_in_progress} in progress",
+            DIM,
+        )
+    )
+    print(
+        style(
+            f"      PlayStation  {playstation_completed} completed, "
+            f"{playstation_in_progress} in progress",
+            DIM,
+        )
     )
 
 
@@ -184,7 +202,7 @@ def print_problem(text: str) -> None:
 def print_update_available(current_version: str, latest_version: str, release_url: str) -> None:
     print()
     print(style(f"  Update available: v{current_version} -> {latest_version}", YELLOW, BOLD))
-    print(style("  Some banner/reward mappings may be incomplete in this version.", YELLOW))
+    print(style("  Some reward or achievement mappings may be incomplete in this version.", YELLOW))
     print(style("  Updating is recommended for the most accurate export labels.", YELLOW))
     print(style(f"  Download when ready: {release_url}", YELLOW))
 

src/nte_history_exporter/constants.py

@@ -13,7 +13,7 @@ ARC_BANNER_NAME = "Arc Miracle Box"
 MYSTERY_BOX_BANNER_ID = "Gashapon_MysteryBox"
 MYSTERY_BOX_BANNER_NAME = "Mystery Box"
 EXPORTER_NAME = "nte-history-exporter"
-EXPORTER_VERSION = "0.2.1"
+EXPORTER_VERSION = "0.4.0"
 
 HISTORY_REQUEST_BANNER = 4220
 HISTORY_REQUEST_LENGTH = 45

src/nte_history_exporter/decoder/achievement.py

@@ -27,7 +27,10 @@ class AchievementRecord:
     def status(self) -> str:
         if self.completed:
             return "completed"
-        return "in_progress" if self.progress else "not_started"
+        # Presence in the server response means the achievement has tracked
+        # state. Compound achievements can have hidden checklist progress while
+        # exposing a numeric progress value of zero to the client.
+        return "in_progress"
 
     @property
     def completed_at(self) -> str | None:

src/nte_history_exporter/decoder/arc.py

@@ -21,7 +21,7 @@ from nte_history_exporter.constants import (
 )
 from nte_history_exporter.decoder.boundary import select_continuous_run_from_page_1
 from nte_history_exporter.decoder.run import fmt_packet_time
-from nte_history_exporter.mappings import ARC_META
+from nte_history_exporter.mappings import ARC_META_BY_CASEFOLD
 from nte_history_exporter.decoder.structured_protocol import (
     StructuredProtocolAssembler,
     StructuredRecord,
@@ -95,7 +95,7 @@ def _parse_legacy_arc_response(response: bytes) -> list[dict[str, Any]]:
         timestamp_raw = response[pos : pos + 8]
         pos += 8
         arc_id = decode_arc_key(name_raw) or name_raw.hex()
-        meta = ARC_META.get(arc_id, {})
+        canonical_id, meta = _resolve_arc_metadata(arc_id)
         try:
             ticks, unix_seconds, timestamp_decoded = decode_arc_timestamp(timestamp_raw)
         except ValueError:
@@ -107,7 +107,7 @@ def _parse_legacy_arc_response(response: bytes) -> list[dict[str, Any]]:
                 "record_len": pos - start,
                 "reward_key_hex": name_raw.hex(),
                 "reward_type": "arc",
-                "reward_id": arc_id,
+                "reward_id": canonical_id,
                 "reward_name": meta.get("name", "UNKNOWN"),
                 "reward_rank": meta.get("rank", ""),
                 "type_key_hex": type_raw.hex(),
@@ -122,18 +122,15 @@ def _parse_legacy_arc_response(response: bytes) -> list[dict[str, Any]]:
     return records
 
 
-def _arc_metadata(arc_id: str) -> dict[str, Any]:
+def _resolve_arc_metadata(arc_id: str) -> tuple[str, dict[str, Any]]:
-    direct = ARC_META.get(arc_id)
+    canonical_id, meta = ARC_META_BY_CASEFOLD.get(arc_id.casefold(), (arc_id, {}))
-    if direct is not None:
+    return canonical_id, meta
-        return direct
-    folded = arc_id.casefold()
-    return next((meta for item_id, meta in ARC_META.items() if item_id.casefold() == folded), {})
 
 
 def structured_arc_rows(structured_rows: list[StructuredRecord]) -> list[dict[str, Any]]:
     records = []
     for structured in structured_rows:
-        meta = _arc_metadata(structured.item_id)
+        canonical_id, meta = _resolve_arc_metadata(structured.item_id)
         records.append(
             {
                 "record_start": structured.record_start,
@@ -141,7 +138,7 @@ def structured_arc_rows(structured_rows: list[StructuredRecord]) -> list[dict[st
                 "record_len": structured.record_end - structured.record_start,
                 "reward_key_hex": "",
                 "reward_type": "arc",
-                "reward_id": structured.item_id,
+                "reward_id": canonical_id,
                 "reward_name": meta.get("name", "UNKNOWN"),
                 "reward_rank": meta.get("rank", ""),
                 "type_key_hex": "",

src/nte_history_exporter/decoder/gear.py

@@ -0,0 +1,499 @@
+"""Decode cartridges, modules and character loadouts from the gameplay UDP flow.
+
+The game speaks Unreal's bit-packed net protocol. Each datagram is a bit stream:
+a 15-byte packet and bunch header, the payload, then two 1 bits - the packet
+handler's terminator and the bit writer's - zero-padded to the next byte
+boundary. Records routinely straddle datagram boundaries, so the payload bit
+ranges have to be spliced; concatenating whole datagrams splices the framing into
+the middle of a record and corrupts it.
+
+Inside the spliced stream, an item record reads:
+
+    <u32 length>  length == len(id) + 1   (self-validating frame check)
+    <ascii id>\0
+    <5 bytes> <8-byte instance id> ... <u32 level at +38 past the id>
+    ... <12-byte owner reference at +312 bits, all zero when unequipped> ...
+    then N stat entries, each:
+        <u32 length> <ascii stat name>\0 <4 zero bytes> <f32 value>
+
+Consecutive entries drift one bit apart, so entries are chained forward from the
+previous one rather than found by global search - that way a neighbouring
+record's entries can never be stolen.
+
+A character block is marked by its `GA_<Name>_Melee|Skill|UltraSkill|QTE` ability
+strings, with the character's level and breakthrough count at fixed bit offsets
+before it, and its equipped modules listed afterwards one entry per occupied
+grid cell.
+
+Every failure path returns empty rather than raising.
+"""
+
+from __future__ import annotations
+
+import re
+import struct
+from dataclasses import dataclass, field
+
+# --- framing ---------------------------------------------------------------
+
+HEADER_BITS = 120
+# Datagrams with no room for a payload are pure acks and carry nothing.
+MIN_PAYLOAD_BITS = 32
+
+# --- record layout ---------------------------------------------------------
+
+LEVEL_OFFSET_BYTES = 38
+MAX_ITEM_LEVEL = 20
+OWNER_OFFSET_BITS = 312
+OWNER_BYTES = 12
+# An item packs its entries tightly; anything further away belongs to something
+# else, so a wider scan would wander into unrelated data.
+CHAIN_SCAN_BITS = 1400
+# Observed entry spacing is (len+12) bytes plus a 1-bit separator, tightest at
+# ~145 bits. Advancing by a floor below that and rescanning avoids overshooting
+# entries that are shorter than their computed width - main stats run two bytes
+# shorter than substats.
+MIN_ENTRY_SPACING_BITS = 100
+MODULE_ENTRIES = 6
+CARTRIDGE_ENTRIES = 5
+
+# --- character layout ------------------------------------------------------
+
+# Bit offsets relative to the start of the character's first ability frame.
+CHARACTER_MARKER_OFFSET_BITS = -466
+CHARACTER_LEVEL_OFFSET_BITS = -434
+CHARACTER_BREAKTHROUGH_OFFSET_BITS = -402
+CHARACTER_MARKER = 67806
+MAX_CHARACTER_LEVEL = 90
+MAX_BREAKTHROUGHS = 10
+# A loadout cell entry stores doubled grid coordinates after the item id.
+CELL_ROW_OFFSET_BITS = 96
+CELL_COL_OFFSET_BITS = 128
+
+# --- vocabulary ------------------------------------------------------------
+
+# Ids are whitelisted: the length prefix validates a frame but cannot catch a
+# single flipped bit inside the name - `Ingantation` and `GetEfviciency` both
+# passed the length check before the whitelist went in.
+ITEM_ID = re.compile(
+    rb"(cell[234]_style[1-6]_\d_(?:Orange|Purple|Blue)"
+    # Psychically must precede Psyche: both ids exist and name different sets.
+    rb"|(?:Attack|Chaos|Cosmos|GetEfficiency|Heal|Incantation|Lakshana|Mag|Nature"
+    rb"|Psychically|Psyche|Shield)_(?:orange|purple|blue))\x00"
+)
+ABILITY_ID = re.compile(rb"GA_([A-Za-z0-9]+)_(?:Melee|Skill|UltraSkill|QTE)\x00")
+
+STAT_NAMES = frozenset({
+    "AtkAdd", "AtkUp", "DefAdd", "DefUp", "HPMaxAdd", "HPMaxUp", "CritBase",
+    "CritDamageBase", "MagBase", "UnbalIntensityBase", "HealUp",
+})
+
+# Substat values are fixed per stat and item size, with no roll variance, so any
+# value off this table means the parse went wrong. Main stats read 0.0.
+CANONICAL_VALUES = frozenset({
+    0.0, 200.0, 300.0, 400.0, 1000.0, 16.0, 24.0, 32.0, 80.0, 48.0, 64.0, 600.0, 800.0,
+    12.0, 18.0, 60.0, 36.0,
+    0.025, 0.0375, 0.05, 0.125, 0.035, 0.0525, 0.07, 0.075, 0.175, 0.02, 0.03, 0.04,
+    0.1, 0.06, 0.08, 0.2, 0.105, 0.14,
+})
+
+# A main stat's displayed value is never transmitted: it is base * (1 + level/5),
+# so +0 shows the base and +20 shows five times it.
+CARTRIDGE_MAIN_BASE = {
+    "HPMaxUp": 0.075, "AtkUp": 0.075, "DefUp": 0.105, "CritBase": 0.06,
+    "CritDamageBase": 0.12, "MagBase": 36.0, "UnbalIntensityBase": 36.0,
+    "DamageUpGeneralBase": 0.075,
+    # Healing Bonus reads 6.90% at +0 in game - the one main that is not round.
+    "HealUp": 0.069,
+}
+# Module mains scale per cell: 56 HP and 4.2 ATK each at +0. Measured in game as
+# Type III 168/12.6 -> 840/63 and Type IV 224/16.8 -> 1120/84. The client floors
+# flat values, which is why ATK displays as 12 and 16 at +0.
+MODULE_HP_PER_CELL = 56.0
+MODULE_ATK_PER_CELL = 4.2
+
+# Confirmed against everness's cartridge boxes, which name all twelve sets. The
+# three the capture has never carried are Psyche, Shield and Heal - NOT Blood,
+# Night and Kingdom, which were a guess and would have decoded as nothing.
+SET_NAMES = {
+    "Attack": "Shadow Creed", "Chaos": "Diabolos", "Cosmos": "Lost Radiance",
+    "GetEfficiency": "Speedy Hedgehog", "Incantation": "Crimson: Twin Butterflies",
+    "Lakshana": "Street Boxer", "Nature": "Fireflies and the Forest",
+    "Psychically": "Quiet Manor", "Mag": "Tiny Big Adventure",
+    "Psyche": "Devil's Blood: Curse", "Heal": "Thea's Night Tavern",
+    "Shield": "Kingdom's Guard",
+}
+
+MODULE_TYPE_NAMES = {2: "II", 3: "III", 4: "IV"}
+
+
+@dataclass(frozen=True)
+class GearItem:
+    instance: str
+    kind: str
+    item_id: str
+    level: int
+    rarity: str
+    main_stats: tuple[tuple[str, float | None], ...]
+    substats: tuple[tuple[str, float], ...]
+    owner_group: str | None
+    shape: str | None = None
+    module_type: str | None = None
+    set_name: str | None = None
+
+
+@dataclass(frozen=True)
+class GearCharacter:
+    key: str
+    level: int | None
+    breakthroughs: int | None
+    board: tuple[tuple[int, int], ...] = ()
+
+
+@dataclass(frozen=True)
+class GearSnapshot:
+    items: tuple[GearItem, ...] = ()
+    characters: tuple[GearCharacter, ...] = ()
+    warnings: tuple[str, ...] = field(default=())
+
+    @property
+    def cartridges(self) -> int:
+        return sum(item.kind == "cartridge" for item in self.items)
+
+    @property
+    def modules(self) -> int:
+        return sum(item.kind == "module" for item in self.items)
+
+    @property
+    def characters_with_loadouts(self) -> int:
+        return sum(bool(character.board) for character in self.characters)
+
+
+def is_stat_name(name: str) -> bool:
+    return name in STAT_NAMES or (name.startswith("DamageUp") and name.endswith("Base"))
+
+
+def bitshift(buf: bytes, shift: int) -> bytes:
+    if shift == 0:
+        return buf
+    return bytes(
+        ((buf[i] >> shift) | (buf[i + 1] << (8 - shift))) & 0xFF
+        for i in range(len(buf) - 1)
+    )
+
+
+class _BitAccumulator:
+    """Append arbitrary bit ranges; the result is a byte buffer, LSB-first."""
+
+    def __init__(self) -> None:
+        self.buf = bytearray()
+        self.nbits = 0
+
+    def append(self, data: bytes, start: int, nbits: int) -> None:
+        if nbits <= 0:
+            return
+        value = (int.from_bytes(data, "little") >> start) & ((1 << nbits) - 1)
+        offset = self.nbits % 8
+        if offset:
+            value <<= offset
+            raw = value.to_bytes((offset + nbits + 7) // 8, "little")
+            self.buf[-1] |= raw[0]
+            self.buf.extend(raw[1:])
+        else:
+            self.buf.extend(value.to_bytes((nbits + 7) // 8, "little"))
+        self.nbits += nbits
+        del self.buf[(self.nbits + 7) // 8 :]
+
+
+def payload_end_bit(datagram: bytes) -> int:
+    """First bit past the payload.
+
+    Unreal ends a packet with two 1 bits then zero-pads to a byte boundary, so
+    the payload stops one bit below the highest set bit of the last non-zero
+    byte.
+    """
+
+    i = len(datagram) - 1
+    while i >= 0 and datagram[i] == 0:
+        i -= 1
+    if i < 0:
+        return 0
+    return i * 8 + datagram[i].bit_length() - 2
+
+
+def splice_datagrams(datagrams) -> bytes:
+    """Concatenate every datagram's payload bits with the framing removed."""
+
+    accumulator = _BitAccumulator()
+    for datagram in datagrams:
+        end = payload_end_bit(datagram)
+        if end - HEADER_BITS >= MIN_PAYLOAD_BITS:
+            accumulator.append(datagram, HEADER_BITS, end - HEADER_BITS)
+    return bytes(accumulator.buf)
+
+
+class _BitView:
+    """Random access to a buffer at arbitrary bit offsets."""
+
+    def __init__(self, buf: bytes) -> None:
+        self.views = [bitshift(buf, shift) for shift in range(8)]
+
+    def at(self, bitpos: int) -> tuple[bytes, int]:
+        return self.views[bitpos % 8], bitpos // 8
+
+    def u32(self, bitpos: int) -> int | None:
+        view, offset = self.at(bitpos)
+        if offset + 4 > len(view):
+            return None
+        return struct.unpack_from("<I", view, offset)[0]
+
+    def f32(self, bitpos: int) -> float | None:
+        view, offset = self.at(bitpos)
+        if offset + 4 > len(view):
+            return None
+        return struct.unpack_from("<f", view, offset)[0]
+
+    def raw(self, bitpos: int, length: int) -> bytes:
+        view, offset = self.at(bitpos)
+        return view[offset : offset + length]
+
+    def text(self, bitpos: int, length: int) -> str | None:
+        view, offset = self.at(bitpos)
+        if offset + length > len(view):
+            return None
+        raw = view[offset : offset + length]
+        if not raw.endswith(b"\0"):
+            return None
+        try:
+            text = raw[:-1].decode("ascii")
+        except UnicodeDecodeError:
+            return None
+        return text if text.isprintable() else None
+
+
+def _validated_frames(view: _BitView, pattern):
+    """Every self-validating <u32 len><name\\0> frame, in bit order."""
+
+    found = []
+    for shift in range(8):
+        haystack = view.views[shift]
+        for match in pattern.finditer(haystack):
+            body = match.group()
+            if match.start() < 4:
+                continue
+            if struct.unpack_from("<I", haystack, match.start() - 4)[0] != len(body):
+                continue
+            found.append((
+                (match.start() - 4) * 8 + shift,   # frame start
+                match,
+                shift,
+                match.end() * 8 + shift,           # first bit past the name
+            ))
+    found.sort()
+    return found
+
+
+def read_entry(view: _BitView, bitpos: int) -> tuple[str, float] | None:
+    """Read one length-prefixed <name, f32> stat entry."""
+
+    declared = view.u32(bitpos)
+    if declared is None or not 4 <= declared <= 64:
+        return None
+    name = view.text(bitpos + 32, declared)
+    if name is None:
+        return None
+    # Layout: <u32 len><name\0><4 zero bytes><f32 value>
+    value = view.f32(bitpos + 32 + declared * 8 + 32)
+    if value is None:
+        return None
+    return name, round(value, 6)
+
+
+def _chain_entries(view: _BitView, start_bit: int, needed: int):
+    """Walk `needed` stat entries forward from just past the item id."""
+
+    cursor, entries = start_bit, []
+    while len(entries) < needed:
+        found = None
+        for delta in range(CHAIN_SCAN_BITS):
+            probe = read_entry(view, cursor + delta)
+            if probe and is_stat_name(probe[0]):
+                found = (probe, cursor + delta)
+                break
+        if not found:
+            return entries
+        entries.append(found[0])
+        cursor = found[1] + MIN_ENTRY_SPACING_BITS
+    return entries
+
+
+def _main_value(kind: str, stat: str, level: int, cells: int | None) -> float | None:
+    if kind == "module":
+        base = {
+            "HPMaxAdd": MODULE_HP_PER_CELL * (cells or 0),
+            "AtkAdd": MODULE_ATK_PER_CELL * (cells or 0),
+        }.get(stat)
+    else:
+        base = CARTRIDGE_MAIN_BASE.get(stat)
+        # Every elemental damage bonus shares the general bonus's base.
+        if base is None and stat.startswith("DamageUp") and stat.endswith("Base"):
+            base = CARTRIDGE_MAIN_BASE["DamageUpGeneralBase"]
+    if base is None:
+        return None
+    value = base * (1 + level / 5)
+    # Flat module stats display floored; percentages keep their precision.
+    return float(int(value)) if kind == "module" else round(value, 4)
+
+
+def _item_from_frame(view: _BitView, frame) -> GearItem | None:
+    bitpos, match, shift, id_end_bit = frame
+    item_id = match.group()[:-1].decode()
+    is_module = item_id.startswith("cell")
+    needed = MODULE_ENTRIES if is_module else CARTRIDGE_ENTRIES
+    main_count = 2 if is_module else 1
+
+    entries = _chain_entries(view, id_end_bit, needed)
+    if len(entries) != needed:
+        return None
+    # Substats never vary, so an off-table value means the chain went astray.
+    if not all(value in CANONICAL_VALUES for _, value in entries):
+        return None
+    # In every record confirmed against in-game values the main stats come
+    # first. A main appearing later means the chain wrapped into a neighbour.
+    if not all(value == 0.0 for _, value in entries[:main_count]):
+        return None
+    if sum(value == 0.0 for _, value in entries) != main_count:
+        return None
+
+    id_end_byte = match.end()
+    haystack = view.views[shift]
+    instance = haystack[id_end_byte + 5 : id_end_byte + 13]
+    if len(instance) != 8:
+        return None
+
+    level_offset = id_end_byte + LEVEL_OFFSET_BYTES
+    if level_offset + 4 > len(haystack):
+        return None
+    level = struct.unpack_from("<I", haystack, level_offset)[0]
+    if level > MAX_ITEM_LEVEL:
+        level = 0
+
+    owner_bit = id_end_bit + OWNER_OFFSET_BITS
+    # Gate on the leading word: trailing bytes of the field carry unrelated data
+    # on some records, so testing the whole span over-reports.
+    owner = view.raw(owner_bit, OWNER_BYTES) if view.u32(owner_bit) else b""
+
+    cells = int(item_id[4]) if is_module else None
+    mains = tuple(
+        (name, _main_value("module" if is_module else "cartridge", name, level, cells))
+        for name, value in entries if value == 0.0
+    )
+    substats = tuple((name, value) for name, value in entries if value != 0.0)
+    rarity = item_id.rsplit("_", 1)[1].lower()
+
+    if is_module:
+        shape = "_".join(item_id.split("_")[:2])
+        return GearItem(
+            instance=instance.hex(), kind="module", item_id=item_id, level=level,
+            rarity=rarity, main_stats=mains, substats=substats,
+            owner_group=owner.hex() or None,
+            shape=shape, module_type=MODULE_TYPE_NAMES.get(cells or 0),
+        )
+    return GearItem(
+        instance=instance.hex(), kind="cartridge", item_id=item_id, level=level,
+        rarity=rarity, main_stats=mains, substats=substats,
+        owner_group=owner.hex() or None,
+        set_name=SET_NAMES.get(item_id.rsplit("_", 1)[0]),
+    )
+
+
+def _character_blocks(view: _BitView) -> list[tuple[int, str]]:
+    """The first ability frame of each character, in stream order."""
+
+    blocks: list[tuple[int, str]] = []
+    for bitpos, match, _shift, _end in _validated_frames(view, ABILITY_ID):
+        name = match.group(1).decode()
+        if not blocks or blocks[-1][1] != name:
+            blocks.append((bitpos, name))
+    return blocks
+
+
+def _character_progress(view: _BitView, block_bit: int) -> tuple[int | None, int | None]:
+    """Level and breakthrough count, validated against the block's marker."""
+
+    if view.u32(block_bit + CHARACTER_MARKER_OFFSET_BITS) != CHARACTER_MARKER:
+        return None, None
+    level = view.u32(block_bit + CHARACTER_LEVEL_OFFSET_BITS)
+    breakthroughs = view.u32(block_bit + CHARACTER_BREAKTHROUGH_OFFSET_BITS)
+    if level is None or not 1 <= level <= MAX_CHARACTER_LEVEL:
+        level = None
+    if breakthroughs is None or breakthroughs > MAX_BREAKTHROUGHS:
+        breakthroughs = None
+    return level, breakthroughs
+
+
+def extract_gear(datagrams) -> GearSnapshot:
+    """Decode every item and character loadout in a captured UDP flow."""
+
+    try:
+        buf = splice_datagrams(datagrams)
+    except (ValueError, MemoryError):
+        return GearSnapshot()
+    if len(buf) < 64:
+        return GearSnapshot()
+
+    view = _BitView(buf)
+    frames = _validated_frames(view, ITEM_ID)
+    blocks = _character_blocks(view)
+
+    items: dict[str, GearItem] = {}
+    # An id frame that chains no stat entries is a loadout reference, not a
+    # record: the same id quoted inside a character's equipped list.
+    loadouts: dict[str, list[tuple[int, int, int]]] = {}
+    for frame in frames:
+        item = _item_from_frame(view, frame)
+        if item is not None:
+            items.setdefault(item.instance, item)
+            continue
+        bitpos, _match, _shift, id_end_bit = frame
+        row = view.u32(id_end_bit + CELL_ROW_OFFSET_BITS)
+        column = view.u32(id_end_bit + CELL_COL_OFFSET_BITS)
+        if not row or not column:
+            continue
+        owner_index = -1
+        for index, (block_bit, _name) in enumerate(blocks):
+            if block_bit > bitpos:
+                break
+            owner_index = index
+        if owner_index >= 0:
+            # Coordinates are doubled; a cell is one square, not four.
+            loadouts.setdefault(blocks[owner_index][1], []).append(
+                (bitpos, row // 2, column // 2)
+            )
+
+    # A character's block is transmitted more than once in a login - the active
+    # team appears early and again in the full roster - so merge by name and
+    # keep the first reading that validated.
+    merged: dict[str, GearCharacter] = {}
+    for block_bit, name in blocks:
+        level, breakthroughs = _character_progress(view, block_bit)
+        board = tuple(sorted({(r, c) for _bit, r, c in loadouts.get(name, [])}))
+        previous = merged.get(name)
+        if previous is None:
+            merged[name] = GearCharacter(name, level, breakthroughs, board)
+            continue
+        merged[name] = GearCharacter(
+            key=name,
+            level=previous.level if previous.level is not None else level,
+            breakthroughs=(
+                previous.breakthroughs
+                if previous.breakthroughs is not None
+                else breakthroughs
+            ),
+            board=previous.board or board,
+        )
+
+    return GearSnapshot(
+        items=tuple(items.values()),
+        characters=tuple(merged.values()),
+    )

src/nte_history_exporter/decoder/mystery_box.py

@@ -24,16 +24,11 @@ from nte_history_exporter.constants import (
 from nte_history_exporter.decoder.boundary import select_continuous_run_from_page_1
 from nte_history_exporter.decoder.protocol import infer_reward_type
 from nte_history_exporter.decoder.run import fmt_packet_time
-from nte_history_exporter.mappings import REWARDS_BY_ID
+from nte_history_exporter.mappings import REWARDS_BY_CASEFOLD
 
 DOTNET_TICKS_PER_SECOND = 10_000_000
 MAX_RECORDS_PER_BLOCK = 100
 MAX_REWARD_ID_LENGTH = 256
-REWARDS_BY_CASEFOLDED_ID = {
-    reward_id.casefold(): reward for reward_id, reward in REWARDS_BY_ID.items()
-}
-
-
 def is_mystery_box_history_request(content: bytes) -> bool:
     if len(content) < MYSTERY_BOX_HISTORY_REQUEST_LENGTH:
         return False
@@ -108,16 +103,15 @@ def _parse_view(data: bytes) -> list[dict[str, Any]]:
             timestamp_raw = data[pos : pos + 8]
             pos += 8
             ticks, unix_seconds, timestamp_decoded = _decode_timestamp(timestamp_raw)
-            reward = REWARDS_BY_ID.get(reward_id) or REWARDS_BY_CASEFOLDED_ID.get(
+            reward = REWARDS_BY_CASEFOLD.get(reward_id.casefold(), {})
-                reward_id.casefold(), {}
+            canonical_reward_id = reward.get("id", reward_id)
-            )
             rows.append(
                 {
                     "record_start": record_start,
                     "record_end": pos,
                     "record_len": pos - record_start,
                     "reward_type": reward.get("type") or infer_reward_type(reward_id),
-                    "reward_id": reward_id,
+                    "reward_id": canonical_reward_id,
                     "reward_name": reward.get("name", ""),
                     "reward_rank": reward.get("rank"),
                     "quantity": quantity,

src/nte_history_exporter/decoder/protocol.py

@@ -15,7 +15,7 @@ from nte_history_exporter.constants import (
     TIMESTAMP_TICKS_PER_SECOND,
     VALID_DICE_FIELDS,
 )
-from nte_history_exporter.mappings import REWARDS_BY_ID
+from nte_history_exporter.mappings import REWARDS_BY_CASEFOLD
 from nte_history_exporter.decoder.structured_protocol import (
     StructuredRecord,
     parse_structured_records,
@@ -61,11 +61,12 @@ def decode_reward_key(raw: bytes) -> str:
 def infer_reward_type(reward_id: str) -> str:
     if not reward_id:
         return ""
-    if reward_id.startswith("fork_"):
+    folded = reward_id.casefold()
+    if folded.startswith("fork_"):
         return "arc"
     if reward_id.isdigit():
         return "character"
-    if reward_id.startswith("Fashion_"):
+    if folded.startswith("fashion_"):
         return "cosmetic"
     return "item"
 
@@ -165,7 +166,7 @@ def classify_result_type(
 ) -> tuple[str, int | None]:
     if dice is None or dice_offset is None:
         return "unknown", None
-    if reward_id == "Dice_ticket_01" and WARP_PIECE_CHASE_PATTERN in chunk_without_marker:
+    if reward_id.casefold() == "dice_ticket_01" and WARP_PIECE_CHASE_PATTERN in chunk_without_marker:
         return "chase_reward", -4
     if dice == 0:
         return "points_gift", 0
@@ -182,13 +183,14 @@ def classify_result_type(
 
 
 def guess_quantity(chunk_hex: str, reward_id: str, result_type: str | None = None) -> int | None:
-    if reward_id == "Dice_ticket_01":
+    folded = reward_id.casefold()
+    if folded == "dice_ticket_01":
         if result_type == "chase_reward":
             return 30
         return 4
-    if reward_id == "DiceNormal":
+    if folded == "dicenormal":
         return 1
-    if reward_id == "Dice_ticket_02":
+    if folded == "dice_ticket_02":
         if "c8b0d4c0" in chunk_hex:
             return 50
         if "c8b0ccc0" in chunk_hex:
@@ -266,7 +268,8 @@ def _decode_aligned_response_records(response_content: bytes) -> list[dict[str,
         elif result_type == "chase_reward":
             dice = -4
             dice_raw = -4
-        reward = REWARDS_BY_ID.get(reward_id, {})
+        reward = _reward_metadata(reward_id)
+        canonical_reward_id = reward.get("id", reward_id)
         timestamp_raw = response_content[marker_offset + len(marker) : marker_offset + len(marker) + 8]
         timestamp_ticks, timestamp_unix, timestamp_decoded = decode_history_timestamp(timestamp_raw)
         chunk_hex = chunk.hex()
@@ -288,7 +291,7 @@ def _decode_aligned_response_records(response_content: bytes) -> list[dict[str,
                 "dice_offset_in_record": dice_offset,
                 "reward_key_hex": key_hex,
                 "reward_type": reward.get("type") or infer_reward_type(reward_id),
-                "reward_id": reward_id,
+                "reward_id": canonical_reward_id,
                 "reward_name": reward.get("name", ""),
                 "reward_rank": reward.get("rank"),
                 "quantity": guess_quantity(chunk_hex, reward_id, result_type),
@@ -335,8 +338,8 @@ def _enrich_heuristic_rows(
         return heuristic_rows
     for heuristic, structured in zip(heuristic_rows, structured_rows):
         if not heuristic.get("reward_id"):
-            heuristic["reward_id"] = structured.item_id
             reward = _reward_metadata(structured.item_id)
+            heuristic["reward_id"] = reward.get("id", structured.item_id)
             heuristic["reward_type"] = reward.get("type") or infer_reward_type(structured.item_id)
             heuristic["reward_name"] = reward.get("name", "")
             heuristic["reward_rank"] = reward.get("rank")
@@ -361,11 +364,7 @@ def _enrich_heuristic_rows(
 
 
 def _reward_metadata(reward_id: str) -> dict[str, Any]:
-    direct = REWARDS_BY_ID.get(reward_id)
+    return REWARDS_BY_CASEFOLD.get(reward_id.casefold(), {})
-    if direct is not None:
-        return direct
-    folded = reward_id.casefold()
-    return next((meta for item_id, meta in REWARDS_BY_ID.items() if item_id.casefold() == folded), {})
 
 
 def structured_monopoly_rows(structured_rows: list[StructuredRecord]) -> list[dict[str, Any]]:
@@ -373,6 +372,7 @@ def structured_monopoly_rows(structured_rows: list[StructuredRecord]) -> list[di
     for row_index, structured in enumerate(structured_rows, start=1):
         dice, dice_raw, result_type, result_source = _structured_result(structured.roll_points_raw)
         reward = _reward_metadata(structured.item_id)
+        canonical_reward_id = reward.get("id", structured.item_id)
         rows.append(
             {
                 "row": row_index,
@@ -391,7 +391,7 @@ def structured_monopoly_rows(structured_rows: list[StructuredRecord]) -> list[di
                 "dice_offset_in_record": None,
                 "reward_key_hex": "",
                 "reward_type": reward.get("type") or infer_reward_type(structured.item_id),
-                "reward_id": structured.item_id,
+                "reward_id": canonical_reward_id,
                 "reward_name": reward.get("name", ""),
                 "reward_rank": reward.get("rank"),
                 "quantity": structured.count,

src/nte_history_exporter/export/gear_export.py

@@ -0,0 +1,99 @@
+"""Gear export envelope.
+
+Kept out of ``json_export`` so the gear feature is a set of added files rather
+than edits to upstream ones - this repo tracks Golumpa's exporter and has to be
+able to merge it.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from nte_history_exporter import __version__
+from nte_history_exporter.constants import EXPORTER_NAME, GAME_NAME
+from nte_history_exporter.decoder.gear import GearCharacter, GearItem, GearSnapshot
+from nte_history_exporter.decoder.server_region import account_region_for_server
+
+
+def build_gear_export_json(
+    snapshot: GearSnapshot,
+    *,
+    source: str = "packet_capture",
+    capture_source: str | None = None,
+    user_uid: str | None = None,
+    server_id: str | None = None,
+) -> dict[str, Any]:
+    scan = {
+        "cartridges": snapshot.cartridges,
+        "modules": snapshot.modules,
+        "characters": len(snapshot.characters),
+        "characters_with_loadouts": snapshot.characters_with_loadouts,
+        "warnings": list(snapshot.warnings),
+    }
+
+    export: dict[str, Any] = {
+        "format": "nte-gear-export",
+        "format_version": 1,
+        "game": GAME_NAME,
+        "source": source,
+    }
+    if capture_source:
+        export["capture_source"] = capture_source
+    export["exporter"] = {"name": EXPORTER_NAME, "version": __version__}
+    export["scan"] = scan
+
+    normalized_user_uid = user_uid.strip() if user_uid else ""
+    normalized_server_id = str(server_id).strip() if server_id else ""
+    if normalized_user_uid:
+        export["user_uid"] = normalized_user_uid
+    if normalized_server_id:
+        export["server_id"] = normalized_server_id
+        account_region = account_region_for_server(normalized_server_id)
+        if account_region:
+            export["account_region"] = account_region
+
+    export["characters"] = [
+        _gear_character_for_export(character) for character in snapshot.characters
+    ]
+    export["items"] = [_gear_item_for_export(item) for item in snapshot.items]
+    return export
+
+
+def _gear_character_for_export(character: GearCharacter) -> dict[str, Any]:
+    record: dict[str, Any] = {
+        "key": character.key,
+        "level": character.level,
+        "breakthroughs": character.breakthroughs,
+    }
+    # Only emitted when a loadout was actually seen; an empty board and an
+    # unequipped character are different states and must not collapse.
+    if character.board:
+        record["board"] = [[row, column] for row, column in character.board]
+    # The capture never ties an owner group to a character name, so the app
+    # resolves this itself rather than being handed a guess.
+    record["owner_group"] = None
+    return record
+
+
+def _gear_item_for_export(item: GearItem) -> dict[str, Any]:
+    record: dict[str, Any] = {
+        "instance": item.instance,
+        "kind": item.kind,
+        "item_id": item.item_id,
+    }
+    if item.shape:
+        record["shape"] = item.shape
+    if item.module_type:
+        record["module_type"] = item.module_type
+    if item.set_name:
+        record["set"] = item.set_name
+    record["level"] = item.level
+    record["rarity"] = item.rarity
+    record["main_stats"] = [
+        {"stat": stat, "value": value} for stat, value in item.main_stats
+    ]
+    record["substats"] = [
+        {"stat": stat, "value": value} for stat, value in item.substats
+    ]
+    record["owner_group"] = item.owner_group
+    return record

src/nte_history_exporter/export/json_export.py

@@ -94,13 +94,11 @@ def build_achievement_export_json(
             "total_achievements": 0,
             "completed_achievements": 0,
             "in_progress_achievements": 0,
-            "not_started_achievements": 0,
         },
         "playstation": {
             "total_achievements": 0,
             "completed_achievements": 0,
             "in_progress_achievements": 0,
-            "not_started_achievements": 0,
         },
     }
     for achievement in achievement_records:
@@ -110,10 +108,8 @@ def build_achievement_export_json(
         summary["total_achievements"] += 1
         if achievement.completed:
             summary["completed_achievements"] += 1
-        elif achievement.progress:
-            summary["in_progress_achievements"] += 1
         else:
-            summary["not_started_achievements"] += 1
+            summary["in_progress_achievements"] += 1
         category, separator, _number = achievement_id.partition("_")
         category_key = category.casefold() if separator else "uncategorized"
         record = {

src/nte_history_exporter/live_capture/gear_collector.py

@@ -0,0 +1,191 @@
+"""Gear capture, held in one object so ``runner`` only has to call it.
+
+This repo tracks Golumpa's exporter, so the gear feature is deliberately shaped
+as added files plus a handful of one-line calls in ``runner.run_live_capture``
+rather than as edits scattered through the capture loop. Everything gear needs -
+flow accumulation, the mid-capture decode, the fallback scan, the console line
+and the export file - lives here.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from datetime import datetime
+from pathlib import Path
+
+from nte_history_exporter.console import BOLD, DIM, GREEN, style
+from nte_history_exporter.decoder.gear import GearSnapshot, extract_gear
+from nte_history_exporter.export.gear_export import build_gear_export_json
+
+
+Flow = tuple[str, int, str, int]
+
+# The login gear burst is roughly 500 KB; the cap only exists so a long
+# session on a busy flow cannot grow without bound.
+MAX_UDP_FLOW_BYTES = 8 * 1024 * 1024
+# Decoding is linear in flow size, so only the few busiest are worth trying.
+MAX_GEAR_FLOW_ATTEMPTS = 4
+# A gear burst arrives in a rush and then stops, so "this flow has gone quiet"
+# is the only mid-stream marker there is. Attempts are gated on the flow being
+# large enough to hold a burst, having been idle, and having grown since it was
+# last tried, so a quiet flow is never decoded twice for the same bytes.
+GEAR_LIVE_MIN_FLOW_BYTES = 64 * 1024
+GEAR_LIVE_IDLE_SECONDS = 1.0
+GEAR_LIVE_MAX_ATTEMPTS = 8
+
+
+def print_gear_captured(
+    cartridges: int,
+    modules: int,
+    characters: int,
+    characters_with_loadouts: int,
+) -> None:
+    print(style("  + ", GREEN, BOLD) + "Gear captured")
+    print(
+        style(
+            f"      Inventory    {cartridges} cartridges, {modules} modules",
+            DIM,
+        )
+    )
+    print(
+        style(
+            f"      Characters   {characters} seen, "
+            f"{characters_with_loadouts} with loadouts",
+            DIM,
+        )
+    )
+
+
+def gear_path(user_uid: str | None = None) -> Path:
+    export_dir = Path("exports")
+    export_dir.mkdir(parents=True, exist_ok=True)
+    uid_prefix = _safe_filename_part(user_uid) if user_uid else "unknown"
+    base = export_dir / f"{uid_prefix}_Gear_{datetime.now():%Y%m%d_%H%M%S}"
+    path = base.with_suffix(".json")
+    counter = 2
+    while path.exists():
+        path = export_dir / f"{base.name}_{counter}.json"
+        counter += 1
+    return path
+
+
+def _safe_filename_part(value: str | None) -> str:
+    # Deliberately a copy of runner's helper rather than an import: importing
+    # runner from here would be circular, and the rule is four lines long.
+    if not value:
+        return "unknown"
+    cleaned = "".join(ch for ch in value.strip() if ch.isalnum() or ch in ("-", "_"))
+    return cleaned or "unknown"
+
+
+class GearCollector:
+    """Accumulates inbound UDP and decodes gear out of it.
+
+    Call ``poll`` once per capture iteration (above any ``packet is None``
+    guard - the pcap read timeout is what keeps it firing after the burst goes
+    quiet), ``observe`` for each packet, ``finalize`` after the loop, and
+    ``write`` to emit the export.
+    """
+
+    def __init__(self, local_ip: str | None) -> None:
+        self._local_ip = local_ip
+        self._flows: dict[Flow, list[bytes]] = {}
+        self._flow_bytes: dict[Flow, int] = {}
+        self._flow_last_seen: dict[Flow, float] = {}
+        self._attempted_bytes: dict[Flow, int] = {}
+        self._attempts = 0
+        self.snapshot: GearSnapshot | None = None
+
+    def observe(self, packet) -> None:
+        """Keep an inbound UDP payload whole.
+
+        Datagram boundaries carry the framing, so payloads are spliced at
+        decode time and never concatenated here.
+        """
+        if (
+            packet is None
+            or packet.protocol != "udp"
+            or not packet.payload
+            or packet.dst_ip != self._local_ip
+        ):
+            return
+        flow: Flow = (packet.src_ip, packet.src_port, packet.dst_ip, packet.dst_port)
+        if self._flow_bytes.get(flow, 0) + len(packet.payload) > MAX_UDP_FLOW_BYTES:
+            return
+        self._flows.setdefault(flow, []).append(packet.payload)
+        self._flow_bytes[flow] = self._flow_bytes.get(flow, 0) + len(packet.payload)
+        self._flow_last_seen[flow] = time.monotonic()
+
+    def poll(self) -> None:
+        """Decode a flow that has stopped growing, and report it immediately."""
+        if self.snapshot is not None or self._attempts >= GEAR_LIVE_MAX_ATTEMPTS:
+            return
+        now = time.monotonic()
+        candidates = [
+            flow
+            for flow, size in self._flow_bytes.items()
+            if size >= GEAR_LIVE_MIN_FLOW_BYTES
+            and self._attempted_bytes.get(flow) != size
+            and now - self._flow_last_seen.get(flow, now) >= GEAR_LIVE_IDLE_SECONDS
+        ]
+        candidates.sort(key=self._flow_bytes.__getitem__, reverse=True)
+
+        for flow in candidates[:MAX_GEAR_FLOW_ATTEMPTS]:
+            self._attempted_bytes[flow] = self._flow_bytes[flow]
+            self._attempts += 1
+            snapshot = extract_gear(self._flows[flow])
+            if snapshot.items:
+                self._report(snapshot)
+                return
+
+    def finalize(self) -> None:
+        """Fallback scan for a flow that never idled.
+
+        Normally ``poll`` has already decoded the burst. Rescanning per packet
+        instead of only here would be quadratic in flow size.
+        """
+        if self.snapshot is not None:
+            return
+        # The gameplay flow dwarfs the others, but history traffic can outweigh
+        # a short login burst, so size only sets the order - the first flow that
+        # actually decodes wins.
+        busiest = sorted(self._flow_bytes, key=self._flow_bytes.__getitem__, reverse=True)
+        for flow in busiest[:MAX_GEAR_FLOW_ATTEMPTS]:
+            snapshot = extract_gear(self._flows[flow])
+            if snapshot.items:
+                self._report(snapshot)
+                return
+
+    def write(
+        self,
+        *,
+        capture_source: str | None = None,
+        user_uid: str | None = None,
+        server_id: str | None = None,
+        source: str = "live_capture",
+    ) -> Path | None:
+        if self.snapshot is None:
+            return None
+        path = gear_path(user_uid)
+        export = build_gear_export_json(
+            self.snapshot,
+            source=source,
+            capture_source=capture_source,
+            user_uid=user_uid,
+            server_id=server_id,
+        )
+        path.write_text(
+            json.dumps(export, ensure_ascii=False, indent=2) + "\n",
+            encoding="utf-8",
+        )
+        return path
+
+    def _report(self, snapshot: GearSnapshot) -> None:
+        self.snapshot = snapshot
+        print_gear_captured(
+            snapshot.cartridges,
+            snapshot.modules,
+            len(snapshot.characters),
+            snapshot.characters_with_loadouts,
+        )

src/nte_history_exporter/live_capture/runner.py

@@ -21,6 +21,7 @@ from nte_history_exporter.export.json_export import (
 )
 from nte_history_exporter.live_capture.backends import open_capture_backend
 from nte_history_exporter.live_capture.diagnostics import new_diagnostics_path, write_capture_diagnostics
+from nte_history_exporter.live_capture.gear_collector import GearCollector
 from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket
 from nte_history_exporter.live_capture.stop_key import StopKeyMonitor
 from nte_history_exporter.live_capture.windows_raw import detect_local_ipv4
@@ -88,14 +89,17 @@ def run_live_capture(
     active_gap_notices: set[str] = set()
     tcp_segments: dict[tuple[str, int, str, int], list[tuple[int, bytes]]] = {}
     achievement_records = []
+    gear = GearCollector(local_ip)
 
     try:
         with StopKeyMonitor() as stop_key:
             for packet in capture.packets():
                 if stop_key.pressed():
                     break
+                gear.poll()  # above the guard: the pcap read timeout drives it
                 if packet is None:
                     continue
+                gear.observe(packet)
                 if (
                     not achievement_records
                     and packet.protocol == "tcp"
@@ -111,12 +115,27 @@ def run_live_capture(
                             reassemble_tcp_segments(tcp_segments[flow])
                         )
                         if achievement_records:
-                            completed = [record for record in achievement_records if record.completed]
+                            in_game = [
+                                record
+                                for record in achievement_records
+                                if not record.achievement_id.casefold().startswith(
+                                    "playstation_"
+                                )
+                            ]
+                            playstation = [
+                                record
+                                for record in achievement_records
+                                if record.achievement_id.casefold().startswith(
+                                    "playstation_"
+                                )
+                            ]
                             console.print_achievements_captured(
-                                len(completed),
+                                sum(record.completed for record in in_game),
+                                sum(record.status == "in_progress" for record in in_game),
+                                sum(record.completed for record in playstation),
                                 sum(
-                                    record.achievement_id.lower().startswith("playstation_")
+                                    record.status == "in_progress"
-                                    for record in completed
+                                    for record in playstation
                                 ),
                             )
                 pair_count_before = len(session.pairs)
@@ -161,6 +180,8 @@ def run_live_capture(
         stats = capture.stats()
         capture.close()
 
+    gear.finalize()
+
     exports = []
     achievement_path = None
     diagnostics_path = None
@@ -168,7 +189,9 @@ def run_live_capture(
         diagnostics_path = new_diagnostics_path()
         write_capture_diagnostics(diagnostics_path, session.diagnostic_report())
     resolved_user_uid = user_uid or session.user_uid
-    if (session.kinds_seen() or achievement_records) and not resolved_user_uid:
+    if (
+        session.kinds_seen() or achievement_records or gear.snapshot
+    ) and not resolved_user_uid:
         resolved_user_uid = console.prompt_user_uid()
     if achievement_records:
         achievement_path = _achievement_path(resolved_user_uid)
@@ -183,6 +206,11 @@ def run_live_capture(
             json.dumps(achievement_export, ensure_ascii=False, indent=2) + "\n",
             encoding="utf-8",
         )
+    gear_path = gear.write(
+        capture_source=CAPTURE_SOURCE_LABELS.get(capture.name, capture.name),
+        user_uid=resolved_user_uid,
+        server_id=session.server_id,
+    )
     resolved_server_id = session.server_id
     if session.kinds_seen() and not resolved_server_id:
         resolved_server_id = console.prompt_server_id()
@@ -236,16 +264,23 @@ def run_live_capture(
             console.print_note("No pull history was captured (this is fine if you only wanted achievements).")
             print()
             console.print_note(f"Export written: {achievement_path}")
+        elif gear_path is not None:
+            console.print_success("Gear export complete.")
+            console.print_note("No pull history was captured (this is fine if you only wanted gear).")
+            print()
         else:
             console.print_problem("No history pages were captured.")
             console.print_note("Reopen a supported history screen and scroll from page 1.")
             console.print_note("If no page messages appear, return to the main menu and re-enter the game.")
+        if gear_path is not None:
+            console.print_note(f"Export written: {gear_path}")
         if diagnostics_path is not None:
             console.print_note(f"Diagnostics written: {diagnostics_path}")
         return {
             "exports": [],
             "diagnostics_path": diagnostics_path,
             "achievement_path": achievement_path,
+            "gear_path": gear_path,
         }
 
     for item in exports:
@@ -267,6 +302,8 @@ def run_live_capture(
     if achievement_path is not None:
         print()
         console.print_note(f"Export written: {achievement_path}")
+    if gear_path is not None:
+        console.print_note(f"Export written: {gear_path}")
     if diagnostics_path is not None:
         console.print_note(f"Diagnostics written: {diagnostics_path}")
 
@@ -282,6 +319,7 @@ def run_live_capture(
         "exports": exports,
         "diagnostics_path": diagnostics_path,
         "achievement_path": achievement_path,
+        "gear_path": gear_path,
     }
 
 

src/nte_history_exporter/mapping_update.py

@@ -324,9 +324,16 @@ def _build_item_mapping(
             candidate_id.isdigit()
             or folded.startswith("fork_")
             or folded.startswith("characterawaken_")
-            or folded.startswith("fashion_vehicle_")
         ):
             continue
+        if folded.startswith("fashion_vehicle_"):
+            illustration = tables["illustrations"].get(candidate_id)
+            if illustration is not None:
+                vehicle_meta = _normalise_vehicle_livery(candidate_id, illustration, translations)
+                if vehicle_meta is None:
+                    continue
+                result[candidate_id] = vehicle_meta
+                continue
         sources = (inventory, capital, appearances, vehicle_inventory)
         match = next((source.get(folded) for source in sources if folded in source), None)
         if match is None:
@@ -343,6 +350,31 @@ def _build_item_mapping(
     return dict(sorted(result.items(), key=lambda pair: (pair[0].casefold(), pair[0])))
 
 
+def _normalise_vehicle_livery(
+    item_id: str,
+    row: Any,
+    translations: dict[str, list[tuple[str, str]]],
+) -> dict[str, str] | None:
+    """Resolve a livery from its illustration, ignoring mismatched placeholder rows."""
+    if not isinstance(row, dict):
+        raise MappingUpdateError(f"NTE_Assets row {item_id} must be an object")
+    head_icon = row.get("HeadIcon")
+    asset_path = head_icon.get("AssetPathName") if isinstance(head_icon, dict) else None
+    if not isinstance(asset_path, str) or not asset_path:
+        raise MappingUpdateError(f"NTE_Assets vehicle livery row {item_id} has no HeadIcon")
+    icon_id = asset_path.rsplit("/", 1)[-1].split(".", 1)[0]
+    if icon_id.casefold() != item_id.casefold():
+        return None
+    item_name = row.get("ItemName_Override")
+    if not isinstance(item_name, dict):
+        raise MappingUpdateError(f"NTE_Assets vehicle livery row {item_id} has no ItemName_Override")
+    return {
+        "type": "cosmetic",
+        "name": _translate_name(item_id, item_name, translations).strip(),
+        "rank": "S",
+    }
+
+
 def _build_achievement_mapping(
     rows: dict[str, Any],
     translations: dict[str, list[tuple[str, str]]],

src/nte_history_exporter/mappings.py

@@ -14,6 +14,9 @@ def load_mapping_file(filename: str) -> dict[str, Any]:
 
 
 ARC_META: dict[str, dict[str, Any]] = load_mapping_file("arcs.json")
+ARC_META_BY_CASEFOLD = {
+    arc_id.casefold(): (arc_id, meta) for arc_id, meta in ARC_META.items()
+}
 CHARACTERS: dict[str, dict[str, Any]] = load_mapping_file("characters.json")
 ITEMS: dict[str, dict[str, Any]] = load_mapping_file("items.json")
 ACHIEVEMENTS: dict[str, dict[str, Any]] = load_mapping_file("achievements.json")
@@ -36,3 +39,6 @@ for _character_id, _meta in CHARACTERS.items():
     }
 for _item_id, _meta in ITEMS.items():
     REWARDS_BY_ID[_item_id] = {"type": _meta["type"], "id": _item_id, "name": _meta["name"], "rank": _meta.get("rank")}
+REWARDS_BY_CASEFOLD = {
+    reward_id.casefold(): reward for reward_id, reward in REWARDS_BY_ID.items()
+}

tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json

@@ -1 +1 @@
-[{"Rows":{"1003":{},"1099":{},"fork_alpha":{},"fork_new":{},"DIceNormal":{},"Dice_ticket_03":{},"Fashion_glide_2000":{},"Characterawaken_1003":{}}}]
+[{"Rows":{"1003":{},"1099":{},"fork_alpha":{},"fork_new":{},"DIceNormal":{},"Dice_ticket_03":{},"Fashion_glide_2000":{},"Characterawaken_1003":{},"Fashion_vehicle_1010_V008":{"HeadIcon":{"AssetPathName":"/Game/UI/UI_Icon/Item/Fashion_vehicle_1010_V008.Fashion_vehicle_1010_V008"},"ItemName_Override":{"TableId":"/Game/Text/ST_VehicleData.ST_VehicleData","Key":"vehicle_1010_livery"}},"Fashion_vehicle_1072_V006":{"HeadIcon":{"AssetPathName":"/Game/UI/UI_Icon/Item/Fashion_vehicle_1036_V043.Fashion_vehicle_1036_V043"},"ItemName_Override":{"TableId":"/Game/Text/ST_VehicleData.ST_VehicleData","Key":"vehicle_placeholder_livery"}}}}]

tests/fixtures/nte_assets/Localization/en/game.json

@@ -18,5 +18,9 @@
   },
   "ST_Appearance": {
     "glider_2000": "Inventory Glider"
+  },
+  "ST_VehicleData": {
+    "vehicle_1010_livery": "Tiger Incoming! - Livery",
+    "vehicle_placeholder_livery": "Sandy Mirage - Livery"
   }
 }

tests/test_achievement.py

@@ -28,7 +28,7 @@ class AchievementPacketTests(unittest.TestCase):
         self.assertEqual(records[0].achievement_id, "Battle_30")
         self.assertTrue(records[0].completed)
         self.assertEqual(records[1].achievement_id, "Playstation_035")
-        self.assertEqual(records[1].status, "not_started")
+        self.assertEqual(records[1].status, "in_progress")
 
     def test_reassembles_tcp_data(self):
         self.assertEqual(

tests/test_arc_decoding.py

@@ -1,8 +1,14 @@
 from tests.support import *  # noqa: F401,F403
-from nte_history_exporter.decoder.arc import annotate_arc_groups
+from nte_history_exporter.decoder.arc import annotate_arc_groups, _resolve_arc_metadata
 
 
 class ArcDecodingTests(unittest.TestCase):
+    def test_arc_mapping_lookup_is_case_insensitive_and_canonical(self):
+        reward_id, metadata = _resolve_arc_metadata("fork_Wushoutieyu")
+
+        self.assertEqual(reward_id, "fork_wushoutieyu")
+        self.assertEqual(metadata, {"name": "Raging Flames", "rank": "S"})
+
     def test_page_split_timestamp_variants_form_one_ten_pull(self):
         rows = [
             {

tests/test_export_contract.py

@@ -1,11 +1,16 @@
 from tests.support import *  # noqa: F401,F403
 from nte_history_exporter import __version__
+from nte_history_exporter.constants import EXPORTER_VERSION
 from nte_history_exporter.decoder.achievement import AchievementRecord
 from nte_history_exporter.export.json_export import build_achievement_export_json
 from nte_history_exporter.live_capture.runner import _achievement_path
 
 
 class ExportContractTests(unittest.TestCase):
+    def test_public_version_references_match(self):
+        self.assertEqual(__version__, "0.4.0")
+        self.assertEqual(EXPORTER_VERSION, __version__)
+
     def test_sanitized_export_omits_raw_packet_fields(self):
         annotated = annotate_groups(fixture_session().build_rows("permanent"))
         export = build_export_json(annotated, [])
@@ -153,13 +158,11 @@ class ExportContractTests(unittest.TestCase):
                     "total_achievements": 2,
                     "completed_achievements": 1,
                     "in_progress_achievements": 1,
-                    "not_started_achievements": 0,
                 },
                 "playstation": {
                     "total_achievements": 1,
                     "completed_achievements": 0,
-                    "in_progress_achievements": 0,
+                    "in_progress_achievements": 1,
-                    "not_started_achievements": 1,
                 },
             },
         )

tests/test_gear_decoding.py

@@ -0,0 +1,747 @@
+"""Gear decoding, driven entirely by synthetic packets.
+
+The wire format is fully understood, so every packet here is constructed from
+the spec rather than captured: a 120-bit header, payload bits, two 1 bits and
+zero padding. No real capture is used - see tests/fixtures/README.md.
+"""
+
+import contextlib
+import io
+import json
+import os
+import struct
+import tempfile
+import unittest
+
+from nte_history_exporter import console
+
+from nte_history_exporter import __version__
+from nte_history_exporter.decoder.gear import (
+    HEADER_BITS,
+    extract_gear,
+    payload_end_bit,
+    splice_datagrams,
+)
+from nte_history_exporter.export.gear_export import build_gear_export_json
+from nte_history_exporter.live_capture import gear_collector, runner
+from nte_history_exporter.live_capture.windows_raw import ParsedIpPacket
+
+from tests import support
+
+
+# --- synthetic wire construction -------------------------------------------
+
+
+def string_frame(text: bytes) -> bytes:
+    """A self-validating <u32 len><ascii\\0> frame."""
+    return struct.pack("<I", len(text) + 1) + text + b"\0"
+
+
+def stat_entry(name: bytes, value: float) -> bytes:
+    """<u32 len><name\\0><4 zero bytes><f32 value>."""
+    return string_frame(name) + b"\0\0\0\0" + struct.pack("<f", value)
+
+
+def poke(buf: bytearray, byte_offset: int, raw: bytes) -> None:
+    if len(buf) < byte_offset + len(raw):
+        buf.extend(b"\0" * (byte_offset + len(raw) - len(buf)))
+    buf[byte_offset : byte_offset + len(raw)] = raw
+
+
+def poke_u32_at_bit(buf: bytearray, bitpos: int, value: int) -> None:
+    """Write a little-endian u32 at an arbitrary bit offset."""
+    needed = (bitpos + 32 + 7) // 8 + 1
+    if len(buf) < needed:
+        buf.extend(b"\0" * (needed - len(buf)))
+    whole = int.from_bytes(buf, "little")
+    whole &= ~(0xFFFFFFFF << bitpos)
+    whole |= value << bitpos
+    packed = whole.to_bytes(len(buf) + 8, "little")
+    buf[:] = packed[: len(buf)]
+
+
+def item_record(
+    item_id: bytes,
+    *,
+    level: int = 20,
+    entries=(),
+    instance: bytes = b"\x11\x22\x33\x44\x55\x66\x77\x88",
+    owner: bytes | None = None,
+) -> bytes:
+    """One item record laid out at the documented offsets past the id frame."""
+    body = bytearray(b"\0" * 64)
+    poke(body, 5, instance)
+    # Level is a u32 at +38 bytes; the owner reference starts at +39 bits-wise
+    # (312 bits) and its first three bytes are always zero, so the two fields
+    # coexist exactly as they do on the wire.
+    poke(body, 38, struct.pack("<I", level))
+    if owner is not None:
+        assert owner[:3] == b"\0\0\0", "owner reference starts with three zero bytes"
+        poke(body, 39, owner)
+    chain = b"".join(stat_entry(name, value) for name, value in entries)
+    return string_frame(item_id) + bytes(body) + chain
+
+
+def character_block(name: bytes, *, level: int, breakthroughs: int) -> bytes:
+    """An ability frame with the progress fields at their negative offsets."""
+    lead = 128  # bytes of room for the fields that sit before the frame
+    buf = bytearray(b"\0" * lead)
+    frame = string_frame(b"GA_" + name + b"_Melee")
+    block_bit = lead * 8
+    poke_u32_at_bit(buf, block_bit - 466, 67806)
+    poke_u32_at_bit(buf, block_bit - 434, level)
+    poke_u32_at_bit(buf, block_bit - 402, breakthroughs)
+    return bytes(buf) + frame
+
+
+def loadout_cell(item_id: bytes, row: int, column: int) -> bytes:
+    """An id frame quoted with board coordinates and no stat chain."""
+    body = bytearray(b"\0" * 32)
+    # Coordinates sit 96 and 128 bits past the id and are stored doubled.
+    poke(body, 12, struct.pack("<I", row * 2))
+    poke(body, 16, struct.pack("<I", column * 2))
+    return string_frame(item_id) + bytes(body)
+
+
+def pack_datagrams(payload: bytes, *, chunk_bits: int = 1 << 20, first_seq: int = 1):
+    """Split a payload bit stream into framed datagrams."""
+    total_bits = len(payload) * 8
+    whole = int.from_bytes(payload, "little")
+    datagrams, start, seq = [], 0, first_seq
+    while start < total_bits:
+        nbits = min(chunk_bits, total_bits - start)
+        chunk = (whole >> start) & ((1 << nbits) - 1)
+
+        header = bytearray(15)
+        header[0:3] = b"\xaa\xbb\xcc"
+        # 14-bit packet sequence at bit 24.
+        header[3] = seq & 0xFF
+        header[4] = (seq >> 8) & 0x3F
+        header[12:15] = b"\x64\x06\xee"
+
+        # payload bits, then the two 1 bits Unreal terminates with.
+        value = int.from_bytes(header, "little")
+        value |= chunk << HEADER_BITS
+        value |= 0b11 << (HEADER_BITS + nbits)
+        width = (HEADER_BITS + nbits + 2 + 7) // 8
+        datagrams.append(value.to_bytes(width, "little"))
+
+        start += nbits
+        seq = (seq + 1) & 0x3FFF
+    return datagrams
+
+
+MODULE_ENTRIES = (
+    (b"HPMaxAdd", 0.0),
+    (b"AtkAdd", 0.0),
+    (b"CritDamageBase", 0.06),
+    (b"CritBase", 0.03),
+    (b"AtkUp", 0.04),
+    (b"HPMaxUp", 0.04),
+)
+
+CARTRIDGE_ENTRIES = (
+    (b"AtkUp", 0.0),
+    (b"CritDamageBase", 0.06),
+    (b"CritBase", 0.03),
+    (b"DefUp", 0.07),
+    (b"HPMaxUp", 0.04),
+)
+
+OWNER = b"\0\0\0\x0d\x0d\x20\x3d\x6d\x00\x00\x00\x00"
+
+
+class FramingTests(unittest.TestCase):
+    def test_payload_end_bit_finds_the_double_terminator(self):
+        [datagram] = pack_datagrams(b"\x01\x02\x03\x04")
+        self.assertEqual(payload_end_bit(datagram), HEADER_BITS + 32)
+
+    def test_splice_round_trips_a_payload_across_datagrams(self):
+        payload = bytes(range(256)) * 3
+        # A chunk width that is not a multiple of 8 forces the accumulator to
+        # rejoin the stream at a bit offset, as the real capture does.
+        datagrams = pack_datagrams(payload, chunk_bits=333)
+        self.assertGreater(len(datagrams), 1)
+        self.assertEqual(splice_datagrams(datagrams)[: len(payload)], payload)
+
+    def test_pure_acks_are_skipped(self):
+        payload = b"\x01\x02\x03\x04\x05\x06\x07\x08"
+        ack = bytes(12)
+        datagrams = pack_datagrams(payload)
+        self.assertEqual(splice_datagrams([ack] + datagrams + [ack])[:8], payload)
+
+
+class ItemRecordTests(unittest.TestCase):
+    def decode(self, payload: bytes, **kwargs):
+        return extract_gear(pack_datagrams(b"\0" * 8 + payload, **kwargs))
+
+    def test_module_record_decodes(self):
+        snapshot = self.decode(
+            item_record(b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES)
+        )
+        [item] = snapshot.items
+        self.assertEqual(item.kind, "module")
+        self.assertEqual(item.item_id, "cell3_style6_1_Orange")
+        self.assertEqual(item.shape, "cell3_style6")
+        self.assertEqual(item.module_type, "III")
+        self.assertEqual(item.level, 20)
+        self.assertEqual(item.rarity, "orange")
+        self.assertEqual(item.instance, "1122334455667788")
+        self.assertIsNone(item.owner_group)
+        # Both mains are computed from the cell count, never transmitted.
+        self.assertEqual(dict(item.main_stats), {"HPMaxAdd": 840.0, "AtkAdd": 63.0})
+        self.assertEqual(len(item.substats), 4)
+
+    def test_cartridge_record_decodes_with_its_set(self):
+        snapshot = self.decode(
+            item_record(b"Incantation_orange", level=20, entries=CARTRIDGE_ENTRIES)
+        )
+        [item] = snapshot.items
+        self.assertEqual(item.kind, "cartridge")
+        self.assertIsNotNone(item.set_name)
+        self.assertEqual(len(item.main_stats), 1)
+        self.assertEqual(len(item.substats), 4)
+
+    def test_owner_reference_is_read_when_equipped(self):
+        snapshot = self.decode(
+            item_record(
+                b"cell3_style6_1_Orange",
+                level=20,
+                entries=MODULE_ENTRIES,
+                owner=OWNER,
+            )
+        )
+        [item] = snapshot.items
+        self.assertEqual(item.owner_group, OWNER.hex())
+        # The owner field overlaps the level word without corrupting it.
+        self.assertEqual(item.level, 20)
+
+    def test_record_spanning_a_datagram_boundary_decodes(self):
+        record = item_record(
+            b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES, owner=OWNER
+        )
+        # Chunks far smaller than the record, at a width that is not a whole
+        # number of bytes: every field lands across at least one boundary.
+        snapshot = self.decode(record, chunk_bits=101)
+        [item] = snapshot.items
+        self.assertEqual(item.item_id, "cell3_style6_1_Orange")
+        self.assertEqual(item.level, 20)
+        self.assertEqual(item.owner_group, OWNER.hex())
+        self.assertEqual(dict(item.main_stats), {"HPMaxAdd": 840.0, "AtkAdd": 63.0})
+
+    def test_unframed_id_is_ignored(self):
+        # Same bytes, wrong length prefix: the frame check must reject it.
+        record = bytearray(
+            item_record(b"cell3_style6_1_Orange", entries=MODULE_ENTRIES)
+        )
+        record[0:4] = struct.pack("<I", 99)
+        self.assertEqual(self.decode(bytes(record)).items, ())
+
+
+class ValidationTests(unittest.TestCase):
+    def decode_one(self, payload: bytes):
+        return extract_gear(pack_datagrams(b"\0" * 8 + payload)).items
+
+    def test_main_stat_after_a_substat_is_rejected(self):
+        scrambled = (
+            (b"HPMaxAdd", 0.0),
+            (b"CritDamageBase", 0.06),
+            (b"AtkAdd", 0.0),
+            (b"CritBase", 0.03),
+            (b"AtkUp", 0.04),
+            (b"HPMaxUp", 0.04),
+        )
+        self.assertEqual(
+            self.decode_one(
+                item_record(b"cell3_style6_1_Orange", entries=scrambled)
+            ),
+            (),
+        )
+
+    def test_off_table_substat_value_is_rejected(self):
+        corrupt = MODULE_ENTRIES[:-1] + ((b"HPMaxUp", 0.0413),)
+        self.assertEqual(
+            self.decode_one(item_record(b"cell3_style6_1_Orange", entries=corrupt)),
+            (),
+        )
+
+    def test_short_entry_chain_is_rejected(self):
+        # Five entries is a cartridge count; a module needs six.
+        self.assertEqual(
+            self.decode_one(
+                item_record(b"cell3_style6_1_Orange", entries=MODULE_ENTRIES[:5])
+            ),
+            (),
+        )
+
+
+class CharacterTests(unittest.TestCase):
+    def test_level_and_breakthroughs_are_read(self):
+        payload = character_block(b"Zankou", level=70, breakthroughs=5)
+        [character] = extract_gear(pack_datagrams(payload)).characters
+        self.assertEqual(character.key, "Zankou")
+        self.assertEqual(character.level, 70)
+        self.assertEqual(character.breakthroughs, 5)
+
+    def test_progress_without_the_marker_is_not_reported(self):
+        payload = bytearray(character_block(b"Zankou", level=70, breakthroughs=5))
+        poke_u32_at_bit(payload, 128 * 8 - 466, 1)  # marker no longer 67806
+        [character] = extract_gear(pack_datagrams(bytes(payload))).characters
+        self.assertIsNone(character.level)
+        self.assertIsNone(character.breakthroughs)
+
+    def test_repeated_block_keeps_the_first_valid_reading(self):
+        block = character_block(b"Zankou", level=70, breakthroughs=5)
+        blank = bytearray(character_block(b"Zankou", level=70, breakthroughs=5))
+        poke_u32_at_bit(blank, 128 * 8 - 466, 1)
+        payload = block + b"\0" * 64 + bytes(blank)
+        [character] = extract_gear(pack_datagrams(payload)).characters
+        self.assertEqual(character.level, 70)
+        self.assertEqual(character.breakthroughs, 5)
+
+
+class LoadoutTests(unittest.TestCase):
+    def test_board_cells_follow_their_character_and_are_halved(self):
+        payload = (
+            character_block(b"Zankou", level=70, breakthroughs=5)
+            + b"\0" * 64
+            + loadout_cell(b"cell3_style6_1_Orange", 1, 2)
+            + b"\0" * 64
+            + loadout_cell(b"cell4_style2_1_Orange", 3, 4)
+        )
+        snapshot = extract_gear(pack_datagrams(payload))
+        [character] = snapshot.characters
+        # A transposed board renders correctly on a symmetric grid, so the
+        # assertion has to pin row and column separately.
+        self.assertEqual(character.board, ((1, 2), (3, 4)))
+        self.assertEqual(snapshot.characters_with_loadouts, 1)
+        # Quoted ids are references, not records.
+        self.assertEqual(snapshot.items, ())
+
+
+class GearExportContractTests(unittest.TestCase):
+    def build(self):
+        payload = (
+            character_block(b"Zankou", level=70, breakthroughs=5)
+            + b"\0" * 64
+            + loadout_cell(b"cell3_style6_1_Orange", 1, 2)
+            + b"\0" * 64
+            + item_record(
+                b"cell3_style6_1_Orange",
+                level=20,
+                entries=MODULE_ENTRIES,
+                owner=OWNER,
+            )
+            + b"\0" * 64
+            + item_record(
+                b"Incantation_orange",
+                level=20,
+                entries=CARTRIDGE_ENTRIES,
+                instance=b"\x99\xaa\xbb\xcc\xdd\xee\xff\x01",
+            )
+        )
+        snapshot = extract_gear(pack_datagrams(payload))
+        return snapshot, build_gear_export_json(
+            snapshot,
+            source="live_capture",
+            capture_source="libpcap",
+            user_uid=" 100200300 ",
+            server_id="1",
+        )
+
+    def test_envelope(self):
+        _snapshot, export = self.build()
+        self.assertEqual(export["format"], "nte-gear-export")
+        self.assertEqual(export["format_version"], 1)
+        self.assertEqual(export["source"], "live_capture")
+        self.assertEqual(export["capture_source"], "libpcap")
+        self.assertEqual(export["exporter"]["version"], __version__)
+        self.assertEqual(export["user_uid"], "100200300")
+        self.assertEqual(export["server_id"], "1")
+
+    def test_scan_counts_match_the_snapshot(self):
+        snapshot, export = self.build()
+        self.assertEqual(
+            export["scan"],
+            {
+                "cartridges": 1,
+                "modules": 1,
+                "characters": 1,
+                "characters_with_loadouts": 1,
+                "warnings": [],
+            },
+        )
+        self.assertEqual(len(export["items"]), len(snapshot.items))
+
+    def test_item_records_carry_raw_stat_ids(self):
+        _snapshot, export = self.build()
+        module = next(i for i in export["items"] if i["kind"] == "module")
+        self.assertEqual(module["shape"], "cell3_style6")
+        self.assertEqual(module["module_type"], "III")
+        self.assertEqual(module["owner_group"], OWNER.hex())
+        self.assertIn(
+            {"stat": "HPMaxAdd", "value": 840.0}, module["main_stats"]
+        )
+        self.assertTrue(all("stat" in s for s in module["substats"]))
+        cartridge = next(i for i in export["items"] if i["kind"] == "cartridge")
+        self.assertIn("set", cartridge)
+        self.assertIsNone(cartridge["owner_group"])
+
+    def test_character_board_and_unresolved_owner_group(self):
+        _snapshot, export = self.build()
+        [character] = export["characters"]
+        self.assertEqual(character["key"], "Zankou")
+        self.assertEqual(character["level"], 70)
+        self.assertEqual(character["breakthroughs"], 5)
+        self.assertEqual(character["board"], [[1, 2]])
+        # The capture cannot join an owner group to a name; emitting a guess
+        # would be worse than emitting the truth.
+        self.assertIsNone(character["owner_group"])
+
+
+# --- runner wiring ---------------------------------------------------------
+
+
+class _StubCapture:
+    name = "libpcap"
+    detail = "documentation range"
+    fallback_reason = ""
+
+    def __init__(self, packets):
+        self._packets = packets
+
+    def packets(self):
+        return iter(self._packets)
+
+    def stats(self):
+        return None
+
+    def close(self):
+        return None
+
+
+LOCAL_IP = "192.0.2.10"
+REMOTE_IP = "198.51.100.20"
+
+
+class RunnerWiringTests(unittest.TestCase):
+    """The gear export must survive being the only thing captured."""
+
+    def run_capture(self, *, user_uid=None):
+        payload = (
+            character_block(b"Zankou", level=70, breakthroughs=5)
+            + b"\0" * 64
+            + loadout_cell(b"cell3_style6_1_Orange", 1, 2)
+            + b"\0" * 64
+            + item_record(
+                b"cell3_style6_1_Orange",
+                level=20,
+                entries=MODULE_ENTRIES,
+                owner=OWNER,
+            )
+        )
+        packets = [
+            ParsedIpPacket(
+                src_ip=REMOTE_IP,
+                dst_ip=LOCAL_IP,
+                src_port=30166,
+                dst_port=49610,
+                payload=datagram,
+                protocol="udp",
+            )
+            for datagram in pack_datagrams(payload, chunk_bits=277)
+        ]
+        # An outbound datagram and a second, quieter inbound flow: the decoder
+        # must pick the busiest inbound flow and ignore the rest.
+        packets.append(
+            ParsedIpPacket(
+                src_ip=LOCAL_IP,
+                dst_ip=REMOTE_IP,
+                src_port=49610,
+                dst_port=30166,
+                payload=b"\x00" * 32,
+                protocol="udp",
+            )
+        )
+        packets.append(
+            ParsedIpPacket(
+                src_ip=REMOTE_IP,
+                dst_ip=LOCAL_IP,
+                src_port=30167,
+                dst_port=49611,
+                payload=b"\x00" * 16,
+                protocol="udp",
+            )
+        )
+
+        prompted = []
+
+        def fake_prompt_user_uid():
+            prompted.append(True)
+            return "100200300"
+
+        with tempfile.TemporaryDirectory() as workdir:
+            previous = os.getcwd()
+            os.chdir(workdir)
+            patched_backend = runner.open_capture_backend
+            patched_prompt = console.prompt_user_uid
+            runner.open_capture_backend = lambda ip, backend: _StubCapture(packets)
+            console.prompt_user_uid = fake_prompt_user_uid
+            try:
+                with contextlib.redirect_stdout(io.StringIO()) as out:
+                    result = runner.run_live_capture(
+                        interface_ip=LOCAL_IP, user_uid=user_uid
+                    )
+                gear_path = result["gear_path"]
+                self.assertIsNotNone(gear_path)
+                export = json.loads(gear_path.read_text(encoding="utf-8"))
+                printed = out.getvalue()
+            finally:
+                runner.open_capture_backend = patched_backend
+                console.prompt_user_uid = patched_prompt
+                os.chdir(previous)
+        return result, export, printed, prompted
+
+    def test_gear_only_capture_still_writes_and_reports_its_export(self):
+        result, export, printed, prompted = self.run_capture()
+        self.assertEqual(result["exports"], [])
+        self.assertEqual(export["format"], "nte-gear-export")
+        self.assertEqual(export["scan"]["modules"], 1)
+        self.assertEqual(export["scan"]["characters_with_loadouts"], 1)
+        self.assertEqual(export["source"], "live_capture")
+        self.assertEqual(export["capture_source"], "libpcap")
+        # The summary branch for "no pull history" has to mention it, or the
+        # file is written and never named.
+        self.assertIn("Gear export complete.", printed)
+        self.assertIn(result["gear_path"].name, printed)
+        self.assertIn("Gear captured", printed)
+        # Gear alone must be enough to opt into the UID prompt.
+        self.assertEqual(prompted, [True])
+        self.assertEqual(export["user_uid"], "100200300")
+        self.assertTrue(result["gear_path"].name.startswith("100200300_Gear_"))
+
+    def test_supplied_uid_skips_the_prompt(self):
+        _result, export, _printed, prompted = self.run_capture(user_uid="999")
+        self.assertEqual(prompted, [])
+        self.assertEqual(export["user_uid"], "999")
+
+
+    def test_gear_export_is_named_alongside_pull_history_exports(self):
+        """The second summary branch: gear must not vanish behind a banner."""
+        fixture = support.load_network_fixture()
+        local_ip = fixture["local_ip"]
+        payload = item_record(
+            b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES, owner=OWNER
+        )
+        packets = [
+            ParsedIpPacket(
+                src_ip=packet.src_ip,
+                dst_ip=packet.dst_ip,
+                src_port=packet.src_port,
+                dst_port=packet.dst_port,
+                payload=packet.payload,
+                protocol=packet.protocol,
+            )
+            for packet in support.fixture_packets("replay")
+        ]
+        packets += [
+            ParsedIpPacket(
+                src_ip=REMOTE_IP,
+                dst_ip=local_ip,
+                src_port=30166,
+                dst_port=49610,
+                payload=datagram,
+                protocol="udp",
+            )
+            for datagram in pack_datagrams(payload, chunk_bits=277)
+        ]
+
+        with tempfile.TemporaryDirectory() as workdir:
+            previous = os.getcwd()
+            os.chdir(workdir)
+            patched_backend = runner.open_capture_backend
+            patched_server = console.prompt_server_id
+            runner.open_capture_backend = lambda ip, backend: _StubCapture(packets)
+            console.prompt_server_id = lambda: ""
+            try:
+                with contextlib.redirect_stdout(io.StringIO()) as out:
+                    result = runner.run_live_capture(
+                        interface_ip=local_ip, user_uid="100200300"
+                    )
+                printed = out.getvalue()
+                gear_path = result["gear_path"]
+                self.assertIsNotNone(gear_path)
+                export = json.loads(gear_path.read_text(encoding="utf-8"))
+            finally:
+                runner.open_capture_backend = patched_backend
+                console.prompt_server_id = patched_server
+                os.chdir(previous)
+
+        self.assertTrue(result["exports"], "the fixture should still yield a banner")
+        self.assertEqual(export["scan"]["modules"], 1)
+        self.assertIn(gear_path.name, printed)
+
+    def test_udp_flow_accumulation_is_bounded(self):
+        self.assertEqual(gear_collector.MAX_UDP_FLOW_BYTES, 8 * 1024 * 1024)
+
+
+class _FakeClock:
+    """Stands in for the `time` module so idleness is exact, not slept for."""
+
+    def __init__(self, now=1000.0):
+        self.now = now
+
+    def monotonic(self):
+        return self.now
+
+    def time(self):
+        return self.now
+
+
+class _TickingCapture(_StubCapture):
+    """Yields a burst, then the read-timeout ticks pcap emits when idle.
+
+    `pending` is how many items are still unyielded, which is what proves a
+    report happened during the capture rather than after it.
+    """
+
+    def __init__(self, packets, clock, *, ticks=3, tick_seconds=5.0):
+        super().__init__(packets)
+        self.clock = clock
+        self.ticks = ticks
+        self.tick_seconds = tick_seconds
+        self.pending = 0
+
+    def packets(self):
+        queued = list(self._packets) + [None] * self.ticks
+        self.pending = len(queued)
+        for item in queued:
+            self.pending -= 1
+            if item is None:
+                self.clock.now += self.tick_seconds
+            yield item
+
+
+class LiveGearReportingTests(unittest.TestCase):
+    """Gear is reported while the capture runs, not only once it stops."""
+
+    def burst_packets(self):
+        payload = (
+            character_block(b"Zankou", level=70, breakthroughs=5)
+            + b"\0" * 64
+            + loadout_cell(b"cell3_style6_1_Orange", 1, 2)
+            + b"\0" * 64
+            + item_record(
+                b"cell3_style6_1_Orange",
+                level=20,
+                entries=MODULE_ENTRIES,
+                owner=OWNER,
+            )
+        )
+        return [
+            ParsedIpPacket(
+                src_ip=REMOTE_IP,
+                dst_ip=LOCAL_IP,
+                src_port=30166,
+                dst_port=49610,
+                payload=datagram,
+                protocol="udp",
+            )
+            for datagram in pack_datagrams(payload, chunk_bits=277)
+        ]
+
+    def run_capture(self, capture_factory, *, min_flow_bytes=1):
+        """Run one live capture against a stub, with the size gate relaxed.
+
+        Synthetic bursts are a few hundred bytes; the real gate exists to keep
+        chatter out, not to describe a minimum decodable size.
+        """
+        clock = _FakeClock()
+        capture = capture_factory(clock)
+        decodes = []
+        real_extract = gear_collector.extract_gear
+        reported_pending = []
+        real_report = gear_collector.print_gear_captured
+
+        def counting_extract(grams):
+            decodes.append(len(grams))
+            return real_extract(grams)
+
+        def recording_report(*args, **kwargs):
+            reported_pending.append(capture.pending)
+            return real_report(*args, **kwargs)
+
+        with tempfile.TemporaryDirectory() as workdir:
+            previous = os.getcwd()
+            os.chdir(workdir)
+            patched_backend = runner.open_capture_backend
+            patched_time = gear_collector.time
+            runner.open_capture_backend = lambda ip, backend: capture
+            gear_collector.time = clock
+            gear_collector.extract_gear = counting_extract
+            gear_collector.print_gear_captured = recording_report
+            previous_min = gear_collector.GEAR_LIVE_MIN_FLOW_BYTES
+            gear_collector.GEAR_LIVE_MIN_FLOW_BYTES = min_flow_bytes
+            try:
+                with contextlib.redirect_stdout(io.StringIO()) as out:
+                    result = runner.run_live_capture(
+                        interface_ip=LOCAL_IP, user_uid="100200300"
+                    )
+                printed = out.getvalue()
+            finally:
+                runner.open_capture_backend = patched_backend
+                gear_collector.time = patched_time
+                gear_collector.extract_gear = real_extract
+                gear_collector.print_gear_captured = real_report
+                gear_collector.GEAR_LIVE_MIN_FLOW_BYTES = previous_min
+                os.chdir(previous)
+        return result, printed, decodes, reported_pending
+
+    def test_gear_is_reported_before_the_capture_ends(self):
+        result, printed, _, reported_pending = self.run_capture(
+            lambda clock: _TickingCapture(self.burst_packets(), clock)
+        )
+
+        self.assertIn("Gear captured", printed)
+        self.assertEqual(len(reported_pending), 1, "reported exactly once")
+        self.assertGreater(
+            reported_pending[0],
+            0,
+            "the report must land while packets are still to come",
+        )
+        self.assertIsNotNone(result["gear_path"])
+
+    def test_a_flow_that_never_goes_idle_still_decodes_at_the_stop(self):
+        result, printed, _, reported_pending = self.run_capture(
+            lambda clock: _TickingCapture(self.burst_packets(), clock, ticks=0)
+        )
+
+        self.assertIn("Gear captured", printed)
+        self.assertEqual(
+            reported_pending, [0], "nothing pending: this is the post-loop fallback"
+        )
+        self.assertIsNotNone(result["gear_path"])
+
+    def test_an_idle_flow_is_not_decoded_twice_for_the_same_bytes(self):
+        # 40 ticks against a flow that never yields gear: without the
+        # already-attempted guard this would decode once per tick.
+        junk = [
+            ParsedIpPacket(
+                src_ip=REMOTE_IP,
+                dst_ip=LOCAL_IP,
+                src_port=30166,
+                dst_port=49610,
+                payload=b"\x00" * 64,
+                protocol="udp",
+            )
+        ]
+        _, _, decodes, _ = self.run_capture(
+            lambda clock: _TickingCapture(junk, clock, ticks=40)
+        )
+
+        # One live attempt for the single flow size, plus the post-loop scan.
+        self.assertEqual(len(decodes), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mapping_update.py

@@ -55,7 +55,7 @@ class MappingUpdateTests(unittest.TestCase):
         self.assertNotIn("fork_retired", result.mappings["arcs.json"])
         self.assertNotIn("OldTicket", result.mappings["items.json"])
         self.assertNotEqual(result.mappings["characters.json"]["1003"], current["characters.json"]["1003"])
-        self.assertEqual(result.report["changes"], {"additions": 5, "updates": 2, "deletions": 3})
+        self.assertEqual(result.report["changes"], {"additions": 6, "updates": 2, "deletions": 3})
         self.assertIs(result.report["safety"]["authoritative_snapshot"], True)
         self.assertIs(result.report["safety"]["deletions_allowed"], True)
         self.assertIs(result.report["safety"]["uid_inputs_touched"], False)
@@ -83,7 +83,11 @@ class MappingUpdateTests(unittest.TestCase):
         self.assertNotIn("DIceNormal", result.mappings["items.json"])
         self.assertNotIn("UnusedItem", result.mappings["items.json"])
         self.assertFalse(any(key.startswith("Characterawaken_") for key in result.mappings["items.json"]))
-        self.assertFalse(any(key.startswith("Fashion_vehicle_") for key in result.mappings["items.json"]))
+        self.assertEqual(
+            result.mappings["items.json"]["Fashion_vehicle_1010_V008"],
+            {"type": "cosmetic", "name": "Tiger Incoming! - Livery", "rank": "S"},
+        )
+        self.assertNotIn("Fashion_vehicle_1072_V006", result.mappings["items.json"])
 
     def test_future_mystery_box_pool_is_discovered_without_hard_coded_event_id(self):
         assets = load_assets(assets_root=SAMPLE_ASSETS)
@@ -113,7 +117,7 @@ class MappingUpdateTests(unittest.TestCase):
         tables["localization"]["ST_Item"].update(
             {"wow_ticket": "Wowzers Ticket", "wow_frame": "Wowzers Frame"}
         )
-        tables["localization"]["ST_VehicleData"] = {"vehicle_wow": "Wowmobile"}
+        tables["localization"]["ST_VehicleData"]["vehicle_wow"] = "Wowmobile"
 
         result = build_mapping_update(sample_current(), replace(assets, tables=tables))
 

tests/test_mystery_box_decoding.py

@@ -71,6 +71,15 @@ class MysteryBoxDecodingTests(unittest.TestCase):
         self.assertEqual(rows[2]["reward_name"], "Beetle Coin")
         self.assertEqual(rows[2]["reward_rank"], "B")
 
+    def test_reward_mapping_lookup_is_case_insensitive_and_canonical(self):
+        timestamp = datetime(2026, 7, 8, 19, 9, 33, tzinfo=timezone.utc)
+        rows = parse_mystery_box_response(
+            mystery_box_response([("Vehicle039", 1, timestamp)])
+        )
+
+        self.assertEqual(rows[0]["reward_id"], "vehicle039")
+        self.assertEqual(rows[0]["reward_name"], "Draco")
+
     def test_live_session_accepts_partial_final_page(self):
         session = LiveHistorySession("192.168.0.10")
         request = mystery_box_request(3)

tests/test_structured_protocol.py

@@ -157,6 +157,12 @@ class StructuredProtocolTests(unittest.TestCase):
         self.assertEqual(rows[0]["secondary_quantity"], 5)
         self.assertEqual(rows[0]["structured_pool_id"], "CardPool_Character")
 
+    def test_structured_reward_mapping_is_case_insensitive_and_canonical(self):
+        row = decode_response_records(monopoly_payload("DICENORMAL,1"))[0]
+
+        self.assertEqual(row["reward_id"], "DiceNormal")
+        self.assertEqual(row["reward_name"], "Fabricated Dice")
+
     def test_structured_monopoly_parser_falls_back_when_heuristic_returns_no_rows(self):
         payload = monopoly_payload("Dice_ticket_02,50", roll_points=0)