33 changed files with 127 additions and 1919 deletions
--- a/.github/workflows/update-mappings.yml
+++ b/.github/workflows/update-mappings.yml
@ -1,4 +1,4 @@
-name: Update mappings
+name: Update reward mappings
 on:
  workflow_dispatch:
@ -14,7 +14,7 @@ permissions:
  pull-requests: write
 concurrency:
-  group: update-mappings
+  group: update-reward-mappings
  cancel-in-progress: false
 jobs:
@ -23,7 +23,7 @@ jobs:
    runs-on: ubuntu-latest
    env:
-      AUTOMATION_BRANCH: dev/automated-mapping-update-${{ github.ref_name }}
+      AUTOMATION_BRANCH: dev/automated-mapping-update
      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.ref_name }}
+          ref: ${{ github.event.repository.default_branch }}
      - name: Get Python
        uses: actions/setup-python@v6
@ -42,7 +42,7 @@ jobs:
      - name: Install project
        run: python -m pip install -e .
-      - name: Rebuild mappings
+      - name: Rebuild reward 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.ref_name }}
+          BASE_BRANCH: ${{ github.event.repository.default_branch }}
          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 and achievement mappings, resolved English text through "
+              "The workflow rebuilt all reward mappings, resolved English names 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 mappings from NTE_Assets" \
+              --title "chore: update reward 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 mappings from NTE_Assets" \
+              --title "chore: update reward mappings from NTE_Assets" \
              --body-file build/mapping-update/pull-request-body.md
          fi
--- a/README.md
+++ b/README.md
@ -4,7 +4,7 @@
 # NTE History Exporter
-Prototype CLI exporter for **Neverness to Everness** pull history and achievements — decodes your own game traffic into sanitized JSON.
+Prototype CLI exporter for **Neverness to Everness** pull history — decodes your own game traffic into sanitized JSON ready for tracker import.
 ![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,17 +23,10 @@ Prototype CLI exporter for **Neverness to Everness** pull history and achievemen
 ## What It Does
-The exporter decodes Permanent Board, Limited Character Board, Arc Miracle Box,
+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.
 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]
-> Export JSON contains decoded history or achievement records and the shareable
+> 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.
 > 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
@ -97,11 +90,7 @@ From source:
 Or simply double-click **`run-exporter.cmd`**.
 > [!IMPORTANT]
-> Start the exporter **before pressing Start on the game's main menu**. This is
+> 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.
 > 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:
@ -109,7 +98,6 @@ 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.
@ -124,22 +112,19 @@ If a page response is missed, the exporter reports the missing page number while
 ### Achievement export
-Start the live exporter before pressing Start on the game's main menu. The
+Start the live exporter before launching the game. The completed achievement
-tracked achievement state is loaded during login, captured automatically, and written to
+list 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 in-progress achievements are included with their current
+Both completed and incomplete 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.
-PlayStation trophy entries are retained and identified separately
+Completed PlayStation trophy entries are retained and identified separately
 even though they are not shown in the in-game achievement list. The console
-reports completed and in-progress totals for in-game and PlayStation records as
+reports the visible and PlayStation counts as soon as the list has been decoded.
 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
@ -206,22 +191,22 @@ The exporter automatically includes the shareable NTE user UID when it appears i
 ## Mapping maintenance
-Reward and achievement metadata can be rebuilt directly from the latest
+Reward metadata can be rebuilt directly from the latest NTE_Assets tables and
-NTE_Assets tables and English translation files. Mapping snapshots may change
+English translation files. The reward snapshot may change IDs while UID inputs
-IDs while UID inputs remain untouched:
+remain untouched:
-Run **Update mappings** from the GitHub Actions tab to generate and test
+Run **Update reward 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 mapping candidates directly from NTE_Assets under
+This rebuilds reviewable reward-map 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
-[Mapping updates](docs/mapping-updates.md) for the source rules and
+[Reward mapping updates](docs/mapping-updates.md) for the source rules and
 review workflow.
 ## Privacy
--- a/docs/export-format.md
+++ b/docs/export-format.md
@ -15,7 +15,7 @@ decoder-only offsets.
  "capture_source": "npcap",
  "exporter": {
    "name": "nte-history-exporter",
-    "version": "0.3.0"
+    "version": "0.2.1"
  },
  "banner": {
    "id": "Lottery_Permanent",
@ -84,12 +84,6 @@ 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
@ -182,79 +176,6 @@ 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
--- a/docs/limitations.md
+++ b/docs/limitations.md
@ -21,13 +21,5 @@ 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.
--- a/docs/mapping-updates.md
+++ b/docs/mapping-updates.md
@ -1,4 +1,4 @@
-# Mapping updates
+# Reward mapping updates
 `tools/update_mappings.py` rebuilds `arcs.json`, `characters.json`,
 `items.json`, and `achievements.json` directly from the current
@ -19,17 +19,15 @@ game whose asset tables evolve over time.
 ### GitHub Actions
-Run **Update mappings** from the repository's Actions tab. **Use workflow
+Run **Update reward mappings** from the repository's Actions tab. The optional
-from** selects the exporter branch whose updater is run and the target branch
+`source_ref` input accepts an NTE_Assets branch, tag, or commit and defaults to
-for the resulting pull request. The `source_ref` input separately selects an
+`main`.
 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-<selected-branch>`. Running it again for the same
+`dev/automated-mapping-update`. Running it again updates the same open pull
-selected branch updates that branch's open pull request. If the authoritative
+request. If the authoritative snapshot has not changed, it reports zero changes
-snapshot has not changed, it reports zero changes and does not create a commit
+and does not create a commit or pull request.
 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
@ -70,7 +68,7 @@ python -m pytest -q
 ```
 For CI, `--check` exits with status 1 whenever the generated snapshot differs
-from the committed mappings. It still writes the staged artifacts.
+from the committed reward mappings. It still writes the staged artifacts.
 ## Source rules
@ -82,9 +80,6 @@ from the committed 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,
@ -94,8 +89,8 @@ from the committed 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 illustration entries are not independent pull rewards
+- Character-awakening and character-vehicle illustration entries are not
-  and are excluded.
+  independent pull rewards and are excluded.
 - Orange, purple, and blue item qualities map to `S`, `A`, and `B`.
 ## UID compatibility boundary
@ -105,7 +100,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 and achievement IDs or display metadata may change or disappear when
+Reward IDs and display metadata may change or disappear when the authoritative
-the authoritative NTE_Assets snapshot changes. Existing exports remain
+NTE_Assets snapshot changes. Existing exported history files remain unchanged;
-unchanged; new exports use the current asset snapshot. UID stability is
+new exports describe rewards using the current asset snapshot. UID stability is
-independent of display mappings.
+independent of reward display mappings.
--- a/docs/packet-format.md
+++ b/docs/packet-format.md
@ -1,21 +1,6 @@
 # Packet Format Notes
-This prototype supports separate Monopoly, Arc/Gashapon, Mystery Box history,
+This prototype supports separate Monopoly, Arc/Gashapon, and Mystery Box history decoders.
 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
@ -101,8 +86,6 @@ 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.
--- a/mappings/arcs.json
+++ b/mappings/arcs.json
@ -10,10 +10,8 @@
  "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"},
--- a/mappings/characters.json
+++ b/mappings/characters.json
@ -9,7 +9,6 @@
  "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"},
@ -19,9 +18,7 @@
  "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"}
 }
--- a/mappings/items.json
+++ b/mappings/items.json
@ -1,51 +1,40 @@
 {
  "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_1072_01": {"type": "cosmetic", "name": "Surfing Among Stars", "rank": "S"},
+  "Fashion_character_1075_01": {"type": "cosmetic", "name": "Summer Dream", "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_1072": {"type": "cosmetic", "name": "Sheepcopter", "rank": "A"},
+  "Fashion_glide_1075": {"type": "cosmetic", "name": "Skyrider", "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": "Three's Company - Livery", "rank": "S"},
+  "Fashion_vehicle_1075_V006": {"type": "cosmetic", "name": "Hidden Dragon - 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"},
  "Fons": {"type": "item", "name": "Fons", "rank": "A"},
  "Frame_MangHe_bsj": {"type": "cosmetic", "name": "Heartbeat Moment", "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"},
+  "Fashion_1051_rpg_level3_Dye1": {"type": "cosmetic", "name": "Glory Days - Braveheart", "rank": "S"},
-  "weeklycloneboss_selectbox_02": {"type": "item", "name": "Anomaly Pilgrimage Material Selection Box", "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"}
 }
--- a/pyproject.toml
+++ b/pyproject.toml
@ -1,7 +1,7 @@
 [project]
 name = "nte-history-exporter"
-version = "0.4.0"
+version = "0.2.1"
-description = "Cross-platform Neverness to Everness pull-history and achievement exporter."
+description = "Cross-platform Neverness to Everness pull-history exporter."
 requires-python = ">=3.10"
 dependencies = []
--- a/src/nte_history_exporter/init.py
+++ b/src/nte_history_exporter/init.py
@ -1 +1 @@
-__version__ = "0.4.0"
+__version__ = "0.2.1"
--- a/src/nte_history_exporter/console.py
+++ b/src/nte_history_exporter/console.py
@ -94,13 +94,9 @@ 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 tracked achievements are captured automatically when")
+    print("    Your completed 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())
@ -110,26 +106,12 @@ 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(
+def print_achievements_captured(total: int, playstation: int) -> None:
-    in_game_completed: int,
+    visible = total - playstation
    in_game_in_progress: int,
    playstation_completed: int,
    playstation_in_progress: int,
 ) -> None:
    print(style("  + ", GREEN, BOLD) + "Achievements captured")
    print(
-        style(
+        style("  + ", GREEN, BOLD)
-            f"      In-game      {in_game_completed} completed, "
+        + "Achievements"
-            f"{in_game_in_progress} in progress",
+        + style(f"  captured {visible} visible + {playstation} PlayStation", DIM)
            DIM,
        )
    )
    print(
        style(
            f"      PlayStation  {playstation_completed} completed, "
            f"{playstation_in_progress} in progress",
            DIM,
        )
    )
@ -202,7 +184,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 reward or achievement mappings may be incomplete in this version.", YELLOW))
+    print(style("  Some banner/reward 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))
--- a/src/nte_history_exporter/constants.py
+++ b/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.4.0"
+EXPORTER_VERSION = "0.2.1"
 HISTORY_REQUEST_BANNER = 4220
 HISTORY_REQUEST_LENGTH = 45
--- a/src/nte_history_exporter/decoder/achievement.py
+++ b/src/nte_history_exporter/decoder/achievement.py
@ -27,10 +27,7 @@ class AchievementRecord:
    def status(self) -> str:
        if self.completed:
            return "completed"
-        # Presence in the server response means the achievement has tracked
+        return "in_progress" if self.progress else "not_started"
        # 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:
--- a/src/nte_history_exporter/decoder/arc.py
+++ b/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_BY_CASEFOLD
+from nte_history_exporter.mappings import ARC_META
 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()
-        canonical_id, meta = _resolve_arc_metadata(arc_id)
+        meta = ARC_META.get(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": canonical_id,
+                "reward_id": arc_id,
                "reward_name": meta.get("name", "UNKNOWN"),
                "reward_rank": meta.get("rank", ""),
                "type_key_hex": type_raw.hex(),
@ -122,15 +122,18 @@ def _parse_legacy_arc_response(response: bytes) -> list[dict[str, Any]]:
    return records
-def _resolve_arc_metadata(arc_id: str) -> tuple[str, dict[str, Any]]:
+def _arc_metadata(arc_id: str) -> dict[str, Any]:
-    canonical_id, meta = ARC_META_BY_CASEFOLD.get(arc_id.casefold(), (arc_id, {}))
+    direct = ARC_META.get(arc_id)
-    return canonical_id, meta
+    if direct is not None:
        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:
-        canonical_id, meta = _resolve_arc_metadata(structured.item_id)
+        meta = _arc_metadata(structured.item_id)
        records.append(
            {
                "record_start": structured.record_start,
@ -138,7 +141,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": canonical_id,
+                "reward_id": structured.item_id,
                "reward_name": meta.get("name", "UNKNOWN"),
                "reward_rank": meta.get("rank", ""),
                "type_key_hex": "",
--- a/src/nte_history_exporter/decoder/gear.py
+++ b/src/nte_history_exporter/decoder/gear.py
@ -1,499 +0,0 @@
 """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()),
    )
--- a/src/nte_history_exporter/decoder/mystery_box.py
+++ b/src/nte_history_exporter/decoder/mystery_box.py
@ -24,11 +24,16 @@ 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_CASEFOLD
+from nte_history_exporter.mappings import REWARDS_BY_ID
 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
@ -103,15 +108,16 @@ 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_CASEFOLD.get(reward_id.casefold(), {})
+            reward = REWARDS_BY_ID.get(reward_id) or REWARDS_BY_CASEFOLDED_ID.get(
-            canonical_reward_id = reward.get("id", reward_id)
+                reward_id.casefold(), {}
            )
            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": canonical_reward_id,
+                    "reward_id": reward_id,
                    "reward_name": reward.get("name", ""),
                    "reward_rank": reward.get("rank"),
                    "quantity": quantity,
--- a/src/nte_history_exporter/decoder/protocol.py
+++ b/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_CASEFOLD
+from nte_history_exporter.mappings import REWARDS_BY_ID
 from nte_history_exporter.decoder.structured_protocol import (
    StructuredRecord,
    parse_structured_records,
@ -61,12 +61,11 @@ def decode_reward_key(raw: bytes) -> str:
 def infer_reward_type(reward_id: str) -> str:
    if not reward_id:
        return ""
-    folded = reward_id.casefold()
+    if reward_id.startswith("fork_"):
    if folded.startswith("fork_"):
        return "arc"
    if reward_id.isdigit():
        return "character"
-    if folded.startswith("fashion_"):
+    if reward_id.startswith("Fashion_"):
        return "cosmetic"
    return "item"
@ -166,7 +165,7 @@ def classify_result_type(
 ) -> tuple[str, int | None]:
    if dice is None or dice_offset is None:
        return "unknown", None
-    if reward_id.casefold() == "dice_ticket_01" and WARP_PIECE_CHASE_PATTERN in chunk_without_marker:
+    if reward_id == "Dice_ticket_01" and WARP_PIECE_CHASE_PATTERN in chunk_without_marker:
        return "chase_reward", -4
    if dice == 0:
        return "points_gift", 0
@ -183,14 +182,13 @@ def classify_result_type(
 def guess_quantity(chunk_hex: str, reward_id: str, result_type: str | None = None) -> int | None:
-    folded = reward_id.casefold()
+    if reward_id == "Dice_ticket_01":
    if folded == "dice_ticket_01":
        if result_type == "chase_reward":
            return 30
        return 4
-    if folded == "dicenormal":
+    if reward_id == "DiceNormal":
        return 1
-    if folded == "dice_ticket_02":
+    if reward_id == "Dice_ticket_02":
        if "c8b0d4c0" in chunk_hex:
            return 50
        if "c8b0ccc0" in chunk_hex:
@ -268,8 +266,7 @@ def _decode_aligned_response_records(response_content: bytes) -> list[dict[str,
        elif result_type == "chase_reward":
            dice = -4
            dice_raw = -4
-        reward = _reward_metadata(reward_id)
+        reward = REWARDS_BY_ID.get(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()
@ -291,7 +288,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": canonical_reward_id,
+                "reward_id": reward_id,
                "reward_name": reward.get("name", ""),
                "reward_rank": reward.get("rank"),
                "quantity": guess_quantity(chunk_hex, reward_id, result_type),
@ -338,8 +335,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")
@ -364,7 +361,11 @@ def _enrich_heuristic_rows(
 def _reward_metadata(reward_id: str) -> dict[str, Any]:
-    return REWARDS_BY_CASEFOLD.get(reward_id.casefold(), {})
+    direct = REWARDS_BY_ID.get(reward_id)
    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]]:
@ -372,7 +373,6 @@ 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": canonical_reward_id,
+                "reward_id": structured.item_id,
                "reward_name": reward.get("name", ""),
                "reward_rank": reward.get("rank"),
                "quantity": structured.count,
--- a/src/nte_history_exporter/export/gear_export.py
+++ b/src/nte_history_exporter/export/gear_export.py
@ -1,99 +0,0 @@
 """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
--- a/src/nte_history_exporter/export/json_export.py
+++ b/src/nte_history_exporter/export/json_export.py
@ -94,11 +94,13 @@ 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:
@ -108,8 +110,10 @@ def build_achievement_export_json(
        summary["total_achievements"] += 1
        if achievement.completed:
            summary["completed_achievements"] += 1
-        else:
+        elif achievement.progress:
            summary["in_progress_achievements"] += 1
        else:
            summary["not_started_achievements"] += 1
        category, separator, _number = achievement_id.partition("_")
        category_key = category.casefold() if separator else "uncategorized"
        record = {
--- a/src/nte_history_exporter/live_capture/gear_collector.py
+++ b/src/nte_history_exporter/live_capture/gear_collector.py
@ -1,191 +0,0 @@
 """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,
        )
--- a/src/nte_history_exporter/live_capture/runner.py
+++ b/src/nte_history_exporter/live_capture/runner.py
@ -21,7 +21,6 @@ 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
@ -89,17 +88,14 @@ 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"
@ -115,27 +111,12 @@ def run_live_capture(
                            reassemble_tcp_segments(tcp_segments[flow])
                        )
                        if achievement_records:
-                            in_game = [
+                            completed = [record for record in achievement_records if record.completed]
                                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(
-                                sum(record.completed for record in in_game),
+                                len(completed),
                                sum(record.status == "in_progress" for record in in_game),
                                sum(record.completed for record in playstation),
                                sum(
-                                    record.status == "in_progress"
+                                    record.achievement_id.lower().startswith("playstation_")
-                                    for record in playstation
+                                    for record in completed
                                ),
                            )
                pair_count_before = len(session.pairs)
@ -180,8 +161,6 @@ def run_live_capture(
        stats = capture.stats()
        capture.close()
    gear.finalize()
    exports = []
    achievement_path = None
    diagnostics_path = None
@ -189,9 +168,7 @@ 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 (
+    if (session.kinds_seen() or achievement_records) and not resolved_user_uid:
        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)
@ -206,11 +183,6 @@ 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()
@ -264,23 +236,16 @@ 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:
@ -302,8 +267,6 @@ 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}")
@ -319,7 +282,6 @@ def run_live_capture(
        "exports": exports,
        "diagnostics_path": diagnostics_path,
        "achievement_path": achievement_path,
        "gear_path": gear_path,
    }
--- a/src/nte_history_exporter/mapping_update.py
+++ b/src/nte_history_exporter/mapping_update.py
@ -324,16 +324,9 @@ 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:
@ -350,31 +343,6 @@ 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]]],
--- a/src/nte_history_exporter/mappings.py
+++ b/src/nte_history_exporter/mappings.py
@ -14,9 +14,6 @@ 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")
@ -39,6 +36,3 @@ 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()
 }
--- a/tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json
+++ b/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":{},"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"}}}}]
+[{"Rows":{"1003":{},"1099":{},"fork_alpha":{},"fork_new":{},"DIceNormal":{},"Dice_ticket_03":{},"Fashion_glide_2000":{},"Characterawaken_1003":{}}}]
--- a/tests/fixtures/nte_assets/Localization/en/game.json
+++ b/tests/fixtures/nte_assets/Localization/en/game.json
@ -18,9 +18,5 @@
  },
  "ST_Appearance": {
    "glider_2000": "Inventory Glider"
  },
  "ST_VehicleData": {
    "vehicle_1010_livery": "Tiger Incoming! - Livery",
    "vehicle_placeholder_livery": "Sandy Mirage - Livery"
  }
 }
--- a/tests/test_achievement.py
+++ b/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, "in_progress")
+        self.assertEqual(records[1].status, "not_started")
    def test_reassembles_tcp_data(self):
        self.assertEqual(
--- a/tests/test_arc_decoding.py
+++ b/tests/test_arc_decoding.py
@ -1,14 +1,8 @@
 from tests.support import *  # noqa: F401,F403
-from nte_history_exporter.decoder.arc import annotate_arc_groups, _resolve_arc_metadata
+from nte_history_exporter.decoder.arc import annotate_arc_groups
 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 = [
            {
--- a/tests/test_export_contract.py
+++ b/tests/test_export_contract.py
@ -1,16 +1,11 @@
 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, [])
@ -158,11 +153,13 @@ 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": 1,
+                    "in_progress_achievements": 0,
                    "not_started_achievements": 1,
                },
            },
        )
--- a/tests/test_gear_decoding.py
+++ b/tests/test_gear_decoding.py
@ -1,747 +0,0 @@
 """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()
--- a/tests/test_mapping_update.py
+++ b/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": 6, "updates": 2, "deletions": 3})
+        self.assertEqual(result.report["changes"], {"additions": 5, "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,11 +83,7 @@ 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.assertEqual(
+        self.assertFalse(any(key.startswith("Fashion_vehicle_") for key in result.mappings["items.json"]))
            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)
@ -117,7 +113,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))
--- a/tests/test_mystery_box_decoding.py
+++ b/tests/test_mystery_box_decoding.py
@ -71,15 +71,6 @@ 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)
--- a/tests/test_structured_protocol.py
+++ b/tests/test_structured_protocol.py
@ -157,12 +157,6 @@ 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)
`@ -1 +1 @@`
	[{"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"}}}}]	`[{"Rows":{"1003":{},"1099":{},"fork_alpha":{},"fork_new":{},"DIceNormal":{},"Dice_ticket_03":{},"Fashion_glide_2000":{},"Characterawaken_1003":{}}}]`