intake: several configured intake channels (mcp, folder, command) all end in quarantine/ (was mcp-upload/), and upload collect pulls the synced-folder and command channels into it (#213); 10.0.0-beta.11

Files changed:
- .gitignore
- CHANGES.md
- VERSION
- docs/explanation/architecture.md
- docs/explanation/pipeline-rationale.md
- docs/how-to/README.md
- docs/how-to/back-up-to-a-synced-folder.md
- docs/how-to/collect-from-a-synced-folder.md
- docs/how-to/ingest-a-source.md
- docs/how-to/set-up-the-mcp-server.md
- docs/reference/commands/README.md
- docs/reference/commands/git.md
- docs/reference/commands/instance-health.md
- docs/reference/commands/raw-material-and-uploads.md
- docs/reference/configuration.md
- instructions/gates.md
- instructions/ingest-queue.md
- instructions/mcp-read-server.md
- instructions/migrations/10.0.0-settings-tiers.md
- instructions/wiki-ingest/SKILL.md
- raw/CONTRACT.md
- tools/CONTRACT.md
- tools/README.md
- tools/chemenu/cli_contract.py
- tools/chemenu/collect.py
- tools/chemenu/commands/backup_cmd.py
- tools/chemenu/commands/docs_verify.py
- tools/chemenu/commands/doctor.py
- tools/chemenu/commands/raw/pending.py
- tools/chemenu/commands/upload_cmd.py
- tools/chemenu/config.py
- tools/chemenu/config_shapes.py
- tools/chemenu/legacy_settings.py
- tools/chemenu/mcp/server.py
- tools/chemenu/reference_doc.py
- tools/chemenu/tests/test_backup.py
- tools/chemenu/tests/test_collect.py
- tools/chemenu/tests/test_config_shapes.py
- tools/chemenu/tests/test_doctor.py
- tools/chemenu/tests/test_legacy_settings.py
- tools/chemenu/tests/test_mcp_server.py
- tools/chemenu/tests/test_upload.py
- tools/chemenu/tests/test_upload_cmd.py
- tools/chemenu/upload.py

Co-Authored-By: Claude Code <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SnAJ7Z3CpVD3PRbN73QtU2

tools/CONTRACT.md

@@ -130,7 +130,8 @@ raw capture           write non-idempotent budget:counted              exit:0,1
 raw status            read  idempotent     budget:counted              exit:0      Report which captured bundles under `raw/` have fallen behind their repository, with each changed file as `A`/`M`/`D`.
 raw pending           read  idempotent     budget:counted              exit:0      List what waits in `incoming/`, oldest first, and name the entry an ingest without an argument takes next.
 raw accept            write non-idempotent budget:counted              exit:0,1    Promote one or more files, or one folder, from `incoming/` into `raw/`.
-upload list           read  idempotent     budget:counted              exit:0      List every MCP submission currently waiting in the quarantine (`mcp-upload/`).
+upload collect        write non-idempotent budget:counted              exit:0,1    Pull material from the configured `folder` and `command` intake channels into the quarantine - never past it.
+upload list           read  idempotent     budget:counted              exit:0      List every submission currently waiting in the quarantine (`quarantine/`), whatever channel it came through.
 upload show           read  idempotent     budget:counted              exit:0,1    Print one submission's manifest in full.
 upload accept         write non-idempotent budget:counted              exit:0,1,42 **Upload Review Gate:** promote a submission's file from quarantine into `incoming/`.
 upload reject         write non-idempotent budget:counted              exit:0,1    Delete a submission's material, keeping only its ledger trail.
@@ -1759,7 +1760,7 @@ List what waits in `incoming/`, oldest first, and name the entry an ingest witho
 
 **NOTES**
 
-- A candidate is a top-level entry of `incoming/`: a single `file`, a `bundle` of top-level files sharing a stem (a `raw fetch` pair), or a `folder` with every file below it. Dotfiles, empty directories and their contents are none; `mcp-upload/` is outside `incoming/` and never listed.
+- A candidate is a top-level entry of `incoming/`: a single `file`, a `bundle` of top-level files sharing a stem (a `raw fetch` pair), or a `folder` with every file below it. Dotfiles, empty directories and their contents are none; `quarantine/` is outside `incoming/` and never listed.
 - Order: oldest first by modification time. A bundle or folder counts as new as its newest file; a tie goes by name. The mtime is when a document last changed only if it was copied with its timestamps kept (`cp -p`, `rsync -a`, an unpacked archive) - for a download or a `raw fetch` it is merely when it was dropped.
 - Each candidate shows its path(s), kind, file count and mtime, and whether `raw accept` would take it as it stands - the same checks, minus `--fidelity`/`--authority`. One it would refuse is listed with the reason and skipped: it needs a human.
 - The default is the first candidate `raw accept` would take, and the output names it.
@@ -1865,9 +1866,61 @@ Promote one or more files, or one folder, from `incoming/` into `raw/`.
 - `wikitool types describe source` - the capture field values
 - `wikitool new source` - the source page for a promoted file
 
+#### `upload collect`
+
+Pull material from the configured `folder` and `command` intake channels into the quarantine - never past it.
+
+**SYNOPSIS**
+
+- `wikitool upload collect [<channel>...]`
+
+**PROPERTIES**
+
+- effect: write
+- idempotent: no
+- atomic: No - per file: one copy into `quarantine/<id>/` (manifest last), then one removal of the source; a channel's failure does not stop the others
+- budget: counted
+- network: no
+
+**EXAMPLES**
+
+- `tools/wikitool upload collect`
+- `tools/wikitool upload collect onedrive`
+
+**EXIT STATUS**
+
+- 0    success
+- 1    Unknown channel name, an `mcp` channel as argument, a malformed `intake` section, or an `mcp-upload/` that still holds submissions
+- 1    A channel or a file failed (folder missing, program exited non-zero, timed out or could not be started, a source that could not be removed)
+
+**ON FAILURE**
+
+- Unknown channel name, an `mcp` channel as argument, a malformed `intake` section, or an `mcp-upload/` that still holds submissions -> Fix the named cause and retry once
+- A channel or a file failed (folder missing, program exited non-zero, timed out or could not be started, a source that could not be removed) -> Not fixed by retrying unchanged - read the per-channel lines, fix the channel's folder or program, then run `collect <channel>` for that channel only
+
+**NEVER**
+
+- Never follow `collect` with `upload accept` and a `--confirm` token the user has not seen and approved.
+
+**NOTES**
+
+- Without arguments, collects from every channel of the `intake` section whose handler is `folder` or `command`; with names, from those channels only. A name no channel has, or the name of the `mcp` channel (which is pushed, not pulled), exits 1 before anything happens.
+- Everything a channel delivers lands in `quarantine/<id>/` unchanged, with a manifest that names `channel`, `handler` and `origin`; the submitter is `unknown`. Nothing reaches `incoming/` or `raw/` - that is `upload accept`, behind its gate, and an agent may run `collect` because collecting gives no trust.
+- `folder`: only the top level of `path`. A file is taken once its modification time and, where the platform has it, its creation time are older than `settle_seconds`, its extension and size fit, and its name passes the filename rules. It is removed from the folder only after its copy is fully written and the source still has the size and sha256 that were read; if it changed meanwhile, nothing is kept and it stays. A file whose sha256 already waits in the quarantine is reported as a duplicate and removed from the folder, because its content is kept. Subdirectories, hidden files (names starting with `.` or `~$`, or the hidden or system attribute on Windows), symbolic links, other extensions and oversized files are left untouched and reported.
+- `command`: the program of `commands.<name>` runs without a shell, with stdin closed, a 300 s timeout and an empty staging directory under `quarantine/.staging/` as its last argument. On exit 0 each regular file at the top of that directory with an allowed extension and size becomes a submission; on anything else nothing is taken and the end of stderr is printed. The staging directory is always removed.
+- Channels are independent: when one fails the others still run, and the exit status is 1 if any channel or file failed. With no pulled channel it prints `nothing to collect` and exits 0.
+- Exits 1, before it does anything, while an `mcp-upload/` of an earlier version still holds submissions, and when the `intake` section is malformed.
+- Not idempotent: a second run finds the folder emptied.
+
+**SEE ALSO**
+
+- `wikitool upload list` - what waits for review afterwards
+- `wikitool upload accept` - the gate out of the quarantine
+- `docs/how-to/collect-from-a-synced-folder.md`
+
 #### `upload list`
 
-List every MCP submission currently waiting in the quarantine (`mcp-upload/`).
+List every submission currently waiting in the quarantine (`quarantine/`), whatever channel it came through.
 
 **SYNOPSIS**
 
@@ -1892,8 +1945,9 @@ List every MCP submission currently waiting in the quarantine (`mcp-upload/`).
 
 **NOTES**
 
-- Lists every submission waiting in `mcp-upload/`, oldest id first: id, filename, size, submitter.
-- Only ever non-empty when the `upload` section of `.wikitool-local.json` opts the checkout into the MCP server's `submit` tool.
+- Lists every submission waiting in `quarantine/`, oldest id first: id, filename, size, channel, submitter.
+- Only ever non-empty when the `intake` section of `.wikitool-local.json` declares a channel.
+- Exits 1, before it does anything, while an `mcp-upload/` of an earlier version still holds submissions - the message names the command that moves it.
 - A submission directory with a corrupt manifest is silently skipped.
 - Never fails; read-only and safe to retry freely.
 
@@ -1933,7 +1987,8 @@ Print one submission's manifest in full.
 
 **NOTES**
 
-- Prints one submission's manifest in full: filename, size, sha256, submitter, submitter source (the header name, not a claim the header was honest), submission time.
+- Prints one submission's manifest in full: filename, size, sha256, submitter, submitter source (the header name or the handler, not a claim the header was honest), submission time, and where it came from: channel, handler, origin (the source file's path, `command:<name>` or `mcp:submit`).
+- A submitter of `unknown` means the channel has no sender to trust - read channel and origin instead.
 - What a reviewer reads before `upload accept`.
 - Read-only.
 
@@ -1967,13 +2022,13 @@ Print one submission's manifest in full.
 **EXIT STATUS**
 
 - 0    success
-- 1    Unknown or malformed submission id, or the submission's file is missing from `mcp-upload/<id>/`
+- 1    Unknown or malformed submission id, or the submission's file is missing from `quarantine/<id>/`
 - 1    `incoming/<filename>` already exists
 - 42   Upload Review Gate: `--confirm` is absent or does not match the manifest's current token
 
 **ON FAILURE**
 
-- Unknown or malformed submission id, or the submission's file is missing from `mcp-upload/<id>/` -> Fix the named argument and retry once
+- Unknown or malformed submission id, or the submission's file is missing from `quarantine/<id>/` -> Fix the named argument and retry once
 - `incoming/<filename>` already exists -> Not fixed by retrying unchanged - rename or clear it first
 - Upload Review Gate: `--confirm` is absent or does not match the manifest's current token -> Show the user the command's full output verbatim and stop. Once they have approved it, run the re-run line the output prints, which carries `--confirm <token>`. Without that token, or with a wrong, invented or superseded one, it exits 42 again with the current state
 
@@ -1983,7 +2038,7 @@ Print one submission's manifest in full.
 
 **NOTES**
 
-- Promotes a submission's file from `mcp-upload/<id>/` into `incoming/`, deletes the quarantine directory, and appends an `accepted` event to `mcp-upload/ledger.jsonl`.
+- Promotes a submission's file from `quarantine/<id>/` into `incoming/`, deletes the quarantine directory, and appends an `accepted` event to `quarantine/ledger.jsonl`.
 - Upload Review Gate: without a matching `--confirm`, exits 42 and prints the manifest in full plus the exact re-run line - one submission at a time. The gate check runs before anything is moved.
 - The token digests id, filename, size, sha256 and submitter, so an edited or superseded manifest invalidates it.
 - An occupied `incoming/<filename>` is an ordinary validation error, not the gate.
@@ -2271,7 +2326,7 @@ Mirror this machine's files that are not in git into `<target>/<machine>/`.
 **NOTES**
 
 - Mirrors `.wikitool-local.json`, `ENVIRONMENT.md` if present, every file in `incoming/` but `.gitkeep`, and the `extra` paths of the `backup` section into `<target>/<machine>/`, then writes the manifest `.chemenu-backup.json` there: time, stack version, machine name, and per file its relative path, size and sha256.
-- Never mirrors `mcp-upload/`, `reports/`, `.wikitool-tools.json`, `tools/.venv/`, the published skills or `.git/` - unless an `extra` path names one of the first ones. The secrets file lies outside the instance and is never read.
+- Never mirrors `quarantine/`, `reports/`, `.wikitool-tools.json`, `tools/.venv/`, the published skills or `.git/` - unless an `extra` path names one of the first ones. The secrets file lies outside the instance and is never read.
 - Writes and deletes only under `<target>/<machine>/`. Deletes only a file the previous manifest lists and the instance no longer has; a file there the manifest does not list stays as it is. A file whose size and sha256 match the previous manifest is not rewritten.
 - A symbolic link or junction inside a mirrored directory is skipped and named in the output, never followed.
 - `publish` runs the same step after every push when `machine_data` is true.
@@ -4018,7 +4073,7 @@ Check that this instance is correctly configured.
 - Legacy settings (`legacy-settings`): one `FAIL` for each per-feature settings file of an earlier version (telemetry, remotes, upload, task tracker) that is still there. Nothing reads them any more, so the settings they hold are not in effect; the fix is `tools/wikitool config migrate`.
 - Synced folder (`sync-folder`): a `WARN` when the git directory, links and junctions resolved, lies under a folder named by `OneDrive`, `OneDriveCommercial` or `OneDriveConsumer` (compared without regard to case on Windows) or under `~/Library/CloudStorage/` (macOS); `OK` otherwise, never a `FAIL`.
 - Backup (`backup`): the `backup` section absent is `OK`, malformed is a `FAIL`. Configured, a `WARN` when `<target>/git` is not a bare repository, `origin` does not push there, or - with `machine_data` true - no machine-data backup exists yet; otherwise `OK` with the age of the last machine-data backup.
-- MCP `submit` tool: whether the `upload` section of the settings is present, absent or malformed, its limits, and how many submissions wait in `mcp-upload/`. Absent is `OK` and means the write path does not exist at all; malformed is a `FAIL`.
+- Intake channels (`upload-intake`): the channels of the `intake` section with their handler, and how many submissions wait in `quarantine/`. Absent is `OK` and means no channel - no `submit` tool, nothing to collect; malformed is a `FAIL`, and so is an `mcp-upload/` of an earlier version that still holds submissions (the message names the command that moves it). A `folder` channel whose `path` does not exist is a `WARN`.
 - Task tracker: the `tasks` section present, absent or malformed - absent is `OK` (no tracker configured), malformed is a `FAIL`. When `WIKITOOL_TASKS_CONFIG` is set, the profile it names is read instead of the `tasks`, `superproductivity` and `caldav` sections and the finding names it; a set variable that names no file is a `FAIL`, never `OK`.
 - For a configured `superproductivity` provider, the configured `access` path's own state: `access: "api"` reports whether its local REST API answers `GET /health` with a ready renderer right now, `access: "snapshot"` whether a backup file is ready. The other access path is never attempted, and neither state is ever a `FAIL`.
 - For a configured `caldav` provider, whether the server is reachable and Basic auth succeeds - never a `FAIL`; only a broken config block is.

tools/README.md

@@ -99,7 +99,7 @@ tools/
     cli.py                 Typer app: registers every command, runs the budget gate, renders `-h`/`--help` from cli_contract
     cli_contract.py         one data record per command (name, synopsis, properties, exit status) - the source `-h`, the index, CONTRACT.md's generated region and the human reference under docs/reference/ render from
     config.py              repo layout: root resolution and every path under it
-    config_shapes.py       one record per hand-written .wikitool-*.json (telemetry, remotes, upload, tasks): what its loader accepts, its defaults and `Expected:` text, and what docs/reference/configuration.md renders; also the sections of the tiered settings, each field with its tier; stdlib-only, since the telemetry policy reads it
+    config_shapes.py       one record per hand-written .wikitool-*.json (telemetry, remotes, intake, tasks): what its loader accepts, its defaults and `Expected:` text, and what docs/reference/configuration.md renders; also the sections of the tiered settings, each field with its tier; stdlib-only, since the telemetry policy reads it
     settings.py            the tiered settings: wikitool.json (instance), .wikitool-local.json (checkout), secrets by reference - placement checked per field, references resolved on access through WIKITOOL_SECRET_<NAME>, the secrets file or `op read`
     api.py                 the in-process entry point - point Chemenu at a corpus and read it
     errors.py              ChemenuError / ValidationError / BackendError

tools/chemenu/cli_contract.py

@@ -288,7 +288,7 @@ GROUPS: tuple[tuple[str, tuple[str, ...]], ...] = (
     )),
     ("Raw material and uploads", (
         "raw fetch", "raw login", "raw derive", "raw capture", "raw status", "raw pending", "raw accept",
-        "upload list", "upload show", "upload accept", "upload reject",
+        "upload collect", "upload list", "upload show", "upload accept", "upload reject",
     )),
     ("Git", (
         "sync", "publish", "backup init", "backup run",

tools/chemenu/collect.py

@@ -0,0 +1,268 @@
+"""Pulling material into the quarantine from the channels that are not pushed
+(Gitea #213): a folder a sync client fills (`folder`), a program of this
+checkout (`command`).
+
+A handler is only a source. It hands over files unchanged; converting or
+deriving stays with `raw derive`, after a human has accepted the submission.
+Everything it takes ends in `chemenu.upload.store`, so nothing reaches
+`incoming/` or `raw/` any other way than `upload accept`.
+
+This module is deliberately not part of `chemenu.upload`: it reads outside the
+quarantine, deletes in the source folder and starts programs, none of which the
+MCP server process may do. Like `upload accept`, it is not importable from there.
+
+What a pull guarantees:
+
+- A source file is removed only after its copy in the quarantine is fully
+  written and the source still has the size and sha256 that were read.
+- Outside the quarantine it changes one thing: it removes files it took, directly
+  in the channel's `path`. It writes nothing there, deletes no directory and
+  follows no symbolic link.
+- Whatever it does not take stays where it is and is reported with its reason.
+
+No `typer`, no `rich`: the CLI renders a `ChannelResult`.
+"""
+from __future__ import annotations
+
+import hashlib
+import os
+import stat
+import subprocess
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Callable, Optional
+
+from chemenu import upload
+from chemenu.errors import ValidationError
+
+COMMAND_TIMEOUT_SECONDS = 300
+_STDERR_TAIL_CHARS = 2000
+_HIDDEN_PREFIXES = (".", "~$")
+_WINDOWS_HIDDEN = getattr(stat, "FILE_ATTRIBUTE_HIDDEN", 0) | getattr(stat, "FILE_ATTRIBUTE_SYSTEM", 0)
+
+
+@dataclass
+class ChannelResult:
+    """What one channel's pull did. `taken` and `duplicates` are
+    `(name, submission id)`, `skipped` is `(name, reason)`, `failed` is
+    `(name, reason)` for a file that should have been taken and was not.
+    `error` is set when the channel as a whole could not be pulled."""
+
+    channel: str
+    handler: str
+    taken: list = field(default_factory=list)
+    duplicates: list = field(default_factory=list)
+    skipped: list = field(default_factory=list)
+    failed: list = field(default_factory=list)
+    error: Optional[str] = None
+
+    @property
+    def ok(self) -> bool:
+        return self.error is None and not self.failed
+
+
+def _tail(stderr: bytes) -> str:
+    text = stderr.decode("utf-8", errors="replace").strip()
+    if len(text) > _STDERR_TAIL_CHARS:
+        text = "..." + text[-_STDERR_TAIL_CHARS:]
+    return text or "(nothing on stderr)"
+
+
+def _digest(path: Path) -> tuple[int, str]:
+    sha = hashlib.sha256()
+    size = 0
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1024 * 1024), b""):
+            sha.update(block)
+            size += len(block)
+    return size, sha.hexdigest()
+
+
+def source_matches(path: Path, size: int, digest: str) -> bool:
+    """Whether `path` still is the file that was read: a regular file of the
+    same size and sha256. The check that gates removing a source."""
+    try:
+        if path.is_symlink() or not path.is_file():
+            return False
+        return _digest(path) == (size, digest)
+    except OSError:
+        return False
+
+
+def _hidden(name: str, mode_attributes: int) -> bool:
+    return name.startswith(_HIDDEN_PREFIXES) or bool(mode_attributes & _WINDOWS_HIDDEN)
+
+
+def _moving(info: os.stat_result, settle_seconds: int, now: float) -> bool:
+    """Whether a file may still be arriving: its modification time or, where
+    the platform has it, its creation time is younger than `settle_seconds`."""
+    stamps = [info.st_mtime]
+    born = getattr(info, "st_birthtime", None)
+    if born is not None:
+        stamps.append(born)
+    return any(now - stamp < settle_seconds for stamp in stamps)
+
+
+def _take(
+    root: Path, channel: upload.Channel, result: ChannelResult, path: Path, name: str, origin: str,
+    *, after: Optional[Callable[[Path], None]] = None, guard_source: bool,
+) -> None:
+    """Read `path`, store it, and call `after(path)` once it is safe in the
+    quarantine. Reports into `result`; never raises for a problem with this
+    one file."""
+    try:
+        before = path.stat()
+        if before.st_size > channel.max_bytes:
+            result.skipped.append((name, f"larger than max_bytes ({channel.max_bytes})"))
+            return
+        data = path.read_bytes()
+    except OSError as exc:
+        result.failed.append((name, f"could not be read ({exc})"))
+        return
+    if len(data) != before.st_size:
+        result.skipped.append((name, "changed while it was read - left for the next run"))
+        return
+    if not data:
+        result.skipped.append((name, "empty"))
+        return
+    digest = hashlib.sha256(data).hexdigest()
+    waiting = upload.digest_to_pending_id(root, digest)
+    if waiting is not None:
+        if after is not None and not source_matches(path, len(data), digest):
+            result.skipped.append((name, "changed while it was read - left for the next run"))
+            return
+        result.duplicates.append((name, waiting))
+        _finish(result, path, name, after)
+        return
+    verify = (lambda: source_matches(path, len(data), digest)) if guard_source else None
+    try:
+        manifest = upload.store(
+            root, channel, filename=name, data=data,
+            submitter=upload.UNKNOWN_SUBMITTER, submitter_source=channel.handler,
+            origin=origin, verify=verify,
+        )
+    except upload.SourceChanged:
+        result.skipped.append((name, "changed while it was copied - nothing was kept, left for the next run"))
+        return
+    except (ValidationError, OSError) as exc:
+        result.failed.append((name, str(exc)))
+        return
+    result.taken.append((name, manifest["id"]))
+    _finish(result, path, name, after, manifest["id"])
+
+
+def _finish(result: ChannelResult, path: Path, name: str, after, submission_id: str = "") -> None:
+    if after is None:
+        return
+    try:
+        after(path)
+    except OSError as exc:
+        kept = f" (stored as {submission_id})" if submission_id else ""
+        result.failed.append((name, f"stays in the source, which could not be emptied{kept}: {exc}"))
+
+
+def collect_folder(root: "Path | str", channel: upload.Channel, now: Optional[float] = None) -> ChannelResult:
+    """Pull the top level of the channel's `path` into the quarantine."""
+    root = Path(root)
+    result = ChannelResult(channel.name, channel.handler)
+    folder = channel.path
+    if folder is None or not folder.is_dir():
+        result.error = f"the folder {folder} does not exist or is not a directory"
+        return result
+    clock = time.time() if now is None else now
+    try:
+        entries = sorted(os.scandir(folder), key=lambda entry: entry.name)
+    except OSError as exc:
+        result.error = f"the folder {folder} could not be read ({exc})"
+        return result
+    for entry in entries:
+        name = entry.name
+        path = folder / name
+        try:
+            info = entry.stat(follow_symlinks=False)
+        except OSError as exc:
+            result.failed.append((name, f"could not be inspected ({exc})"))
+            continue
+        if entry.is_symlink():
+            result.skipped.append((name, "symbolic link"))
+        elif stat.S_ISDIR(info.st_mode):
+            result.skipped.append((name, "subdirectory"))
+        elif _hidden(name, getattr(info, "st_file_attributes", 0)):
+            result.skipped.append((name, "hidden"))
+        elif not stat.S_ISREG(info.st_mode):
+            result.skipped.append((name, "not a regular file"))
+        elif Path(name).suffix.lower() not in channel.allowed_extensions:
+            result.skipped.append((name, "extension not allowed"))
+        elif info.st_size > channel.max_bytes:
+            result.skipped.append((name, f"larger than max_bytes ({channel.max_bytes})"))
+        elif _moving(info, channel.settle_seconds, clock):
+            result.skipped.append((name, f"still in motion (younger than {channel.settle_seconds} s)"))
+        else:
+            try:
+                upload.sanitize_filename(name)
+            except ValidationError as exc:
+                result.skipped.append((name, f"name refused ({exc})"))
+                continue
+            _take(root, channel, result, path, name, str(path), after=lambda p: p.unlink(), guard_source=True)
+    return result
+
+
+def collect_command(root: "Path | str", channel: upload.Channel) -> ChannelResult:
+    """Run the channel's program with an empty staging directory as its last
+    argument; take each regular file it left at the top of it."""
+    root = Path(root)
+    result = ChannelResult(channel.name, channel.handler)
+    origin = f"command:{channel.command_name}"
+    staging = upload.staging_dir(root)
+    try:
+        where = f"the command {channel.command_name!r}"
+        try:
+            ran = subprocess.run(
+                [*channel.command, str(staging)], capture_output=True, stdin=subprocess.DEVNULL,
+                timeout=COMMAND_TIMEOUT_SECONDS, check=False,
+            )
+        except FileNotFoundError:
+            result.error = f"{where} could not be started: {channel.command[0]} does not exist"
+            return result
+        except subprocess.TimeoutExpired:
+            result.error = f"{where} did not finish within {COMMAND_TIMEOUT_SECONDS} s"
+            return result
+        except OSError as exc:
+            result.error = f"{where} could not be started ({exc})"
+            return result
+        if ran.returncode != 0:
+            result.error = f"{where} ended with exit {ran.returncode}. Its stderr:\n{_tail(ran.stderr)}"
+            return result
+        for entry in sorted(os.scandir(staging), key=lambda e: e.name):
+            name = entry.name
+            path = staging / name
+            if entry.is_symlink():
+                result.skipped.append((name, "symbolic link"))
+            elif not entry.is_file(follow_symlinks=False):
+                result.skipped.append((name, "not a regular file"))
+            elif Path(name).suffix.lower() not in channel.allowed_extensions:
+                result.skipped.append((name, "extension not allowed"))
+            else:
+                try:
+                    upload.sanitize_filename(name)
+                except ValidationError as exc:
+                    result.skipped.append((name, f"name refused ({exc})"))
+                    continue
+                _take(root, channel, result, path, name, origin, guard_source=False)
+    finally:
+        upload.drop_staging(root, staging)
+    return result
+
+
+def collect(root: "Path | str", channel: upload.Channel) -> ChannelResult:
+    """One channel, whatever its handler. Never raises for a failure of the
+    channel: that is the result's `error`, so the others can still run."""
+    try:
+        if channel.handler == "folder":
+            return collect_folder(root, channel)
+        return collect_command(root, channel)
+    except (ValidationError, OSError) as exc:
+        failed = ChannelResult(channel.name, channel.handler)
+        failed.error = str(exc)
+        return failed

tools/chemenu/commands/backup_cmd.py

@@ -112,7 +112,7 @@ def init_command():
         "but `.gitkeep`, and the `extra` paths of the `backup` section into "
         "`<target>/<machine>/`, then writes the manifest `.chemenu-backup.json` there: time, "
         "stack version, machine name, and per file its relative path, size and sha256.",
-        "Never mirrors `mcp-upload/`, `reports/`, `.wikitool-tools.json`, `tools/.venv/`, the "
+        "Never mirrors `quarantine/`, `reports/`, `.wikitool-tools.json`, `tools/.venv/`, the "
         "published skills or `.git/` - unless an `extra` path names one of the first ones. The "
         "secrets file lies outside the instance and is never read.",
         "Writes and deletes only under `<target>/<machine>/`. Deletes only a file the previous "

tools/chemenu/commands/docs_verify.py

@@ -167,7 +167,7 @@ REQUIRED_IGNORE_CANARIES = (
     # The MCP `submit` tool's quarantine (Gitea #32) - stronger than
     # `incoming/` above: read by no command in the ordinary pipeline, not
     # only uncommitted. No type subdirectory either, for the same reason.
-    "mcp-upload/probe.pdf",
+    "quarantine/probe.pdf",
     # The live suite's tracker profiles (Gitea #156) - references to credentials, one file per tracker.
     ".wikitool-tasks.d/probe.json",
     # `raw capture`'s git cache (Gitea #177) - fetched repositories, recomputable,

tools/chemenu/commands/doctor.py

@@ -641,39 +641,48 @@ def _telemetry_file_deviations() -> list[str]:
 
 
 def check_upload_intake() -> Check:
-    """Whether the MCP `submit` tool is armed for this checkout, and how full
-    its quarantine is.
+    """Which intake channels this checkout has, how many submissions wait in
+    the quarantine, and whether the quarantine of earlier versions
+    (`mcp-upload/`) still holds any.
 
     Absent is the *safe* default here, unlike `check_publish_remotes`'s "any
-    push target passes" absence: no `upload` section means the write
-    path does not exist at all, not that it is unrestricted - so this never
-    `FAIL`s on a missing section. It does `FAIL` on one that parses to something
-    invalid, because a broken opt-in must not silently disable the very
-    limits it exists to enforce.
+    push target passes" absence: no `intake` section means no channel and no
+    `submit` tool, not that anything is unrestricted - so this never `FAIL`s on
+    a missing section. It does `FAIL` on one that parses to something invalid,
+    because a broken opt-in must not silently disable the very limits it exists
+    to enforce, and on an `mcp-upload/` with content, which no command reads
+    any more.
     """
     from chemenu import upload as upload_module
 
+    legacy = upload_module.legacy_dir_problem(config.ROOT)
+    if legacy:
+        return Check("upload-intake", "FAIL", legacy, "Move the directory as the message says")
     try:
-        cfg = upload_module.read_config(config.ROOT)
+        channels = upload_module.read_channels(config.ROOT)
     except ChemenuError as exc:
         return Check(
             "upload-intake", "FAIL", str(exc),
-            "Fix the upload section as the message says - a broken one is not treated as "
-            "'no limits'",
+            "Fix the intake section as the message says - a broken one is not treated as "
+            "'no channel'",
         )
-    if cfg is None:
+    if not channels:
         return Check(
             "upload-intake", "OK",
-            "submit tool not registered - no upload section in the settings",
+            "no channel - no intake section in the settings, so no submit tool and nothing to collect",
         )
     pending = upload_module.list_submissions(config.ROOT)
-    return Check(
-        "upload-intake", "OK",
-        f"submit tool armed (identity header {cfg.identity_header!r}, up to "
-        f"{cfg.max_bytes:,} byte(s), {cfg.submissions_per_day}/day and "
-        f"{cfg.bytes_per_day:,} byte(s)/day per submitter); "
-        f"{len(pending)} submission(s) waiting in {rel_path(config.UPLOAD_DIR)}",
-    )
+    described = ", ".join(f"{channel.name} ({channel.handler})" for channel in channels.values())
+    detail = f"channels: {described}; {len(pending)} submission(s) waiting in {rel_path(config.UPLOAD_DIR)}"
+    missing = [channel for channel in channels.values()
+               if channel.handler == "folder" and not (channel.path and channel.path.is_dir())]
+    if missing:
+        names = ", ".join(f"{channel.name} ({channel.path})" for channel in missing)
+        return Check(
+            "upload-intake", "WARN", f"{detail}; the folder does not exist: {names}",
+            "Create the folder or correct `path` of the channel in .wikitool-local.json",
+        )
+    return Check("upload-intake", "OK", detail)
 
 
 def check_tasks_provider() -> Check:
@@ -1086,9 +1095,11 @@ def run_doctor() -> list[Check]:
         "Configured, a `WARN` when `<target>/git` is not a bare repository, `origin` does not "
         "push there, or - with `machine_data` true - no machine-data backup exists yet; "
         "otherwise `OK` with the age of the last machine-data backup.",
-        "MCP `submit` tool: whether the `upload` section of the settings is present, absent or "
-        "malformed, its limits, and how many submissions wait in `mcp-upload/`. Absent is `OK` "
-        "and means the write path does not exist at all; malformed is a `FAIL`.",
+        "Intake channels (`upload-intake`): the channels of the `intake` section with their "
+        "handler, and how many submissions wait in `quarantine/`. Absent is `OK` and means no "
+        "channel - no `submit` tool, nothing to collect; malformed is a `FAIL`, and so is an "
+        "`mcp-upload/` of an earlier version that still holds submissions (the message names the "
+        "command that moves it). A `folder` channel whose `path` does not exist is a `WARN`.",
         "Task tracker: the `tasks` section present, absent or malformed - absent is `OK` "
         "(no tracker configured), malformed is a `FAIL`. When `WIKITOOL_TASKS_CONFIG` is set, "
         "the profile it names is read instead of the `tasks`, `superproductivity` and `caldav` "

tools/chemenu/commands/raw/pending.py

@@ -131,7 +131,7 @@ def _pending_candidates() -> list[_Candidate]:
         "A candidate is a top-level entry of `incoming/`: a single `file`, a `bundle` of "
         "top-level files sharing a stem (a `raw fetch` pair), or "
         "a `folder` with every file below it. Dotfiles, empty directories and their contents "
-        "are none; `mcp-upload/` is outside `incoming/` and never listed.",
+        "are none; `quarantine/` is outside `incoming/` and never listed.",
         "Order: oldest first by modification time. A bundle or folder counts as new as its "
         "newest file; a tie goes by name. The mtime is when a document last changed only if "
         "it was copied with its timestamps kept (`cp -p`, `rsync -a`, an unpacked archive) - "

tools/chemenu/commands/upload_cmd.py

@@ -1,12 +1,14 @@
-"""`wikitool upload list|show|accept|reject` - the human review side of the
-MCP submission quarantine (Gitea #32).
+"""`wikitool upload list|show|accept|reject|collect` - the quarantine every intake
+channel ends in (Gitea #32, #213): the pull from the channels that are not
+pushed, and the human review side.
 
 Everything that decides *whether* a submission was accepted or written at all
 lives in `chemenu.upload` - importable from the MCP server, stdlib only. What
 lives here instead is CLI-only by construction: the **Upload Review Gate**
 (`accept` refuses with Exit 42 until a human has seen the submission and
-re-runs with the printed `--confirm` token), and the two read commands a
-reviewer uses to look before clearing it. None of the four is importable
+re-runs with the printed `--confirm` token), the two read commands a
+reviewer uses to look before clearing it, and `collect`, which reads outside
+the quarantine and starts programs. None of the five is importable
 from `chemenu.mcp.server` - they sit under `chemenu.commands`, the same
 boundary every other write command is already kept out by (AGENTS.md
 invariant 6, `instructions/gates.md`).
@@ -18,15 +20,16 @@ from typing import Optional
 
 import typer
 
-from chemenu import cli_contract, config, upload
+from chemenu import cli_contract, collect as collect_module, config, upload
 from chemenu.commands._util import fail, needs_clearance, rel_path, success
 from chemenu.errors import ValidationError
 
-app = typer.Typer(help="Review, promote or reject MCP submissions waiting in mcp-upload/.")
+app = typer.Typer(help="Collect from the pulled intake channels; review, promote or reject what waits in quarantine/.")
 
 
 def _call(fn, *args, **kwargs):
     try:
+        upload.refuse_legacy_dir(config.ROOT)
         return fn(*args, **kwargs)
     except ValidationError as exc:
         fail(str(exc))
@@ -35,7 +38,7 @@ def _call(fn, *args, **kwargs):
 @app.command("list")
 @cli_contract.record(cli_contract.CommandRecord(
     path="upload list",
-    summary="List every MCP submission currently waiting in the quarantine (`mcp-upload/`).",
+    summary="List every submission currently waiting in the quarantine (`quarantine/`), whatever channel it came through.",
     synopsis=(cli_contract.Variant(usage="upload list [--json]"),),
     properties=cli_contract.Properties(
         effect=cli_contract.Effect.READ,
@@ -44,10 +47,11 @@ def _call(fn, *args, **kwargs):
         budget=cli_contract.Budget.COUNTED,
     ),
     notes=(
-        "Lists every submission waiting in `mcp-upload/`, oldest id first: id, filename, size, "
-        "submitter.",
-        "Only ever non-empty when the `upload` section of `.wikitool-local.json` opts the checkout into the MCP "
-        "server's `submit` tool.",
+        "Lists every submission waiting in `quarantine/`, oldest id first: id, filename, size, "
+        "channel, submitter.",
+        "Only ever non-empty when the `intake` section of `.wikitool-local.json` declares a channel.",
+        "Exits 1, before it does anything, while an `mcp-upload/` of an earlier version still holds "
+        "submissions - the message names the command that moves it.",
         "A submission directory with a corrupt manifest is silently skipped.",
         "Never fails; read-only and safe to retry freely.",
     ),
@@ -64,18 +68,18 @@ def _call(fn, *args, **kwargs):
 def upload_list_command(
     json_out: bool = typer.Option(False, "--json", help="Print every waiting submission as JSON"),
 ):
-    """List every submission currently waiting in mcp-upload/, oldest first."""
+    """List every submission currently waiting in quarantine/, oldest first."""
     manifests = _call(upload.list_submissions, config.ROOT)
     if json_out:
         typer.echo(json_module.dumps(manifests, indent=2))
         return
     if not manifests:
-        typer.echo("Nothing is waiting in mcp-upload/.")
+        typer.echo("Nothing is waiting in quarantine/.")
         return
     for manifest in manifests:
         typer.echo(
             f"{manifest['id']}  {manifest['filename']}  {manifest['size']}B  "
-            f"from {manifest['submitter']}"
+            f"via {manifest.get('channel', '?')}  from {manifest['submitter']}"
         )
 
 
@@ -92,8 +96,11 @@ def upload_list_command(
     ),
     notes=(
         "Prints one submission's manifest in full: filename, size, sha256, submitter, "
-        "submitter source (the header name, not a claim the header was honest), submission "
-        "time.",
+        "submitter source (the header name or the handler, not a claim the header was honest), "
+        "submission time, and where it came from: channel, handler, origin (the source file's "
+        "path, `command:<name>` or `mcp:submit`).",
+        "A submitter of `unknown` means the channel has no sender to trust - read channel and "
+        "origin instead.",
         "What a reviewer reads before `upload accept`.",
         "Read-only.",
     ),
@@ -119,7 +126,8 @@ def upload_show_command(
     if json_out:
         typer.echo(json_module.dumps(manifest, indent=2))
         return
-    for key in ("id", "filename", "size", "sha256", "submitter", "submitter_source", "submitted_at"):
+    for key in ("id", "filename", "size", "sha256", "submitter", "submitter_source", "submitted_at",
+                "channel", "handler", "origin"):
         typer.echo(f"{key}: {manifest.get(key)}")
 
 
@@ -134,6 +142,9 @@ def _clearance_message(manifest: dict, token: str, stale: Optional[str]) -> str:
         f"  submitter:        {manifest['submitter']}",
         f"  submitter_source: {manifest['submitter_source']}",
         f"  submitted_at:     {manifest['submitted_at']}",
+        f"  channel:          {manifest.get('channel')}",
+        f"  handler:          {manifest.get('handler')}",
+        f"  origin:           {manifest.get('origin')}",
         "",
     ]
     if stale:
@@ -165,8 +176,8 @@ def _clearance_message(manifest: dict, token: str, stale: Optional[str]) -> str:
         gates=("upload-review",),
     ),
     notes=(
-        "Promotes a submission's file from `mcp-upload/<id>/` into `incoming/`, deletes the "
-        "quarantine directory, and appends an `accepted` event to `mcp-upload/ledger.jsonl`.",
+        "Promotes a submission's file from `quarantine/<id>/` into `incoming/`, deletes the "
+        "quarantine directory, and appends an `accepted` event to `quarantine/ledger.jsonl`.",
         "Upload Review Gate: without a matching `--confirm`, exits 42 and prints the manifest "
         "in full plus the exact re-run line - one submission at a time. The gate check runs "
         "before anything is moved.",
@@ -177,7 +188,7 @@ def _clearance_message(manifest: dict, token: str, stale: Optional[str]) -> str:
     failures=(
         cli_contract.Failure(
             cause="Unknown or malformed submission id, or the submission's file is missing from "
-            "`mcp-upload/<id>/`",
+            "`quarantine/<id>/`",
             reaction="Fix the named argument and retry once",
         ),
         cli_contract.Failure(
@@ -261,3 +272,115 @@ def upload_reject_command(
     """Delete a submission's material, keeping its ledger trail."""
     _call(upload.reject, config.ROOT, submission_id, reason)
     success(f"Rejected '{submission_id}': {reason}")
+
+
+def _result_lines(result: collect_module.ChannelResult) -> list[str]:
+    head = f"{result.channel} ({result.handler}): "
+    if result.error:
+        return [head + "FAILED - " + result.error]
+    lines = [head + f"{len(result.taken)} taken, {len(result.duplicates)} duplicate, "
+             f"{len(result.skipped)} skipped, {len(result.failed)} failed"]
+    lines += [f"  taken      {name}  ->  {submission_id}" for name, submission_id in result.taken]
+    lines += [f"  duplicate  {name}  (already waiting as {submission_id})" for name, submission_id in result.duplicates]
+    lines += [f"  skipped    {name}  ({reason})" for name, reason in result.skipped]
+    lines += [f"  failed     {name}  ({reason})" for name, reason in result.failed]
+    return lines
+
+
+@app.command("collect")
+@cli_contract.record(cli_contract.CommandRecord(
+    path="upload collect",
+    summary="Pull material from the configured `folder` and `command` intake channels into the "
+    "quarantine - never past it.",
+    synopsis=(cli_contract.Variant(usage="upload collect [<channel>...]"),),
+    properties=cli_contract.Properties(
+        effect=cli_contract.Effect.WRITE,
+        idempotent=cli_contract.Idempotent.NO,
+        atomic="No - per file: one copy into `quarantine/<id>/` (manifest last), then one removal "
+        "of the source; a channel's failure does not stop the others",
+        budget=cli_contract.Budget.COUNTED,
+    ),
+    notes=(
+        "Without arguments, collects from every channel of the `intake` section whose handler is "
+        "`folder` or `command`; with names, from those channels only. A name no channel has, or "
+        "the name of the `mcp` channel (which is pushed, not pulled), exits 1 before anything happens.",
+        "Everything a channel delivers lands in `quarantine/<id>/` unchanged, with a manifest that "
+        "names `channel`, `handler` and `origin`; the submitter is `unknown`. Nothing reaches "
+        "`incoming/` or `raw/` - that is `upload accept`, behind its gate, and an agent may run "
+        "`collect` because collecting gives no trust.",
+        "`folder`: only the top level of `path`. A file is taken once its modification time and, "
+        "where the platform has it, its creation time are older than `settle_seconds`, its "
+        "extension and size fit, and its name passes the filename rules. It is removed from the "
+        "folder only after its copy is fully written and the source still has the size and "
+        "sha256 that were read; if it changed meanwhile, nothing is kept and it stays. A file whose "
+        "sha256 already waits in the quarantine is reported as a duplicate and removed from the "
+        "folder, because its content is kept. Subdirectories, hidden files (names starting with "
+        "`.` or `~$`, or the hidden or system attribute on Windows), symbolic links, other "
+        "extensions and oversized files are left untouched and reported.",
+        "`command`: the program of `commands.<name>` runs without a shell, with stdin closed, a "
+        "300 s timeout and an empty staging directory under `quarantine/.staging/` as its last "
+        "argument. On exit 0 each regular file at the top of that directory with an allowed "
+        "extension and size becomes a submission; on anything else nothing is taken and the end of "
+        "stderr is printed. The staging directory is always removed.",
+        "Channels are independent: when one fails the others still run, and the exit status is 1 "
+        "if any channel or file failed. With no pulled channel it prints `nothing to collect` "
+        "and exits 0.",
+        "Exits 1, before it does anything, while an `mcp-upload/` of an earlier version still holds "
+        "submissions, and when the `intake` section is malformed.",
+        "Not idempotent: a second run finds the folder emptied.",
+    ),
+    failures=(
+        cli_contract.Failure(
+            cause="Unknown channel name, an `mcp` channel as argument, a malformed `intake` section, "
+            "or an `mcp-upload/` that still holds submissions",
+            reaction="Fix the named cause and retry once",
+        ),
+        cli_contract.Failure(
+            cause="A channel or a file failed (folder missing, program exited non-zero, timed out or "
+            "could not be started, a source that could not be removed)",
+            reaction="Not fixed by retrying unchanged - read the per-channel lines, fix the channel's "
+            "folder or program, then run `collect <channel>` for that channel only",
+        ),
+    ),
+    examples=(
+        "tools/wikitool upload collect",
+        "tools/wikitool upload collect onedrive",
+    ),
+    never=(
+        "Never follow `collect` with `upload accept` and a `--confirm` token the user has not seen "
+        "and approved.",
+    ),
+    see_also=(
+        "`wikitool upload list` - what waits for review afterwards",
+        "`wikitool upload accept` - the gate out of the quarantine",
+        "`docs/how-to/collect-from-a-synced-folder.md`",
+    ),
+))
+def upload_collect_command(
+    names: Optional[list[str]] = typer.Argument(
+        None, help="Channels to collect from; all pulled channels when none is named"
+    ),
+):
+    """Pull from the folder and command channels into the quarantine."""
+    channels = _call(upload.read_channels, config.ROOT)
+    pulled = {name: channel for name, channel in channels.items() if channel.pulled}
+    chosen = list(names or [])
+    for name in chosen:
+        if name not in channels:
+            fail(f"No channel '{name}' in the `intake` section (channels: "
+                 f"{', '.join(channels) or 'none'}). Nothing was collected.")
+        if not channels[name].pulled:
+            fail(f"The channel '{name}' has the handler mcp: it is pushed through the `submit` tool, "
+                 "there is nothing to collect. Nothing was collected.")
+    selected = [pulled[name] for name in dict.fromkeys(chosen)] if chosen else list(pulled.values())
+    if not selected:
+        typer.echo("nothing to collect")
+        return
+    ok = True
+    for channel in selected:
+        result = collect_module.collect(config.ROOT, channel)
+        ok = ok and result.ok
+        for line in _result_lines(result):
+            typer.echo(line)
+    if not ok:
+        raise typer.Exit(code=1)

tools/chemenu/config.py

@@ -62,6 +62,9 @@ def _root() -> Path:
     return Path(assigned) if assigned is not None else resolve_root()
 
 
+# The quarantine's name; `UPLOAD_DIR` below is that directory under the root.
+UPLOAD_DIRNAME = "quarantine"
+
 # Everything under the root, as a name -> relative-path table rather than as
 # assignments. One place to read, and the only place that has to know a derived
 # path exists at all.
@@ -72,10 +75,10 @@ _DERIVED = {
     "REPORTS_DIR": ("reports",),
     "WORK_DIR": ("work",),
     "INSTRUCTIONS_DIR": ("instructions",),
-    # The MCP upload quarantine (Gitea #32) - never `raw/` and never `incoming/`,
-    # see raw/CONTRACT.md "Getting a file in". Gitignored; the server process is
-    # the only writer.
-    "UPLOAD_DIR": ("mcp-upload",),
+    # The quarantine every intake channel ends in (Gitea #32, #213) - never `raw/`
+    # and never `incoming/`, see raw/CONTRACT.md "Getting a file in". Gitignored
+    # twice over: the root `.gitignore` and a `.gitignore` of its own.
+    "UPLOAD_DIR": (UPLOAD_DIRNAME,),
     # Generated copies of the skill directories under `instructions/`. Both are
     # gitignored: they are build output, and a fresh clone publishes them with
     # `wikitool instructions sync` (see instructions/bootstrap.md).
@@ -83,6 +86,10 @@ _DERIVED = {
     "CLAUDE_SKILLS_DIR": (".claude", "skills"),
 }
 
+# What the quarantine was called before 10.0.0. Nothing reads or writes it any
+# more; a directory of this name with content is reported, never moved.
+LEGACY_UPLOAD_DIRNAME = "mcp-upload"
+
 # The generated files, derived from `KB_DIR` rather than from the root: a
 # caller that repoints only the corpus directory must not be left with a log
 # and a catalog belonging to a different tree.

tools/chemenu/config_shapes.py

@@ -136,6 +136,11 @@ class Shape:
     examples: tuple[dict, ...]
     absent: str = ""
     malformed: str = ""
+    # Judges the section as a whole once every field passed - for a rule that
+    # spans fields, such as a reference from one field to an entry of another.
+    # Gets the coerced values; returns a message that starts with the dotted
+    # place it is about (the section key is put in front), or None.
+    section_check: Optional[Callable[[dict], Optional[str]]] = None
 
     def field(self, name: str) -> Field:
         for candidate in self.fields:
@@ -278,6 +283,119 @@ def fetchers_problem(value: dict) -> Optional[str]:
     return None
 
 
+INTAKE_HANDLER_MCP = "mcp"
+INTAKE_HANDLER_FOLDER = "folder"
+INTAKE_HANDLER_COMMAND = "command"
+INTAKE_HANDLERS = (INTAKE_HANDLER_MCP, INTAKE_HANDLER_FOLDER, INTAKE_HANDLER_COMMAND)
+INTAKE_SETTLE_SECONDS = 60
+INTAKE_DEFAULT_IDENTITY_HEADER = "X-Forwarded-User"
+# The keys every channel has, then the ones only its handler has.
+_CHANNEL_COMMON = ("handler", "max_bytes", "allowed_extensions")
+_CHANNEL_KEYS = {
+    INTAKE_HANDLER_MCP: ("identity_header", "quota_submissions_per_day", "quota_bytes_per_day"),
+    INTAKE_HANDLER_FOLDER: ("path", "settle_seconds"),
+    INTAKE_HANDLER_COMMAND: ("command",),
+}
+_CHANNEL_REQUIRED = {
+    INTAKE_HANDLER_MCP: ("quota_submissions_per_day", "quota_bytes_per_day"),
+    INTAKE_HANDLER_FOLDER: ("path",),
+    INTAKE_HANDLER_COMMAND: ("command",),
+}
+
+
+def mcp_channel_keys() -> tuple[str, ...]:
+    """The keys a channel with the handler `mcp` takes."""
+    return _CHANNEL_COMMON[1:] + _CHANNEL_KEYS[INTAKE_HANDLER_MCP]
+
+
+def _is_int(value: Any) -> bool:
+    return isinstance(value, int) and not isinstance(value, bool)
+
+
+def _channel_problem(name: str, channel: Any) -> Optional[str]:
+    where = f"the channel {name!r}"
+    if not isinstance(channel, dict):
+        return f"{where} must be an object"
+    handler = channel.get("handler")
+    if handler not in INTAKE_HANDLERS:
+        return f"{where} needs `handler` to be one of {', '.join(INTAKE_HANDLERS)}; got {handler!r}"
+    own = _CHANNEL_KEYS[handler]
+    stray = sorted(set(channel) - set(_CHANNEL_COMMON) - set(own))
+    if stray:
+        return (f"{where} (handler {handler}) has {', '.join(f'`{key}`' for key in stray)}, which "
+                f"that handler does not take - its fields are {', '.join(_CHANNEL_COMMON + own)}")
+    for key in _CHANNEL_REQUIRED[handler] + ("max_bytes", "allowed_extensions"):
+        if key not in channel:
+            return f"{where} (handler {handler}) is missing `{key}`"
+    if not _is_int(channel["max_bytes"]) or channel["max_bytes"] <= 0:
+        return f"{where}: `max_bytes` must be a positive integer"
+    exts = channel["allowed_extensions"]
+    if (not isinstance(exts, list) or not exts
+            or not all(isinstance(ext, str) and ext.startswith(".") and len(ext) > 1 for ext in exts)):
+        return f"{where}: `allowed_extensions` must be a non-empty array of strings that each start with `.`"
+    if handler == INTAKE_HANDLER_MCP:
+        header = channel.get("identity_header", INTAKE_DEFAULT_IDENTITY_HEADER)
+        if not isinstance(header, str) or not header.strip():
+            return f"{where}: `identity_header` must be a non-empty string"
+        for key in ("quota_submissions_per_day", "quota_bytes_per_day"):
+            if not _is_int(channel[key]) or channel[key] <= 0:
+                return f"{where}: `{key}` must be a positive integer"
+    elif handler == INTAKE_HANDLER_FOLDER:
+        path = channel["path"]
+        if not isinstance(path, str) or not path.strip():
+            return f"{where}: `path` must be a non-empty string"
+        problem = absolute_path_problem(path)
+        if problem:
+            return f"{where}: `path` {problem}"
+        settle = channel.get("settle_seconds", INTAKE_SETTLE_SECONDS)
+        if not _is_int(settle) or settle < 0:
+            return f"{where}: `settle_seconds` must be an integer of 0 or more"
+    else:
+        ref = channel["command"]
+        if not isinstance(ref, str) or not _FETCHER_NAME.fullmatch(ref):
+            return (f"{where}: `command` names an entry of `commands` - lower-case letters, digits and "
+                    "hyphens")
+    return None
+
+
+def intake_channels_problem(value: dict) -> Optional[str]:
+    if not value:
+        return "must hold at least one channel"
+    mcp = []
+    for name, channel in value.items():
+        if not _FETCHER_NAME.fullmatch(name):
+            return (f"has the channel name {name!r} - a channel's name is lower-case letters, digits "
+                    "and hyphens")
+        problem = _channel_problem(name, channel)
+        if problem:
+            return problem
+        if channel["handler"] == INTAKE_HANDLER_MCP:
+            mcp.append(name)
+    if len(mcp) > 1:
+        return (f"has {len(mcp)} channels with handler mcp ({', '.join(mcp)}) - the MCP server has "
+                "one `submit` tool, so at most one channel takes that handler")
+    return None
+
+
+def intake_commands_problem(value: dict) -> Optional[str]:
+    for name, command in value.items():
+        if not _FETCHER_NAME.fullmatch(name):
+            return f"has the name {name!r} - a command's name is lower-case letters, digits and hyphens"
+        if (not isinstance(command, list) or not command
+                or not all(isinstance(part, str) and part for part in command)):
+            return f"{name} must be a non-empty array of non-empty strings: the program, then its arguments"
+    return None
+
+
+def intake_section_problem(values: dict) -> Optional[str]:
+    commands = values.get("commands") or {}
+    for name, channel in values["channels"].items():
+        if channel["handler"] == INTAKE_HANDLER_COMMAND and channel["command"] not in commands:
+            return (f"channels.{name}.command names {channel['command']!r}, which `commands` does not "
+                    "define")
+    return None
+
+
 def _positive(value: int) -> Optional[str]:
     return None if value > 0 else "must be positive"
 
@@ -359,39 +477,66 @@ REMOTES = Shape(
     "read as \"no restriction\".",
 )
 
-UPLOAD = Shape(
-    key="upload",
-    title="MCP submissions",
-    purpose="Turns on the MCP read server's `submit` tool and sets its limits.",
+INTAKE = Shape(
+    key="intake",
+    title="Intake channels",
+    purpose="Declares the channels material reaches the quarantine through: the MCP server's "
+    "`submit` tool, a folder a sync client fills, or a program of this checkout. Whatever a "
+    "channel delivers waits in `quarantine/` until a human accepts it.",
     fields=(
         Field(
-            "identity_header", "string", "The request header the authenticating middleware sets to "
-            "the submitter's identity on every request.",
-            default="X-Forwarded-User", json_type="string", coerce=str, empty_is_missing=True,
-            lenient="An empty value means the default.", tier=CHECKOUT, gate=True,
-        ),
-        positive_integer("max_bytes", "Largest single submission, in bytes.", required=True,
-                         tier=CHECKOUT, gate=True),
-        Field(
-            "allowed_extensions", "array of strings", "The file extensions `submit` accepts, each "
-            "starting with `.`. Compared case-insensitively.",
-            required=True, json_type="array", coerce=extensions, check=dotted_extensions,
-            lenient="Entries are read as strings and lower-cased; duplicates collapse.",
+            "channels", "object", "Channel name (lower-case letters, digits and hyphens) to "
+            "channel. The name goes into the manifest of what the channel delivers and is the "
+            "argument of `upload collect`. Every channel has `handler` (`mcp`, `folder` or "
+            "`command`), `max_bytes` (the largest single file) and `allowed_extensions` (each "
+            "starting with `.`, compared case-insensitively). Then per handler: `mcp` takes "
+            "`identity_header` (the request header the authenticating middleware sets to the "
+            f"submitter's identity, default `{INTAKE_DEFAULT_IDENTITY_HEADER}`) and, required, "
+            "`quota_submissions_per_day` and `quota_bytes_per_day` (per submitter, over any "
+            "rolling 24 hours); `folder` takes `path` (required: an absolute path outside the "
+            "instance, read without descending into subdirectories) and `settle_seconds` (default "
+            f"{INTAKE_SETTLE_SECONDS}: a file whose modification or creation time is younger is "
+            "left alone); `command` takes `command` (required: the name of an entry of "
+            "`commands`). At most one channel has the handler `mcp`. A field the handler does not "
+            "take is an error.",
+            required=True, json_type="object", coerce=object_value, check=intake_channels_problem,
             tier=CHECKOUT, gate=True,
         ),
-        positive_integer("quota_submissions_per_day", "Submissions one submitter may make over any "
-                         "rolling 24 hours.", required=True, tier=CHECKOUT, gate=True),
-        positive_integer("quota_bytes_per_day", "Bytes one submitter may submit over any rolling "
-                         "24 hours.", required=True, tier=CHECKOUT, gate=True),
+        Field(
+            "commands", "object", "Command name to command: an array of the program and its "
+            "arguments, run by the `command` handler without a shell, with stdin closed and a "
+            "timeout of 300 s, and a staging directory appended as the last argument. Exit 0 "
+            "turns each regular file at the top level of that directory into a submission; "
+            "anything else takes nothing over and prints the end of the program's stderr.",
+            default={}, json_type="object", coerce=object_value, check=intake_commands_problem,
+            tier=CHECKOUT, names_command=True,
+        ),
     ),
     examples=({
-        "identity_header": "X-Forwarded-User", "max_bytes": 10485760,
-        "allowed_extensions": [".md", ".pdf"],
-        "quota_submissions_per_day": 20, "quota_bytes_per_day": 52428800,
+        "channels": {
+            "mcp": {
+                "handler": "mcp", "identity_header": INTAKE_DEFAULT_IDENTITY_HEADER,
+                "max_bytes": 10485760, "allowed_extensions": [".md", ".pdf"],
+                "quota_submissions_per_day": 20, "quota_bytes_per_day": 52428800,
+            },
+            "onedrive": {
+                "handler": "folder",
+                "path": "C:\\Users\\<user>\\OneDrive - <company>\\Chemenu-Eingang",
+                "max_bytes": 52428800, "allowed_extensions": [".pdf", ".docx", ".msg", ".eml", ".md"],
+                "settle_seconds": 60,
+            },
+            "scanner": {
+                "handler": "command", "command": "scanner-pull",
+                "max_bytes": 52428800, "allowed_extensions": [".pdf"],
+            },
+        },
+        "commands": {"scanner-pull": ["pwsh", "-NoProfile", "-File", "C:\\tools\\scanner-pull.ps1"]},
     },),
-    absent="The `submit` tool is not registered: the write path does not exist.",
-    malformed="An error. The MCP server does not start and `doctor` reports `FAIL`. A broken section "
-    "is never read as \"no limits\".",
+    absent="No channel: the `submit` tool is not registered and `upload collect` has nothing to "
+    "collect. A valid state; `doctor` reports `OK`.",
+    malformed="An error. The MCP server does not start, `upload collect` stops with exit 1 and "
+    "`doctor` reports `FAIL`. A broken section is never read as \"no channel\".",
+    section_check=intake_section_problem,
 )
 
 TASKS = Shape(
@@ -672,7 +817,7 @@ BACKUP = Shape(
 # `.wikitool-local.json` (`chemenu.settings`), read at call time so a test can
 # stand in a section of its own.
 SETTINGS_SECTIONS: tuple[Shape, ...] = (
-    TELEMETRY, REMOTES, UPLOAD, TASKS, SUPERPRODUCTIVITY, CALDAV, SITES, FETCH, BACKUP,
+    TELEMETRY, REMOTES, INTAKE, TASKS, SUPERPRODUCTIVITY, CALDAV, SITES, FETCH, BACKUP,
 )
 
 # The sections `WIKITOOL_TASKS_CONFIG` replaces as a whole: a profile names a
@@ -747,6 +892,10 @@ def check(shape: Shape, data: Any) -> Checked:
             ))
         out.values[spec.name] = value
         out.given.add(spec.name)
+    if shape.section_check is not None and not out.problems:
+        problem = shape.section_check(out.values)
+        if problem:
+            out.problems.append(Problem("", problem))
     out.unknown = sorted(set(data) - {spec.name for spec in shape.fields})
     return out
 

tools/chemenu/legacy_settings.py

@@ -10,7 +10,7 @@ into `wikitool.json`, `.wikitool-local.json` and the secrets file -
 |---|---|
 | `.wikitool-telemetry.json` | section `telemetry` |
 | `.wikitool-remotes.json` | section `remotes` |
-| `.wikitool-upload.json` | section `upload` (`quota.*` is flat: `quota_*`) |
+| `.wikitool-upload.json` | section `intake`, channel `mcp` with `handler: "mcp"` (`quota.*` is flat: `quota_*`) |
 | `.wikitool-tasks.json` | sections `tasks` (`thresholds.*` is flat), `superproductivity`, `caldav` |
 
 What the migration promises, and how it keeps it:
@@ -49,7 +49,7 @@ TASKS_FILE = ".wikitool-tasks.json"
 FILES = {
     "telemetry": TELEMETRY_FILE,
     "remotes": REMOTES_FILE,
-    "upload": UPLOAD_FILE,
+    "intake": UPLOAD_FILE,
     "tasks": TASKS_FILE,
 }
 
@@ -199,7 +199,14 @@ def read_plan(root: "Path | str") -> Plan:
     if path.exists():
         plan.files.append(path)
         data = _flat(_read_old(path, lenient=False, plan=plan), "quota", "quota_", UPLOAD_FILE)
-        _take(plan, config_shapes.UPLOAD, data, file=UPLOAD_FILE)
+        if isinstance(data, dict):
+            known = set(config_shapes.mcp_channel_keys())
+            ignored = [key for key in data if key not in known and key != "schema"]
+            if ignored:
+                plan.notes.append(f"{UPLOAD_FILE}: {', '.join(ignored)} was not read before and is not migrated")
+            data = {key: value for key, value in data.items() if key in known}
+            data = {"handler": config_shapes.INTAKE_HANDLER_MCP, **data}
+        _take(plan, config_shapes.INTAKE, {"channels": {"mcp": data}}, file=UPLOAD_FILE)
 
     path = root / TASKS_FILE
     if path.exists():

tools/chemenu/mcp/server.py

@@ -1,5 +1,5 @@
 """The MCP server: `search`, `types`, `lint`, `status` over `kb/`, plus an
-opt-in write path, `submit`, into a quarantine (Gitea #32).
+opt-in write path, `submit`, into a quarantine (Gitea #32, #213).
 
 Chemenu's second consumer. The CLI and this are two adapters over one core -
 `chemenu.api.Corpus` - so a question answered here and the same question asked
@@ -15,12 +15,12 @@ test asserts it by importing this module in a clean interpreter and looking at
 `sys.modules`.
 
 **The sixth, `submit`, writes by a positive list instead of an absence.** It
-exists only when the `upload` section of `.wikitool-local.json` opts this
-checkout into it (absent by default) - not merely hidden, but
+exists only when the `intake` section of `.wikitool-local.json` declares a
+channel with the handler `mcp` (absent by default) - not merely hidden, but
 never registered on the server, the same distinction #19 draws for every
 other write function. When it is registered, every byte it writes still goes
 through exactly one choke point, `chemenu.upload._write_atomic_within`, which
-resolves the target and refuses anything outside `mcp-upload/`. That
+resolves the target and refuses anything outside `quarantine/`. That
 directory is read by no other command in the ordinary pipeline - promoting a
 submission out of it is `wikitool upload accept`, gated (Exit 42) and run by
 a human, never by this process. See `instructions/ingest-queue.md`.
@@ -128,18 +128,21 @@ def build_server(
     if check_trace:
         check_trace_destination(corpus.root)
 
-    # A malformed `upload` section is a start failure, not "no limits":
+    # A malformed `intake` section is a start failure, not "no limits":
     # this section decides whether an unauthenticated write path is offered at
     # all, so it is read once, up front, rather than lazily inside the tool
     # closure where a broken section would only surface on the first `submit`
     # call. `upload_cfg` is None exactly when the tool below is not defined.
+    # So is a quarantine of an earlier version that still holds submissions:
+    # starting beside it would leave them where no command looks.
+    upload.refuse_legacy_dir(corpus.root)
     upload_cfg = upload.read_config(corpus.root)
 
     read_only_note = (
         "Five tools are structurally read-only - there is no tool that writes among them."
         if upload_cfg is None
         else "Five tools are structurally read-only. The sixth, 'submit', writes only into "
-        "a quarantine ('mcp-upload/') no other tool or command reads - nothing here can "
+        "a quarantine ('quarantine/') no other tool or command reads - nothing here can "
         "reach kb/. A submission needs a human to promote it (see instructions/ingest-queue.md)."
     )
     server = MCPServer(

tools/chemenu/reference_doc.py

@@ -67,7 +67,7 @@ CONFIG_REGIONS = (
     ("reference-config-tasks", "TASKS"),
     ("reference-config-superproductivity", "SUPERPRODUCTIVITY"),
     ("reference-config-caldav", "CALDAV"),
-    ("reference-config-upload", "UPLOAD"),
+    ("reference-config-intake", "INTAKE"),
     ("reference-config-sites", "SITES"),
     ("reference-config-fetch", "FETCH"),
     ("reference-config-backup", "BACKUP"),

tools/chemenu/tests/test_backup.py

@@ -47,7 +47,7 @@ def instance(tmp_path, monkeypatch):
     (root / "incoming").mkdir()
     (root / "incoming" / ".gitkeep").write_text("", encoding="utf-8")
     (root / ".gitignore").write_text(
-        "/incoming/*\n!/incoming/.gitkeep\n/.wikitool-local.json\n/ENVIRONMENT.md\n/mcp-upload/\n"
+        "/incoming/*\n!/incoming/.gitkeep\n/.wikitool-local.json\n/ENVIRONMENT.md\n/quarantine/\n"
         "/reports/\n/.wikitool-tools.json\n", encoding="utf-8")
     (root / "README.md").write_text("init\n", encoding="utf-8")
     _git(root, "add", "-A")
@@ -222,7 +222,7 @@ def _fill(root: Path) -> None:
     (root / "incoming" / "paper.pdf").write_bytes(b"%PDF-1.7 fixture")
     (root / "incoming" / "old").mkdir()
     (root / "incoming" / "old" / "note.md").write_text("note\n", encoding="utf-8")
-    for relative in ("mcp-upload/sub-1/x.md", "reports/lint.md", ".wikitool-tools.json"):
+    for relative in ("quarantine/sub-1/x.md", "reports/lint.md", ".wikitool-tools.json"):
         (root / relative).parent.mkdir(parents=True, exist_ok=True)
         (root / relative).write_text("not mirrored\n", encoding="utf-8")
     (root / ".claude").mkdir()

tools/chemenu/tests/test_collect.py

@@ -0,0 +1,297 @@
+"""Tests for the pulled intake channels (`chemenu/collect.py`) and `wikitool
+upload collect`: the `folder` and `command` handlers, and what the command does
+across several channels."""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+import typer
+
+from chemenu import collect, config, upload
+from chemenu.commands.upload_cmd import upload_collect_command, upload_list_command
+
+OLD = time.time() - 3600
+
+
+@pytest.fixture
+def root(tmp_path, monkeypatch):
+    instance = tmp_path / "instance"
+    instance.mkdir()
+    (instance / "incoming").mkdir()
+    monkeypatch.setattr(config, "ROOT", instance)
+    return instance
+
+
+@pytest.fixture
+def drop(tmp_path):
+    folder = tmp_path / "drop"
+    folder.mkdir()
+    return folder
+
+
+def _put(folder: Path, name: str, data: bytes = b"content\n", age: float = 3600) -> Path:
+    path = folder / name
+    path.write_bytes(data)
+    stamp = time.time() - age
+    os.utime(path, (stamp, stamp))
+    return path
+
+
+def _folder_channel(drop: Path, **overrides) -> upload.Channel:
+    fields = dict(
+        name="onedrive", handler="folder", max_bytes=1024, allowed_extensions=(".md", ".pdf"),
+        path=drop, settle_seconds=60,
+    )
+    fields.update(overrides)
+    return upload.Channel(**fields)
+
+
+def _command_channel(argv: list[str], **overrides) -> upload.Channel:
+    fields = dict(
+        name="mail", handler="command", max_bytes=1024, allowed_extensions=(".md", ".pdf"),
+        command_name="scan", command=tuple(argv),
+    )
+    fields.update(overrides)
+    return upload.Channel(**fields)
+
+
+def _script(body: str) -> list[str]:
+    return [sys.executable, "-c", f"import sys, pathlib\nout = pathlib.Path(sys.argv[1])\n{body}"]
+
+
+def _write_intake(root: Path, body: dict) -> None:
+    (root / config.LOCAL_SETTINGS_FILENAME).write_text(
+        json.dumps({"schema": 1, "intake": body}), encoding="utf-8")
+
+
+def _tree(folder: Path) -> dict[str, bytes]:
+    return {str(p.relative_to(folder)): p.read_bytes() for p in sorted(folder.rglob("*")) if p.is_file()}
+
+
+def _manifests(root: Path) -> list[dict]:
+    return upload.list_submissions(root)
+
+
+# --- folder -----------------------------------------------------------------
+
+def test_a_folder_pull_stores_each_file_and_empties_the_source(root, drop):
+    _put(drop, "a.md", b"alpha\n")
+    _put(drop, "b.pdf", b"bravo\n")
+    result = collect.collect(root, _folder_channel(drop))
+    assert result.ok and len(result.taken) == 2
+    assert _tree(drop) == {}
+    manifests = {m["filename"]: m for m in _manifests(root)}
+    assert set(manifests) == {"a.md", "b.pdf"}
+    first = manifests["a.md"]
+    assert (first["channel"], first["handler"]) == ("onedrive", "folder")
+    assert first["origin"] == str(drop / "a.md")
+    assert (first["submitter"], first["submitter_source"]) == ("unknown", "folder")
+    assert (root / "quarantine" / first["id"] / "a.md").read_bytes() == b"alpha\n"
+
+
+def test_a_folder_pull_leaves_what_it_does_not_take_untouched(root, drop):
+    (drop / "sub").mkdir()
+    (drop / "sub" / "inner.md").write_bytes(b"inner")
+    _put(drop, ".x.pdf", b"hidden")
+    _put(drop, "~$a.docx", b"lock")
+    _put(drop, "notes.exe", b"bad extension")
+    _put(drop, "big.md", b"x" * 2000)
+    target = _put(drop, "real.md", b"target")
+    link = drop / "link.md"
+    link.symlink_to(target)
+    target.unlink()
+    before = _tree(drop)
+    result = collect.collect(root, _folder_channel(drop))
+    assert result.ok and result.taken == [] and _manifests(root) == []
+    assert _tree(drop) == before
+    assert link.is_symlink()
+    reasons = dict(result.skipped)
+    assert reasons["sub"] == "subdirectory"
+    assert reasons[".x.pdf"] == "hidden" and reasons["~$a.docx"] == "hidden"
+    assert reasons["notes.exe"] == "extension not allowed"
+    assert reasons["link.md"] == "symbolic link"
+    assert "max_bytes" in reasons["big.md"]
+
+
+def test_a_file_younger_than_settle_seconds_waits_for_the_next_run(root, drop):
+    _put(drop, "fresh.md", b"arriving", age=5)
+    result = collect.collect(root, _folder_channel(drop, settle_seconds=60))
+    assert result.taken == [] and "still in motion" in dict(result.skipped)["fresh.md"]
+    assert (drop / "fresh.md").read_bytes() == b"arriving"
+    os.utime(drop / "fresh.md", (OLD, OLD))
+    later = collect.collect(root, _folder_channel(drop, settle_seconds=60))
+    assert [name for name, _ in later.taken] == ["fresh.md"]
+
+
+def test_a_source_that_changes_during_the_copy_is_kept_and_nothing_is_stored(root, drop, monkeypatch):
+    source = _put(drop, "a.md", b"one\n")
+    monkeypatch.setattr(collect, "source_matches", lambda *args: False)
+    result = collect.collect(root, _folder_channel(drop))
+    assert result.taken == [] and "changed while it was copied" in dict(result.skipped)["a.md"]
+    assert source.read_bytes() == b"one\n"
+    assert _manifests(root) == []
+    assert [p for p in (root / "quarantine").iterdir() if p.is_dir()] == []
+
+
+def test_a_duplicate_is_reported_with_the_existing_id_and_the_source_is_removed(root, drop):
+    _put(drop, "a.md", b"same\n")
+    first = collect.collect(root, _folder_channel(drop))
+    existing = first.taken[0][1]
+    _put(drop, "again.md", b"same\n")
+    second = collect.collect(root, _folder_channel(drop))
+    assert second.taken == [] and second.duplicates == [("again.md", existing)]
+    assert not (drop / "again.md").exists()
+    assert len(_manifests(root)) == 1
+
+
+def test_a_missing_folder_fails_the_channel(root, tmp_path):
+    result = collect.collect(root, _folder_channel(tmp_path / "gone"))
+    assert not result.ok and "does not exist" in result.error
+
+
+# --- command ----------------------------------------------------------------
+
+def test_a_command_hands_over_the_files_it_left_in_the_staging_directory(root):
+    channel = _command_channel(_script("(out / 'mail.md').write_bytes(b'from a script\\n')"))
+    result = collect.collect(root, channel)
+    assert result.ok and [name for name, _ in result.taken] == ["mail.md"]
+    manifest = _manifests(root)[0]
+    assert (manifest["channel"], manifest["handler"], manifest["origin"]) == ("mail", "command", "command:scan")
+    assert (manifest["submitter"], manifest["submitter_source"]) == ("unknown", "command")
+    assert not (root / "quarantine" / ".staging").exists()
+
+
+def test_a_failing_command_takes_nothing_and_shows_its_stderr_tail(root):
+    channel = _command_channel(_script(
+        "(out / 'half.md').write_bytes(b'x')\nsys.stderr.write('boom ' * 600 + 'THE END')\nsys.exit(3)"))
+    result = collect.collect(root, channel)
+    assert not result.ok and result.taken == []
+    assert "exit 3" in result.error and result.error.rstrip().endswith("THE END")
+    assert len(result.error) < 2500
+    assert _manifests(root) == []
+    assert not (root / "quarantine" / ".staging").exists()
+
+
+def test_a_command_that_times_out_fails_its_channel_and_cleans_up(root, monkeypatch):
+    monkeypatch.setattr(collect, "COMMAND_TIMEOUT_SECONDS", 1)
+    channel = _command_channel(_script("(out / 'a.md').write_bytes(b'x')\nimport time\ntime.sleep(30)"))
+    result = collect.collect(root, channel)
+    assert not result.ok and "did not finish" in result.error
+    assert _manifests(root) == []
+    assert not (root / "quarantine" / ".staging").exists()
+
+
+def test_a_command_that_does_not_exist_fails_its_channel(root, tmp_path):
+    result = collect.collect(root, _command_channel([str(tmp_path / "nope.sh")]))
+    assert not result.ok and "does not exist" in result.error
+
+
+def test_a_command_result_skips_what_is_not_a_plain_allowed_file(root):
+    channel = _command_channel(_script(
+        "(out / 'sub').mkdir()\n(out / 'x.exe').write_bytes(b'x')\n(out / 'ok.md').write_bytes(b'ok')"))
+    result = collect.collect(root, channel)
+    assert [name for name, _ in result.taken] == ["ok.md"]
+    assert dict(result.skipped) == {"sub": "not a regular file", "x.exe": "extension not allowed"}
+
+
+# --- upload collect ---------------------------------------------------------
+
+def _two_channels(root, drop):
+    _write_intake(root, {
+        "commands": {"scan": _script("(out / 'mail.md').write_bytes(b'mail\\n')")},
+        "channels": {
+            "onedrive": {"handler": "folder", "path": str(drop), "max_bytes": 1024,
+                         "allowed_extensions": [".md"]},
+            "mail": {"handler": "command", "command": "scan", "max_bytes": 1024,
+                     "allowed_extensions": [".md"]},
+            "server": {"handler": "mcp", "max_bytes": 1024, "allowed_extensions": [".md"],
+                       "quota_submissions_per_day": 1, "quota_bytes_per_day": 1000},
+        },
+    })
+
+
+def test_collect_without_a_pull_channel_has_nothing_to_do(root, capsys):
+    upload_collect_command(names=None)
+    assert "nothing to collect" in capsys.readouterr().out
+
+
+def test_collect_pulls_every_pull_channel_and_reports_per_channel(root, drop, capsys):
+    _two_channels(root, drop)
+    _put(drop, "a.md", b"alpha\n")
+    upload_collect_command(names=None)
+    out = capsys.readouterr().out
+    assert "onedrive (folder)" in out and "mail (command)" in out
+    assert {m["channel"] for m in _manifests(root)} == {"onedrive", "mail"}
+
+
+def test_collect_names_one_channel_and_leaves_the_others(root, drop):
+    _two_channels(root, drop)
+    _put(drop, "a.md", b"alpha\n")
+    upload_collect_command(names=["mail"])
+    assert {m["channel"] for m in _manifests(root)} == {"mail"}
+    assert (drop / "a.md").exists()
+
+
+def test_a_failing_channel_does_not_stop_the_others_but_the_exit_is_1(root, drop, tmp_path, capsys):
+    _two_channels(root, drop)
+    body = json.loads((root / config.LOCAL_SETTINGS_FILENAME).read_text())
+    body["intake"]["channels"]["onedrive"]["path"] = str(tmp_path / "gone")
+    (root / config.LOCAL_SETTINGS_FILENAME).write_text(json.dumps(body), encoding="utf-8")
+    with pytest.raises(typer.Exit) as excinfo:
+        upload_collect_command(names=None)
+    assert excinfo.value.exit_code == 1
+    assert "FAILED" in capsys.readouterr().out
+    assert {m["channel"] for m in _manifests(root)} == {"mail"}
+
+
+@pytest.mark.parametrize("names", [["nope"], ["server"], ["mail", "nope"], ["mail", "server"]])
+def test_an_unknown_or_mcp_argument_exits_1_before_any_effect(root, drop, names):
+    _two_channels(root, drop)
+    _put(drop, "a.md", b"alpha\n")
+    before = _tree(drop)
+    with pytest.raises(typer.Exit) as excinfo:
+        upload_collect_command(names=names)
+    assert excinfo.value.exit_code == 1
+    assert _tree(drop) == before
+    assert _manifests(root) == []
+
+
+def test_a_malformed_intake_section_exits_1(root):
+    _write_intake(root, {"channels": {"a": {"handler": "folder"}}})
+    with pytest.raises(typer.Exit) as excinfo:
+        upload_collect_command(names=None)
+    assert excinfo.value.exit_code == 1
+
+
+def test_the_command_refuses_while_a_legacy_quarantine_holds_submissions(root, drop):
+    _two_channels(root, drop)
+    _put(drop, "a.md", b"alpha\n")
+    (root / "mcp-upload" / "sub-1").mkdir(parents=True)
+    (root / "mcp-upload" / "sub-1" / "a.md").write_bytes(b"waiting")
+    with pytest.raises(typer.Exit) as excinfo:
+        upload_collect_command(names=None)
+    assert excinfo.value.exit_code == 1
+    assert (drop / "a.md").exists()
+    with pytest.raises(typer.Exit):
+        upload_list_command(json_out=False)
+
+
+# --- the quarantine's own .gitignore ----------------------------------------
+
+def _git(root: Path, *args: str) -> str:
+    return subprocess.run(["git", *args], cwd=root, check=True, capture_output=True, text=True).stdout
+
+
+def test_collected_material_shows_in_no_git_status_even_without_a_root_entry(root, drop):
+    _git(root, "init", "-q")
+    _put(drop, "a.md", b"alpha\n")
+    collect.collect(root, _folder_channel(drop))
+    collect.collect(root, _command_channel(_script("(out / 'm.md').write_bytes(b'mail')")))
+    assert (root / "quarantine" / ".gitignore").read_text().strip() == "*"
+    assert _git(root, "status", "--porcelain", "--untracked-files=all") == ""

tools/chemenu/tests/test_config_shapes.py

@@ -29,10 +29,10 @@ runner = CliRunner()
 
 PACKAGE = Path(config_shapes.__file__).parent
 
-UPLOAD_OK = {
-    "identity_header": "X-User", "max_bytes": 1000, "allowed_extensions": [".md", ".PDF"],
+INTAKE_OK = {"channels": {"mcp": {
+    "handler": "mcp", "identity_header": "X-User", "max_bytes": 1000, "allowed_extensions": [".md", ".PDF"],
     "quota_submissions_per_day": 5, "quota_bytes_per_day": 5000,
-}
+}}}
 TASKS_OK = {"provider": "caldav", "stalled_waiting_days": 14, "unpaged_project_weeks": 3,
             "someday_stale_months": 5}
 CALDAV_OK = {"url": "https://dav.example/", "username": "u", "app_password": "pw",
@@ -78,7 +78,7 @@ def test_the_telemetry_policy_reads_the_record():
 
 def test_one_record_per_section():
     assert [shape.key for shape in config_shapes.SETTINGS_SECTIONS] == [
-        "telemetry", "remotes", "upload", "tasks", "superproductivity", "caldav", "sites", "fetch",
+        "telemetry", "remotes", "intake", "tasks", "superproductivity", "caldav", "sites", "fetch",
         "backup",
     ]
 
@@ -126,24 +126,30 @@ def _swap(monkeypatch, attr, shape):
     )
 
 
-def test_upload_follows_a_field_made_required_or_optional(root, monkeypatch):
+def test_intake_follows_a_field_made_required(root, monkeypatch):
     from chemenu import upload
 
-    original = config_shapes.UPLOAD
-    write_settings(root, monkeypatch, upload=UPLOAD_OK)
+    original = config_shapes.INTAKE
+    write_settings(root, monkeypatch, intake=INTAKE_OK)
     assert upload.read_config(root).max_bytes == 1000
-    _swap(monkeypatch, "UPLOAD", _add_required(original))
+    _swap(monkeypatch, "INTAKE", _add_required(original))
     with pytest.raises(ValidationError, match="added is missing"):
         upload.read_config(root)
 
-    _swap(monkeypatch, "UPLOAD", original)
-    (root / config.LOCAL_SETTINGS_FILENAME).unlink()
-    (root / config.INSTANCE_SETTINGS_FILENAME).unlink()
-    write_settings(root, monkeypatch, upload={k: v for k, v in UPLOAD_OK.items() if k != "max_bytes"})
-    with pytest.raises(ValidationError):
+
+def test_intake_rules_across_fields_are_checked_by_the_record(root, monkeypatch):
+    from chemenu import upload
+
+    write_settings(root, monkeypatch, intake={"channels": {"mcp": {"handler": "mcp", "max_bytes": 1}}})
+    with pytest.raises(ValidationError, match="quota_submissions_per_day"):
         upload.read_config(root)
-    _swap(monkeypatch, "UPLOAD", _not_required(original, "max_bytes", 99))
-    assert upload.read_config(root).max_bytes == 99
+
+
+@pytest.mark.parametrize("key", ["intake.channels", "intake.commands"])
+def test_the_intake_fields_are_not_settable_by_config_set(root, key):
+    result = runner.invoke(app, ["config", "set", key, "{}"])
+    assert result.exit_code == 42, result.output
+    assert not (root / config.LOCAL_SETTINGS_FILENAME).exists()
 
 
 def test_tasks_follows_a_field_made_required(root, monkeypatch):

tools/chemenu/tests/test_doctor.py

@@ -350,27 +350,52 @@ def test_publish_remotes_warns_on_several_remotes_without_an_allowlist(instance)
 def test_upload_intake_ok_when_no_section(instance):
     checks = doctor.run_doctor()
     assert _status(checks, "upload-intake") == "OK"
-    assert "not registered" in _detail(checks, "upload-intake")
+    assert "no channel" in _detail(checks, "upload-intake")
 
 
-def test_upload_intake_ok_and_armed_when_section_is_valid(instance, monkeypatch):
-    write_settings(config.ROOT, monkeypatch, upload={
-        "max_bytes": 100, "allowed_extensions": [".md"],
-        "quota_submissions_per_day": 1, "quota_bytes_per_day": 100,
-    })
+_MCP = {"handler": "mcp", "max_bytes": 100, "allowed_extensions": [".md"],
+        "quota_submissions_per_day": 1, "quota_bytes_per_day": 100}
+
+
+def test_upload_intake_names_each_channel_with_its_handler(instance, monkeypatch, tmp_path_factory):
+    folder = tmp_path_factory.mktemp("synced")
+    write_settings(config.ROOT, monkeypatch, intake={"channels": {
+        "mcp": _MCP,
+        "onedrive": {"handler": "folder", "path": str(folder), "max_bytes": 100, "allowed_extensions": [".md"]},
+    }})
     checks = doctor.run_doctor()
     assert _status(checks, "upload-intake") == "OK"
     detail = _detail(checks, "upload-intake")
-    assert "armed" in detail
+    assert "mcp (mcp)" in detail and "onedrive (folder)" in detail
     assert "0 submission(s) waiting" in detail
 
 
+def test_upload_intake_warns_when_a_folder_does_not_exist(instance, monkeypatch, tmp_path_factory):
+    write_settings(config.ROOT, monkeypatch, intake={"channels": {
+        "onedrive": {"handler": "folder", "path": str(tmp_path_factory.mktemp("outside") / "gone"), "max_bytes": 100,
+                     "allowed_extensions": [".md"]},
+    }})
+    checks = doctor.run_doctor()
+    assert _status(checks, "upload-intake") == "WARN"
+    assert "onedrive" in _detail(checks, "upload-intake")
+
+
 def test_upload_intake_fails_on_a_malformed_section(instance, monkeypatch):
-    write_settings(config.ROOT, monkeypatch, upload={"max_bytes": "many"})
+    write_settings(config.ROOT, monkeypatch, intake={"channels": {"mcp": {"handler": "mcp", "max_bytes": "many"}}})
     checks = doctor.run_doctor()
     assert _status(checks, "upload-intake") == "FAIL"
 
 
+def test_upload_intake_fails_on_a_leftover_mcp_upload_directory(instance):
+    waiting = config.ROOT / "mcp-upload" / "sub-1"
+    waiting.mkdir(parents=True)
+    (waiting / "a.md").write_bytes(b"waiting")
+    checks = doctor.run_doctor()
+    assert _status(checks, "upload-intake") == "FAIL"
+    assert "mv mcp-upload quarantine" in _detail(checks, "upload-intake")
+    assert (waiting / "a.md").read_bytes() == b"waiting"
+
+
 def test_tasks_provider_ok_when_no_section(instance):
     checks = doctor.run_doctor()
     assert _status(checks, "tasks-provider") == "OK"

tools/chemenu/tests/test_legacy_settings.py

@@ -30,6 +30,9 @@ APP_PASSWORD = "sekrit-app-password-0815"
 THRESHOLDS = {"stalled_waiting_days": 14, "unpaged_project_weeks": 3, "someday_stale_months": 5}
 
 # name -> (old files by file name, what is in effect afterwards)
+_MCP_CHANNEL = {"name": "mcp", "handler": "mcp", "path": None, "settle_seconds": 60,
+                "command_name": "", "command": ()}
+
 CASES = {
     "telemetry-full": (
         {".wikitool-telemetry.json": {"schema": 1, "enabled": False, "max_session_bytes": 1000,
@@ -54,7 +57,7 @@ CASES = {
             "schema": 1, "identity_header": "X-Forwarded-User", "max_bytes": 2048,
             "allowed_extensions": [".md", ".txt"],
             "quota": {"submissions_per_day": 4, "bytes_per_day": 9000}}},
-        {"upload": {"identity_header": "X-Forwarded-User", "max_bytes": 2048,
+        {"upload": {**_MCP_CHANNEL, "identity_header": "X-Forwarded-User", "max_bytes": 2048,
                     "allowed_extensions": (".md", ".txt"), "submissions_per_day": 4,
                     "bytes_per_day": 9000}},
     ),
@@ -185,7 +188,8 @@ def _expected(case: dict) -> dict:
     for key, value in case.items():
         if key == "upload" and isinstance(value, str):
             value = {
-                "identity_header": _default("upload", "identity_header"),
+                **_MCP_CHANNEL,
+                "identity_header": upload.DEFAULT_IDENTITY_HEADER,
                 "max_bytes": 100, "allowed_extensions": (".md",),
                 "submissions_per_day": 1, "bytes_per_day": 5,
             }
@@ -210,6 +214,17 @@ def test_every_shape_keeps_its_effective_value(checkout, case):
     assert _normal(_effective(checkout)) == _normal(_expected(expected))
 
 
+def test_the_old_upload_file_becomes_the_mcp_channel_of_the_intake_section(checkout):
+    _write_old(checkout, CASES["upload-full"][0])
+    legacy_settings.migrate(checkout)
+    data = json.loads((checkout / config.LOCAL_SETTINGS_FILENAME).read_text(encoding="utf-8"))
+    assert "upload" not in data
+    assert data["intake"] == {"channels": {"mcp": {
+        "handler": "mcp", "identity_header": "X-Forwarded-User", "max_bytes": 2048,
+        "allowed_extensions": [".md", ".txt"],
+        "quota_submissions_per_day": 4, "quota_bytes_per_day": 9000}}}
+
+
 # --- b: only the tools file remains; no secret in the tree or the index -------
 
 
@@ -376,7 +391,7 @@ def test_the_telemetry_hook_neither_reads_nor_refuses_over_an_old_file(checkout)
 
 
 @pytest.mark.parametrize("key,value", [
-    ("upload.max_bytes", "100"),
+    ("intake.channels", '{"mcp": {"handler": "mcp"}}'),
     ("remotes.allowed_push_urls", '["ssh://git@example.net/a.git"]'),
 ])
 def test_config_set_on_a_gated_section_exits_42_and_writes_nothing(checkout, key, value):

tools/chemenu/tests/test_mcp_server.py

@@ -9,8 +9,8 @@ The acceptance criteria that are not about a value coming back:
   importable from the server - structural, not filtered;
 - every response carries the commit it was computed from;
 - telemetry cannot land inside the served checkout;
-- the sixth tool, `submit`, exists only when the `upload` section of `.wikitool-local.json` opts a
-  checkout in, writes only under `mcp-upload/`, and never resolves an
+- the sixth tool, `submit`, exists only when the `intake` section of `.wikitool-local.json` opts a
+  checkout in, writes only under `quarantine/`, and never resolves an
   identity from anywhere but the request's own headers.
 """
 from __future__ import annotations
@@ -44,13 +44,14 @@ from chemenu.mcp.server import (  # noqa: E402  - after the skip guard
 
 _UPLOAD_CFG = {
     "schema": 1,
-    "upload": {
+    "intake": {"channels": {"mcp": {
+        "handler": "mcp",
         "identity_header": "X-Forwarded-User",
         "max_bytes": 1024,
         "allowed_extensions": [".md"],
         "quota_submissions_per_day": 10,
         "quota_bytes_per_day": 100_000,
-    },
+    }}},
 }
 
 
@@ -368,13 +369,38 @@ def test_submit_is_present_once_armed(corpus):
     assert names == {"search", "types", "describe_type", "lint", "status", "submit"}
 
 
+def test_submit_is_absent_when_the_section_has_only_pulled_channels(corpus, tmp_path_factory):
+    folder = tmp_path_factory.mktemp("synced")
+    section = {"schema": 1, "intake": {"channels": {"onedrive": {
+        "handler": "folder", "path": str(folder), "max_bytes": 1024, "allowed_extensions": [".md"]}}}}
+    (corpus / ".wikitool-local.json").write_text(json.dumps(section), encoding="utf-8")
+    names = {tool.name for tool in asyncio.run(build_server(corpus, check_trace=False).list_tools())}
+    assert "submit" not in names
+
+
+def test_submit_carries_the_limits_of_the_mcp_channel(corpus):
+    _arm_upload(corpus)
+    tools = asyncio.run(build_server(corpus, check_trace=False).list_tools())
+    submit = next(tool for tool in tools if tool.name == "submit")
+    assert "1024 bytes" in submit.description and ".md" in submit.description
+
+
+def test_a_leftover_mcp_upload_directory_refuses_to_build(corpus):
+    _arm_upload(corpus)
+    (corpus / "mcp-upload" / "sub-1").mkdir(parents=True)
+    (corpus / "mcp-upload" / "sub-1" / "a.md").write_bytes(b"waiting")
+    with pytest.raises(ValidationError, match="mcp-upload"):
+        build_server(corpus, check_trace=False)
+    assert (corpus / "mcp-upload" / "sub-1" / "a.md").read_bytes() == b"waiting"
+
+
 def test_a_malformed_upload_config_refuses_to_build(corpus):
     (corpus / ".wikitool-local.json").write_text("{not json", encoding="utf-8")
     with pytest.raises(ValidationError):
         build_server(corpus, check_trace=False)
 
 
-def test_submit_writes_only_into_mcp_upload_incoming_stays_untouched(corpus):
+def test_submit_writes_only_into_the_quarantine_incoming_stays_untouched(corpus):
     _arm_upload(corpus)
     server = build_server(corpus, check_trace=False)
     before = _tree(corpus)
@@ -391,15 +417,16 @@ def test_submit_writes_only_into_mcp_upload_incoming_stays_untouched(corpus):
 
     assert manifest["submitter"] == "alice"
     assert manifest["submitter_source"] == "X-Forwarded-User"
+    assert (manifest["channel"], manifest["handler"], manifest["origin"]) == ("mcp", "mcp", "mcp:submit")
 
     after = _tree(corpus)
     changed = {p for p in after if after.get(p) != before.get(p)}
     assert changed
-    assert all(p.startswith("mcp-upload/") for p in changed)
+    assert all(p.startswith("quarantine/") for p in changed)
     assert list((corpus / "incoming").iterdir()) == []
 
     # git itself is untouched - the fixture's own committed tree (kb/, types/)
-    # is unchanged, and HEAD did not move. `mcp-upload/` and the config file
+    # is unchanged, and HEAD did not move. `quarantine/` and the config file
     # are new, untracked paths (this fixture carries no .gitignore of its
     # own, unlike the real repo), which `git status` reports and is exactly
     # what a real checkout's `.gitignore` would hide - not a write to git.

tools/chemenu/tests/test_upload.py

@@ -20,8 +20,10 @@ CONTENT = b"hello world, this is a test submission\n"
 CONTENT_B64 = base64.b64encode(CONTENT).decode("ascii")
 
 
-def _cfg(**overrides) -> upload.UploadConfig:
+def _cfg(**overrides) -> upload.Channel:
     defaults = dict(
+        name="mcp",
+        handler="mcp",
         identity_header="X-Forwarded-User",
         max_bytes=1024,
         allowed_extensions=(".md", ".pdf"),
@@ -29,7 +31,7 @@ def _cfg(**overrides) -> upload.UploadConfig:
         bytes_per_day=10_000,
     )
     defaults.update(overrides)
-    return upload.UploadConfig(**defaults)
+    return upload.Channel(**defaults)
 
 
 def _tree(root: Path) -> dict[str, bytes]:
@@ -53,15 +55,21 @@ def test_sanitize_filename_accepts_a_bare_name():
     assert upload.sanitize_filename("report.md") == "report.md"
 
 
-# --- read_config -------------------------------------------------------------
+# --- read_config / read_channels -----------------------------------------------
 
 def test_read_config_absent_is_none(tmp_path):
     assert upload.read_config(tmp_path) is None
+    assert upload.read_channels(tmp_path) == {}
 
 
 def _section(tmp_path, **fields):
+    channel = dict({"handler": "mcp"}, **fields)
+    _intake(tmp_path, {"channels": {"mcp": channel}})
+
+
+def _intake(tmp_path, body):
     (tmp_path / config.LOCAL_SETTINGS_FILENAME).write_text(
-        json.dumps({"schema": 1, "upload": fields}), encoding="utf-8")
+        json.dumps({"schema": 1, "intake": body}), encoding="utf-8")
 
 
 GOOD = {"max_bytes": 100, "allowed_extensions": [".md"],
@@ -110,6 +118,7 @@ def test_read_config_valid_file_parses(tmp_path):
     _section(tmp_path, identity_header="X-User", max_bytes=2048, allowed_extensions=[".MD", ".pdf"],
              quota_submissions_per_day=3, quota_bytes_per_day=999)
     cfg = upload.read_config(tmp_path)
+    assert cfg.name == "mcp" and cfg.handler == "mcp"
     assert cfg.identity_header == "X-User"
     assert cfg.max_bytes == 2048
     assert cfg.allowed_extensions == (".md", ".pdf")
@@ -128,13 +137,70 @@ def test_read_config_refuses_a_file_of_an_earlier_version(tmp_path):
         upload.read_config(tmp_path)
 
 
+def test_two_mcp_channels_are_an_error(tmp_path):
+    channel = dict({"handler": "mcp"}, **GOOD)
+    _intake(tmp_path, {"channels": {"a": channel, "b": channel}})
+    with pytest.raises(ValidationError, match="mcp"):
+        upload.read_channels(tmp_path)
+
+
+def test_a_field_that_does_not_match_the_handler_is_an_error(tmp_path, tmp_path_factory):
+    outside = tmp_path_factory.mktemp("outside")
+    channel = {"handler": "folder", "path": str(outside), "max_bytes": 100, "allowed_extensions": [".md"],
+               "identity_header": "X-User"}
+    _intake(tmp_path, {"channels": {"f": channel}})
+    with pytest.raises(ValidationError, match="identity_header"):
+        upload.read_channels(tmp_path)
+
+
+def test_a_command_channel_needs_a_commands_entry(tmp_path):
+    channel = {"handler": "command", "command": "scanner", "max_bytes": 100, "allowed_extensions": [".md"]}
+    _intake(tmp_path, {"channels": {"s": channel}})
+    with pytest.raises(ValidationError, match="scanner"):
+        upload.read_channels(tmp_path)
+
+
+def test_a_folder_path_must_be_absolute_and_outside_the_instance(tmp_path, tmp_path_factory):
+    base = {"handler": "folder", "max_bytes": 100, "allowed_extensions": [".md"]}
+    _intake(tmp_path, {"channels": {"f": dict(base, path="relative/dir")}})
+    with pytest.raises(ValidationError, match="absolute"):
+        upload.read_channels(tmp_path)
+    inside = tmp_path / "drop"
+    _intake(tmp_path, {"channels": {"f": dict(base, path=str(inside))}})
+    with pytest.raises(ValidationError, match="inside the instance"):
+        upload.read_channels(tmp_path)
+    outside = tmp_path_factory.mktemp("outside")
+    _intake(tmp_path, {"channels": {"f": dict(base, path=str(outside))}})
+    channel = upload.read_channels(tmp_path)["f"]
+    assert channel.handler == "folder" and channel.path == outside
+    assert channel.settle_seconds == 60
+    assert upload.read_config(tmp_path) is None
+
+
+def test_a_leftover_mcp_upload_directory_is_refused_and_untouched(tmp_path):
+    legacy = tmp_path / "mcp-upload" / "sub-1"
+    legacy.mkdir(parents=True)
+    (legacy / "a.md").write_bytes(b"waiting")
+    before = _tree(tmp_path)
+    problem = upload.legacy_dir_problem(tmp_path)
+    assert "mv mcp-upload quarantine" in problem and "Move-Item" in problem
+    with pytest.raises(ValidationError, match="mcp-upload"):
+        upload.refuse_legacy_dir(tmp_path)
+    assert _tree(tmp_path) == before
+
+
+def test_an_empty_mcp_upload_directory_is_no_problem(tmp_path):
+    (tmp_path / "mcp-upload").mkdir()
+    assert upload.legacy_dir_problem(tmp_path) is None
+
+
 # --- the write choke point ----------------------------------------------------
 
 def test_write_primitive_refuses_a_relative_escape(tmp_path):
     with pytest.raises(ValidationError):
         upload._write_atomic_within(tmp_path, Path("../outside.txt"), b"x")
     assert not (tmp_path / "outside.txt").exists()
-    assert not (tmp_path / "mcp-upload").exists()
+    assert not (tmp_path / "quarantine").exists()
 
 
 def test_write_primitive_refuses_an_absolute_path(tmp_path):
@@ -146,7 +212,7 @@ def test_write_primitive_refuses_an_absolute_path(tmp_path):
 def test_write_primitive_refuses_a_symlink_escape(tmp_path):
     outside = tmp_path / "outside"
     outside.mkdir()
-    upload_dir = tmp_path / "mcp-upload"
+    upload_dir = tmp_path / "quarantine"
     upload_dir.mkdir()
     (upload_dir / "escape").symlink_to(outside, target_is_directory=True)
 
@@ -173,12 +239,12 @@ def test_submit_happy_path_writes_manifest_and_content(tmp_path):
     assert manifest["submitter_source"] == "X-Forwarded-User"
     assert manifest["size"] == len(CONTENT)
 
-    submission_dir = tmp_path / "mcp-upload" / manifest["id"]
+    submission_dir = tmp_path / "quarantine" / manifest["id"]
     assert (submission_dir / "report.md").read_bytes() == CONTENT
     stored_manifest = json.loads((submission_dir / "manifest.json").read_text())
     assert stored_manifest == manifest
 
-    ledger_lines = (tmp_path / "mcp-upload" / "ledger.jsonl").read_text().splitlines()
+    ledger_lines = (tmp_path / "quarantine" / "ledger.jsonl").read_text().splitlines()
     assert len(ledger_lines) == 1
     event = json.loads(ledger_lines[0])
     assert event["event"] == "submitted"
@@ -342,10 +408,10 @@ def test_promote_moves_file_deletes_dir_and_ledgers_accepted(tmp_path):
 
     assert dest == tmp_path / "incoming" / "a.md"
     assert dest.read_bytes() == CONTENT
-    assert not (tmp_path / "mcp-upload" / manifest["id"]).exists()
+    assert not (tmp_path / "quarantine" / manifest["id"]).exists()
 
     events = [json.loads(line) for line in
-              (tmp_path / "mcp-upload" / "ledger.jsonl").read_text().splitlines()]
+              (tmp_path / "quarantine" / "ledger.jsonl").read_text().splitlines()]
     assert [e["event"] for e in events] == ["submitted", "accepted"]
 
     # nothing else in the tree changed
@@ -363,7 +429,7 @@ def test_promote_refuses_when_the_file_is_missing_from_disk(tmp_path):
     defense in depth, since nothing in this module deletes a content file
     without its manifest."""
     manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
-    (tmp_path / "mcp-upload" / manifest["id"] / "a.md").unlink()
+    (tmp_path / "quarantine" / manifest["id"] / "a.md").unlink()
 
     with pytest.raises(ValidationError):
         upload.promote(tmp_path, manifest["id"])
@@ -378,7 +444,7 @@ def test_promote_refuses_when_incoming_already_has_the_name(tmp_path):
         upload.promote(tmp_path, manifest["id"])
 
     # nothing moved: quarantine still holds it, incoming/ untouched
-    assert (tmp_path / "mcp-upload" / manifest["id"] / "a.md").read_bytes() == CONTENT
+    assert (tmp_path / "quarantine" / manifest["id"] / "a.md").read_bytes() == CONTENT
     assert (tmp_path / "incoming" / "a.md").read_text() == "already here"
 
 
@@ -388,9 +454,9 @@ def test_reject_deletes_material_and_ledgers_with_reason(tmp_path):
     manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
     upload.reject(tmp_path, manifest["id"], "license unclear")
 
-    assert not (tmp_path / "mcp-upload" / manifest["id"]).exists()
+    assert not (tmp_path / "quarantine" / manifest["id"]).exists()
     events = [json.loads(line) for line in
-              (tmp_path / "mcp-upload" / "ledger.jsonl").read_text().splitlines()]
+              (tmp_path / "quarantine" / "ledger.jsonl").read_text().splitlines()]
     assert events[-1]["event"] == "rejected"
     assert events[-1]["reason"] == "license unclear"
     assert events[-1]["sha256"] == manifest["sha256"]
@@ -401,7 +467,7 @@ def test_reject_requires_a_reason(tmp_path):
     with pytest.raises(ValidationError):
         upload.reject(tmp_path, manifest["id"], "")
     # nothing deleted
-    assert (tmp_path / "mcp-upload" / manifest["id"]).exists()
+    assert (tmp_path / "quarantine" / manifest["id"]).exists()
 
 
 def test_reject_unknown_id_is_rejected(tmp_path):

tools/chemenu/tests/test_upload_cmd.py

@@ -20,7 +20,9 @@ from chemenu.commands.upload_cmd import (
 CONTENT = b"a test submission for the CLI layer\n"
 CONTENT_B64 = base64.b64encode(CONTENT).decode("ascii")
 
-_CFG = upload.UploadConfig(
+_CFG = upload.Channel(
+    name="mcp",
+    handler="mcp",
     identity_header="X-Forwarded-User",
     max_bytes=1024,
     allowed_extensions=(".md",),
@@ -50,7 +52,7 @@ def test_accept_without_a_token_needs_clearance(root, capsys):
     assert manifest["sha256"] in out
     assert "--confirm" in out
     # nothing moved
-    assert (root / "mcp-upload" / manifest["id"]).exists()
+    assert (root / "quarantine" / manifest["id"]).exists()
     assert not (root / "incoming" / manifest["filename"]).exists()
 
 
@@ -59,7 +61,7 @@ def test_accept_with_a_stale_token_needs_clearance_again(root):
     with pytest.raises(typer.Exit) as excinfo:
         upload_accept_command(submission_id=manifest["id"], confirm="not-the-real-token")
     assert excinfo.value.exit_code == EXIT_NEEDS_CLEARANCE
-    assert (root / "mcp-upload" / manifest["id"]).exists()
+    assert (root / "quarantine" / manifest["id"]).exists()
 
 
 def test_accept_with_the_right_token_promotes(root):
@@ -67,13 +69,13 @@ def test_accept_with_the_right_token_promotes(root):
     token = upload.confirm_token(manifest)
     upload_accept_command(submission_id=manifest["id"], confirm=token)
     assert (root / "incoming" / manifest["filename"]).read_bytes() == CONTENT
-    assert not (root / "mcp-upload" / manifest["id"]).exists()
+    assert not (root / "quarantine" / manifest["id"]).exists()
 
 
 def test_reject_deletes_and_needs_no_gate(root):
     manifest = _seed(root)
     upload_reject_command(submission_id=manifest["id"], reason="license unclear")
-    assert not (root / "mcp-upload" / manifest["id"]).exists()
+    assert not (root / "quarantine" / manifest["id"]).exists()
 
 
 def test_reject_without_a_reason_is_a_validation_error(root):
@@ -81,7 +83,7 @@ def test_reject_without_a_reason_is_a_validation_error(root):
     with pytest.raises(typer.Exit) as excinfo:
         upload_reject_command(submission_id=manifest["id"], reason="")
     assert excinfo.value.exit_code == 1
-    assert (root / "mcp-upload" / manifest["id"]).exists()
+    assert (root / "quarantine" / manifest["id"]).exists()
 
 
 def test_list_json_matches_pending_manifests(root, capsys):

tools/chemenu/upload.py

@@ -1,5 +1,5 @@
-"""The MCP write path's one choke point: `mcp-upload/`, and nothing else
-(Gitea #32).
+"""The quarantine every intake channel ends in: `quarantine/`, and nothing else
+(Gitea #32, #213).
 
 `#19`'s MCP read server has an absence property - nothing under
 `chemenu.commands` is importable from it, so a write function does not exist
@@ -8,29 +8,34 @@ being true by itself: something in the server process now writes. What
 replaces it is not an absence but a **positive list**, enforced in code
 rather than promised in prose:
 
-    The server process may write into exactly one directory - `mcp-upload/`
+    The server process may write into exactly one directory - `quarantine/`
     under the served root - and every write in this module resolves its
     target and refuses anything that lands outside it.
 
 `_write_atomic_within()` is that one choke point: every file this module
 writes (a submitted file, its manifest) goes through it. The ledger append is
 the one exception, and it is a narrower case of the same rule rather than a
-gap in it - its destination is a hardcoded constant (`mcp-upload/ledger.jsonl`),
+gap in it - its destination is a hardcoded constant (`quarantine/ledger.jsonl`),
 never a caller-supplied name, so there is no path to sanitise in the first
 place.
 
-Two stages, two different grants of trust:
+Several channels (the `intake` section) end in the same place. A channel has
+a handler: `mcp` is the `submit` tool, which pushes; `folder` and `command`
+are pulled by `wikitool upload collect` (`chemenu.collect`), which reads a
+folder or runs a program and hands each file to `store()` here. Nothing a
+channel delivers reaches `incoming/` or `raw/` any other way:
 
-    mcp-upload/<id>/   material nobody has looked at - `submit` writes here
+    quarantine/<id>/   material nobody has looked at - every channel writes here
        |  wikitool upload accept <id>     <- a human decides (Exit 42 gate)
     incoming/          the ordinary local intake (Gitea #58/#67)
        |  wikitool raw accept ...
     raw/
 
 `upload accept`/`upload reject` (the reviewer commands, `commands/upload_cmd.py`)
-are **not** importable from here or from `chemenu.mcp.server` - they live under
-`chemenu.commands`, structurally unreachable from the server, same as every
-other write command #19 already keeps out.
+and everything that reads outside the quarantine or starts a program
+(`chemenu.collect`) are **not** importable from here or from
+`chemenu.mcp.server` - they live outside it, structurally unreachable from the
+server, same as every other write command #19 already keeps out.
 
 Stdlib only, like `chemenu.telemetry` - this module is imported by the MCP
 server process on every request, not only at CLI dispatch.
@@ -47,12 +52,14 @@ import secrets
 import shutil
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 
 from chemenu import config, config_shapes, legacy_settings, settings
 from chemenu.errors import ValidationError
 
-DEFAULT_IDENTITY_HEADER = config_shapes.UPLOAD.field("identity_header").default
+DEFAULT_IDENTITY_HEADER = config_shapes.INTAKE_DEFAULT_IDENTITY_HEADER
+GITIGNORE_NAME = ".gitignore"
+STAGING_DIRNAME = ".staging"
 
 # Submissions older than this stop counting toward a submitter's quota. A
 # rolling window rather than a calendar day - "resets at midnight" is a
@@ -62,22 +69,48 @@ _QUOTA_WINDOW = datetime.timedelta(days=1)
 
 _FORBIDDEN_FILENAME_CHARS = ("/", "\\", "\x00")
 
+UNKNOWN_SUBMITTER = "unknown"
+
 
 @dataclass(frozen=True)
-class UploadConfig:
-    """The opt-in, read from the `upload` section of `.wikitool-local.json`."""
+class Channel:
+    """One channel of the `intake` section of `.wikitool-local.json`. Only the
+    fields of its handler are meaningful; the others keep their defaults."""
 
-    identity_header: str
+    name: str
+    handler: str
     max_bytes: int
     allowed_extensions: tuple[str, ...]
-    submissions_per_day: int
-    bytes_per_day: int
+    identity_header: str = DEFAULT_IDENTITY_HEADER
+    submissions_per_day: int = 0
+    bytes_per_day: int = 0
+    path: Optional[Path] = None
+    settle_seconds: int = config_shapes.INTAKE_SETTLE_SECONDS
+    command_name: str = ""
+    command: tuple[str, ...] = ()
+
+    @property
+    def pulled(self) -> bool:
+        return self.handler != config_shapes.INTAKE_HANDLER_MCP
 
 
-def read_config(root: "Path | str") -> Optional[UploadConfig]:
-    """The upload opt-in for `root`, or `None` when the `upload` section is not
-    set - which means the write path does not exist, not that it is
-    unrestricted.
+def _folder_problem(name: str, raw: str, root: "Path | str") -> Optional[str]:
+    if not os.path.isabs(raw):
+        return f"intake.channels.{name}.path {raw!r} is not an absolute path on this machine"
+    real_root = Path(root).resolve()
+    resolved = Path(raw).resolve()
+    try:
+        resolved.relative_to(real_root)
+    except ValueError:
+        return None
+    return (f"intake.channels.{name}.path {raw!r} lies inside the instance - a folder the channel "
+            "empties belongs outside it")
+
+
+def read_channels(root: "Path | str") -> dict[str, Channel]:
+    """Every channel of the `intake` section for `root`, by name, in the order
+    the file lists them - empty when the section is not set, which means no
+    channel exists, not that one is unrestricted.
 
     A malformed section is a `ValidationError`, never a silent "no limits": it
     decides whether an unauthenticated write path is offered at all, so a
@@ -89,17 +122,79 @@ def read_config(root: "Path | str") -> Optional[UploadConfig]:
     does not exist. The file of an earlier version that is still there raises
     `LegacySettingsLeftover` rather than reading as "not configured".
     """
-    legacy_settings.refuse_leftover(root, "upload")
-    section = settings.load(root, only=(config_shapes.UPLOAD.key,)).section(config_shapes.UPLOAD.key)
+    legacy_settings.refuse_leftover(root, "intake")
+    section = settings.load(root, only=(config_shapes.INTAKE.key,)).section(config_shapes.INTAKE.key)
     if not section.configured:
+        return {}
+    commands = section["commands"]
+    channels: dict[str, Channel] = {}
+    problems: list[str] = []
+    for name, raw in section["channels"].items():
+        handler = raw["handler"]
+        common = dict(
+            name=name, handler=handler, max_bytes=raw["max_bytes"],
+            allowed_extensions=config_shapes.extensions(raw["allowed_extensions"]),
+        )
+        if handler == config_shapes.INTAKE_HANDLER_MCP:
+            channels[name] = Channel(
+                identity_header=raw.get("identity_header", DEFAULT_IDENTITY_HEADER),
+                submissions_per_day=raw["quota_submissions_per_day"],
+                bytes_per_day=raw["quota_bytes_per_day"], **common)
+        elif handler == config_shapes.INTAKE_HANDLER_FOLDER:
+            problem = _folder_problem(name, raw["path"], root)
+            if problem:
+                problems.append(problem)
+                continue
+            channels[name] = Channel(
+                path=Path(raw["path"]),
+                settle_seconds=raw.get("settle_seconds", config_shapes.INTAKE_SETTLE_SECONDS), **common)
+        else:
+            channels[name] = Channel(
+                command_name=raw["command"], command=tuple(commands[raw["command"]]), **common)
+    if problems:
+        raise ValidationError("; ".join(problems))
+    return channels
+
+
+def mcp_channel(channels: dict[str, Channel]) -> Optional[Channel]:
+    """The channel with the handler `mcp` - at most one exists - or `None`."""
+    for channel in channels.values():
+        if channel.handler == config_shapes.INTAKE_HANDLER_MCP:
+            return channel
+    return None
+
+
+def read_config(root: "Path | str") -> Optional[Channel]:
+    """The `mcp` channel for `root`, or `None` when there is none - the `submit`
+    tool does not exist then. A malformed section raises like `read_channels`."""
+    return mcp_channel(read_channels(root))
+
+
+def legacy_dir_problem(root: "Path | str") -> Optional[str]:
+    """What to tell the operator when the quarantine of earlier versions
+    (`mcp-upload/`) still holds something, or `None`. Nothing is moved for
+    them: material that waits for review is not ours to relocate."""
+    legacy = Path(root) / config.LEGACY_UPLOAD_DIRNAME
+    try:
+        if not legacy.is_dir() or not any(legacy.iterdir()):
+            return None
+    except OSError:
         return None
-    return UploadConfig(
-        identity_header=section["identity_header"],
-        max_bytes=section["max_bytes"],
-        allowed_extensions=section["allowed_extensions"],
-        submissions_per_day=section["quota_submissions_per_day"],
-        bytes_per_day=section["quota_bytes_per_day"],
-    )
+    new = config.UPLOAD_DIRNAME
+    if (Path(root) / new).exists():
+        how = (f"{new}/ exists already, so move the entries of {config.LEGACY_UPLOAD_DIRNAME}/ into it "
+               f"one by one and delete the empty {config.LEGACY_UPLOAD_DIRNAME}/ afterwards")
+    else:
+        how = (f"POSIX: `mv {config.LEGACY_UPLOAD_DIRNAME} {new}`; "
+               f"pwsh: `Move-Item {config.LEGACY_UPLOAD_DIRNAME} {new}` - run in the instance root")
+    return (f"{config.LEGACY_UPLOAD_DIRNAME}/ still holds submissions from an earlier version, and the "
+            f"quarantine is {new}/ now ({how}). Nothing was changed.")
+
+
+def refuse_legacy_dir(root: "Path | str") -> None:
+    problem = legacy_dir_problem(root)
+    if problem:
+        raise ValidationError(problem)
 
 
 def sanitize_filename(name: str) -> str:
@@ -153,12 +248,26 @@ def _parse_iso(value: Any) -> Optional[datetime.datetime]:
 
 
 def _upload_root(root: "Path | str") -> Path:
-    return (Path(root) / "mcp-upload").resolve()
+    return (Path(root) / config.UPLOAD_DIRNAME).resolve()
+
+
+def _ensure_quarantine(root: "Path | str") -> Path:
+    """The quarantine directory, created with a `.gitignore` of its own that
+    ignores everything in it - so material nobody has looked at is never
+    committed, not even from a checkout whose root `.gitignore` predates the
+    directory's name. The file is a constant written in place, never a
+    caller-supplied name."""
+    base = _upload_root(root)
+    base.mkdir(parents=True, exist_ok=True)
+    marker = base / GITIGNORE_NAME
+    if not marker.exists():
+        marker.write_text("*\n", encoding="utf-8", newline="\n")
+    return base
 
 
 def _write_atomic_within(root: "Path | str", relative: Path, data: bytes) -> Path:
     """The one choke point every content/manifest write in this module goes
-    through: resolve the target, refuse anything outside `mcp-upload/` - a
+    through: resolve the target, refuse anything outside `quarantine/` - a
     resolved path also closes a symlink or a `..` in `relative` - then write
     it via a temp file plus `os.replace` so a reader never observes a partial
     file."""
@@ -167,7 +276,8 @@ def _write_atomic_within(root: "Path | str", relative: Path, data: bytes) -> Pat
     try:
         target.relative_to(base)
     except ValueError:
-        raise ValidationError(f"refusing to write outside mcp-upload/: {relative}")
+        raise ValidationError(f"refusing to write outside {config.UPLOAD_DIRNAME}/: {relative}")
+    _ensure_quarantine(root)
     target.parent.mkdir(parents=True, exist_ok=True)
     tmp = target.with_name(f".{target.name}.{secrets.token_hex(4)}.tmp")
     tmp.write_bytes(data)
@@ -177,7 +287,7 @@ def _write_atomic_within(root: "Path | str", relative: Path, data: bytes) -> Pat
 
 def _pending_dir(root: "Path | str", submission_id: str) -> Path:
     """The quarantine directory for `submission_id`, refusing a `submission_id`
-    that would resolve outside `mcp-upload/` - this one comes from a CLI
+    that would resolve outside `quarantine/` - this one comes from a CLI
     argument, a human, not the generator above, so it gets the same
     containment check as a write."""
     base = _upload_root(root)
@@ -190,12 +300,11 @@ def _pending_dir(root: "Path | str", submission_id: str) -> Path:
 
 
 def _append_ledger(root: "Path | str", event: dict) -> None:
-    """Append one event to `mcp-upload/ledger.jsonl` - the destination is this
+    """Append one event to `quarantine/ledger.jsonl` - the destination is this
     literal constant, never a caller-supplied path, which is what makes an
     append-mode write (rather than `_write_atomic_within`'s replace) safe:
     there is nothing here for untrusted input to redirect."""
-    base = _upload_root(root)
-    base.mkdir(parents=True, exist_ok=True)
+    base = _ensure_quarantine(root)
     path = base / "ledger.jsonl"
     line = json.dumps(event, sort_keys=True) + "\n"
     with path.open("a", encoding="utf-8", newline="\n") as fh:
@@ -238,7 +347,7 @@ def _pending_manifests(root: "Path | str") -> list[dict]:
 def digest_to_pending_id(root: "Path | str", digest: str) -> Optional[str]:
     """The id of a submission already waiting with this exact sha256, if any.
     Only *pending* submissions are consulted - an accepted or rejected one no
-    longer has a manifest under `mcp-upload/` - so a re-submission after a
+    longer has a manifest under `quarantine/` - so a re-submission after a
     rejection is not blocked by this check."""
     for manifest in _pending_manifests(root):
         if manifest.get("sha256") == digest:
@@ -246,13 +355,15 @@ def digest_to_pending_id(root: "Path | str", digest: str) -> Optional[str]:
     return None
 
 
-def _check_quota(root: "Path | str", cfg: UploadConfig, submitter: str, size: int) -> None:
+def _check_quota(root: "Path | str", cfg: Channel, submitter: str, size: int) -> None:
     window_start = datetime.datetime.now(datetime.timezone.utc) - _QUOTA_WINDOW
     count = 0
     total_bytes = 0
     for event in _read_ledger(root):
         if event.get("event") != "submitted" or event.get("submitter") != submitter:
             continue
+        if event.get("handler", config_shapes.INTAKE_HANDLER_MCP) != config_shapes.INTAKE_HANDLER_MCP:
+            continue  # a pulled channel has no submitter, and no quota
         when = _parse_iso(event.get("time"))
         if when is None or when < window_start:
             continue
@@ -270,9 +381,73 @@ def _check_quota(root: "Path | str", cfg: UploadConfig, submitter: str, size: in
         )
 
 
+class SourceChanged(Exception):
+    """The `verify` hook of `store()` found that the source no longer matches
+    what was read - nothing of the submission was kept."""
+
+
+def store(
+    root: "Path | str",
+    channel: Channel,
+    *,
+    filename: str,
+    data: bytes,
+    submitter: str,
+    submitter_source: str,
+    origin: str,
+    verify: Optional[Callable[[], bool]] = None,
+) -> dict:
+    """Write one already-checked submission into the quarantine - its file, its
+    manifest, a `submitted` ledger event - and return the manifest.
+
+    Every channel ends here, so the manifest says where the file came from:
+    `channel`, `handler` and `origin` (the source file's path, `command:<name>`
+    or `mcp:submit`). `verify`, when given, runs after the file is fully
+    written and before anything marks the submission as existing: if it
+    returns false the half submission is removed and `SourceChanged` is
+    raised, so a source that moved under the copy is never kept as if it
+    had been read cleanly."""
+    clean_name = sanitize_filename(filename)
+    digest = hashlib.sha256(data).hexdigest()
+    submission_id = new_submission_id()
+    _write_atomic_within(root, Path(submission_id) / clean_name, data)
+    if verify is not None and not verify():
+        shutil.rmtree(_pending_dir(root, submission_id), ignore_errors=True)
+        raise SourceChanged(clean_name)
+    manifest = {
+        "schema": 1,
+        "id": submission_id,
+        "filename": clean_name,
+        "size": len(data),
+        "sha256": digest,
+        "submitter": submitter,
+        "submitter_source": submitter_source,
+        "submitted_at": _now_iso(),
+        "channel": channel.name,
+        "handler": channel.handler,
+        "origin": origin,
+    }
+    _write_atomic_within(
+        root,
+        Path(submission_id) / "manifest.json",
+        (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode("utf-8"),
+    )
+    _append_ledger(root, {
+        "event": "submitted",
+        "id": submission_id,
+        "channel": channel.name,
+        "handler": channel.handler,
+        "submitter": submitter,
+        "time": manifest["submitted_at"],
+        "size": len(data),
+        "sha256": digest,
+    })
+    return manifest
+
+
 def submit(
     root: "Path | str",
-    cfg: UploadConfig,
+    cfg: Channel,
     *,
     filename: str,
     content_b64: str,
@@ -338,32 +513,31 @@ def submit(
 
     _check_quota(root, cfg, submitter, len(data))
 
-    submission_id = new_submission_id()
-    _write_atomic_within(root, Path(submission_id) / clean_name, data)
-    manifest = {
-        "schema": 1,
-        "id": submission_id,
-        "filename": clean_name,
-        "size": len(data),
-        "sha256": digest,
-        "submitter": submitter,
-        "submitter_source": cfg.identity_header,
-        "submitted_at": _now_iso(),
-    }
-    _write_atomic_within(
-        root,
-        Path(submission_id) / "manifest.json",
-        (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode("utf-8"),
+    return store(
+        root, cfg, filename=clean_name, data=data, submitter=submitter,
+        submitter_source=cfg.identity_header, origin="mcp:submit",
     )
-    _append_ledger(root, {
-        "event": "submitted",
-        "id": submission_id,
-        "submitter": submitter,
-        "time": manifest["submitted_at"],
-        "size": len(data),
-        "sha256": digest,
-    })
-    return manifest
+
+
+def staging_dir(root: "Path | str") -> Path:
+    """A new, empty directory for a `command` channel's program to fill, under
+    `quarantine/.staging/` - inside the quarantine, so a program that writes
+    nowhere else leaves nothing outside it. The caller removes it."""
+    base = _ensure_quarantine(root)
+    parent = base / STAGING_DIRNAME
+    parent.mkdir(exist_ok=True)
+    path = parent / secrets.token_hex(8)
+    path.mkdir()
+    return path
+
+
+def drop_staging(root: "Path | str", path: Path) -> None:
+    """Remove a staging directory and, once nothing else is staged, `.staging/`."""
+    shutil.rmtree(path, ignore_errors=True)
+    try:
+        (_upload_root(root) / STAGING_DIRNAME).rmdir()
+    except OSError:
+        pass
 
 
 def list_submissions(root: "Path | str") -> list[dict]:
@@ -375,7 +549,7 @@ def list_submissions(root: "Path | str") -> list[dict]:
 def read_manifest(root: "Path | str", submission_id: str) -> dict:
     path = _pending_dir(root, submission_id) / "manifest.json"
     if not path.is_file():
-        raise ValidationError(f"No submission '{submission_id}' is waiting in mcp-upload/.")
+        raise ValidationError(f"No submission '{submission_id}' is waiting in quarantine/.")
     try:
         data = json.loads(path.read_text(encoding="utf-8"))
     except (OSError, json.JSONDecodeError) as exc:
@@ -430,6 +604,8 @@ def promote(root: "Path | str", submission_id: str) -> Path:
     _append_ledger(root, {
         "event": "accepted",
         "id": submission_id,
+        "channel": manifest.get("channel"),
+        "handler": manifest.get("handler"),
         "submitter": manifest.get("submitter"),
         "time": _now_iso(),
         "size": manifest.get("size"),
@@ -449,6 +625,8 @@ def reject(root: "Path | str", submission_id: str, reason: str) -> None:
     _append_ledger(root, {
         "event": "rejected",
         "id": submission_id,
+        "channel": manifest.get("channel"),
+        "handler": manifest.get("handler"),
         "submitter": manifest.get("submitter"),
         "time": _now_iso(),
         "size": manifest.get("size"),