Add stage-3 content triage: sort unknown files into binary/ and text/

Unknown files are no longer terminal. Stage 1 now routes :unknown onto a
dedicated queue (with the same blocking backpressure as the known queue),
and a third worker pool sorts each file into data/binary/ or data/text/
using a NUL-byte sniff of the first 8000 bytes.

- content.jl: is_binary content sniff (stage 3)
- worker.jl: handle_unknown_job; stage-1 routes unknown with backpressure;
  KNOWN_ENQUEUE_RETRY_SECONDS -> ROUTE_ENQUEUE_RETRY_SECONDS (serves both)
- config.jl: unknown_worker_count/queue_capacity, binary_dir, text_dir + env
- FileServer.jl: unknown queue, pool, stage-aware recovery, drain ordering
- tests for is_binary and handle_unknown_job; tmp_config isolates new dirs
- README: three-stage pipeline

Project.toml

@@ -14,6 +14,13 @@ Oxygen = "df9a0d86-3283-4920-82dc-4555fc0d1d8b"
 UUIDs = "cf7118a7-6976-5b1a-9a39-7adc72f591a4"
 Zygote = "e88e6eb3-aa80-5325-afca-941959d7151f"
 
+[extras]
+JSON3 = "0f8b85d8-7281-11e9-16c2-39a750bddbf1"
+Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
+
+[targets]
+test = ["Test", "JSON3"]
+
 [compat]
 HTTP = "1.11.0"
 JLD2 = "0.6.4"

README.md

@@ -11,6 +11,10 @@ classifier that labels it **known** (a file type resembling the training set) or
 
 ## Architecture
 
+The pipeline is three stages, each with its own bounded queue and its own worker
+pool (tuned independently, since classification is CPU-bound, enrichment is
+process-/IO-bound, and content triage is cheap IO):
+
 ```
         POST /upload (multipart)
                  │
@@ -21,33 +25,113 @@ classifier that labels it **known** (a file type resembling the training set) or
         └────────┬─────────┘     enqueue reference (non-blocking)
                  │                        │
                  ▼                        ▼
-          202 + job IDs           ┌───────────────┐
-          (503 if full)           │  work queue   │  bounded, thread-safe
-                                  │  (Channel-ish)│
-                                  └───────┬───────┘
-                                          │ dequeue
-                          ┌───────────────┼───────────────┐
-                          ▼               ▼               ▼
-                      worker 1        worker 2   …    worker N   (Threads.@spawn)
-                          │
-              success ────┴──► data/done/<uuid>-<name>
-              failure  ───────► data/failed/<uuid>-<name>
+          202 + job IDs        ┌────────────────────┐
+          (503 if full)        │  stage-1 queue     │  classification
+                               └─────────┬──────────┘
+                                         │ dequeue
+                     ┌───────────────────┼───────────────────┐
+                     ▼                   ▼                    ▼
+              classify wkr 1      classify wkr 2   …   classify wkr N
+                     │
+        ┌────────────┴────────────┐
+    :unknown                    :known
+        │  move to data/unknown/,  │  move to data/known/, then
+        ▼  then enqueue (blocking) ▼  enqueue (blocking backpressure)
+  ┌────────────────────┐        ┌────────────────────┐
+  │  unknown queue     │        │  known queue       │  enrichment
+  └─────────┬──────────┘        └─────────┬──────────┘
+            │ dequeue                     │ dequeue
+   ┌────────┼────────┐        ┌───────────┼───────────┐
+   ▼        ▼        ▼        ▼           ▼           ▼
+ unk 1  unk 2 … unk K   known wkr 1  known wkr 2 … known wkr M
+   │  binary-vs-text sniff       │  exiftool → normalized sidecar
+   ├─► data/binary/<uuid>-<name> success ──┴──► data/done/<uuid>-<name>
+   └─► data/text/<uuid>-<name>              data/done/<uuid>-<name>.meta.json
+                                            (sidecar-first commit)
+                                 failure  ───────► data/failed/<uuid>-<name>
 ```
 
+Stages 2 (enrichment) and 3 (content triage) run in parallel: stage 1 feeds both
+the known and unknown queues.
+
 Key properties:
 
 - **Fast intake:** the queue only ever carries small references; file bytes live
   on disk, so memory stays flat regardless of file size.
-- **Backpressure:** the queue is bounded (default 1000). When full, uploads get
-  `503 Service Unavailable` instead of silently piling up.
-- **Crash-resilient:** files survive on disk. On startup, anything left in
-  `data/spool/` is re-enqueued (`recovered = N` in the log).
+- **Backpressure:** each queue is bounded (default 1000). When the *intake* queue
+  is full, uploads get `503 Service Unavailable`. When the *known* queue is full,
+  the stage-1 worker blocks and retries (a classified file is never dropped).
+- **Crash-resilient:** files survive on disk. On startup, recovery is
+  stage-aware: leftovers in `data/spool/` re-enter classification, `data/known/`
+  re-enter enrichment, and `data/unknown/` re-enter content triage (`recovered` /
+  `recovered_known` / `recovered_unknown` in the log), so a file resumes at its
+  correct stage instead of restarting from scratch.
 - **Graceful shutdown:** SIGINT (Ctrl-C) and SIGTERM (systemd/Docker/k8s `stop`)
-  both stop accepting uploads, drain the queue, wait for in-flight files to
-  finish, then exit. (See "Shutdown" below for one cosmetic caveat on SIGTERM.)
+  both stop accepting uploads, then drain the stages *in order* — close the
+  stage-1 queue and wait out the classify workers (the only producer of the known
+  *and* unknown queues) before closing those two queues and waiting out the
+  enrich and content-triage workers.
+  (See "Shutdown" below for one cosmetic caveat on SIGTERM.)
 - **Safe filenames:** client-supplied names are sanitized and prefixed with a
   server-minted UUID before touching the filesystem (no path traversal).
 
+### Metadata enrichment (stage 2)
+
+Files the classifier labels **known** are handed to a second pool that extracts
+metadata with [`exiftool`](https://exiftool.org/) (`exiftool -json -G -n`) —
+chosen because no native Julia library comes close to its multi-format coverage.
+The output is normalized into a small, stable, documented schema and written as a
+JSON **sidecar** next to the file in `data/done/`, e.g.
+`data/done/<uuid>-<name>.meta.json`. The original bytes are never modified.
+
+> **Prerequisite:** `exiftool` must be on `PATH` (e.g. `apt install
+> libimage-exiftool-perl`). The server **fails fast at startup** if it's missing.
+
+Sidecar top-level fields (all nullable — present only when available), plus the
+complete raw `exiftool` object under `raw`:
+
+| Field | Meaning |
+|---|---|
+| `id`, `original_name` | job id and client-supplied name |
+| `file_type`, `mime_type` | e.g. `PDF` / `application/pdf` |
+| `file_size` | bytes (authoritative, from intake — not exiftool) |
+| `created_date`, `modified_date` | content timestamps |
+| `author` | person (`Author`/`Artist`/`By-line`) |
+| `created_by` | authoring app/tool (`Producer`/`CreatorTool`/`Creator`/`Software`/…) |
+| `dimensions` | `{width, height}` for media |
+| `duration` | seconds, for audio/video |
+| `page_count` | for documents |
+| `error` | set on a *degraded* sidecar (see below) |
+| `raw` | full `exiftool` output |
+
+Each normalized field is a coalesce over a priority list of exiftool tags
+(`src/metadata.jl`); extend a field by appending tag names. If extraction fails
+or `exiftool` times out (`FS_EXIFTOOL_TIMEOUT`, default 30s), the file still
+completes to `data/done/` with a **degraded sidecar** — `file_size`/`file_type`
+plus an `error` note — rather than being quarantined, because it's still a wanted
+known file. Only genuine I/O errors (can't write the sidecar or move the file)
+send it to `data/failed/`.
+
+The sidecar is committed **before** the file is moved into `data/done/`, so a
+file's presence there always implies its sidecar is already present; a crash in
+between leaves only a harmless orphan sidecar, and recovery re-enriches
+idempotently.
+
+### Content triage (stage 3)
+
+Files the classifier labels **unknown** are handed to a third pool that sorts
+them into two coarse buckets so downstream tooling can treat them differently:
+
+- **`data/binary/`** — the file looks like binary data.
+- **`data/text/`** — the file looks like text.
+
+The test is the classic **NUL-byte sniff** (the same heuristic `git` and
+`file(1)` use): read the first 8000 bytes and, if any is NUL, call it binary,
+else text. It's cheap (no full read) and reliable in practice — text encodings
+don't embed NUL bytes, while binary formats almost always do near the start. An
+empty file has no NUL, so it's treated as text. This is deliberately simple for
+now; richer handling can hang off either bucket later (`src/content.jl`).
+
 ## The queue seam (→ RabbitMQ later)
 
 The HTTP handler and workers only ever call `enqueue!`, `dequeue!`, and
@@ -130,18 +214,27 @@ init, so the artifact is exactly regenerable from the same inputs.
 
 | Variable            | Default        | Meaning                                  |
 |---------------------|----------------|------------------------------------------|
-| `FS_HOST`           | `127.0.0.1`    | Bind address                             |
-| `FS_PORT`           | `8080`         | Port                                     |
-| `FS_WORKERS`        | `nthreads()`   | Number of worker tasks                   |
-| `FS_QUEUE_CAPACITY` | `1000`         | Max pending jobs before `503`            |
-| `FS_SPOOL_DIR`      | `data/spool`   | Incoming files (pending)                 |
-| `FS_DONE_DIR`       | `data/done`    | Files after successful processing        |
-| `FS_FAILED_DIR`     | `data/failed`  | Files whose processing threw             |
-| `FS_MODEL_PATH`     | `model/classifier.jld2` | Classifier artifact loaded at startup |
+| `FS_HOST`                 | `127.0.0.1`    | Bind address                                 |
+| `FS_PORT`                 | `8080`         | Port                                         |
+| `FS_WORKERS`              | `nthreads()`   | Stage-1 (classification) worker tasks        |
+| `FS_QUEUE_CAPACITY`       | `1000`         | Max pending intake jobs before `503`         |
+| `FS_KNOWN_WORKERS`        | `nthreads()`   | Stage-2 (enrichment) worker tasks            |
+| `FS_KNOWN_QUEUE_CAPACITY` | `1000`         | Max pending enrichment jobs (then backpressure) |
+| `FS_UNKNOWN_WORKERS`      | `nthreads()`   | Stage-3 (content triage) worker tasks        |
+| `FS_UNKNOWN_QUEUE_CAPACITY` | `1000`       | Max pending triage jobs (then backpressure)  |
+| `FS_SPOOL_DIR`            | `data/spool`   | Incoming files (pending classification)      |
+| `FS_KNOWN_DIR`            | `data/known`   | Classified-known, awaiting enrichment        |
+| `FS_UNKNOWN_DIR`          | `data/unknown` | Classified-unknown, awaiting content triage  |
+| `FS_BINARY_DIR`           | `data/binary`  | Stage-3 sink: unknown files that look binary  |
+| `FS_TEXT_DIR`             | `data/text`    | Stage-3 sink: unknown files that look like text |
+| `FS_DONE_DIR`             | `data/done`    | Enriched known files (+ `.meta.json`)        |
+| `FS_FAILED_DIR`           | `data/failed`  | Files whose processing threw                 |
+| `FS_MODEL_PATH`           | `model/classifier.jld2` | Classifier artifact loaded at startup |
+| `FS_EXIFTOOL_TIMEOUT`     | `30`           | Seconds before a stuck exiftool is killed    |
 
-> To get real parallelism, start Julia with enough threads (`-t N`) to match
-> `FS_WORKERS`. If `FS_WORKERS` exceeds available threads you'll get a warning
-> and workers will share threads.
+> To get real parallelism, start Julia with enough threads (`-t N`) to cover all
+> pools. If `FS_WORKERS + FS_KNOWN_WORKERS + FS_UNKNOWN_WORKERS` exceeds available
+> threads you'll get a warning (non-fatal) and workers will share threads.
 
 ## Usage
 
@@ -173,7 +266,9 @@ src/
   spool.jl        filename sanitizing, spool/move, startup recovery
   model.jl        NN architecture + byte→feature mapping (shared with trainer)
   classify.jl     load artifact + classify a file at inference time
-  worker.jl       worker loop + per-job processing (classify + move)
+  metadata.jl     exiftool extraction + normalized sidecar (stage 2)
+  content.jl      binary-vs-text sniff for unknown files (stage 3)
+  worker.jl       parametrized worker loop + classify/enrich/triage handlers
   server.jl       HTTP routes/handlers
 bin/
   server.jl       entry point

src/FileServer.jl

@@ -14,13 +14,17 @@ include("queue.jl")
 include("spool.jl")
 include("model.jl")     # build_model() + read_features(); shared with bin/train.jl
 include("classify.jl")  # Classifier + load_classifier/classify (needs model.jl)
+include("metadata.jl")  # exiftool extraction + sidecar enrichment (stage 2)
+include("content.jl")   # binary-vs-text triage for unknown files (stage 3)
 include("worker.jl")
 
 # Globals the HTTP handlers read at request time. Set once in `run`, before the
 # server starts accepting connections. Declared after the includes above so the
 # `Config`/`ChannelQueue` types exist.
 const CONFIG = Ref{Config}()
-const QUEUE  = Ref{ChannelQueue}()
+const QUEUE  = Ref{ChannelQueue}()          # stage-1 (classification) queue; HTTP intake enqueues here
+const KNOWN_QUEUE = Ref{ChannelQueue}()     # stage-2 (enrichment) queue; stage-1 workers enqueue here
+const UNKNOWN_QUEUE = Ref{ChannelQueue}()   # stage-3 (content triage) queue; stage-1 workers enqueue here
 const CLASSIFIER = Ref{Classifier}()  # loaded once at startup, shared read-only across workers
 
 include("server.jl")   # registers routes (references CONFIG/QUEUE at call time)
@@ -68,23 +72,45 @@ function run(; overrides...)
     cfg = config_from_env(; overrides...)
     ensure_dirs(cfg)
 
-    if cfg.worker_count > Threads.nthreads()
-        @warn "worker_count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" worker_count=cfg.worker_count nthreads=Threads.nthreads()
+    # All pools draw from the same OS threads. Warn on the *combined* size (still
+    # allowed): oversubscription just means tasks share threads, not a failure.
+    total_workers = cfg.worker_count + cfg.known_worker_count + cfg.unknown_worker_count
+    if total_workers > Threads.nthreads()
+        @warn "combined worker count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" classify_workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count total=total_workers nthreads=Threads.nthreads()
     end
 
-    queue = ChannelQueue(cfg.queue_capacity)
-    CONFIG[] = cfg
-    QUEUE[]  = queue
+    # exiftool is a hard prerequisite for stage-2 enrichment. Fail fast at
+    # startup rather than discover it missing on the first known file.
+    assert_exiftool()
+
+    queue         = ChannelQueue(cfg.queue_capacity)
+    known_queue   = ChannelQueue(cfg.known_queue_capacity)
+    unknown_queue = ChannelQueue(cfg.unknown_queue_capacity)
+    CONFIG[]        = cfg
+    QUEUE[]         = queue
+    KNOWN_QUEUE[]   = known_queue
+    UNKNOWN_QUEUE[] = unknown_queue
 
     # Load the classifier before serving. Fail fast: a server that silently
     # doesn't classify is a worse surprise than a clear startup error.
     CLASSIFIER[] = load_classifier(cfg.model_path)
     @info "loaded classifier" path=cfg.model_path
 
-    recovered = recover_spool!(cfg, queue)
-    @info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count capacity=cfg.queue_capacity recovered=recovered
+    # Stage-aware recovery: re-drive each stage's leftovers onto its own queue so
+    # files resume where they were, not from scratch. spool/ → stage-1,
+    # known/ → stage-2, unknown/ → stage-3.
+    recovered         = recover_dir!(cfg.spool_dir, queue)
+    recovered_known   = recover_dir!(cfg.known_dir, known_queue)
+    recovered_unknown = recover_dir!(cfg.unknown_dir, unknown_queue)
+    @info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity recovered=recovered recovered_known=recovered_known recovered_unknown=recovered_unknown
 
-    workers = [Threads.@spawn worker_loop(i, cfg, queue) for i in 1:cfg.worker_count]
+    workers = [Threads.@spawn worker_loop(i, cfg, queue,
+                   (job, c, wid) -> handle_classify_job(job, c, wid, known_queue, unknown_queue))
+               for i in 1:cfg.worker_count]
+    known_workers = [Threads.@spawn worker_loop(i, cfg, known_queue, handle_known_job)
+                     for i in 1:cfg.known_worker_count]
+    unknown_workers = [Threads.@spawn worker_loop(i, cfg, unknown_queue, handle_unknown_job)
+                       for i in 1:cfg.unknown_worker_count]
 
     register_routes()
     serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false)
@@ -99,10 +125,15 @@ function run(; overrides...)
     drained = Threads.Atomic{Bool}(false)
     function drain()
         Threads.atomic_xchg!(drained, true) && return   # run at most once
-        @info "draining queue and stopping workers"
-        terminate()               # stop accepting new HTTP requests
-        close!(queue)             # let workers drain buffered jobs, then exit
-        foreach(wait, workers)
+        @info "draining queues and stopping workers"
+        terminate()                    # 1. stop accepting new HTTP requests
+        close!(queue)                  # 2. no new classify jobs; stage-1 drains buffered
+        foreach(wait, workers)         # 3. wait out stage-1 — the ONLY producer of BOTH the
+                                       #    known and unknown queues — so nothing else enqueues
+        close!(known_queue)            # 4. now safe to close the downstream queues
+        close!(unknown_queue)
+        foreach(wait, known_workers)   # 5. wait out stage-2 and stage-3
+        foreach(wait, unknown_workers)
         @info "shutdown complete"
     end
     atexit(drain)

src/config.jl

@@ -7,10 +7,25 @@ Base.@kwdef struct Config
     port::Int           = 8080
     worker_count::Int   = Threads.nthreads()
     queue_capacity::Int = 1000
-    spool_dir::String   = "data/spool"   # files land here on intake (pending)
-    done_dir::String    = "data/done"    # files move here after successful processing
+    # Stage 2 (enrichment) has its own pool + queue: exiftool work is process-spawn
+    # and I/O bound, a different cost profile than the CPU-bound Lux classify, so
+    # the two pools are tuned independently.
+    known_worker_count::Int   = Threads.nthreads()
+    known_queue_capacity::Int = 1000
+    # Stage 3 (content triage) also gets its own pool + queue: sorting an
+    # unrecognized file into binary/ vs text/ is cheap I/O, tuned independently
+    # of the classify and enrich pools.
+    unknown_worker_count::Int   = Threads.nthreads()
+    unknown_queue_capacity::Int = 1000
+    spool_dir::String   = "data/spool"   # files land here on intake (pending classification)
+    known_dir::String   = "data/known"   # classified-known, awaiting enrichment (stage 2)
+    unknown_dir::String = "data/unknown" # classified-unknown, awaiting content triage (stage 3)
+    binary_dir::String  = "data/binary"  # stage-3 sink: unknown files that look like binary data
+    text_dir::String    = "data/text"    # stage-3 sink: unknown files that look like text
+    done_dir::String    = "data/done"    # fully enriched known files (+ .meta.json sidecars)
     failed_dir::String  = "data/failed"  # files move here if a worker throws
     model_path::String  = "model/classifier.jld2"  # committed classifier artifact, loaded at startup
+    exiftool_timeout::Int = 30           # seconds before a stuck exiftool is killed → degraded sidecar
 end
 
 """
@@ -22,26 +37,43 @@ and for `FileServer.run(; port=...)`).
 
 Recognised variables:
   FS_HOST, FS_PORT, FS_WORKERS, FS_QUEUE_CAPACITY,
-  FS_SPOOL_DIR, FS_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH
+  FS_KNOWN_WORKERS, FS_KNOWN_QUEUE_CAPACITY,
+  FS_UNKNOWN_WORKERS, FS_UNKNOWN_QUEUE_CAPACITY,
+  FS_SPOOL_DIR, FS_KNOWN_DIR, FS_UNKNOWN_DIR, FS_BINARY_DIR, FS_TEXT_DIR,
+  FS_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH, FS_EXIFTOOL_TIMEOUT
 """
 function config_from_env(; host=nothing, port=nothing, worker_count=nothing,
-                          queue_capacity=nothing, spool_dir=nothing,
-                          done_dir=nothing, failed_dir=nothing, model_path=nothing)
+                          queue_capacity=nothing, known_worker_count=nothing,
+                          known_queue_capacity=nothing, unknown_worker_count=nothing,
+                          unknown_queue_capacity=nothing, spool_dir=nothing,
+                          known_dir=nothing, unknown_dir=nothing, binary_dir=nothing,
+                          text_dir=nothing, done_dir=nothing,
+                          failed_dir=nothing, model_path=nothing, exiftool_timeout=nothing)
     Config(
-        host           = something(host,           get(ENV, "FS_HOST", "127.0.0.1")),
-        port           = something(port,           parse(Int, get(ENV, "FS_PORT", "8080"))),
-        worker_count   = something(worker_count,   parse(Int, get(ENV, "FS_WORKERS", string(Threads.nthreads())))),
-        queue_capacity = something(queue_capacity, parse(Int, get(ENV, "FS_QUEUE_CAPACITY", "1000"))),
-        spool_dir      = something(spool_dir,      get(ENV, "FS_SPOOL_DIR", "data/spool")),
-        done_dir       = something(done_dir,       get(ENV, "FS_DONE_DIR", "data/done")),
-        failed_dir     = something(failed_dir,     get(ENV, "FS_FAILED_DIR", "data/failed")),
-        model_path     = something(model_path,     get(ENV, "FS_MODEL_PATH", "model/classifier.jld2")),
+        host                   = something(host,                   get(ENV, "FS_HOST", "127.0.0.1")),
+        port                   = something(port,                   parse(Int, get(ENV, "FS_PORT", "8080"))),
+        worker_count           = something(worker_count,           parse(Int, get(ENV, "FS_WORKERS", string(Threads.nthreads())))),
+        queue_capacity         = something(queue_capacity,         parse(Int, get(ENV, "FS_QUEUE_CAPACITY", "1000"))),
+        known_worker_count     = something(known_worker_count,     parse(Int, get(ENV, "FS_KNOWN_WORKERS", string(Threads.nthreads())))),
+        known_queue_capacity   = something(known_queue_capacity,   parse(Int, get(ENV, "FS_KNOWN_QUEUE_CAPACITY", "1000"))),
+        unknown_worker_count   = something(unknown_worker_count,   parse(Int, get(ENV, "FS_UNKNOWN_WORKERS", string(Threads.nthreads())))),
+        unknown_queue_capacity = something(unknown_queue_capacity, parse(Int, get(ENV, "FS_UNKNOWN_QUEUE_CAPACITY", "1000"))),
+        spool_dir              = something(spool_dir,              get(ENV, "FS_SPOOL_DIR", "data/spool")),
+        known_dir              = something(known_dir,              get(ENV, "FS_KNOWN_DIR", "data/known")),
+        unknown_dir            = something(unknown_dir,            get(ENV, "FS_UNKNOWN_DIR", "data/unknown")),
+        binary_dir             = something(binary_dir,             get(ENV, "FS_BINARY_DIR", "data/binary")),
+        text_dir               = something(text_dir,               get(ENV, "FS_TEXT_DIR", "data/text")),
+        done_dir               = something(done_dir,               get(ENV, "FS_DONE_DIR", "data/done")),
+        failed_dir             = something(failed_dir,             get(ENV, "FS_FAILED_DIR", "data/failed")),
+        model_path             = something(model_path,             get(ENV, "FS_MODEL_PATH", "model/classifier.jld2")),
+        exiftool_timeout       = something(exiftool_timeout,       parse(Int, get(ENV, "FS_EXIFTOOL_TIMEOUT", "30"))),
     )
 end
 
-"Create the spool/done/failed directories if they don't already exist."
+"Create all the pipeline-stage directories if they don't already exist."
 function ensure_dirs(cfg::Config)
-    for d in (cfg.spool_dir, cfg.done_dir, cfg.failed_dir)
+    for d in (cfg.spool_dir, cfg.known_dir, cfg.unknown_dir, cfg.binary_dir,
+              cfg.text_dir, cfg.done_dir, cfg.failed_dir)
         mkpath(d)
     end
     return nothing

src/content.jl

@@ -0,0 +1,24 @@
+# Stage-3 content triage for unknown files.
+#
+# A file that stage-1 couldn't recognize is still sorted into one of two coarse
+# buckets so downstream tooling can treat them differently: `text/` for
+# human-readable content, `binary/` for everything else. The test is the classic
+# "NUL byte in the first sniff window" heuristic that git and file(1) use — cheap
+# (no full read), and reliable in practice: text encodings don't embed NUL bytes,
+# while binary formats almost always do near the start.
+
+const CONTENT_SNIFF_BYTES = 8000
+
+"""
+    is_binary(path) -> Bool
+
+Classify a file as binary (`true`) or text (`false`) by sniffing its first
+`CONTENT_SNIFF_BYTES` bytes for a NUL byte. An empty file has no NUL, so it is
+treated as text.
+"""
+function is_binary(path::AbstractString)::Bool
+    open(path, "r") do io
+        chunk = read(io, CONTENT_SNIFF_BYTES)
+        return any(==(0x00), chunk)
+    end
+end

src/metadata.jl

@@ -0,0 +1,207 @@
+# Stage-2 metadata extraction and enrichment.
+#
+# Known files are enriched by shelling out to `exiftool -json -G` (the only tool
+# with broad, multi-format coverage — there is no comparable native Julia
+# library), then normalizing its output into a small, stable, documented schema
+# that downstream consumers can rely on, while preserving the full raw dump.
+#
+# exiftool being installed is a hard startup prerequisite (see `assert_exiftool`,
+# called from `run`). A per-file extraction failure or hang does NOT quarantine
+# the file — it produces a *degraded* sidecar recording what we know plus the
+# error, because a file that passed classification is wanted regardless of
+# whether we could read its metadata.
+
+"Throw at startup if the `exiftool` binary isn't on PATH — fail fast rather than discover it per file."
+function assert_exiftool()
+    try
+        Base.run(pipeline(`exiftool -ver`; stdout=devnull, stderr=devnull))
+    catch
+        error("exiftool not found on PATH (install libimage-exiftool-perl / exiftool). It is required for stage-2 metadata enrichment.")
+    end
+    return nothing
+end
+
+# Each normalized field is a coalesce over exiftool tag names, tried in order;
+# the first present, non-empty value wins. exiftool with `-G` prefixes tags by
+# group (e.g. "EXIF:Software"), so we match on the bare tag name after the last
+# colon. Extend a field simply by appending tag names here.
+#
+# Note the documented `Creator` ambiguity: in PDF it's the authoring app, but
+# elsewhere it's the person. We take the simple route — `Creator` feeds
+# `created_by` only, and `author` relies on the person-specific tags.
+const CREATED_BY_TAGS = ["Producer", "CreatorTool", "Creator", "Software", "Application", "Encoder", "EncodingTool", "HostComputer"]
+const AUTHOR_TAGS      = ["Author", "Artist", "By-line", "Owner", "Artist"]
+const CREATED_DATE_TAGS  = ["DateTimeOriginal", "CreateDate", "MediaCreateDate", "CreationDate"]
+const MODIFIED_DATE_TAGS = ["ModifyDate", "FileModifyDate"]
+
+"fsync an open file descriptor, throwing on failure — used to make a write durable before a rename commits it."
+function fsync_fd(fd)
+    ccall(:fsync, Cint, (Cint,), fd) == 0 || error("fsync failed: $(Base.Libc.strerror())")
+    return nothing
+end
+
+"fsync a directory so a rename into it survives a crash (the rename, not just the file bytes, must be persisted)."
+function fsync_dir(dir::AbstractString)
+    dfd = ccall(:open, Cint, (Cstring, Cint), dir, 0)   # O_RDONLY
+    dfd < 0 && error("cannot open dir for fsync: $dir ($(Base.Libc.strerror()))")
+    try
+        fsync_fd(dfd)
+    finally
+        ccall(:close, Cint, (Cint,), dfd)
+    end
+    return nothing
+end
+
+"Strip exiftool's `-G` group prefix (`EXIF:Software` → `Software`) so lookups are group-agnostic."
+strip_group(tag::AbstractString) = String(last(split(tag, ':')))
+
+"Return the first present, non-empty value among `tags` in the group-stripped map, or `nothing`."
+function coalesce_tag(bytag::Dict{String,Any}, tags)
+    for t in tags
+        v = get(bytag, t, nothing)
+        v === nothing && continue
+        s = string(v)
+        isempty(strip(s)) && continue
+        return v
+    end
+    return nothing
+end
+
+"""
+    run_exiftool(path, timeout) -> Union{Dict{String,Any},Nothing}
+
+Run `exiftool -json -G` on `path`, returning the parsed tag object, or `nothing`
+on non-zero exit, unparseable output, or timeout. The subprocess is killed after
+`timeout` seconds so one pathological file can't wedge a worker forever.
+"""
+function run_exiftool(path::AbstractString, timeout::Integer)
+    out = IOBuffer()
+    # -json: machine output; -G: group-prefixed tags; -n: numeric (unformatted)
+    # values so sizes/durations are numbers, not display strings.
+    proc = Base.run(pipeline(`exiftool -json -G -n $path`; stdout=out, stderr=devnull); wait=false)
+
+    # Kill the process if it overruns the timeout. `t` polls rather than blocking
+    # so we can `kill` a hung exiftool; the poll interval bounds shutdown latency.
+    killed = Ref(false)
+    t = Threads.@spawn begin
+        waited = 0.0
+        while process_running(proc) && waited < timeout
+            sleep(0.1); waited += 0.1
+        end
+        if process_running(proc)
+            killed[] = true
+            kill(proc, Base.SIGTERM)
+            # Escalate: a process that ignores/defers SIGTERM would otherwise pin
+            # the worker forever on the wait(proc) below, defeating the timeout.
+            grace = 0.0
+            while process_running(proc) && grace < 2.0
+                sleep(0.1); grace += 0.1
+            end
+            process_running(proc) && kill(proc, Base.SIGKILL)
+        end
+    end
+    wait(proc)
+    wait(t)
+
+    (killed[] || !success(proc)) && return nothing
+
+    parsed = try
+        JSON3.read(String(take!(out)))
+    catch
+        return nothing
+    end
+    # exiftool -json emits a one-element array of objects (one per input file).
+    (parsed isa AbstractVector && !isempty(parsed)) || return nothing
+    return Dict{String,Any}(String(strip_group(String(k))) => v for (k, v) in pairs(parsed[1]))
+end
+
+"""
+    build_metadata(job, cfg) -> NamedTuple
+
+Extract and normalize metadata for a known file. Always returns a sidecar
+payload: on extraction success, the normalized fields plus the full raw dump; on
+failure/timeout, a *degraded* payload with what we know from the Job plus an
+`error` note. `file_size` always comes from the Job (authoritative), never
+exiftool.
+"""
+function build_metadata(job::Job, cfg::Config)
+    bytag = run_exiftool(job.path, cfg.exiftool_timeout)
+
+    if bytag === nothing
+        return (
+            id            = job.id,
+            original_name = job.original_name,
+            file_type     = nothing,
+            mime_type     = nothing,
+            file_size     = job.size,
+            created_date  = nothing,
+            modified_date = nothing,
+            author        = nothing,
+            created_by    = nothing,
+            dimensions    = nothing,
+            duration      = nothing,
+            page_count    = nothing,
+            error         = "exiftool extraction failed or timed out",
+            raw           = nothing,
+        )
+    end
+
+    return normalize_metadata(job, bytag)
+end
+
+"Build the normalized sidecar payload from a successful exiftool tag map."
+function normalize_metadata(job::Job, bytag::Dict{String,Any})
+    w = get(bytag, "ImageWidth", nothing)
+    h = get(bytag, "ImageHeight", nothing)
+    dims = (w !== nothing && h !== nothing) ? (; width = w, height = h) : nothing
+
+    return (
+        id            = job.id,
+        original_name = job.original_name,
+        file_type     = get(bytag, "FileType", get(bytag, "FileTypeExtension", nothing)),
+        mime_type     = get(bytag, "MIMEType", nothing),
+        file_size     = job.size,                              # authoritative, from intake
+        created_date  = coalesce_tag(bytag, CREATED_DATE_TAGS),
+        modified_date = coalesce_tag(bytag, MODIFIED_DATE_TAGS),
+        author        = coalesce_tag(bytag, AUTHOR_TAGS),
+        created_by    = coalesce_tag(bytag, CREATED_BY_TAGS),
+        dimensions    = dims,
+        duration      = get(bytag, "Duration", get(bytag, "MediaDuration", nothing)),
+        page_count    = get(bytag, "PageCount", nothing),
+        error         = nothing,
+        raw           = bytag,
+    )
+end
+
+"""
+    finalize_known!(cfg, job, meta) -> (file_dest, sidecar_dest)
+
+Commit an enriched known file to `done/` with the sidecar-first ordering so the
+invariant *"a file in done/ implies its sidecar is already there"* always holds.
+
+Sequence: write `<name>.meta.json` to a temp name, fsync its bytes, rename it
+into place, fsync `done/` so the rename itself is durable, THEN move the file
+into `done/`. A crash between the two leaves only a harmless orphan sidecar in
+`done/` while the file stays in `known/`, so stage-aware recovery re-drives it
+and overwrites the sidecar — idempotent. The fsyncs make the ordering hold
+across power loss, not just process crashes.
+"""
+function finalize_known!(cfg::Config, job::Job, meta)
+    base        = basename(job.path)
+    sidecar     = joinpath(cfg.done_dir, string(base, ".meta.json"))
+    tmp_sidecar = string(sidecar, ".tmp")
+
+    # Write to a temp name then rename, so a reader in done/ never sees a partial
+    # sidecar and a crash mid-write can't masquerade as a committed one.
+    open(tmp_sidecar, "w") do io
+        write(io, JSON3.write(meta))
+        flush(io)
+        fsync_fd(fd(io))                   # durably persist bytes before the rename
+    end
+    mv(tmp_sidecar, sidecar; force=true)   # sidecar committed first
+    fsync_dir(cfg.done_dir)                # persist the rename itself, not just the bytes
+
+    file_dest = move_to(cfg.done_dir, job) # file arrival = commit point
+    return (file_dest, sidecar)
+end
+

src/spool.jl

@@ -42,17 +42,22 @@ end
 const UUID_LEN = 36
 
 """
-    recover_spool!(cfg, queue) -> Int
+    recover_dir!(dir, queue) -> Int
 
-Re-enqueue any files already sitting in the spool directory (left by a crash,
-a hard shutdown, or an intake that never got processed). This is the payoff of
-spooling to disk: a restart resumes work instead of stranding it. Returns the
-number of files recovered.
+Re-enqueue any files sitting in `dir` (left by a crash, hard shutdown, or an
+intake that never finished) onto `queue`. This is the payoff of spooling to
+disk: a restart resumes work instead of stranding it. Stage-aware recovery uses
+one call per stage — `spool/` → stage-1 queue, `known/` → known queue — so each
+file re-enters at the correct stage rather than being reclassified from scratch.
+Returns the number of files recovered.
+
+Skips `.meta.json` sidecars: those are stage-2 output, not work to redo.
 """
-function recover_spool!(cfg::Config, queue::JobQueue)::Int
+function recover_dir!(dir::AbstractString, queue::JobQueue)::Int
     n = 0
-    for path in sort(readdir(cfg.spool_dir; join=true))
+    for path in sort(readdir(dir; join=true))
         isfile(path) || continue
+        endswith(path, ".meta.json") && continue   # sidecar, not a work item
         fname = basename(path)
         if length(fname) > UUID_LEN + 1
             id   = fname[1:UUID_LEN]

src/worker.jl

@@ -1,39 +1,103 @@
-# Worker task: pull jobs off the queue and process them. One of these runs per
-# configured worker, each as its own `Threads.@spawn`'d task.
+# Worker tasks: pull jobs off a queue and process them. The loop scaffolding
+# (dequeue-until-drained, try/catch, quarantine-on-throw) is identical for every
+# stage, so `worker_loop` is parametrized with a `handler` and reused. Today
+# there are three stages:
+#
+#   stage 1  handle_classify_job  spool/  → classify → known/ (+known queue) | unknown/ (+unknown queue)
+#   stage 2  handle_known_job     known/  → exiftool enrich → done/ (+ .meta.json)
+#   stage 3  handle_unknown_job   unknown/ → binary-vs-text sniff → binary/ | text/
+#
+# Adding a stage later is just another queue + pool + handler; the loop below
+# doesn't change.
+
+# How long a stage-1 worker backs off before retrying an enqueue onto a full
+# downstream queue (known or unknown). Blocking backpressure: a classified file
+# is never dropped, so the stage-1 worker parks until the next stage makes room.
+# Keeps intake decoupled — the HTTP path's `enqueue!` stays non-blocking; only
+# this worker-to-worker handoff blocks.
+const ROUTE_ENQUEUE_RETRY_SECONDS = 0.05
 
 """
-    handle_job(job, cfg, worker_id)
+    handle_classify_job(job, cfg, worker_id, known_queue, unknown_queue)
 
-Do the work for a single job, then move the file to `done/`.
+Stage 1. Classify the spooled file and route it to the next stage's queue,
+retrying on a full queue rather than dropping the file:
+  * `:known`   → move to `known/`,   enqueue onto the known queue for stage 2.
+  * `:unknown` → move to `unknown/`, enqueue onto the unknown queue for stage 3.
 
-For now the "work" is just logging the received filename to prove the flow —
-this is the seam where real heavy-lifting will go later.
+In both cases move first so the file physically lives in its stage dir before the
+reference is visible downstream; the moved path becomes the routed job's location.
+
+Sub-`MIN_FILE_BYTES` files short-circuit to `:unknown` inside `classify`.
 """
-function handle_job(job::Job, cfg::Config, worker_id::Int)
-    # Classify the spooled file (annotate-only for now: the result is logged but
-    # every file still moves to done/ regardless of known/unknown). Sub-32-byte
-    # files short-circuit to :unknown inside classify without touching the model.
+function handle_classify_job(job::Job, cfg::Config, worker_id::Int,
+                             known_queue::JobQueue, unknown_queue::JobQueue)
     classification = classify(CLASSIFIER[], job.path)
-    @info "received file" worker=worker_id id=job.id name=job.original_name size=job.size classification=classification
-    dest = move_to(cfg.done_dir, job)
-    @info "completed" worker=worker_id id=job.id dest=dest
+    @info "classified file" worker=worker_id id=job.id name=job.original_name size=job.size classification=classification
+
+    if classification === :known
+        dest = move_to(cfg.known_dir, job)
+        routed = Job(job.id, job.original_name, dest, job.size, job.received_at)
+        while !enqueue!(known_queue, routed)
+            sleep(ROUTE_ENQUEUE_RETRY_SECONDS)   # known queue full → back off, don't drop
+        end
+        @info "routed to enrichment" worker=worker_id id=job.id dest=dest
+    else
+        dest = move_to(cfg.unknown_dir, job)
+        routed = Job(job.id, job.original_name, dest, job.size, job.received_at)
+        while !enqueue!(unknown_queue, routed)
+            sleep(ROUTE_ENQUEUE_RETRY_SECONDS)   # unknown queue full → back off, don't drop
+        end
+        @info "routed to content triage" worker=worker_id id=job.id dest=dest
+    end
     return nothing
 end
 
 """
-    worker_loop(worker_id, cfg, queue)
+    handle_known_job(job, cfg, worker_id)
 
-Consume jobs until the queue is closed and drained. A failure on one job is
-logged and the file is quarantined in `failed/` — it must never kill the
-worker, or the pool would silently shrink.
+Stage 2. Extract metadata (exiftool, with timeout) and enrich: build the
+normalized sidecar and commit both to `done/` sidecar-first. Extraction
+failure/timeout yields a *degraded* sidecar (the file is still a wanted known
+file), so the only way to land in `failed/` is a genuine I/O error writing the
+sidecar or moving the file — handled by `worker_loop`'s quarantine.
 """
-function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue)
+function handle_known_job(job::Job, cfg::Config, worker_id::Int)
+    meta = build_metadata(job, cfg)
+    file_dest, sidecar = finalize_known!(cfg, job, meta)
+    @info "enriched" worker=worker_id id=job.id dest=file_dest sidecar=basename(sidecar) file_type=meta.file_type created_by=meta.created_by degraded=(meta.error !== nothing)
+    return nothing
+end
+
+"""
+    handle_unknown_job(job, cfg, worker_id)
+
+Stage 3. Sort an unrecognized file into a coarse content bucket by sniffing its
+first bytes: `binary/` if it looks like binary data, `text/` otherwise. Terminal
+— there is no further stage. Simple by design for now; richer handling can hang
+off either bucket later.
+"""
+function handle_unknown_job(job::Job, cfg::Config, worker_id::Int)
+    binary = is_binary(job.path)
+    dest = move_to(binary ? cfg.binary_dir : cfg.text_dir, job)
+    @info "sorted unknown" worker=worker_id id=job.id name=job.original_name kind=(binary ? :binary : :text) dest=dest
+    return nothing
+end
+
+"""
+    worker_loop(worker_id, cfg, queue, handler)
+
+Consume jobs from `queue` until it is closed and drained, running `handler` on
+each. A failure on one job is logged and the file quarantined in `failed/` — it
+must never kill the worker, or the pool would silently shrink.
+"""
+function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue, handler)
     @info "worker started" worker=worker_id
     while true
         job = dequeue!(queue)
         job === nothing && break   # queue closed and drained → exit
         try
-            handle_job(job, cfg, worker_id)
+            handler(job, cfg, worker_id)
         catch e
             @error "processing failed" worker=worker_id id=job.id name=job.original_name exception=(e, catch_backtrace())
             try

test/runtests.jl

@@ -0,0 +1,215 @@
+using Test
+using FileServer
+using JSON3
+
+# Pull internals into scope. These aren't exported (only `run` is), but the
+# whole risk profile of this pipeline lives in these functions, so we test them
+# directly rather than only through the HTTP surface.
+using FileServer: Job, Config, ChannelQueue, enqueue!, dequeue!, length,
+                  sanitize_filename, recover_dir!, normalize_metadata,
+                  build_metadata, finalize_known!, run_exiftool,
+                  is_binary, handle_unknown_job
+
+# A minimal, valid 1×1 PNG. Lets the real-exiftool tests assert stable facts
+# (FileType == "PNG", 1×1 dimensions) that don't drift across exiftool versions.
+const PNG_1x1 = UInt8[137,80,78,71,13,10,26,10,0,0,0,13,73,72,68,82,0,0,0,1,0,
+    0,0,1,8,6,0,0,0,31,21,196,137,0,0,0,11,73,68,65,84,120,218,99,100,248,255,
+    191,30,0,5,132,2,127,194,91,30,42,0,0,0,0,73,69,78,68,174,66,96,130]
+
+"Build a Config whose data dirs all live under a fresh temp directory."
+function tmp_config(root; kwargs...)
+    cfg = Config(;
+        spool_dir   = joinpath(root, "spool"),
+        known_dir   = joinpath(root, "known"),
+        unknown_dir = joinpath(root, "unknown"),
+        binary_dir  = joinpath(root, "binary"),
+        text_dir    = joinpath(root, "text"),
+        done_dir    = joinpath(root, "done"),
+        failed_dir  = joinpath(root, "failed"),
+        kwargs...,
+    )
+    FileServer.ensure_dirs(cfg)
+    return cfg
+end
+
+@testset "FileServer" begin
+
+    @testset "sanitize_filename" begin
+        @test sanitize_filename("report.pdf") == "report.pdf"
+        # Directory components and traversal are stripped, not preserved.
+        @test sanitize_filename("../../etc/passwd") == "passwd"
+        @test sanitize_filename("/abs/path/x.txt") == "x.txt"
+        # Leading dots removed so "..", ".hidden" can't sneak through.
+        @test sanitize_filename("..") == "unnamed"
+        @test sanitize_filename(".hidden") == "hidden"
+        # Unsafe chars collapse to underscores; empty falls back to "unnamed".
+        @test sanitize_filename("a b&c*.d") == "a_b_c_.d"
+        @test sanitize_filename("") == "unnamed"
+        # Length is capped.
+        @test Base.length(sanitize_filename("a"^500)) == FileServer.MAX_NAME_LEN
+    end
+
+    @testset "normalize_metadata" begin
+        job = Job("id-1", "photo.jpg", "/data/known/id-1-photo.jpg", 4242, 0.0)
+        # Group-prefixed tags as exiftool -G emits them are already group-stripped
+        # by run_exiftool before reaching normalize_metadata, so keys are bare.
+        bytag = Dict{String,Any}(
+            "FileType"   => "JPEG",
+            "MIMEType"   => "image/jpeg",
+            "ImageWidth" => 800,
+            "ImageHeight"=> 600,
+            "Author"     => "Ada Lovelace",
+            "Creator"    => "Acrobat",        # feeds created_by, not author
+            "CreateDate" => "2020:01:02 03:04:05",
+            "ModifyDate" => "2020:01:02 03:04:06",
+            "PageCount"  => 12,
+        )
+        m = normalize_metadata(job, bytag)
+        @test m.file_type == "JPEG"
+        @test m.mime_type == "image/jpeg"
+        @test m.dimensions == (width = 800, height = 600)
+        @test m.author == "Ada Lovelace"
+        @test m.created_by == "Acrobat"
+        @test m.created_date == "2020:01:02 03:04:05"
+        @test m.page_count == 12
+        @test m.error === nothing
+        @test m.raw === bytag
+        # file_size is authoritative from the Job, never from exiftool.
+        @test m.file_size == 4242
+    end
+
+    @testset "normalize_metadata: missing tags degrade to nothing" begin
+        job = Job("id-2", "blob.bin", "/data/known/id-2-blob.bin", 7, 0.0)
+        m = normalize_metadata(job, Dict{String,Any}())
+        @test m.file_type === nothing
+        @test m.dimensions === nothing   # neither width nor height present
+        @test m.author === nothing
+        @test m.file_size == 7
+        @test m.error === nothing        # empty-but-present dict is still "success"
+    end
+
+    @testset "build_metadata: degraded on extraction failure" begin
+        mktempdir() do root
+            cfg = tmp_config(root; exiftool_timeout=5)
+            # Point at a nonexistent file → exiftool exits non-zero → degraded.
+            job = Job("id-3", "gone.dat", joinpath(cfg.known_dir, "id-3-gone.dat"), 99, 0.0)
+            m = build_metadata(job, cfg)
+            @test m.error !== nothing
+            @test m.file_type === nothing
+            @test m.raw === nothing
+            @test m.file_size == 99      # still authoritative from the Job
+            @test m.id == "id-3"
+        end
+    end
+
+    @testset "run_exiftool: real extraction on a PNG" begin
+        mktempdir() do root
+            p = joinpath(root, "pixel.png")
+            write(p, PNG_1x1)
+            bytag = run_exiftool(p, 30)
+            @test bytag !== nothing
+            @test bytag["FileType"] == "PNG"
+            @test bytag["ImageWidth"] == 1
+            @test bytag["ImageHeight"] == 1
+        end
+    end
+
+    @testset "finalize_known!: sidecar-first commit, end to end" begin
+        mktempdir() do root
+            cfg = tmp_config(root)
+            # A real known-stage file to enrich.
+            src = joinpath(cfg.known_dir, "id-9-pixel.png")
+            write(src, PNG_1x1)
+            job = Job("id-9", "pixel.png", src, Base.length(PNG_1x1), 0.0)
+
+            meta = build_metadata(job, cfg)
+            file_dest, sidecar = finalize_known!(cfg, job, meta)
+
+            # File moved into done/, original gone from known/.
+            @test isfile(file_dest)
+            @test dirname(file_dest) == cfg.done_dir
+            @test !isfile(src)
+
+            # Sidecar committed alongside it, valid JSON, no leftover .tmp.
+            @test isfile(sidecar)
+            @test endswith(sidecar, ".meta.json")
+            @test !isfile(string(sidecar, ".tmp"))
+            parsed = JSON3.read(read(sidecar, String))
+            @test parsed.file_type == "PNG"
+            @test parsed.file_size == Base.length(PNG_1x1)
+            @test parsed.error === nothing
+        end
+    end
+
+    @testset "is_binary: NUL-byte sniff" begin
+        mktempdir() do root
+            # Plain text → text.
+            txt = joinpath(root, "notes.txt")
+            write(txt, "hello, world\nsecond line\n")
+            @test is_binary(txt) == false
+
+            # A NUL byte anywhere in the sniff window → binary.
+            bin = joinpath(root, "blob.dat")
+            write(bin, UInt8[0x01, 0x02, 0x00, 0x03])
+            @test is_binary(bin) == true
+
+            # Empty file has no NUL → treated as text.
+            empty = joinpath(root, "empty")
+            write(empty, UInt8[])
+            @test is_binary(empty) == false
+
+            # A NUL past the sniff window is not seen → still text.
+            far = joinpath(root, "far.txt")
+            write(far, vcat(fill(UInt8('a'), FileServer.CONTENT_SNIFF_BYTES), UInt8[0x00]))
+            @test is_binary(far) == false
+        end
+    end
+
+    @testset "handle_unknown_job: routes to binary/ and text/" begin
+        mktempdir() do root
+            cfg = tmp_config(root)
+
+            # A binary file (embedded NUL) lands in binary/.
+            bpath = joinpath(cfg.unknown_dir, "id-b-blob.dat")
+            write(bpath, UInt8[0x00, 0xFF, 0x10])
+            bjob = Job("id-b", "blob.dat", bpath, filesize(bpath), 0.0)
+            handle_unknown_job(bjob, cfg, 1)
+            @test isfile(joinpath(cfg.binary_dir, "id-b-blob.dat"))
+            @test !isfile(bpath)
+
+            # A text file lands in text/.
+            tpath = joinpath(cfg.unknown_dir, "id-t-notes.log")
+            write(tpath, "just some log text\n")
+            tjob = Job("id-t", "notes.log", tpath, filesize(tpath), 0.0)
+            handle_unknown_job(tjob, cfg, 1)
+            @test isfile(joinpath(cfg.text_dir, "id-t-notes.log"))
+            @test !isfile(tpath)
+        end
+    end
+
+    @testset "recover_dir!: re-enqueues work, skips sidecars" begin
+        mktempdir() do root
+            dir = joinpath(root, "known"); mkpath(dir)
+            uuid = "0123456789abcdef0123456789abcdef0123"   # 36 chars
+            work = joinpath(dir, string(uuid, "-report.pdf"))
+            write(work, "x")
+            write(joinpath(dir, string(uuid, "-report.pdf.meta.json")), "{}")  # sidecar
+            write(joinpath(dir, "shortname"), "y")          # no uuid prefix
+
+            q = ChannelQueue(10)
+            n = recover_dir!(dir, q)
+            @test n == 2                       # the two real files, not the sidecar
+            @test length(q) == 2
+
+            jobs = [dequeue!(q), dequeue!(q)]   # sorted by filename on recovery
+            # "0123...-report.pdf" sorts before "shortname".
+            @test jobs[1].id == uuid
+            @test jobs[1].original_name == "report.pdf"
+            @test jobs[1].path == work
+            # File with no uuid prefix keeps its whole name; gets a minted id.
+            @test jobs[2].original_name == "shortname"
+            @test !isempty(jobs[2].id)
+        end
+    end
+
+end