diff --git a/.github/workflows/pob-codes-feed-proof.yml b/.github/workflows/pob-codes-feed-proof.yml
new file mode 100644
index 00000000000..aa97ba3a168
--- /dev/null
+++ b/.github/workflows/pob-codes-feed-proof.yml
@@ -0,0 +1,32 @@
+name: PoB Codes feed integration proof
+on:
+  pull_request:
+    branches: [tests-branch, dev]
+    paths:
+      - spec/FetchTestBuilds.py
+      - spec/GenerateBuilds.lua
+      - spec/DiffOutput.lua
+      - spec/TestBuilds/**
+      - tests/test_pob_codes_feed.py
+      - .github/workflows/pob-codes-feed-proof.yml
+  workflow_dispatch:
+permissions:
+  contents: read
+jobs:
+  existing_generator:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+      - name: Prove adapter, existing generator, and existing comparator
+        env:
+          POB_FEED_DOCKER_TEST: '1'
+        run: python -m unittest discover -s tests -p test_pob_codes_feed.py -v
+      - name: Validate and decode the live monthly batch (manual only)
+        if: github.event_name == 'workflow_dispatch'
+        run: python spec/FetchTestBuilds.py --output "$RUNNER_TEMP/pob-codes-inputs"
diff --git a/spec/FetchTestBuilds.py b/spec/FetchTestBuilds.py
new file mode 100644
index 00000000000..cdbecaf6454
--- /dev/null
+++ b/spec/FetchTestBuilds.py
@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Decode one PoB Codes batch for the existing GenerateBuilds.lua input path."""
+import argparse
+import base64
+import hashlib
+import json
+from pathlib import Path
+import re
+import tempfile
+import urllib.request
+import zlib
+from xml.etree import ElementTree
+
+FEED = "https://api.pob.codes/test-builds"
+MAX_BATCH, MAX_CODE, MAX_XML = 16 * 1024 * 1024, 150 * 1024, 4 * 1024 * 1024
+
+
+def decode_batch(data):
+    if len(data) > MAX_BATCH:
+        raise ValueError("batch too large")
+    batch = json.loads(data)
+    if batch.get("schemaVersion") != 1 or batch.get("requestedCount") != 100:
+        raise ValueError("unsupported feed schema")
+    builds = batch.get("builds")
+    if not isinstance(builds, list) or not 1 <= len(builds) <= 100 or batch.get("count") != len(builds):
+        raise ValueError("invalid build count")
+    decoded = {}
+    for entry in builds:
+        code, key = entry["code"], entry["sha256"]
+        if not isinstance(code, str) or not 0 < len(code) <= MAX_CODE or not re.fullmatch(r"[A-Za-z0-9_+/=-]+", code):
+            raise ValueError("invalid build code")
+        if hashlib.sha256(code.encode("ascii")).hexdigest() != key or key + ".xml" in decoded:
+            raise ValueError("hash mismatch or duplicate build")
+        packed = base64.b64decode(code + "=" * (-len(code) % 4), altchars=b"-_", validate=True)
+        xml = None
+        for window in (zlib.MAX_WBITS, -zlib.MAX_WBITS):
+            try:
+                stream = zlib.decompressobj(window)
+                xml = stream.decompress(packed, MAX_XML + 1)
+            except zlib.error:
+                continue
+            if len(xml) > MAX_XML or stream.unconsumed_tail or not stream.eof or stream.unused_data:
+                raise ValueError("oversized, incomplete, or trailing compressed data")
+            break
+        if xml is None:
+            raise ValueError("invalid compressed build")
+        text = xml.decode("utf-8")
+        if "\0" in text or re.search(r"<!\s*(DOCTYPE|ENTITY)\b", text, re.I):
+            raise ValueError("unsupported XML declaration")
+        root = ElementTree.fromstring(text)
+        if root.tag != "PathOfBuilding" or root.find("Build") is None:
+            raise ValueError("not a PoE 1 build")
+        decoded[key + ".xml"] = xml
+    return decoded
+
+
+def write_inputs(data, output):
+    decoded = decode_batch(data)  # Validate the entire batch before writing anything.
+    output = Path(output).resolve()
+    if output.exists():
+        raise ValueError("output must be a new directory")
+    with tempfile.TemporaryDirectory(dir=output.parent) as temporary:
+        staged = Path(temporary) / "inputs"
+        staged.mkdir()
+        for name, xml in decoded.items():
+            (staged / name).write_bytes(xml)
+        staged.rename(output)
+    return len(decoded)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--batch", type=Path, help="Saved JSON batch; otherwise fetch the public API once")
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    if args.batch:
+        with args.batch.open("rb") as source:
+            data = source.read(MAX_BATCH + 1)
+    else:
+        request = urllib.request.Request(FEED, headers={"Accept": "application/json", "User-Agent": "PoB-feed-proof/1"})
+        with urllib.request.urlopen(request, timeout=30) as response:
+            if response.status != 200 or response.headers.get_content_type() != "application/json":
+                raise ValueError("unexpected API response")
+            data = response.read(MAX_BATCH + 1)
+    print("Validated and decoded %d feed builds" % write_inputs(data, args.output))
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as error:
+        raise SystemExit("Feed download/decoding failed (%s); use a new output directory and retry when ready" % type(error).__name__)
diff --git a/spec/GenerateBuilds.lua b/spec/GenerateBuilds.lua
index 9ed6c9de924..de46bf9f6f4 100644
--- a/spec/GenerateBuilds.lua
+++ b/spec/GenerateBuilds.lua
@@ -1,3 +1,13 @@
+-- Optional decoded feed directory; the existing link and fixture paths remain the default.
+local inputDir = os.getenv("BUILDINPUTDIR")
+if inputDir then
+    assert(not os.getenv("BUILDLINKS"), "Choose BUILDINPUTDIR or BUILDLINKS")
+    local originalLoadDB = build.LoadDB
+    function build:LoadDB(...)
+        assert(not originalLoadDB(self, ...), "Feed build import failed")
+    end
+end
+
 local function fetchBuilds(path)
     local lastDLtime = GetTime()
     local co = coroutine.create(function(path)
@@ -67,13 +77,17 @@ local function fetchBuilds(path)
     end
 end
 
-for testBuild in fetchBuilds("../spec/TestBuilds") do
+for testBuild in fetchBuilds(inputDir or "../spec/TestBuilds") do
     local filePath = (os.getenv("BUILDCACHEPREFIX") or "/tmp") .. "/" .. testBuild.filename
     local startTime = GetTime()
 
     -- Compute the build
     print("[+] Computing " .. filePath)
-    loadBuildFromXML(testBuild.xml)
+    loadBuildFromXML(testBuild.xml, inputDir and testBuild.filename or nil)
+    if inputDir then
+        assert(build.buildName == testBuild.filename and build.targetVersion, "Feed build initialization incomplete")
+        assert(build.calcsTab and build.calcsTab.mainOutput, "Feed build has no calculated output")
+    end
     local calcDuration = GetTime() - startTime
     print("[-] Computed " .. filePath .. " in " .. calcDuration .. "ms")
 
diff --git a/spec/POB_CODES_FEED_PROOF.md b/spec/POB_CODES_FEED_PROOF.md
new file mode 100644
index 00000000000..1a54cf7805d
--- /dev/null
+++ b/spec/POB_CODES_FEED_PROOF.md
@@ -0,0 +1,38 @@
+# PoB Codes feed integration proof
+
+This opt-in proof connects one batch from `https://api.pob.codes/test-builds`
+to the existing `GenerateBuilds.lua` generator. It leaves `BuildDiff.sh`,
+`DiffOutput.lua`, Docker Compose, and the existing scheduled workflows in place.
+It adds no FIFO, scheduled feed refresh, persisted corpus, or reporting system.
+
+The adapter validates every code/hash and bounded XML before creating a new
+input directory. It never replaces existing inputs. It makes one public request;
+there is no API secret. On failure, retry later with a new output directory.
+
+Run from the repository root with Python 3 and Docker:
+
+```sh
+POB_FEED_DOCKER_TEST=1 python3 -m unittest discover -s tests -p test_pob_codes_feed.py -v
+python3 spec/FetchTestBuilds.py --output /tmp/pob-codes-inputs
+mkdir /tmp/pob-codes-output
+chmod 777 /tmp/pob-codes-output
+docker compose run --rm --no-TTY -v /tmp/pob-codes-inputs:/inputs:ro -v /tmp/pob-codes-output:/outputs -e BUILDINPUTDIR=/inputs -e BUILDCACHEPREFIX=/outputs busted-tests timeout 300 busted --lua=luajit -r generate
+```
+
+`--batch saved.json` reads an offline API response instead. `BUILDINPUTDIR` selects
+decoded files and rejects simultaneous `BUILDLINKS`; unset it for the original
+behavior. In this opt-in mode, failed imports and incomplete initialization fail
+instead of saving a default build as a successful feed input.
+
+The test wraps an existing fixture in the API format, decodes it, calculates it
+twice with the existing generator, checks equality with `DiffOutput.lua`, then
+changes one saved stat to prove differences are detected. An unsupported target
+version must fail. PR tests use saved fixtures; manual workflow runs also check
+the live download/decoding boundary. No live calculation is implied by that check.
+
+The old tests-branch runtime does not support current exports. A live calculation
+may therefore fail even when download/decoding succeeds. Runtime compatibility
+must be resolved before a recurring modern-build comparison is enabled. This
+PR provides the starting proof; FIFO and broader runner changes can be reviewed
+separately. Manual workflow dispatch requires the workflow to be registered on
+the repository's default branch; the local commands work before that merge.
diff --git a/tests/test_pob_codes_feed.py b/tests/test_pob_codes_feed.py
new file mode 100644
index 00000000000..bae1ef26e1b
--- /dev/null
+++ b/tests/test_pob_codes_feed.py
@@ -0,0 +1,101 @@
+import base64
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tempfile
+import unittest
+import zlib
+from xml.etree import ElementTree
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "spec"))
+from FetchTestBuilds import decode_batch, write_inputs, MAX_XML
+
+
+def batch_for(xml):
+    code = base64.urlsafe_b64encode(zlib.compress(xml)).decode("ascii")
+    return {"schemaVersion": 1, "batchId": "fixture", "period": "2026-09",
+            "generatedAt": "2026-09-01T00:00:00.000Z", "patchVersion": "3.25",
+            "requestedCount": 100, "count": 1,
+            "builds": [{"code": code, "sha256": hashlib.sha256(code.encode()).hexdigest()}]}
+
+
+class FeedTests(unittest.TestCase):
+    def setUp(self):
+        self.xml = (ROOT / "spec/TestBuilds/OccVortex.xml").read_bytes()
+        self.batch = batch_for(self.xml)
+
+    def test_existing_fixture_survives_the_wire_format_exactly(self):
+        self.assertEqual(list(decode_batch(json.dumps(self.batch)).values()), [self.xml])
+
+    def test_invalid_batches_do_not_create_output(self):
+        bad_hash = batch_for(self.xml)
+        bad_hash["builds"][0]["sha256"] = "0" * 64
+        duplicate = batch_for(self.xml)
+        duplicate["builds"] *= 2
+        duplicate["count"] = 2
+        for batch in (bad_hash, duplicate, {**self.batch, "count": 0},
+                      {**self.batch, "schemaVersion": 2}, batch_for(b"not XML"),
+                      batch_for(b'<!DOCTYPE x><PathOfBuilding><Build/></PathOfBuilding>'),
+                      batch_for(b"x" * (MAX_XML + 1))):
+            with self.subTest(batch=batch.get("schemaVersion")), tempfile.TemporaryDirectory() as tmp:
+                output = Path(tmp) / "inputs"
+                with self.assertRaises((ValueError, ElementTree.ParseError)):
+                    write_inputs(json.dumps(batch), output)
+                self.assertFalse(output.exists())
+
+    def test_existing_inputs_are_never_overwritten(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            output = Path(tmp) / "inputs"
+            self.assertEqual(write_inputs(json.dumps(self.batch), output), 1)
+            before = {p.name: p.read_bytes() for p in output.iterdir()}
+            with self.assertRaises(ValueError):
+                write_inputs(json.dumps(self.batch), output)
+            self.assertEqual({p.name: p.read_bytes() for p in output.iterdir()}, before)
+
+    @unittest.skipUnless(os.environ.get("POB_FEED_DOCKER_TEST") == "1", "opt-in existing Docker generator proof")
+    def test_existing_generator_and_comparator(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp = Path(tmp)
+            write_inputs(json.dumps(self.batch), tmp / "inputs")
+            for name in ("first", "second", "bad-output"):
+                (tmp / name).mkdir()
+            command = ["docker", "run", "--rm", "--network", "none",
+                       "--mount", "type=bind,source=%s,target=/workdir,readonly" % ROOT,
+                       "--mount", "type=bind,source=%s,target=/proof" % tmp,
+                       "-w", "/workdir", "-e", "HOME=/tmp"]
+
+            def docker(arguments, environment=()):
+                return subprocess.run(command + list(environment) +
+                    ["ghcr.io/paliak/busted-tests:latest"] + arguments, timeout=330,
+                    stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+
+            def generate(inputs, output):
+                return docker(["timeout", "300", "busted", "--lua=luajit", "-r", "generate"],
+                    ["-e", "BUILDINPUTDIR=/proof/" + inputs, "-e", "BUILDCACHEPREFIX=/proof/" + output])
+
+            for output in ("first", "second"):
+                result = generate("inputs", output)
+                self.assertEqual(result.returncode, 0, result.stdout)
+                self.assertEqual(len(list((tmp / output).glob("*.build"))), 1)
+            name = next((tmp / "first").glob("*.build")).name
+            args = ["luajit", "spec/DiffOutput.lua", "/proof/first/" + name, "/proof/second/" + name]
+            self.assertEqual(docker(args).returncode, 0)
+            saved = (tmp / "second" / name).read_text()
+            changed, count = re.subn(r'(<PlayerStat stat="[^"]+" value=")[^"]+', r'\g<1>123456789', saved, count=1)
+            self.assertEqual(count, 1, "generator must save calculated stats")
+            (tmp / "second" / name).write_text(changed)
+            self.assertEqual(docker(args).returncode, 1, "existing comparator must detect changed stats")
+            bad = re.sub(br'targetVersion="[^"]+"', b'targetVersion="unsupported"', self.xml, count=1)
+            write_inputs(json.dumps(batch_for(bad)), tmp / "bad-input")
+            result = generate("bad-input", "bad-output")
+            self.assertNotEqual(result.returncode, 0, result.stdout)
+            self.assertFalse(list((tmp / "bad-output").glob("*.build")))
+
+
+if __name__ == "__main__":
+    unittest.main()