diff --git a/CHANGELOG.md b/CHANGELOG.md
index 650ff91..f4bf0cd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,8 +4,14 @@
 
 ### Added
 
+- Added `--zip-parallelism` for controlling open/read ZIP concurrency independently from XML entry worker count.
+- Added opt-in `--largest-first` ZIP scheduling. Word2vec/NOW exports now keep argument order by default, avoiding an automatic size-based corpus-domain bias.
 - Krill tars can now be used as *input* and updated in place of a full re-export ("merge mode"): listing an existing `.krill.tar` among the inputs with `-t krill` streams it to a new tar (`<name>.updated.krill.tar`, or `-o`), updated from the remaining inputs. Two use cases: (1) adding or fixing metadata (stand-off `<standOff>` files, `<xenoData>`, header ZIPs) when the original KorAP-XML ZIPs are no longer available, and (2) replacing or adding single annotation foundries (e.g. an improved wiki-taxonomy topic-domain classification) without touching the rest. Texts unaffected by the new inputs are copied through byte-identically without being decompressed; affected texts are patched surgically at the JSON level and recompressed in parallel, and end up identical to a full re-export with the same inputs. New metadata fields replace same-key fields (including legacy/corrected name pairs such as `textClass`/`dmozDomain`) and are appended otherwise; `creationDate`/`pubDate` are only replaced when a new text-level header is supplied, so corpus-level headers cannot clobber per-text dates. Texts present in the new inputs but missing from the tar are ignored with a warning; base tokenization/structure entries are ignored as well. Tars produced by korapxml2krill (Perl) are supported. The input tar is never modified, and the regular Krill export path is unchanged.
 
+### Fixed
+
+- Stabilized large `word2vec`/`NOW` exports: ZIP readers now share one bounded XML-entry backlog instead of multiplying the backlog by every open ZIP, plain-output document claims are released when each ZIP closes instead of growing for the whole corpus, annotation ZIPs are not re-reading the same base ZIP for surface output, and unused late annotation layers no longer repopulate cleaned document state.
+
 ## [v4.1.0] - 2026-06-19
 
 ### Added
diff --git a/Readme.md b/Readme.md
index 74ac4c1..da07587 100644
--- a/Readme.md
+++ b/Readme.md
@@ -47,6 +47,8 @@
 
 - `-t FORMAT`, `--to FORMAT`: Output format (`zip`, `conllu`, `w2v`, `now`, `krill`)
 - `-j N`, `--jobs N`, `--threads N`: Number of threads/jobs to use
+- `--zip-parallelism N`: Maximum ZIP files read concurrently (default: up to 8); XML entry work still uses `--threads`
+- `--largest-first`: Schedule larger input ZIPs first (opt-in; the default is argument order)
 - `-T TAGGER[:MODEL]`, `--tag-with TAGGER[:MODEL]`: POS tagger and optional model
 - `-P PARSER[:MODEL]`, `--parse-with PARSER[:MODEL]`: Parser and optional model  
 - `-f`, `--force`: Overwrite existing output files
@@ -134,20 +136,21 @@
 
 - `--lemma-only`: For `-t w2v` and `-t now`, skip loading `data.xml` and output only lemmas from `morpho.xml`. This reduces memory and speeds up throughput.
 - `--sequential`: Process entries inside each zip sequentially (zips can still run in parallel). Recommended for `w2v`/`now` to keep locality and lower memory.
+- Parallel `w2v`/`now` runs schedule ZIPs in argument order and use a single bounded entry backlog across all open ZIPs. Completion and output order can still vary when multiple ZIPs run concurrently. Use `--largest-first` to opt into size-descending scheduling when minimizing the low-parallelism tail matters more than corpus order. For reproducible server runs, set both `--threads` and `--zip-parallelism` explicitly.
 - `--exclude-zip-glob GLOB` (repeatable): Skip zip basenames that match the glob (e.g., `--exclude-zip-glob 'w?d24.tree_tagger.zip'`).
 
 Example for large NOW export with progress and exclusions:
 
 ```
 KORAPXMLTOOL_XMX=64g KORAPXMLTOOL_MODELS_PATH=/data/models KORAPXMLTOOL_JAVA_OPTS="-XX:+UseG1GC -Djdk.util.zip.disableMemoryMapping=true -Djdk.util.zip.reuseInflater=true" \
-     ./build/bin/korapxmltool -l info -j 100 \
+     ./build/bin/korapxmltool -l info -j 100 --zip-parallelism 8 \
      --lemma-only --sequential -t now \
      --exclude-zip-glob 'w?d24.tree_tagger.zip' \
      /vol/corpora/DeReKo/current/KorAP/zip/*24.tree_tagger.zip | pv > dach2024.lemma.txt
 ```
 
 At INFO level the tool logs:
-- The zip processing order with file sizes (largest-first in `--lemma-only`).
+- The ZIP processing order with file sizes (argument order by default, or size-descending with `--largest-first`).
 - For each zip: start message including its size and a completion line with cumulative progress, ETA and average MB/s.
 
 ### Conversion to Krill (KoralQuery) JSON format
diff --git a/app/src/main/kotlin/de/ids_mannheim/korapxmltools/KorapXmlTool.kt b/app/src/main/kotlin/de/ids_mannheim/korapxmltools/KorapXmlTool.kt
index 520ad94..4002198 100644
--- a/app/src/main/kotlin/de/ids_mannheim/korapxmltools/KorapXmlTool.kt
+++ b/app/src/main/kotlin/de/ids_mannheim/korapxmltools/KorapXmlTool.kt
@@ -265,6 +265,31 @@
         System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", threads.toString())
     }
 
+    var zipParallelism: Int = 0  // 0 = auto-detect after --threads is resolved
+
+    @Option(
+        names = ["--zip-parallelism"],
+        paramLabel = "ZIPS",
+        description = [
+            "Maximum number of ZIP files read concurrently. Default: up to 8; entry processing still uses --threads."
+        ]
+    )
+    fun configureZipParallelism(parallelism: Int) {
+        if (parallelism < 1) {
+            throw ParameterException(spec.commandLine(), String.format(Locale.ROOT,
+                "Invalid value `%d' for option '--zip-parallelism': must be at least 1", parallelism))
+        }
+        zipParallelism = parallelism
+    }
+
+    @Option(
+        names = ["--largest-first"],
+        description = [
+            "Schedule larger input ZIPs first. By default ZIPs are scheduled in argument order."
+        ]
+    )
+    var largestFirst: Boolean = false
+
 
     @Option(
         names = ["--sequential"],
@@ -693,6 +718,8 @@
         if (outputFile != null)       sb.appendLine("  Output file:    $outputFile")
         if (outputDir != ".")         sb.appendLine("  Output dir:     $outputDir")
         sb.appendLine("  Threads:        $maxThreads")
+        sb.appendLine("  ZIP parallelism:${if (zipParallelism > 0) "  $zipParallelism" else "  auto"}")
+        sb.appendLine("  ZIP order:      ${if (largestFirst) "largest first" else "argument order"}")
         sb.appendLine("  Log level:      $logLevel")
 
         // Annotation options
@@ -1070,6 +1097,21 @@
             else -> outputFormat.name
         }
 
+    internal fun effectiveZipParallelism(zipCount: Int): Int {
+        if (zipCount <= 0) return 1
+        val requested = if (zipParallelism > 0) zipParallelism else minOf(maxThreads, 8)
+        return requested.coerceIn(1, zipCount)
+    }
+
+    internal fun orderZipInputsForProcessing(zips: Array<String>): Array<String> =
+        if (largestFirst) {
+            zips.sortedByDescending { zipSizes[it] ?: 0L }.toTypedArray()
+        } else {
+            zips.copyOf()
+        }
+
+    internal fun streamingOutputClaimCount(): Int = streamingOutputClaims.values.sumOf { it.size }
+
     internal fun canUseStaxTextParsing(): Boolean =
         outputFormat == OutputFormat.CONLLU ||
             outputFormat == OutputFormat.WORD2VEC ||
@@ -1171,6 +1213,12 @@
 
     // Single priority-based executor for all entry processing
     private var entryExecutor: java.util.concurrent.ExecutorService? = null
+    private var streamEntryPermits: java.util.concurrent.Semaphore? = null
+    private val streamEntriesInFlight = AtomicInteger(0)
+    private val streamEntriesPeak = AtomicInteger(0)
+    private val activeZipWorkers = AtomicInteger(0)
+    private val peakActiveZipWorkers = AtomicInteger(0)
+    private val textStreamWorkerNumber = AtomicInteger(0)
 
     private val MONTH_ORDER = mapOf(
         "JAN" to 1, "FEB" to 2, "MAR" to 3, "MRZ" to 3, "APR" to 4,
@@ -1389,6 +1437,11 @@
     // Track which texts have been output to avoid counting duplicates (thread-safe)
     val outputTexts: MutableSet<String> = ConcurrentHashMap.newKeySet()
 
+    // Plain streaming output only needs duplicate protection while one ZIP is open.
+    // Keeping every emitted ID in outputTexts made memory grow with total corpus size.
+    private val streamingOutputClaims: ConcurrentHashMap<String, MutableSet<String>> = ConcurrentHashMap()
+    private val streamingZipUsers: ConcurrentHashMap<String, AtomicInteger> = ConcurrentHashMap()
+
     // Scheduled executor for periodic scanning
     var incrementalOutputScheduler: java.util.concurrent.ScheduledExecutorService? = null
 
@@ -1413,8 +1466,31 @@
         return if (File(baseZip).exists()) baseZip else null
     }
 
+    internal fun planPlainSurfaceTextInputs(zips: Array<String>): Array<String> {
+        if ((outputFormat != OutputFormat.WORD2VEC && outputFormat != OutputFormat.NOW) || useLemma) {
+            return zips
+        }
+        val planned = LinkedHashMap<String, String>()
+        zips.forEach { input ->
+            val usefulInput = input.correspondingBaseZip() ?: input
+            val identity = try {
+                File(usefulInput).canonicalPath
+            } catch (_: Exception) {
+                File(usefulInput).absolutePath
+            }
+            planned.putIfAbsent(identity, usefulInput)
+        }
+        return planned.values.toTypedArray()
+    }
+
     fun korapxml2conllu(args: Array<String>) {
         outputTexts.clear()
+        streamingOutputClaims.clear()
+        streamingZipUsers.clear()
+        streamEntriesInFlight.set(0)
+        streamEntriesPeak.set(0)
+        activeZipWorkers.set(0)
+        peakActiveZipWorkers.set(0)
         firstTextOutputLogged.set(false)
         // Reset Krill state for fresh run (important for tests)
         if (outputFormat == OutputFormat.KRILL) {
@@ -1468,12 +1544,16 @@
             LOGGER.info("Initialized work-stealing scheduler with $maxThreads worker threads for Krill output")
         } else if (canUseArchiveOrderTextStreaming()) {
             entryExecutor = java.util.concurrent.Executors.newFixedThreadPool(maxThreads) { r ->
-                Thread(r, "TextStreamWorker-${Thread.currentThread().threadId()}")
+                Thread(r, "TextStreamWorker-${textStreamWorkerNumber.incrementAndGet()}")
             }
+            // One global budget bounds all open ZIPs together. The old per-ZIP budget
+            // grew quadratically with --threads (128 * 512 queued entries at -j 128).
+            val maxStreamEntries = maxOf(maxThreads * 2, 32)
+            streamEntryPermits = java.util.concurrent.Semaphore(maxStreamEntries, true)
             val textParserMode = if (shouldParseDataXmlWithStax()) "StAX" else "DOM"
             LOGGER.info(
                 "Initialized ${textStreamingModeLabel()} streaming mode: archive-order entries, parallel entry processing, " +
-                    "no text-ID scheduling, data.xml via $textParserMode"
+                    "global in-flight limit=$maxStreamEntries, data.xml via $textParserMode"
             )
         } else {
             // For other formats, use priority-based executor
@@ -1691,6 +1771,17 @@
                 LOGGER.info("Excluded $excluded of $before zip(s) by glob(s): ${excludeZipGlobs.joinToString(", ")}")
             }
         }
+        if (canUseArchiveOrderTextStreaming() && !useLemma) {
+            val before = zips.size
+            zips = planPlainSurfaceTextInputs(zips)
+            val skipped = before - zips.size
+            if (skipped > 0) {
+                LOGGER.info(
+                    "Surface ${textStreamingModeLabel()} output reduced $before inputs to ${zips.size} unique base ZIP(s); " +
+                        "annotation ZIPs do not affect surface tokens"
+                )
+            }
+        }
         // Initialize zip progress tracking and sizes
         startTimeMillis = System.currentTimeMillis()
         processedZipBytes.set(0)
@@ -1702,10 +1793,10 @@
             registerZipProgress(zip, try { File(zip).length() } catch (_: Exception) { 0L })
         }
         totalZipBytes = zipSizes.values.sum()
-        // In lemma-only mode, process largest zips first
-        if (lemmaOnly) {
-            zips = zips.sortedByDescending { zipSizes[it] ?: 0L }.toTypedArray()
-        }
+        // Keep caller-controlled order by default. Size-descending scheduling can
+        // improve worker utilization near the end, but may put one large corpus
+        // domain at the beginning of word2vec training data, so it is opt-in.
+        zips = orderZipInputsForProcessing(zips)
         zips.forEachIndexed { index, zip -> zipOrdinals[zip] = index + 1 }
 
         // Log zip order with sizes so the user can verify sorting
@@ -1752,9 +1843,13 @@
 
         if (maxThreads > 1) {
             val foundry = getFoundryFromZipFileNames(zips)
-            val parallelism = maxThreads.coerceAtLeast(1)
+            val parallelism = effectiveZipParallelism(zips.size)
             if (canUseArchiveOrderTextStreaming()) {
-                LOGGER.info("Processing zips in ${textStreamingModeLabel()} streaming mode; zip parallelism=$parallelism; entry order=archive")
+                LOGGER.info(
+                    "Processing zips in ${textStreamingModeLabel()} streaming mode; " +
+                        "zip parallelism=$parallelism; entry threads=$maxThreads; " +
+                        "ZIP order=${if (largestFirst) "largest first" else "argument order"}"
+                )
             } else {
                 LOGGER.info("Processing zips with ordered queue; parallelism=$parallelism; entries ${if (sequentialInZip) "sequential" else "parallel"}")
             }
@@ -1766,6 +1861,13 @@
             }
         }
 
+        if (canUseArchiveOrderTextStreaming()) {
+            LOGGER.info(
+                "Text streaming scheduler summary: peak ZIP workers=${peakActiveZipWorkers.get()}, " +
+                    "peak entries in flight=${streamEntriesPeak.get()}, retained claims=${streamingOutputClaimCount()}"
+            )
+        }
+
         } finally {
         // Signal work-stealing scheduler that all foundries have been submitted
         if (workStealingSchedulerActive) {
@@ -2027,6 +2129,8 @@
         repeat(parallelism) {
             executor.submit {
                 active.incrementAndGet()
+                val activeNow = activeZipWorkers.incrementAndGet()
+                peakActiveZipWorkers.accumulateAndGet(activeNow, ::maxOf)
                 try {
                     while (true) {
                         val zipPath = queue.poll(100, java.util.concurrent.TimeUnit.MILLISECONDS)
@@ -2048,6 +2152,7 @@
                     }
                 } finally {
                     active.decrementAndGet()
+                    activeZipWorkers.decrementAndGet()
                 }
             }
         }
@@ -2863,14 +2968,16 @@
                 } else {
                     foundry  // Keep original foundry for non-krill formats
                 }
+                val requireRelatedMorpho =
+                    if (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) useLemma else true
                 if (useJavaZipForTextStreaming()) {
                     openJavaZipFile(zip).use { zipFile ->
                         LOGGER.info("Using ${textStreamingModeLabel()} streaming mode for $zip: archive-order entries, no text-ID sorting")
-                        processZipEntriesStreaming(zipFile, zip, zipFoundry, true)
+                        processZipEntriesStreaming(zipFile, zip, zipFoundry, requireRelatedMorpho)
                     }
                 } else {
                     openZipFile(zip).use { zipFile ->
-                        processZipEntriesWithPool(zipFile, zip, zipFoundry, true)
+                        processZipEntriesWithPool(zipFile, zip, zipFoundry, requireRelatedMorpho)
                     }
                 }
             }
@@ -2929,10 +3036,12 @@
                 } else {
                     foundry  // Keep original foundry for non-krill formats
                 }
+                val requireRelatedMorpho =
+                    if (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) useLemma else true
                 if (useJavaZipForTextStreaming()) {
                     openJavaZipFile(zip).use { zipFile ->
                         LOGGER.info("Using ${textStreamingModeLabel()} streaming mode for $zip: archive-order entries, no text-ID sorting")
-                        processZipEntriesStreaming(zipFile, zip, zipFoundry, true)
+                        processZipEntriesStreaming(zipFile, zip, zipFoundry, requireRelatedMorpho)
                     }
                 } else {
                     openZipFile(zip).use { zipFile ->
@@ -2941,7 +3050,7 @@
                             .filter { extractMetadataRegex.isNotEmpty() || !it.name.contains("header.xml") }
                             .sortedBy { getTextIdFromPath(it.name) }
                             .forEach { zipEntry ->
-                                processZipEntry(zipFile, zip, zipFoundry, zipEntry, true)
+                                processZipEntry(zipFile, zip, zipFoundry, zipEntry, requireRelatedMorpho)
                             }
                     }
                 }
@@ -2990,10 +3099,18 @@
             val pct = (done * 100.0 / total).coerceIn(0.0, 100.0)
             val humanSpeed = String.format(Locale.ROOT, "%.2f MB/s", speedBytesPerSec / (1024.0 * 1024.0))
             val etaStr = if (etaSeconds >= 0) formatDuration(etaSeconds) else "unknown"
+            val entryPool = entryExecutor as? java.util.concurrent.ThreadPoolExecutor
+            val schedulerState = if (canUseArchiveOrderTextStreaming()) {
+                ", workers{zip=${activeZipWorkers.get()},entry=${entryPool?.activeCount ?: 0}," +
+                    "queued=${entryPool?.queue?.size ?: 0},inFlight=${streamEntriesInFlight.get()}," +
+                    "claims=${streamingOutputClaims.values.sumOf { it.size }}}"
+            } else {
+                ""
+            }
             LOGGER.info(
                 "Finished zip ${if (ord>0) ord else "?"}/$totalZips: ${zipFilePath} " +
                         "(${humanBytes(size)}). Progress: ${String.format(Locale.ROOT, "%.1f", pct)}%, " +
-                        "ETA ${etaStr} at ${humanSpeed}"
+                        "ETA ${etaStr} at ${humanSpeed}$schedulerState"
             )
         } catch (e: Exception) {
             LOGGER.fine("Failed to log zip progress for $zipFilePath: ${e.message}")
@@ -3042,74 +3159,118 @@
         }
     }
 
-    private fun tryProcessReadyText(docId: String, foundry: String): Boolean {
-        if (!outputTexts.add(docId)) return false
+    private fun tryProcessReadyText(zipPath: String, docId: String, foundry: String): Boolean {
+        val claims = if (canUseArchiveOrderTextStreaming()) {
+            streamingOutputClaims.computeIfAbsent(zipPath) { ConcurrentHashMap.newKeySet() }
+        } else {
+            outputTexts
+        }
+        if (!claims.add(docId)) return false
         return try {
             processText(docId, foundry)
             noteFirstTextOutput(docId)
             true
         } catch (t: Throwable) {
-            outputTexts.remove(docId)
+            claims.remove(docId)
+            throw t
+        }
+    }
+
+    private fun beginStreamingZip(zipPath: String) {
+        streamingZipUsers.computeIfAbsent(zipPath) { AtomicInteger(0) }.incrementAndGet()
+    }
+
+    private fun endStreamingZip(zipPath: String) {
+        val users = streamingZipUsers[zipPath] ?: return
+        if (users.decrementAndGet() == 0) {
+            streamingZipUsers.remove(zipPath, users)
+            streamingOutputClaims.remove(zipPath)
+        }
+    }
+
+    private fun isRelevantStreamingEntry(name: String, zipFoundry: String): Boolean {
+        if (name.contains("header.xml")) return extractMetadataRegex.isNotEmpty()
+        if (outputFormat != OutputFormat.WORD2VEC && outputFormat != OutputFormat.NOW) return true
+        return when {
+            name.endsWith("data.xml") -> !lemmaOnly
+            name.endsWith("tokens.xml") || name.endsWith("structure.xml") -> true
+            // A base ZIP may use morpho.xml as its tokenization layer. Surface-form
+            // output extracts only those spans and does not retain morpho features.
+            name.endsWith("morpho.xml") -> useLemma || zipFoundry == "base"
+            else -> false
+        }
+    }
+
+    private fun submitStreamingEntry(phaser: java.util.concurrent.Phaser, task: () -> Unit) {
+        val permits = streamEntryPermits
+        permits?.acquireUninterruptibly()
+        phaser.register()
+        val inFlight = streamEntriesInFlight.incrementAndGet()
+        streamEntriesPeak.accumulateAndGet(inFlight) { current, update -> maxOf(current, update) }
+        try {
+            entryExecutor!!.execute {
+                try {
+                    task()
+                } finally {
+                    streamEntriesInFlight.decrementAndGet()
+                    permits?.release()
+                    phaser.arriveAndDeregister()
+                }
+            }
+        } catch (t: Throwable) {
+            streamEntriesInFlight.decrementAndGet()
+            permits?.release()
+            phaser.arriveAndDeregister()
             throw t
         }
     }
 
     private fun processZipEntriesStreaming(zipFile: ApacheZipFile, zipPath: String, foundry: String, waitForMorpho: Boolean) {
-        LOGGER.fine("Streaming NOW entries in archive order for $zipPath without text-ID sorting")
+        LOGGER.fine("Streaming ${textStreamingModeLabel()} entries in archive order for $zipPath")
         val enumEntries = zipFile.entries
+        val zipFoundry = getFoundryFromZipFileName(zipPath)
         val phaser = java.util.concurrent.Phaser(1)
-        val maxInFlight = maxOf(maxThreads * 4, 32)
-        val permits = java.util.concurrent.Semaphore(maxInFlight)
-
-        while (enumEntries.hasMoreElements()) {
-            val entry = enumEntries.nextElement()
-            if (extractMetadataRegex.isEmpty() && entry.name.contains("header.xml")) continue
-            if (entryExecutor != null && maxThreads > 1 && !sequentialInZip) {
-                permits.acquireUninterruptibly()
-                phaser.register()
-                entryExecutor!!.execute {
-                    try {
+        beginStreamingZip(zipPath)
+        try {
+            while (enumEntries.hasMoreElements()) {
+                val entry = enumEntries.nextElement()
+                if (!isRelevantStreamingEntry(entry.name, zipFoundry)) continue
+                if (entryExecutor != null && maxThreads > 1 && !sequentialInZip) {
+                    submitStreamingEntry(phaser) {
                         processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
-                    } finally {
-                        permits.release()
-                        phaser.arriveAndDeregister()
                     }
+                } else {
+                    processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
                 }
-            } else {
-                processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
             }
+            phaser.arriveAndAwaitAdvance()
+        } finally {
+            endStreamingZip(zipPath)
         }
-
-        phaser.arriveAndAwaitAdvance()
     }
 
     private fun processZipEntriesStreaming(zipFile: ZipFile, zipPath: String, foundry: String, waitForMorpho: Boolean) {
-        LOGGER.fine("Streaming NOW entries in archive order for $zipPath without text-ID sorting")
+        LOGGER.fine("Streaming ${textStreamingModeLabel()} entries in archive order for $zipPath")
         val enumEntries = zipFile.entries()
+        val zipFoundry = getFoundryFromZipFileName(zipPath)
         val phaser = java.util.concurrent.Phaser(1)
-        val maxInFlight = maxOf(maxThreads * 4, 32)
-        val permits = java.util.concurrent.Semaphore(maxInFlight)
-
-        while (enumEntries.hasMoreElements()) {
-            val entry = enumEntries.nextElement()
-            if (extractMetadataRegex.isEmpty() && entry.name.contains("header.xml")) continue
-            if (entryExecutor != null && maxThreads > 1 && !sequentialInZip) {
-                permits.acquireUninterruptibly()
-                phaser.register()
-                entryExecutor!!.execute {
-                    try {
+        beginStreamingZip(zipPath)
+        try {
+            while (enumEntries.hasMoreElements()) {
+                val entry = enumEntries.nextElement()
+                if (!isRelevantStreamingEntry(entry.name, zipFoundry)) continue
+                if (entryExecutor != null && maxThreads > 1 && !sequentialInZip) {
+                    submitStreamingEntry(phaser) {
                         processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
-                    } finally {
-                        permits.release()
-                        phaser.arriveAndDeregister()
                     }
+                } else {
+                    processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
                 }
-            } else {
-                processZipEntry(zipFile, zipPath, foundry, entry, waitForMorpho)
             }
+            phaser.arriveAndAwaitAdvance()
+        } finally {
+            endStreamingZip(zipPath)
         }
-
-        phaser.arriveAndAwaitAdvance()
     }
 
     private fun processZipEntriesWithPool(zipFile: ApacheZipFile, zipPath: String, foundry: String, waitForMorpho: Boolean) {
@@ -3507,8 +3668,12 @@
                     }
 
                     "morpho.xml" -> {
-                        waitForMorpho = true
-                        fnames[docId] = zipEntry.name
+                        if (useLemma || (outputFormat != OutputFormat.WORD2VEC && outputFormat != OutputFormat.NOW)) {
+                            waitForMorpho = true
+                        }
+                        val surfaceTextOutput =
+                            (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) && !useLemma
+                        if (!surfaceTextOutput) fnames[docId] = zipEntry.name
                         LOGGER.info("Processing morpho.xml for $docId with foundry=$annotationFoundry from ${zipEntry.name}")
                         val fsSpans: NodeList = doc.getElementsByTagName("span")
                         val morphoSpans = extractMorphoSpans(fsSpans)
@@ -3522,6 +3687,13 @@
                                 tokens[docId] = morphoTokens
                                 collectKrillTokensFromMorpho(docId, morphoFoundry, morphoTokens)
                             }
+                        } else if ((outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) && !useLemma) {
+                            // Surface-form text output needs morpho.xml only as a token-span
+                            // fallback for base ZIPs without tokens.xml. Do not retain the much
+                            // larger lemma/POS map after the text may already have been emitted.
+                            if (_foundry == "base" && tokens[docId] == null) {
+                                tokens[docId] = extractSpans(fsSpans, docId)
+                            }
                         } else {
                             // For other formats, use the shared morpho map
                             // Merge with existing morpho data (e.g., from dependency.xml)
@@ -3646,7 +3818,7 @@
                     && (extractMetadataRegex.isEmpty() || metadata[docId] != null)
                 ) {
                     LOGGER.fine("All data ready for $docId, calling processText")
-                    tryProcessReadyText(docId, annotationFoundry)
+                    tryProcessReadyText(zipPath, docId, annotationFoundry)
                 } else {
                     LOGGER.fine("NOT ready to process $docId yet: textOK=${texts[docId] != null || !textRequired}, sentencesOK=${sentences[docId] != null}, tokensOK=${tokens[docId] != null}, morphoOK=${!morphoRequired || morpho[docId] != null}")
                 }
@@ -3716,7 +3888,7 @@
                         && (!morphoRequired || morpho[docId] != null)
                     ) {
                         LOGGER.info("Processing text (meta-ready): $docId in thread ${Thread.currentThread().threadId()}")
-                        tryProcessReadyText(docId, foundry)
+                        tryProcessReadyText(zipPath, docId, foundry)
                     }
                 }
             }
@@ -3786,7 +3958,14 @@
                     }
                 }
                 "morpho.xml" -> {
-                    fnames[docId] = zipEntry.name
+                    val surfaceTextOutput =
+                        (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) && !useLemma
+                    if (surfaceTextOutput && tokens[docId] != null) {
+                        // Standard base ZIPs already supplied tokens.xml. Avoid parsing and
+                        // allocating an unused lemma/POS map from a later morpho layer.
+                        return
+                    }
+                    if (!surfaceTextOutput) fnames[docId] = zipEntry.name
                     val (morphoSpans, allSpans) = extractMorphoSpansStax(reader)
 
                     if (outputFormat == OutputFormat.KRILL) {
@@ -3796,6 +3975,11 @@
                             tokens[docId] = allSpans
                             collectKrillTokensFromMorpho(docId, morphoFoundry, allSpans)
                         }
+                    } else if (surfaceTextOutput) {
+                        // See the DOM path above: surface output uses only token offsets.
+                        if (_foundry == "base" && tokens[docId] == null) {
+                            tokens[docId] = allSpans
+                        }
                     } else {
                         val morphoMap = synchronized(morpho) {
                             morpho.getOrPut(docId) { morphoSpans }
@@ -3854,7 +4038,9 @@
                 processedTextsPerZip.getOrPut(zipPath) { mutableSetOf() }.add(docId)
             }
             
-            val effectiveWaitForMorpho = if (fileName == "morpho.xml") true else waitForMorpho
+            val surfaceTextOutput =
+                (outputFormat == OutputFormat.WORD2VEC || outputFormat == OutputFormat.NOW) && !useLemma
+            val effectiveWaitForMorpho = if (fileName == "morpho.xml" && !surfaceTextOutput) true else waitForMorpho
             
              val finalMorphoRequired = when {
                 taggerName != null || parserName != null -> false
@@ -3875,7 +4061,7 @@
                 && (!finalMorphoRequired || morpho[docId] != null)
                 && (extractMetadataRegex.isEmpty() || metadata[docId] != null)
             ) {
-                tryProcessReadyText(docId, annotationFoundry)
+                tryProcessReadyText(zipPath, docId, annotationFoundry)
             }
 
         } catch (e: Exception) {
@@ -3956,7 +4142,18 @@
             if (outputFormat == OutputFormat.KORAP_XML && annotationWorkerPool == null) {
                 formatKorapXmlOutput(getMorphoFoundry(), docId)
             } else {
-                formatConlluOutput(foundry, docId)
+                // In archive-order mode, data/structure/morpho entries for one text
+                // finish concurrently. Derive an input annotation foundry from the
+                // retained annotation filename so the last base-layer task cannot
+                // nondeterministically relabel custom morpho data as "base".
+                val inputFile = fnames[docId]
+                val outputFoundry = if (inputFile != null &&
+                    (inputFile.endsWith("morpho.xml") || inputFile.endsWith("dependency.xml"))) {
+                    annotationFoundryFor(inputFile, File(inputFile).name, foundry)
+                } else {
+                    foundry
+                }
+                formatConlluOutput(outputFoundry, docId)
             }
         }
 
@@ -4142,7 +4339,9 @@
             val max = rt.maxMemory() / (1024 * 1024)
             LOGGER.info(
                 "MEM-STATS docs=${count} usedMB=${used} totalMB=${total} maxMB=${max} " +
-                        "maps{texts=${texts.size},tokens=${tokens.size},sentences=${sentences.size},morpho=${morpho.size}}"
+                        "maps{texts=${texts.size},tokens=${tokens.size},sentences=${sentences.size},morpho=${morpho.size}} " +
+                        "stream{zipActive=${activeZipWorkers.get()},entriesInFlight=${streamEntriesInFlight.get()}," +
+                        "entriesPeak=${streamEntriesPeak.get()},claims=${streamingOutputClaims.values.sumOf { it.size }}}"
             )
         } catch (e: Exception) {
             LOGGER.warning("Failed to log memory stats: ${e.message}")
diff --git a/app/src/test/kotlin/de/ids_mannheim/korapxmltools/GeneralFeaturesTest.kt b/app/src/test/kotlin/de/ids_mannheim/korapxmltools/GeneralFeaturesTest.kt
index 5a7bf73..e307009 100644
--- a/app/src/test/kotlin/de/ids_mannheim/korapxmltools/GeneralFeaturesTest.kt
+++ b/app/src/test/kotlin/de/ids_mannheim/korapxmltools/GeneralFeaturesTest.kt
@@ -158,6 +158,57 @@
     }
 
     @Test
+    fun textStreamingSeparatesZipAndEntryParallelism() {
+        val tool = KorapXmlTool()
+        tool.maxThreads = 128
+
+        assertEquals(8, tool.effectiveZipParallelism(2600))
+        tool.zipParallelism = 3
+        assertEquals(3, tool.effectiveZipParallelism(2600))
+        assertEquals(2, tool.effectiveZipParallelism(2))
+    }
+
+    @Test
+    fun zipParallelismMustBePositive() {
+        val exitCode = debug(arrayOf(
+            "-t", "w2v", "--zip-parallelism", "0", loadResource("wdf19.zip").path
+        ))
+
+        assertTrue(exitCode != 0)
+        assertContains(errContent.toString(), "--zip-parallelism': must be at least 1")
+    }
+
+    @Test
+    fun zipSchedulingKeepsArgumentOrderByDefaultAndLargestFirstIsOptIn() {
+        val tool = KorapXmlTool()
+        val small = "/tmp/small.zip"
+        val large = "/tmp/large.zip"
+        tool.registerZipProgress(small, 10L)
+        tool.registerZipProgress(large, 100L)
+
+        assertEquals(
+            listOf(small, large),
+            tool.orderZipInputsForProcessing(arrayOf(small, large)).toList()
+        )
+
+        tool.largestFirst = true
+        assertEquals(
+            listOf(large, small),
+            tool.orderZipInputsForProcessing(arrayOf(small, large)).toList()
+        )
+    }
+
+    @Test
+    fun surfaceWord2VecUsesEachBaseZipOnlyOnce() {
+        val tool = KorapXmlTool()
+        tool.outputFormat = OutputFormat.WORD2VEC
+        val base = loadResource("goe.zip").path
+        val annotation = loadResource("goe.tree_tagger.zip").path
+
+        assertEquals(listOf(base), tool.planPlainSurfaceTextInputs(arrayOf(annotation, base)).toList())
+    }
+
+    @Test
     fun singleBaseConlluOutputCanUseArchiveOrderStreaming() {
         val tool = KorapXmlTool()
         tool.outputFormat = OutputFormat.CONLLU
diff --git a/app/src/test/kotlin/de/ids_mannheim/korapxmltools/Word2VecFormatterTest.kt b/app/src/test/kotlin/de/ids_mannheim/korapxmltools/Word2VecFormatterTest.kt
index 0e676bc..507bdc3 100644
--- a/app/src/test/kotlin/de/ids_mannheim/korapxmltools/Word2VecFormatterTest.kt
+++ b/app/src/test/kotlin/de/ids_mannheim/korapxmltools/Word2VecFormatterTest.kt
@@ -5,6 +5,7 @@
 import java.io.ByteArrayOutputStream
 import java.io.PrintStream
 import java.net.URL
+import picocli.CommandLine
 import kotlin.test.Test
 import kotlin.test.assertContains
 import kotlin.test.assertFalse
@@ -116,4 +117,29 @@
         assertContains(out, "automatique")
         assertFalse(out.contains("Gedanken"))
     }
+
+    @Test
+    fun streamingClaimsAreReleasedAfterZipCompletes() {
+        val tool = KorapXmlTool()
+        val exitCode = CommandLine(tool).execute(
+            "-t", "w2v", "-j", "4", loadResource("wdf19.zip").path
+        )
+
+        assertTrue(exitCode == 0)
+        assertTrue(tool.outputTexts.isEmpty(), "Plain w2v must not retain all emitted document IDs")
+        assertTrue(tool.streamingOutputClaimCount() == 0, "Per-ZIP claims must be released when the ZIP closes")
+    }
+
+    @Test
+    fun surfaceWord2VecSupportsMorphoTokenizationWithoutRetainingMorpho() {
+        val tool = KorapXmlTool()
+        val exitCode = CommandLine(tool).execute(
+            "-t", "w2v", "-j", "2", loadResource("dck_sample.zip").path
+        )
+
+        assertTrue(exitCode == 0)
+        assertTrue(outContent.size() > 0, "Custom morpho tokenization should produce surface output")
+        assertTrue(tool.morpho.isEmpty(), "Surface w2v must not retain unused morpho maps")
+        assertTrue(tool.fnames.isEmpty(), "Late optional layers must not repopulate per-document state")
+    }
 }